Day 5：加记忆——让Agent记住对话
昨天的问题
昨天的Agent有一个致命缺陷——它是"金鱼记忆"。
你问："ORD001"
Agent回："订单状态运输中"
你接着问："那这个能退货吗？"
Agent回："请问您说的是哪个订单？"
它把上一轮对话忘得一干二净。
这不是Agent的错——目前每次调用 customer_service_agent() 都是独立的一次请求，没有任何状态被保存下来。要解决这个问题，我们需要两样东西：状态机（管理Agent的运行流程）和检查点（保存对话历史）。
今天做什么
用 LangGraph 重构Agent。LangGraph把Agent的每一步定义为"节点"，节点之间的跳转定义为"边"。
我们定义四个节点：
- 分类节点：判断用户意图（查订单？问政策？其他？）
- 订单节点：处理订单相关
- 政策节点：走RAG
- 兜底节点：处理其他
关键创新：编译时传入 MemorySaver——检查点存储器。每次对话结束，Agent的状态自动存盘。下次同一个 thread_id 进来，自动恢复之前的状态。
你需要安装
bash
pip install langgraph📁 今天的代码目录
customer_service_agent/
├── day1.py
├── day2.py
├── day3.py
├── day4.py
├── tools.py (Day4的工具)
└── day5.py ← 新增！LangGraph重写，80行python
# day5.py — Day4的工具和RAG重构为LangGraph状态机
from langchain.tools import tool
from datetime import datetime
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import FakeEmbeddings
# ===== Day4的工具和RAG（保持不变）=====
ORDERS = {
"ORD001": {"status": "运输中", "item": "无线耳机", "price": 299, "date": "2025-01-18"},
"ORD002": {"status": "已签收", "item": "手机壳", "price": 39, "date": "2025-01-10"},
"ORD003": {"status": "待发货", "item": "机械键盘", "price": 599, "date": "2025-01-21"},
}
@tool
def lookup_order(order_id: str) -> str:
order = ORDERS.get(order_id.upper())
if not order: return f"未找到订单 {order_id}。"
return f"📦 {order_id} | {order['item']} | ¥{order['price']} | {order['status']}"
@tool
def check_return(order_id: str) -> str:
order = ORDERS.get(order_id.upper())
if not order: return "订单不存在。"
if order["status"] != "已签收": return f"状态为'{order['status']}'，请先确认收货。"
days = (datetime.now() - datetime.strptime(order["date"], "%Y-%m-%d")).days
msg = f"✅ 可以退货" if days <= 7 else f"❌ 已超7天({days}天)"
return f"订单{order_id}：{msg}"
FAQ = "## 退货政策\n7天内可退货，邮费公司承担。\n## 退款流程\n申请退款后24小时审核，3-5工作日到账。"
vectorstore = Chroma.from_documents(
RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=30).create_documents([FAQ]),
FakeEmbeddings(size=128)
)
def search_kb(q):
docs = vectorstore.similarity_search(q, k=2)
return "\n".join([d.page_content for d in docs]) if docs else ""
# ===== Day4代码结束 =====
# ===== Day5新增：LangGraph状态机 =====
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
from langchain_core.messages import HumanMessage, AIMessage
import operator
# 1. 定义Agent的状态——这是LangGraph的核心
class AgentState(TypedDict):
"""
AgentState 是Agent的"记忆体"。
messages: 对话历史（每次追加，不覆盖）
intent: 当前分类结果（可选，用于路由）
"""
messages: Annotated[list, operator.add] # operator.add 表示追加而非替换
intent: str # 当前意图
# 2. 定义四个节点——每个节点是一个纯函数
def classify_node(state: AgentState) -> dict:
"""节点：分析用户消息，判断意图"""
last_msg = state["messages"][-1].content
# 判断规则
intent = "other"
for oid in ORDERS:
if oid in last_msg.upper():
intent = "order"; break
if any(w in last_msg for w in ["退货", "退款", "换货", "政策", "流程"]):
intent = "policy"
print(f" 🔍 意图识别：{intent}")
return {"intent": intent}
def handle_order_node(state: AgentState) -> dict:
"""节点：处理订单相关"""
last_msg = state["messages"][-1].content
for oid in ORDERS:
if oid in last_msg.upper():
if "退货" in last_msg or "退" in last_msg:
result = check_return.invoke(oid)
else:
result = lookup_order.invoke(oid)
return {"messages": [AIMessage(content=result)]}
return {"messages": [AIMessage(content="请提供您的订单号（如 ORD001）。")]}
def handle_policy_node(state: AgentState) -> dict:
"""节点：处理政策咨询——走RAG"""
last_msg = state["messages"][-1].content
knowledge = search_kb(last_msg)
if knowledge:
return {"messages": [AIMessage(content=f"根据公司政策：\n{knowledge}")]}
return {"messages": [AIMessage(content="正在为您查询相关政策，请稍候。")]}
def handle_other_node(state: AgentState) -> dict:
"""节点：处理一般对话"""
return {"messages": [AIMessage(content="您好！我是小智。可以帮您查订单、退货政策。有什么需要？")]}
# 3. 构建图——节点和边
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("classify", classify_node)
workflow.add_node("order", handle_order_node)
workflow.add_node("policy", handle_policy_node)
workflow.add_node("other", handle_other_node)
# 路由函数：根据意图决定下一步
def router(state: AgentState) -> str:
intent = state.get("intent", "other")
return intent if intent in ["order", "policy"] else "other"
# 条件边：分类后根据意图跳转
workflow.add_conditional_edges("classify", router, {
"order": "order",
"policy": "policy",
"other": "other"
})
# 终止边：处理完就结束
workflow.add_edge("order", END)
workflow.add_edge("policy", END)
workflow.add_edge("other", END)
# 入口点
workflow.set_entry_point("classify")
# 4. 编译——MemorySaver是记忆的关键
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
# ===== 测试多轮对话 =====
if __name__ == "__main__":
# 每个用户一个 thread_id——不同用户互不干扰
config = {"configurable": {"thread_id": "user-Alice"}}
print("=== 第一轮对话 ===")
r1 = app.invoke(
{"messages": [HumanMessage(content="ORD001")]},
config
)
print(f"👤: ORD001")
print(f"🤖: {r1['messages'][-1].content}\n")
print("=== 第二轮对话（没提订单号）===")
r2 = app.invoke(
{"messages": [HumanMessage(content="那这个能退货吗？")]},
config
)
print(f"👤: 那这个能退货吗？")
print(f"🤖: {r2['messages'][-1].content}\n")
print("=== 第三轮对话 ===")
r3 = app.invoke(
{"messages": [HumanMessage(content="退货需要什么条件？")]},
config
)
print(f"👤: 退货需要什么条件？")
print(f"🤖: {r3['messages'][-1].content}")运行
bash
python day5.py你应该看到：
=== 第一轮对话 ===
🔍 意图识别：order
👤: ORD001
🤖: 📦 ORD001 | 无线耳机 | ¥299 | 运输中
=== 第二轮对话（没提订单号）===
🔍 意图识别：order
👤: 那这个能退货吗？
🤖: 订单ORD001：状态为'运输中'，请先确认收货。
=== 第三轮对话 ===
🔍 意图识别：policy
👤: 退货需要什么条件？
🤖: 根据公司政策：7天内可退货，邮费公司承担...注意第二轮——用户没说ORD001，但Agent知道"这个"指的就是上一轮的订单。这就是检查点+thread_id的作用。
你学到了什么
LangGraph把Agent变成了"状态机"。 每个节点做一件事，边定义了流程。这让代码结构清晰——想加一个新功能？加一个节点，连一条边就行。
MemorySaver是记忆的引擎。 它把每次对话的状态序列化存起来。同样的 thread_id 进来，自动恢复历史。不同 thread_id 之间完全隔离——Alice看不到Bob的对话。这就是生产环境中"多用户会话管理"的雏形。
明天的预告
今天的工具函数是硬编码在Agent内部的。明天用MCP改造——把工具变成独立的标准服务，让任何系统都能调用。
Day 5完成。你的Agent现在有记忆了。它记得你上一轮说了什么，就像一个真正的客服一样。
