Spaces:
Sleeping
Sleeping
Integrate lead_scoring_agent: score leads at start, inject scoring fields into asset flow
c5d8b23 verified | """ | |
| Recovery/Enrichment Agent | |
| Implements prioritized enrichment source config, rate limiting, retry/backoff, queue/handoff, and full action/decision logging for leads enrichment and recovery workflows. | |
| Integrates lead_scoring_agent to score leads at the start of processing and inject scoring fields into the asset flow for analytics and UI visibility. | |
| """ | |
| import time | |
| import logging | |
| from enum import Enum | |
| from typing import List, Dict, Any, Optional | |
| from lead_scoring_agent import score_lead | |
| # Agent config | |
| class EnrichmentSource(Enum): | |
| API = "api" | |
| BROWSER_AGENT = "browser_agent" | |
| HUMAN_REVIEW = "human_review" | |
| default_agent_config = { | |
| "priority_sources": [EnrichmentSource.API, EnrichmentSource.BROWSER_AGENT, EnrichmentSource.HUMAN_REVIEW], | |
| "rate_limit_per_minute": 10, | |
| "max_retries": 3, | |
| "retry_backoff_seconds": 5, | |
| } | |
| # Audit and log actions/decisions/edge cases | |
| def log_event(event_type: str, info: dict): | |
| logging.info({"event_type": event_type, **info}) | |
| # Simulate lead batch handoff and enrichment | |
| class LeadBatchHandler: | |
| def __init__(self, config=default_agent_config): | |
| self.config = config | |
| self.queue = [] | |
| self.history = [] | |
| self.rate_limit_counter = 0 | |
| self.last_reset = time.time() | |
| def reset_rate_limit(self): | |
| now = time.time() | |
| if now - self.last_reset > 60: | |
| self.rate_limit_counter = 0 | |
| self.last_reset = now | |
| log_event("rate_limit_reset", {}) | |
| def handle_lead_batch(self, leads: List[Dict[str, Any]]): | |
| results = [] | |
| for lead in leads: | |
| result = self._process_lead(lead) | |
| results.append(result) | |
| return results | |
| def _process_lead(self, lead: Dict[str, Any]) -> Dict[str, Any]: | |
| # Score the lead at the start of processing and inject scoring fields | |
| scoring_data = score_lead(lead) | |
| lead.update({ | |
| "lead_score": scoring_data.get("score", 0), | |
| "score_tier": scoring_data.get("tier", "unknown"), | |
| "score_factors": scoring_data.get("factors", {}), | |
| "scored_at": time.time() | |
| }) | |
| log_event("lead_scored", {"lead": lead, "scoring": scoring_data}) | |
| self.reset_rate_limit() | |
| for source in self.config["priority_sources"]: | |
| for attempt in range(self.config["max_retries"]): | |
| if self.rate_limit_counter >= self.config["rate_limit_per_minute"]: | |
| wait_time = 60 - (time.time() - self.last_reset) | |
| if wait_time > 0: | |
| log_event("rate_limit_wait", {"wait": wait_time}) | |
| time.sleep(wait_time) | |
| self.reset_rate_limit() | |
| log_event("enrich_attempt", {"source": source.value, "lead": lead, "attempt": attempt+1}) | |
| self.rate_limit_counter += 1 | |
| success, data = self._enrich_from_source(source, lead) | |
| if success: | |
| log_event("enrich_success", {"source": source.value, "lead": lead}) | |
| # Preserve scoring fields in enriched result | |
| return {"lead": lead, "enriched": data, "lead_score": lead["lead_score"], "score_tier": lead["score_tier"]} | |
| else: | |
| log_event("enrich_failure", {"source": source.value, "lead": lead, "attempt": attempt+1}) | |
| time.sleep(self.config["retry_backoff_seconds"]) | |
| self._escalate_to_human(lead) | |
| return {"lead": lead, "enriched": None, "escalated": True, "lead_score": lead["lead_score"], "score_tier": lead["score_tier"]} | |
| def _enrich_from_source(self, source: EnrichmentSource, lead: Dict[str, Any]) -> (bool, Optional[dict]): | |
| # Simulation logic; replace with real API/browser/human integration | |
| if source == EnrichmentSource.API: | |
| return False, None # simulate API failure | |
| if source == EnrichmentSource.BROWSER_AGENT: | |
| return True, {"company": "Simulated Corp", "status": "Browser enrichment success"} | |
| if source == EnrichmentSource.HUMAN_REVIEW: | |
| return True, {"company": "Simulated Corp", "status": "Human reviewed"} | |
| return False, None | |
| def _escalate_to_human(self, lead: Dict[str, Any]): | |
| self.queue.append(lead) | |
| log_event("escalate", {"lead": lead}) | |
| def get_audit_log(self): | |
| # Return all logged history (simulate) | |
| pass | |
| # Config interface sample | |
| def get_sample_config() -> dict: | |
| return default_agent_config | |
| # Usage sample | |
| def simulate_lead_handoff(): | |
| handler = LeadBatchHandler() | |
| leads = [{"first_name": "Mila", "company": "Sample Inc"}, {"first_name": "Rob", "company": "Unknown Ltd"}] | |
| results = handler.handle_lead_batch(leads) | |
| print(results) | |
| # add more logic for audit/fallback/escalation as needed | |
| if __name__ == "__main__": | |
| simulate_lead_handoff() |