LangChain：长期记忆的使用与读取

为 LangChain 智能体添加长期记忆，使其能够在不同对话和会话之间存储和回忆数据。

长期记忆可以让智能体能够跨不同的对话和会话存储和回忆信息。

为 LangChain 智能体添加长期记忆，使其能够在不同对话和会话之间存储和回忆数据。

长期记忆让你的智能体能够跨不同的对话和会话存储和回忆信息。与作用域限定在单个线程内的短期记忆不同，长期记忆可以跨线程持久存在，并可随时被召回。长期记忆构建于 LangGraph 存储之上，它将数据保存为由命名空间和键组织的 JSON 文档。

使用方法

要为智能体添加长期记忆，需创建一个存储（store）并将其传递给 create_agent：

内存存储

from langchain.agents import create_agent
from langchain_core.runnables import Runnable
from langgraph.store.memory import InMemoryStore

# InMemoryStore 将数据保存到内存字典中。生产环境中请使用数据库支持的存储。
store = InMemoryStore()

agent: Runnable = create_agent(
    "claude-sonnet-4-6",
    tools=[],
    store=store,  # 将存储对象注入智能体，使智能体及工具能够读写长期记忆
)

代码说明：上述代码演示了如何创建一个最简单的支持长期记忆的智能体。InMemoryStore 适用于开发和测试，数据仅保存在内存中，进程结束后即丢失。生产环境应使用数据库支持的存储（如 PostgresStore）。创建存储后，通过 store 参数传递给 create_agent，智能体内部以及后续定义的工具都可以通过 runtime.store 访问该存储对象。

PostgreSQL存储

pip install langgraph-checkpoint-postgres

• 需要先安装 langgraph 的 PostgreSQL 扩展。
• 生产环境建议使用连接池或异步驱动（如 asyncpg），并配置合适的连接参数。
• setup() 表结构创建应放在数据库迁移脚本中，避免应用运行时意外重建表。

# 导入 LangChain 的智能体创建函数
from langchain.agents import create_agent
# 导入 Runnable 类型提示，用于标注智能体为可运行对象
from langchain_core.runnables import Runnable
# 导入基于 PostgreSQL 的长期记忆存储实现（生产环境推荐使用）
# type: ignore[import-not-found] 表示忽略导入缺失的错误（通常由类型检查工具触发）
from langgraph.store.postgres import PostgresStore  # type: ignore[import-not-found]

# 定义 PostgreSQL 数据库连接字符串
# 格式：postgresql://用户名:密码@主机:端口/数据库名?sslmode=disable
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"

# 使用 with 语句管理 PostgresStore 连接，确保退出时自动关闭连接
with PostgresStore.from_conn_string(DB_URI) as store:
    # 执行数据库初始化：创建存储所需的表结构和索引（只需运行一次，后续可注释）
    store.setup()

    # 创建支持长期记忆的 LangChain 智能体
    # 参数：
    #   - "claude-sonnet-4-6": 使用的大语言模型
    #   - tools=[]: 当前未绑定任何工具（可按需添加）
    #   - store=store: 注入 PostgreSQL 存储，使智能体具备跨会话持久化记忆的能力
    agent: Runnable = create_agent(
        "claude-sonnet-4-6",
        tools=[],
        store=store,
    )

代码说明：本段代码展示了如何为 LangChain 智能体配置基于 PostgreSQL 的长期记忆存储，从而替代仅适用于开发/测试的 InMemoryStore。

• 数据库连接：DB_URI 定义了 PostgreSQL 的连接信息。示例中使用了本地默认配置（用户 postgres，密码 postgres，数据库 postgres），实际生产环境中应通过环境变量或密钥管理服务（如 AWS Secrets Manager）获取连接字符串，避免硬编码敏感信息。
• 存储初始化：PostgresStore.from_conn_string(DB_URI) 根据连接字符串创建存储实例。with 语句确保智能体使用完毕后数据库连接被正确关闭，防止资源泄漏。store.setup() 会在数据库中创建必要的表（如 store 表）、索引以及向量支持所需的结构。注意：setup() 应当只执行一次（例如在应用部署或数据库迁移脚本中），多次调用虽然不会出错，但会增加不必要的开销。
• 智能体创建：create_agent 接收 store 参数后，会将该存储注入到智能体及其所有工具中。这样，智能体就能够通过 runtime.store 在对话中读写长期记忆——例如记住用户的偏好、历史交互或业务数据，并在不同会话、不同线程中共享这些信息。与 InMemoryStore 不同，PostgresStore 的数据会持久化到磁盘，即使智能体重启或水平扩展到多台服务器，记忆也不会丢失。
• 依赖提示：使用 PostgresStore 需要安装额外的依赖，如 psycopg（或 psycopg2）以及 langgraph[postgres]。若缺少依赖，Python 导入时会抛出 ModuleNotFoundError。代码中的 # type: ignore[import-not-found] 仅用于抑制静态类型检查器（如 mypy 或 pyright）的报错，不影响运行时行为。

