Create outreach_agent.py

outreach_agent.py

@@ -0,0 +1,136 @@
+import logging
+from typing import Dict, Any, Optional, List, Callable
+
+class OutreachAgent:
+    def __init__(self, channel_handlers: Optional[Dict[str, Callable]]=None):
+        self.logger = logging.getLogger("OutreachAgent")
+        self.lead_assets = {}
+        self.analytics = []
+        self.channel_handlers = channel_handlers or {}
+
+    def intake_lead(self, lead_profile: Dict[str, Any]) -> Dict[str, Any]:
+        enriched_profile = self.validate_and_enrich_profile(lead_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.
+        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")