Spaces:
Sleeping
Sleeping
| """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" | |
| ) | |
| 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 | |
| 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}") |