"""Human Validation Agent Module This module provides functionality for human review and validation of leads that require manual intervention. It integrates with: - Lead queue system (pulls leads marked with 'needs_human_review') - Lead scoring agent (scores each lead profile at action start) - Downstream agents (receives enrichment and scoring metadata) Key Features: - Pull leads with 'needs_human_review' flag from queue - Score each lead profile using lead_scoring_agent - Batch review interface with multiple resolution options - Pass scoring metadata and enrichment to downstream handlers - 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 try: from lead_scoring_agent import LeadScoringAgent, ScoreResult except ImportError: raise ImportError( "lead_scoring_agent module not found. " "Ensure it is available in the same namespace." ) 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) # Scoring metadata added during review action scoring_result: Optional[Dict] = Field( None, description="Lead scoring result from lead_scoring_agent" ) enrichment_data: Optional[Dict] = Field( None, description="Enriched profile data from lead_scoring_agent" ) @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 HumanValidationAgent: """Agent for handling human validation and review of leads with scoring integration.""" def __init__(self): """Initialize the validation agent with scoring capability.""" self.review_history = [] self.validation_errors = [] self.scoring_agent = LeadScoringAgent() def score_lead_profile(self, lead_id: str, lead_data: Dict) -> Dict: """ Score a lead profile using the lead_scoring_agent. Args: lead_id: The lead identifier lead_data: Dictionary containing lead profile information Returns: Dictionary with score_result and enrichment_data """ try: # Call the lead scoring agent to score the profile score_result = self.scoring_agent.score_lead(lead_data) enrichment = self.scoring_agent.get_enrichment_data(lead_id) return { 'score_result': score_result, 'enrichment_data': enrichment, 'scored_at': datetime.now().isoformat() } except Exception as e: self.validation_errors.append({ 'lead_id': lead_id, 'error': f"Scoring failed: {str(e)}", 'timestamp': datetime.now().isoformat() }) return {'error': str(e)} def process_review_action(self, review: LeadReview) -> Dict: """ Process a single lead review action with scoring. Args: review: LeadReview instance with review details Returns: Dictionary with processing result and scoring metadata """ try: # Score the lead profile at action start scoring_info = self.score_lead_profile( review.lead_id, review.edited_data or {} ) # Add scoring metadata to review review.scoring_result = scoring_info.get('score_result') review.enrichment_data = scoring_info.get('enrichment_data') # Store review with scoring data self.review_history.append({ 'lead_id': review.lead_id, 'action': review.action, 'reviewer_id': review.reviewer_id, 'reviewed_at': review.reviewed_at.isoformat(), 'scoring_metadata': { 'score_result': review.scoring_result, 'enrichment_data': review.enrichment_data }, 'timestamp': datetime.now().isoformat() }) return { 'status': 'success', 'lead_id': review.lead_id, 'action': review.action, 'scoring_metadata': { 'score_result': review.scoring_result, 'enrichment_data': review.enrichment_data } } except Exception as e: error_entry = { 'lead_id': review.lead_id, 'error': str(e), 'timestamp': datetime.now().isoformat() } self.validation_errors.append(error_entry) return {'status': 'error', 'error': str(e)} def process_batch_review(self, batch_request: BatchReviewRequest) -> Dict: """ Process multiple lead reviews in batch with scoring for each. Args: batch_request: BatchReviewRequest with list of reviews Returns: Dictionary with batch processing results including scoring data """ results = [] for review in batch_request.reviews: try: result = self.process_review_action(review) results.append(result) except ValidationError as e: self.validation_errors.append({ 'batch_id': batch_request.batch_id, 'error': str(e), 'timestamp': datetime.now().isoformat() }) results.append({ 'status': 'validation_error', 'error': str(e) }) return { 'batch_id': batch_request.batch_id, 'processed_count': len(results), 'results': results } def pull_leads_for_review(self, limit: int = 50) -> List[Dict]: """ Pull leads marked for human review from queue. Args: limit: Maximum number of leads to pull Returns: List of leads with 'needs_human_review' flag """ # Integration point with lead queue system return [] def get_review_stats(self) -> Dict: """ Get statistics on review activities and scoring. Returns: Dictionary with review statistics and error counts """ return { 'total_reviews': len(self.review_history), 'total_errors': len(self.validation_errors), 'recent_reviews': self.review_history[-10:] if self.review_history else [] } def export_review_history(self, filepath: str = "review_history.json") -> None: """ Export review history with scoring metadata to JSON file. Args: filepath: Path where to save the history file """ 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 with scoring. Args: agent: HumanValidationAgent instance Returns: Dictionary with interface functions that include scoring integration """ 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 with scoring 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}")