LangGraph从零搭建客服Agent：多轮对话+工具调用全流程

来源：互联网 2026-07-04 06:32:07

基于LangGraph构建客服Agent，通过图结构管理多轮对话与工具调用流程。核心节点包括意图分类、订单查询、RAG检索和人工转接，利用条件边路由、Checkpoint保持对话记忆，实现专业高效的自动客服服务。

用 LangGraph 从零搭一个客服 Agent：多轮对话 + 工具调用全流程

这篇文章将介绍如何使用 LangGraph 从零开始搭建一个真正可用的客服 Agent，支持多轮对话和工具调用。

这个 Agent 要解决什么问题

想象一下真实的客服场景：

长期稳定更新的攒劲资源： >>>点此立即查看<<<

LangGraph从零搭建客服Agent：多轮对话+工具调用全流程

"我的订单什么时候发货？"——这类问题需要查询订单系统
"你们的退款政策是什么？"——需要检索知识库
"我要投诉！"——需要转人工处理
各种追问——上下文不能丢失

传统写法通常堆砌 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 的几个核心概念会变得清晰：

概念	在本例中的体现
State	`CustomerServiceState`，存储消息、意图、工具结果
Node	分类、查询、生成各自独立的函数
条件边	意图分类后路由到不同工具节点
Reducer	`operator.add` 保证消息历史追加不覆盖
Checkpoint	`MemorySa ver` + `thread_id` 实现多轮记忆
interrupt	人工转接时暂停流程等待介入