Agent 长期记忆

长期记忆介绍

长期记忆是跨越会话和线程共享的存储系统。与短期记忆局限于 thread_id 不同，长期记忆的数据可以在任何线程中被召回，实现真正的跨会话信息持久化。

在 LangChain 中，长期记忆通过 Store 组件实现。Store 允许将记忆保存为 JSON 文档，通过 namespace（命名空间） 和 key（键） 进行组织管理：

namespace：类似文件夹，用于逻辑分组
key：命名空间内文档的唯一标识

Store 支持以下四种操作：

操作	说明
`put`	写入记忆
`get`	读取记忆
`delete`	删除记忆
`search`	搜索记忆

基本操作示例

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

namespace = ("user1", "preferences")

# 写入记忆
store.put(namespace, "fruit", {"likes": ["苹果", "香蕉"], "dislikes": ["橙子"]})

print("=== 读取 fruit 记忆 ===")
memory = store.get(namespace, "fruit")
print(memory)

store.put(namespace, "chat", {"language": "中文", "emotion": "高兴"})
print("=== 读取 chat 记忆 ===")
memory = store.get(namespace, "chat")
print(memory)

print("=== 搜索所有记忆 ===")
memories = store.search(namespace)
print(memories)

# 更新记忆（相同 key 的 put 会覆盖原值）
store.put(namespace, "fruit", {"likes": ["橘子", "葡萄"], "dislikes": ["草莓"]})
print("=== 读取更新后的 fruit 记忆 ===")
memory = store.get(namespace, "fruit")
print(memory)

要点

namespace 为元组格式，建议包含用户 ID 和上下文特征
相同 key 的 put 会覆盖原值
value 是字典结构

Agent 中使用长期记忆方式

长期记忆的核心作用是跨会话数据共享。Agent 通过将 Store 注入到不同会话中实现这一能力。

使用内存存储长期记忆

使用 InMemoryStore + ToolRuntime + context_schema 实现内存级长期记忆：

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

store = InMemoryStore()
checkpointer = InMemorySaver()

# 预存用户信息
store.put(("users",), "user_123", {"name": "张三", "age": 28, "city": "北京", "hobby": "编程、阅读"})
store.put(("users",), "user_456", {"name": "李四", "age": 32, "city": "上海", "hobby": "旅游、摄影"})

@tool
def get_user_info(runtime: ToolRuntime[UserContext]) -> str:
    """从长期记忆中获取当前用户的信息"""
    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 "未找到该用户的信息。"

agent = create_agent(
    model=deepseek_llm,
    tools=[get_user_info],
    checkpointer=checkpointer,
    store=store,
    context_schema=UserContext
)

print("=== 长期记忆的跨线程读取 ===")

print("线程1 - 用户123询问信息:")
result1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我的个人信息是什么？"}]},
    config={"configurable": {"thread_id": "thread_1"}},
    context=UserContext(user_id="user_123")
)
print(f"Agent回复: {result1['messages'][-1].content}\n")

print("线程2 - 相同的用户，不同的线程:")
result2 = agent.invoke(
    {"messages": [{"role": "user", "content": "再告诉我一次我的信息"}]},
    config={"configurable": {"thread_id": "thread_2"}},
    context=UserContext(user_id="user_123")
)
print(f"Agent回复: {result2['messages'][-1].content}\n")

print("线程3 - 用户456询问信息:")
result3 = agent.invoke(
    {"messages": [{"role": "user", "content": "我的信息是什么？"}]},
    config={"configurable": {"thread_id": "thread_3"}},
    context={"user_id": "user_456"}
)
print(f"Agent回复: {result3['messages'][-1].content}")

要点

create_agent 通过 store 参数指定长期记忆，context_schema 指定上下文类型。

使用数据库存储长期记忆

使用 MySQL（PyMySQLStore）实现持久化长期记忆。

前置准备：

create database langchain_db;

安装依赖：

pip install langgraph-checkpoint-mysql==3.0.0 pymysql==1.1.2 cryptography==46.0.3
pip install aiomysql==0.3.2 asyncmy==0.2.11

代码示例：

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 init_llm import deepseek_llm

class UserContext(BaseModel):
    user_id: str

@tool
def get_user_info(runtime: ToolRuntime[UserContext]) -> str:
    """从长期记忆中获取当前用户的信息"""
    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 "未找到该用户的信息。"

DB_URI = "mysql+pymysql://root:123456@localhost:3306/langchain_db?charset=utf8mb4"

