Spaces:
Sleeping
Sleeping
Integrate DataEnrichmentManager into outreach_agent - add enrichment to contact validation
97376e1 verified | import logging | |
| from typing import Dict, Any, Optional, List, Callable | |
| # Import lead_scoring_agent for lead scoring integration | |
| from lead_scoring_agent import score_lead | |
| from data_enrichment import DataEnrichmentManager | |
| class OutreachAgent: | |
| """ | |
| OutreachAgent handles lead intake, message tailoring, and multi-channel delivery. | |
| Integrates lead_scoring_agent to score leads at intake for prioritization and routing. | |
| """ | |
| def __init__(self, username: str, channel_handlers: Optional[Dict[str, Callable]]=None): self.logger = logging.getLogger("OutreachAgent") | |
| self.lead_assets = {} | |
| self.analytics = [] | |
| self.channel_handlers = channel_handlers or {} | |
| self.username = username | |
| self.enrichment_manager = DataEnrichmentManager(username) | |
| def intake_lead(self, lead_profile: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| Intake lead with scoring integration as first logic step. | |
| Calls score_lead(profile) to enrich profile with scoring metadata. | |
| """ | |
| # FIRST STEP: Score the lead using lead_scoring_agent | |
| scored_profile = score_lead(lead_profile) | |
| self.logger.info(f"Lead scored: {scored_profile.get('name', 'Unknown')} - Score: {scored_profile.get('lead_score', 'N/A')}") | |
| # Enrich profile with additional data | |
| enriched_profile = self.validate_and_enrich_profile(scored_profile) | |
| self.logger.info(f"Intake for {enriched_profile.get('name', 'Unknown Lead')}") | |
| asset_id = enriched_profile.get('id') | |
| if asset_id: | |
| self.lead_assets[asset_id] = enriched_profile | |
| return enriched_profile | |
| def validate_and_enrich_profile(self, profile: Dict[str, Any]) -> Dict[str, Any]: | |
| # Enrich with historical, persona, engagements, etc. | |
| # Preserve scoring metadata from lead_scoring_agent | |
| # ENRICHMENT STEP: Use DataEnrichmentManager to enrich profile with 14 APIs | |
| try: | |
| enriched_data = self.enrichment_manager.enrich_contact_full(profile) | |
| if enriched_data: | |
| # Merge enriched data into profile (preserve existing fields) | |
| for key, value in enriched_data.items(): | |
| if key not in profile or not profile[key]: # Only update if field is missing | |
| profile[key] = value | |
| self.logger.info(f"Profile enriched with {len(enriched_data)} data sources") | |
| except Exception as e: | |
| self.logger.warning(f"Enrichment failed: {e}. Continuing with available data.") | |
| required_fields = ['persona', 'channel_fit'] | |
| for field in required_fields: | |
| if field not in profile: | |
| profile[field] = 'unspecified' | |
| # Simulate enrichment logic | |
| profile['validation'] = 'success' | |
| if 'recent_posts' not in profile: | |
| profile['recent_posts'] = [] | |
| return profile | |
| def select_template(self, profile: Dict[str, Any]) -> str: | |
| persona = profile.get('persona', 'Generic') | |
| pain = profile.get('pain_opportunity', 'N/A') | |
| buyer_signals = profile.get('buyer_signals', []) | |
| channel = profile.get('channel_fit', 'email') | |
| # Template selection logic (expandable) | |
| template = f"Hi {{name}}, as a {{persona}}, we noticed: {{pain}}. {{buyer_signals}}. Ready to discuss via {{channel}}?" | |
| return template | |
| def tailor_message(self, profile: Dict[str, Any], template: str) -> str: | |
| variables = { | |
| "name": profile.get("name", "there"), | |
| "persona": profile.get("persona", "Professional"), | |
| "pain": profile.get("pain_opportunity", "key considerations"), | |
| "buyer_signals": ', '.join(profile.get("buyer_signals", [])), | |
| "channel": profile.get("channel_fit", "email"), | |
| "context": profile.get("context_snippets", ""), | |
| "recent_posts": '\n'.join(profile.get("recent_posts", [])), | |
| "complaints": ', '.join(profile.get("customer_complaint_tags", [])), | |
| "hooks": ', '.join(profile.get("competitor_industry_hooks", [])), | |
| "urgency": profile.get("urgency_triggers", "") | |
| } | |
| message = template.format(**variables) | |
| # Attach recent posts, urgency, hooks etc with nuance | |
| if variables["context"]: | |
| message += f"\nContext: {variables['context']}" | |
| if variables["recent_posts"]: | |
| message += f"\nRecent posts: {variables['recent_posts']}" | |
| if variables["complaints"]: | |
| message += f"\nCustomer complaints: {variables['complaints']}" | |
| if variables["hooks"]: | |
| message += f"\nIndustry/competitor hooks: {variables['hooks']}" | |
| if variables["urgency"]: | |
| message += f"\nUrgency triggers: {variables['urgency']}" | |
| return message | |
| def deliver(self, profile: Dict[str, Any], message: str, channel: str) -> Dict[str, Any]: | |
| # Channel handler pattern for extensibility | |
| handler = self.channel_handlers.get(channel, self.default_handler) | |
| result = handler(profile, message) | |
| self.logger.info(f"Delivered to {profile.get('name')} via {channel}") | |
| self.update_analytics(profile, channel, result) | |
| return result | |
| def default_handler(self, profile: Dict[str, Any], message: str) -> Dict[str, Any]: | |
| # Simulate sending logic | |
| return {"status": "sent", "channel": profile.get('channel_fit', 'email'), "message": message} | |
| def browser_social_actions(self, profile: Dict[str, Any], action: str, extra: Optional[Dict[str, Any]]=None) -> Dict[str, Any]: | |
| # Simulate browser agent: connect, DM, comment, like, milestone outreach | |
| valid_actions = ["connect", "dm", "comment", "like", "milestone"] | |
| if action not in valid_actions: | |
| raise ValueError(f"Unsupported social action: {action}") | |
| self.logger.info(f"Performed {action} for {profile.get('name')}") | |
| result = {"action": action, "status": "success", "lead": profile.get('id')} | |
| self.update_analytics(profile, "social_action", result) | |
| return result | |
| def update_analytics(self, profile: Dict[str, Any], channel: str, result: Dict[str, Any]): | |
| entry = { | |
| "lead_id": profile.get("id"), | |
| "channel": channel, | |
| "result": result | |
| } | |
| self.analytics.append(entry) | |
| if profile.get("id") in self.lead_assets: | |
| self.lead_assets[profile.get("id")]["last_engagement"] = entry | |
| def feedback_loop(self, profile: Dict[str, Any], feedback: str): | |
| # Store feedback and auto-update asset | |
| pid = profile.get("id") | |
| if pid and pid in self.lead_assets: | |
| self.lead_assets[pid]["feedback"] = feedback | |
| self.logger.info(f"Feedback for {profile.get('name')}: {feedback}") | |
| def get_transcript(self, profile_id: str) -> Dict[str, Any]: | |
| asset = self.lead_assets.get(profile_id) | |
| return { | |
| "profile": asset, | |
| "last_message": asset.get("last_engagement", {}).get("result", {}).get("message", "") if asset else "", | |
| "engagement_stats": [a for a in self.analytics if a.get("lead_id") == profile_id] | |
| } | |
| # Example usage handler for email | |
| def email_handler(profile, message): | |
| # Integrate real email API here | |
| return {"status": "sent_via_email", "recipient": profile.get("email", "n/a"), "message": message} | |
| # Example handlers for expansion for LinkedIn, SMS, browser agent, etc. | |
| def linkedin_handler(profile, message): | |
| # Integrate real LinkedIn API calls here | |
| return {"status": "sent_via_linkedin", "recipient": profile.get("linkedin", "n/a"), "message": message} | |
| def sms_handler(profile, message): | |
| # Integrate real SMS API calls here | |
| return {"status": "sent_via_sms", "recipient": profile.get("phone", "n/a"), "message": message} | |
| # Usage: | |
| # agent = OutreachAgent(channel_handlers={"email": email_handler, "linkedin": linkedin_handler, "sms": sms_handler}) | |
| # lead = agent.intake_lead({...}) | |
| # template = agent.select_template(lead) | |
| # personalized_msg = agent.tailor_message(lead, template) | |
| # agent.deliver(lead, personalized_msg, channel="email") | |
| # agent.browser_social_actions(lead, "connect") |