Customer Service Bot

A multi-tier customer service system with intelligent routing, escalation handling, and human-in-the-loop review. This example demonstrates LLM agents, programmatic agents, human agents, conditional routing, and intelligence functions working together.

Architecture

Customer Message
    │
    ▼
┌─────────────┐  category == 'technical'  ┌──────────────┐
│   Triage     │──────────────────────────▶│  Technical    │
│ (gpt-4o-mini)│                           │  (gpt-4o)    │
│              │  category == 'billing'    │              │
│ yields:      │──────────────────────────▶│ yields:      │
│  category    │                           │  solution    │
│  urgency     │  satisfaction < 4         │  resolved    │──── resolved == false ──▶ Human Reviewer
│  satisfaction│──────────────────────────▶└──────────────┘
└─────────────┘        │
                       │
                       ▼
               ┌──────────────┐
               │   Billing     │
               │ (gpt-4o-mini) │
               │               │
               │ yields:       │
               │  resolution   │──── needs_approval == true ──▶ Human Reviewer
               │  needs_approval│
               └──────────────┘

YAML Configuration

The agency's configuration (paste it into Studio's YAML view, or import it once from a config.yaml):

name: "customer_service"
version: "1.0.0"
architecture: "managed_mesh"

manager:
  enabled: true
  model: "gpt-4o-mini"
  coordination_rules:
    max_response_time: 30

timeouts:
  mesh_call_default_timeout_s: 60

mesh:
  max_retries: 2

agents:
  # Step 1: Classify and route the customer's message
  triage:
    name: "triage"
    model: "gpt-4o-mini"
    temperature: 0.1
    max_tokens: 300
    prompt: |
      You are a customer service triage agent.

      Classify the customer's message:
      - category: "technical", "billing", or "general"
      - urgency: 1 (low) to 10 (critical)
      - satisfaction: estimated satisfaction 1-10 based on tone

      Be accurate. Angry or frustrated customers get lower satisfaction scores.
    yields:
      category: "string"
      urgency: "number"
      satisfaction: "number"
      summary: "string"
    can_call:
      - agent: "technical_support"
        condition: "category == 'technical'"
        call_immediately: true
      - agent: "billing_support"
        condition: "category == 'billing'"
        call_immediately: true
      - agent: "human_reviewer"
        condition: "satisfaction < 4"

  # Step 2a: Handle technical issues
  technical_support:
    name: "technical_support"
    model: "gpt-4o"
    temperature: 0.3
    max_tokens: 800
    tools: ["current_time"]
    prompt: |
      You are a technical support specialist.

      Diagnose the customer's issue and provide a step-by-step solution.
      If you cannot resolve it, set resolved to false for human escalation.
    yields:
      solution: "string"
      resolved: "boolean"
      complexity: "string"
    can_call:
      - agent: "human_reviewer"
        condition: "resolved == false"

  # Step 2b: Handle billing issues
  billing_support:
    name: "billing_support"
    model: "gpt-4o-mini"
    temperature: 0.0
    max_tokens: 500
    prompt: |
      You are a billing specialist.

      Handle billing inquiries: explain charges, process adjustments,
      resolve payment issues. For refunds over $100, set needs_approval to true.
    yields:
      resolution: "string"
      amount: "number"
      needs_approval: "boolean"
    can_call:
      - agent: "human_reviewer"
        condition: "needs_approval == true"

  # Step 3: Human escalation endpoint
  human_reviewer:
    name: "human_reviewer"
    agent_type: "human"
    is_human_powered: true
    human_interface: "api"
    human_timeout_seconds: 1800
    yields:
      decision: "string"
      notes: "string"

From 2.4.482 the store is set only in .env, never in the YAML:

# .env
LEAFMESH_AGENCY_NAME=customer_service
LEAFMESH_STORE=redis
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
REDIS_SESSION_TTL=3600

Python Implementation

Create customer_service.py:

import asyncio
from leafmesh import LeafMesh

leafmesh = LeafMesh.from_store()


# --- Triage Intelligence ---
# Function name "triage" matches the YAML agent name — auto_discover finds it

async def triage(llm_response, input_data, context):
    """Enhance triage with keyword-based classification"""

    message = input_data.get("message", "").lower()

    # Keyword-based category detection
    tech_keywords = ["error", "bug", "broken", "crash", "login", "not working"]
    billing_keywords = ["charge", "refund", "payment", "invoice", "subscription"]

    category = llm_response.get("category", "general")

    # Override LLM if keywords are strong signals
    if any(kw in message for kw in tech_keywords):
        category = "technical"
    elif any(kw in message for kw in billing_keywords):
        category = "billing"

    # Estimate satisfaction from tone
    angry_words = ["angry", "furious", "terrible", "worst", "hate"]
    frustrated_words = ["frustrated", "annoyed", "disappointed", "upset"]

    satisfaction = llm_response.get("satisfaction", 7)
    if any(w in message for w in angry_words):
        satisfaction = min(satisfaction, 2)
    elif any(w in message for w in frustrated_words):
        satisfaction = min(satisfaction, 4)

    return {
        "category": category,
        "urgency": llm_response.get("urgency", 5),
        "satisfaction": satisfaction,
        "summary": llm_response.get("summary", message[:100])
    }


