LangGraph客服Agent搭建指南：多轮对话与工具调用实战

用 LangGraph 从零构建生产级客服 Agent：多轮对话与工具调用完整实现

这篇教程深入讲解如何利用 LangGraph 从底层搭建一个支持多轮对话与工具调用的生产型客服 Agent，覆盖意图分类、订单查询、知识库检索与人工转接等真实场景。

这个 Agent 要解决什么核心痛点

设想一个典型客服场景中涉及的任务拆分：

• "我的订单什么时候发货？"——需要对接订单系统实时查询
• "你们的退款政策是什么？"——需要从知识库（FAQ/RAG）中检索
• "我要投诉！"——必须触发人工客服转接流程
• 后续追问——上下文记忆绝对不能丢失

传统实现靠层层 if/else 和手动维护消息列表，扩展性差且状态管理混乱。LangGraph 将整个流程建模为有向图，逻辑清晰，状态自动流转。

流程设计

整体流程采用管道式图结构：

用户输入
     ↓
意图分类节点（订单/退款/产品/转人工）
     ↓
┌──────────┬────────────┬──────────┐
查询工具节点  RAG检索节点  人工转接节点
└──────────┴────────────┴──────────┘
     ↓
（合并）Claude 生成回答节点
     ↓
满意度判断（继续对话？）
     ↓ 是 → 回到意图分类（循环）
     ↓ 否
结束 / 存档

该图核心利用 LangGraph 的关键特性：条件边（根据意图分支）、工具节点（订单查询/RAG）、Checkpoint（多轮记忆）以及 interrupt（人工转接中断）。

环境准备

pip install langgraph langchain-anthropic langchain-community chromadb
export ANTHROPIC_API_KEY="sk-ant-api03-..."

Step 1：定义 State

State 充当全图的“工作内存”，所有节点从中读取并写入数据。

from typing import TypedDict, Annotated, Optional
from langchain_core.messages import BaseMessage
import operator

class CustomerServiceState(TypedDict):
    # 对话历史（operator.add 追加而非覆盖）
    messages: Annotated[list[BaseMessage], operator.add]
    # 意图分类结果
    intent: Optional[str]
    # 工具查询结果
    tool_result: Optional[str]
    # 是否需要人工介入
    needs_human: bool
    # 对话轮次
    turn_count: int

Annotated[list, operator.add] 是 LangGraph 的 reducer 机制，多轮对话中只追加新消息，历史记录自动保留，这是实现持久化对话记忆的基础。

Step 2：意图分类节点

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage

model = ChatAnthropic(model="claude-sonnet-4-20250514")

def classify_intent(state: CustomerServiceState) -> dict:
    """解析用户最新消息，判定意图类型"""
    last_user_msg = ""
    for msg in reversed(state["messages"]):
        if isinstance(msg, HumanMessage):
            last_user_msg = msg.content
            break

    classify_prompt = f"""分析用户消息，返回以下之一：
- "order"：涉及订单查询、物流、发货
- "refund"：涉及退款、退货、投诉
- "product"：涉及产品功能、使用方法、价格
- "human"：用户明确要求转人工，或情绪激动

用户消息：{last_user_msg}

只返回对应的英文标签，不要其他内容。"""

    response = model.invoke([SystemMessage(content=classify_prompt)])
    intent = response.content.strip().lower()

    # 容错处理
    if intent not in ["order", "refund", "product", "human"]:
        intent = "product"

    return {"intent": intent, "turn_count": state.get("turn_count", 0) + 1}

Step 3：工具节点

订单查询节点

def query_order_tool(state: CustomerServiceState) -> dict:
    """模拟调用订单系统 API"""
    # 实际项目中替换为真实 API 调用
    last_msg = state["messages"][-1].content

    # 简单模拟：从消息中提取订单号
    import re
    order_match = re.search(r'(ORD|订单)[- ]?(\d+)', last_msg, re.IGNORECASE)
    if order_match:
        order_id = order_match.group(2)
        result = f"订单 {order_id} 状态：已发货，预计明天到达，快递单号：SF1234567890"
    else:
        result = "未找到订单号，请提供您的订单编号（如 ORD12345）"

    return {"tool_result": result}

RAG 知识库检索节点

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings

# 初始化知识库（实际项目中提前构建好）
def build_knowledge_base():
    docs = [
        "退款政策：商品签收后7天内可申请无理由退款，需保持商品完好。",
        "发货时间：工作日下单，48小时内发货；节假日顺延。",
        "退款流程：在订单页点击「申请退款」，填写原因，1-3个工作日审核。",
        "产品保修：所有产品提供1年质保，人为损坏不在范围内。",
    ]
    embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
    vectorstore = Chroma.from_texts(docs, embeddings)
    return vectorstore

# vectorstore = build_knowledge_base()
# 实际使用时取消注释

def rag_retrieval_tool(state: CustomerServiceState) -> dict:
    """从知识库检索相关信息"""
    last_msg = state["messages"][-1].content

    # 实际项目中：
    # results = vectorstore.similarity_search(last_msg, k=2)
    # result = "\n".join([doc.page_content for doc in results])
    # 这里简单模拟
    result = "根据知识库：商品签收后7天内可申请无理由退款，退款1-3个工作日到账。"

    return {"tool_result": result}

人工转接节点

def transfer_to_human(state: CustomerServiceState) -> dict:
    """标记需要人工介入"""
    return {"needs_human": True, "tool_result": "正在为您转接人工客服，预计等待时间 2-3 分钟..."}

Step 4：Claude 生成回答节点

from langchain_core.messages import AIMessage