然后，Tools 可以使用 runtime.store 参数从存储中读取和写入数据。请参阅在工具中读取长期记忆和在工具中写入长期记忆中的示例。

如需深入了解内存类型（语义记忆、情景记忆、程序性记忆）以及写入记忆的策略，请参阅内存概念指南。

记忆存储

LangGraph 将长期记忆作为 JSON 文档存储在存储中。

每个记忆都在一个自定义的命名空间（类似于文件夹）和一个唯一的键（类似于文件名）下进行组织。命名空间通常包含用户或组织 ID 或其他标签，以便于组织信息。

这种结构支持记忆的分层组织。随后，可以通过内容过滤器支持跨命名空间搜索。

关于内存存储的更多信息，请参见持久化指南。

内存存储

from collections.abc import Sequence
from langgraph.store.base import IndexConfig
from langgraph.store.memory import InMemoryStore

# 定义嵌入函数，用于将文本转换为向量，实现语义搜索
def embed(texts: Sequence[str]) -> list[list[float]]:
    # 替换为实际的嵌入函数或 LangChain 嵌入对象
    return [[1.0, 2.0] for _ in texts]  # 示例：返回固定维度的向量

# InMemoryStore 将数据保存到内存字典中。生产环境中请使用数据库支持的存储。
store = InMemoryStore(index=IndexConfig(embed=embed, dims=2))  # 启用向量索引，dims=2 表示嵌入向量的维度

user_id = "my-user"
application_context = "chitchat"               # 场景上下文，例如闲聊、技术支持等
namespace = (user_id, application_context)     # 命名空间为元组，支持层级结构

# 在指定命名空间下存储一个记忆项（键为 "a-memory"）
store.put(
    namespace,
    "a-memory",
    {
        "rules": [
            "User likes short, direct language",
            "User only speaks English & python",
        ],
        "my-key": "my-value",
    },
)

# 通过 ID 获取“记忆”
item = store.get(namespace, "a-memory")

# 在该命名空间内搜索“记忆”，基于内容等价性进行过滤，并按向量相似度排序
items = store.search(
    namespace,
    filter={"my-key": "my-value"},   # 精确过滤：只返回 my-key 等于 my-value 的记忆
    query="language preferences"      # 查询文本，会使用 embed 函数转换为向量并与存储的文档向量做相似度搜索
)

代码说明：此代码块展示了存储的核心操作：put（写入）、get（读取）、search（语义搜索）。

• 命名空间：使用元组 (user_id, application_context) 来组织记忆，例如可以按用户 -> 场景的层级隔离不同数据。
• 索引配置：IndexConfig 提供了向量索引能力，使得 search 方法能够根据语义相似性返回最相关的记忆项。生产环境中应使用真正的嵌入模型（如 OpenAIEmbeddings）替换示例中的 embed 函数。
• 搜索过滤：filter 参数支持对存储的 JSON 文档中的字段进行精确匹配，query 参数用于向量相似度排序。两者可同时使用。

PostgreSQL存储

pip install langgraph-checkpoint-postgres

# 导入 Sequence 类型，用于标注嵌入函数的参数类型
from collections.abc import Sequence

# 导入存储索引配置类，用于启用向量语义搜索
from langgraph.store.base import IndexConfig
# 导入 PostgreSQL 长期记忆存储（生产环境推荐），忽略类型检查器可能报的缺少导入错误
from langgraph.store.postgres import PostgresStore  # type: ignore[import-not-found]


# 定义嵌入函数，将文本序列转换为向量
# 实际使用时需替换为真实的嵌入模型（如 OpenAIEmbeddings 或 HuggingFace 嵌入）
def embed(texts: Sequence[str]) -> list[list[float]]:
    # 示例实现：为每个文本返回固定值 [1.0, 2.0]，演示向量维度为2
    return [[1.0, 2.0] for _ in texts]


# PostgreSQL 数据库连接字符串（请根据实际环境修改用户名、密码、主机、数据库名）
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"

