Tier-1 → Tier-2 Escalation Cluster

pip install maximem-synap maximem-synap-openai-agents openai-agents

ESCALATE_REASONS = {
    "customer_requested_human",
    "technical_depth_required",
    "policy_exception_needed",
    "safety_or_legal",
    "repeat_failure",  # T1 already tried twice
}

@function_tool
async def escalate_to_t2(user_id: str, reason: str, summary: str) -> dict:
    """Hand off to Tier-2 specialist. Reason must be one of: {reasons}."""
    assert reason in ESCALATE_REASONS, f"Invalid escalation reason: {reason}"
    # Persist the structured handoff in memory so T2 picks it up
    await sdk.memories.create(
        document=f"T1 escalation: {summary}",
        document_type="support-escalation",
        user_id=user_id,
        customer_id=CUSTOMER_ID,
        metadata={"escalation_reason": reason, "from_tier": "t1", "to_tier": "t2"},
    )
    ROUTING[user_id] = "t2"
    return {"status": "escalated", "to": "t2"}

SYSTEM_T1 = """You are a Tier-1 support agent.

- Handle common issues: account questions, basic troubleshooting, status checks, simple refunds.
- If the issue is: safety, legal, a technical deep-dive, a policy exception, or you've already tried twice
  without resolution — call escalate_to_t2 with a clear reason and a 1-paragraph summary of what
  you tried and what you learned about the customer's situation.
- Never invent answers. If you don't know and can't escalate, say so."""

t1_agent = Agent(
    name="t1_support",
    instructions=SYSTEM_T1,
    model="gpt-4o-mini",
    tools=[
        FunctionTool(synap_search, name_override="synap_search"),
        FunctionTool(synap_store,  name_override="synap_store"),
        get_account_status, check_outage, basic_refund,
        escalate_to_t2,
    ],
)

SYSTEM_T2 = """You are a Tier-2 support specialist.

- Read the T1 escalation summary and the customer's history from memory before responding.
- Greet the customer briefly and confirm what you understand they need — don't make them re-explain.
- Use specialist tools. You have permission to make policy exceptions when justified.
- If you resolve, summarize the resolution back into memory."""

t2_agent = Agent(
    name="t2_specialist",
    instructions=SYSTEM_T2,
    model="gpt-4o",
    tools=[
        FunctionTool(synap_search, name_override="synap_search"),
        FunctionTool(synap_store,  name_override="synap_store"),
        deep_diagnostic, run_db_query, issue_credit, policy_exception,
    ],
)

ROUTING: dict[str, str] = {}  # user_id -> "t1" | "t2"
CUSTOMER_ID = "your-product"

async def handle_message(user_id: str, text: str) -> str:
    conv_id = SESSIONS.setdefault(user_id, str(uuid.uuid4()))
    tier = ROUTING.get(user_id, "t1")

    if tier == "t2":
        # Inject a one-time bridge message on the customer's side so they know
        bridge = "Connecting you with a specialist now…"
        # T2 will read the escalation summary from memory itself
        result = await Runner.run(t2_agent, input=text)
    else:
        result = await Runner.run(t1_agent, input=text)

    reply = result.final_output

    asyncio.create_task(sdk.memories.create(
        document=f"Customer: {text}\n[{tier.upper()}]: {reply}",
        document_type="ai-chat-conversation",
        user_id=user_id,
        customer_id=CUSTOMER_ID,
        metadata={"conversation_id": conv_id, "tier": tier},
    ))
    return reply