智能客服系统

本实战把前面学的 RAG、条件边、HITL 人机交互、检查点串成一个能跑的客服系统：识别意图 → 知识库问答 → 不满意转人工。完整代码用 Mock FAQ + 本地向量库，不依赖外部服务，复制即可运行。

一、需求分析

一个像样的客服系统至少要能干这几件事：

意图识别：区分用户是闲聊、咨询产品、还是投诉/转人工。
知识库问答：对咨询类问题，去 FAQ 库检索并作答。
满意度判断：回答后判断用户是否满意，不满意再尝试或转人工。
转人工（HITL）：兜底场景，把对话交给真人，等真人接入前先 interrupt。
多轮对话：用 checkpointer 记住上下文，跨轮次连贯。

二、架构设计

mermaid

flowchart TD
    S([START]) --> CLS[intent_classify 意图分类]
    CLS -- 闲聊/咨询 --> RET[retrieve_kb 检索FAQ]
    CLS -- 转人工 --> HUMAN[handoff 转人工]
    RET --> GEN[generate 回答生成]
    GEN --> SAT[satisfy_check 满意度判断]
    SAT -- 满意 --> E([END])
    SAT -- 不满意且未超阈值 --> RET
    SAT -- 多次不满意 --> HUMAN
    HUMAN --> W([interrupt 等待人工])

关键设计点：

意图分类是入口：所有消息先过意图识别，分流到不同分支。
检索-生成-满意度是循环：不满意就重新检索回答，带 retry_count 刹车。
转人工用 interrupt：进入 handoff 节点后图暂停，等人工处理后再恢复。详见 HITL 章节。

三、状态设计

客服状态要承载多轮对话和分支信号，字段较多。新手记一条：状态字段按"节点产出"来设计，每个节点对应一两个它要读/写的字段。

python

from typing import TypedDict, List, Literal
from langchain_core.messages import BaseMessage
from langchain_core.documents import Document

class CustomerServiceState(TypedDict):
    messages: List[BaseMessage]    # 对话历史（多轮）
    user_input: str                # 本轮用户输入
    intent: Literal["chat", "faq", "human"]  # 意图
    documents: List[Document]      # 检索到的 FAQ
    answer: str                    # 本轮回答
    satisfied: bool                # 用户是否满意
    retry_count: int               # 重试次数
    need_human: bool               # 是否转人工
    thread_id: str                 # 会话标识（checkpointer 用）

四、知识库准备（Mock FAQ）

用一份内置 FAQ，构造本地向量库。真实项目从数据库或工单系统加载。

python

from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

faqs = [
    "退货政策：商品签收后 7 天内可申请退货，需保持商品完好；生鲜食品不支持退货。",
    "物流时效：默认顺丰发货，江浙沪次日达，其他地区 2-3 天；偏远地区加 1-2 天。",
    "发票开具：下单时可填写发票抬头，电子发票在发货后 24 小时内发送至邮箱。",
    "会员等级：消费累计满 1000 元升银卡，满 5000 元升金卡，享专属折扣。",
    "支付方式：支持微信、支付宝、银行卡、信用卡，暂不支持货到付款。",
]
faq_docs = [Document(page_content=f) for f in faqs]
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
faq_store = InMemoryVectorStore(embeddings)
faq_store.add_documents(faq_docs)
faq_retriever = faq_store.as_retriever(search_kwargs={"k": 2})

五、节点实现

1. 意图分类节点

python

from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class Intent(BaseModel):
    intent: str = Field(description="意图：chat(闲聊) / faq(咨询) / human(转人工)")

intent_llm = llm.with_structured_output(Intent)

def intent_classify(state: CustomerServiceState) -> dict:
    """识别用户意图"""
    user_input = state["user_input"]
    history = state.get("messages", [])
    # 把最近 4 条历史拼进 prompt，提高意图判断准确率
    hist_text = "\n".join(f"{type(m).__name__}: {m.content}" for m in history[-4:])
    prompt = (
        "根据用户最新输入判断意图，分类：\n"
        "- chat：闲聊、问候\n"
        "- faq：咨询商品/订单/政策等业务问题\n"
        "- human：明确要求转人工、投诉、或连续不满\n\n"
        f"历史对话：\n{hist_text}\n\n最新输入：{user_input}\n\n只输出意图词。"
    )
    result = intent_llm.invoke(prompt)
    intent = result.intent.strip().lower()
    if intent not in ("chat", "faq", "human"):
        intent = "faq"   # 兜底按业务咨询处理
    return {"intent": intent}

2. 检索节点

闲聊和咨询都走检索（闲聊检索也可能命中冷知识，无害），转人工跳过。

python

def retrieve_kb(state: CustomerServiceState) -> dict:
    """从 FAQ 库检索相关条目"""
    docs = faq_retriever.invoke(state["user_input"])
    return {"documents": docs, "retry_count": state.get("retry_count", 0) + 1}