# 使用 with 语句创建 PostgresStore 连接，确保自动释放资源
# 同时配置索引：embed 函数和向量维度 dims=2，使存储支持语义搜索
with PostgresStore.from_conn_string(
    DB_URI,
    index=IndexConfig(embed=embed, dims=2),  # type: ignore[arg-type]  # 忽略类型检查器对 embed 参数类型的提示
) as store:
    # 初始化数据库：创建必要的表、索引等（首次运行或数据库结构变更时执行）
    store.setup()

    # 定义用户标识和场景上下文，构成记忆的命名空间（类似于文件夹）
    user_id = "my-user"
    application_context = "chitchat"          # 例如：闲聊场景
    namespace = (user_id, application_context)  # 命名空间使用元组，支持层级组织

    # 在指定命名空间下写入一条记忆，键为 "a-memory"，值为包含规则和自定义键值的 JSON 对象
    store.put(
        namespace,
        "a-memory",
        {
            "rules": [
                "User likes short, direct language",
                "User only speaks English & python",
            ],
            "my-key": "my-value",
        },
    )

    # 根据命名空间和键精确获取记忆项
    item = store.get(namespace, "a-memory")

    # 在命名空间内进行语义搜索：
    # - filter: 精确过滤字段 my-key 等于 my-value 的记忆
    # - query: 查询文本 "language preferences"，会通过 embed 函数转换为向量，然后与存储的文档向量计算相似度排序
    items = store.search(
        namespace, filter={"my-key": "my-value"}, query="language preferences"
    )

代码说明：这段代码演示了如何配置并使用 PostgreSQL 支持的长期记忆存储，并启用向量语义搜索功能。与之前的内存存储示例相比，这里使用了 PostgresStore 并传入了 IndexConfig。

• 嵌入函数
  embed 函数用于将文本转换为向量，是语义搜索的核心。示例中返回固定值（[1.0, 2.0]）仅为演示，实际生产环境应使用真实的嵌入模型（例如 OpenAIEmbeddings 或 HuggingFaceEmbeddings）。向量维度 dims=2 需与嵌入函数输出的维度一致。

• 索引配置
  IndexConfig(embed=embed, dims=2) 启用向量索引。这会在数据库中创建额外的向量存储结构，使得 store.search() 方法能够根据查询文本的语义相似度返回最相关的记忆项，而不仅仅是字符串匹配。

• 命名空间与键

  • 命名空间 (user_id, application_context) 采用元组形式，支持多级组织（例如 ("user_123", "chat", "session_abc")）。

  • store.put() 写入记忆，store.get() 精确读取。

  • store.search() 支持两种过滤：filter 参数用于对 JSON 字段进行精确匹配；query 参数用于向量相似度搜索。两者可以同时使用。

• 依赖与初始化

  • 使用 PostgresStore 需要安装 psycopg（或 asyncpg）以及 langgraph[postgres] 扩展。

  • store.setup() 负责创建数据库表（如 store 表、向量索引表等）。建议在应用部署或数据库迁移脚本中执行，后续运行可注释或跳过。

• 资源管理
  with 语句确保在退出代码块后数据库连接被正确关闭，避免连接泄漏。生产环境通常使用连接池，但 from_conn_string 配合 with 也能满足简单脚本或开发场景。

典型应用场景：

• 为聊天机器人长期保存用户偏好（如语言风格、兴趣领域）。
• 跨会话记忆业务上下文，例如客服系统中记住用户之前的问题。
• 结合语义搜索，根据用户当前问题自动检索历史相关记忆。

在工具中读取长期记忆

读取内存记忆

# 导入 dataclass 装饰器，用于定义上下文数据结构
from dataclasses import dataclass

# 导入智能体创建函数
from langchain.agents import create_agent
# 导入工具运行时类型和工具装饰器
from langchain.tools import ToolRuntime, tool
# 导入 Runnable 类型提示，用于标注智能体对象
from langchain_core.runnables import Runnable
# 导入内存存储（仅用于开发/测试，生产环境请使用数据库存储）
from langgraph.store.memory import InMemoryStore


# 定义上下文数据类，包含用户ID，智能体会将此类实例注入到工具中
@dataclass
class Context:
    user_id: str


# InMemoryStore 将数据保存到内存字典中。生产环境中请使用数据库支持的存储（如 PostgresStore）
store = InMemoryStore()

# 使用 put 方法向存储中写入示例数据
store.put(
    ("users",),          # 命名空间：将相关数据分组（users 命名空间用于存放用户数据）
    "user_123",          # 命名空间内的键：这里使用用户 ID 作为键
    {                    # 要存储的数据（任意 JSON 可序列化的字典）
        "name": "John Smith",
        "language": "English",
    },
)


