How to Build a Multi-Agent Customer Support Router in Python

A tiered multi-agent customer support system that classifies queries, routes to specialist bots, uses cheap models for simple questions and expensive ones only for complex issues, and escalates to human agents when automated handling fails.

Patterns used: Supervisor, TalkerReasoner, HumanInTheLoop, BoundedExecution, GuardrailChain, RouterMiddleware

---

Requirements

• Functional — classify every incoming query into billing, technical, account, or escalate; route to the right specialist; hand off to a human when automated handling isn't appropriate.
• Non-functional — most queries should resolve without a senior/expensive model; classification itself needs to be cheap since it runs on every single query.
• Audit — every routed query should be traceable to the classification decision that sent it there.
• Not required — no persistent memory across separate support sessions in this recipe (each query is classified independently).

Architecture decisions

Decision	Why	Why not the alternative
Supervisor for routing	Categories (billing/technical/account/escalate) are fixed and known in advance.	Orchestrator-Workers implies discovering the right specialists per-input; here the categories never change.
Classifier on fast, specialists on smart	Classification is a cheap pattern-match; specialist responses need real reasoning quality.	Using smart for classification alone would multiply the highest-volume call in the system by ~10x cost for no quality benefit.
One workflow (route), not a nested Supervisor-inside-TalkerReasoner	Supervisor.routes expects Agent objects, not Pattern objects — nesting patterns inside pattern slots isn't supported by the schema.	The _fast tier agents are declared but used outside this workflow (e.g. a cost-optimized fallback), documented directly in the blueprint's own comment.

Four-pillar mapping

Requirement	Pillar	Capability
Classify and route by category	Execution	Supervisor pattern
Track daily routing spend	Observability	observability.cost_budget
Trace each classify/specialist call	Observability	observability.tracing
Escalate to human when needed	Execution	escalation_writer agent + downstream HumanInTheLoop (Python implementation)

Blueprint (declarative form)

The real, verified file at examples/cookbook/customer-support/support_router/blueprint.yaml, compiled against PyAgentAdapter as part of this repo's test suite:

api_version: pyagent/v1
metadata:
  name: support-router
  version: 1.0.0
  description: Classify customer queries, route to specialist bots, escalate to human when needed

providers:
  fast:  { model: gpt-4o-mini }
  smart: { model: claude-sonnet-4-20250514 }

agents:
  supervisor:   { provider: fast,  prompt: "Classify as: billing, technical, account, escalate." }
  billing_deep: { provider: smart, prompt: "Senior billing specialist — complex cases." }
  tech_deep:    { provider: smart, prompt: "Senior engineer — deep technical debug." }
  account_deep: { provider: smart, prompt: "Senior account specialist — SSO, compliance." }
  escalation_writer: { provider: fast, prompt: "Summarize for human handoff." }

workflows:
  route:
    pattern: supervisor
    agents:
      classifier: supervisor
      routes: { billing: billing_deep, technical: tech_deep, account: account_deep, escalate: escalation_writer }

observability:
  tracing: { enabled: true }
  cost_budget: { daily_usd: 200.0, alert_threshold: 0.8 }

pyagent-blueprint validate support-router.yaml
pyagent-blueprint test support-router.yaml

Production checklist

Ran this exact blueprint through PyAgentAdapter.compile() and inspected the real diagnostics:

• ✅ The routing workflow runs as declared — route compiles and executes with no diagnostics on the workflow structure itself.
• ⚠️ observability.cost_budget is declared but not auto-enforced — compiling emits BUDGET_UNSUPPORTED: the $200/day budget is recorded but not enforced. Wire real enforcement via graph.wire_cost_tracker(tracker).
• The _fast tier agents and the TalkerReasoner tiering shown in the Python implementation below aren't expressed in the blueprint's route workflow — Supervisor.routes requires Agent targets, not nested patterns, so the cost-tiering happens in code, not the declared spec. A real, documented schema limitation, not an oversight.

