# Customer-Facing Agents

Agents that interact directly with customers: support, sales, onboarding, and personalization.

---

## Customer-Facing Agent Capabilities

| Capability | Description | Channels |
|---|---|---|
| **Customer Support** | Answer questions, resolve issues | Chat, email, voice, social |
| **Sales Assistance** | Product recommendation, quotes, demos | Chat, email, video |
| **Onboarding** | Guide new users, setup, training | In-app, email, video |
| **Account Management** | Billing, subscriptions, upgrades | Chat, email, portal |
| **Technical Support** | Troubleshooting, escalation | Chat, screen share, remote |
| **Personalization** | Recommendations, content, offers | In-app, email, push |
| **Feedback Collection** | Surveys, NPS, feature requests | In-app, email, SMS |
| **Crisis Management** | Outage communication, apology | All channels |

---

## Agent Architectures for Customer-Facing

### 1. Support Triage Agent
```python
class SupportTriageAgent:
    def __init__(self):
        self.classifier = IssueClassifier()
        self.routing_rules = RoutingRules()
        self.escalation = EscalationManager()
    
    async def triage(self, ticket: SupportTicket) -> TriageResult:
        # 1. Classify issue
        classification = await self.classifier.classify(
            text=ticket.content,
            metadata=ticket.metadata
        )
        
        # 2. Determine priority
        priority = self.calculate_priority(classification, ticket.customer_tier)
        
        # 3. Check for known issues
        known_issue = await self.check_known_issues(classification)
        if known_issue:
            return TriageResult(
                action="auto_resolve",
                resolution=known_issue.workaround,
                priority=priority
            )
        
        # 4. Check if self-service possible
        if self.can_self_serve(classification):
            return TriageResult(
                action="self_serve",
                articles=self.find_help_articles(classification),
                priority=priority
            )
        
        # 5. Route to specialist
        assignee = self.routing_rules.route(classification)
        return TriageResult(
            action="route",
            assignee=assignee,
            priority=priority
        )
    
    def calculate_priority(self, classification: Classification, tier: CustomerTier) -> Priority:
        base = classification.base_priority
        if tier == "enterprise":
            return max(base, Priority.HIGH)
        elif tier == "premium":
            return max(base, Priority.MEDIUM)
        return base
```

### 2. Conversational Support Agent
```python
class ConversationalSupportAgent:
    def __init__(self):
        self.knowledge_base = KnowledgeBase()
        self.tools = {
            "account_lookup": AccountLookupTool(),
            "order_lookup": OrderLookupTool(),
            "refund": RefundTool(),
            "subscription": SubscriptionTool(),
            "technical_diagnostic": DiagnosticTool(),
            "escalate": EscalationTool()
        }
        self.guardrails = Guardrails()
    
    async def respond(self, message: CustomerMessage, context: ConversationContext) -> AgentResponse:
        # 1. Check guardrails
        guardrail_check = await self.guardrails.check(message, context)
        if not guardrail_check.allowed:
            return AgentResponse(
                text=guardrail_check.block_message,
                action="blocked"
            )
        
        # 2. Check for escalation triggers
        if self.should_escalate(message, context):
            return await self.escalate(message, context)
        
        # 3. Determine action
        action = await self.determine_action(message, context)
        
        if action.type == "answer":
            answer = await self.generate_answer(message, context)
            return AgentResponse(text=answer, action="answer")
        
        elif action.type == "tool":
            result = await self.execute_tool(action.tool, action.args)
            return await self.formulate_response(result)
        
        elif action.type == "handoff":
            return await self.handoff(action.target_agent, message, context)
        
        elif action.type == "clarify":
            return AgentResponse(
                text=action.question,
                action="clarify"
            )
    
    async def generate_answer(self, message: CustomerMessage, context: ConversationContext) -> str:
        # Retrieve relevant knowledge
        kb_results = await self.knowledge_base.search(
            query=message.text,
            customer_context=context.customer_profile,
            k=5
        )
        
        # Generate grounded answer
        prompt = f"""
        Customer: {message.text}
        Context: {context.summary}
        Knowledge: {kb_results}
        
        Provide a helpful, empathetic response. Cite sources.
        If unsure, say so and offer to escalate.
        """
        return await self.llm.generate(prompt)
```