# 定义工具：通过运行时上下文获取用户信息
@tool
def get_user_info(runtime: ToolRuntime[Context]) -> str:
    """查找用户信息。"""
    # runtime.store 与创建智能体时传入的 store 是同一个对象
    assert runtime.store is not None
    # 从上下文中获取当前用户 ID
    user_id = runtime.context.user_id
    # 从存储中检索数据：在 ("users",) 命名空间下，使用 user_id 作为键
    # 返回值为 StoreValue 对象（包含 value 和元数据），若不存在则返回 None
    user_info = runtime.store.get(("users",), user_id)
    # 如果找到数据则返回其字符串形式，否则返回 "Unknown user"
    return str(user_info.value) if user_info else "Unknown user"


# 创建支持长期记忆的智能体
agent: Runnable = create_agent(
    model="openai:gpt-5.4",          # 使用的大语言模型
    tools=[get_user_info],           # 工具列表，包含刚定义的工具
    store=store,                     # 传入存储对象，使智能体在运行工具时能够访问存储
    context_schema=Context,          # 声明上下文的数据结构，用于运行时注入
)

# 运行智能体：通过 context 参数传递当前用户上下文
agent.invoke(
    {"messages": [{"role": "user", "content": "look up user information"}]},
    context=Context(user_id="user_123"),
)

代码说明：这段代码演示了如何在 LangChain 智能体的工具内部读取长期记忆。它使用内存存储（InMemoryStore）作为示例，展示了从存储中检索与当前用户相关的信息。

核心流程：

1. 定义上下文
   使用 @dataclass 定义 Context 类，其中包含 user_id 字段。智能体在调用 invoke 时通过 context 参数传入该对象，并在运行工具时将其提供给 runtime.context。
2. 初始化存储
   store = InMemoryStore() 创建内存存储。为演示目的，预先调用 store.put() 写入了一条示例用户数据（用户 ID user_123 对应的姓名和语言偏好）。注意：InMemoryStore 的数据仅存在于进程内存中，进程结束后即丢失。生产环境应使用 PostgresStore 等数据库存储。
3. 定义工具
   get_user_info 是一个通过 @tool 装饰器定义的函数。它接受一个 runtime: ToolRuntime[Context] 参数，智能体会自动注入该参数。工具内部：
   • 通过 runtime.store 访问存储对象（与创建智能体时传入的 store 相同）。
   • 从 runtime.context.user_id 获取当前用户的 ID。
   • 调用 store.get(("users",), user_id) 精确查询存储。命名空间 ("users",) 与写入时保持一致，键为用户 ID。
   • 如果查询到结果（返回 StoreValue 对象），则提取其 .value 属性并转换为字符串返回；否则返回 "Unknown user"。
4. 创建智能体
   调用 create_agent 时，将 store 和 context_schema=Context 传入。这使得智能体内部能够正确管理存储，并在调用工具时构建 ToolRuntime 对象，其中包含 store 和 context。
5. 运行智能体
   通过 agent.invoke 发送用户消息，并同时提供 context=Context(user_id="user_123")。智能体会自动将用户消息与上下文一起处理，当需要调用 get_user_info 工具时，runtime.context.user_id 会被设置为 "user_123"，从而检索到预先存储的用户信息。

关键点：

• 存储的命名空间：元组 ("users",) 是一个一级命名空间。你也可以使用更深的元组（如 ("users", "active")）来细分数据。
• 存储的键：这里使用了用户 ID 作为键，使得每个用户的信息可以独立存储和读取。
• 工具中的断言：assert runtime.store is not None 用于类型收窄，因为 ToolRuntime.store 在静态类型中是可选的，但在传入 store 后实际运行时一定存在。
• 返回值处理：store.get() 可能返回 None，因此需要判空。

扩展方向：

• 写入记忆：类似地，可以在工具中调用 store.put() 来保存用户提供的信息（例如用户说“我叫张三”时，智能体调用工具写入存储）。
• 语义搜索：如果存储配置了索引（IndexConfig），可以使用 store.search() 根据查询文本的语义相关性检索记忆，而不仅仅是精确键查找。
• 生产部署：将 InMemoryStore 替换为 PostgresStore 或 RedisStore，并处理好数据库连接池和事务。

读取PostgreSQL记忆

# 导入 dataclass 装饰器，用于定义上下文数据结构
from dataclasses import dataclass

# 导入智能体创建函数
from langchain.agents import create_agent
# 导入工具运行时类型和工具装饰器
from langchain.tools import ToolRuntime, tool
# 导入 Runnable 类型提示，用于标注智能体对象
from langchain_core.runnables import Runnable
# 导入 PostgreSQL 长期记忆存储（生产环境推荐），忽略类型检查器可能报的缺少导入错误
from langgraph.store.postgres import PostgresStore  # type: ignore[import-not-found]


