生产级客服 Agent · Agent Atlas

客服 Agent 的难点不是「生成一段回复」，而是政策、订单、审批、幂等和审计。这个教程用 FastAPI 做一个能真正跑起来的最小生产版本。

最终形态

bash

uvicorn app.main:app --reload

接口：

方法	路径	作用
POST	`/tickets`	创建工单，分类、检索、草拟
GET	`/tickets/{id}`	查看工单、草稿、风险标记
POST	`/tickets/{id}/approve`	人工批准/修改/拒绝
GET	`/tickets/{id}/events`	SSE 实时进度
GET	`/health`	健康检查

流程：

text

接入 → 分类 → 检索政策/订单 → 草拟 → 规则校验
     → 低风险自动发送 / 高风险等待审批 → 幂等写回 → 审计

项目结构

text

support/
├─ app/
│  ├─ main.py         # FastAPI
│  ├─ db.py           # SQLite
│  ├─ llm.py          # 模型封装
│  ├─ policies.py     # 政策检索
│  ├─ service.py      # 业务编排
│  └─ schemas.py      # Pydantic
├─ tests/test_flow.py
└─ requirements.txt

依赖：

bash

pip install fastapi uvicorn pydantic openai pytest httpx

Step 1：数据库

app/db.py：

python

from __future__ import annotations

import json
import sqlite3
import time
from pathlib import Path

DB_PATH = Path("support.db")


def conn() -> sqlite3.Connection:
    c = sqlite3.connect(DB_PATH, check_same_thread=False)
    c.row_factory = sqlite3.Row
    return c


def init() -> None:
    with conn() as c:
        c.executescript(
            """
            CREATE TABLE IF NOT EXISTS tickets (
              id TEXT PRIMARY KEY,
              customer_id TEXT NOT NULL,
              message TEXT NOT NULL,
              category TEXT,
              priority TEXT,
              status TEXT NOT NULL,
              draft TEXT,
              risk TEXT,
              created_at REAL NOT NULL,
              updated_at REAL NOT NULL
            );
            CREATE TABLE IF NOT EXISTS events (
              id INTEGER PRIMARY KEY AUTOINCREMENT,
              ticket_id TEXT NOT NULL,
              type TEXT NOT NULL,
              payload TEXT NOT NULL,
              created_at REAL NOT NULL
            );
            CREATE TABLE IF NOT EXISTS refunds (
              idempotency_key TEXT PRIMARY KEY,
              ticket_id TEXT NOT NULL,
              amount REAL NOT NULL,
              refund_id TEXT NOT NULL,
              created_at REAL NOT NULL
            );
            """
        )


def create_ticket(ticket_id: str, customer_id: str, message: str) -> None:
    now = time.time()
    with conn() as c:
        c.execute(
            "INSERT INTO tickets (id, customer_id, message, status, created_at, updated_at) VALUES (?,?,?,?,?,?)",
            (ticket_id, customer_id, message, "received", now, now),
        )


def update_ticket(ticket_id: str, **fields) -> None:
    fields["updated_at"] = time.time()
    keys = ", ".join(f"{k}=?" for k in fields)
    with conn() as c:
        c.execute(f"UPDATE tickets SET {keys} WHERE id=?", [*fields.values(), ticket_id])


def get_ticket(ticket_id: str) -> dict | None:
    with conn() as c:
        row = c.execute("SELECT * FROM tickets WHERE id=?", (ticket_id,)).fetchone()
    return dict(row) if row else None


def add_event(ticket_id: str, type_: str, payload: dict) -> None:
    with conn() as c:
        c.execute(
            "INSERT INTO events (ticket_id, type, payload, created_at) VALUES (?,?,?,?)",
            (ticket_id, type_, json.dumps(payload, ensure_ascii=False), time.time()),
        )


def events_after(ticket_id: str, after_id: int = 0) -> list[dict]:
    with conn() as c:
        rows = c.execute(
            "SELECT * FROM events WHERE ticket_id=? AND id>? ORDER BY id",
            (ticket_id, after_id),
        ).fetchall()
    return [
        {"id": r["id"], "type": r["type"], "payload": json.loads(r["payload"]), "at": r["created_at"]}
        for r in rows
    ]

为什么用 SQLite：教程需要能本地一键跑通。生产可换 PostgreSQL，表结构和事务语义基本一致。

