Customer Support Swarm with Human Approval and Memory in 10xGraph

"""Customer support swarm with approval for large refunds and saved memory.

Triage (Gemini), billing (OpenAI) and technical support (Anthropic) hand the
customer off to each other. Refunds above the threshold pause the run and ask
you to approve in the terminal. State persists in SQLite, memory in local Qdrant.

Run from the repo root: python examples/use_cases/customer_support/main.py
"""

from __future__ import annotations

import asyncio

from agents import build_memory_store, build_swarm
from config import CHECKPOINT_DB
from dotenv import load_dotenv

from tenxgraph.core.state import Message
from tenxgraph.storage.checkpointer import SqliteCheckpointer
from tenxgraph.utils import ResponseGranularity, pending_interrupt


load_dotenv()


def ask_reviewer(request) -> dict:
    """Show the pending approval to a person and return the resume value."""
    print(f"\n[approval needed] {request.message}")
    print(f"  details: {request.value}")
    answer = input("  approve? [y/N] ").strip().lower()
    note = input("  note for the customer (optional): ").strip()
    return {"approved": answer == "y", "note": note}


async def send(app, text: str, config: dict) -> None:
    """Send one customer message and keep resuming while the graph asks for approval."""
    result = await app.ainvoke(
        {"messages": [Message.text_message(text)]},
        config,
        response_granularity=ResponseGranularity.FULL,
    )
    # A paused thread has a pending interrupt on its state. Resume the same
    # thread with the answer; the refund tool re-runs and interrupt() returns it.
    while (request := pending_interrupt(result["state"])) is not None:
        result = await app.ainvoke(
            {"resume": ask_reviewer(request), "resume_interrupt_id": request.id},
            config,
            response_granularity=ResponseGranularity.FULL,
        )

    replies = [m for m in result["messages"] if m.role == "assistant" and m.text()]
    print("assistant:", replies[-1].text() if replies else "(no reply)")


async def main() -> None:
    CHECKPOINT_DB.parent.mkdir(parents=True, exist_ok=True)
    store = build_memory_store()
    app = build_swarm(store).compile(
        checkpointer=SqliteCheckpointer(CHECKPOINT_DB),
        store=store,
    )

    customer_id = input("Customer id [C-1001]: ").strip() or "C-1001"
    # The same thread_id on a later run continues this conversation from SQLite.
    config = {"thread_id": f"support-{customer_id}", "user_id": customer_id}
    print("Type a message, or 'quit' to exit.")

    try:
        while (text := input("\nyou: ").strip()) not in {"quit", "exit"}:
            if text:
                await send(app, text, config)
    except (EOFError, KeyboardInterrupt):
        pass
    finally:
        await app.aclose()


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

"""Builds the support swarm: three agents, three companies, one shared memory store.

SwarmAgent wires peer-to-peer handoffs. Each member is an ordinary `Agent`
with its own model, tools and memory settings.
"""

from __future__ import annotations

import prompts
from config import MEMORY_COLLECTION, QDRANT_PATH
from tools import (
    issue_refund,
    lookup_customer,
    lookup_order,
    run_diagnostics,
    search_knowledge_base,
)

from tenxgraph.core.graph import Agent, ToolNode
from tenxgraph.prebuilt.agent import SwarmAgent
from tenxgraph.prebuilt.agent.swarm import SwarmMemberConfig
from tenxgraph.storage.store import (
    MemoryConfig,
    OpenAIEmbedding,
    QdrantStore,
    UserMemoryConfig,
)


def build_memory_store() -> QdrantStore:
    """Local file-based Qdrant. Embeddings come from OpenAI (needs OPENAI_API_KEY)."""
    QDRANT_PATH.mkdir(parents=True, exist_ok=True)
    return QdrantStore(
        embedding=OpenAIEmbedding(model="text-embedding-3-small"),
        path=str(QDRANT_PATH),
        collection=MEMORY_COLLECTION,
    )


