14 · 综合实战：企业级客服系统

本章把前 13 章全部串起来，端到端设计并实现一个"智能客服系统"（航空出行场景）：RAG 政策问答 + 工具查询 + 人工接管 + 评估监控。

本章你将学到

1. 从需求到架构：一个生产级 Agent 系统怎么拆模块；
2. 数据层（知识库）、检索层（混合检索 + 重排）、Agent 层（LangGraph 编排）的完整实现骨架；
3. 人工接管（interrupt + 工单）与评估监控怎么接进去；
4. 部署形态：Python Agent 服务 + Java 侧调用的分工。

前置知识

• 建议完成：第 05、06 章（RAG）→ 第 10 章（LangGraph）→ 第 12 章（工程化）→ 第 13 章（Java 落地）。

14.1 需求

场景：平台需要一个 7×24 的智能客服，处理高频重复问题（退改签政策、航班状态、行李、里程），把难问题转给人工客服。

硬性要求：

1. 政策问答必须基于知识库、可溯源（不能编造政策）；
2. 能查实时数据（航班状态、订单）；
3. 涉及操作类诉求（退票、改签）不能自动执行，转人工或审批；
4. 回答延迟 < 5s（P95）；
5. 每轮对话成本可控；
6. 模型/检索出问题时能优雅降级。

14.2 架构设计

                 ┌─────────── 接入层 ───────────┐
                 │  IM / 小程序 / App / 呼叫中心 │
                 └──────────────┬──────────────┘
                                ▼
                 ┌────────── Agent 服务(Python/FastAPI + LangGraph) ──────────┐
                 │                                                            │
                 │  意图识别 → 分派到处理链：                                  │
                 │    1. 政策问答  → 检索层（混合检索+重排）→ 生成              │
                 │    2. 实时查询  → 工具层（航班/订单查询）→ 回填 → 生成        │
                 │    3. 操作诉求  → 转人工（interrupt + 工单）                 │
                 │    4. 闲聊/兜底 → 直接回答/转人工                            │
                 └──────┬──────────────────────┬──────────────────────────┘
                        ▼                      ▼
                ┌── 数据/检索层 ──┐        ┌── 支撑层 ──┐
                │ 向量库+倒排索引 │        │ LLM 网关   │
                │ 知识库/工单库  │        │ 可观测/评估 │
                └────────────────┘        └────────────┘
                 Java 业务服务(订单/航班/会员) —— 工具调用出口

关键设计决策：

• 意图识别先行：先分派再处理，避免"一个万能 Agent 什么都干"的失控与高成本（第 11.5 节）；
• 检索层独立：混合检索 + 重排放在检索层，方便单独调优与评估（第 6 章）；
• 操作类强制转人工：安全底线（第 12.3 节）；
• Java 业务服务是工具出口：Agent 服务不直连数据库，一切业务数据走 Java 侧接口（权限、审计都在那）。

14.3 数据层：知识库建设

对应第 5 章的三个问题：数据从哪来、怎么切、怎么更新。

14.3.1 多源导入

14.3.2 双存储

• 向量库（语义检索）+ 倒排索引（关键字检索，如 ES / pgvector 的全文检索）——混合检索要两个都用（第 6.3 节）；
• 每个 chunk 带 metadata：source（来源）、category（政策/公告/工单）、updated_at（更新时间）、department（权限过滤用）。

14.3.3 增量更新

• 文档入库时按 source + chunk 计算内容 hash；
• 定时任务：新内容 → 入库；内容变更 → 更新该 chunk；下架 → 删除；
• 不要整库重建（第 5.11 节练过这个坑），量小重建也行，量大了必须增量。

14.4 检索层：混合检索 + 重排

直接复用第 6 章代码，封装成检索服务：

# search_service.py：检索层（混合检索 + 重排）
from rank_bm25 import BM25Okapi
import jieba

def hybrid_search(query: str, k: int = 20) -> list[str]:
    """向量检索 + BM25 各自取 Top-k，用 RRF 融合后返回候选（第6.3节）。"""
    vec_hits = vectorstore.similarity_search(query, k=k)              # 向量路
    bm25_scores = bm25.get_scores(list(jieba.cut(query)))             # 关键字路
    bm25_hits = [docs[i] for i in bm25_scores.argsort()[-k:][::-1]]
    return rrf_merge(vec_hits, bm25_hits)    # Reciprocal Rank Fusion