with (
    PyMySQLSaver.from_conn_string(DB_URI) as checkpointer,
    PyMySQLStore.from_conn_string(DB_URI) as store
):
    checkpointer.setup()
    store.setup()

    store.put(("users",), "user_123", {"name": "张三", "age": 28, "city": "北京", "hobby": "编程、阅读"})
    store.put(("users",), "user_456", {"name": "李四", "age": 32, "city": "上海", "hobby": "旅游、摄影"})

    agent = create_agent(
        model=deepseek_llm,
        tools=[get_user_info],
        checkpointer=checkpointer,
        store=store,
        context_schema=UserContext
    )

    print("=== 长期记忆的跨线程读取 ===")
    print("线程1 - 用户123询问信息:")
    result1 = agent.invoke(
        {"messages": [{"role": "user", "content": "我的个人信息是什么？"}]},
        config={"configurable": {"thread_id": "thread_1"}},
        context=UserContext(user_id="user_123")
    )
    print(f"Agent回复: {result1['messages'][-1].content}\n")

    print("线程2 - 相同的用户，不同的线程:")
    result2 = agent.invoke(
        {"messages": [{"role": "user", "content": "再告诉我一次我的信息"}]},
        config={"configurable": {"thread_id": "thread_2"}},
        context=UserContext(user_id="user_123")
    )
    print(f"Agent回复: {result2['messages'][-1].content}\n")

    print("线程3 - 用户456询问信息:")
    result3 = agent.invoke(
        {"messages": [{"role": "user", "content": "我的信息是什么？"}]},
        config={"configurable": {"thread_id": "thread_3"}},
        context=UserContext(user_id="user_456")
    )
    print(f"Agent回复: {result3['messages'][-1].content}")

要点

PyMySQLStore.from_conn_string 创建持久化连接；store.setup() 自动建表。

工具中实现长期记忆读写

通过 ToolRuntime 将 store 和 context 注入工具，实现长期记忆的读写操作。

使用 args_schema 定义工具输入，save_user_preference 负责写入偏好，get_user_preferences 负责读取所有偏好：

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 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="具体偏好内容，如'红色'、'中国美食'等")

store = InMemoryStore()
checkpointer = InMemorySaver()

@tool(args_schema=UserPreference)
def save_user_preference(category: str, preference: str, runtime: ToolRuntime) -> str:
    """将用户偏好保存到长期记忆中"""
    user_id = runtime.context.user_id
    namespace = (user_id, "preferences")
    memory_id = str(uuid.uuid4())
    memory_value = {"category": category, "preference": preference}
    runtime.store.put(namespace, memory_id, memory_value)
    return f"已成功保存你的{category}偏好：{preference}"

@tool
def get_user_preferences(runtime: ToolRuntime) -> 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)

memory_agent = create_agent(
    model=deepseek_llm,
    tools=[save_user_preference, get_user_preferences],
    checkpointer=checkpointer,
    store=store,
    context_schema=UserContext
)

print("=== 完整演示：长期记忆的写入与跨线程读取 ===")

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}")

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']}")

短期记忆和长期记忆区别总结

对比维度	短期记忆	长期记忆
作用域	线程/会话范围（`thread_id` 绑定）	跨线程/会话（自定义 `namespace`）
核心目的	保证单次对话的连贯性和上下文感知	实现跨对话的个性化、知识积累和持续学习
主要存储内容	对话的原始历史和当前会话状态数据	从交互中提炼的结构化知识
管理组件	Checkpointer（`InMemorySaver`、`PyMySQLSaver`）	Store（`InMemoryStore`、`PyMySQLStore`）
生命周期	随线程创建开始，随线程销毁结束	独立于线程，除非显式删除否则永久存在
访问方式	自动管理，每步自动持久化并恢复	手动控制，需在工具或逻辑中显式调用 `store.put()`/`get()`
典型应用场景	维持聊天上下文	记住用户身份、偏好、历史行为

记忆综合案例 - 电商客服助手

本案例整合短期记忆与长期记忆，构建一个完整的电商客服助手，具备以下能力：

记住当前会话状态（短期记忆）：当前查询的订单号
了解用户长期偏好（长期记忆）：偏好的商品类型和名称
处理多轮对话：通过消息摘要管理长对话上下文
工具调用：查询用户信息、订单信息、更新偏好、推荐商品
错误捕获：@wrap_tool_call 中间件

前置准备：MySQL 建库和安装依赖（与上文相同）。

完整代码

"""
智能电商客服助手
功能：结合短期记忆、长期记忆、消息摘要，提供个性化客服服务。
"""
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

warnings.filterwarnings("ignore", category=UserWarning, module="pydantic.main")

# 1. 定义 Context Schema
class UserContext(BaseModel):
    user_id: str = Field(description="用户的唯一标识符")
    channel: str = Field(description="用户咨询渠道，如: APP, Web, 小程序")