def build_swarm(store: QdrantStore) -> SwarmAgent:
    # Memory is keyed by config["user_id"], so each customer only sees their own notes.
    # PRELOAD: the store is searched once per run and the hits are added to the
    # request, so the model needs no memory tool (good for a cheap triage model).
    preload = MemoryConfig(
        store=store,
        retrieval_mode="preload",
        limit=5,
        user_memory=UserMemoryConfig(),
        agent_memory=None,
    )
    # POSTLOAD: the model gets `user_memory_tool` and decides when to search
    # or remember. The specialists use this so they can write notes.
    postload = MemoryConfig(
        store=store,
        retrieval_mode="postload",
        limit=5,
        user_memory=UserMemoryConfig(),
        agent_memory=None,
    )

    triage = Agent(
        model="gemini-2.5-flash",
        system_prompt=prompts.TRIAGE,
        tool_node=ToolNode([lookup_customer]),
        memory=preload,
        reasoning_config=None,  # routing is easy; skip thinking tokens
    )
    billing = Agent(
        model="gpt-5-mini",
        system_prompt=prompts.BILLING,
        tool_node=ToolNode([lookup_order, issue_refund]),
        memory=postload,
    )
    technical = Agent(
        model="claude-sonnet-5-5",
        system_prompt=prompts.TECHNICAL,
        tool_node=ToolNode([search_knowledge_base, run_diagnostics]),
        memory=postload,
    )

    return SwarmAgent(
        members={
            "TRIAGE": SwarmMemberConfig(
                agent=triage,
                can_handoff_to=["BILLING", "TECHNICAL"],
                description="Front desk. Greets customers and routes them.",
            ),
            "BILLING": SwarmMemberConfig(
                agent=billing,
                can_handoff_to=["TECHNICAL"],
                description="Billing: charges, invoices and refunds.",
            ),
            "TECHNICAL": SwarmMemberConfig(
                agent=technical,
                can_handoff_to=["BILLING"],
                description="Technical support: errors, API problems, how-to questions.",
            ),
        },
        entry="TRIAGE",
    )

"""Support tools: customer lookup, orders, refunds with approval, and diagnostics.

`issue_refund` is the human-in-the-loop part. `interrupt()` stops the run and
saves it; the CLI shows the request to a person and resumes the same thread
with their answer.
"""

from __future__ import annotations

import json
from typing import Any

from config import REFUND_APPROVAL_THRESHOLD
from data import CUSTOMERS, KNOWLEDGE_BASE, ORDERS, REFUNDS

from tenxgraph.utils import interrupt


# The `config` parameter is filled in by 10xGraph with the run config. Reading
# the customer from `config["user_id"]` (set by the CLI, not by the model) means
# the model can never look up another customer's data.

MAX_HITS = 3
MIN_WORD = 3


def lookup_customer(config: dict[str, Any]) -> str:
    """Return the signed-in customer's profile and plan."""
    customer = CUSTOMERS.get(config.get("user_id", ""))
    return json.dumps(customer) if customer else json.dumps({"error": "unknown customer"})


def lookup_order(order_id: str, config: dict[str, Any]) -> str:
    """Look up one order by id, if it belongs to the signed-in customer."""
    order = ORDERS.get(order_id)
    if not order or order["customer_id"] != config.get("user_id"):
        return json.dumps({"error": f"order {order_id} not found"})
    return json.dumps({"order_id": order_id, **order, "refunded": REFUNDS.get(order_id, 0.0)})


def issue_refund(order_id: str, amount: float, reason: str, config: dict[str, Any]) -> str:
    """Refund part or all of an order. Large refunds need human approval."""
    order = ORDERS.get(order_id)
    if not order or order["customer_id"] != config.get("user_id"):
        return json.dumps({"error": f"order {order_id} not found"})
    if amount <= 0 or amount + REFUNDS.get(order_id, 0.0) > order["amount"]:
        return json.dumps({"error": "refund exceeds the order amount"})

    if amount > REFUND_APPROVAL_THRESHOLD:
        # On the first run this raises and pauses the graph. When the thread is
        # resumed, this whole function runs again and interrupt() returns the
        # reviewer's answer. Keep side effects (the refund) below this call.
        decision = interrupt(
            {"order_id": order_id, "amount": amount, "reason": reason},
            message=f"Approve a ${amount:.2f} refund on {order_id}?",
            reason="refund_approval",
            response_schema={
                "type": "object",
                "properties": {"approved": {"type": "boolean"}, "note": {"type": "string"}},
                "required": ["approved"],
            },
        )
        # None means the reviewer cancelled instead of answering.
        if not decision or not decision.get("approved"):
            note = (decision or {}).get("note", "no reason given")
            return json.dumps({"status": "declined", "note": note})

    REFUNDS[order_id] = REFUNDS.get(order_id, 0.0) + amount
    return json.dumps({"status": "refunded", "order_id": order_id, "amount": amount})


def search_knowledge_base(query: str) -> str:
    """Search help articles by keyword and return the best matches."""
    words = {w.lower() for w in query.split() if len(w) > MIN_WORD}
    hits = [
        article
        for article in KNOWLEDGE_BASE
        if words & set((article["title"] + " " + article["body"]).lower().split())
    ]
    return json.dumps(hits[:MAX_HITS] or [{"title": "No article found", "body": ""}])