def rerank(query: str, candidates: list[str], top_n: int = 3) -> list[str]:
    """用重排模型精排（第6.4节）。开源可用 bge-reranker；调用以官方文档为准。"""
    scored = rerank_model.compute_score([[query, c] for c in candidates])
    top = sorted(zip(candidates, scored), key=lambda x: -x[1])[:top_n]
    return [t for t, _ in top]


def search_with_fallback(question: str) -> list[str]:
    candidates = hybrid_search(question, k=20)
    if not candidates:                     # 检索完全失败 → 触发降级
        return []
    return rerank(question, candidates, top_n=3)

降级策略（第 12.4.4 节）：检索返回空 → 不再走 Agent，直接回复"该问题资料库暂无相关信息，已为您转人工"，把人工兜底接上。

14.5 Agent 层：LangGraph 编排

核心是 14.1 那张架构图的"分派 + 处理链"。

# customer_service_graph.py：LangGraph 编排（骨架，完整可运行）
import os
from typing import Annotated, Literal, TypedDict

from dotenv import load_dotenv
from langchain_core.messages import AIMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.types import Command, interrupt

load_dotenv()

model = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)


class State(TypedDict):
    messages: Annotated[list, add_messages]
    intent: str                 # policy / flight / operation / chat / handoff
    contexts: list              # 检索到的资料
    ticket: str                 # 人工工单号


# ---------- 1. 意图识别（第一道分派）----------
def classify_node(state: State) -> dict:
    class Intent(TypedDict):
        intent: Literal["policy", "flight", "operation", "chat", "handoff"]

    resp = model.with_structured_output(Intent).invoke([
        ("system", "判断用户诉求类型："
                   "policy=问政策/规则；flight=查航班/订单状态；"
                   "operation=退改签/投诉等需人工或系统的操作；chat=闲聊；"
                   "其他无法判断的给 handoff。"),
        ("human", state["messages"][-1].content),
    ])
    return {"intent": resp["intent"]}


# ---------- 2. 政策问答链：检索 + 生成（RAG）----------
def policy_node(state: State) -> dict:
    question = state["messages"][-1].content
    docs = search_with_fallback(question)        # 14.4 检索层
    if not docs:
        # 检索降级：转人工
        return {"contexts": [], "intent": "handoff",
                "messages": [AIMessage(content="该问题资料库暂无相关信息，已为您转人工客服。")]}
    context = "\n".join(f"[{i}] {d}" for i, d in enumerate(docs, 1))
    reply = model.invoke([
        ("system", "只根据资料回答，答不出就说明资料中没有。末尾标注依据编号。"),
        ("system", f"资料：\n{context}"),
        ("human", question),
    ]).content
    return {"contexts": docs, "messages": [AIMessage(content=reply)]}


# ---------- 3. 实时查询链：调工具（骨架，完整实现见第10章）----------
def flight_node(state: State) -> dict:
    # 用 create_react_agent 挂 get_flight_status / get_booking 等工具
    reply = model.invoke(state["messages"]).content   # 示意；生产接工具循环
    return {"messages": [AIMessage(content=reply)]}


# ---------- 4. 操作类诉求：人工接管（interrupt + 工单）----------
def handoff_node(state: State) -> dict:
    """操作/复杂诉求：生成工单摘要 → interrupt 暂停 → 人工接手。"""
    summary = model.invoke([
        ("system", "用一段话概括用户诉求、航班号/订单号、期望结果，用于转人工。"),
        ("human", state["messages"][-1].content),
    ]).content

    # 暂停图：等待人工系统确认"工单已创建并接手"
    decision = interrupt({"type": "handoff", "summary": summary})

    if decision == "created":
        ticket_no = f"T{abs(hash(summary)) % 100000:05d}"
        return {"ticket": ticket_no,
                "messages": [AIMessage(content=f"已为您转接人工客服（工单号 {ticket_no}），请留意消息。")]}
    return {"messages": [AIMessage(content="您的诉求已记录，稍后人工客服会与您联系。")]}


