重点介绍Agent长期记忆，短期记忆使Agent能够在单次会话中维持对话的连贯性，而长期记忆则赋予了Agent跨会话学习和积累知识的能力。这样Agent能够构建持久的用户画像，形成经验库，并持续优化自身行为，从而提供真正个性化、智能化的服务。
长期记忆介绍
长期记忆是一种用于存储用户特定信息或应用级数据的系统，其核心特点是跨越会话和线程共享。与短期记忆局限于单一线程（thread_id）不同，长期记忆中的数据可以被任何时间、在任何线程中召回。其存储范围（作用域）被定义在自定义的命名空间中，而非单个线程ID内。
在LangChain框架中，长期记忆通过存储（Store）组件来实现，Store允许你将记忆保存为JSON文档，并通过命名空间（namespace）和键（Key）进行组织和管理，便于后续的检索、更新与删除。
- 命名空间（namespace）：类似于文件夹，用于对记忆进行逻辑分组（例如，按应用场景划分）。
- 键（key）：命名空间内每个文档的唯一标识符。
Store存储支持基本的put（写入）、get（读取）、delete（删除）、search（搜索）操作，如下案例中演示基于内存store存储的基本操作。
from langgraph.store.memory import InMemoryStore
# 初始化一个内存存储store
store = InMemoryStore()
# 定义命名空间：通常包含用户ID和上下文
namespace = ("user1", "preferences")
# 写入一个记忆
store.put(
namespace,
"fruit", # 键
{"likes": ["苹果", "香蕉"], "dislikes": ["橙子"]} # 值 (JSON文档)
)
print("=== 读取 fruit 记忆 ===")
# 读取记忆
memory = store.get(namespace, "fruit")
print(memory)
# 写入另一个记忆
store.put(
namespace,
"chat", # 键
{"language": "中文", "emotion": "高兴"} # 值 (JSON文档)
)
print("=== 读取 chat 记忆 ===")
# 读取记忆
memory = store.get(namespace, "chat")
print(memory)
print("=== 搜索所有记忆 ===")
# 搜索所有记忆
memories = store.search(namespace)
print(memories)
# 更新记忆
store.put(
namespace,
"fruit", # 键
{"likes": ["橘子", "葡萄"], "dislikes": ["草莓"]} # 值 (JSON文档)
)
print("=== 读取更新后的 fruit 记忆 ===")
# 读取更新后的记忆
memory = store.get(namespace, "fruit")
print(memory)
以上代码运行结果如下：
以上代码注意点如下：
- 命名空间（namespace）格式为元组格式，建议包含用户ID和上下文特征，这样的层级化命名空间可以更好隔离数据、避免命名空间冲突。
- 键（key）：在命名空间中唯一的一个key，store中put对应key的值后，后续再put相同的key的值，会覆盖当前key的值。
- 键（key）对应的value的值是一个字典结构，该字典中 k,v 内容由用户定义。
- store.put 方法可以传入命名空间（元组类型）、key（字符串类型）、value（字典类型）三个参数。
- store.get方法可以指定namespace和key参数获取对应的value值；store.search方法可以指定namespace参数获取该命名空间中所有的value。
Agent中使用长期记忆方式
长期记忆的核心作用是实现跨会话、跨节点的数据共享。在 LangChain 中，单个 Agent 通过将 Store注入到不同会话（thread_id）中，实现跨会话状态持久化。Agent中设置长期记忆的方式和短期记忆类似，长期记忆可以存在内存或者数据库中，下面重点介绍在Agent中如何使用长期记忆。
此外，长期记忆更多使用在 LangGraph 构建的工作流中，流中不同节点通过共享的 Store访问同一份长期记忆。关于LangGraph中长期记忆在后续进行介绍。
使用内存存储长期记忆
在测试环境中，通常使用内存型的 InMemoryStore 对象，这种方式简单易用但程序重启后数据会丢失。
如下代码中使用内存存储长期记忆，该案例中创建好 InMemoryStore 后，首先向该对象中存储一些数据作为长期记忆，后续在工具中获取上下文Context中用户信息，根据不同用户查询不同的长期记忆信息，实现在相同/不同的thread_id中读取相同长期记忆内容。
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, ToolRuntime
from pydantic import BaseModel
from init_llm import deepseek_llm
class UserContext(BaseModel):
user_id: str
# 1. 初始化存储并预先存入一些长期记忆
store = InMemoryStore()
checkpointer = InMemorySaver()
# 预先在长期记忆中存储一些用户信息
store.put(
("users",), # 命名空间：用户数据
"user_123", # 键：用户ID
{"name": "张三", "age": 28, "city": "北京", "hobby": "编程、阅读"} # 值：用户信息
)
store.put(
("users",),
"user_456",
{"name": "李四", "age": 32, "city": "上海", "hobby": "旅游、摄影"}
)
# 2. 定义读取长期记忆的工具
@tool
def get_user_info(runtime: ToolRuntime[UserContext]) -> str:
"""
从长期记忆中获取当前用户的信息。
Returns:
str: 用户的详细信息
"""
# 从runtime中获取store和context
store = runtime.store
user_id = runtime.context.user_id
# 从长期记忆中读取用户信息
user_data = store.get(("users",), user_id)
if user_data:
# 将字典格式化为字符串
info = user_data.value
return f"用户信息：姓名-{info['name']}, 年龄-{info['age']}, 城市-{info['city']}, 爱好-{info['hobby']}"
else:
return "未找到该用户的信息。"
# 3. 创建带有长期记忆读取工具的Agent
agent = create_agent(
model=deepseek_llm,
tools=[get_user_info],
checkpointer=checkpointer,
store=store,
context_schema=UserContext
)
# 4. 演示：在不同线程中读取相同的长期记忆
print("=== 长期记忆的跨线程读取 ===")
# 线程1：用户123询问自己的信息
print("线程1 - 用户123询问信息:")
result1 = agent.invoke(
{"messages": [{"role": "user", "content": "我的个人信息是什么？"}]},
config={"configurable": {"thread_id": "thread_1"}},
context=UserContext(user_id="user_123") # 用户ID决定访问哪个长期记忆
)
print(f"Agent回复: {result1['messages'][-1].content}\n")
# 线程2：另一个线程，相同的用户
print("线程2 - 相同的用户，不同的线程:")
result2 = agent.invoke(
{"messages": [{"role": "user", "content": "再告诉我一次我的信息"}]},
config={"configurable": {"thread_id": "thread_2"}}, # 线程ID不同
context=UserContext(user_id="user_123") # 但用户ID相同
)
print(f"Agent回复: {result2['messages'][-1].content}\n")
# 线程3：另一个用户
print("线程3 - 用户456询问信息:")
result3 = agent.invoke(
{"messages": [{"role": "user", "content": "我的信息是什么？"}]},
config={"configurable": {"thread_id": "thread_3"}},
context={"user_id": "user_456"} # 用户ID不同，访问不同的长期记忆
)
print(f"Agent回复: {result3['messages'][-1].content}")
以上代码运行结果如下：
以上代码注意：Agent创建中通过Store参数指定长期记忆对象，使用长期记忆时，可以同时使用短期记忆（checkpointer指定），也可以不使用短期记忆，但一般两者会同时使用。
使用数据库存储长期记忆
在生产环境中，推荐使用数据库支持的Store，以确保数据的持久化和多实例部署的支持。
如下案例中使用mysql数据库来存储长期记忆，完成此案例需要提前安装好mysql数据库（默认已经安装mysql8），并且进行数据库创建和安装必要python依赖，具体如下：
1) 在mysql中创建langchain_db数据库
#进入mysql navicate客户端，创建mysql数据库langchain_db
create database langchain_db;
2) 在当前python环境中安装如下依赖
#安装必要依赖
conda activate langchain_v1.2
python -m pip install langgraph-checkpoint-mysql==3.0.0 pymysql==1.1.2 cryptography==46.0.3
python -m pip install aiomysql==0.3.2 asyncmy==0.2.11
使用mysql数据库存储长期记忆代码如下：
from langchain.agents import create_agent
from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver
from langchain.tools import tool, ToolRuntime
from langgraph.store.mysql import PyMySQLStore
from pydantic import BaseModel
from utils.llm_util import deepseek_llm
from urllib.parse import quote_plus
class UserContext(BaseModel):
user_id: str
# 1. 定义读取长期记忆的工具
@tool
def get_user_info(runtime: ToolRuntime[UserContext]) -> str:
"""
从长期记忆中获取当前用户的信息。
Returns:
str: 用户的详细信息
"""
# 从runtime中获取store和context
store = runtime.store
user_id = runtime.context.user_id
# 从长期记忆中读取用户信息
user_data = store.get(("users",), user_id)
if user_data:
# 将字典格式化为字符串
info = user_data.value
return f"用户信息：姓名-{info['name']}, 年龄-{info['age']}, 城市-{info['city']}, 爱好-{info['hobby']}"
else:
return "未找到该用户的信息。"
password = quote_plus("Root#*123")
# 配置 MySQL 连接
DB_URI = (
f"mysql+pymysql://root:{password}"
"@localhost:3306/langchain_db?charset=utf8mb4"
)
with (
PyMySQLSaver.from_conn_string(DB_URI) as checkpointer,
PyMySQLStore.from_conn_string(DB_URI) as store
):
# 自动创建checkpointer数据库表（首次运行）
checkpointer.setup()
# 2. 自动创建store数据库表（首次运行）
store.setup()
# 3. 预先在长期记忆中存储一些用户信息
store.put(
("users",), # 命名空间：用户数据
"user_123", # 键：用户ID
{"name": "张三", "age": 28, "city": "北京", "hobby": "编程、阅读"}, # 值：用户信息
)
store.put(
("users",),
"user_456",
{"name": "李四", "age": 32, "city": "上海", "hobby": "旅游、摄影"},
)
# 4. 创建带有长期记忆读取工具的Agent
agent = create_agent(
model=deepseek_llm,
tools=[get_user_info],
checkpointer=checkpointer,
store=store,
context_schema=UserContext
)
# 5. 演示：在不同线程中读取相同的长期记忆
print("=== 长期记忆的跨线程读取 ===")
# 线程1：用户123询问自己的信息
print("线程1 - 用户123询问信息:")
result1 = agent.invoke(
{"messages": [{"role": "user", "content": "我的个人信息是什么？"}]},
config={"configurable": {"thread_id": "thread_1"}},
context=UserContext(user_id="user_123") # 用户ID决定访问哪个长期记忆
)
print(f"Agent回复: {result1['messages'][-1].content}\n")
# 线程2：另一个线程，相同的用户
print("线程2 - 相同的用户，不同的线程:")
result2 = agent.invoke(
{"messages": [{"role": "user", "content": "再告诉我一次我的信息"}]},
config={"configurable": {"thread_id": "thread_2"}}, # 线程ID不同
context=UserContext(user_id="user_123") # 但用户ID相同
)
print(f"Agent回复: {result2['messages'][-1].content}\n")
# 线程3：另一个用户
print("线程3 - 用户456询问信息:")
result3 = agent.invoke(
{"messages": [{"role": "user", "content": "我的信息是什么？"}]},
config={"configurable": {"thread_id": "thread_3"}},
context=UserContext(user_id="user_456") # 用户ID不同，访问不同的长期记忆
)
print(f"Agent回复: {result3['messages'][-1].content}")
以上代码运行结果如下：
运行代码后进入到Mysql数据库中可以看到对应的数据库表和数据：
以上代码需要注意如下几点：
“with PyMySQLStore.from_conn_string(DB_URI) as store:”通过连接字符串DB_URI创建与MySQL数据库的持久化连接，用于管理长期记忆的存储。
使用mysql存储长期记忆需要提前在数据库中创建对应的数据库，然后代码首次运行执行“store.setup()”（首次运行需要，首次运行后可以不再执行该代码）会自动在该数据库中创建对应数据库表。
也可以使用其他数据库进行短期记忆的持久化存储，例如使用postgresql存储，需要安装“pip install langgraph-checkpoint-postgres==3.0.4”，具体代码参考：https://docs.langchain.com/oss/python/langchain/short-term-memory#in-production
持久化存储支持的数据库可以通过 https://pypi.org/search/?o=&q=langgraph-checkpoint&page=2 查看，搜索 langgraph-checkpoint-* 查看对应需要安装的依赖和使用方式。
工具中实现长期记忆读写
长期记忆的核心使用场景是在工具中进行读写操作。LangChain通过ToolRuntime将store和context注入到工具函数中，使工具能够访问和修改长期记忆。如下案例中通过工具使用长期记忆记录用户偏好。
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, ToolRuntime
from typing import TypedDict, Literal
from pydantic import BaseModel, Field
from init_llm import deepseek_llm
import uuid
class UserContext(BaseModel):
user_id: str
# 定义工具输入的类型
class UserPreference(BaseModel):
category: Literal["color", "food", "music"] = Field(description=" 用户偏好类别，必须是 'color', 'food', 'music' 中的一个")
preference: str = Field(description="具体偏好内容，如'红色'、'中国美食'等")
# 1. 初始化存储
store = InMemoryStore()
checkpointer = InMemorySaver()
# 2. 定义写入长期记忆的工具
@tool(args_schema=UserPreference)
def save_user_preference(category: str, preference: str, runtime: ToolRuntime) -> str:
"""
将用户偏好保存到长期记忆中。
Args:
category: 用户偏好类别，必须是 "color", "food", "music" 中的一个
preference: 具体偏好内容，如'红色'、'中国美食'等
runtime: ToolRuntime # 包含长期记忆存储和上下文
Returns:
str: 操作结果描述
"""
user_id = runtime.context.user_id
# 创建命名空间：(user_id, "preferences")
namespace = (user_id, "preferences")
# 生成唯一记忆ID
memory_id = str(uuid.uuid4())
# 准备要保存的数据
memory_value = {
"category": category,
"preference": preference,
}
# 保存到长期记忆
runtime.store.put(namespace, memory_id, memory_value)
return f"已成功保存你的{category}偏好：{preference}"
# 3. 定义读取用户偏好的工具
@tool
def get_user_preferences(runtime: ToolRuntime) -> str:
"""
从长期记忆中获取用户特定类别的所有偏好。
Returns:
str: 用户的偏好列表
"""
user_id = runtime.context.user_id
namespace = (user_id, "preferences")
# 搜索该命名空间下的所有记忆
memories = runtime.store.search(namespace)
if not memories:
return f"您还没有保存过偏好"
print("memories:", memories)
# 格式化所有偏好为字符串列表
preferences_list = []
for mem in memories:
pref = mem.value
preferences_list.append(f"- 种类：{pref['category']}，偏好：{pref['preference']}")
return f"你的偏好有：\n" + "\n".join(preferences_list)
# 4. 创建带有长期记忆读写工具的Agent
memory_agent = create_agent(
model=deepseek_llm,
tools=[save_user_preference, get_user_preferences],
checkpointer=checkpointer,
store=store,
context_schema=UserContext
)
# 5. 演示：完整的长时期记忆读写流程
print("=== 完整演示：长期记忆的写入与跨线程读取 ===")
# 第一轮：用户保存颜色偏好（线程1）
print("第一轮（线程1）：用户保存颜色偏好")
result1 = memory_agent.invoke(
{"messages": [{"role": "user", "content": "请记住我喜欢的颜色是蓝色"}]},
config={"configurable": {"thread_id": "thread1"}},
context=UserContext(user_id="current_user")
)
print(f"Agent回复: {result1['messages'][-1].content}")
# 第二轮：用户保存食物偏好（同一线程）
print("第二轮（同一线程）：用户保存食物偏好")
result2 = memory_agent.invoke(
{"messages": [{"role": "user", "content": "我还喜欢的食物是意大利面"}]},
config={"configurable": {"thread_id": "thread1"}}, # 同一线程
context=UserContext(user_id="current_user")
)
print(f"Agent回复: {result2['messages'][-1].content}")
# 第三轮：在新线程中查询所有偏好
print("第三轮（新线程）：查询我的所有偏好")
result3 = memory_agent.invoke(
{"messages": [{"role": "user", "content": "告诉我我都喜欢什么颜色和食物"}]},
config={"configurable": {"thread_id": "thread2"}}, # 新线程
context=UserContext(user_id="current_user") # 相同用户
)
print(f"Agent回复: {result3['messages'][-1].content}")
# 直接验证：从store中读取数据
print("=== 直接验证：从长期记忆存储中读取数据 ===")
# 读取颜色偏好
color_memories = store.search(("current_user", "preferences"))
print(f"长期记忆中存储的颜色偏好: {[m.value for m in color_memories if m.value['category'] == 'color']}")
# 读取食物偏好
food_memories = store.search(("current_user", "preferences"))
print(f"长期记忆中存储的食物偏好: {[m.value for m in food_memories if m.value['category'] == 'food']}")
代码运行结果如下：
短期记忆和长期记忆区别总结
如下是短期记忆和长期记忆区别：
| 对比维度 | 短期记忆（Short-term Memory） | 长期记忆（Long-term Memory） |
|---|---|---|
| 作用域 | 线程/会话范围记忆 (Thread-scoped)。与单个会话线程 (thread_id) 绑定。 | 跨线程/会话记忆。存储在自定义命名空间 (namespace) 中，可被多个线程共享。 |
| 核心目的 | 保证单次对话的连贯性和上下文感知。 | 实现跨对话的个性化、知识积累和持续学习。 |
| 主要存储内容 | 对话的原始历史 (messages列表) 以及当前会话的状态数据。 | 从交互中提炼的结构化知识（如用户事实、行为经验、优化规则）。 |
| 管理组件 | 检查点 (Checkpointer)，如 InMemorySaver、PyMySQLSaver。 | 存储 (Store)，如 InMemoryStore、PostgresStore。 |
| 生命周期 | 随线程的创建而开始，随线程的销毁（或超时清理）而结束。 | 独立于任何特定线程，除非被显式删除，否则永久或长期存在。 |
| 访问方式 | 自动管理。Agent的状态在每个步骤后自动持久化到检查点，并在下次恢复。 | 手动控制。必须在工具(Tool) 或自定义逻辑中，通过代码显式地调用 store.put()或 store.get()。 |
| 典型应用场景 | 维持聊天上下文，让Agent记得用户在当前对话中刚说过的话。 | 记住用户的身份、偏好、历史行为。 |
记忆综合案例-电商客服助手
如下案例是一个智能电商客服助手案例，该助手具备如下能力：
- 记住当前会话状态（短期记忆）：用户正在查询的订单号。
- 了解用户长期偏好（长期记忆）：用户偏好的商品类型和商品名称。
- 处理多轮复杂对话：通过消息摘要管理长对话上下文。
- 工具调用：查询用户信息、查询订单信息、更新用户偏好、基于用户偏好给用户推荐商品。
- 工具调用错误捕获：当工具调用错误时，通过中间件进行错误捕获，返回友好提示。
该案例中长期和短期记忆使用MySQL进行存储，需要在MySQL中创建对应的库：
#进入mysql navicate客户端，创建mysql数据库langchain_db
drop database langchain_db;
create database langchain_db;
在项目中安装必要依赖：
#安装必要依赖
conda activate langchain_v1.2
python -m pip install langgraph-checkpoint-mysql==3.0.0 pymysql==1.1.2 cryptography==46.0.3
python - m pip install aiomysql==0.3.2 asyncmy==0.2.11
案例代码如下：
"""
智能电商客服助手
功能：结合短期记忆、长期记忆、消息摘要，提供流式、个性化的客服服务。
前置准备：
1. 安装依赖: pip install langchain langgraph langgraph-checkpoint-mysql pymysql
2. 创建MySQL数据库: CREATE DATABASE langchain_memory_db;
"""
import uuid
import warnings
from typing import List, Optional
from pydantic import BaseModel, Field
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import SummarizationMiddleware, wrap_tool_call
from langchain_core.tools import tool
from langchain_core.messages import ToolMessage
from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver
from langgraph.store.mysql.pymysql import PyMySQLStore
from langgraph.prebuilt import ToolRuntime
from langgraph.types import Command
from init_llm import deepseek_llm
# 禁用Pydantic序列化警告
warnings.filterwarnings("ignore", category=UserWarning, module="pydantic.main")
# ========== 1. 定义Context 上下文 Schema==========
class UserContext(BaseModel):
"""定义调用Agent时传入的静态上下文信息"""
user_id: str = Field(description="用户的唯一标识符")
channel: str = Field(description="用户咨询渠道，如: APP, Web, 小程序")
# ========== 2. 定义自定义短期记忆状态 (继承AgentState) ==========
class CustomerSessionState(AgentState):
"""自定义短期记忆状态，用于管理单次会话中的动态信息"""
current_order_id: str # 用户当前正在查询的订单号
# ========== 3. 模拟订单数据库 数据 ==========
MOCK_DATABASE = {
"orders": {
"order001": {"order_id": "order001", "status": "已发货", "product": "智能手机",
"preference_context": "华为手机P70"},
"order002": {"order_id": "order002", "status": "待支付", "product": "智能手表",
"preference_context": "Apple Watch Series 8"},
}
}
# ========== 4. 定义工具 ==========
@tool
def get_user_info(runtime: ToolRuntime) -> str:
"""
获取用户当前用户信息
Args:
runtime (ToolRuntime): 包含上下文信息的运行时环境
Returns:
str: 用户当前用户信息
"""
print("get_user_info 中 runtime:", runtime)
# 从上下文中获取当前用户ID
current_user_id = runtime.context.user_id
# 从上下文中获取用户咨询渠道
user_channel = runtime.context.channel
# 从状态中获取用户当前正在查询的订单号
state = runtime.state
if "current_order_id" in state:
current_order_id = state["current_order_id"]
else:
current_order_id = "无"
# 获取当前用户信息
return f"用户ID: {current_user_id}, 咨询渠道: {user_channel}, 当前查询订单号: {current_order_id}"
@tool
def query_order_status(order_id: str, runtime: ToolRuntime) -> Command:
"""
查询用户订单状态
Args:
order_id (str): 用户订单号
runtime (ToolRuntime): 包含上下文信息的运行时环境
Returns:
Command: 包含更新操作的命令对象：状态中更新当前订单ID，并返回订单信息（状态、商品、用户偏好）
"""
# 查询订单状态
order_info = MOCK_DATABASE["orders"].get(order_id)
if not order_info:
return Command(
update={
"messages": [
ToolMessage(
content=f"错误：订单 [{order_id}] 不存在",
tool_call_id=runtime.tool_call_id
)
]
}
)
updates = {
"current_order_id": order_id,
"messages": [
ToolMessage(
content=f"订单 [{order_id}] 状态: {order_info['status']}, 商品: {order_info['product']}。"
f"需要进行用户偏好更新，用户偏好: {order_info['preference_context']}",
tool_call_id=runtime.tool_call_id
)
]
}
return Command(update=updates)
@tool
def update_user_preference(category: str, liked_item: str, runtime: ToolRuntime) -> str:
"""
更新用户长期偏好
Args:
category (str): 商品类别，如: 手机、配件
liked_item (str): 用户喜欢的具体商品
runtime (ToolRuntime): 包含上下文信息的运行时环境
Returns:
str: 确认更新结果
"""
user_id = runtime.context.user_id
namespace = (f"user_{user_id}", "preferences")
key = str(uuid.uuid4())
value_to_store = {
"category": category,
"liked_item": liked_item,
}
# 写入到长期记忆
runtime.store.put(namespace, key, value_to_store)
return f"已成功将您的偏好记录到长期记忆: 喜欢 {category} 类的 {liked_item}。"
@tool
def get_recommendation(runtime: ToolRuntime) -> str:
"""
获取用户推荐商品
Args:
runtime (ToolRuntime): 包含上下文信息的运行时环境
Returns:
str: 包含推荐商品信息的字符串
"""
user_id = runtime.context.user_id
current_order = runtime.state.get("current_order_id", "未知订单")
namespace = (f"user_{user_id}", "preferences")
prefs = runtime.store.search(namespace)
pref_list = []
if prefs:
for p in prefs[-3:]: # 仅取最近3条偏好记录，[-3:] 表示取最后3条记录
pref_list.append(f"{p.value.get('category')}({p.value.get('liked_item')})")
return f"基于用户当前的订单 [{current_order}] 和长期偏好 {pref_list if pref_list else '无'}，为用户推荐相关配件或类似风格商品。"
@wrap_tool_call
def handle_tool_errors(request, handler):
"""使用自定义消息处理工具执行错误"""
try:
return handler(request)
except Exception as e:
# 向模型返回自定义错误消息
return ToolMessage(
content=f"调用工具错误:请稍后重试，错误信息：({str(e)})",
tool_call_id=request.tool_call["id"]
)
# ========== 5. 创建Agent，控制台交互循环 ==========
DB_URI = "mysql+pymysql://root:123456@localhost:3306/langchain_db?charset=utf8mb4"
# 初始化MySQL存储 (短期记忆Checkpointer 和 长期记忆Store)
with (
PyMySQLSaver.from_conn_string(DB_URI) as checkpointer,
PyMySQLStore.from_conn_string(DB_URI) as store
):
# 首次运行时自动建表
checkpointer.setup()
store.setup()
# 创建Agent
agent = create_agent(
model=deepseek_llm,
tools=[get_user_info, query_order_status, update_user_preference, get_recommendation],
system_prompt="""
你是一个智能电商客服助手，具备回答用户咨询、获取用户信息、查询订单状态、更新用户偏好和推荐商品功能。"
获取用户信息请调用 get_user_info 工具。
查询订单状态请调用 query_order_status 工具，查询到订单状态后，还需要调用 update_user_preference 工具更新用户偏好。
更新用户偏好请调用 update_user_preference 工具。
获取推荐商品请调用 get_recommendation 工具。
""",
checkpointer=checkpointer,
store=store,
state_schema=CustomerSessionState,
context_schema=UserContext,
middleware=[
SummarizationMiddleware(
model=deepseek_llm,
summary_prompt="请总结以下对话内容：{messages}",
trigger=("messages", 10), # 每10条消息触发一次摘要
keep=("messages", 5), # 保留最后5条消息
),
handle_tool_errors
],
)
# 控制台交互循环 (流式调用)
print("=" * 50)
print("智能电商客服助手")
print("功能: 查询订单、更新偏好、获取推荐。")
print("输入 'quit' 或 '退出' 结束对话。")
print("=" * 50)
# 初始化用户上下文
user_context = UserContext(user_id="customer_001", channel="Web")
# 会话线程ID
config = {"configurable": {"thread_id": "session_01"}}
# 对话循环
while True:
try:
user_input = input("[你]: ").strip()
if user_input.lower() in ['quit', 'exit', '退出', 'q']:
print("客服助手: 感谢你的咨询，再见！")
break
# 过滤空输入
if not user_input:
continue
# 准备输入消息
input_data = {"messages": {"role": "user", "content": user_input}}
print("[客服助手]: ")
# 流式调用Agent
for chunk in agent.stream(input_data, config=config, context=user_context):
# print("chunk:", chunk)
for step, data in chunk.items(): # 遍历dict的key-value对
# print("step:", step)
# print("data:", data)
# 只有当 step为model或者tools时，才打印消息
if step in ["model", "tools"]:
message = data["messages"][-1]
message.pretty_print()
except Exception as e:
print(f"调用过程中出现错误: {e}")
代码运行后，对于thread_id为session_01 时，进行如下对话：
给我查询订单 order001 信息
给我推荐一些商品
我喜欢 苹果电脑 尤其是 mac pro
查询我的信息
我还喜欢 索尼的xm5耳机
给我推荐一些商品
停止程序后，切换thread_id为session_02后，进行如下对话：
查询我的信息
给我推荐一些商品
可以看到长期记忆信息在不同的会话之间是共享的，而短期记忆只针对每个thread_id生效。
此外，以上代码运行中，会出现如下警告：
UserWarning: Pydantic serializer warnings:
PydanticSerializationUnexpectedValue(Expected `none` - serialized value may not be as expected [field_name='context', input_value=UserContext(user_id='customer_001', channel='Web'), input_type=UserContext])
return self.__pydantic_serializer__.to_python(
该警告是将Pydantic模型对象直接传递给了LangChain Agent的context参数，而LangChain内部在处理这个对象时，期望的是基本的Python数据类型（如字典），而不是Pydantic模型对象，导致了序列化警告。去除该警告，可以在项目开始导入：“warnings.filterwarnings(“ignore”, category=UserWarning, module=”pydantic.main”)”