def generate_response(state: CustomerServiceState) -> dict:
    """融合工具结果与对话历史，生成最终回答"""
    tool_result = state.get("tool_result", "")
    system_prompt = f"""你是一个专业、友好的客服助手。
{ f'参考信息：{tool_result}' if tool_result else '' }
请基于对话历史和参考信息，给出简洁、准确、有帮助的回答。如果是退款或投诉，语气要更加体贴。"""
    messages_for_llm = [SystemMessage(content=system_prompt)] + state["messages"]
    response = model.invoke(messages_for_llm)
    return {"messages": [response]}

Step 5：满意度判断（条件边逻辑）

def route_by_intent(state: CustomerServiceState) -> str:
    """意图分类后的路由"""
    intent = state.get("intent", "product")
    if intent == "human":
        return "human"
    elif intent in ["order", "refund"]:
        return "order_tool"
    else:
        return "rag_tool"

def check_conversation_end(state: CustomerServiceState) -> str:
    """判断对话是否应该结束"""
    # 超过 10 轮自动结束
    if state.get("turn_count", 0) >= 10:
        return "end"
    # 已转人工则结束自动流程
    if state.get("needs_human", False):
        return "end"
    # 否则等待用户继续输入（实际产品中这里接用户新输入）
    return "end"  # 单次调用演示中直接结束

Step 6：组装 Graph

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySa ver

builder = StateGraph(CustomerServiceState)

# 注册所有节点
builder.add_node("classify", classify_intent)
builder.add_node("order_tool", query_order_tool)
builder.add_node("rag_tool", rag_retrieval_tool)
builder.add_node("human_transfer", transfer_to_human)
builder.add_node("generate", generate_response)
builder.add_node("check_end", lambda s: s)  # 透传节点，仅用于路由

# 入口
builder.set_entry_point("classify")

# 条件边：意图分类 → 三个分支
builder.add_conditional_edges(
    "classify",
    route_by_intent,
    {
        "order_tool": "order_tool",
        "rag_tool": "rag_tool",
        "human": "human_transfer",
    }
)

# 三个分支都汇聚到 generate
builder.add_edge("order_tool", "generate")
builder.add_edge("rag_tool", "generate")
builder.add_edge("human_transfer", "generate")

# generate 之后检查是否结束
builder.add_edge("generate", "check_end")
builder.add_conditional_edges(
    "check_end",
    check_conversation_end,
    {"end": END}
)

# 编译，挂载 checkpoint（多轮记忆的关键）
memory = MemorySa ver()
graph = builder.compile(checkpointer=memory)

Step 7：运行 Agent

def chat(user_input: str, session_id: str = "default"):
    """单轮对话入口"""
    config = {"configurable": {"thread_id": session_id}}
    result = graph.invoke(
        {"messages": [HumanMessage(content=user_input)], "needs_human": False, "turn_count": 0},
        config=config
    )
    # 取最后一条 AI 消息
    for msg in reversed(result["messages"]):
        if isinstance(msg, AIMessage):
            return msg.content
    return ""

# 模拟多轮对话（同一 session_id 自动携带上文）
session = "user_001"
print("用户：我的订单 ORD98765 发货了吗？")
print("客服：", chat("我的订单 ORD98765 发货了吗？", session))
print("\n用户：那大概几点能到？")
print("客服：", chat("那大概几点能到？", session))  # 自动记住上文是 ORD98765
print("\n用户：退款政策是怎样的？")
print("客服：", chat("退款政策是怎样的？", session))

输出示例：

用户：我的订单 ORD98765 发货了吗？
客服：您好！订单 ORD98765 已经发货，快递单号 SF1234567890，预计明天到达，您可以用这个单号实时查询物流进度。

用户：那大概几点能到？
客服：根据顺丰的配送情况，一般明天上午10点前会尝试第一次派送。如果不方便，可以在快递单号页面预约派送时间。

用户：退款政策是怎样的？
客服：我们的退款政策是：商品签收后7天内可申请无理由退款，退款将在1-3个工作日内原路返回。如需申请，在订单页点击「申请退款」即可。

几个容易忽略的实现细节

Reducer 是记忆持久化的基石：messages 字段使用 operator.add 而非默认覆盖，这是多轮对话“记住”历史的核心机制。若省略 Annotated，每次 invoke 会清空历史消息。

Checkpoint 与 thread_id 绑定：同一 thread_id 的所有调用共享状态。切换 thread_id 即切换会话，完全隔离。生产环境建议改用 SqliteSa ver 实现持久化：

from langgraph.checkpoint.sqlite import SqliteSa ver
memory = SqliteSa ver.from_conn_string("./customer_service.db")

人工转接搭配 interrupt 实现暂停等待：上面实现仅打标记，实际产品中配合 interrupt_before 控制流更优雅：

graph = builder.compile(checkpointer=memory, interrupt_before=["human_transfer"])
# 执行到人工转接节点前自动暂停，等待人工确认后可继续

流式输出支持实时对话界面：直接替换为 graph.stream() 即可：

for chunk in graph.stream({"messages": [HumanMessage(content="我要退款")]}, config=config):
    for node_name, output in chunk.items():
        if "messages" in output:
            print(f"[{node_name}]", output["messages"][-1].content[:80])

小结

通过这个客服 Agent 实例，LangGraph 的核心概念一目了然：

这套架构可直接扩展：增加情感分析节点、敏感词过滤节点、并行多路工具调用等。图的特性让每次扩展都是局部修改，不影响其他节点。

参考资料

• LangGraph 官方文档
• LangGraph Human-in-the-loop 文档
• GitHub 示例仓库