"""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}")