Create Human Validation Agent module with batch review, schema validation, and queue integration

human_validation_agent.py

@@ -0,0 +1,375 @@
+"""Human Validation Agent Module
+
+This module provides functionality for human review and validation of leads
+that require manual intervention. It integrates with the lead queue system
+to pull leads marked with 'needs_human_review', presents batch review options,
+and handles resolution workflows.
+
+Key Features:
+- Pull leads with 'needs_human_review' flag from queue
+- Batch review interface with multiple resolution options
+- Schema validation for actions
+- Feedback and guardrails for invalid actions
+- Handoff to downstream queue or archival
+"""
+
+import json
+from typing import List, Dict, Optional, Literal
+from datetime import datetime
+from pydantic import BaseModel, Field, ValidationError, validator
+
+
+class LeadReview(BaseModel):
+    """Schema for a single lead review action."""
+    lead_id: str = Field(..., description="Unique identifier for the lead")
+    action: Literal["approve", "edit", "hold", "scrub"] = Field(
+        ..., description="Action to take on the lead"
+    )
+    reason: Optional[str] = Field(
+        None, description="Reason for action (required for hold and scrub)"
+    )
+    edited_data: Optional[Dict] = Field(
+        None, description="Updated lead data (required for edit action)"
+    )
+    reviewer_id: str = Field(..., description="ID of the person reviewing")
+    reviewed_at: datetime = Field(default_factory=datetime.now)
+
+    @validator('reason')
+    def validate_reason(cls, v, values):
+        """Ensure reason is provided for hold and scrub actions."""
+        if values.get('action') in ['hold', 'scrub'] and not v:
+            raise ValueError(f"Reason is required for {values.get('action')} action")
+        return v
+
+    @validator('edited_data')
+    def validate_edited_data(cls, v, values):
+        """Ensure edited_data is provided for edit action."""
+        if values.get('action') == 'edit' and not v:
+            raise ValueError("edited_data is required for edit action")
+        return v
+
+
+class BatchReviewRequest(BaseModel):
+    """Schema for batch review request."""
+    reviews: List[LeadReview] = Field(..., description="List of lead reviews")
+    batch_id: Optional[str] = Field(None, description="Optional batch identifier")
+
+
+class Lead(BaseModel):
+    """Schema for a lead in the queue."""
+    lead_id: str
+    status: str
+    needs_human_review: bool = False
+    data: Dict
+    created_at: datetime
+    updated_at: datetime
+    review_reason: Optional[str] = None
+
+
+class HumanValidationAgent:
+    """Agent for handling human validation of leads."""
+
+    def __init__(self, queue_connection=None):
+        """
+        Initialize the Human Validation Agent.
+        
+        Args:
+            queue_connection: Connection to the lead queue system
+        """
+        self.queue_connection = queue_connection
+        self.review_history = []
+        self.validation_errors = []
+
+    def pull_leads_for_review(self, limit: int = 50) -> List[Lead]:
+        """
+        Pull all leads with 'needs_human_review' flag from the queue.
+        
+        Args:
+            limit: Maximum number of leads to pull
+            
+        Returns:
+            List of leads requiring human review
+        """
+        try:
+            # In production, this would query the actual queue
+            # For now, we return a simulated structure
+            leads = []
+            
+            if self.queue_connection:
+                # Query leads with needs_human_review=True
+                query = {
+                    "needs_human_review": True,
+                    "status": {"$in": ["pending_review", "flagged"]}
+                }
+                leads = self.queue_connection.find(query).limit(limit)
+            
+            return [Lead(**lead) for lead in leads]
+            
+        except Exception as e:
+            self.validation_errors.append({
+                "error": f"Failed to pull leads: {str(e)}",
+                "timestamp": datetime.now().isoformat()
+            })
+            return []
+
+    def validate_review_action(self, review: LeadReview) -> tuple[bool, Optional[str]]:
+        """
+        Validate a single review action against schema and business rules.
+        
+        Args:
+            review: LeadReview object to validate
+            
+        Returns:
+            Tuple of (is_valid, error_message)
+        """
+        try:
+            # Schema validation is handled by Pydantic
+            # Additional business logic validation can be added here
+            
+            if review.action == "edit":
+                # Validate edited data has required fields
+                required_fields = ["name", "email", "company"]
+                for field in required_fields:
+                    if field not in review.edited_data:
+                        return False, f"Missing required field in edited_data: {field}"
+            
+            if review.action == "scrub":
+                # Ensure reason is substantive
+                if len(review.reason) < 10:
+                    return False, "Scrub reason must be at least 10 characters"
+            
+            return True, None
+            
+        except ValidationError as e:
+            return False, str(e)
+        except Exception as e:
+            return False, f"Validation error: {str(e)}"
+
+    def process_review_action(self, review: LeadReview) -> Dict:
+        """
+        Process a validated review action and update lead status.
+        
+        Args:
+            review: Validated LeadReview object
+            
+        Returns:
+            Dictionary with processing results
+        """
+        result = {
+            "lead_id": review.lead_id,
+            "action": review.action,
+            "status": "processing",
+            "timestamp": datetime.now().isoformat()
+        }
+        
+        try:
+            if review.action == "approve":
+                # Move to approved queue for downstream processing
+                result["next_queue"] = "approved_leads"
+                result["status"] = "approved"
+                
+            elif review.action == "edit":
+                # Update lead data and move to approved queue
+                result["next_queue"] = "approved_leads"
+                result["status"] = "edited_and_approved"
+                result["updated_fields"] = list(review.edited_data.keys())
+                
+            elif review.action == "hold":
+                # Keep in review queue with hold status
+                result["next_queue"] = "on_hold"
+                result["status"] = "on_hold"
+                result["reason"] = review.reason
+                
+            elif review.action == "scrub":
+                # Move to scrubbed/archived queue
+                result["next_queue"] = "scrubbed_leads"
+                result["status"] = "scrubbed"
+                result["reason"] = review.reason
+            
+            # Record in history
+            self.review_history.append({
+                "lead_id": review.lead_id,
+                "action": review.action,
+                "reviewer_id": review.reviewer_id,
+                "timestamp": review.reviewed_at.isoformat()
+            })
+            
+            return result
+            
+        except Exception as e:
+            result["status"] = "error"
+            result["error"] = str(e)
+            self.validation_errors.append(result)
+            return result
+
+    def process_batch_review(self, batch_request: BatchReviewRequest) -> Dict:
+        """
+        Process a batch of review actions with validation and error handling.
+        
+        Args:
+            batch_request: BatchReviewRequest containing multiple reviews
+            
+        Returns:
+            Dictionary with batch processing results
+        """
+        results = {
+            "batch_id": batch_request.batch_id or f"batch_{datetime.now().timestamp()}",
+            "total_reviews": len(batch_request.reviews),
+            "successful": 0,
+            "failed": 0,
+            "results": [],
+            "errors": []
+        }
+        
+        for review in batch_request.reviews:
+            # Validate each review
+            is_valid, error_msg = self.validate_review_action(review)
+            
+            if not is_valid:
+                results["failed"] += 1
+                results["errors"].append({
+                    "lead_id": review.lead_id,
+                    "error": error_msg,
+                    "feedback": self._get_validation_feedback(error_msg)
+                })
+                continue
+            
+            # Process valid review
+            result = self.process_review_action(review)
+            
+            if result["status"] == "error":
+                results["failed"] += 1
+                results["errors"].append(result)
+            else:
+                results["successful"] += 1
+                results["results"].append(result)
+        
+        return results
+
+    def _get_validation_feedback(self, error_msg: str) -> str:
+        """
+        Generate helpful feedback for validation errors.
+        
+        Args:
+            error_msg: The validation error message
+            
+        Returns:
+            User-friendly feedback message
+        """
+        feedback_map = {
+            "reason is required": "Please provide a reason for this action. This helps maintain audit trail and clarity.",
+            "edited_data is required": "When editing a lead, you must provide the updated data fields.",
+            "Missing required field": "Edited lead data must include all required fields: name, email, and company.",
+            "must be at least 10 characters": "Please provide a more detailed reason to ensure proper documentation."
+        }
+        
+        for key, feedback in feedback_map.items():
+            if key.lower() in error_msg.lower():
+                return feedback
+        
+        return "Please check your input and try again. Refer to the API documentation for correct format."
+
+    def get_review_stats(self) -> Dict:
+        """
+        Get statistics on review activity.
+        
+        Returns:
+            Dictionary with review statistics
+        """
+        total_reviews = len(self.review_history)
+        
+        if total_reviews == 0:
+            return {
+                "total_reviews": 0,
+                "by_action": {},
+                "error_count": len(self.validation_errors)
+            }
+        
+        action_counts = {}
+        for review in self.review_history:
+            action = review["action"]
+            action_counts[action] = action_counts.get(action, 0) + 1
+        
+        return {
+            "total_reviews": total_reviews,
+            "by_action": action_counts,
+            "error_count": len(self.validation_errors),
+            "success_rate": (total_reviews / (total_reviews + len(self.validation_errors))) * 100
+        }
+
+    def export_review_history(self, filepath: str = "review_history.json"):
+        """
+        Export review history to a JSON file.
+        
+        Args:
+            filepath: Path to save the review history
+        """
+        with open(filepath, 'w') as f:
+            json.dump({
+                "review_history": self.review_history,
+                "validation_errors": self.validation_errors,
+                "stats": self.get_review_stats()
+            }, f, indent=2)
+
+
+# API Interface Functions
+def create_review_interface(agent: HumanValidationAgent):
+    """
+    Create a simple interface for batch review operations.
+    
+    Args:
+        agent: HumanValidationAgent instance
+        
+    Returns:
+        Dictionary with interface functions
+    """
+    return {
+        "get_pending_reviews": lambda limit=50: agent.pull_leads_for_review(limit),
+        "submit_review": lambda review_data: agent.process_review_action(
+            LeadReview(**review_data)
+        ),
+        "submit_batch_review": lambda batch_data: agent.process_batch_review(
+            BatchReviewRequest(**batch_data)
+        ),
+        "get_stats": lambda: agent.get_review_stats(),
+        "export_history": lambda filepath="review_history.json": agent.export_review_history(filepath)
+    }
+
+
+if __name__ == "__main__":
+    # Example usage
+    agent = HumanValidationAgent()
+    
+    # Example batch review
+    sample_batch = {
+        "reviews": [
+            {
+                "lead_id": "lead_001",
+                "action": "approve",
+                "reviewer_id": "reviewer_123"
+            },
+            {
+                "lead_id": "lead_002",
+                "action": "edit",
+                "edited_data": {
+                    "name": "Updated Name",
+                    "email": "updated@email.com",
+                    "company": "Updated Corp"
+                },
+                "reviewer_id": "reviewer_123"
+            },
+            {
+                "lead_id": "lead_003",
+                "action": "scrub",
+                "reason": "Lead is a competitor and not a valid prospect for our services",
+                "reviewer_id": "reviewer_123"
+            }
+        ]
+    }
+    
+    try:
+        batch_request = BatchReviewRequest(**sample_batch)
+        results = agent.process_batch_review(batch_request)
+        print(json.dumps(results, indent=2))
+    except ValidationError as e:
+        print(f"Validation Error: {e}")