Step 2：政策检索与分类

app/policies.py：

python

POLICIES = {
    "refund_window": "退款需在签收后 7 天内申请，商品需保持完好。",
    "refund_eta": "审核通过后 1–3 个工作日原路退回。",
    "refund_limit": "单笔自动退款上限 100 元，超过需主管审批。",
    "cross_border": "跨境订单退款需额外提供清关证明，时效 7–15 个工作日。",
    "warranty": "电子类商品保修 12 个月，人为损坏不在保修范围。",
}


def search_policy(query: str) -> dict:
    q = query.lower()
    hits = [{"id": k, "text": v} for k, v in POLICIES.items() if q in k or q in v]
    return {"ok": True, "summary": f"命中 {len(hits)} 条政策", "items": hits}

app/llm.py：

python

from __future__ import annotations

import json
import os

from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    api_key=os.getenv("OPENAI_API_KEY"),
)
MODEL = os.getenv("MODEL", "gpt-4.1-mini")


def json_call(system: str, user: str, schema: dict) -> dict:
    resp = client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
        response_format={"type": "json_object"},
        temperature=0,
        max_tokens=800,
    )
    data = json.loads(resp.choices[0].message.content)
    # 教程只做最小校验，生产用 Pydantic/JSON Schema
    for key in schema.get("required", []):
        if key not in data:
            raise ValueError(f"缺少字段：{key}")
    return data

分类：

python

def classify(message: str, order: dict | None) -> dict:
    return json_call(
        system="""你是客服分类器。
只输出 JSON：category、priority、needs_human、reason、confidence。
category ∈ billing/technical/refund/other。
涉及退款、投诉、法律、账号安全时 needs_human=true。
confidence 低于 0.6 时 needs_human=true。""",
        user=f"用户消息：{message}\n订单：{json.dumps(order, ensure_ascii=False) if order else '无'}",
        schema={"required": ["category", "priority", "needs_human", "reason", "confidence"]},
    )

Step 3：草拟与规则校验

python

def draft_reply(message: str, order: dict, policies: list[dict]) -> dict:
    return json_call(
        system="""你是客服回复助手。
只根据给定订单与政策草拟回复。
要求：先结论后步骤；金额/时效必须与资料一致；
不承诺政策之外的内容；不透露内部工具或其他客户信息；
不超过 200 字。输出 JSON：reply、amount、needs_refund、risk_flags、citations。""",
        user=f"""用户：{message}
订单：{json.dumps(order, ensure_ascii=False)}
政策：{json.dumps(policies, ensure_ascii=False)}""",
        schema={"required": ["reply", "needs_refund", "risk_flags", "citations"]},
    )


def validate_draft(draft: dict, order: dict) -> list[str]:
    errors: list[str] = []
    amount = draft.get("amount")
    if draft.get("needs_refund"):
        if amount is None:
            errors.append("退款金额缺失")
        elif abs(float(amount) - float(order["paid_amount"])) > 0.01:
            errors.append("退款金额与订单实付不一致")
        if order.get("days_since_delivery", 999) > 7:
            errors.append("超过 7 天退款窗口")
    if "一定" in draft.get("reply", "") or "保证" in draft.get("reply", ""):
        errors.append("包含绝对化承诺")
    return errors

Step 4：幂等退款

python

class PaymentGateway:
    def refund(self, order_id: str, amount: float) -> dict:
        # 教程用 mock；真实实现调用支付网关
        return {"refund_id": f"RF-{order_id}-{int(amount * 100)}", "eta_days": 2}


payments = PaymentGateway()


def create_refund(ticket_id: str, amount: float) -> dict:
    key = f"ticket:{ticket_id}:refund"
    with conn() as c:
        existing = c.execute("SELECT * FROM refunds WHERE idempotency_key=?", (key,)).fetchone()
    if existing:
        return {"ok": True, "duplicate": True, "refund_id": existing["refund_id"]}

    order = get_ticket(ticket_id)
    if order is None:
        return {"ok": False, "error": "TICKET_NOT_FOUND"}
    if amount > 100:
        return {"ok": False, "error": "NEEDS_MANAGER_APPROVAL"}

    result = payments.refund(ticket_id, amount)
    with conn() as c:
        c.execute(
            "INSERT INTO refunds (idempotency_key, ticket_id, amount, refund_id, created_at) VALUES (?,?,?,?,?)",
            (key, ticket_id, amount, result["refund_id"], time.time()),
        )
    add_event(ticket_id, "refund_created", {"refund_id": result["refund_id"], "amount": amount})
    return {"ok": True, "duplicate": False, **result}

