Document Understanding Platform

1class LLMService:
2    """Multi-provider LLM extraction with graceful fallback."""
3    
4    def __init__(self, db: Session = None):
5        # Initialize clients based on available API keys
6        self.anthropic_client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
7        self.openai_client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
8        self.azure_openai_service = AzureOpenAIService()
9        
10        self.default_provider = os.getenv("DEFAULT_LLM_PROVIDER", "anthropic")
11        self.model_version = f"{self.default_provider}_{self.default_model}_v1.0"
12    
13    def extract_fields(self, ocr_text: str, provider: str = None) -> Dict[str, Any]:
14        """Extract fields with configurable field definitions from database."""
15        
16        # Get field definitions (required vs optional)
17        required_fields = self.field_service.get_required_fields()
18        optional_fields = self.field_service.get_optional_fields()
19        
20        # Route to appropriate provider
21        if provider == "azure_openai":
22            return self.azure_openai_service.extract_fields(ocr_text, all_fields)
23        elif provider == "anthropic":
24            result = self._extract_with_anthropic(prompt, model)
25        elif provider == "openai":
26            result = self._extract_with_openai(prompt, model)
27        
28        # Calculate confidence scores per field
29        confidence_scores = self._calculate_confidence_scores(extracted_data)
30        
31        return {
32            'extracted_fields': extracted_data,
33            'confidence_scores': confidence_scores,
34            'requires_review': self._requires_review(extracted_data, confidence_scores),
35            'model_version': self.model_version  # For RL tracking
36        }

1class DocumentQualityService:
2    """Multi-metric document quality assessment."""
3    
4    def assess_document_quality(self, file_path: str) -> Dict[str, Any]:
5        # Image quality metrics
6        quality_metrics = self._assess_image_quality(image_path, image)
7        
8        # Text quality via OCR confidence
9        text_metrics = self._assess_text_quality(image_path)
10        
11        # Weighted overall score
12        overall_score = self._calculate_overall_score(quality_metrics, text_metrics)
13        
14        return {
15            "image_dpi": quality_metrics.get("dpi", 0),
16            "image_clarity_score": quality_metrics.get("clarity_score", 0.0),
17            "text_density_score": text_metrics.get("text_density", 0.0),
18            "overall_quality_score": overall_score,
19            "quality_issues": self._identify_issues(quality_metrics, text_metrics),
20            "recommendations": self._generate_recommendations(...)
21        }
22    
23    def _assess_image_quality(self, image_path: str, pil_image: Image) -> Dict:
24        # Laplacian variance for clarity
25        clarity_score = cv2.Laplacian(gray, cv2.CV_64F).var()
26        
27        # Contrast and brightness analysis
28        contrast = gray.std()
29        brightness_score = 1.0 - abs(gray.mean() - 128) / 128.0
30        
31        # Noise estimation via median filter comparison
32        noise_level = self._estimate_noise_level(gray)
33        
34        return {"clarity_score": clarity_score, "contrast": contrast, ...}

1class WorkflowService:
2    """Business rule validation and intelligent assignment."""
3    
4    def validate_business_rules(self, document_id: int) -> Dict[str, Any]:
5        """Validate against all active business rules."""
6        
7        active_rules = self.db.query(BusinessRule).filter(
8            BusinessRule.is_active == True
9        ).all()
10        
11        violations = []
12        for rule in active_rules:
13            violation = self._validate_single_rule(document, rule)
14            if violation:
15                # Store in database for audit trail
16                rule_violation = BusinessRuleViolation(
17                    document_id=document_id,
18                    rule_id=rule.id,
19                    violation_details=violation,
20                    severity=rule.severity
21                )
22                self.db.add(rule_violation)
23                
24                if rule.severity == "error":
25                    violations.append(violation)
26        
27        return {"has_violations": len(violations) > 0, "violations": violations}
28    
29    def _validate_cross_field_rule(self, document, rule, rule_def) -> Optional[Dict]:
30        """Cross-field validation (e.g., denial + auth number conflict)."""
31        
32        if rule_def.get("logic") == "denial_no_auth_number":
33            denial_reason = document.extracted_fields.get("denial_reason")
34            auth_number = document.extracted_fields.get("authorization_number")
35            
36            if denial_reason and auth_number:
37                return {
38                    "rule_name": rule.name,
39                    "issue": "Denied documents should not have authorization numbers",
40                    "severity": rule.severity
41                }

1class ReinforcementLearningService:
2    """Human feedback capture for continuous model improvement."""
3    
4    def record_human_feedback(
5        self,
6        document_id: int,
7        field_name: str,
8        original_value: str,
9        corrected_value: str,
10        feedback_type: str,  # correction, addition, removal, confirmation
11        model_version: str
12    ) -> HumanFeedback:
13        
14        # Calculate reward based on feedback type
15        reward_score = self._calculate_reward(
16            feedback_type, original_value, corrected_value
17        )
18        
19        feedback = HumanFeedback(
20            document_id=document_id,
21            field_name=field_name,
22            original_value=original_value,
23            corrected_value=corrected_value,
24            feedback_type=feedback_type,
25            reward_score=reward_score,
26            model_version=model_version,
27            ocr_context=ocr_context  # For training context
28        )
29        
30        # Update model performance metrics
31        self._update_model_performance(model_version, field_name, feedback_type)
32        
33        return feedback
34    
35    def _calculate_reward(self, feedback_type: str, original: str, corrected: str) -> float:
36        """Reward shaping for RL training data."""
37        if feedback_type == "confirmation":
38            return 1.0   # Model was correct
39        elif feedback_type == "correction":
40            return -0.5  # Partial penalty (had something, was wrong)
41        elif feedback_type == "addition":
42            return -0.8  # Missed field entirely
43        elif feedback_type == "removal":
44            return -0.3  # Hallucinated field
45        return 0.0