Customer-Facing Agents

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

Customer-Facing Agent Capabilities

Agent Architectures for Customer-Facing

1. Support Triage Agent

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

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

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

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)

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

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

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

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

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

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

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

Analytics Dashboard

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

• Sales Agents — Sales-specific agents
• Function Calling — Tools for support
• MCP Protocol — MCP for integrations
• API Integration — CRM, helpdesk APIs
• Browser Automation — Screen sharing, co-browsing
• Code Execution — Technical diagnostics
• API Integration — API integrations
• Coordination Patterns — Escalation patterns
• Delegation Handoff — Handoff to humans
• Safety Coordination — Guardrails
• Human in the Loop — Human-in-loop
• Observability — Metrics, dashboards
• Security — PII, compliance
• Reliability — Uptime, reliability
• Secure Coding — PII handling
• Supply Chain Security — 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.