# 定义上下文数据类，包含用户ID，智能体会将此类实例注入到工具中
@dataclass
class Context:
    user_id: str


# PostgreSQL 数据库连接字符串（请根据实际环境修改）
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"

# 使用 with 语句管理 PostgresStore 连接，确保退出时自动释放资源
with PostgresStore.from_conn_string(DB_URI) as store:
    # 初始化数据库：创建必要的表结构和索引（首次运行或数据库结构变更时执行）
    store.setup()

    # 写入一条示例用户数据到存储
    # 命名空间为 ("users",)，键为用户 ID "user_123"，值为用户信息字典
    store.put(("users",), "user_123", {"name": "John Smith", "language": "English"})

    # 定义工具：通过运行时上下文获取当前用户信息
    @tool
    def get_user_info(runtime: ToolRuntime[Context]) -> str:
        """查找用户信息。"""
        # runtime.store 与创建智能体时传入的 store 是同一个对象
        assert runtime.store is not None
        # 从存储中检索当前用户的信息：命名空间 ("users",)，键为 runtime.context.user_id
        user_info = runtime.store.get(("users",), runtime.context.user_id)
        # 如果找到数据则返回其字符串形式，否则返回 "Unknown user"
        return str(user_info.value) if user_info else "Unknown user"

    # 创建支持长期记忆的智能体
    agent: Runnable = create_agent(
        "claude-sonnet-4-6",          # 使用的大语言模型
        tools=[get_user_info],         # 工具列表
        store=store,                   # 传入 PostgreSQL 存储，使智能体具备持久化记忆能力
        context_schema=Context,        # 声明上下文的数据结构，用于运行时注入
    )

    # 运行智能体：发送用户消息，并通过 context 参数传递当前用户上下文
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "look up user information"}]},
        context=Context(user_id="user_123"),
    )

代码说明：这段代码演示了如何在生产环境中为 LangChain 智能体配置 PostgreSQL 长期记忆存储，并在工具内部读取记忆。与之前使用 InMemoryStore 的示例不同，这里使用了 PostgresStore，确保记忆数据在智能体重启或不同服务器之间依然持久存在。

核心流程

1. 定义上下文
   Context 数据类包含 user_id 字段。智能体在调用 invoke 时通过 context 参数传入该对象，并在运行工具时提供给 runtime.context。
2. 初始化 PostgreSQL 存储
   • 使用 PostgresStore.from_conn_string(DB_URI) 创建基于连接字符串的存储实例。
   • with 语句确保在代码块退出时数据库连接被正确关闭，避免资源泄漏。
   • store.setup() 创建必要的数据库表（如 store 表）和索引。此操作只需执行一次（例如在应用部署或数据库迁移脚本中），后续运行可注释或跳过。
   • store.put(("users",), "user_123", {...}) 向存储中写入一条示例用户数据。命名空间 ("users",) 用于组织用户信息，键为 user_123，值为包含姓名和语言偏好的字典。
3. 定义工具
   get_user_info 是一个通过 @tool 装饰器定义的函数。它接受 runtime: ToolRuntime[Context] 参数，智能体会自动注入。工具内部：
   • 通过 runtime.store 访问存储对象（与创建智能体时传入的 store 相同）。
   • 从 runtime.context.user_id 获取当前用户的 ID。
   • 调用 store.get(("users",), user_id) 精确查询用户信息。
   • 如果查询到结果（StoreValue 对象），则提取其 .value 属性并转换为字符串返回；否则返回 "Unknown user"。
4. 创建智能体
   • 指定模型 "claude-sonnet-4-6"。
   • 将 get_user_info 工具加入工具列表。
   • 传入 store 参数，使智能体能够访问长期记忆。
   • 指定 context_schema=Context，让智能体知道上下文的结构。
5. 运行智能体
   • 发送用户消息 "look up user information"。
   • 同时提供 context=Context(user_id="user_123")。
   • 智能体会解析用户意图，决定调用 get_user_info 工具，并传入正确的上下文，最终从数据库中检索出预先存储的用户信息。

关键点：

• 持久化：PostgresStore 将数据写入 PostgreSQL 数据库，即使进程终止或服务重启，记忆也不会丢失。这适用于需要长期保存用户偏好、历史记录等信息的场景。
• 资源管理：with 语句确保数据库连接在使用后正确关闭。生产环境中，推荐使用连接池（例如通过 asyncpg 或 psycopg2.pool）来提高并发性能。
• 存储操作：store.put 和 store.get 分别对应写入和精确读取。命名空间和键的设计应能唯一标识一组记忆。
• 工具注入：ToolRuntime 类型携带了 store 和 context，工具内部无需关心这些依赖的来源，只需使用即可。
• 初始化：store.setup() 只需要在数据库未初始化时调用一次。可以在代码中判断表是否存在，也可以使用数据库迁移工具（如 Alembic）来管理。

