"""
Pattern 269: Validation Pattern with LangChain/LangGraph

Purpose: Data validation with rule checking and constraint enforcement.
Provides intelligent validation orchestration with LLM-powered rule analysis.
"""

from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json


class ValidationType(Enum):
    """Types of validation."""
    TYPE = "type"
    RANGE = "range"
    FORMAT = "format"
    SCHEMA = "schema"
    BUSINESS = "business"
    CUSTOM = "custom"


class ValidationSeverity(Enum):
    """Severity of validation issues."""
    ERROR = "error"
    WARNING = "warning"
    INFO = "info"


@dataclass
class ValidationRule:
    """Validation rule definition."""
    name: str
    validation_type: ValidationType
    validator: Callable
    error_message: str
    severity: ValidationSeverity = ValidationSeverity.ERROR
    field: Optional[str] = None


@dataclass
class ValidationResult:
    """Result of validation."""
    field: str
    rule: str
    passed: bool
    error_message: Optional[str] = None
    severity: ValidationSeverity = ValidationSeverity.ERROR
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ValidationInsights:
    """LLM-generated validation insights."""
    overall_assessment: str
    critical_issues: List[str]
    recommendations: List[str]
    fix_suggestions: List[str]


class ValidationState(TypedDict):
    """State for validation workflow."""
    data: Dict[str, Any]
    rules: List[ValidationRule]
    field_results: Dict[str, List[ValidationResult]]
    all_passed: bool
    errors: List[str]
    warnings: List[str]
    insights: Optional[ValidationInsights]
    validation_summary: Dict[str, Any]


class IntelligentValidator:
    """
    Intelligent Validator with LangGraph workflow integration.
    
    Features:
    - Multiple validation types
    - LLM-powered rule analysis
    - Intelligent fix suggestions
    - Severity classification
    - Batch validation
    - Custom validators
    - Pattern detection
    - Compliance checking
    """
    
    def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.5):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        
        # Rules
        self.rules: Dict[str, List[ValidationRule]] = {}
        
        # History
        self.validation_history: List[ValidationResult] = []
        
        # Build workflow
        self.workflow = self._build_workflow()
        
        # Statistics
        self.stats = {
            "total_validations": 0,
            "passed": 0,
            "failed": 0,
            "validation_types": {t.value: 0 for t in ValidationType}
        }
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow."""
        workflow = StateGraph(ValidationState)
        
        workflow.add_node("parse_rules", self._parse_rules_node)
        workflow.add_node("validate_fields", self._validate_fields_node)
        workflow.add_node("aggregate_results", self._aggregate_results_node)
        workflow.add_node("generate_insights", self._generate_insights_node)
        workflow.add_node("recommend_fixes", self._recommend_fixes_node)
        
        workflow.set_entry_point("parse_rules")
        workflow.add_edge("parse_rules", "validate_fields")
        workflow.add_edge("validate_fields", "aggregate_results")
        workflow.add_edge("aggregate_results", "generate_insights")
        workflow.add_edge("generate_insights", "recommend_fixes")
        workflow.add_edge("recommend_fixes", END)
        
        return workflow.compile()
    
    def _parse_rules_node(self, state: ValidationState) -> ValidationState:
        """Parse and organize validation rules."""
        # Organize rules by field
        all_rules = []
        for field, rules in self.rules.items():
            all_rules.extend(rules)
        
        state["rules"] = all_rules
        state["field_results"] = {}
        
        return state
    
    def _validate_fields_node(self, state: ValidationState) -> ValidationState:
        """Validate all fields."""
        data = state["data"]
        field_results = {}
        
        # Validate each field
        for field, value in data.items():
            if field not in self.rules:
                continue
            
            field_results[field] = []
            
            for rule in self.rules[field]:
                self.stats["total_validations"] += 1
                self.stats["validation_types"][rule.validation_type.value] += 1
                
                try:
                    passed = rule.validator(value)
                    
                    result = ValidationResult(
                        field=field,
                        rule=rule.name,
                        passed=passed,
                        error_message=None if passed else rule.error_message,
                        severity=rule.severity
                    )
                    
                    if passed:
                        self.stats["passed"] += 1
                    else:
                        self.stats["failed"] += 1
                    
                    field_results[field].append(result)
                    self.validation_history.append(result)
                
                except Exception as e:
                    result = ValidationResult(
                        field=field,
                        rule=rule.name,
                        passed=False,
                        error_message=f"Validation error: {str(e)}",
                        severity=ValidationSeverity.ERROR
                    )
                    field_results[field].append(result)
                    self.validation_history.append(result)
                    self.stats["failed"] += 1
        
        state["field_results"] = field_results
        return state
    
    def _aggregate_results_node(self, state: ValidationState) -> ValidationState:
        """Aggregate validation results."""
        field_results = state["field_results"]
        
        errors = []
        warnings = []
        all_passed = True
        
        for field, results in field_results.items():
            for result in results:
                if not result.passed:
                    if result.severity == ValidationSeverity.ERROR:
                        errors.append(f"{field}: {result.error_message}")
                        all_passed = False
                    elif result.severity == ValidationSeverity.WARNING:
                        warnings.append(f"{field}: {result.error_message}")
        
        state["errors"] = errors
        state["warnings"] = warnings
        state["all_passed"] = all_passed
        
        # Create summary
        state["validation_summary"] = {
            "total_fields": len(state["data"]),
            "validated_fields": len(field_results),
            "total_rules": sum(len(results) for results in field_results.values()),
            "errors": len(errors),
            "warnings": len(warnings),
            "passed": all_passed
        }
        
        return state
    
    def _generate_insights_node(self, state: ValidationState) -> ValidationState:
        """Generate validation insights using LLM."""
        errors = state["errors"]
        warnings = state["warnings"]
        summary = state["validation_summary"]
        
        if not errors and not warnings:
            state["insights"] = ValidationInsights(
                overall_assessment="All validations passed successfully",
                critical_issues=[],
                recommendations=["Data is valid and ready for processing"],
                fix_suggestions=[]
            )
            return state
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert in data validation and quality assurance."),
            ("user", """Analyze these validation results and provide insights:

