把智能客服做成可控 Agent：从意图识别、工具调用到人工接管

用户消息
  -> 意图识别与信息抽取
  -> 置信度/风险判断
  -> 会话状态机
  -> 白名单工具调用
  -> 结果校验
  -> 回复生成或人工接管
  -> 全链路审计

{
  "session_id": "s-1024",
  "intent": "refund",
  "slots": {
    "order_id": "OD20250308001",
    "reason": null
  },
  "stage": "collecting_information",
  "risk_level": "medium",
  "tool_history": []
}

from dataclasses import dataclass, field
from typing import Any, Callable
import json
import re


@dataclass
class Session:
    session_id: str
    slots: dict[str, Any] = field(default_factory=dict)
    events: list[dict[str, Any]] = field(default_factory=list)

    def record(self, event: str, **data: Any) -> None:
        self.events.append({"event": event, **data})


def query_order(order_id: str) -> dict[str, str]:
    # 示例数据；生产环境应调用只读订单服务。
    return {
        "order_id": order_id,
        "status": "shipped",
        "message": "订单已发出，正在运输中"
    }


TOOLS: dict[str, Callable[..., dict[str, str]]] = {
    "query_order": query_order,
}


def classify_intent(text: str) -> tuple[str, float]:
    if any(word in text for word in ("物流", "快递", "到哪", "订单状态")):
        return "order_query", 0.92
    if any(word in text for word in ("退款", "退货", "不要了")):
        return "refund", 0.88
    return "unknown", 0.35


def extract_order_id(text: str) -> str | None:
    match = re.search(r"OD\d{8,}", text.upper())
    return match.group(0) if match else None


def handoff(session: Session, reason: str) -> dict[str, Any]:
    session.record("handoff", reason=reason)
    return {
        "action": "handoff",
        "reply": "这个问题需要人工进一步处理，我已整理当前对话信息。",
        "reason": reason,
    }


def handle(session: Session, text: str) -> dict[str, Any]:
    intent, confidence = classify_intent(text)
    session.record("intent_detected", intent=intent, confidence=confidence)

    if confidence < 0.70:
        return handoff(session, "low_intent_confidence")

    order_id = extract_order_id(text) or session.slots.get("order_id")
    if order_id:
        session.slots["order_id"] = order_id

    if intent == "refund":
        # 退款会改变业务状态，示例中不允许 Agent 直接执行。
        return handoff(session, "state_changing_operation")

    if intent == "order_query" and not order_id:
        return {
            "action": "ask_user",
            "reply": "请提供以 OD 开头的订单号，我来查询物流状态。"
        }

    tool = TOOLS["query_order"]
    result = tool(order_id=order_id)
    session.record("tool_called", tool="query_order", result=result)
    return {"action": "reply", "reply": result["message"], "data": result}


if __name__ == "__main__":
    session = Session(session_id="demo-001")
    for message in ("帮我看看物流，订单 OD20250308001", "我想退款"):
        response = handle(session, message)
        print(json.dumps(response, ensure_ascii=False, indent=2))
    print(json.dumps(session.events, ensure_ascii=False, indent=2))

python agent.py