3. 回答生成节点

python

from langchain_core.prompts import ChatPromptTemplate

gen_prompt = ChatPromptTemplate.from_template(
    "你是某电商客服。根据 FAQ 资料友好回答用户问题，"
    "资料不足就如实说并建议转人工。回答简短口语化。\n\n"
    "FAQ：\n{context}\n\n用户问题：{question}\n回答："
)

def generate(state: CustomerServiceState) -> dict:
    """根据 FAQ 生成客服回答"""
    context = "\n".join(d.page_content for d in state["documents"]) or "（无相关FAQ）"
    resp = llm.invoke(gen_prompt.invoke({
        "context": context, "question": state["user_input"]
    }))
    return {"answer": resp.content}

4. 满意度判断节点

让 LLM 根据对话判断用户是否满意。注意：用户本轮可能还没说话，满意度判断的是"上一轮回答后用户的反应"。简单起见，这里用启发式：如果用户输入里带"不行/没用/人工/投诉"判不满意。

python

def satisfy_check(state: CustomerServiceState) -> dict:
    """判断用户对回答是否满意（启发式 + LLM 兜底）"""
    text = state["user_input"]
    # 启发式：明显的负面词直接判不满意
    neg_keywords = ["不行", "没用", "不对", "人工", "投诉", "转人工", "没解决", "还是不行"]
    if any(w in text for w in neg_keywords):
        return {"satisfied": False}

    # LLM 兜底判断
    verdict = llm.invoke(
        f"判断用户对客服回答是否满意，只回 yes 或 no。\n"
        f"用户输入：{text}\n回答：{state.get('answer','')}"
    ).content.strip().lower()
    return {"satisfied": verdict.startswith("yes")}

5. 转人工节点（HITL interrupt）

这是本实战的高光。用 interrupt 让图暂停，等人工接入。

python

from langgraph.types import interrupt, Command

def handoff(state: CustomerServiceState) -> Command:
    """转人工：暂停图，等待人工处理"""
    # 把对话摘要和原因交给人工
    summary = f"用户问题：{state['user_input']}\n已尝试回答：{state.get('answer','无')}"
    # interrupt 会暂停图执行，把 value 抛给调用方
    human_input = interrupt({
        "reason": "need_human",
        "summary": summary,
        "retry_count": state.get("retry_count", 0),
    })
    # 人工恢复后会用 Command(resume=...) 把结果传回 human_input
    return {
        "answer": f"[人工客服] {human_input}",
        "need_human": False,
    }

六、条件边与路由

python

def route_by_intent(state: CustomerServiceState) -> str:
    """按意图路由"""
    if state["intent"] == "human":
        return "handoff"
    return "retrieve_kb"   # chat / faq 都先检索

def route_after_satisfy(state: CustomerServiceState) -> str:
    """满意度判断后路由"""
    if state.get("satisfied"):
        return END
    # 不满意且重试次数未超 2 次 → 重新检索回答
    if state.get("retry_count", 0) < 2:
        return "retrieve_kb"
    # 多次不满意 → 转人工
    return "handoff"

七、组装图 + checkpointer

python

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

gb = StateGraph(CustomerServiceState)
for n, f in [
    ("intent_classify", intent_classify),
    ("retrieve_kb", retrieve_kb),
    ("generate", generate),
    ("satisfy_check", satisfy_check),
    ("handoff", handoff),
]:
    gb.add_node(n, f)

gb.add_edge(START, "intent_classify")
gb.add_conditional_edges("intent_classify", route_by_intent, {
    "handoff": "handoff", "retrieve_kb": "retrieve_kb"
})
gb.add_edge("retrieve_kb", "generate")
gb.add_edge("generate", "satisfy_check")
gb.add_conditional_edges("satisfy_check", route_after_satisfy, {
    END: END, "retrieve_kb": "retrieve_kb", "handoff": "handoff"
})
gb.add_edge("handoff", END)

# MemorySaver 支持多轮：同一 thread_id 的对话共享状态
memory = MemorySaver()
service_app = gb.compile(checkpointer=memory)

checkpointer=memory 让图按 thread_id 保存中间状态，下一轮对话能读到上一轮的 messages、retry_count 等。生产环境换成持久化检查点（SQLite/Postgres），见持久化与检查点。

八、完整可运行示例

把上面所有片段拼起来，加上测试对话。下面是完整可运行版本（省略已展示的 imports 和节点，只补主流程）：

python

# ---- 主流程：多轮对话 + 转人工演示 ----
config = {"configurable": {"thread_id": "user-001"}}