扩展建议：

• 语义搜索：如果需要根据查询内容的相似性检索记忆，可以在创建 PostgresStore 时传入 index=IndexConfig(embed=embed_func, dims=vector_dim)，然后使用 store.search() 方法。
• 写入记忆：类似地，可以定义另一个工具，接收用户提供的信息并调用 store.put() 写入存储。
• 错误处理：在实际应用中，应添加 try-except 块处理数据库连接失败、序列化错误等异常。

在工具中写入长期记忆

记忆写入内存

# 导入 dataclass 装饰器，用于定义上下文数据结构
from dataclasses import dataclass

# 导入智能体创建函数
from langchain.agents import create_agent
# 导入工具运行时类型和工具装饰器
from langchain.tools import ToolRuntime, tool
# 导入 Runnable 类型提示，用于标注智能体对象
from langchain_core.runnables import Runnable
# 导入内存存储（仅用于开发/测试，生产环境请使用数据库存储）
from langgraph.store.memory import InMemoryStore
# 导入 TypedDict，用于定义 LLM 期望的参数结构
from typing_extensions import TypedDict

# InMemoryStore 将数据保存到内存字典中。生产环境中请使用数据库支持的存储（如 PostgresStore）
store = InMemoryStore()


# 定义上下文数据类，包含用户ID，智能体会将此类实例注入到工具中
@dataclass
class Context:
    user_id: str


# TypedDict 定义了用户信息的结构，帮助 LLM 理解如何调用工具
class UserInfo(TypedDict):
    name: str   # 用户姓名


# 定义工具：允许智能体更新用户信息（适用于聊天应用）
@tool
def save_user_info(user_info: UserInfo, runtime: ToolRuntime[Context]) -> str:
    """保存用户信息。"""
    # runtime.store 与创建智能体时传入的 store 是同一个对象
    assert runtime.store is not None
    store = runtime.store               # 获取存储引用
    user_id = runtime.context.user_id   # 从上下文中获取当前用户ID
    # 将数据存入存储：命名空间 ("users",)，键为 user_id，值为 user_info 字典
    store.put(("users",), user_id, dict(user_info))
    return "Successfully saved user info."


# 创建支持长期记忆的智能体
agent: Runnable = create_agent(
    model="openai:gpt-5.4",          # 使用的大语言模型
    tools=[save_user_info],          # 工具列表
    store=store,                     # 传入存储对象，使智能体在运行工具时能够访问存储
    context_schema=Context,          # 声明上下文的数据结构，用于运行时注入
)

# 运行智能体：用户说“我叫 John Smith”
agent.invoke(
    {"messages": [{"role": "user", "content": "My name is John Smith"}]},
    # 在 context 中传递 user_id，以标识要更新哪个用户的信息
    context=Context(user_id="user_123"),
)

# 可以直接访问存储来获取刚刚保存的值，用于验证
item = store.get(("users",), "user_123")

代码说明：这段代码演示了如何在 LangChain 智能体的工具内部写入长期记忆。它使用内存存储（InMemoryStore）作为示例，展示了智能体如何从用户对话中提取信息并将其保存到存储中，以便后续对话使用。

核心流程：

1. 定义上下文
   使用 @dataclass 定义 Context 类，包含 user_id 字段。智能体在调用 invoke 时通过 context 参数传入该对象，并在运行工具时将其提供给 runtime.context。这确保了每个工具调用都知道是针对哪个用户的操作。

2. 定义输入结构
   UserInfo 是一个 TypedDict，它告诉 LLM：调用 save_user_info 工具时需要提供一个包含 name 字段的字典。LLM 会根据用户消息（例如“My name is John Smith”）自动提取该信息并填充参数。

3. 初始化存储
   store = InMemoryStore() 创建内存存储。此处没有预先写入数据，因为工具将在运行时动态写入。注意：InMemoryStore 仅用于开发/测试，生产环境应替换为 PostgresStore 或 RedisStore 等持久化存储。

4. 定义工具
   save_user_info 是一个通过 @tool 装饰器定义的函数，它接受两个参数：

   • user_info: UserInfo：由 LLM 从对话中提取并填充的数据。
   • runtime: ToolRuntime[Context]：由智能体自动注入的运行时对象，提供 store 和 context 访问。

   工具内部：

   • 通过 runtime.store 获取存储对象。
   • 从 runtime.context.user_id 获取当前用户 ID。
   • 调用 store.put(("users",), user_id, dict(user_info)) 将用户信息写入存储。命名空间为 ("users",)，键为用户 ID，值为包含 name 的字典。
   • 返回成功消息。