Validation Summary:
- Total Fields: {total_fields}
- Validated Fields: {validated_fields}
- Total Rules: {total_rules}
- Errors: {error_count}
- Warnings: {warning_count}

Errors:
{errors}

Warnings:
{warnings}

Provide analysis as JSON with:
- overall_assessment: Brief assessment of data quality
- critical_issues: List of most critical issues (2-3 items)
- recommendations: List of recommendations (2-3 items)
- fix_suggestions: List of specific fixes (2-3 items)

Focus on actionable insights.""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "total_fields": summary["total_fields"],
            "validated_fields": summary["validated_fields"],
            "total_rules": summary["total_rules"],
            "error_count": summary["errors"],
            "warning_count": summary["warnings"],
            "errors": "\n".join(errors) if errors else "None",
            "warnings": "\n".join(warnings) if warnings else "None"
        })
        
        try:
            insights_data = json.loads(response.content)
            state["insights"] = ValidationInsights(
                overall_assessment=insights_data.get("overall_assessment", "Analysis complete"),
                critical_issues=insights_data.get("critical_issues", []),
                recommendations=insights_data.get("recommendations", []),
                fix_suggestions=insights_data.get("fix_suggestions", [])
            )
        except:
            state["insights"] = ValidationInsights(
                overall_assessment="Validation completed with issues",
                critical_issues=errors[:3] if errors else [],
                recommendations=["Review validation errors", "Fix data issues"],
                fix_suggestions=["Check field values", "Ensure correct data types"]
            )
        
        return state
    
    def _recommend_fixes_node(self, state: ValidationState) -> ValidationState:
        """Generate fix recommendations."""
        # Insights already contain fix suggestions from LLM
        return state
    
    def add_rule(
        self,
        field: str,
        name: str,
        validation_type: ValidationType,
        validator: Callable,
        error_message: str,
        severity: ValidationSeverity = ValidationSeverity.ERROR
    ):
        """Add validation rule."""
        rule = ValidationRule(
            name=name,
            validation_type=validation_type,
            validator=validator,
            error_message=error_message,
            severity=severity,
            field=field
        )
        
        if field not in self.rules:
            self.rules[field] = []
        
        self.rules[field].append(rule)
    
    def validate(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Validate data through workflow."""
        initial_state: ValidationState = {
            "data": data,
            "rules": [],
            "field_results": {},
            "all_passed": True,
            "errors": [],
            "warnings": [],
            "insights": None,
            "validation_summary": {}
        }
        
        final_state = self.workflow.invoke(initial_state)
        
        return {
            "valid": final_state["all_passed"],
            "errors": final_state["errors"],
            "warnings": final_state["warnings"],
            "summary": final_state["validation_summary"],
            "insights": final_state["insights"],
            "field_results": final_state["field_results"]
        }
    
    def is_valid(self, data: Dict[str, Any]) -> bool:
        """Check if data is valid."""
        result = self.validate(data)
        return result["valid"]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get validation statistics."""
        return {
            "name": self.name,
            "rules_count": sum(len(rules) for rules in self.rules.values()),
            **self.stats
        }


class ValidationPattern:
    """
    Comprehensive Validation Pattern coordinator with LangChain/LangGraph.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.validators: Dict[str, IntelligentValidator] = {}
        self.model_name = model_name
    
    def create_validator(self, name: str) -> IntelligentValidator:
        """Create a validator."""
        validator = IntelligentValidator(name, model_name=self.model_name)
        self.validators[name] = validator
        return validator
    
    def validate(
        self,
        validator_name: str,
        data: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Run validation."""
        if validator_name not in self.validators:
            return {"error": f"Validator {validator_name} not found"}
        
        validator = self.validators[validator_name]
        result = validator.validate(data)
        
        return {
            "validator": validator_name,
            **result
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get overall statistics."""
        total_validations = sum(v.stats["total_validations"] for v in self.validators.values())
        total_passed = sum(v.stats["passed"] for v in self.validators.values())
        total_failed = sum(v.stats["failed"] for v in self.validators.values())
        
        return {
            "total_validators": len(self.validators),
            "total_validations": total_validations,
            "total_passed": total_passed,
            "total_failed": total_failed
        }


if __name__ == "__main__":
    print("=== Validation Pattern with LangChain/LangGraph Demo ===\n")
    
    print("Scenario 1: User Data Validation")
    print("-" * 50)
    pattern = ValidationPattern(model_name="gpt-4")
    
    user_validator = pattern.create_validator("UserValidator")
    
    # Add validation rules
    print("\nAdding validation rules...")
    
    # Type validation
    user_validator.add_rule(
        "age",
        "type_check",
        ValidationType.TYPE,
        lambda x: isinstance(x, int),
        "Age must be an integer",
        ValidationSeverity.ERROR
    )
    
    # Range validation
    user_validator.add_rule(
        "age",
        "range_check",
        ValidationType.RANGE,
        lambda x: 0 <= x  0,
        "Price must be positive",
        ValidationSeverity.ERROR
    )
    
    product_validator.add_rule(
        "price",
        "price_reasonable",
        ValidationType.BUSINESS,
        lambda x: x = 0,
        "Quantity cannot be negative",
        ValidationSeverity.ERROR
    )
    
    product_validator.add_rule(
        "sku",
        "sku_format",
        ValidationType.FORMAT,
        lambda x: isinstance(x, str) and len(x) >= 3,
        "SKU must be at least 3 characters",
        ValidationSeverity.ERROR
    )
    
    print("\nValidation 3: Valid product")
    result3 = pattern.validate("ProductValidator", {
        "price": 29.99,
        "quantity": 100,
        "sku": "PROD-001"
    })
    
    print(f"Valid: {result3['valid']}")
    print(f"Summary: {result3['summary']}")
    
    print("\n" + "=" * 50)
    print("\nValidation 4: Product with warnings")
    result4 = pattern.validate("ProductValidator", {
        "price": 999999.99,  # High price (warning)
        "quantity": 50,
        "sku": "LUXURY-001"
    })
    
    print(f"Valid: {result4['valid']}")
    print(f"Summary: {result4['summary']}")
    if result4['warnings']:
        print(f"\nWarnings:")
        for warning in result4['warnings']:
            print(f"  - {warning}")
    
    print("\n" + "=" * 50)
    print("\nScenario 3: Transaction Validation")
    print("-" * 50)
    
    transaction_validator = pattern.create_validator("TransactionValidator")
    
    # Transaction rules
    transaction_validator.add_rule(
        "amount",
        "amount_positive",
        ValidationType.RANGE,
        lambda x: x > 0,
        "Transaction amount must be positive",
        ValidationSeverity.ERROR
    )
    
    transaction_validator.add_rule(
        "currency",
        "currency_valid",
        ValidationType.FORMAT,
        lambda x: x in ["USD", "EUR", "GBP", "JPY"],
        "Invalid currency code",
        ValidationSeverity.ERROR
    )
    
    transaction_validator.add_rule(
        "account_id",
        "account_format",
        ValidationType.SCHEMA,
        lambda x: isinstance(x, str) and len(x) == 10,
        "Account ID must be 10 characters",
        ValidationSeverity.ERROR
    )
    
    print("\nValidation 5: Valid transaction")
    result5 = pattern.validate("TransactionValidator", {
        "amount": 100.50,
        "currency": "USD",
        "account_id": "ACC1234567"
    })
    
    print(f"Valid: {result5['valid']}")
    print(f"Summary: {result5['summary']}")
    
    print("\n" + "=" * 50)
    print("\nStatistics")
    print("-" * 50)
    
    overall_stats = pattern.get_statistics()
    print(f"\nOverall Statistics:")
    print(f"  Total Validators: {overall_stats['total_validators']}")
    print(f"  Total Validations: {overall_stats['total_validations']}")
    print(f"  Total Passed: {overall_stats['total_passed']}")
    print(f"  Total Failed: {overall_stats['total_failed']}")
    
    print(f"\nUser Validator Statistics:")
    user_stats = user_validator.get_statistics()
    for key, value in user_stats.items():
        print(f"  {key}: {value}")
    
    print(f"\nProduct Validator Statistics:")
    product_stats = product_validator.get_statistics()
    for key, value in product_stats.items():
        print(f"  {key}: {value}")
    
    print("\n" + "=" * 50)
    print("Validation pattern demonstration complete!")
    print("=" * 50)

    

Agentic Architecture Pattern: Validation

Pattern 269: Validation Pattern – Line-by-Line Explanation

Overview

This pattern implements comprehensive data validation with rule checking, constraint enforcement, and LLM-powered fix suggestions. It supports multiple validation types, severity classification, and intelligent error analysis.

Code Explanation

Lines 1-15: Imports

"""
Pattern 269: Validation Pattern with LangChain/LangGraph

Purpose: Data validation with rule checking and constraint enforcement.
Provides intelligent validation orchestration with LLM-powered rule analysis.
"""

Explanation: Module docstring – pattern for intelligent data validation.

from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json

Explanation: Standard imports for validation with LLM integration.

Lines 17-33: Enum Definitions

class ValidationType(Enum):
    """Types of validation."""
    TYPE = "type"
    RANGE = "range"
    FORMAT = "format"
    SCHEMA = "schema"
    BUSINESS = "business"
    CUSTOM = "custom"

Explanation: Six validation types: – TYPE: Check data type (int, str, etc.) – RANGE: Check value within bounds (age between 0-120) – FORMAT: Check pattern (email, phone, regex) – SCHEMA: Check structure (required fields, nested objects) – BUSINESS: Check business rules (order total > line items sum) – CUSTOM: Custom validation functions

class ValidationSeverity(Enum):
    """Severity of validation issues."""
    ERROR = "error"      # Blocks processing
    WARNING = "warning"  # Should be addressed
    INFO = "info"        # Informational only

Explanation: Three severity levels for validation issues.

Lines 35-61: Dataclasses

@dataclass
class ValidationRule:
    """Validation rule definition."""
    name: str
    validation_type: ValidationType
    validator: Callable
    error_message: str
    severity: ValidationSeverity = ValidationSeverity.ERROR
    field: Optional[str] = None

Explanation: Defines a validation rule: – name: Human-readable identifier – validation_type: Category of validation – validator: Callable that returns True/False – error_message: Message when validation fails – severity: How serious the failure is – field: Optional specific field this rule applies to

Example: ValidationRule("age_range", ValidationType.RANGE, lambda x: 0 <= x <= 120, "Age must be 0-120")

@dataclass
class ValidationResult:
    """Result of validation."""
    field: str
    rule: str
    passed: bool
    error_message: Optional[str] = None
    severity: ValidationSeverity = ValidationSeverity.ERROR
    timestamp: datetime = field(default_factory=datetime.now)

Explanation: Result of applying one rule to one field. Tracks success/failure and error details.

@dataclass
class ValidationInsights:
    """LLM-generated validation insights."""
    overall_assessment: str
    critical_issues: List[str]
    recommendations: List[str]
    fix_suggestions: List[str]

Explanation: AI-generated analysis of validation failures with actionable fixes.

Lines 63-74: TypedDict State

class ValidationState(TypedDict):
    """State for validation workflow."""
    data: Dict[str, Any]
    rules: List[ValidationRule]
    field_results: Dict[str, List[ValidationResult]]
    all_passed: bool
    errors: List[str]
    warnings: List[str]
    insights: Optional[ValidationInsights]
    validation_summary: Dict[str, Any]

Explanation: Workflow state tracking data, rules, results per field, and AI insights.

Lines 76-117: IntelligentValidator Initialization

class IntelligentValidator:
    """
    Intelligent Validator with LangGraph workflow integration.

    Features:
    - Multiple validation types
    - LLM-powered rule analysis
    - Intelligent fix suggestions
    - Severity classification
    - Batch validation
    - Custom validators
    - Pattern detection
    - Compliance checking
    """

Explanation: Main validator class with AI-powered analysis.

def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.5):
    self.name = name
    self.llm = ChatOpenAI(model=model_name, temperature=temperature)

    # Rules
    self.rules: Dict[str, List[ValidationRule]] = {}

    # History
    self.validation_history: List[ValidationResult] = []

    # Build workflow
    self.workflow = self._build_workflow()

    # Statistics
    self.stats = {
        "total_validations": 0,
        "passed": 0,
        "failed": 0,
        "validation_types": {t.value: 0 for t in ValidationType}
    }

Explanation: Initialize with: – Rules organized by field – Validation history for trend analysis – Workflow for processing – Statistics tracking

Lines 119-137: Workflow Building

def _build_workflow(self) -> StateGraph:
    """Build LangGraph workflow."""
    workflow = StateGraph(ValidationState)

    workflow.add_node("parse_rules", self._parse_rules_node)
    workflow.add_node("validate_fields", self._validate_fields_node)
    workflow.add_node("aggregate_results", self._aggregate_results_node)
    workflow.add_node("generate_insights", self._generate_insights_node)
    workflow.add_node("recommend_fixes", self._recommend_fixes_node)

Explanation: Create 5-node workflow: 1. Parse rules: Organize rules by field 2. Validate fields: Apply all rules to data 3. Aggregate results: Collect errors/warnings 4. Generate insights: LLM analyzes failures 5. Recommend fixes: LLM suggests corrections

    workflow.set_entry_point("parse_rules")
    workflow.add_edge("parse_rules", "validate_fields")
    workflow.add_edge("validate_fields", "aggregate_results")
    workflow.add_edge("aggregate_results", "generate_insights")
    workflow.add_edge("generate_insights", "recommend_fixes")
    workflow.add_edge("recommend_fixes", END)

    return workflow.compile()

Explanation: Linear flow through all nodes.

Lines 139-150: Parse Rules Node

def _parse_rules_node(self, state: ValidationState) -> ValidationState:
    """Parse and organize validation rules."""
    # Organize rules by field
    all_rules = []
    for field, rules in self.rules.items():
        all_rules.extend(rules)

    state["rules"] = all_rules
    state["field_results"] = {}

    return state

Explanation: Flatten rules from field-organized dictionary into list, initialize results dictionary.

Lines 152-213: Validate Fields Node

def _validate_fields_node(self, state: ValidationState) -> ValidationState:
    """Validate all fields."""
    data = state["data"]
    field_results = {}

    # Validate each field
    for field, value in data.items():
        if field not in self.rules:
            continue

        field_results[field] = []

Explanation: Extract data and initialize results per field. Skip fields without rules.

        for rule in self.rules[field]:
            self.stats["total_validations"] += 1
            self.stats["validation_types"][rule.validation_type.value] += 1

            try:
                passed = rule.validator(value)

                result = ValidationResult(
                    field=field,
                    rule=rule.name,
                    passed=passed,
                    error_message=None if passed else rule.error_message,
                    severity=rule.severity
                )

                if passed:
                    self.stats["passed"] += 1
                else:
                    self.stats["failed"] += 1

                field_results[field].append(result)
                self.validation_history.append(result)

Explanation: For each rule on field: 1. Call validator function with value 2. Create ValidationResult 3. Update statistics 4. Record in field results and history

            except Exception as e:
                result = ValidationResult(
                    field=field,
                    rule=rule.name,
                    passed=False,
                    error_message=f"Validation error: {str(e)}",
                    severity=ValidationSeverity.ERROR
                )
                field_results[field].append(result)
                self.validation_history.append(result)
                self.stats["failed"] += 1

    state["field_results"] = field_results
    return state

Explanation: Handle exceptions in validator – treat as validation failure. Store results in state.

Lines 215-247: Aggregate Results Node

def _aggregate_results_node(self, state: ValidationState) -> ValidationState:
    """Aggregate validation results."""
    field_results = state["field_results"]

    errors = []
    warnings = []

Explanation: Initialize error and warning lists.

    # Collect errors and warnings
    for field, results in field_results.items():
        for result in results:
            if not result.passed:
                message = f"{field}: {result.error_message}"

                if result.severity == ValidationSeverity.ERROR:
                    errors.append(message)
                elif result.severity == ValidationSeverity.WARNING:
                    warnings.append(message)

Explanation: Iterate through all results, categorize failures by severity.

    # Check if all passed
    all_passed = len(errors) == 0

    state["errors"] = errors
    state["warnings"] = warnings
    state["all_passed"] = all_passed

    state["validation_summary"] = {
        "total_fields": len(field_results),
        "errors": len(errors),
        "warnings": len(warnings),
        "passed": all_passed
    }

    return state

Explanation: Determine overall pass/fail (errors block processing), create summary.

Lines 249-325: Generate Insights Node (LLM-Powered)

def _generate_insights_node(self, state: ValidationState) -> ValidationState:
    """Generate LLM-powered validation insights."""
    errors = state["errors"]
    warnings = state["warnings"]

    if not errors and not warnings:
        state["insights"] = ValidationInsights(
            overall_assessment="All validations passed",
            critical_issues=[],
            recommendations=[],
            fix_suggestions=[]
        )
        return state

Explanation: If no issues, skip LLM analysis.

    # Prepare data for LLM
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an expert in data validation and quality assurance."),
        ("user", """Analyze these validation failures:

Validation: {name}
Errors: {errors}
Warnings: {warnings}

Data Summary:
{data_summary}

Provide analysis as JSON with:
- overall_assessment: Brief assessment of data quality
- critical_issues: List of most critical problems
- recommendations: List of recommendations to fix issues
- fix_suggestions: Specific actionable fixes for each error

Focus on root causes and patterns, not just individual errors.""")
    ])

Explanation: Ask LLM to analyze validation failures, identify patterns, suggest fixes.

    chain = prompt | self.llm
    response = chain.invoke({
        "name": self.name,
        "errors": "\n".join(f"- {e}" for e in errors),
        "warnings": "\n".join(f"- {w}" for w in warnings),
        "data_summary": str(state["validation_summary"])
    })

Explanation: Invoke LLM with error list and summary.

    try:
        insights_data = json.loads(response.content)

        state["insights"] = ValidationInsights(
            overall_assessment=insights_data.get("overall_assessment", "Analysis completed"),
            critical_issues=insights_data.get("critical_issues", []),
            recommendations=insights_data.get("recommendations", []),
            fix_suggestions=insights_data.get("fix_suggestions", [])
        )

    except Exception as e:
        state["insights"] = ValidationInsights(
            overall_assessment=f"Validation completed with {len(errors)} errors and {len(warnings)} warnings",
            critical_issues=errors[:3],
            recommendations=["Review validation failures and correct data"],
            fix_suggestions=[]
        )

Explanation: Parse JSON response into insights object. Fallback if parsing fails.

    return state

Lines 327-345: Recommend Fixes Node

def _recommend_fixes_node(self, state: ValidationState) -> ValidationState:
    """Recommend specific fixes."""
    insights = state["insights"]

    if not insights or not insights.fix_suggestions:
        return state

    # Fixes already generated in insights
    # This node could be extended to apply automatic fixes

    return state

Explanation: Placeholder for automatic fix application. Currently fixes are just suggestions from LLM.

Lines 347-409: Public Methods

def add_rule(self, field: str, rule: ValidationRule):
    """Add validation rule for field."""
    if field not in self.rules:
        self.rules[field] = []
    self.rules[field].append(rule)

Explanation: Register validation rule for specific field.

def add_type_rule(self, field: str, expected_type: type, error_message: str = None):
    """Add type validation rule."""
    rule = ValidationRule(
        name=f"{field}_type",
        validation_type=ValidationType.TYPE,
        validator=lambda x: isinstance(x, expected_type),
        error_message=error_message or f"{field} must be {expected_type.__name__}",
        severity=ValidationSeverity.ERROR
    )
    self.add_rule(field, rule)

Explanation: Convenience method for type checking. Example: add_type_rule("age", int).

def add_range_rule(
    self,
    field: str,
    min_value: Optional[Any] = None,
    max_value: Optional[Any] = None,
    error_message: str = None
):
    """Add range validation rule."""
    def range_validator(x):
        if min_value is not None and x < min_value:
            return False
        if max_value is not None and x > max_value:
            return False
        return True

    rule = ValidationRule(
        name=f"{field}_range",
        validation_type=ValidationType.RANGE,
        validator=range_validator,
        error_message=error_message or f"{field} must be between {min_value} and {max_value}",
        severity=ValidationSeverity.ERROR
    )
    self.add_rule(field, rule)

Explanation: Convenience method for range checking. Example: add_range_rule("age", 0, 120).

def validate(self, data: Dict[str, Any]) -> Dict[str, Any]:
    """Validate data through workflow."""
    initial_state: ValidationState = {
        "data": data,
        "rules": [],
        "field_results": {},
        "all_passed": False,
        "errors": [],
        "warnings": [],
        "insights": None,
        "validation_summary": {}
    }

    final_state = self.workflow.invoke(initial_state)

    return {
        "passed": final_state["all_passed"],
        "errors": final_state["errors"],
        "warnings": final_state["warnings"],
        "insights": final_state["insights"],
        "summary": final_state["validation_summary"],
        "field_results": final_state["field_results"]
    }

Explanation: Main entry point – execute workflow and return results.

Lines 434-515: Demo Code

Scenario 1: User Registration Validation

validator = pattern.create_validator("UserRegistration")

validator.add_type_rule("username", str)
validator.add_type_rule("age", int)
validator.add_type_rule("email", str)

validator.add_range_rule("age", min_value=18, max_value=100)

def email_format_validator(email: str) -> bool:
    import re
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return bool(re.match(pattern, email))

validator.add_format_rule("email", email_format_validator, "Invalid email format")

Explanation: Register multiple validation types: – Type checking (username, age, email) – Range checking (age 18-100) – Format checking (email regex)

Scenario 2: Valid Data

valid_data = {
    "username": "john_doe",
    "age": 25,
    "email": "john@example.com"
}

result1 = pattern.validate("UserRegistration", valid_data)

Explanation: All validations pass, no errors or warnings.

Scenario 3: Invalid Data

invalid_data = {
    "username": "jane_doe",
    "age": 150,  # Invalid: too old
    "email": "invalid-email"  # Invalid: bad format
}

result2 = pattern.validate("UserRegistration", invalid_data)

Explanation: Multiple validation failures trigger LLM analysis and fix suggestions.

Scenario 4: Business Rule Validation

order_validator = pattern.create_validator("OrderValidation")

def order_total_matches(data: Dict) -> bool:
    """Business rule: total must match sum of line items."""
    items = data.get("items", [])
    stated_total = data.get("total", 0)
    calculated_total = sum(item.get("price", 0) * item.get("quantity", 0) for item in items)
    return abs(stated_total - calculated_total) < 0.01

order_validator.add_business_rule("order_total", order_total_matches, "Total must match line items")

order_data = {
    "total": 100.0,
    "items": [
        {"price": 10.0, "quantity": 5},  # 50.0
        {"price": 25.0, "quantity": 2}   # 50.0
    ]
}

result3 = pattern.validate("OrderValidation", order_data)

Explanation: Business rule validation ensures data consistency across fields.

Validation Type Comparison

Key Design Patterns

1. Chain of Responsibility: Rules applied sequentially
2. Strategy Pattern: Different validation types use different strategies
3. Template Method: Common workflow with type-specific validators
4. Observer Pattern: Validation history tracks all validations
5. Specification Pattern: Rules specify validation criteria

Performance Considerations

• Rule Application: O(fields × rules_per_field) – linear in rules
• Type Checking: O(1) – instant with isinstance
• Range Checking: O(1) – simple comparisons
• Format (Regex): O(n) – where n is string length
• Business Rules: O(custom) – depends on complexity
• LLM Analysis: O(LLM) – ~1-2 seconds for insights

Best Practices: – Order rules by speed (type → range → format → business) – Fail fast for ERROR severity – Use WARNING for non-blocking issues – Cache validation results for immutable data – Batch validations when possible

Extension Points

1. Add schema validation (JSON Schema, XML Schema)
2. Implement async validators for external checks (API calls)
3. Add validation caching for expensive checks
4. Implement automatic fix application
5. Add cross-field validation (password confirmation)
6. Support conditional validation (if field X, then validate Y)
7. Add i18n support for error messages