def chat(user_input: str):
    """单轮对话封装"""
    print(f"\n用户：{user_input}")
    # 注意：多轮里 messages 由外部维护，这里简化处理
    result = service_app.invoke(
        {"user_input": user_input, "messages": [], "retry_count": 0},
        config=config,
        # 转人工会 interrupt，需要特殊处理
    )
    print(f"客服：{result['answer']}")
    return result

if __name__ == "__main__":
    # 第一轮：FAQ 咨询
    chat("你们的退货政策是什么样的？")
    # 第二轮：不满意 → 触发重试或转人工
    chat("这个政策不行，我要退货怎么办？")
    # 第三轮：明确转人工
    try:
        chat("我要找人工客服！")
    except GraphInterrupt:
        print(">>> 图已暂停，等待人工接入...")

处理 interrupt

当走到 handoff 节点，interrupt 会抛 GraphInterrupt（被 service_app.invoke 内部捕获，图状态停在 handoff 节点）。检查人工接入后，用 Command(resume=...) 恢复：

python

from langgraph.types import Command
from langgraph.errors import GraphInterrupt

# 检测到 interrupt 后，人工给个回复
human_reply = "您好，我是人工客服小王，您的退货申请已受理，1 小时内会有专员联系您。"

# 恢复图执行
result = service_app.invoke(
    Command(resume=human_reply),
    config=config,
)
print("客服：", result["answer"])
# 输出：[人工客服] 您好，我是人工客服小王...

Command(resume=...) 是 LangGraph 0.2+ 恢复 interrupt 的标准方式，resume 的值会作为 interrupt(...) 的返回值进入 handoff 节点。

九、测试示例对话

跑一轮完整对话，观察状态流：

text

用户：退货政策什么样的？
[intent_classify] intent=faq
[retrieve_kb] 命中 2 条 FAQ
[generate] 回答：签收 7 天内可退，需保持完好...
[satisfy_check] satisfied=True → END
客服：签收 7 天内可退，需保持完好；生鲜不支持退货哦~

用户：这个不行，我要投诉
[intent_classify] intent=human
[handoff] interrupt → 等待人工
>>> 图已暂停，等待人工接入...
（人工接入后 Command(resume=...)）
客服：[人工客服] 您好，我是小王，已为您受理...

十、踩坑与优化方向

踩坑

意图分类不准：用户说"我要退货"可能被判 faq 也可能被判 human。解决：prompt 里给典型示例；连续 2 轮 faq 都不满意自动转 human。
interrupt 后忘记 resume：图停在 handoff，下次同 thread_id 调用会接着暂停。必须用 Command(resume=...) 推进，不能直接 invoke 新输入。
messages 维护：示例为简洁没把 answer/user_input 累积进 messages，真实多轮要写一个 reducer 把每轮对话追加进去，否则上下文丢失。参考 Reducer 归约器。
MemorySaver 不持久：进程重启会话就没了。生产用 SQLite/Postgres checkpointer。

优化方向

加工具调用：对"查订单"类问题，给客服 agent 加 query_order 工具查真实数据库，比纯 FAQ 强。
流式输出：用 service_app.stream 边生成边推给前端，体验更好。见流式输出。
接工单系统：handoff 时自动建工单，把 summary 写进工单字段，人工在工单系统里处理。
满意度模型微调：启发式 + LLM 判断有延迟，可训练一个小分类模型专门做满意度判断。

十一、小结

智能客服 = 意图分类 + FAQ 检索生成 + 满意度判断 + 转人工(HITL)，全部用 StateGraph 串成一张带循环和分支的图。
interrupt + Command(resume=...) 是实现"转人工等接入"的标准手段。
MemorySaver 让多轮对话共享状态，生产换持久化 checkpointer。
任何循环都要有 retry_count 刹车，避免不满意就无限重试。

下一篇代码分析助手换个场景，用 create_react_agent 构建一个能自主探索代码库的智能体。

智能客服系统 ​

一、需求分析 ​

二、架构设计 ​

三、状态设计 ​

四、知识库准备（Mock FAQ） ​

五、节点实现 ​

1. 意图分类节点 ​

2. 检索节点 ​

3. 回答生成节点 ​

4. 满意度判断节点 ​

5. 转人工节点（HITL interrupt） ​

六、条件边与路由 ​

七、组装图 + checkpointer ​

八、完整可运行示例 ​

处理 interrupt ​

九、测试示例对话 ​

十、踩坑与优化方向 ​

踩坑 ​

优化方向 ​

十一、小结 ​

智能客服系统

一、需求分析

二、架构设计

三、状态设计

四、知识库准备（Mock FAQ）

五、节点实现

1. 意图分类节点

2. 检索节点

3. 回答生成节点

4. 满意度判断节点

5. 转人工节点（HITL interrupt）

六、条件边与路由

七、组装图 + checkpointer

八、完整可运行示例

处理 interrupt

九、测试示例对话

十、踩坑与优化方向

踩坑

优化方向

十一、小结