---

Architecture

flowchart TD
    C[Customer Query] --> G[Input Guardrails\nPII redaction]
    G --> S[Supervisor\nClassify intent]

    S -->|billing| B[TalkerReasoner\nBilling Bot]
    S -->|technical| T[TalkerReasoner\nTech Support]
    S -->|account| A[TalkerReasoner\nAccount Bot]
    S -->|escalate| H[Human-in-the-Loop]

    B -->|easy| BR1[Fast Response\ngpt-4o-mini]
    B -->|complex| BR2[Deep Response\nclaude-sonnet]
    T -->|easy| TR1[Quick Fix\ngpt-4o-mini]
    T -->|complex| TR2[Full Debug\nclaude-sonnet]
    H --> HR[Human Agent\nvia ticket queue]

---

Implementation

import asyncio
from pyagent_patterns.base import Agent, Message
from pyagent_patterns.orchestration import Supervisor
from pyagent_patterns.advanced import TalkerReasoner, HumanInTheLoop
from pyagent_patterns.advanced.human_in_the_loop import HumanDecision
from pyagent_patterns.recovery import BoundedExecution
from pyagent_patterns.guardrails import GuardrailChain, PIIGuard, LengthGuard
from pyagent_router.middleware import RouterMiddleware
from pyagent_providers import AnthropicLLM, OpenAILLM

# ── LLMs ──────────────────────────────────────────────────────────────────────
fast_llm  = OpenAILLM("gpt-4o-mini")
smart_llm = AnthropicLLM("claude-sonnet-4-20250514")

model_registry = {
    "gpt-4o-mini":              fast_llm,
    "claude-sonnet-4-20250514": smart_llm,
}
router = RouterMiddleware(model_registry=model_registry)

# ── Guardrails ─────────────────────────────────────────────────────────────────
input_guard = GuardrailChain([
    LengthGuard(max_chars=3_000, truncate=True),
    PIIGuard(redact=True),   # protect customer PII in logs
])

# ── Tier 1: Billing bot (TalkerReasoner) ─────────────────────────────────────
billing_bot = TalkerReasoner(
    talker=router.wrap(
        Agent("billing_fast", fast_llm,
              system_prompt=(
                  "You are a billing support agent. Answer quickly and clearly. "
                  "You handle: invoice questions, payment methods, refund policies, "
                  "subscription changes, and pricing. Be concise — 2-3 sentences max."
              )),
    ),
    reasoner=router.wrap(
        Agent("billing_deep", smart_llm,
              system_prompt=(
                  "You are a senior billing specialist. Handle complex cases: "
                  "disputed charges, partial refunds, multi-seat subscription adjustments, "
                  "enterprise billing questions. Provide step-by-step resolution."
              )),
    ),
    handoff_threshold=5,   # difficulty ≥ 5 goes to reasoner
)

# ── Tier 2: Technical support bot ────────────────────────────────────────────
tech_bot = TalkerReasoner(
    talker=router.wrap(
        Agent("tech_fast", fast_llm,
              system_prompt=(
                  "You are a tech support agent. Handle common issues: login problems, "
                  "password resets, browser compatibility, basic integrations. "
                  "Give step-by-step instructions."
              )),
    ),
    reasoner=router.wrap(
        Agent("tech_deep", smart_llm,
              system_prompt=(
                  "You are a senior engineer doing technical support. Handle complex issues: "
                  "API integration failures, webhook debugging, performance problems, "
                  "data sync issues, custom configurations. Ask clarifying questions if needed."
              )),
    ),
    handoff_threshold=4,
)