幂等键由服务端生成，不要让模型填。模型可能每次生成不同的 key，等于没有幂等。

Step 5：FastAPI 接口

app/schemas.py：

python

from pydantic import BaseModel, Field


class TicketIn(BaseModel):
    customer_id: str
    message: str = Field(min_length=1, max_length=4000)
    order_id: str | None = None


class ApproveIn(BaseModel):
    approver: str
    action: str = Field(pattern="^(approve|reject|edit)$")
    reply: str | None = None

app/main.py：

python

from __future__ import annotations

import asyncio
import json
import uuid

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse

from . import db
from .llm import classify, draft_reply
from .policies import search_policy
from .schemas import ApproveIn, TicketIn
from .service import create_refund, validate_draft

app = FastAPI(title="Support Agent")
db.init()

ORDERS = {
    "A1024": {"order_id": "A1024", "paid_amount": 299.0, "days_since_delivery": 3, "status": "delivered"},
    "B2048": {"order_id": "B2048", "paid_amount": 1200.0, "days_since_delivery": 9, "status": "delivered"},
}


@app.post("/tickets")
def create_ticket(body: TicketIn):
    ticket_id = f"T-{uuid.uuid4().hex[:8]}"
    db.create_ticket(ticket_id, body.customer_id, body.message)
    db.add_event(ticket_id, "received", {"message": body.message})

    order = ORDERS.get(body.order_id or "")
    cls = classify(body.message, order)
    db.update_ticket(ticket_id, category=cls["category"], priority=cls["priority"])
    db.add_event(ticket_id, "classified", cls)

    policies = search_policy(body.message)["items"]
    draft = draft_reply(body.message, order or {}, policies)
    errors = validate_draft(draft, order or {"paid_amount": 0, "days_since_delivery": 0})

    if errors:
        draft["risk_flags"] = list(set(draft.get("risk_flags", []) + errors))
    status = "waiting_approval" if (cls["needs_human"] or draft.get("needs_refund") or errors) else "ready"
    db.update_ticket(ticket_id, draft=json.dumps(draft, ensure_ascii=False), risk=json.dumps(errors), status=status)
    db.add_event(ticket_id, "drafted", {"status": status, "draft": draft, "errors": errors})

    return {"ticket_id": ticket_id, "status": status, "draft": draft, "errors": errors}


@app.get("/tickets/{ticket_id}")
def get_ticket(ticket_id: str):
    ticket = db.get_ticket(ticket_id)
    if not ticket:
        raise HTTPException(404, "ticket not found")
    ticket["events"] = db.events_after(ticket_id)
    return ticket


@app.post("/tickets/{ticket_id}/approve")
def approve(ticket_id: str, body: ApproveIn):
    ticket = db.get_ticket(ticket_id)
    if not ticket:
        raise HTTPException(404, "ticket not found")
    if ticket["status"] not in {"waiting_approval", "ready"}:
        raise HTTPException(409, f"invalid status: {ticket['status']}")

    draft = json.loads(ticket["draft"] or "{}")
    if body.action == "reject":
        db.update_ticket(ticket_id, status="rejected")
        db.add_event(ticket_id, "rejected", {"approver": body.approver})
        return {"ok": True, "status": "rejected"}

    if body.action == "edit":
        if not body.reply:
            raise HTTPException(400, "reply required when editing")
        draft["reply"] = body.reply

    refund_result = None
    if draft.get("needs_refund"):
        refund_result = create_refund(ticket_id, float(draft.get("amount", 0)))

    db.update_ticket(ticket_id, status="sent", draft=json.dumps(draft, ensure_ascii=False))
    db.add_event(ticket_id, "approved", {"approver": body.approver, "reply": draft["reply"], "refund": refund_result})
    return {"ok": True, "status": "sent", "reply": draft["reply"], "refund": refund_result}