5. 创建智能体
   调用 create_agent 时传入 model、tools、store 和 context_schema。其中 store 使智能体能够访问持久化记忆，context_schema 告知智能体上下文的结构。

6. 运行智能体
   用户发送消息 "My name is John Smith"，同时通过 context=Context(user_id="user_123") 指定用户 ID。智能体会将消息和上下文一起处理：

   • LLM 理解用户正在提供姓名信息。
   • LLM 决定调用 save_user_info 工具，并生成参数 user_info={"name": "John Smith"}。
   • 智能体调用该工具，传入 runtime（其中 runtime.context.user_id 为 "user_123"）。
   • 工具执行 store.put(...)，将数据写入内存存储。
   • 工具返回成功消息，LLM 可能将其作为回复的一部分呈现给用户。

7. 验证
   store.get(("users",), "user_123") 直接访问存储，可以看到刚刚写入的 {"name": "John Smith"}。

关键点：

• 写入记忆：与读取记忆的工具（get_user_info）不同，此工具展示了如何动态写入用户提供的信息。这是构建能够“记住”用户偏好的对话智能体的基础。
• TypedDict 的作用：UserInfo 为 LLM 提供了清晰的参数结构，提高了工具调用的准确性和可靠性。LLM 会自动将用户消息中的姓名提取并映射到 name 字段。
• 存储的命名空间：("users",) 作为命名空间，使得用户数据集中存储，便于管理。如果需要区分不同应用场景（如“闲聊” vs “技术支持”），可以使用更复杂的命名空间元组。
• 上下文传递：context_schema=Context 和 invoke 时的 context 参数配合，确保了工具能够获取到当前用户 ID，从而正确地将信息归属于正确的用户。
• 持久化：示例中使用 InMemoryStore，进程重启后数据会丢失。生产环境中应替换为 PostgresStore 并调用 store.setup() 初始化数据库表。

扩展建议：

• 更丰富的数据：UserInfo 可以包含更多字段，例如 age、preferences 等，LLM 会在对话中识别并填充。
• 读取记忆：可以同时提供 get_user_info 工具，让智能体既能读取也能写入，形成完整的记忆循环。
• 语义搜索：如果存储配置了向量索引（IndexConfig），还可以实现基于语义的模糊查询，例如“找回用户提到的关于编程语言偏好的信息”。

记忆写入PostgreSQL

# 导入 dataclass 装饰器，用于定义上下文数据结构
from dataclasses import dataclass

# 导入智能体创建函数
from langchain.agents import create_agent
# 导入工具运行时类型和工具装饰器
from langchain.tools import ToolRuntime, tool
# 导入 Runnable 类型提示，用于标注智能体对象
from langchain_core.runnables import Runnable
# 导入 PostgreSQL 长期记忆存储（生产环境推荐），忽略类型检查器可能的导入报错
from langgraph.store.postgres import PostgresStore  # type: ignore[import-not-found]
# 导入 TypedDict，用于定义 LLM 期望的工具参数结构
from typing_extensions import TypedDict


# 定义上下文数据类，包含用户ID，智能体会将此类实例注入到工具中
@dataclass
class Context:
    user_id: str


# TypedDict 定义了用户信息的结构，帮助 LLM 理解如何调用工具
class UserInfo(TypedDict):
    name: str   # 用户姓名


# 定义工具：允许智能体更新用户信息（适用于聊天应用）
@tool
def save_user_info(user_info: UserInfo, runtime: ToolRuntime[Context]) -> str:
    """保存用户信息。"""
    # runtime.store 与创建智能体时传入的 store 是同一个对象
    assert runtime.store is not None
    # 将数据存入存储：命名空间 ("users",)，键为当前用户的 ID，值为 user_info 字典
    runtime.store.put(("users",), runtime.context.user_id, dict(user_info))
    return "Successfully saved user info."


# PostgreSQL 数据库连接字符串（请根据实际环境修改用户名、密码、主机、数据库名）
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"