def run_diagnostics(service: str, config: dict[str, Any]) -> str:
    """Run a health check for one of the customer's services (api, webhooks, login)."""
    # Fake result; a real version would query your monitoring system.
    status = {"api": "degraded: rate limited", "webhooks": "ok", "login": "ok"}
    return json.dumps(
        {
            "customer": config.get("user_id"),
            "service": service,
            "status": status.get(service, "unknown service"),
        }
    )

"""System prompts for the three support agents.

Each prompt is short on purpose. The handoff tools (`transfer_to_billing`,
`transfer_to_technical`, ...) are added by SwarmAgent; the prompts only say
when to use them.
"""

from __future__ import annotations


def _prompt(text: str) -> list[dict[str, str]]:
    return [{"role": "system", "content": text}]


TRIAGE = _prompt(
    "You are the front desk of a software company's support team. Greet the customer, "
    "call lookup_customer once to see their plan, then decide who should handle the "
    "request. Billing, invoices, charges and refunds go to billing. Bugs, errors, API "
    "problems and how-to questions go to technical support. Answer simple questions "
    "yourself. Do not attempt refunds or diagnostics."
)

BILLING = _prompt(
    "You are the billing specialist. Use lookup_order to check a charge before you act. "
    "Only call issue_refund for an order the customer owns, with a clear reason. "
    "Refunds above the approval threshold are reviewed by a human; if the refund tool "
    "reports it was declined, explain that politely and offer alternatives. "
    "Use user_memory_tool with action='remember' to save lasting facts, such as a "
    "preferred contact method or a past billing dispute. If the problem is technical, "
    "transfer to technical support."
)

TECHNICAL = _prompt(
    "You are the technical support engineer. Search the knowledge base first, run "
    "diagnostics when the customer reports an error, and give numbered steps. "
    "Use user_memory_tool with action='search' to recall earlier problems this customer "
    "had, and action='remember' to save what you resolved. If the problem turns out to "
    "be about charges or refunds, transfer to billing."
)

"""Fake customer, order and refund data for the support demo.

Replace these dicts with calls to your CRM or order system. Everything is in
memory, so refunds issued during a run reset when the process exits.
"""

from __future__ import annotations

from typing import Any


CUSTOMERS: dict[str, dict[str, Any]] = {
    "C-1001": {"name": "Maya Patel", "email": "[email protected]", "plan": "Pro"},
    "C-1002": {"name": "Tomas Reyes", "email": "[email protected]", "plan": "Starter"},
}

ORDERS: dict[str, dict[str, Any]] = {
    "ORD-5001": {
        "customer_id": "C-1001",
        "item": "Pro plan, annual",
        "amount": 240.00,
        "status": "paid",
        "charged_on": "2026-10-01",
    },
    "ORD-5002": {
        "customer_id": "C-1001",
        "item": "Extra seat, monthly",
        "amount": 12.00,
        "status": "paid",
        "charged_on": "2026-10-15",
    },
    "ORD-6001": {
        "customer_id": "C-1002",
        "item": "Starter plan, monthly",
        "amount": 19.00,
        "status": "paid",
        "charged_on": "2026-10-20",
    },
}

# Refunds issued during this process: order_id -> amount.
REFUNDS: dict[str, float] = {}

KNOWLEDGE_BASE: list[dict[str, str]] = [
    {
        "title": "Login loops back to the sign-in page",
        "body": "Clear cookies for the app domain, then sign in again. If SSO is on, "
        "check that the identity provider session has not expired.",
    },
    {
        "title": "API returns 429 Too Many Requests",
        "body": "Starter allows 60 requests/minute, Pro allows 600. Back off with "
        "exponential delay and honour the Retry-After header.",
    },
    {
        "title": "Webhook deliveries are delayed",
        "body": "Deliveries retry for 24 hours. Check the endpoint returns 2xx within "
        "10 seconds and is not blocking on slow work.",
    },
]

"""Settings for the support demo. Only API keys come from the environment."""

from __future__ import annotations

from pathlib import Path


HERE = Path(__file__).parent

# Refunds above this amount (USD) pause the run for human approval.
REFUND_APPROVAL_THRESHOLD = 50.0

# Conversation state is saved here, so closing the CLI and starting it again
# with the same customer id resumes the same thread.
CHECKPOINT_DB = HERE / "data" / "support_checkpoints.db"

# Local, file-based Qdrant: no Docker or server needed.
QDRANT_PATH = HERE / "data" / "qdrant"
MEMORY_COLLECTION = "support_customer_memory"