Skip to content

06 长期记忆

上一篇文章介绍了,短期记忆以及短期记忆的用法,本文将会介绍长期记忆的用法,短期记忆的作用范围是同一thread_id中,也就是在同一会话中,而长期记忆是跨会话的、用户级别的记忆信息存储机制。

在LangChain中,长期记忆基于LangGraph的Store实现,Store支持设置命名空间、key将长期记忆信息以JSON文档的形式进行保存。

一、在工具中读取长期记忆

代码示例如下:

python

from dataclasses import dataclass

from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langchain_core.runnables import Runnable
from langgraph.store.memory import InMemoryStore
import dotenv

dotenv.load_dotenv()


# 1.定义用户上下文
@dataclass
class Context:
    user_id: str


# 2.定义store并添加数据
store = InMemoryStore()

store.put(
    ("users",),
    "user_001",
    {
        "name": "大志",
        "language": "中文",
        "job": "Java高级开发工程师",
        "interest": ["LangChain", "AI Agent", "公众号写作"]
    },
)


# 3.定义工具信息
@tool
def get_user_info(runtime: ToolRuntime[Context]) -> str:
    """获取用户信息"""
    user_id = runtime.context.user_id

    user_info = runtime.store.get(("users",), user_id)

    if not user_info:
        return "未找到该用户信息"

    return f"""
            用户信息如下:
            {user_info.value}
            """


# 4.创建agent
agent: Runnable = create_agent(
    model="deepseek-v4-flash",
    tools=[get_user_info],
    store=store,
    context_schema=Context,
)

# 5.调用Agent
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "帮我查一下这个用户的信息,并简单介绍一下他"
            }
        ]
    },
    context=Context(user_id="user_001"),
)

print(result["messages"][-1].content)

执行结果:

image-20260612112838286

二、在工具中写入长期记忆

代码示例如下:

python
from dataclasses import dataclass

from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langchain_core.runnables import Runnable
from langgraph.store.memory import InMemoryStore
import dotenv

dotenv.load_dotenv()


# 1.定义用户上下文
@dataclass
class Context:
    user_id: str


# 2.定义store并添加数据
store = InMemoryStore()

store.put(
    ("users",),
    "user_001",
    {
        "name": "大志",
        "language": "中文",
        "job": "Java高级开发工程师",
        "interest": ["LangChain", "AI Agent", "公众号写作"]
    },
)

# 3.定义工具:写入长期记忆信息
@tool
def update_user_memory(runtime: ToolRuntime[Context], info: str) -> str:
    """写入用户长期记忆"""

    user_id = runtime.context.user_id

    # 1.读取已有记忆
    old = runtime.store.get(("users",), user_id)

    old_value = old.value if old else {}

    # 2.合并新记忆
    new_memory = {
        **old_value,
        "extra_info": info
    }

    # 3.写入长期记忆
    runtime.store.put(
        ("users",),
        user_id,
        new_memory
    )

    return f"""
            已写入长期记忆:
            {new_memory}
            """


# 4.创建agent
agent: Runnable = create_agent(
    model="deepseek-v4-flash",
    tools=[update_user_memory],
    store=store,
    context_schema=Context,
)


# 5.调用Agent
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "帮我记住:我最近在学习微服务架构,并且在用 Spring Cloud"
            }
        ]
    },
    context=Context(user_id="user_001"),
)

# 6.获取store内容
print(store.get(("users",), "user_001"))

执行结果如下:

json
Item(namespace=['users'], key='user_001', value={'name': '大志', 'language': '中文', 'job': 'Java高级开发工程师', 'interest': ['LangChain', 'AI Agent', '公众号写作'], 'extra_info': '用户最近在学习微服务架构,并且在用 Spring Cloud。'}, created_at='2026-06-12T04:15:40.629007+00:00', updated_at='2026-06-12T04:15:40.629011+00:00')