---
title: 使用LangGraph构建AI客服Agent：完整代码实战
date: 2026-05-21
categories: [AI, Agent, 实战教程]
tags: [LangGraph, AI Agent, 客服系统, Python, 教程]
---
使用LangGraph构建AI客服Agent：完整代码实战
> 本文首发于 Agent评测站，一个独立AI智能体工具评测平台。文章同步发布于CSDN、知乎，欢迎收藏讨论。
LangGraph是构建多步骤AI Agent的框架，核心思想是把Agent的执行流程建模成"有向图"——节点是操作，边是流转条件。
这篇文章带你在1小时内搭一个可用的客服Agent。它能理解用户意图、查订单、查库存、转人工，所有代码都能直接跑。
环境准备
# 创建项目目录
mkdir langgraph-customer-service && cd langgraph-customer-service
# Python 3.10+
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# 安装依赖
pip install langgraph langchain-openai langchain-core
需要准备一个OpenAI兼容的API Key（DeepSeek、百度千帆等也支持）。把Key配置到环境变量：
export OPENAI_API_KEY="sk-your-api-key-here"
export OPENAI_BASE_URL="https://api.deepseek.com/v1" # 用DeepSeek示例，也可以用OpenAI的
第一步：定义Agent的状态图
LangGraph的核心是StateGraph。先定义Agent运行时需要维护的状态数据结构。
# agent.py
from typing import TypedDict, List, Dict, Optional, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
import json
import operator
# ===== 定义状态 =====
class AgentState(TypedDict):
messages: Annotated[List[dict], operator.add] # 对话历史
user_id: Optional[str] # 当前用户
intent: Optional[str] # 识别到的意图
order_data: Optional[dict] # 查询到的订单数据
inventory_data: Optional[dict] # 查询到的库存数据
need_human: bool # 是否需要转人工
final_response: Optional[str] # 最终回复
# ===== 初始化大模型 =====
llm = ChatOpenAI(
model="deepseek-chat",
temperature=0.3,
max_tokens=2000,
)
这段代码定义了Agent的核心数据结构。messages存对话历史，intent存分析出的用户意图，order_data和inventory_data存业务查询结果，need_human标记是否需要转人工。
第二步：实现各个节点
节点1：意图识别
用户发来消息后，先判断他想干什么：
def analyze_intent(state: AgentState) -> dict:
"""识别用户意图"""
last_message = state["messages"][-1]["content"]
# 用大模型做意图分类
prompt = f"""分析用户消息的意图，只返回一个词：
- order_query: 查订单、物流、配送
- inventory_query: 查库存、有没有货
- complaint: 投诉、售后、退款
- human: 想找人工客服
- greeting: 打招呼
- other: 其他
用户消息：{last_message}
返回格式：{{"intent": "分类结果"}}"""
response = llm.invoke([SystemMessage(content=prompt)])
try:
result = json.loads(response.content)
intent = result.get("intent", "other")
except:
intent = "other"
print(f"[意图识别] {intent}")
return {"intent": intent}
这个节点把用户消息发给大模型做分类，返回一个结构化的意图标签。实际生产中可以用few-shot示例提高准确率。
节点2：订单查询
如果意图是查订单，调用这个节点：
def query_order(state: AgentState) -> dict:
"""查询订单信息（模拟数据）"""
# 实际项目中这里调订单系统的API
mock_orders = {
"1001": {"status": "已发货", "logistics": "顺丰快递 SF1234567890",
"eta": "2026-05-23", "items": "Hermes Agent授权 × 1"},
"1002": {"status": "待支付", "total": 3980, "create_time": "2026-05-20"},
"1003": {"status": "已完成", "delivered_at": "2026-05-15",
"items": "技术合伙人方案 × 1"},
}
last_message = state["messages"][-1]["content"]
# 让LLM从消息中提取订单号
extract_prompt = f"""从下面消息中提取订单号（只有数字部分），如果没有则返回null：
消息：{last_message}
返回格式：{{"order_id": "数字或null"}}"""
response = llm.invoke([SystemMessage(content=extract_prompt)])
try:
result = json.loads(response.content)
order_id = result.get("order_id")
except:
order_id = None
if order_id and order_id in mock_orders:
return {"order_data": mock_orders[order_id]}
else:
return {"order_data": {"error": "未找到订单号或订单不存在"}}
节点3：库存查询
def query_inventory(state: AgentState) -> dict:
"""查询库存（模拟数据）"""
mock_inventory = {
"hermes_agent": {"name": "Hermes Agent授权", "stock": 98, "price": "咨询客服"},
"edge_gateway": {"name": "边缘AI网关", "stock": 23, "price": "咨询客服"},
"consulting": {"name": "技术合伙人方案", "stock": "不限", "price": "咨询客服"},
}
last_message = state["messages"][-1]["content"]
# 让LLM从消息中提取产品名称
extract_prompt = f"""从下面消息中提取咨询的产品（关键词），如果没找到返回null：
消息：{last_message}
可用产品：hermes_agent（Hermes Agent）, edge_gateway（边缘AI网关）, consulting（技术合伙人）
返回格式：{{"product": "产品key或null"}}"""
response = llm.invoke([SystemMessage(content=extract_prompt)])
try:
result = json.loads(response.content)
product = result.get("product")
except:
product = None
if product and product in mock_inventory:
return {"inventory_data": mock_inventory[product]}
else:
return {"inventory_data": {"info": "请告诉我您想咨询哪个产品"}}
节点4：判断是否转人工
def check_human_needed(state: AgentState) -> dict:
"""判断是否需要转人工"""
intent = state.get("intent", "other")
last_message = state["messages"][-1]["content"]
# 需要转人工的情况：用户明确要求、投诉、或多次查询失败
need_human = False
if intent == "human" or intent == "complaint":
need_human = True
# 如果订单查询失败也转人工
order_data = state.get("order_data", {})
if order_data and order_data.get("error"):
# 第一次失败可以继续，第二次转人工
if len(state["messages"]) > 4:
need_human = True
return {"need_human": need_human}
节点5：生成回复
def generate_response(state: AgentState) -> dict:
"""根据所有信息生成最终回复"""
intent = state.get("intent", "other")
order_data = state.get("order_data", {})
inventory_data = state.get("inventory_data", {})
need_human = state.get("need_human", False)
# 构建上下文
context = f"意图: {intent}\n"
if order_data:
context += f"订单数据: {json.dumps(order_data, ensure_ascii=False)}\n"
if inventory_data:
context += f"库存数据: {json.dumps(inventory_data, ensure_ascii=False)}\n"
if need_human:
context += "状态: 需要转人工\n"
# 如果是转人工，直接返回提示
if need_human:
response = "您的问题已转接人工客服，请稍等，客服人员将在1分钟内接入。您也可以拨打客服热线 400-xxx-xxxx。"
return {"final_response": response}
prompt = f"""你是一个客服AI，基于以下数据回复客户。
{context}
历史对话：
{state['messages'][-3:]}
要求：
1. 回复简洁（不超过100字）
2. 语气礼貌但不过分热情
3. 如果是查询类问题，直接给结果
4. 不要编造不存在的功能"""
response = llm.invoke([
SystemMessage(content=prompt),
HumanMessage(content=state["messages"][-1]["content"])
])
return {"final_response": response.content}
第三步：搭建执行流程
把上面所有节点串起来的核心代码：
# ===== 构建图 =====
def build_agent():
workflow = StateGraph(AgentState)
# 注册节点
workflow.add_node("analyze_intent", analyze_intent)
workflow.add_node("query_order", query_order)
workflow.add_node("query_inventory", query_inventory)
workflow.add_node("check_human", check_human_needed)
workflow.add_node("generate_response", generate_response)
# 设置入口点
workflow.set_entry_point("analyze_intent")
# 定义条件边 —— 根据意图分流
def route_by_intent(state: AgentState) -> Literal["query_order", "query_inventory", "check_human"]:
intent = state.get("intent", "other")
if intent == "order_query":
return "query_order"
elif intent == "inventory_query":
return "query_inventory"
else:
return "check_human"
workflow.add_conditional_edges(
"analyze_intent",
route_by_intent,
{
"query_order": "query_order",
"query_inventory": "query_inventory",
"check_human": "check_human",
}
)
# 固定边
workflow.add_edge("query_order", "check_human")
workflow.add_edge("query_inventory", "check_human")
workflow.add_edge("check_human", "generate_response")
workflow.add_edge("generate_response", END)
# 编译
memory = MemorySaver()
return workflow.compile(checkpointer=memory)
# ===== 运行Agent =====
agent = build_agent()
def chat(user_message: str, thread_id: str = "default"):
"""发送消息给Agent并获取回复"""
config = {"configurable": {"thread_id": thread_id}}
result = agent.invoke(
{"messages": [{"role": "user", "content": user_message}]},
config
)
return result.get("final_response", "抱歉，没有理解您的问题")
# ===== 测试运行 =====
if __name__ == "__main__":
# 测试1：查订单
print("用户: 查一下我的订单1001")
print(f"Agent: {chat('查一下我的订单1001')}")
print()
# 测试2：查库存
print("用户: Hermes Agent还有货吗？")
print(f"Agent: {chat('Hermes Agent还有货吗？')}")
print()
# 测试3：转人工
print("用户: 我要投诉，转人工客服")
print(f"Agent: {chat('我要投诉，转人工客服')}")
第四步：执行效果
运行上述代码，输出如下：
用户: 查一下我的订单1001
[意图识别] order_query
Agent: 您的订单1001已发货，物流信息：顺丰快递SF1234567890，预计5月23日送达。包含商品：Hermes Agent授权 × 1。
用户: Hermes Agent还有货吗？
[意图识别] inventory_query
Agent: Hermes Agent授权当前库存充足（98件），您可以随时下单。如有需要，请访问官网了解详情。
用户: 我要投诉，转人工客服
[意图识别] complaint
Agent: 您的问题已转接人工客服，请稍等，客服人员将在1分钟内接入。您也可以拨打客服热线 400-xxx-xxxx。
进阶优化
1. 加入对话历史
LangGraph的MemorySaver已经自动保存了对话历史。修改chat函数可以支持多轮对话上下文：
def chat_with_history(user_message: str, thread_id: str = "default"):
config = {"configurable": {"thread_id": thread_id}}
# 读取历史
history = agent.get_state(config)
result = agent.invoke(
{"messages": [{"role": "user", "content": user_message}]},
config
)
return result.get("final_response", "抱歉，没有理解您的问题")
2. 添加超时保护
客服Agent不能无限制调用API：
import asyncio
from functools import wraps
def timeout_handler(seconds=10):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 在消息中注入超时检查
return func(*args, **kwargs)
return wrapper
return decorator
3. 连接真实业务系统
把mock数据换成真实API调用：
import requests
def query_order_real(state: AgentState) -> dict:
"""真实的订单查询"""
last_message = state["messages"][-1]["content"]
# 提取订单号（同上）
# ...
if order_id:
# 调订单系统API
resp = requests.get(
f"https://api.youcompany.com/orders/{order_id}",
headers={"Authorization": "Bearer YOUR_TOKEN"},
timeout=5
)
if resp.status_code == 200:
return {"order_data": resp.json()}
return {"order_data": {"error": "订单查询失败"}}
4. 人类介入模式
有些场景需要人工确认后再执行：
def human_confirmation(state: AgentState) -> Literal["confirm", "reject"]:
"""等待人工确认（生产环境通过WebSocket/消息队列实现）"""
# 这里简化处理——如果是退款类操作，挂起等待人工确认
if state.get("intent") == "complaint":
print("⚠️ [等待人工确认中...]")
return "reject" # 先不自动回复，等人
return "confirm"
完整项目结构
langgraph-customer-service/
├── agent.py # 主程序（上面所有代码）
├── requirements.txt # 依赖
├── .env # API Key配置
├── tests/
│ └── test_agent.py # 单元测试
└── README.md
总结
用LangGraph搭Agent的核心思路就三步：
- 定义状态：想清楚Agent需要维护哪些数据
- 实现节点函数：每个节点只做一件事
- 画流程图：用
add_conditional_edges定义智能路由
这个客服Agent只是一个起点。同样的模式可以套用于代码审查Agent、数据分析Agent、文档处理Agent——换节点逻辑就行，图结构基本不变。
代码已开源，可以在GitHub上搜索"langgraph-customer-service"找到完整项目。如果有问题欢迎评论区交流。