@app.get("/tickets/{ticket_id}/events")
async def stream_events(ticket_id: str):
    async def gen():
        last = 0
        while True:
            events = db.events_after(ticket_id, last)
            for e in events:
                last = e["id"]
                yield f"id: {e['id']}\nevent: {e['type']}\ndata: {json.dumps(e['payload'], ensure_ascii=False)}\n\n"
            ticket = db.get_ticket(ticket_id)
            if ticket and ticket["status"] in {"sent", "rejected", "failed"}:
                break
            await asyncio.sleep(1)

    return StreamingResponse(gen(), media_type="text/event-stream")

Step 6：测试

tests/test_flow.py：

python

from fastapi.testclient import TestClient

from app import db
from app.main import app

client = TestClient(app)


def test_refund_flow_is_idempotent(monkeypatch):
    # 用固定草稿绕过真实模型
    monkeypatch.setattr("app.main.classify", lambda *a, **k: {"category": "refund", "priority": "normal", "needs_human": True, "reason": "test", "confidence": 0.9})
    monkeypatch.setattr("app.main.draft_reply", lambda *a, **k: {"reply": "可以退。", "needs_refund": True, "amount": 99.0, "risk_flags": [], "citations": ["refund_window"]})
    monkeypatch.setattr("app.main.validate_draft", lambda *a, **k: [])

    r = client.post("/tickets", json={"customer_id": "u1", "message": "退款", "order_id": "A1024"})
    assert r.status_code == 200
    tid = r.json()["ticket_id"]
    assert r.json()["status"] == "waiting_approval"

    approved = client.post(f"/tickets/{tid}/approve", json={"approver": "alice", "action": "approve"})
    assert approved.status_code == 200
    assert approved.json()["refund"]["ok"] is True

    again = client.post(f"/tickets/{tid}/approve", json={"approver": "alice", "action": "approve"})
    assert again.status_code == 409  # 已发送，不能重复审批

再补一个直接测 create_refund 两次调用的用例，断言第二次 duplicate=True 且 refund_id 相同。

Step 7：运行

bash

uvicorn app.main:app --reload

创建工单：

bash

curl -X POST localhost:8000/tickets \
  -H "content-type: application/json" \
  -d '{"customer_id":"u1","message":"订单 A1024 想退款","order_id":"A1024"}'

审批：

bash

curl -X POST localhost:8000/tickets/T-xxxx/approve \
  -H "content-type: application/json" \
  -d '{"approver":"alice","action":"approve"}'

订阅事件：

bash

curl -N localhost:8000/tickets/T-xxxx/events

生产化还缺什么

项目	教程版本	生产要求
认证	无	OAuth / 内部 SSO
数据库	SQLite	PostgreSQL + 迁移
队列	同步	异步任务 + 重试
模型	单次调用	超时、重试、路由、降级
政策	固定字典	版本化知识库 + 权限
审批	接口	工单系统 + 通知
审计	events 表	不可篡改日志
观测	事件	trace + 指标 + 告警
限流	无	用户/租户配额

验收清单

创建工单能分类、草拟、标记风险
高风险进入等待审批，不自动发送
审批支持批准/修改/拒绝
退款幂等，重复调用不重复退
SSE 能实时看到进度
测试覆盖主流程与幂等
政策与金额经过代码校验
有审计事件

常见错误

让模型决定是否退款

模型只生成建议；金额、时效、权限由代码判断。

审批只给一个「同意」按钮

审批者看不到金额、政策依据、风险标记，只能盲签。

退款不幂等

超时重试会退两次。幂等键由服务端生成。

政策写死在提示词里

更新要发版，还容易与知识库不一致。政策应检索，并带版本号。

小结

客服 Agent = 分类闸门 + 政策检索 + 草拟 + 规则校验 + 人工审批 + 幂等写回。
高风险动作不要自动执行。
审批界面要展示判断依据。
先影子运行，再建议，最后低风险自动。

下一篇进入代码场景：让 Agent 读仓库、改代码、跑测试。

#最终形态

#项目结构

#Step 1：数据库

#Step 2：政策检索与分类

#Step 3：草拟与规则校验

#Step 4：幂等退款

#Step 5：FastAPI 接口

#Step 6：测试

#Step 7：运行

#生产化还缺什么

#验收清单

#常见错误

#小结

相关阅读

最终形态

项目结构

Step 1：数据库

Step 2：政策检索与分类

Step 3：草拟与规则校验

Step 4：幂等退款

Step 5：FastAPI 接口

Step 6：测试

Step 7：运行

生产化还缺什么

验收清单

常见错误

小结