level-1 of generator and validator for notes

SYSTEM_ARCHITECTURE_BOXES.md

@@ -2421,3 +2421,142 @@ FILE DOWNLOAD
 │xlsx                 │
 └─────────────────────┘
 ```
+
+
+
+
+
+
+**generator-validator**
+┌─────────────────┐
+│   API Request   │
+│  POST /notes-llm│
+│  with file      │
+└─────────┬───────┘
+          │
+          ▼
+┌─────────────────┐
+│ create_notes_   │
+│ pipeline()      │
+│                 │
+│ → LLMNotesGen   │
+│ → NotesValidator│
+└─────────┬───────┘
+          │
+          ▼
+┌─────────────────┐     ┌─────────────────┐
+│Generator-Validator│◄───│  Max 3 Attempts │
+│     Pipeline     │     │                 │
+│                 │     │                 │
+│ ┌─────────────┐ │     └─────────────────┘
+│ │  Attempt    │ │             │ No
+│ │  Counter=0  │ │             ▼
+│ └─────────────┘ │     ┌─────────────────┐
+└─────────┬───────┘     │   Return Best   │
+          │             │     Result      │
+          ▼             └─────────────────┘
+┌─────────────────┐
+│   Generate      │
+│   (LLM)         │
+│                 │
+│ - Call langgraph│
+│ - Use RLHF if   │
+│   requested     │
+│ - Track attempt │
+└─────────┬───────┘
+          │
+          ▼
+┌─────────────────┐
+│   Validate      │
+│   Quality       │
+│                 │
+│ - File exists   │
+│ - Size >1KB     │
+│ - Metadata OK   │
+│ - RLHF quality  │
+│ - Score 0.0-1.0 │
+└─────────┬───────┘
+          │
+    ┌─────┴─────┐
+    │           │
+    ▼           ▼
+┌─────────┐ ┌─────────┐
+│ Valid?  │ │ Invalid │
+│ Score   │ │ Score   │
+│ ≥0.6    │ │ <0.6    │
+└─────┬───┘ └─────┬───┘
+      │           │
+      ▼           ▼
+┌─────────┐ ┌─────────┐
+│ Return   │ │ Refine  │
+│ Success  │ │ & Retry │
+│ with     │ │         │
+│ Metadata │ │ - Use   │
+│ Headers  │ │   feedback│
+└─────────┘ └─────┬───┘
+                 │
+                 ▼
+            ┌─────────┐
+            │Increment │
+            │ Attempt  │
+            │ Counter  │
+            └─────┬───┘
+                 │
+                 └─────────────┐
+                               ▼
+                      ┌─────────────────┐
+                      │   Continue to   │
+                      │   Next Attempt  │
+                      └─────────────────┘
+
+
+**refine and retry**
+
+┌─────────────────┐
+│   Attempt 1     │
+│   Generate      │
+│   → Validate    │
+│   Score: 0.4    │  ❌ FAIL (< 0.6)
+└─────────┬───────┘
+          │
+          ▼
+┌─────────────────┐
+│   Refinement    │
+│   Analysis      │
+│                 │
+│ Feedback:       │
+│ - "Low quality" │
+│ - "Small file"  │
+└─────────┬───────┘
+          │
+    ┌─────┴─────┐
+    │           │
+    ▼           ▼
+┌─────────┐ ┌─────────┐
+│Quality   │ │Other     │
+│Issue?    │ │Issue?    │
+│          │ │          │
+│"quality" │ │File size │
+│in feedback│ │Metadata  │
+└─────┬───┘ └─────┬───┘
+      │           │
+      ▼           ▼
+┌─────────┐ ┌─────────┐
+│Switch to │ │Simple    │
+│RLHF      │ │Retry     │
+│Mode      │ │          │
+│          │ │Use same  │
+│use_rlhf= │ │config    │
+│true      │ │          │
+└─────┬───┘ └─────┬───┘
+      │           │
+      └─────┬─────┘
+            │
+            ▼
+┌─────────────────┐
+│   Attempt 2     │
+│   Generate      │
+│   (Improved)    │
+│   → Validate    │
+│   Score: 0.8    │  ✅ SUCCESS (≥ 0.6)
+└─────────────────┘

agents/generator_validator.py

@@ -0,0 +1,386 @@
+"""
+Generator-Validator Pattern Implementation for Financial Notes
+Implements formal Generator-Validator pattern with iterative refinement
+"""
+import os
+import json
+import logging
+from abc import ABC, abstractmethod
+from typing import Dict, Any, List, Optional, Tuple
+from dataclasses import dataclass
+from datetime import datetime
+import subprocess
+import shutil
+
+logger = logging.getLogger(__name__)
+
+@dataclass
+class ValidationResult:
+    """Result of validation process"""
+    is_valid: bool
+    score: float
+    feedback: List[str]
+    suggestions: List[str]
+    metadata: Dict[str, Any]
+
+@dataclass
+class GenerationResult:
+    """Result of generation process"""
+    success: bool
+    output_path: Optional[str]
+    data: Optional[Dict[str, Any]]
+    error: Optional[str]
+    metadata: Dict[str, Any]
+
+class BaseGenerator(ABC):
+    """Abstract base class for financial statement generators"""
+
+    def __init__(self, max_attempts: int = 3):
+        self.max_attempts = max_attempts
+        self.attempts_made = 0
+
+    @abstractmethod
+    def generate(self, file_path: str, **kwargs) -> GenerationResult:
+        """Generate financial statement from input file"""
+        pass
+
+    @abstractmethod
+    def refine(self, previous_result: GenerationResult, feedback: List[str]) -> GenerationResult:
+        """Refine generation based on validation feedback"""
+        pass
+
+class BaseValidator(ABC):
+    """Abstract base class for financial statement validators"""
+
+    @abstractmethod
+    def validate(self, generation_result: GenerationResult) -> ValidationResult:
+        """Validate the generated financial statement"""
+        pass
+
+    @abstractmethod
+    def get_validation_criteria(self) -> List[str]:
+        """Return list of validation criteria"""
+        pass
+
+class LLMNotesGenerator(BaseGenerator):
+    """Generator for AI-powered financial notes"""
+
+    def __init__(self, max_attempts: int = 3, use_rlhf: bool = False):
+        super().__init__(max_attempts)
+        self.use_rlhf = use_rlhf
+
+    def generate(self, file_path: str, **kwargs) -> GenerationResult:
+        """Generate notes using AI/LLM approach"""
+        try:
+            self.attempts_made += 1
+            execution_id = f"notes_llm_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{self.attempts_made}"
+
+            # Choose workflow based on RLHF preference
+            if self.use_rlhf:
+                from agents.rlhf_workflows import run_rlhf_workflow
+                result = run_rlhf_workflow(file_path, "notes-llm")
+            else:
+                from agents.langgraph import run_workflow
+                result = run_workflow(file_path, "notes-llm")
+
+            if result["status"] == "success":
+                return GenerationResult(
+                    success=True,
+                    output_path=result["result"]["output_xlsx_path"],
+                    data=result["result"],
+                    error=None,
+                    metadata={
+                        "execution_id": execution_id,
+                        "generation_method": "llm",
+                        "use_rlhf": self.use_rlhf,
+                        "attempt": self.attempts_made,
+                        "rlhf_metadata": result["result"].get("rlhf_metadata", {})
+                    }
+                )
+            else:
+                return GenerationResult(
+                    success=False,
+                    output_path=None,
+                    data=None,
+                    error=result.get("error", "Unknown error"),
+                    metadata={
+                        "execution_id": execution_id,
+                        "generation_method": "llm",
+                        "use_rlhf": self.use_rlhf,
+                        "attempt": self.attempts_made
+                    }
+                )
+
+        except Exception as e:
+            logger.error(f"LLM Notes generation failed: {e}")
+            return GenerationResult(
+                success=False,
+                output_path=None,
+                data=None,
+                error=str(e),
+                metadata={
+                    "execution_id": f"error_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
+                    "generation_method": "llm",
+                    "use_rlhf": self.use_rlhf,
+                    "attempt": self.attempts_made
+                }
+            )
+
+    def refine(self, previous_result: GenerationResult, feedback: List[str]) -> GenerationResult:
+        """Refine LLM notes generation based on feedback"""
+        logger.info(f"Refining LLM notes generation with feedback: {feedback}")
+
+        # For LLM generation, we can try different approaches:
+        # 1. Switch to RLHF if not already using it
+        # 2. Retry with different parameters
+        # 3. Use fallback models
+
+        if not self.use_rlhf and "quality" in str(feedback).lower():
+            # If quality issues and not using RLHF, try RLHF
+            logger.info("Switching to RLHF for better quality")
+            original_rlhf = self.use_rlhf
+            self.use_rlhf = True
+            result = self.generate(previous_result.data.get("file_path") if previous_result.data else None)
+            self.use_rlhf = original_rlhf  # Reset for future calls
+            return result
+        else:
+            # Otherwise, just retry
+            return self.generate(previous_result.data.get("file_path") if previous_result.data else None)
+    """Generator for AI-powered financial notes"""
+
+    def __init__(self, max_attempts: int = 3, use_rlhf: bool = False):
+        super().__init__(max_attempts)
+        self.use_rlhf = use_rlhf
+
+    def generate(self, file_path: str, **kwargs) -> GenerationResult:
+        """Generate notes using AI/LLM approach"""
+        try:
+            self.attempts_made += 1
+            execution_id = f"notes_llm_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{self.attempts_made}"
+
+            # Choose workflow based on RLHF preference
+            if self.use_rlhf:
+                from agents.rlhf_workflows import run_rlhf_workflow
+                result = run_rlhf_workflow(file_path, "notes-llm")
+            else:
+                from agents.langgraph import run_workflow
+                result = run_workflow(file_path, "notes-llm")
+
+            if result["status"] == "success":
+                return GenerationResult(
+                    success=True,
+                    output_path=result["result"]["output_xlsx_path"],
+                    data=result["result"],
+                    error=None,
+                    metadata={
+                        "execution_id": execution_id,
+                        "generation_method": "llm",
+                        "use_rlhf": self.use_rlhf,
+                        "attempt": self.attempts_made,
+                        "rlhf_metadata": result["result"].get("rlhf_metadata", {})
+                    }
+                )
+            else:
+                return GenerationResult(
+                    success=False,
+                    output_path=None,
+                    data=None,
+                    error=result.get("error", "Unknown error"),
+                    metadata={
+                        "execution_id": execution_id,
+                        "generation_method": "llm",
+                        "use_rlhf": self.use_rlhf,
+                        "attempt": self.attempts_made
+                    }
+                )
+
+        except Exception as e:
+            logger.error(f"LLM Notes generation failed: {e}")
+            return GenerationResult(
+                success=False,
+                output_path=None,
+                data=None,
+                error=str(e),
+                metadata={
+                    "execution_id": f"error_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
+                    "generation_method": "llm",
+                    "use_rlhf": self.use_rlhf,
+                    "attempt": self.attempts_made
+                }
+            )
+
+    def refine(self, previous_result: GenerationResult, feedback: List[str]) -> GenerationResult:
+        """Refine LLM notes generation based on feedback"""
+        logger.info(f"Refining LLM notes generation with feedback: {feedback}")
+
+        # For LLM generation, we can try different approaches:
+        # 1. Switch to RLHF if not already using it
+        # 2. Retry with different parameters
+        # 3. Use fallback models
+
+        if not self.use_rlhf and "quality" in str(feedback).lower():
+            # If quality issues and not using RLHF, try RLHF
+            logger.info("Switching to RLHF for better quality")
+            original_rlhf = self.use_rlhf
+            self.use_rlhf = True
+            result = self.generate(previous_result.data.get("file_path") if previous_result.data else None)
+            self.use_rlhf = original_rlhf  # Reset for future calls
+            return result
+        else:
+            # Otherwise, just retry
+            return self.generate(previous_result.data.get("file_path") if previous_result.data else None)
+
+class NotesValidator(BaseValidator):
+    """Validator for financial notes quality"""
+
+    def validate(self, generation_result: GenerationResult) -> ValidationResult:
+        """Validate generated financial notes"""
+        if not generation_result.success or not generation_result.output_path:
+            return ValidationResult(
+                is_valid=False,
+                score=0.0,
+                feedback=["Generation failed - no output produced"],
+                suggestions=["Retry generation process"],
+                metadata={"validation_type": "basic"}
+            )
+
+        feedback = []
+        suggestions = []
+        score = 1.0  # Start with perfect score
+
+        try:
+            # Check if output file exists
+            if not os.path.exists(generation_result.output_path):
+                feedback.append("Output file does not exist")
+                suggestions.append("Check file generation process")
+                score -= 0.5
+
+            # Check file size (reasonable minimum)
+            if os.path.exists(generation_result.output_path):
+                file_size = os.path.getsize(generation_result.output_path)
+                if file_size < 1000:  # Less than 1KB
+                    feedback.append("Output file is unusually small")
+                    suggestions.append("Verify data processing and Excel generation")
+                    score -= 0.3
+
+            # Check for common financial notes elements
+            # This would require reading the Excel file, but for now we'll do basic checks
+
+            # Check metadata completeness
+            metadata = generation_result.metadata
+            if not metadata.get("execution_id"):
+                feedback.append("Missing execution ID in metadata")
+                score -= 0.1
+
+            if generation_result.metadata.get("generation_method") == "llm":
+                # Additional checks for LLM-generated content
+                rlhf_metadata = metadata.get("rlhf_metadata", {})
+                if rlhf_metadata and rlhf_metadata.get("predicted_quality", 0) < 0.5:
+                    feedback.append("Low quality score from RLHF validation")
+                    suggestions.append("Consider regenerating with RLHF enabled")
+                    score -= 0.4
+
+            # Ensure score doesn't go below 0
+            score = max(0.0, score)
+
+            return ValidationResult(
+                is_valid=score >= 0.6,  # 60% threshold for validity
+                score=score,
+                feedback=feedback,
+                suggestions=suggestions,
+                metadata={
+                    "validation_type": "comprehensive",
+                    "file_size": os.path.getsize(generation_result.output_path) if os.path.exists(generation_result.output_path) else 0,
+                    "has_rlhf": bool(generation_result.metadata.get("rlhf_metadata"))
+                }
+            )
+
+        except Exception as e:
+            logger.error(f"Validation failed: {e}")
+            return ValidationResult(
+                is_valid=False,
+                score=0.0,
+                feedback=[f"Validation error: {str(e)}"],
+                suggestions=["Check system logs for detailed error information"],
+                metadata={"validation_type": "error"}
+            )
+
+    def get_validation_criteria(self) -> List[str]:
+        """Return list of validation criteria"""
+        return [
+            "Output file exists and is accessible",
+            "File size is reasonable (>1KB)",
+            "Metadata contains required fields",
+            "For LLM generation: quality score meets threshold",
+            "RLHF metadata present when RLHF is enabled",
+            "No critical errors in generation process"
+        ]
+
+class GeneratorValidatorPipeline:
+    """Main pipeline that orchestrates Generator-Validator pattern"""
+
+    def __init__(self, generator: BaseGenerator, validator: BaseValidator):
+        self.generator = generator
+        self.validator = validator
+        self.generation_history = []
+        self.validation_history = []
+
+    def process(self, file_path: str, **kwargs) -> Tuple[GenerationResult, ValidationResult]:
+        """Process file through generator-validator pipeline"""
+        logger.info("Starting Generator-Validator pipeline")
+
+        best_result = None
+        best_validation = None
+
+        for attempt in range(self.generator.max_attempts):
+            logger.info(f"Attempt {attempt + 1}/{self.generator.max_attempts}")
+
+            # Generate
+            generation_result = self.generator.generate(file_path, **kwargs)
+            self.generation_history.append(generation_result)
+
+            # Validate
+            validation_result = self.validator.validate(generation_result)
+            self.validation_history.append(validation_result)
+
+            logger.info(f"Generation success: {generation_result.success}, Validation score: {validation_result.score}")
+
+            # Keep track of best result
+            if best_result is None or (generation_result.success and validation_result.score > (best_validation.score if best_validation else 0)):
+                best_result = generation_result
+                best_validation = validation_result
+
+            # If validation passes, return immediately
+            if validation_result.is_valid:
+                logger.info("Validation passed - returning result")
+                return generation_result, validation_result
+
+            # If not the last attempt, try to refine
+            if attempt < self.generator.max_attempts - 1:
+                logger.info(f"Validation failed - refining with feedback: {validation_result.feedback}")
+                # Reset attempts counter for refinement
+                original_attempts = self.generator.attempts_made
+                self.generator.attempts_made = 0  # Reset for refinement
+                generation_result = self.generator.refine(generation_result, validation_result.feedback)
+                self.generator.attempts_made = original_attempts
+
+        logger.info("All attempts completed - returning best result")
+        return best_result, best_validation
+
+    def get_processing_summary(self) -> Dict[str, Any]:
+        """Get summary of the processing pipeline"""
+        return {
+            "total_attempts": len(self.generation_history),
+            "successful_generations": sum(1 for g in self.generation_history if g.success),
+            "validation_scores": [v.score for v in self.validation_history],
+            "best_score": max([v.score for v in self.validation_history]) if self.validation_history else 0,
+            "generation_methods": list(set([g.metadata.get("generation_method") for g in self.generation_history if g.metadata])),
+            "validation_criteria": self.validator.get_validation_criteria()
+        }
+
+def create_notes_pipeline(use_rlhf: bool = False) -> GeneratorValidatorPipeline:
+    """Factory function to create LLM-based pipeline for notes generation"""
+    generator = LLMNotesGenerator(use_rlhf=use_rlhf)
+    validator = NotesValidator()
+
+    return GeneratorValidatorPipeline(generator, validator)

app.py

@@ -5,6 +5,7 @@ import os
 import shutil
 import logging
 import json
+from agents.generator_validator import create_notes_pipeline
 from agents.langgraph import run_workflow
 from agents.rlhf_workflows import run_rlhf_workflow
 from agents.rlhf_routes import rlhf_router
@@ -37,29 +38,60 @@ router = APIRouter()
 
 @router.post("/notes-llm")
 async def notes_llm_route(file: UploadFile = File(...), use_rlhf: bool = Query(False)):
+    """Generate AI-powered financial notes using Generator-Validator pattern"""
     file_path = f"data/input/{file.filename}"
     os.makedirs("data/input", exist_ok=True)
     with open(file_path, "wb") as buffer:
         shutil.copyfileobj(file.file, buffer)
-    
-    # Choose workflow based on RLHF preference - using "notes-llm" as statement type
-    if use_rlhf:
-        result = run_rlhf_workflow(file_path, "notes-llm")
-    else:
-        result = run_workflow(file_path, "notes-llm")
-    
-    if result["status"] == "success":
-        response = FileResponse(result["result"]["output_xlsx_path"], filename=os.path.basename(result["result"]["output_xlsx_path"]))
-        
-        # Add RLHF metadata to headers if available
-        if "rlhf_metadata" in result.get("result", {}):
-            rlhf_data = result["result"]["rlhf_metadata"]
-            response.headers["X-RLHF-Statement-ID"] = str(rlhf_data.get("statement_id", ""))
-            response.headers["X-RLHF-Quality-Score"] = str(rlhf_data.get("predicted_quality", ""))
-            response.headers["X-RLHF-Confidence"] = str(rlhf_data.get("confidence_score", ""))
-        
-        return response
-    raise HTTPException(status_code=500, detail=result["error"])
+
+    try:
+        # Create Generator-Validator pipeline for LLM notes
+        pipeline = create_notes_pipeline(use_rlhf=use_rlhf)
+
+        # Process through pipeline
+        generation_result, validation_result = pipeline.process(file_path)
+
+        # Log processing summary
+        summary = pipeline.get_processing_summary()
+        logger.info(f"LLM Notes Pipeline Summary: {summary}")
+
+        if generation_result.success and validation_result.is_valid:
+            response = FileResponse(
+                generation_result.output_path,
+                filename=os.path.basename(generation_result.output_path)
+            )
+
+            # Add comprehensive metadata to headers
+            response.headers["X-Generation-Method"] = "llm"
+            response.headers["X-Validation-Score"] = str(validation_result.score)
+            response.headers["X-Attempts-Made"] = str(generation_result.metadata.get("attempt", 1))
+            response.headers["X-Execution-ID"] = generation_result.metadata.get("execution_id", "")
+
+            # Add RLHF metadata if available
+            if use_rlhf and "rlhf_metadata" in generation_result.metadata:
+                rlhf_data = generation_result.metadata["rlhf_metadata"]
+                response.headers["X-RLHF-Statement-ID"] = str(rlhf_data.get("statement_id", ""))
+                response.headers["X-RLHF-Quality-Score"] = str(rlhf_data.get("predicted_quality", ""))
+                response.headers["X-RLHF-Confidence"] = str(rlhf_data.get("confidence_score", ""))
+
+            # Add validation feedback
+            if validation_result.feedback:
+                response.headers["X-Validation-Feedback"] = json.dumps(validation_result.feedback)
+
+            return response
+        else:
+            error_detail = {
+                "generation_error": generation_result.error,
+                "validation_feedback": validation_result.feedback,
+                "validation_score": validation_result.score,
+                "attempts_made": generation_result.metadata.get("attempt", 1),
+                "processing_summary": summary
+            }
+            raise HTTPException(status_code=500, detail=json.dumps(error_detail))
+
+    except Exception as e:
+        logger.error(f"LLM Notes pipeline failed: {e}")
+        raise HTTPException(status_code=500, detail=f"Pipeline processing failed: {str(e)}")
 
 
 @router.post("/notes")