git:20260503.72c1838 to git:20260516.97ec6c2
264 added, 725 removed. Audit A to A.
# LLM Validation Module
## Overview
- The `infrastructure/llm/validation/` directory contains validation utilities for ensuring the quality, consistency, and reliability of LLM-generated content. These modules provide checks for content quality, format compliance, structural integrity, and output validation across all LLM operations in the research template system.
+ The `infrastructure/llm/validation/` directory contains validation utilities for
+ ensuring the quality, consistency, and reliability of LLM-generated content.
+ All validation is performed on LLM *output* (responses), not on prompts.
+ Public symbols are importable directly from `infrastructure.llm.validation`.
+
## Directory Structure
```mermaid
flowchart LR
V[/infrastructure/llm/validation//]
- V --> META[AGENTS.md · __init__.py]
- V --> CORE[core.py<br/>Core validation framework]
- V --> FMT[format.py<br/>Format compliance checking]
- V --> REP[repetition.py<br/>Repetition & redundancy detection]
- V --> STR[structure.py<br/>Structural validation]
+ V --> META[AGENTS.md · README.md · __init__.py]
+ V --> CORE[core.py<br/>JSON · length · structure · repetition · formatting]
+ V --> DET[detection.py<br/>Repetition detection algorithms]
+ V --> FMT[format.py<br/>Off-topic & format compliance]
+ V --> REP[repetition.py<br/>Public re-export of detection.py]
+ V --> SIM[similarity.py<br/>Internal similarity helpers]
+ V --> STR[structure.py<br/>Section validation]
classDef d fill:#0f172a,stroke:#0f172a,color:#fff
classDef code fill:#1e3a8a,stroke:#0f172a,color:#fff
classDef doc fill:#0f766e,stroke:#0f172a,color:#fff
class V d
- class CORE,FMT,REP,STR code
+ class CORE,DET,FMT,REP,SIM,STR code
class META doc
```
## Key Components
- ### Core Validation Framework (`core.py`)
-
- **Foundation classes and interfaces for validation:**
-
- #### Validation Base Classes
-
- **Abstract Validator Interface:**
-
- ```python
- class BaseValidator(ABC):
- """Abstract base class for all validators."""
-
- def __init__(self, config: ValidationConfig = None):
- self.config = config or ValidationConfig()
-
- @abstractmethod
- def validate(self, content: str, context: Dict[str, Any] = None) -> ValidationResult:
- """Validate content and return results."""
- pass
-
- def _create_result(self, passed: bool, issues: List[str] = None,
- metadata: Dict[str, Any] = None) -> ValidationResult:
- """Create standardized validation result."""
- return ValidationResult(
- validator_name=self.__class__.__name__,
- passed=passed,
- issues=issues or [],
- metadata=metadata or {},
- timestamp=datetime.now().isoformat()
- )
- ```
-
- **Validation Result Structure:**
-
- ```python
- @dataclass
- class ValidationResult:
- """Result of validation operation."""
-
- validator_name: str
- passed: bool
- issues: List[str]
- metadata: Dict[str, Any]
- timestamp: str
- severity: str = "medium" # low, medium, high, critical
-
- def __post_init__(self):
- """Set severity based on issue count and types."""
- if not self.passed:
- self.severity = self._determine_severity()
-
- def _determine_severity(self) -> str:
- """Determine issue severity."""
- if any("critical" in issue.lower() for issue in self.issues):
- return "critical"
- elif len(self.issues) > 5:
- return "high"
- elif len(self.issues) > 2:
- return "medium"
- else:
- return "low"
- ```
-
- #### Validation Configuration
-
- **Configurable Validation Parameters:**
-
- ```python
- @dataclass
- class ValidationConfig:
- """Configuration for validation operations."""
-
- # General settings
- strict_mode: bool = False
- enable_logging: bool = True
-
- # Content thresholds
- min_content_length: int = 10
- max_content_length: int = 100000
-
- # Quality thresholds
- min_quality_score: float = 0.6
- repetition_threshold: float = 0.3
-
- # Performance settings
- timeout: float = 30.0
- cache_results: bool = True
- ```
+ ### Core Validation (`core.py`)
- ### Format Validation (`format.py`)
+ All functions are importable from `infrastructure.llm.validation.core` or from
+ the package root `infrastructure.llm.validation`.
- **Format compliance and structure validation:**
+ #### Error contract
- #### Markdown Format Validator
+ - **Schema-level validators** (`validate_json`, `validate_structure`,
+ `validate_complete` in STRUCTURED mode) raise `ValidationError` on failure;
+ callers cannot recover from invalid structure.
+ - **Signal validators** (`validate_length`, `validate_short_response`,
+ `validate_long_response`, `validate_formatting`) return `bool`; callers
+ choose to warn, log, or retry.
+ - `validate_complete` raises `ValidationError` for structural problems (empty
+ content, bad schema) and returns `bool` for SHORT/LONG format failures.
- **Markdown Structure Validation:**
+ #### Functions
```python
- class MarkdownFormatValidator(BaseValidator):
- """Validate markdown format compliance and structure."""
-
- def validate(self, content: str, context: Dict[str, Any] = None) -> ValidationResult:
- """Validate markdown formatting."""
+ from infrastructure.llm.validation.core import (
+ validate_json,
+ validate_length,
+ estimate_tokens,
+ validate_short_response,
+ validate_long_response,
+ validate_structure,
+ validate_citations,
+ validate_formatting,
+ validate_complete,
+ validate_no_repetition,
+ clean_repetitive_output,
+ )
+ from infrastructure.llm.core.config import ResponseMode
- issues = []
+ # Parse JSON output; strips markdown fences before parsing.
+ # Raises ValidationError on invalid JSON.
+ data = validate_json(content)
- # Check header hierarchy
- issues.extend(self._validate_header_hierarchy(content))
+ # Check character length bounds; returns bool.
+ ok = validate_length(content, min_len=0, max_len=None)
- # Check link validity
- issues.extend(self._validate_links(content))
+ # Heuristic token estimate (1 token ≈ 4 chars).
+ tokens = estimate_tokens(content)
- # Check code block formatting
- issues.extend(self._validate_code_blocks(content))
+ # Validate short response (< 150 tokens by default); returns bool.
+ ok = validate_short_response(content, max_tokens=150)
- # Check table formatting
- issues.extend(self._validate_tables(content))
+ # Validate long response (> 500 tokens by default); returns bool.
+ ok = validate_long_response(content, min_tokens=500)
- # Check list consistency
- issues.extend(self._validate_lists(content))
+ # Validate dict against a JSON-Schema-style schema dict.
+ # Returns True or raises ValidationError.
+ validate_structure(data_dict, schema)
- passed = len(issues) == 0
- return self._create_result(passed, issues)
+ # Extract citations matching (Author Year), [1], or @key patterns.
+ citations: list[str] = validate_citations(content)
- def _validate_header_hierarchy(self, content: str) -> List[str]:
- """Validate header level progression."""
+ # Lightweight formatting quality check (!!! ??? double spaces).
+ # Returns bool; logs warning on failure.
+ ok = validate_formatting(content)
- issues = []
- lines = content.split('\n')
- last_level = 0
+ # Composite validator — dispatches to the right check for each ResponseMode.
+ # mode: SHORT | LONG | STRUCTURED | RAW | STANDARD
+ # Returns True / False for SHORT and LONG; True or raises for others.
+ ok = validate_complete(content, mode=ResponseMode.STANDARD, schema=None)
- for line in lines:
- if line.startswith('#'):
- level = len(line) - len(line.lstrip('#'))
- if level > last_level + 1:
- issues.append(f"Skipped header level: {line.strip()}")
- last_level = level
+ # Repetition gate — wraps detect_repetition.
+ # Returns (is_valid: bool, details: dict).
+ is_valid, details = validate_no_repetition(content, max_allowed_ratio=0.3)
- return issues
+ # Remove repeated sections from output using "balanced" mode.
+ cleaned = clean_repetitive_output(content, max_repetitions=2)
```
- #### Academic Format Validator
+ ### Format Compliance (`format.py`)
- **Academic Writing Standards:**
+ Detects off-topic drift and conversational AI phrases that indicate poor
+ response quality or hallucination.
```python
- class AcademicFormatValidator(BaseValidator):
- """Validate academic writing format and conventions."""
-
- def validate(self, content: str, context: Dict[str, Any] = None) -> ValidationResult:
- """Validate academic formatting standards."""
-
- issues = []
-
- # Check citation format consistency
- issues.extend(self._validate_citations(content))
+ from infrastructure.llm.validation.format import (
+ is_off_topic,
+ has_on_topic_signals,
+ detect_conversational_phrases,
+ check_format_compliance,
+ OFF_TOPIC_PATTERNS_START,
+ OFF_TOPIC_PATTERNS_ANYWHERE,
+ CONVERSATIONAL_PATTERNS,
+ ON_TOPIC_SIGNALS,
+ )
- # Check reference formatting
- issues.extend(self._validate_references(content))
+ # Two-tier off-topic check.
+ # 1. Returns False immediately if ≥2 ON_TOPIC_SIGNALS match.
+ # 2. Then checks OFF_TOPIC_PATTERNS_START (first 100 chars).
+ # 3. Then checks OFF_TOPIC_PATTERNS_ANYWHERE.
+ off_topic: bool = is_off_topic(text)
- # Check figure/table references
- issues.extend(self._validate_cross_references(content))
+ # True if ≥2 on-topic signals are present (overrides off-topic detection).
+ on_topic: bool = has_on_topic_signals(text)
- # Check section structure
- issues.extend(self._validate_academic_structure(content))
+ # Returns list of matched conversational phrases (up to 50 chars each).
+ phrases: list[str] = detect_conversational_phrases(text)
- passed = len(issues) == 0
- return self._create_result(passed, issues)
+ # Full format compliance check.
+ # Returns (is_compliant: bool, issues: list[str], details: dict).
+ is_compliant, issues, details = check_format_compliance(response)
```
- ### Repetition Detection (`repetition.py`)
+ **Pattern constants** (all `list[str]` of regex patterns):
- **Content redundancy and repetition analysis:**
+ | Constant | Purpose |
+ |----------|---------|
+ | `OFF_TOPIC_PATTERNS_START` | Email/letter headers, casual greetings, book intro phrases — checked at response start |
+ | `OFF_TOPIC_PATTERNS_ANYWHERE` | AI refusal phrases, self-identification, external URLs, code-focused responses |
+ | `CONVERSATIONAL_PATTERNS` | Chatbot-style phrases that indicate poor formal review quality |
+ | `ON_TOPIC_SIGNALS` | Manuscript review markers (`## overview`, `the manuscript`, etc.) |
- #### Repetition Detector
+ ### Repetition Detection (`repetition.py` / `detection.py`)
- **Advanced Repetition Analysis:**
+ `repetition.py` re-exports the public API from `detection.py`. Import from
+ `repetition` — `detection.py` and `similarity.py` are internal.
```python
- class RepetitionDetector(BaseValidator):
- """Detect content repetition and redundancy."""
-
- def validate(self, content: str, context: Dict[str, Any] = None) -> ValidationResult:
- """Analyze content for repetition."""
-
- issues = []
-
- # Sentence-level repetition
- sentence_issues = self._detect_sentence_repetition(content)
- issues.extend(sentence_issues)
-
- # Phrase-level repetition
- phrase_issues = self._detect_phrase_repetition(content)
- issues.extend(phrase_issues)
-
- # Word frequency analysis
- word_issues = self._analyze_word_frequency(content)
- issues.extend(word_issues)
-
- # Structural repetition
- structural_issues = self._detect_structural_repetition(content)
- issues.extend(structural_issues)
-
- passed = len(issues) == 0
- return self._create_result(passed, issues)
+ from infrastructure.llm.validation.repetition import (
+ RepetitionResult,
+ detect_repetition,
+ calculate_unique_content_ratio,
+ deduplicate_sections,
+ )
- def _detect_sentence_repetition(self, content: str) -> List[str]:
- """Detect repeated sentences."""
+ # RepetitionResult is a NamedTuple with three fields.
+ result: RepetitionResult = detect_repetition(text)
+ print(result.found) # bool
+ print(result.examples) # list[str] — first 100 chars of each duplicate
+ print(result.unique_ratio) # float 0.0–1.0
- issues = []
- sentences = self._split_sentences(content)
- sentence_counts = Counter(sentences)
+ # Positional unpacking is supported (7 call sites rely on it):
+ found, examples, ratio = detect_repetition(text)
- for sentence, count in sentence_counts.items():
- if count > 1 and len(sentence.strip()) > 20: # Ignore very short sentences
- repetition_ratio = count / len(sentences)
- if repetition_ratio > self.config.repetition_threshold:
- issues.append(f"Repeated sentence ({count} times): '{sentence.strip()[:50]}...'")
+ # Unique content ratio (lower = more repetitive).
+ ratio: float = calculate_unique_content_ratio(text, chunk_size=200)
- return issues
+ # Remove repeated sections from output.
+ # mode: "conservative" (≥0.9 similarity, ≥3 repeats before removal)
+ # "balanced" (uses caller-supplied thresholds as-is)
+ # "aggressive" (≤0.7 similarity, ≤1 repeat before removal)
+ cleaned = deduplicate_sections(
+ text,
+ max_repetitions=2,
+ mode="conservative",
+ similarity_threshold=0.85,
+ min_content_preservation=0.7,
+ )
```
- #### Semantic Similarity Detection
-
- **Meaning-Based Repetition:**
-
- ```python
- class SemanticRepetitionDetector(BaseValidator):
- """Detect semantic repetition using similarity analysis."""
-
- def validate(self, content: str, context: Dict[str, Any] = None) -> ValidationResult:
- """Detect semantically similar content."""
-
- issues = []
-
- # Split content into segments
- segments = self._segment_content(content)
-
- # Calculate pairwise similarities
- similarities = self._calculate_similarities(segments)
-
- # Find highly similar segments
- for i, j in combinations(range(len(segments)), 2):
- if similarities[i][j] > 0.8: # High similarity threshold
- issues.append(f"Highly similar content segments: {i+1} and {j+1}")
-
- passed = len(issues) == 0
- return self._create_result(passed, issues)
- ```
+ `detect_repetition` uses header-based section splitting (H1–H3, triple
+ newlines, paragraphs) and a hybrid similarity score combining Jaccard,
+ TF-cosine, and 3-gram overlap.
### Structure Validation (`structure.py`)
- **Content organization and structural integrity:**
-
- #### Document Structure Validator
-
- **Structure Analysis:**
-
```python
- class DocumentStructureValidator(BaseValidator):
- """Validate document structure and organization."""
-
- def validate(self, content: str, context: Dict[str, Any] = None) -> ValidationResult:
- """Validate document structural integrity."""
-
- issues = []
-
- # Check required sections
- issues.extend(self._validate_required_sections(content, context))
-
- # Check section ordering
- issues.extend(self._validate_section_order(content, context))
-
- # Check content distribution
- issues.extend(self._validate_content_distribution(content))
-
- # Check transition quality
- issues.extend(self._validate_transitions(content))
-
- passed = len(issues) == 0
- return self._create_result(passed, issues)
-
- def _validate_required_sections(self, content: str, context: Dict[str, Any]) -> List[str]:
- """Check for required sections based on document type."""
-
- issues = []
- doc_type = context.get('document_type', 'general') if context else 'general'
-
- # Define required sections by document type
- required_sections = {
- 'research_paper': ['introduction', 'methods', 'results', 'discussion'],
- 'review': ['summary', 'analysis', 'conclusions'],
- 'manuscript': ['abstract', 'introduction', 'methods', 'results', 'discussion']
- }
+ from infrastructure.llm.validation.structure import (
+ validate_section_completeness,
+ extract_structured_sections,
+ validate_response_structure,
+ )
- required = required_sections.get(doc_type, [])
- content_lower = content.lower()
+ # Check that required markdown headers are present.
+ # flexible=True accepts semantic equivalents (e.g. "overview" matches "## Overview").
+ # Returns (is_complete: bool, missing: list[str], details: dict).
+ is_complete, missing, details = validate_section_completeness(
+ response,
+ required_headers=["## Overview", "## Results"],
+ flexible=True,
+ )
- for section in required:
- if section not in content_lower:
- issues.append(f"Missing required section: {section}")
+ # Parse markdown headers into a dict of {header_text: section_content}.
+ sections: dict[str, str] = extract_structured_sections(response)
- return issues
+ # Composite check: word count + section completeness.
+ # Returns (is_valid: bool, issues: list[str], details: dict).
+ is_valid, issues, details = validate_response_structure(
+ response,
+ required_headers=["## Overview", "## Results"],
+ min_word_count=200,
+ max_word_count=5000,
+ flexible_headers=True,
+ )
```
- #### Content Flow Validator
-
- **Logical Flow and Coherence:**
-
- ```python
- class ContentFlowValidator(BaseValidator):
- """Validate content flow and logical coherence."""
-
- def validate(self, content: str, context: Dict[str, Any] = None) -> ValidationResult:
- """Validate content logical flow."""
-
- issues = []
-
- # Check topic consistency
- issues.extend(self._validate_topic_consistency(content))
-
- # Check argument progression
- issues.extend(self._validate_argument_progression(content))
-
- # Check conclusion alignment
- issues.extend(self._validate_conclusion_alignment(content))
+ ### Similarity Helpers (`similarity.py`)
- # Check transition quality
- issues.extend(self._validate_transition_quality(content))
+ **Internal module — do not import directly.** Used by `detection.py`.
- passed = len(issues) == 0
- return self._create_result(passed, issues)
- ```
+ Provides `_jaccard_similarity`, `_tf_cosine_similarity`, `_sequence_similarity`,
+ `_calculate_similarity` (hybrid combiner), and `_normalize_for_comparison`.
## Validation Integration
- ### Composite Validation System
-
- **Multi-Validator Orchestration:**
-
- ```python
- class ValidationOrchestrator:
- """Orchestrate multiple validators for validation."""
-
- def __init__(self, validators: List[BaseValidator] = None):
- self.validators = validators or self._create_default_validators()
-
- def _create_default_validators(self) -> List[BaseValidator]:
- """Create default set of validators."""
- return [
- MarkdownFormatValidator(),
- AcademicFormatValidator(),
- RepetitionDetector(),
- DocumentStructureValidator(),
- ContentFlowValidator()
- ]
-
- def validate_comprehensive(self, content: str,
- context: Dict[str, Any] = None) -> ComprehensiveValidationResult:
- """Run all validators and aggregate results."""
-
- all_results = []
- all_issues = []
-
- for validator in self.validators:
- try:
- result = validator.validate(content, context)
- all_results.append(result)
- all_issues.extend(result.issues)
- except Exception as e:
- logger.error(f"Validator {validator.__class__.__name__} failed: {e}")
- # Continue with other validators
-
- # Aggregate results
- overall_passed = all(result.passed for result in all_results)
- highest_severity = max((result.severity for result in all_results),
- key=lambda x: ['low', 'medium', 'high', 'critical'].index(x))
-
- return ComprehensiveValidationResult(
- overall_passed=overall_passed,
- individual_results=all_results,
- all_issues=all_issues,
- highest_severity=highest_severity
- )
- ```
-
- ### LLM Response Validation
-
- **Post-Generation Validation:**
-
- ```python
- # Integration with LLM core
- from infrastructure.llm.core import LLMClient
-
- class ValidatingLLMClient(LLMClient):
- """LLM client with built-in response validation."""
-
- def __init__(self, *args, validator: ValidationOrchestrator = None, **kwargs):
- super().__init__(*args, **kwargs)
- self.validator = validator or ValidationOrchestrator()
-
- def query_with_validation(self, prompt: str, **kwargs) -> ValidatedResponse:
- """Query with automatic validation."""
-
- # Generate response
- response = self.query(prompt, **kwargs)
-
- # Validate response
- validation_result = self.validator.validate_comprehensive(
- response,
- context={'source': 'llm_response', 'prompt': prompt}
- )
-
- return ValidatedResponse(
- content=response,
- validation_result=validation_result
- )
- ```
-
- ## Testing
-
- ### Validator Testing
-
- **Individual Validator Tests:**
-
- ```python
- def test_markdown_format_validator():
- """Test markdown format validation."""
-
- validator = MarkdownFormatValidator()
-
- # Valid markdown
- valid_content = "# Header\n\nSome content with a sample link."
- result = validator.validate(valid_content)
- assert result.passed
-
- # Invalid markdown (skipped header levels)
- invalid_content = "# Level 1\n\n### Level 3 (skipped level 2)"
- result = validator.validate(invalid_content)
- assert not result.passed
- assert "Skipped header level" in str(result.issues)
- ```
-
- **Repetition Detection Tests:**
-
- ```python
- def test_repetition_detector():
- """Test repetition detection."""
-
- validator = RepetitionDetector()
-
- # Content with repetition
- repetitive_content = "This is a test. This is a test. This is a test."
- result = validator.validate(repetitive_content)
- assert not result.passed
- assert len(result.issues) > 0
-
- # Content without significant repetition
- unique_content = "This is the first sentence. Here is another sentence. Finally, a third sentence."
- result = validator.validate(unique_content)
- assert result.passed
- ```
-
- ### Integration Testing
-
- **Validation Tests:**
-
- ```python
- def test_validation_orchestrator():
- """Test validation orchestration."""
-
- orchestrator = ValidationOrchestrator()
-
- # Test content
- test_content = """
- # Introduction
-
- This paper presents research on machine learning.
-
- ## Methods
-
- We used Python for implementation.
-
- ## Results
-
- The results show improvement.
-
- ## Discussion
-
- This is a good result. This is a good result. This is a good result.
- """
-
- context = {'document_type': 'research_paper'}
-
- # Run validation
- result = orchestrator.validate_comprehensive(test_content, context)
-
- # Should detect some issues (repetition)
- assert not result.overall_passed
- assert len(result.all_issues) > 0
- assert result.highest_severity in ['low', 'medium', 'high', 'critical']
- ```
-
- ## Performance Considerations
-
- ### Efficient Validation
-
- **Optimized Validation Strategies:**
-
- ```python
- class CachingValidator(BaseValidator):
- """Validator with result caching for performance."""
-
- def __init__(self, *args, **kwargs):
- super().__init__(*args, **kwargs)
- self._cache = {}
-
- def validate(self, content: str, context: Dict[str, Any] = None) -> ValidationResult:
- """Validate with caching."""
-
- cache_key = self._generate_cache_key(content, context)
-
- if cache_key in self._cache:
- return self._cache[cache_key]
-
- result = super().validate(content, context)
-
- # Cache result (with size limits)
- if len(self._cache) < 100: # Max cache size
- self._cache[cache_key] = result
-
- return result
-
- def _generate_cache_key(self, content: str, context: Dict[str, Any]) -> str:
- """Generate cache key from content and context."""
- import hashlib
- key_data = content + str(sorted(context.items()) if context else "")
- return hashlib.md5(key_data.encode()).hexdigest()
- ```
-
- ### Parallel Validation
-
- **Concurrent Validation Processing:**
-
- ```python
- import concurrent.futures
-
- class ParallelValidationOrchestrator(ValidationOrchestrator):
- """Run validators in parallel for better performance."""
-
- def validate_comprehensive_parallel(self, content: str,
- context: Dict[str, Any] = None,
- max_workers: int = 4) -> ComprehensiveValidationResult:
- """Run validators in parallel."""
-
- with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
- # Submit all validation tasks
- future_to_validator = {
- executor.submit(validator.validate, content, context): validator
- for validator in self.validators
- }
-
- # Collect results
- all_results = []
- all_issues = []
-
- for future in concurrent.futures.as_completed(future_to_validator):
- validator = future_to_validator[future]
- try:
- result = future.result(timeout=30.0)
- all_results.append(result)
- all_issues.extend(result.issues)
- except Exception as e:
- logger.error(f"Validator {validator.__class__.__name__} failed: {e}")
-
- # Aggregate results (same as sequential version)
- overall_passed = all(result.passed for result in all_results)
- highest_severity = max((result.severity for result in all_results),
- key=lambda x: ['low', 'medium', 'high', 'critical'].index(x))
-
- return ComprehensiveValidationResult(
- overall_passed=overall_passed,
- individual_results=all_results,
- all_issues=all_issues,
- highest_severity=highest_severity
- )
- ```
-
- ## Error Handling
-
- ### Validation Failure Handling
-
- **Robust Error Recovery:**
-
- ```python
- def validate_with_error_handling(self, content: str,
- context: Dict[str, Any] = None) -> ValidationResult:
- """Validate with error handling."""
-
- try:
- # Input validation
- if not isinstance(content, str):
- raise ValidationError("Content must be a string")
-
- if len(content) == 0:
- return self._create_result(False, ["Content is empty"])
-
- # Perform validation
- return self.validate(content, context)
-
- except ValidationError as e:
- logger.error(f"Validation input error: {e}")
- return self._create_result(False, [str(e)])
-
- except Exception as e:
- logger.error(f"Unexpected validation error: {e}")
- return self._create_result(False, [f"Validation failed: {str(e)}"])
- ```
-
- ### Validation Result Processing
-
- **Result Interpretation and Action:**
+ ### Post-Generation Validation
```python
- def process_validation_result(result: ValidationResult) -> ValidationAction:
- """Process validation result and determine action."""
-
- if result.passed:
- return ValidationAction.ACCEPT
-
- # Determine action based on severity and issues
- if result.severity == "critical":
- return ValidationAction.REJECT
- elif result.severity == "high":
- return ValidationAction.FLAG_FOR_REVIEW
- elif result.severity == "medium":
- if len(result.issues) > 3:
- return ValidationAction.REQUIRE_FIXES
- else:
- return ValidationAction.FLAG_FOR_REVIEW
- else: # low
- return ValidationAction.ACCEPT_WITH_NOTES
- ```
+ from infrastructure.llm.validation import (
+ validate_complete,
+ validate_no_repetition,
+ is_off_topic,
+ clean_repetitive_output,
+ validate_response_structure,
+ )
+ from infrastructure.llm.core.config import ResponseMode
+ from infrastructure.core.exceptions import ValidationError
- ## Usage Examples
+ def validate_llm_response(response: str, required_headers: list[str] | None = None) -> str:
+ """Validate and clean an LLM response. Returns cleaned text or raises."""
- ### Basic Validation
+ # Reject off-topic / hallucinated output
+ if is_off_topic(response):
+ raise ValidationError("Response is off-topic")
- **Simple Content Validation:**
+ # Clean excessive repetition
+ is_valid, details = validate_no_repetition(response)
+ if not is_valid:
+ response = clean_repetitive_output(response)
- ```python
- from infrastructure.llm.validation import MarkdownFormatValidator
+ # Structural check
+ if required_headers:
+ is_valid, issues, _ = validate_response_structure(response, required_headers)
+ if not is_valid:
+ raise ValidationError(f"Structure issues: {issues}")
- validator = MarkdownFormatValidator()
- content = "# Header\n\nSome content with a sample link."
- result = validator.validate(content)
+ # Final composite check (raises on empty content)
+ validate_complete(response, mode=ResponseMode.STANDARD)
- if result.passed:
- print("Content is valid")
- else:
- print(f"Validation issues: {result.issues}")
+ return response
```
- ### Comprehensive Validation
-
- **Multi-Validator Assessment:**
+ ### JSON Response Validation
```python
- from infrastructure.llm.validation import ValidationOrchestrator
-
- orchestrator = ValidationOrchestrator()
- content = "# Research Paper\n\n## Introduction\n\nThis is the introduction..."
- context = {'document_type': 'research_paper'}
+ from infrastructure.llm.validation import validate_json, validate_structure
+ from infrastructure.core.exceptions import ValidationError
- result = orchestrator.validate_comprehensive(content, context)
+ schema = {
+ "required": ["title", "summary"],
+ "properties": {
+ "title": {"type": "string"},
+ "summary": {"type": "string"},
+ },
+ }
- print(f"Overall validation: {'PASSED' if result.overall_passed else 'FAILED'}")
- print(f"Highest severity: {result.highest_severity}")
- print(f"Total issues: {len(result.all_issues)}")
+ try:
+ data = validate_json(response_text) # strips ```json fences, parses
+ validate_structure(data, schema) # required keys + basic type check
+ except ValidationError as e:
+ print(f"Structured response invalid: {e}")
```
- ### LLM Response Validation Example
-
- **Post-Generation Quality Check:**
+ ### Repetition Cleaning Pipeline
```python
- from infrastructure.llm.validation import ValidatingLLMClient
+ from infrastructure.llm.validation import (
+ detect_repetition,
+ deduplicate_sections,
+ calculate_unique_content_ratio,
+ )
- client = ValidatingLLMClient()
- response = client.query_with_validation("Write a research summary")
+ # Inspect first
+ found, examples, ratio = detect_repetition(response_text)
+ print(f"Unique ratio: {ratio:.0%}, duplicates found: {len(examples)}")
- if response.validation_result.overall_passed:
- print("Response passed validation")
- print(f"Content: {response.content}")
+ # Clean with different strategies
+ if ratio < 0.5:
+ # Aggressively repetitive — use balanced mode
+ cleaned = deduplicate_sections(response_text, mode="balanced", similarity_threshold=0.8)
else:
- print("Response failed validation:")
- for issue in response.validation_result.all_issues:
- print(f" - {issue}")
+ # Mildly repetitive — conservative to preserve valid content
+ cleaned = deduplicate_sections(response_text, mode="conservative")
```
- ## Configuration
+ ## Testing
- ### Custom Validation Settings
+ ### Unit Tests
- **Custom Validation Settings:**
+ Tests live in `tests/infra_tests/` and follow the no-mocks policy: use real
+ strings and computed values.
```python
- from infrastructure.llm.validation.core import ValidationConfig
-
- config = ValidationConfig(
- strict_mode=True,
- min_content_length=50,
- max_content_length=50000,
- min_quality_score=0.8,
- repetition_threshold=0.2,
- enable_logging=True,
- cache_results=True
+ from infrastructure.llm.validation import (
+ validate_json,
+ detect_repetition,
+ is_off_topic,
+ validate_section_completeness,
)
+ from infrastructure.core.exceptions import ValidationError
+ import pytest
- validator = MarkdownFormatValidator(config)
- ```
+ def test_validate_json_strips_fences():
+ content = "```json\n{\"key\": \"value\"}\n```"
+ data = validate_json(content)
+ assert data == {"key": "value"}
- ### Environment Configuration
+ def test_validate_json_raises_on_bad_input():
+ with pytest.raises(ValidationError):
+ validate_json("not json at all")
- **Runtime Validation Settings:**
+ def test_detect_repetition_finds_duplicates():
+ text = "## Introduction\nSame content.\n\n## Introduction\nSame content."
+ found, examples, ratio = detect_repetition(text)
+ assert found
+ assert ratio < 1.0
- ```bash
- # Validation behavior
- export LLM_VALIDATION_STRICT_MODE=false
- export LLM_VALIDATION_MIN_CONTENT_LENGTH=10
- export LLM_VALIDATION_MAX_CONTENT_LENGTH=100000
+ def test_is_off_topic_rejects_ai_refusal():
+ assert is_off_topic("I can't help with that request.")
- # Quality thresholds
- export LLM_VALIDATION_MIN_QUALITY_SCORE=0.6
- export LLM_VALIDATION_REPETITION_THRESHOLD=0.3
+ def test_is_off_topic_accepts_manuscript_review():
+ review = "## Overview\nThe manuscript presents strong methodology.\nThe authors clearly..."
+ assert not is_off_topic(review)
- # Performance settings
- export LLM_VALIDATION_TIMEOUT=30.0
- export LLM_VALIDATION_CACHE_RESULTS=true
+ def test_validate_section_completeness():
+ response = "## Overview\nContent.\n\n## Results\nData."
+ ok, missing, _ = validate_section_completeness(response, ["## Overview", "## Results"])
+ assert ok
+ assert missing == []
```
- ## Future Enhancements
-
- ### Advanced Validation Features
-
- **Planned Improvements:**
-
- - **Machine Learning-Based Validation**: ML models for content quality assessment
- - **Domain-Specific Validators**: Specialized validators for different research fields
- - **Real-time Validation**: Streaming validation during content generation
- - **Collaborative Validation**: Multi-user validation workflows
-
- **Integration Features:**
-
- - **IDE Integration**: Real-time validation in text editors
- - **API Integration**: Validation as a service for external tools
- - **Batch Validation**: Process multiple documents efficiently
- - **Validation Reports**: Detailed HTML/PDF validation reports
-
- ## Troubleshooting
-
- ### Common Validation Issues
-
- **False Positives:**
+ ## Error Handling
- ```python
- # Adjust validation sensitivity
- config = ValidationConfig(
- strict_mode=False, # Less strict validation
- repetition_threshold=0.4 # Higher repetition threshold
- )
+ `ValidationError` is defined in `infrastructure.core._exceptions_core` and
+ re-exported from `infrastructure.core.exceptions`. It is raised by:
- validator = RepetitionDetector(config)
- ```
+ - `validate_json` — invalid JSON
+ - `validate_structure` — missing required field or wrong type
+ - `validate_complete` — empty content, missing schema in STRUCTURED mode, or failed structure check
- **Performance Issues:**
+ Signal validators (`validate_length`, `validate_formatting`, etc.) never raise;
+ they return `False` and log a warning.
```python
- # Optimize for performance
- config = ValidationConfig(
- cache_results=True, # Enable caching
- timeout=10.0 # Shorter timeout
- )
-
- orchestrator = ValidationOrchestrator()
- orchestrator = ParallelValidationOrchestrator() # Use parallel processing
- ```
-
- **Configuration Issues:**
+ from infrastructure.core.exceptions import ValidationError
- ```python
- # Validate configuration
try:
- config = ValidationConfig(min_content_length=-1) # Invalid
- except ValueError as e:
- print(f"Configuration error: {e}")
- config = ValidationConfig() # Use defaults
- ```
-
- ### Debug Validation
-
- **Verbose Validation Logging:**
-
- ```python
- import logging
- logging.basicConfig(level=logging.DEBUG)
-
- # Enable debug logging for validators
- config = ValidationConfig(enable_logging=True)
- validator = DocumentStructureValidator(config)
-
- result = validator.validate(content, context)
- # Check logs for detailed validation steps
+ data = validate_json(response_text)
+ validate_structure(data, schema)
+ except ValidationError as e:
+ # e.context contains structured details (field name, expected/got types, etc.)
+ print(f"Validation failed: {e}")
```
## See Also
**Related Documentation:**
- [`../core/AGENTS.md`](../core/AGENTS.md) - LLM core functionality
- - [`../templates/AGENTS.md`](../templates/AGENTS.md) - Template system
- [`../review/AGENTS.md`](../review/AGENTS.md) - Review generation
**System Documentation:**
- [`../../../AGENTS.md`](../../../AGENTS.md) - system overview
- [`../../../docs/development/testing/testing-guide.md`](../../../docs/development/testing/testing-guide.md) - Testing and validation guide