### 3. Sales Assistant Agent
```python
class SalesAssistantAgent:
    def __init__(self):
        self.product_catalog = ProductCatalog()
        self.pricing_engine = PricingEngine()
        self.crm = CRMIntegration()
    
    async def handle_inquiry(self, inquiry: SalesInquiry) -> SalesResponse:
        # 1. Qualify lead
        qualification = await self.qualify_lead(inquiry)
        
        if not qualification.qualified:
            return SalesResponse(
                text="Thanks for your interest! Let me ask a few questions...",
                action="qualify",
                questions=qualification.questions
            )
        
        # 2. Match products
        matches = await self.product_catalog.match(qualification.needs)
        
        # 3. Generate quote
        quote = await self.pricing_engine.generate_quote(
            products=matches,
            volume=inquiry.estimated_volume,
            contract_term=inquiry.contract_term
        )
        
        # 4. Schedule demo if requested
        if inquiry.wants_demo:
            demo = await self.schedule_demo(inquiry.contact, matches)
        
        return SalesResponse(
            text=self.format_response(matches, quote, demo),
            action="quote",
            quote=quote,
            next_steps=["Review quote", "Schedule demo", "Start trial"]
        )
```

### 4. Onboarding Agent
```python
class OnboardingAgent:
    def __init__(self):
        self.steps = self._load_onboarding_flow()
        self.progress_tracker = ProgressTracker()
    
    async def start_onboarding(self, user: User) -> OnboardingPlan:
        # Personalize flow
        plan = self.personalize_flow(user)
        
        # Initialize progress
        await self.progress_tracker.start(user.id, plan)
        
        # Send first step
        first_step = plan.steps[0]
        return OnboardingPlan(
            current_step=first_step,
            total_steps=len(plan.steps),
            estimated_time=plan.estimated_time
        )
    
    async def complete_step(self, user_id: str, step_id: str, 
                           user_input: UserInput) -> StepResult:
        step = self.steps[step_id]
        
        # Validate input
        validation = await step.validate(user_input)
        if not validation.valid:
            return StepResult(success=False, error=validation.error)
        
        # Process
        result = await step.execute(user_input)
        
        # Track progress
        await self.progress_tracker.complete_step(user_id, step_id, result)
        
        # Next step
        next_step = self.get_next_step(step_id)
        if next_step:
            return StepResult(
                success=True,
                next_step=next_step,
                progress=await self.progress_tracker.get_progress(user_id)
            )
        else:
            # Onboarding complete
            return StepResult(
                success=True,
                completed=True,
                celebration=self.generate_celebration()
            )
```

---

## Channel-Specific Implementations

### 1. Chat (Web/Mobile)
```python
class ChatSupportAgent:
    def __init__(self, websocket: WebSocket):
        self.ws = websocket
        self.session = ChatSession()
    
    async def handle_message(self, message: ChatMessage):
        # Typing indicator
        await self.send_typing()
        
        # Process
        response = await self.agent.respond(message, self.session.context)
        
        # Stream response
        if response.stream:
            async for chunk in self.agent.stream_response(response):
                await self.ws.send_json({"type": "chunk", "content": chunk})
        else:
            await self.ws.send_json({"type": "message", "content": response.text})
        
        # Update session
        self.session.add_exchange(message, response)
```