# 2. 自定义短期记忆状态
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:
    """获取用户当前用户信息"""
    current_user_id = runtime.context.user_id
    user_channel = runtime.context.channel
    state = runtime.state
    current_order_id = state.get("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:
    """查询用户订单状态"""
    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']}。需要进行用户偏好更新，用户偏好: {order_info['preference_context']}",
            tool_call_id=runtime.tool_call_id
        )]
    }
    return Command(update=updates)

@tool
def update_user_preference(preference: str, runtime: ToolRuntime) -> str:
    """更新用户的商品偏好到长期记忆"""
    user_id = runtime.context.user_id
    namespace = (user_id, "preferences")
    memory_id = str(uuid.uuid4())
    runtime.store.put(namespace, memory_id, {"preference": preference, "source": "order_interaction"})
    return f"已更新用户偏好: {preference}"

@tool
def recommend_products(runtime: ToolRuntime) -> str:
    """基于用户偏好推荐商品"""
    user_id = runtime.context.user_id
    namespace = (user_id, "preferences")
    memories = runtime.store.search(namespace, limit=10)
    if not memories:
        return "暂无用户偏好记录，推荐热门商品：iPhone 16、小米手环、AirPods Pro"
    prefs = [m.value.get("preference", "") for m in memories]
    return f"基于您的偏好 {prefs}，为您推荐相关商品"

# 5. 工具错误捕获中间件
@wrap_tool_call
def handle_tool_error(tool_call, tool_result, state, runtime):
    """捕获工具调用错误，返回友好提示"""
    if isinstance(tool_result, Exception):
        return ToolMessage(
            content=f"工具调用出错：{str(tool_result)}，请稍后重试",
            tool_call_id=tool_call["id"]
        )
    return tool_result

DB_URI = "mysql+pymysql://root:123456@localhost:3306/langchain_db?charset=utf8mb4"

def run_ecommerce_agent():
    with (
        PyMySQLSaver.from_conn_string(DB_URI) as checkpointer,
        PyMySQLStore.from_conn_string(DB_URI) as store
    ):
        checkpointer.setup()
        store.setup()

        agent = create_agent(
            model=deepseek_llm,
            tools=[get_user_info, query_order_status, update_user_preference, recommend_products],
            state_schema=CustomerSessionState,
            system_prompt="你是电商客服助手。查询订单后自动更新用户偏好到长期记忆，推荐商品前先查询用户偏好。",
            middleware=[
                handle_tool_error,
                SummarizationMiddleware(model=deepseek_llm, trigger=('messages', 10), keep=('messages', 4)),
            ],
            checkpointer=checkpointer,
            store=store,
            context_schema=UserContext
        )

        # 场景1：用户查询订单
        print("=" * 60)
        print("场景1：用户查询订单")
        print("=" * 60)
        config1 = {"configurable": {"thread_id": "session_001"}}
        response = agent.invoke(
            {"messages": [{"role": "user", "content": "帮我查一下订单order001的状态"}], "current_order_id": ""},
            config=config1,
            context=UserContext(user_id="user_wang", channel="App")
        )
        print(f"客服: {response['messages'][-1].content}")

        # 场景2：用户请求推荐
        print("\n" + "=" * 60)
        print("场景2：用户请求推荐商品")
        print("=" * 60)
        response = agent.invoke(
            {"messages": [{"role": "user", "content": "有什么商品推荐给我吗？"}]},
            config=config1,
            context=UserContext(user_id="user_wang", channel="App")
        )
        print(f"客服: {response['messages'][-1].content}")

        # 场景3：新会话，跨会话记忆
        print("\n" + "=" * 60)
        print("场景3：新会话，测试跨会话记忆")
        print("=" * 60)
        config2 = {"configurable": {"thread_id": "session_002"}}
        response = agent.invoke(
            {"messages": [{"role": "user", "content": "我之前买过什么类型的商品？推荐类似的"}], "current_order_id": ""},
            config=config2,
            context=UserContext(user_id="user_wang", channel="Web")
        )
        print(f"客服: {response['messages'][-1].content}")

if __name__ == "__main__":
    run_ecommerce_agent()

案例总结

能力	实现组件	作用
当前对话连贯	Checkpointer（短期记忆）	维持多轮对话上下文
跨会话记忆	Store（长期记忆）	记住偏好、反馈
长对话管理	SummarizationMiddleware	消息摘要控制 token
错误处理	`@wrap_tool_call`	工具错误友好提示
用户隔离	namespace + context_schema	不同用户数据互不干扰

长期记忆介绍​

基本操作示例​

Agent 中使用长期记忆方式​

使用内存存储长期记忆​

使用数据库存储长期记忆​

工具中实现长期记忆读写​

短期记忆和长期记忆区别总结​

记忆综合案例 - 电商客服助手​

完整代码​

案例总结​