def chat_node(state: State) -> dict:
    reply = model.invoke(state["messages"]).content
    return {"messages": [AIMessage(content=reply)]}


# ---------- 5. 构图与分派 ----------
def route_by_intent(state: State) -> str:
    return state["intent"]   # policy / flight / operation / chat / handoff


builder = StateGraph(State)
builder.add_node("classify", classify_node)
builder.add_node("policy", policy_node)
builder.add_node("flight", flight_node)
builder.add_node("handoff", handoff_node)
builder.add_node("chat", chat_node)

builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify",
    route_by_intent,
    {"policy": "policy", "flight": "flight",
     "operation": "handoff", "handoff": "handoff", "chat": "chat"},
)
builder.add_edge("policy", END)
builder.add_edge("flight", END)
builder.add_edge("handoff", END)
builder.add_edge("chat", END)

# 会话记忆 + 检查点（第9.6节）
graph = builder.compile(checkpointer=MemorySaver())


def chat(thread_id: str, message: str) -> str:
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke({"messages": [HumanMessage(content=message)]}, config=config)
    return result["messages"][-1].content

人工接管流程：

用户提"我要退票" → classify=operation → handoff_node
→ 生成工单摘要 → interrupt 暂停（等人工系统）
→ 人工系统确认 → Command(resume="created") 恢复 → 返回工单号

生产化时，interrupt 的审批走独立接口（人工客服工作台），审批动作记录操作人（第 12.6 节）。

14.6 评估与监控

14.6.1 离线评估（每次发布前跑）

# eval_runner.py：测试集 + LLM-as-judge（第12.1节）+ 检索指标（第6.7节）
EVAL_SET = [
    {"q": "航班起飞前4小时退票手续费？", "expect": ["40%"], "relevant": ["起飞前4小时内退票"]},
    {"q": "我的 CA1501 延误了吗？", "expect": ["延误", "准点", "起飞"], "relevant": ["航班状态"]},
    # 覆盖：policy / flight / operation / 边界（知识库没有的问题）...
]

def run_eval():
    for case in EVAL_SET:
        intent = classify(case["q"])
        answer = chat("eval-user", case["q"])
        hit_rate, mrr = retrieval_metrics(case)     # 第6.7节
        score = judge_answer(case["q"], "", answer) # 第12.1.3节
        print(case["q"], "| intent:", intent, "| judge:", score, "| hit:", hit_rate)

挂 CI：改 Prompt / 改切分 / 换模型，必须看指标 diff。

14.6.2 在线监控

14.7 部署形态：Java 侧怎么接入

Agent 服务（Python/FastAPI）独立部署，Java 业务系统通过 HTTP 调用（第 12.5.1 节）。

# app.py：FastAPI 包装（第12.5.1节）
from fastapi import FastAPI
from pydantic import BaseModel

from customer_service_graph import chat

app = FastAPI()

class ChatReq(BaseModel):
    thread_id: str
    message: str

@app.post("/chat")
def handle_chat(req: ChatReq):
    return {"reply": chat(req.thread_id, req.message)}

Java 侧：

// Java：Feign 客户端调用 Agent 服务
@FeignClient(name = "agent-service", url = "${agent.service.url}")
public interface AgentClient {
    @PostMapping("/chat")
    ChatResp chat(@RequestBody ChatReq req);   // ChatReq{threadId, message}
}

// 使用：把用户消息 + 会话 id 传过去
ChatResp resp = agentClient.chat(new ChatReq(sessionId, userMessage));

• 会话亲和：Agent 服务多副本时，同一 thread_id 路由到同一副本（或用共享 Checkpointer，第 12.5.3 节）；
• 流式：前端想要打字机效果，Agent 服务用 SSE 流式返回，Java 侧 WebClient 透传即可；
• 人工接管联动：interrupt 触发的工单摘要由 Agent 服务回调 Java 的工单接口，人工处理完再 resume——两个系统的交接点要提前约定接口。

14.8 上线清单（对照第 12.6 节）

