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

使用者痛点

传统客服程序往往面临以下难题：

• 业务逻辑散落在很多 if/else 代码中，后期需求变更时极易出错且难以维护。
• 对话上下文需要手动管理。忘记保存历史会导致机器人“忘记前文”，使用者体验差。
• 工具调用与对话流程耦合紧密，新增或替换工具时必须改动大量代码。
• 人工转接往往是硬编码的分支，缺乏灵活的中断/恢复机制。

LangGraph 把整个对话抽象成一张有向图节点负责单一职责。边负责流转与条件分支，状态自动持久化，让上述痛点迎刃而解。

这个 Agent 要解决什么问题

一个真实的客服场景，使用者可能问：

• "我的订单什么时候发货？"
• "
• "我要投诉！"
• 还有各种追问…，

用传统写法就是一堆 if/else 加手动维护消息列表。今天用 LangGraph 来做。整个流程用图来描述，逻辑清晰，状态自动管理。不过，

流程设计

整个 Agent 的流程如下：

使用者输入 ↓意图分类节点
↓
┌──────────┬────────────┬──────────┐
查询工具节点 RAG检索节点 人工转接节点
└──────────┴────────────┴──────────┘
↓Claude 生成回答节点
↓满意度判断
↓ 是 → 回到意图分类
↓ 否结束 / 存档

这里有几个 LangGraph 的主要用法：条件边工具节点Checkpointinterrupt。

环境准备

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

Step ：定义 State

Status 是贯穿全图的"工作内存"，所有节点都从这里读数据、往这里写数据：

from typing import TypedDict,Annotated。Optional
from langchain_core.messages import BaseMessage
import operator
class CustomerServiceState:
# 对话历史
messages: Annotated,operator.add]
# 意图分类结果
intent这方面，Optional
# 工具查询结果
tool_result: Optional
# 是否需要人工介入
needs_human: bool
# 对话轮次
turn_count: int

Annotated 是 LangGraph 的 reducer 机制: 多轮对话中每次只追加新消息，而不是覆盖整个列表，历史记录自动保留。

Step ：意图分类节点

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage,SystemMessage
model = ChatAnthropic
def classify_intent -> dict:
"""分析使用者最新消息。判断意图类型"""
last_user_msg = ""
for msg in reversed:
if isinstance:
last_user_msg = msg.content
break
classify_prompt = f"""
分析使用者消息，返回以下之一：
- "order"：涉及订单查询、物流、发货
- "refund"：涉及退款、退货、投诉
- "product"：涉及产品功能、使用方法、价格
- "human"：使用者明确要求转人工，或情绪激动
从使用者消息来看，{last_user_msg}
只返回对应的英文标签，不要其他内容。"""
response = model.invoke])
intent = response.content.strip.lower
# 容错处理
if intent not in :
intent = "product"
return {
"intent": intent,"turn_count": state.get + 1,"needs_human": False,"tool_result": None,"messages": state。}

Step ：工具节点

订单查询节点

def query_order_tool -> dict:
"""模拟调用订单程序 API"""
# 实际项目中替换为真实 API 调用
last_msg = state.content
# 简单模拟：从消息中提取订单号
import re
order_match = re.search?',last_msg,re.IGNORECASE)
if order_match:
order_id = order_match.group
result = f"订单 {order_id} 状态：已发货，预计明天到达，快递单号：SF1234567890"
说到else，result = "未找到订单号。
请提供您的订单编号"
return {"tool_result": result}

RAG 知识库检索节点

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
# 初始化知识库
def build_knowledge_base:
docs =
embeddings = HuggingFaceEmbeddings
vectorstore = Chroma.from_texts
return vectorstore
# vectorstore = build_knowledge_base # 实际使用时取消注释
def rag_retrieval_tool -> dict:
"""从知识库检索相关信息"""
last_msg = state.content
# 实际项目中：
# results = vectorstore.similarity_search
# result = "
".join
# 这里简单模拟返回固定答案：
result = ("根据知识库：商品签收后7天内可申请无理由退款，"
"退款1-3个工作日到账。")
return {"tool_result": result}

人工转接节点

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

Step ：Claude 生成回答节点

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

Step ：满意度判断

def route_by_intent -> str:
"""意图分类后的路由"""
intent = state.get
if intent == "human":
return "human"
elif intent in :
return "order_tool"
else的观点是，return "rag_tool"
def check_conversation_end -> str:
"""判断对话是否应该结束"""
# 超过 5 轮自动结束
if state.get>= 5:
return "end"
# 已转人工则结束自动流程
if state.get:
return "end"
# 否则继续等待使用者输入
return "end"

Step ：组装 Graph

from langgraph.graph import StateGraph,END
from langgraph.checkpoint.memory import MemorySaver
builder = StateGraph
# 注册所有节点
builder.add_node
builder.add_node
builder.add_node
builder.add_node
builder.add_node
builder.add_node # 透传，仅用于路由
# 设置入口
builder.set_entry_point
# 条件边：意图分类 → 三个分支
builder.add_conditional_edges(
"classify"。route_by_intent,{
"order_tool": "order_tool","rag_tool": "rag_tool","human": "human_transfer",},)
# 三个分支都汇聚到 generate
builder.add_edge
builder.add_edge
builder.add_edge
# generate 后检查是否结束
builder.add_edge
builder.add_conditional_edges(
“check_end”,check_conversation_end,{"end": END},)
# 编译并挂载 checkpoint
memory = MemorySaver
graph = builder.compile

Step ：运行 Agent

def chat -> str:
"""单轮对话入口"""
config = {“configurable”: {“threadid”: sessionid}}
result = graph.invoke(
{
“messages”:,“needshuman”: False,“turncount”: 0,},config=config,)
# 返回最终一条 AI 消息
for msg in reversed:
if isinstance:
return msg.content
return “”

session=“user_001” print print) print print) print print)

从输出示例来看，

text 再看使用者，我的订单 ORD98765 发货了吗？说到客服，您好，订单 ORD98765 已经发货，快递单号 SF1234567890。预计明天到达 ... 从使用者来看，那大概几点能到？客服这方面，根据顺丰的配送情况。一般明天上午10点前会尝试第一次派送 ... 从客服来看，我们的退款政策是…

几个容易忽略的细节

Reducer 很关键：messages 字段使用 operator.add 而不是默认覆盖，这是多轮对话能"记住"历史的底层原因。如果不加 Annotated 每次 invoke 都会把历史消息清空。

Checkpoint 和 thread_id : 同一个 thread_id 的所有调用共享状态。切换 thread_id 就是切换 使用者 / 会话,完全 隔离。生产 环境 建议 换 成 SqliteSaver 做 持久 化 : python from langgraph.checkpoint.sqlite import SqliteSaver memory=SqliteSaver.from_conn_string 人工 转 接 用 interrupt :上面的实现是打标记，实际 产品里 配 合 interrupt_before 效果 更 好 : python graph= builder.compile 流式 输出 :直接换成 `graph.stream` 即可。适 合 做 实时 对话 界面 : python for chunk in graph.stream( {"messages":},config=config): for node_name,output in chunk.items: if “messages” in output : print

小结

这套结构 可以直接 : 加 一个 ”情感 分析“ 节点 、 加 一个 ”敏感词过滤“ 节点 、 加 并行 的 多路 工具 调用…，老实说，图 的结构让 每一次 都 是 局部 改动，不影响 其他 节点。

参考资料 :