# --- Technical Support Intelligence ---
# Function name "technical_support" matches the YAML agent name — auto_discover finds it

async def technical_support(llm_response, input_data, context):
    """Add diagnostic metadata to technical responses"""

    solution = llm_response.get("solution", "")

    # Assess complexity
    complex_terms = ["database", "api", "server", "configuration", "network"]
    simple_terms = ["restart", "refresh", "clear cache", "update"]

    complexity = "simple"
    if any(t in solution.lower() for t in complex_terms):
        complexity = "complex"
    elif not any(t in solution.lower() for t in simple_terms):
        complexity = "medium"

    return {
        "solution": solution,
        "resolved": llm_response.get("resolved", True),
        "complexity": complexity
    }


# --- Billing Support Intelligence ---
# Function name "billing_support" matches the YAML agent name — auto_discover finds it

async def billing_support(llm_response, input_data, context):
    """Enforce approval thresholds for billing actions"""
    import re

    message = input_data.get("message", "")
    resolution = llm_response.get("resolution", "")

    # Extract amounts mentioned
    amounts = re.findall(r'\$?(\d+\.?\d*)', message)
    amount = float(amounts[0]) if amounts else 0.0

    # Require approval for amounts over $100
    needs_approval = amount > 100

    return {
        "resolution": resolution,
        "amount": amount,
        "needs_approval": needs_approval
    }


# --- Main Entry Point ---

async def main():
    print("=== Customer Service Bot ===\n")

    await leafmesh.start()
    print("Bot started: triage -> technical/billing -> human_reviewer\n")

    # Test 1: Technical issue
    print("--- Test 1: Technical Issue ---")
    result = await leafmesh.mesh_call(
        "triage",
        input_data={"message": "My login is not working, I keep getting error 403"},
        session_id="customer_001"
    )
    print(f"Result: {result}\n")

    # Test 2: Billing issue
    print("--- Test 2: Billing Issue ---")
    result2 = await leafmesh.mesh_call(
        "triage",
        input_data={"message": "I was charged $150 twice for my subscription"},
        session_id="customer_002"
    )
    print(f"Result: {result2}\n")

    # Test 3: Angry customer
    print("--- Test 3: Angry Customer ---")
    result3 = await leafmesh.mesh_call(
        "triage",
        input_data={"message": "This is terrible! I'm furious about your service!"},
        session_id="customer_003"
    )
    print(f"Result: {result3}\n")

    # Test 4: General question
    print("--- Test 4: General Question ---")
    result4 = await leafmesh.mesh_call(
        "triage",
        input_data={"message": "What are your business hours?"},
        session_id="customer_004"
    )
    print(f"Result: {result4}\n")

    await leafmesh.stop()
    print("=== Bot stopped ===")


if __name__ == "__main__":
    asyncio.run(main())

How the Pipeline Executes

Test 1: "My login is not working, I keep getting error 403"

Triage classifies: category: "technical", urgency: 7, satisfaction: 6
can_call condition category == 'technical' triggers technical_support (call_immediately)
Technical Support diagnoses the 403 error and provides login steps
Returns resolved: true, complexity: "medium"
No can_call match — pipeline ends

Test 2: "I was charged $150 twice for my subscription"

Triage classifies: category: "billing", urgency: 6, satisfaction: 4
can_call condition category == 'billing' triggers billing_support
Billing Support processes the refund request
Intelligence function detects $150 > $100, sets needs_approval: true
can_call condition needs_approval == true routes to human_reviewer
Human Reviewer receives the case for manual approval

Test 3: "This is terrible! I'm furious about your service!"

Triage classifies: category: "general", satisfaction: 2 (angry keywords detected)
can_call condition satisfaction < 4 routes directly to human_reviewer
Human Reviewer handles the escalation

Test 4: "What are your business hours?"

Triage classifies: category: "general", urgency: 1, satisfaction: 7
No can_call conditions match — triage handles it directly
Returns general information

Expected Output

=== Customer Service Bot ===

Bot started: triage -> technical/billing -> human_reviewer

--- Test 1: Technical Issue ---
Result: {'solution': 'Try clearing your browser cache...', 'resolved': True, 'complexity': 'medium'}

--- Test 2: Billing Issue ---
Result: {'decision': '...', 'notes': '...'}   # Routed to human for $150 approval

--- Test 3: Angry Customer ---
Result: {'decision': '...', 'notes': '...'}   # Routed to human (satisfaction < 4)

--- Test 4: General Question ---
Result: {'category': 'general', 'urgency': 1, 'satisfaction': 7, 'summary': 'Business hours inquiry'}

Key Patterns

Pattern	Where
Multi-tier routing	Triage classifies, then routes to specialists
Keyword enhancement	Intelligence functions override LLM when keywords are strong signals
Threshold-based escalation	`satisfaction < 4` routes to human reviewer
Approval workflows	Billing amounts > $100 require human approval
Human-in-the-loop	`agent_type: "human"` for manual review and decisions
Mixed agent types	LLM agents for analysis, human agent for escalation

Next Steps

Data Pipeline — Programmatic agents for ETL workflows
Math Pipeline — Simpler 3-agent example with verification
Self-Healing — Automatic failure recovery
Agent Configuration — Complete YAML reference

LeafMesh — Multi-tier support with human escalation