LangGraph Customer Support Multi-Agent Workflow

This example shows how to build a customer support workflow with CAIP Agents SDK and LangGraph.

It is based on the SDK example file:

caip-agents-sdk/examples/langgraph_customer_support.py

What This Example Demonstrates

The example builds a multi-agent customer support workflow where:

• An orchestrator routes incoming customer requests
• A billing worker handles invoices, charges, refunds, and account status
• A technical worker handles diagnostics, device issues, and support tickets
• A general worker handles simple product or policy questions
• An escalation flow pauses for human approval before continuing
• Checkpoint-backed HITL with interrupt(), get_state(), and resume()
• Structured LangGraph event streaming with astream_events() for routing/tool visibility

Workflow Shape

Customer message
    |
    v
Orchestrator
    |-- billing   --> Billing worker    --> Billing tools    --> Billing worker --> END
    |-- technical --> Technical worker  --> Technical tools  --> Technical worker --> END
    |-- general   --> General worker                                        --> END
    |-- escalate  --> Escalation gate    --> Escalation worker              --> END

The developer provides the graph topology. The SDK handles agent configuration, LLM setup, thread creation, tool wrapping, message persistence, and execution through the unified client.

Prerequisites

Create an agent in CAIP Portal and configure the required environment variables:

CAIP_API_KEY is a unified key for both Agents API and LLM API access. See CAIP API Key Authentication to obtain your key.

export CAIP_ENV=development
export CAIP_API_KEY=your_caip_api_key
export CAIP_AGENT_ID=your_agent_id
export CAIP_REGION=ROW  # Optional: ROW (default) or CN

Optional variables used by the example:

export CAIP_DEMO_MODEL=gpt-4o-mini
export CAIP_CHECKPOINTER=memory      # memory | postgres | redis
export CAIP_PG_DSN=postgresql://...  # required when CAIP_CHECKPOINTER=postgres
export CAIP_REDIS_URL=redis://...    # required when CAIP_CHECKPOINTER=redis

1. Define Domain Tools

The example starts with plain Python functions. The SDK turns these functions into LangGraph-compatible tools when they are registered.

def check_account_status(customer_id: str) -> str:
    """Return the billing status and active plan for a customer account."""
    data = {
        "C001": "Active - Plan: Premium Plus, Next billing: 2026-07-15",
        "C002": "Suspended - Payment overdue since 2026-06-01",
    }
    return data.get(customer_id, f"Account {customer_id}: Active - Standard Plan")


def process_refund(customer_id: str, amount: float, reason: str) -> str:
    """Process a refund request for the given customer and amount."""
    if amount > 500:
        return "Refund exceeds the self-service limit. Escalation is required."
    return f"Refund of {amount} approved for customer {customer_id}. Reason: {reason}"


def run_diagnostic(customer_id: str, device_type: str, symptom: str) -> str:
    """Run a remote diagnostic on the customer's device or service endpoint."""
    return f"Diagnostic complete for {device_type}. Symptom: {symptom}."

2. Group Tools by Domain

Group tools by the worker that should be allowed to use them.

BILLING_TOOLS = [check_account_status, process_refund]
TECHNICAL_TOOLS = [run_diagnostic]

This keeps each worker focused and avoids giving every tool to every node.

3. Build the Orchestrator

The orchestrator classifies each customer request into one route.

ORCHESTRATOR_PROMPT = """
You are a customer support intake router.
Classify the latest customer message into one of these labels:

- billing
- technical
- general
- escalate

Respond with exactly one word.
""".strip()

The route is returned with Command(goto=...):

target = "escalation_gate" if intent == "escalate" else f"{intent}_worker"
return Command(goto=target, update={"route": intent})

4. Add Human-in-the-loop (HITL) with an Escalation Gate

The example uses LangGraph interrupt() to pause execution when human approval is required.

def escalation_gate_node(state):
    decision = interrupt(
        {
            "type": "escalation_approval",
            "reason": "Customer requested manager escalation.",
            "options": ["approve", "decline"],
        }
    )

    if str(decision).strip().lower() in {"approve", "yes", "true"}:
        return Command(goto="escalation_worker")

    return Command(goto="general_worker", update={"route": "general"})

When the graph pauses, the application can resume it after a human decision:

result = await client.run("I want to speak to a manager now", config=checkpoint_config)

if result.is_interrupted():
    state_config = {"configurable": {"thread_id": checkpoint_config["configurable"]["thread_id"]}}
    checkpoint_state = await client.get_state(config=state_config)
    print(checkpoint_state.values.get("route"))
    result = await client.resume("approve", config=checkpoint_config)

5. Register Tools and Create the Agent

The SDK owns the LLM setup and passes the configured LLM plus registered tools to the graph factory.

from langgraph.checkpoint.memory import MemorySaver
from caip_agents_sdk import CAIPAgentsClient

caip_client = CAIPAgentsClient()
checkpointer = MemorySaver()

client = caip_client.create_agent(
    framework="langgraph",
    agent_id="your_agent_id",
    checkpointer=checkpointer,
    graph_factory=build_customer_support_graph,
)

for tool in BILLING_TOOLS + TECHNICAL_TOOLS:
    client.add_tool_plain(tool)

await client.initialize()

6. Create a Thread and Run

The example uses a CAIP thread as the persistent conversation context.

thread = await caip_client.create_thread(
    agent_id="your_agent_id",
    thread_data={"title": "Customer Support Demo Session", "status": "open"},
)

client.thread_id = thread.threadId
checkpoint_config = client.create_checkpoint_config(thread.threadId, checkpoint_ns="support")

result = await client.run(
    "[Customer C001]: Why was I charged twice this month?",
    recursion_limit=20,
    config=checkpoint_config,
)

print(result.output)

7. Stream Structured LangGraph Events

The SDK now exposes LangGraph event streaming through astream_events().

event_result = await client.astream_events(
    {"messages": [HumanMessage(content="[Customer C003]: I need a manager now.")]},
    config=checkpoint_config,
    recursion_limit=20,
)

async for event in event_result:
    if event.get("type") == "route":
        print(f"route -> {event.get('route')}")
    elif event.get("type") == "tool_start":
        print(f"tool -> {event.get('name')}")
    elif event.get("type") == "interrupt":
        print(f"interrupt -> {event.get('interrupts')}")

8. Run the Full Example

From the SDK repository:

python examples/langgraph_customer_support.py

9. Optional: Add Route Evals in Your Project

The SDK sample currently focuses on runtime orchestration, HITL, and event streaming. For production rollout, add route-level evals in your own codebase to verify orchestrator quality over time.

FUNCTIONAL_EVAL_CASES = [
    ("C001", "Why was I charged twice this month?", "billing"),
    ("C002", "My internet keeps dropping every hour.", "technical"),
    ("C001", "What are your support hours?", "general"),
    ("C003", "I want to speak to a manager now!", "escalate"),
]

The eval gate can stop execution if routing accuracy is below the expected threshold.

gate_passed, eval_report = await run_functional_eval_gate(
    client=client,
    checkpoint_config=checkpoint_config,
    min_pass_rate=0.75,
)

if not gate_passed:
    raise RuntimeError("Functional eval gate failed. Aborting execution.")

What to Learn from This Example

• Keep orchestration in the graph factory
• Register tools before initialize()
• Bind tools only to the workers that need them
• Use Command(goto=...) for routing
• Use interrupt() and resume() for HITL
• Use astream_events() when you need route/tool/interrupt visibility during execution
• Add route evals in your project pipeline before production rollout

For a simpler explanation of the same pattern, see Build a Multi-Agent Workflow.