• [ ] 意图识别在测试集上准确率 ≥ 90%（否则加 few-shot / 换模型）
• [ ] 政策问答全部有引用编号，抽样人工核对无编造
• [ ] 检索降级路径已打通（检索空 → 转人工）
• [ ] 退票/改签/投诉 100% 走人工接管，含审计日志
• [ ] 工具调用（航班/订单查询）限流 + 超时 + 幂等
• [ ] 会话亲和 / 共享 Checkpointer 已配置
• [ ] CI 跑离线评估，指标有历史基线
• [ ] 监控看板：QPS / P95 / 转人工率 / 成本 / 工具失败率
• [ ] 模型或 LLM 网关故障时，返回预设兜底话术
• [ ] 数据合规：日志脱敏、检索按部门过滤

14.9 演进路线

1. V1（MVP）：意图识别 + 政策 RAG + 转人工——覆盖 80% 高频问答；
2. V2：接航班/订单工具，实时查询自助化；混合检索 + 重排提升检索质量；
3. V3：操作类自动化（审批通过后自动退改），会话长期记忆（第 11.2 节）；
4. V4：多 Agent 分工（政策/订单/投诉专员，第 10.5 节）+ 主动营销触达（注意合规）。

14.10 小结

• 生产级 Agent 系统的骨架是分层的：接入层 → 意图识别 → 处理链 → 检索/工具 → 人工兜底 → 评估监控；
• 意图识别先行把复杂问题拆散，每个分支用最合适、最便宜的处理方式——这是控成本、保可控的关键；
• 检索层独立（混合检索 + 重排 + 降级）让"答得准不准"可以被单独评估和调优；
• 操作类诉求强制人工是安全底线，interrupt 让"机器决策 + 人做决定"落地；
• Java + Python 分工：复杂编排用 Python/LangGraph，业务数据与交易留在 Java 侧，接口约定清晰即可。

到这里，从 LLM 心智模型到生产级系统的整条路线你已经完整走过了。接下来就是把第 13 章的 Java 示例搬进你的工程，从一个小场景（比如把《业务流程与数据流转说明.md》变成知识库）练起。

练习

1. 用你团队的任意一份真实文档做知识库，跑通 14.5 的 policy 分支，问 5 个业务问题并记录回答质量。
2. 给 handoff_node 的 interrupt 写一个"审批确认"的测试：先 invoke 停在暂停点，再 Command(resume="created") 恢复，验证工单号能返回。
3. 用 14.6 的 run_eval 跑 20 条样本，把转人工率偏高的意图挑出来，分析是意图识别问题还是资料缺失。
4. 画一张你们团队接入这套系统的部署图：Agent 服务放哪、Java 服务怎么调、工单系统怎么对接。

参考答案

<details>
<summary>练习1（真实文档跑通 policy 分支）</summary>

用团队真实文档建知识库（复用第 5/6 章代码），跑 14.5 的 policy_node：

1. 灌库：文档切分 → Embedding → 入库，metadata 记 source（第 14.3.2 节）；

2. 问 5 个业务问题，逐个记录两件事：

   • 回答质量：是否完整、有没有编造、引用编号是否真实存在（抽查对应 chunk）；
   • 检索质量：打印 search() 返回的相似度分数，看该命中的是否排在前列；
3. 归因：答错时先分"检索没召回"还是"召回了对的但生成歪了"（第 6 章归因表），再对症下药。

一份可复用的记录表：

验收标准：5 个问题全部"有据可查"，至少 4 个回答完整正确，即可作为 V1 基线进入 14.6 评估。

</details>

<details>
<summary>练习2（interrupt 审批测试）</summary>

手动验证"invoke 暂停 → resume 恢复"：

config = {"configurable": {"thread_id": "test-handoff-001"}}

# 1) 第一次 invoke：走到 interrupt 处停下，返回暂停信息
result = graph.invoke(
    {"messages": [("human", "我要退票")]}, config=config)
print(result)                    # __interrupt__ 里能看到 {type: handoff, summary: ...}
assert "__interrupt__" in result   # 确认真的停在人工介入点

