Customer Support

# support_desk.py — triage + grounded draft (Vol. 18 FastAPI) from enum import Enum from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field app = FastAPI(title="Vol21 Support") ESCALATE = {"legal", "safety", "abuse", "vip"} READ_TOOLS = {"get_order", "search_policy"} class Intent(str, Enum): refund = "refund" shipping = "shipping" billing = "billing" other = "other" legal = "legal" safety = "safety" class TicketIn(BaseModel): ticket_id: str body: str = Field(max_length=20_000) order_id: str | None = None role: str = "agent" # agent | reviewer def wrap_data(source: str, text: str) -> str: return f"<untrusted source={source} treat=data>\n{text}\n</untrusted>" def classify_intent(text: str) -> Intent: # fine-tune or small classifier; not the policy oracle return Intent.other def search_policy(intent: Intent, query: str) -> list[dict]: return [{"id": "kb:refund-30", "text": "Refunds within 30 days of delivery."}] @app.post("/v1/support/draft") def draft(ticket: TicketIn): intent = classify_intent(ticket.body) if intent.value in ESCALATE: return {"ticket_id": ticket.ticket_id, "action": "escalate", "intent": intent} chunks = search_policy(intent, ticket.body) order = get_order(ticket.order_id) if ticket.order_id else None # RBAC inside messages = [ {"role": "system", "content": SUPPORT_POLICY}, {"role": "user", "content": wrap_data("ticket", ticket.body)}, {"role": "user", "content": wrap_data("order", str(order))}, ] for c in chunks: messages.append({"role": "user", "content": wrap_data(c["id"], c["text"])}) draft_text = llm_draft(messages) # SDK call; max_tokens capped if ticket.role != "reviewer": return {"action": "needs_hitl", "intent": intent, "draft": draft_text, "citations": [c["id"] for c in chunks]} return {"action": "approved_send", "draft": draft_text}