# ── Tier 3: Account management bot ───────────────────────────────────────────
account_bot = TalkerReasoner(
    talker=router.wrap(
        Agent("account_fast", fast_llm,
              system_prompt=(
                  "You are an account support agent. Handle: username changes, "
                  "email updates, team member management, permissions, and SSO setup."
              )),
    ),
    reasoner=router.wrap(
        Agent("account_deep", smart_llm,
              system_prompt=(
                  "You are a senior account specialist. Handle: GDPR data requests, "
                  "account mergers, complex permission structures, enterprise SSO, "
                  "data export and deletion requests."
              )),
    ),
    handoff_threshold=6,
)

# ── Tier 4: Human escalation ──────────────────────────────────────────────────
def queue_human_review(output: str, metadata: dict) -> HumanDecision:
    """Route to human agent via your support queue (e.g. Zendesk, Linear)."""
    ticket_id = _create_support_ticket(
        summary=output[:200],
        priority="high" if "urgent" in output.lower() else "normal",
        metadata=metadata,
    )
    print(f"[TICKET CREATED] #{ticket_id}")
    # Return a holding response while human picks it up
    return HumanDecision(
        approved=True,
        modified_output=(
            f"I've escalated your issue to our specialist team. "
            f"Ticket #{ticket_id} has been created. "
            f"You'll hear back within 2 business hours."
        ),
    )

human_handler = HumanInTheLoop(
    agent=router.wrap(
        Agent("human_prep", fast_llm,
              system_prompt=(
                  "Prepare a concise summary for the human support agent: "
                  "1. Customer issue (1 sentence) "
                  "2. What was already tried "
                  "3. Recommended action"
              )),
    ),
    review_fn=queue_human_review,
    high_risk_keywords=["legal", "lawsuit", "fraud", "hacked", "emergency"],
)

# ── Classifier (Supervisor routing) ──────────────────────────────────────────
classifier = Agent(
    "classifier", fast_llm,
    system_prompt=(
        "Classify customer support queries into exactly one category. "
        "Reply with ONLY the category name.\n"
        "Categories:\n"
        "  billing   — payments, invoices, refunds, subscriptions\n"
        "  technical — bugs, errors, API, integrations, performance\n"
        "  account   — login, permissions, team, SSO, data requests\n"
        "  escalate  — angry customers, legal threats, complex edge cases, "
        "              anything you're unsure about"
    ),
)

supervisor = Supervisor(
    classifier=classifier,
    routes={
        "billing":   billing_bot,
        "technical": tech_bot,
        "account":   account_bot,
        "escalate":  human_handler,
    },
)

# ── Recovery wrapper ──────────────────────────────────────────────────────────
safe_supervisor = BoundedExecution(
    pattern=supervisor,
    fallback=Agent(
        "fallback_agent", fast_llm,
        system_prompt=(
            "You are a helpful support agent. A technical issue occurred with our "
            "routing system. Apologize briefly and ask the customer to try again "
            "or contact support@example.com."
        ),
    ),
    max_retries=2,
    timeout_seconds=25.0,
)

# ── Main handler ──────────────────────────────────────────────────────────────
async def handle_query(customer_query: str) -> dict:
    # Guardrail check
    check = input_guard.check(customer_query)
    if not check.passed:
        return {"response": "I'm sorry, I couldn't process your message.", "blocked": True}
    safe_query = check.sanitized_content or customer_query

    # Run through support workflow
    result = await safe_supervisor.run(safe_query)

    return {
        "response":       result.output,
        "category":       result.metadata.get("route"),
        "model_used":     result.metadata.get("routed_model", "unknown"),
        "recovery_level": result.metadata.get("recovery_level", 0),
        "escalated":      result.metadata.get("route") == "escalate",
    }


def _create_support_ticket(summary: str, priority: str, metadata: dict) -> str:
    """Create a Zendesk ticket and return the ticket number as a string."""
    import httpx, os
    r = httpx.post(
        os.environ["ZENDESK_URL"] + "/api/v2/tickets.json",
        json={"ticket": {"subject": summary[:80], "priority": priority,
                         "comment": {"body": summary}}},
        auth=(os.environ["ZENDESK_EMAIL"] + "/token", os.environ["ZENDESK_TOKEN"]),
        timeout=15.0,
    )
    r.raise_for_status()
    return str(r.json()["ticket"]["id"])