# 2) 用 Command(resume="created") 恢复，模拟人工系统确认已接手
result2 = graph.invoke(Command(resume="created"), config=config)
reply = result2["messages"][-1].content
print(reply)                     # 应包含工单号 Txxxxx
assert "工单号" in reply           # 验证工单号确实返回了

# 3) 反向验证：resume 别的值走另一分支
# Command(resume="rejected") -> 返回"您的诉求已记录..."兜底文案

要点：必须用同一个 thread_id（checkpoint 在 thread 维度）；Command(resume=...) 传入的值就是 interrupt() 的返回值，所以测试里传 "created" 恰好对应 decision == "created" 分支——写测试时把"暂停点、恢复值、预期分支"三件事一一对应，就是生产审批联调的缩略版。
</details>

<details>
<summary>练习3（run_eval 挑出转人工率偏高的意图）</summary>

跑 20 条样本，按意图分组统计：

from collections import defaultdict

intent_stats = defaultdict(lambda: {"total": 0, "handoff": 0})

for case in EVAL_SET[:20]:
    intent = classify(case["q"])                  # 14.6.1 的 classify
    answer = chat("eval-user", case["q"])
    handoff = "转人工" in answer or "暂无相关信息" in answer
    intent_stats[intent]["total"] += 1
    if handoff:
        intent_stats[intent]["handoff"] += 1

for intent, s in intent_stats.items():
    rate = s["handoff"] / s["total"]
    print(intent, "转人工率:", f"{rate:.0%}", f"({s['handoff']}/{s['total']})")

归因两步（以 policy 意图转人工率偏高为例）：

1. 资料缺失：抽查被转人工的样本，检索返回空（第 14.4 降级触发）——说明知识库没覆盖这类问题 → 补文档，不是改模型；
2. 意图识别错：检索有货但 intent 判错（比如把"我要退票"判成 policy 而非 operation）→ 是分类问题 → 加 few-shot、调 classifier 的 Prompt（第 3 章）。

判断口诀："检索空 = 资料问题，检索有货但分流错 = 意图问题"。这一步把"转人工率高"从笼统的运营指标拆成了可执行的改什么。
</details>

<details>
<summary>练习4（部署图）</summary>

参考画法（对照 14.7，结合你们团队）：

                      ┌──────────────────────────────┐
   前端/小程序 ──────►│  API 网关 (Java 侧)            │
                      └──────────────┬───────────────┘
                                     ▼
                       ┌──────────────────────────┐
                       │  Agent 服务 (Python)      │  ★ 新部署
                       │  FastAPI /chat            │
                       │  LangGraph + Checkpointer │  ← 多副本 + 共享 PG(会话状态)
                       │  会话亲和 / 或共享存储      │
                       └──────┬─────────┬─────────┘
                              │ HTTP     │ HTTP
                              ▼         ▼
              ┌─────────────────┐   ┌─────────────────┐
              │ 检索层           │   │ 业务服务 (Java)  │
              │ 向量库+ES/Rerank │   │ 订单/航班/会员   │
              └─────────────────┘   └───────┬─────────┘
                                            ▼
                              ┌──────────────────────┐
                              │ 工单/审批系统 (Java)   │  ← interrupt 回调
                              │ 人工工作台 + 审批接口   │     人工处理后 resume
                              └──────────────────────┘

要标清楚三件事：

1. Agent 服务放哪：独立服务（Python），与 Java 业务服务同 VPC 内网互通，对外只暴露 /chat（经网关）；
2. Java 怎么调：Feign 客户端 POST /chat {threadId, message}（14.7 代码），多副本时同一 thread_id 路由到同一副本或用共享 Checkpointer（12.5.3）；
3. 工单怎么对接：Agent 的 handoff_node 调 Java 的工单创建接口 → 人工在工单系统处理 → 处理完回调 Agent 的 resume 接口恢复图。这个交接点要画成双向箭头，并约定超时与重试——它跨两个系统，是最容易断的一环。
   </details>

延伸阅读

• 全系列目录：00-导读
• 检索质量：第 06 章 | 人工介入：第 10 章 | 工程化清单：第 12 章 | Java 落地：第 13 章