# 使用 with 语句管理 PostgresStore 连接，确保退出时自动释放资源
with PostgresStore.from_conn_string(DB_URI) as store:
    # 初始化数据库：创建必要的表结构和索引（首次运行或数据库结构变更时执行）
    store.setup()

    # 创建支持长期记忆的智能体
    agent: Runnable = create_agent(
        "claude-sonnet-4-6",      # 使用的大语言模型
        tools=[save_user_info],    # 工具列表（仅包含保存用户信息的工具）
        store=store,               # 传入 PostgreSQL 存储，使智能体具备持久化记忆能力
        context_schema=Context,    # 声明上下文的数据结构，用于运行时注入
    )

    # 运行智能体：用户发送消息 "My name is John Smith"
    agent.invoke(
        {"messages": [{"role": "user", "content": "My name is John Smith"}]},
        context=Context(user_id="user_123"),   # 在上下文中传递用户 ID，标识要更新哪个用户的信息
    )

代码说明：这段代码演示了如何在生产环境中为 LangChain 智能体配置 PostgreSQL 长期记忆存储，并在工具内部动态写入记忆。与使用 InMemoryStore 的示例不同，这里使用了 PostgresStore，确保用户信息即使在智能体重启或服务扩缩容后依然持久存在。

核心流程

1. 定义上下文
   Context 数据类包含 user_id 字段。智能体在调用 invoke 时通过 context 参数传入该对象，并在运行工具时提供给 runtime.context。这确保了工具能够知道当前是哪个用户在提供信息。
2. 定义工具输入结构
   UserInfo 是一个 TypedDict，它告诉 LLM：调用 save_user_info 工具时需要提供一个包含 name 字段的字典。LLM 会根据用户消息（例如“My name is John Smith”）自动提取该信息并填充参数。
3. 定义写入工具
   save_user_info 工具：
   • 接受 user_info: UserInfo（由 LLM 自动填充）和 runtime: ToolRuntime[Context]（由智能体自动注入）。
   • 通过 runtime.store 访问 PostgreSQL 存储。
   • 调用 store.put(("users",), runtime.context.user_id, dict(user_info)) 将用户信息写入数据库。命名空间为 ("users",)，键为当前用户的 ID（例如 "user_123"），值为包含用户姓名的字典。
   • 返回成功消息。
4. 初始化 PostgreSQL 存储
   • 使用 PostgresStore.from_conn_string(DB_URI) 创建存储实例。
   • with 语句确保数据库连接在代码块结束后被正确关闭。
   • store.setup() 创建必要的数据库表（如 store 表）和索引。注意：此操作只需在首次部署或数据库结构变更时执行一次，后续运行可注释或跳过。
5. 创建并运行智能体
   • 调用 create_agent 时传入模型 "claude-sonnet-4-6"、工具 [save_user_info]、store 和 context_schema=Context。
   • 用户发送消息 "My name is John Smith"，同时通过 context=Context(user_id="user_123") 指定当前用户 ID。
   • LLM 理解到用户在提供姓名，于是调用 save_user_info 工具，生成参数 user_info={"name": "John Smith"}。
   • 工具执行后，数据库中会存储一条与用户 ID "user_123" 关联的姓名记录。
   • 工具返回成功消息，LLM 可能将其转换为自然语言回复给用户（例如“好的，我已经记住您的名字是 John Smith”）。

关键点：

• 持久化写入：store.put() 将数据写入 PostgreSQL，后续任何会话或线程都可以通过 store.get(("users",), user_id) 读取到该信息。
• 资源管理：with 语句确保数据库连接正确释放。生产环境通常使用连接池（例如通过 psycopg2.pool 或 asyncpg），这里为了简洁使用 with 也是可行的。
• 数据库初始化：store.setup() 只需要运行一次。在实际生产部署中，推荐将数据库表的创建放在独立的迁移脚本中（例如使用 Alembic），避免每次启动应用都尝试创建表。
• TypedDict 的价值：UserInfo 为 LLM 提供了明确的工具参数结构，提高了调用的准确性。LLM 会从用户消息中自动提取 name 字段的值并正确映射。
• 灵活性：可以轻松扩展 UserInfo 以包含更多字段（如 age、preferences），LLM 会自动识别并填充。

内存和PgStore存储的对比

特性	InMemoryStore（前例）	PostgresStore（本例）
数据持久化	仅在进程生命周期内存在	永久持久化到磁盘
适用环境	开发、测试	生产环境
跨进程/服务器共享	否	是
额外依赖	无	PostgreSQL 驱动 （psycopg 或 asyncpg）

扩展建议：

• 添加读取工具：可以定义一个 get_user_info 工具，让智能体在需要时主动读取之前存储的用户信息。
• 语义搜索：如需根据相似度检索记忆，可在创建 PostgresStore 时传入 index=IndexConfig(embed=embed_func, dims=vector_dim)，并使用 store.search() 方法。
• 错误处理：在实际应用中应添加 try-except 块，处理数据库连接失败、序列化错误等异常情况。