# ── Run it ────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    queries = [
        "Why was I charged twice this month?",
        "My API webhooks stopped firing after the latest update",
        "I need to delete all my data immediately under GDPR",
        "This is unacceptable, I'm calling my lawyer",
    ]

    async def demo():
        for query in queries:
            print(f"\nQuery: {query}")
            result = await handle_query(query)
            print(f"Category:  {result['category']}")
            print(f"Model:     {result['model_used']}")
            print(f"Response:  {result['response'][:120]}...")

    asyncio.run(demo())

---

Expected Output

Query: Why was I charged twice this month?
Category:  billing
Model:     gpt-4o-mini   ← easy billing question, fast model
Response:  This can happen if two subscriptions are active simultaneously or if a
           payment retry was processed. To confirm, please check your invoices at
           Settings → Billing → Invoice history...

Query: My API webhooks stopped firing after the latest update
Category:  technical
Model:     claude-sonnet-4-20250514   ← complex technical issue, smart model
Response:  Webhook failures after an update are often caused by: 1) SSL certificate
           changes, 2) payload format changes in the new version, 3) endpoint timeout
           thresholds. Let's debug this step by step: First, check your webhook logs...

Query: I need to delete all my data immediately under GDPR
Category:  account
Model:     claude-sonnet-4-20250514   ← GDPR = high difficulty, smart model
Response:  We take GDPR requests seriously. Here's the process: 1) Submit a formal
           Data Subject Access Request at privacy.example.com/dsar. 2) We'll confirm
           receipt within 24 hours. 3) Deletion completes within 30 days per GDPR Art. 17...

Query: This is unacceptable, I'm calling my lawyer
Category:  escalate
Model:     gpt-4o-mini + human
Response:  I've escalated your issue to our specialist team. Ticket #SUP-847291 has
           been created. You'll hear back within 2 business hours.
[TICKET CREATED] #SUP-847291

---

Customization

Add a knowledge base tool

from pyagent_patterns.advanced import ReAct

def search_kb(query: str) -> str:
    # Search your Confluence/Notion/Help Center
    return kb_client.search(query, top_k=3)

tech_bot_with_kb = ReAct(
    agent=Agent("tech_kb", smart_llm, system_prompt="Answer using the knowledge base."),
    tools={"search_kb": search_kb},
    max_steps=3,
)

Multi-turn conversation

conversation_history: list[Message] = []

async def chat(user_message: str) -> str:
    conversation_history.append(Message.user(user_message))
    result = await safe_supervisor.run(conversation_history)
    conversation_history.append(Message.assistant(result.output))
    return result.output

SLA-based routing

def route_by_sla(query: str, customer_tier: str) -> str:
    if customer_tier == "enterprise":
        return "escalate"   # always get human for enterprise
    if "urgent" in query.lower() or "down" in query.lower():
        return "escalate"
    return None   # let the classifier decide

---

Cost Profile

Query type	Typical model	Avg cost	Volume (1k/day)
Simple billing	gpt-4o-mini	$0.0003	$9/mo
Complex billing	claude-sonnet	$0.003	$90/mo
Simple tech	gpt-4o-mini	$0.0003	$9/mo
Complex tech	claude-sonnet	$0.004	$120/mo
GDPR / legal	claude-sonnet	$0.005	$150/mo
Human escalation	gpt-4o-mini + human	$0.0003 + agent time	$9/mo + agent time
Blended average	mix	~$0.001	~$30/mo

Routing saves ~70% vs always using claude-sonnet for everything.

---

See Also

• Supervisor pattern
• TalkerReasoner pattern
• HumanInTheLoop pattern
• Guardrails Guide
• Recovery Guide