Spaces:
Sleeping
Sleeping
Create Human Validation Agent module with batch review, schema validation, and queue integration
28abf5b verified | """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) | |
| 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 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}") | |