### 2. Email Support Agent
```python
class EmailSupportAgent:
    def __init__(self, imap: IMAPClient, smtp: SMTPClient):
        self.imap = imap
        self.smtp = smtp
    
    async def process_inbox(self):
        emails = await self.imap.fetch_unread()
        
        for email in emails:
            # Parse
            ticket = self.parse_email(email)
            
            # Check for auto-reply loops
            if self.is_auto_reply(email):
                continue
            
            # Process
            response = await self.agent.respond(ticket)
            
            # Send reply
            await self.send_reply(email, response)
            
            # Mark read
            await self.imap.mark_read(email.uid)
    
    async def send_reply(self, original: Email, response: AgentResponse):
        reply = EmailMessage()
        reply["To"] = original["From"]
        reply["Subject"] = f"Re: {original['Subject']}"
        reply["In-Reply-To"] = original["Message-ID"]
        reply["References"] = original["References"]
        reply.set_content(response.text)
        
        await self.smtp.send(reply)
```

### 3. Voice Support Agent
```python
class VoiceSupportAgent:
    def __init__(self, stt: STTClient, tts: TTSClient, llm: LLM):
        self.stt = stt
        self.tts = tts
        self.llm = llm
    
    async def handle_call(self, call: VoiceCall):
        await call.answer()
        
        # Greeting
        await call.play(await self.tts.synthesize(
            "Hello! How can I help you today?"
        ))
        
        while call.active:
            # Listen
            audio = await call.record(max_duration=30)
            transcript = await self.stt.transcribe(audio)
            
            # Process
            response = await self.llm.generate(f"""
                Customer: {transcript}
                Context: {self.get_context()}
                
                Respond naturally. Keep responses concise for voice.
            """)
            
            # Speak
            await call.play(await self.tts.synthesize(transcript))
        
        await call.hangup()
```

---

## Personalization

### Recommendation Engine
```python
class PersonalizationEngine:
    def __init__(self):
        self.collaborative = CollaborativeFiltering()
        self.content_based = ContentBasedFiltering()
        self.contextual = ContextualBandit()
    
    async def recommend(self, user: User, context: Context, k: int = 10) -> list[Recommendation]:
        # Blend multiple approaches
        collab_recs = await self.collaborative.recommend(user, k*2)
        content_recs = await self.content_based.recommend(user, k*2)
        contextual_recs = await self.contextual.recommend(user, context, k*2)
        
        # Blend with weights
        blended = self.blend_recommendations(
            collab_recs, content_recs, contextual_recs,
            weights=[0.4, 0.3, 0.3]
        )
        
        # Apply business rules
        filtered = self.apply_business_rules(blended, user)
        
        return filtered[:k]
```

### Dynamic Content
```python
class DynamicContentEngine:
    def __init__(self):
        self.templates = TemplateEngine()
        self.rules = PersonalizationRules()
    
    async def render(self, template_name: str, user: User, context: Context) -> str:
        # Get personalization data
        profile = await self.build_profile(user)
        
        # Select variant
        variant = self.rules.select_variant(template_name, profile)
        
        # Render
        return await self.templates.render(
            f"{template_name}/{variant}",
            user=user,
            profile=profile,
            context=context
        )
```

---

## Quality & Safety

### Guardrails
```python
class CustomerFacingGuardrails:
    def __init__(self):
        self.pii_detector = PIIDetector()
        self.toxicity = ToxicityDetector()
        self.policy = PolicyEngine()
    
    async def check_input(self, message: str, context: Context) -> GuardrailResult:
        checks = await asyncio.gather(
            self.pii_detector.check(message),
            self.toxicity.check(message),
            self.policy.check(message, context)
        )
        
        return GuardrailResult(
            allowed=all(c.allowed for c in checks),
            violations=[c for c in checks if not c.allowed]
        )
    
    async def check_output(self, response: str, context: Context) -> GuardrailResult:
        # Check for PII leakage
        if self.pii_detector.contains_pii(response):
            return GuardrailResult(allowed=False, violation="PII in response")
        
        # Check for medical/legal/financial advice
        if self.requires_disclaimer(response):
            return GuardrailResult(
                allowed=False, 
                violation="Requires professional disclaimer"
            )
        
        # Check tone
        if not self.is_empathetic(response):
            return GuardrailResult(
                allowed=False,
                violation="Response not empathetic enough"
            )
        
        return GuardrailResult(allowed=True)
```

### Escalation
```python
class EscalationManager:
    def __init__(self):
        self.escalation_paths = {
            "billing": ["billing_specialist", "billing_manager"],
            "technical": ["tier2_support", "engineering"],
            "complaint": ["retention_specialist", "manager"],
            "legal": ["legal_team"],
            "security": ["security_team"]
        }
    
    async def escalate(self, reason: EscalationReason, context: Context) -> EscalationResult:
        path = self.escalation_paths.get(reason.category, ["general_manager"])
        
        for target in path:
            specialist = await self.get_available_specialist(target)
            if specialist:
                await self.transfer(specialist, context)
                return EscalationResult(transferred_to=target)
        
        # Fallback: queue for human
        await self.queue_for_human(context)
        return EscalationResult(queued=True)
```

---

## Metrics & Analytics

### Key Metrics
| Metric | Target | Measurement |
|---|---|---|
| **First Response Time** | < 1 min (chat), < 1 hr (email) | Timestamp diff |
| **Resolution Time** | < 24 hrs | Ticket open → resolved |
| **First Contact Resolution** | > 70% | Resolved in 1 interaction |
| **CSAT** | > 4.5/5 | Post-interaction survey |
| **NPS** | > 50 | Quarterly survey |
| **Escalation Rate** | < 10% | Escalated / total |
| **Bot Deflection** | > 40% | Bot resolved / total |
| **Agent Utilization** | 70-80% | Active time / scheduled |

### Analytics Dashboard
```python
class SupportAnalytics:
    def __init__(self, event_store: EventStore):
        self.events = event_store
    
    async def get_metrics(self, window: TimeWindow) -> SupportMetrics:
        events = await self.events.query(
            start=window.start,
            end=window.end
        )
        
        return SupportMetrics(
            volume=len(events),
            avg_response_time=self.avg_response_time(events),
            avg_resolution_time=self.avg_resolution_time(events),
            fcr_rate=self.fcr_rate(events),
            csat=await self.get_csat(window),
            escalation_rate=self.escalation_rate(events),
            top_issues=self.top_issues(events),
            agent_performance=self.agent_performance(events)
        )
```

---

## Related Notes

- [[Domains/Sales-Agents.md]] — Sales-specific agents
- [[ToolUse/Function-Calling.md]] — Tools for support
- [[ToolUse/MCP-Protocol.md]] — MCP for integrations
- [[ToolUse/API-Integration.md]] — CRM, helpdesk APIs
- [[ToolUse/Browser-Automation.md]] — Screen sharing, co-browsing
- [[ToolUse/Code-Execution.md]] — Technical diagnostics
- [[ToolUse/API-Integration.md]] — API integrations
- [[MultiAgent/Coordination-Patterns.md]] — Escalation patterns
- [[MultiAgent/Delegation-Handoff.md]] — Handoff to humans
- [[MultiAgent/Safety-Coordination.md]] — Guardrails
- [[Architectures/Human-in-the-Loop.md]] — Human-in-loop
- [[Production/Observability.md]] — Metrics, dashboards
- [[Production/Security.md]] — PII, compliance
- [[Production/Reliability.md]] — Uptime, reliability
- [[Security/Secure-Coding.md]] — PII handling
- [[Security/Supply-Chain-Security.md]] — Vendor security

---

*Sources: Zendesk, Intercom, Salesforce Service Cloud, HubSpot, Freshdesk, Gorgias, Help Scout, LiveChat, Drift, Intercom, Ada, Ultimate.ai, Forethought, Sierra, PolyAI, Replicant, Observe.AI, Balto, Cresta, Uniphore, ASAPP, Level AI, Customer Service Benchmark reports, Contact Center best practices, CCaaS platforms.*