核心定位

LangMem 填补了 LLM 代理在以下方面的能力空白：

资料来源：src/langmem/prompts/types.py:1-15

整体架构

LangMem 的架构围绕三大核心模块展开，每个模块负责不同的记忆或优化职责：

graph TD
    A[LangMem] --> B[knowledge<br/>长期记忆模块]
    A --> C[short_term<br/>短期记忆模块]
    A --> D[prompts<br/>提示优化模块]
    
    B --> B1[extraction.py<br/>记忆提取管理]
    B --> B2[tools.py<br/>记忆操作工具]
    
    C --> C1[summarization.py<br/>对话摘要]
    
    D --> D1[optimization.py<br/>优化器工厂]
    D --> D2[gradient.py<br/>梯度优化器]
    D --> D3[prompt.py<br/>提示模板]
    D --> D4[types.py<br/>类型定义]

核心数据类型

LangMem 使用 TypeScript 风格的 TypedDict 定义数据结构，确保类型安全和清晰的接口契约。

资料来源：src/langmem/prompts/types.py:1-90

Prompt（提示定义）

class Prompt(TypedDict, total=False):
    name: Required[str]                    # 提示名称，唯一标识
    prompt: Required[str]                 # 提示内容
    update_instructions: str | None       # 更新指南
    when_to_update: str | None            # 更新依赖条件

AnnotatedTrajectory（标注轨迹）

用于记录对话历史及其反馈信息：

class AnnotatedTrajectory(typing.NamedTuple):
    messages: list[AnyMessage]           # 对话消息列表
    feedback: dict[str, typing.Any]      # 反馈信息（如 developer_feedback、score）

优化器输入类型

资料来源：src/langmem/prompts/types.py:50-120

长期记忆模块

长期记忆模块负责从对话中提取、存储和检索用户偏好及知识信息。

资料来源：src/langmem/knowledge/extraction.py:1-100

核心组件

#### MemoryManager（记忆管理器）

create_memory_manager 函数创建核心的记忆管理器：

def create_memory_manager(
    model: str | BaseChatModel,           # 主模型，用于提取和综合
    schemas: list[type],                  # 记忆的 Pydantic schema 定义
    *,
    instructions: str | None = None,      # 自定义提取指令
    enable_inserts: bool = True,          # 允许创建新记忆
    enable_updates: bool = True,          # 允许更新现有记忆
    enable_deletes: bool = True,          # 允许删除过时记忆
) -> MemoryManager

功能特性：

• 插入（Insert）：从新对话中创建新的记忆条目
• 更新（Update）：修改现有记忆的内容或属性
• 删除（Delete）：移除不再相关或过时的记忆
• 多步迭代：支持设置 max_steps 进行深度提取和综合

资料来源：src/langmem/knowledge/extraction.py:80-120

#### MemoryStoreManager（存储管理器）

create_memory_store_manager 创建与 LangGraph 存储层集成的管理器：

def create_memory_store_manager(
    model: str | BaseChatModel,           # 主模型
    schemas: list[type],                  # 记忆 schema
    *,
    query_model: str | None = None,      # 用于搜索的专用模型（更快更便宜）
    query_limit: int = 10,                # 搜索返回结果数量
    namespace: tuple[str, ...] = ("memories", "{langgraph_user_id}"),
    store: BaseStore | None = None,       # LangGraph 存储后端
) -> MemoryStoreManager

搜索机制：

1. 直接嵌入新消息进行相似度搜索
2. 或使用专用查询模型（query_model）生成优化的搜索查询

资料来源：src/langmem/knowledge/extraction.py:50-80

记忆搜索工具

资料来源：src/langmem/knowledge/tools.py:1-80

create_memory_searcher 函数创建语义搜索管道：

def create_memory_searcher(
    model: str | BaseChatModel,
    prompt: str = "Search for distinct memories relevant to different aspects of the provided context.",
    namespace: tuple[str, ...] = ("memories", "{langgraph_user_id}"),
) -> Runnable[MessagesState, typing.Awaitable[list[SearchItem]]]

记忆管理工具

LangMem 提供可集成到 LangGraph Agent 的工具：

#### 记忆操作工具

create_manage_memory_tool(
    namespace: str | tuple[str, ...],
    schema: type[BaseModel],              # 记忆的 schema 类
    actions_permitted: list[Literal["create", "update", "upsert", "delete"]],
    store: BaseStore | None = None,
)

支持的操作类型：

#### 记忆搜索工具

create_search_memory_tool(
    namespace: str | tuple[str, ...],
    store: BaseStore | None = None,
)

支持参数：

• query: 自然语言搜索查询
• limit: 返回结果数量限制
• offset: 分页偏移
• filter: 元数据过滤条件

资料来源：src/langmem/knowledge/tools.py:80-150

短期记忆模块

短期记忆模块处理对话过程中的实时信息压缩和摘要。

资料来源：src/langmem/short_term/summarization.py:1-80

SummarizationNode（摘要节点）

用于 LangGraph 工作流中的消息摘要：

summarization_node = SummarizationNode(
    model=summarization_model,            # 专用的轻量摘要模型
    max_tokens=256,                        # 摘要最大 token 数
    max_tokens_before_summary=256,         # 触发摘要的消息长度阈值
    max_summary_tokens=128,                # 单次摘要最大 token
)

RunningSummary（运行摘要）

状态管理对象，用于跟踪和管理对话摘要：

class RunningSummary:
    summary: str | None                    # 当前摘要内容
    key_points: list[str]                  # 关键要点列表
    message_count: int                     # 已处理消息数

典型使用流程：

1. 在 LangGraph 状态图中添加 summarization_node
2. 节点接收 summarized_messages_key 输入
3. 返回包含摘要消息和上下文更新的状态

资料来源：src/langmem/short_term/summarization.py:20-100

提示优化模块

提示优化模块基于对话历史和反馈自动改进系统提示。

资料来源：src/langmem/prompts/optimization.py:1-80

优化策略类型

单提示优化器

def create_prompt_optimizer(
    model: str | BaseChatModel,
    kind: Literal["gradient", "metaprompt", "prompt_memory"] = "gradient",
    config: dict | None = None,
) -> PromptOptimizer

输入结构：

{
    "trajectories": [
        AnnotatedTrajectory(
            messages=[...],
            feedback={"clarity": "needs more structure"}
        )
    ],
    "prompt": "You are an astronomy expert"
}

资料来源：src/langmem/prompts/optimization.py:80-120

多提示优化器

def create_multi_prompt_optimizer(
    model: str | BaseChatModel,
    kind: Literal["gradient", "metaprompt", "prompt_memory"] = "gradient",
    config: dict | None = None,
) -> MultiPromptOptimizer

输入结构：

{
    "trajectories": [...],                 # 多个标注轨迹
    "prompts": [
        Prompt(name="research", prompt="Research the given topic"),
        Prompt(name="summarize", prompt="Summarize findings"),
    ]
}

梯度优化器详解

资料来源：src/langmem/prompts/gradient.py:1-60

梯度优化器通过迭代的"假设-验证"循环改进提示：

class GradientOptimizerConfig(TypedDict, total=False):
    gradient_prompt: str                   # 梯度优化提示模板
    metaprompt: str                        # 元提示模板
    max_reflection_steps: int              # 最大反思步数
    min_reflection_steps: int              # 最小反思步数

优化流程：

1. 分析当前提示和对话轨迹
2. 生成性能不佳的假设原因
3. 基于假设提出调整建议
4. 应用最小侵入性的修改
5. 迭代验证直到满足停止条件

资料来源：src/langmem/prompts/gradient.py:20-80

提示模板

资料来源：src/langmem/prompts/prompt.py:1-50

LangMem 预定义了两类优化提示模板：

#### 单轨迹反思模板

INSTRUCTION_REFLECTION_PROMPT = """You are helping an AI agent improve. You can do this by changing their system prompt.

These is their current prompt:
<current_prompt>
{current_prompt}
</current_prompt>

Here was the agent's trajectory:
<trajectory>
{trajectory}
</trajectory>

Here is the user's feedback:

<feedback>
{feedback}
</feedback>
...
"""

#### 多轨迹批量反思模板

INSTRUCTION_REFLECTION_MULTIPLE_PROMPT = """..."""
# 使用 <data> 标签批量包含多个轨迹-反馈对

响应格式：

class GeneralResponse(TypedDict):
    logic: str                             # 优化逻辑说明
    update_prompt: bool                    # 是否需要更新
    new_prompt: str                        # 新提示内容

与 LangGraph 集成

LangMem 被设计为与 LangGraph 无缝协作，利用 LangGraph 的以下组件：

命名空间机制

记忆使用层级命名空间组织：

namespace = ("memories", "{langgraph_user_id}")
# 示例: ("memories", "user123")

命名空间模板支持运行时占位符替换，确保多用户数据隔离。

后台反射执行器

from langmem import ReflectionExecutor

reflection = ReflectionExecutor(manager, store=store)

ReflectionExecutor 支持将记忆处理任务后台化：

sequenceDiagram
    participant Agent
    participant Background
    participant Store

    Agent->>Agent: process message
    Agent-->>User: response
    Agent->>Background: schedule enrichment<br/>(after_seconds=0)
    Note over Background,Store: Memory processing happens<br/>in background thread

API 速查表

创建函数一览

总结

LangMem 提供了一套完整的记忆和提示优化解决方案：

• 长期记忆：通过结构化 schema 和语义搜索实现持久化知识管理
• 短期记忆：通过实时摘要保持对话上下文的高效利用
• 提示优化：通过多种策略（梯度、元提示、记忆学习）持续改进代理行为

该库与 LangGraph 深度集成，既可以作为独立组件使用，也能作为 LangGraph 代理的能力扩展。

概述

LangMem 是一个用于构建具有记忆能力的大语言模型应用的 Python 库。该库提供了一套完整的工具，用于管理对话记忆、提示优化和短期记忆总结等功能。安装与配置章节将详细介绍如何正确设置 LangMem 环境、配置依赖项、以及初始化各类管理器组件。

环境要求

Python 版本要求

LangMem 需要 Python 3.10 或更高版本。建议使用 Python 3.11 以获得最佳性能和兼容性。

核心依赖

LangMem 的核心依赖包括：

安装方式

使用 uv 安装（推荐）

uv add langmem

使用 pip 安装

pip install langmem

从源码安装

cd examples/standalone_examples
uv venv
source .venv/bin/activate
uv sync

环境变量配置

API 密钥设置

使用 LangMem 前需要配置相应的 API 密钥：

export OPENAI_API_KEY=your_api_key_here
export ANTHROPIC_API_KEY=your_api_key_here  # 如使用 Anthropic 模型

核心组件配置

记忆管理器 (Memory Manager)

create_memory_manager 函数用于创建记忆管理器，支持插入、更新和删除记忆操作。资料来源：src/langmem/knowledge/extraction.py:1-200

#### 基础配置示例

from langmem import create_memory_manager
from pydantic import BaseModel

class PreferenceMemory(BaseModel):
    kind: str = "PreferenceMemory"
    preference: str
    context: str

manager = create_memory_manager(
    "anthropic:claude-3-5-sonnet-latest",
    schemas=[PreferenceMemory],
)

#### 配置参数说明

记忆存储管理器 (Memory Store Manager)

create_memory_store_manager 函数创建支持持久化存储的记忆管理器，支持同步和异步操作。资料来源：src/langmem/knowledge/extraction.py:1-200

#### 配置示例

from langmem import create_memory_store_manager
from langgraph.store.memory import InMemoryStore
from langgraph.func import entrypoint

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)

manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",
    query_model="anthropic:claude-3-5-haiku-latest",
    query_limit=10,
    namespace=("memories", "{langgraph_user_id}"),
)

#### 命名空间配置

命名空间使用模板占位符，支持运行时配置：

namespace = ("memories", "{langgraph_user_id}")

在调用时通过配置传入实际值：

config = {"configurable": {"langgraph_user_id": "user123"}}

记忆搜索器 (Memory Searcher)

create_memory_searcher 创建自动生成查询的记忆搜索管道。资料来源：src/langmem/knowledge/extraction.py:1-200

from langmem import create_memory_searcher

searcher = create_memory_searcher(
    "anthropic:claude-3-5-sonnet-latest",
    prompt="Search for distinct memories relevant to different aspects of the provided context.",
    namespace=("memories", "{langgraph_user_id}"),
)

记忆管理工具 (Manage Memory Tool)

create_manage_memory_tool 创建可集成到 LangGraph ReAct Agent 的记忆管理工具。资料来源：src/langmem/knowledge/tools.py:1-100

from langmem import create_manage_memory_tool
from langgraph.prebuilt import create_react_agent

agent = create_react_agent(
    "anthropic:claude-3-5-sonnet-latest",
    tools=[
        create_manage_memory_tool(
            namespace=("memories", "{langgraph_user_id}"),
            actions_permitted=["create", "update", "delete"],
        ),
    ],
    store=store,
)

提示优化器 (Prompt Optimizer)

LangMem 提供多种提示优化策略。资料来源：src/langmem/prompts/optimization.py:1-150

#### 单一提示优化

from langmem import create_prompt_optimizer

optimizer = create_prompt_optimizer("anthropic:claude-3-5-sonnet-latest")

better_prompt = await optimizer.ainvoke({
    "trajectories": [(conversation, feedback)],
    "prompt": "You are an astronomy expert",
})

#### 多提示优化

from langmem import create_multi_prompt_optimizer

optimizer = create_multi_prompt_optimizer(
    "anthropic:claude-3-5-sonnet-latest",
    kind="metaprompt",  # 或 "gradient", "prompt_memory"
    config={"max_reflection_steps": 3, "min_reflection_steps": 1},
)

#### 优化类型配置

短期记忆总结

SummarizationNode 用于 LangGraph 状态图中的消息总结。资料来源：src/langmem/short_term/summarization.py:1-100

from langmem.short_term import SummarizationNode, RunningSummary

summarization_node = SummarizationNode(
    model=summarization_model,
    max_tokens=256,
    max_tokens_before_summary=256,
    max_summary_tokens=128,
)

数据类型定义

Prompt 类型

用于结构化提示管理和优化的 TypedDict。资料来源：src/langmem/prompts/types.py:1-80

from langmem.prompts.types import Prompt

prompt = Prompt(
    name="extract_entities",
    prompt="Extract key entities from the text:",
    update_instructions="Make minimal changes",
    when_to_update="If there seem to be errors",
)

AnnotatedTrajectory 类型

包含可选反馈的对话历史记录。资料来源：src/langmem/prompts/types.py:1-80

from langmem.prompts.types import AnnotatedTrajectory

trajectory = AnnotatedTrajectory(
    messages=[
        {"role": "user", "content": "What pizza is good?"},
        {"role": "assistant", "content": "Try LangPizza"},
    ],
    feedback={"developer_feedback": "too pushy", "score": 0},
)

架构流程图

记忆管理流程

graph TD
    A[用户消息] --> B[记忆管理器]
    B --> C{操作类型}
    C -->|创建| D[提取新记忆]
    C -->|更新| E[合并现有记忆]
    C -->|删除| F[移除过时记忆]
    D --> G[存储到向量数据库]
    E --> G
    F --> G
    G --> H[返回结果]

提示优化流程

graph TD
    A[对话历史] --> B[提示优化器]
    B --> C[元提示优化]
    B --> D[梯度优化]
    B --> E[记忆驱动优化]
    C --> F[优化后的提示]
    D --> F
    E --> F

集成配置

LangGraph 集成

LangMem 可以与 LangGraph 的 ReAct Agent 无缝集成：

from langgraph.prebuilt import create_react_agent
from langmem import create_memory_store_manager

manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",
    namespace=("memories", "{langgraph_user_id}"),
)

agent = create_react_agent(
    "anthropic:claude-3-5-sonnet-latest",
    tools=[create_manage_memory_tool(namespace=("memories", "{langgraph_user_id}"))],
    store=store,
)

后台执行配置

使用 ReflectionExecutor 在后台异步执行记忆处理：

from langmem import create_memory_store_manager, ReflectionExecutor

reflection = ReflectionExecutor(manager, store=store)

最佳实践

1. 模型选择：主模型建议使用功能强大的模型（如 claude-3-5-sonnet），查询模型可使用更快速的版本（如 claude-3-5-haiku）
2. 命名空间设计：使用有意义的层级命名空间，便于记忆的组织和检索
3. Schema 定义：为不同类型的记忆定义清晰的 Pydantic Schema，确保数据结构的一致性
4. 错误处理：实现适当的错误处理机制，处理 API 超时和配额限制
5. 异步操作：优先使用异步方法（ainvoke, aput, asearch）以提高应用性能

功能概述

热路径快速开始模块主要提供以下核心能力：

核心 API

create_manage_memory_tool

创建记忆管理工具的主函数，定义在 src/langmem/knowledge/tools.py 中。

def create_manage_memory_tool(
    namespace: tuple[str, ...] = ("memories", "{langgraph_user_id}"),
    actions_permitted: typing.Sequence[
        typing.Literal["create", "update", "delete"]
    ] = ("create", "update", "delete"),
    schema: type[BaseModel] | None = None,
    store: BaseStore | None = None,
) -> Tool:

参数说明：

资料来源：src/langmem/knowledge/tools.py:36-43

工作流程

基础集成流程

graph TD
    A[创建工具] --> B[配置命名空间]
    B --> C[设置操作权限]
    C --> D[集成到 Agent]
    D --> E[运行时执行]

工具执行流程

当智能体调用记忆管理工具时，内部执行以下流程：

sequenceDiagram
    participant Agent as LangGraph Agent
    participant Tool as Memory Tool
    participant Store as BaseStore

    Agent->>Tool: 调用 manage_memory
    Tool->>Tool: 解析命名空间
    Tool->>Tool: 验证操作权限
    alt action == create
        Tool->>Store: put(key, value)
    else action == update
        Tool->>Store: update(key, value)
    else action == delete
        Tool->>Store: delete(key)
    end
    Store-->>Tool: 操作结果
    Tool-->>Agent: 返回结果

使用示例

基础用法

创建支持所有操作的记忆管理工具：

from langmem.knowledge import create_manage_memory_tool
from langgraph.prebuilt import create_react_agent

tool = create_manage_memory_tool(
    namespace=("memories", "{langgraph_user_id}"),
    actions_permitted=["create", "update", "delete"],
)

agent = create_react_agent(
    "anthropic:claude-3-5-sonnet-latest",
    tools=[tool],
    store=store,
)

资料来源：src/langmem/knowledge/tools.py:65-82

仅创建操作

限制工具只能创建新记忆：

tool = create_manage_memory_tool(
    namespace=("memories", "{user_id}"),
    actions_permitted=["create"],  # 仅允许创建
)

配合系统提示词使用

在 LangGraph 中获取记忆并注入到系统提示词：

from langgraph.prebuilt import create_react_agent
from langgraph.config import get_config, get_store

def prompt(state):
    config = get_config()
    memories = get_store().search(
        ("memories", config["configurable"]["langgraph_user_id"]),
    )
    system_prompt = f"""You are a helpful assistant.
<memories>
{memories}
</memories>"""
    system_message = {"role": "system", "content": system_prompt}
    return [system_message, *state["messages"]]

agent = create_react_agent(
    "anthropic:claude-3-5-sonnet-latest",
    tools=[create_manage_memory_tool(namespace=("memories", "{langgraph_user_id}"))],
    store=store,
)

资料来源：src/langmem/knowledge/tools.py:85-106

配置选项详解

命名空间配置

命名空间支持层级结构和运行时变量替换：

资料来源：src/langmem/knowledge/tools.py:36

操作权限矩阵

自定义 Schema

可以使用 Pydantic 模型定义记忆的数据结构：

from pydantic import BaseModel

class UserProfile(BaseModel):
    name: str
    age: int
    preferences: dict[str, str]

tool = create_manage_memory_tool(
    schema=UserProfile,
    actions_permitted=["create", "update"],
)

底层实现

函数签名与实现

create_manage_memory_tool 内部通过组合多个组件实现：

def create_manage_memory_tool(...):
    namespacer = utils.NamespaceTemplate(namespace)
    if not actions_permitted:
        raise ValueError("actions_permitted cannot be empty")
    action_type = typing.Literal[actions_permitted]
    default_action = "create" if "create" in actions_permitted else actions_permitted[0]
    initial_store = store

    async def amanage_memory(
        content: typing.Optional[schema] = None,
        action: action_type = default_action,
        *,
        id: typing.Optional[uuid.UUID] = None,
    ):
        store = _get_store(initial_store)
        # ... 执行相应操作

资料来源：src/langmem/knowledge/tools.py:48-65

异步管理函数

amanage_memory 是核心的异步处理函数：

与其他模块的关系

graph LR
    A[hot_path_quickstart] --> B[knowledge/tools.py]
    A --> C[prompts/types.py]
    A --> D[knowledge/extraction.py]
    
    B --> E[create_memory_manager]
    B --> F[create_memory_searcher]
    
    C --> G[Prompt 类型定义]
    C --> H[OptimizerInput 类型]
    
    D --> I[MemoryStoreManager]
    D --> J[SearchItem]

相关模块

最佳实践

1. 限制操作权限

根据业务需求限制允许的操作，避免过度权限：

# 推荐：按需授权
tool = create_manage_memory_tool(
    actions_permitted=["create"],  # 只允许创建
)

2. 使用 Schema 验证

对记忆内容进行结构化定义：

class PreferenceMemory(BaseModel):
    preference: str
    context: str
    category: str = "general"

tool = create_manage_memory_tool(schema=PreferenceMemory)

3. 合理设计命名空间

# 推荐：用户隔离
namespace=("app", "users", "{user_id}", "memories")

# 避免：过于扁平
namespace=("memories",)

错误处理

总结

热路径快速开始模块为 LangGraph 应用提供了便捷的记忆管理工具创建方式。通过 create_manage_memory_tool，开发者可以：

1. 快速集成：几行代码即可添加工具到智能体
2. 灵活配置：通过参数控制操作权限和数据结构
3. 类型安全：支持 Schema 验证和类型提示
4. 无缝衔接：与 LangGraph 生态完美配合

该模块是 LangMem 库的核心组件之一，适用于需要用户级记忆管理的对话应用场景。

概述

LangMem 的后台处理功能允许在独立线程中异步执行内存管理操作，从而避免阻塞主应用程序流程。通过 ReflectionExecutor 类，开发者可以将记忆提取和管理任务后台化，提升应用程序的响应速度和整体性能。

核心组件

ReflectionExecutor

ReflectionExecutor 是 LangMem 提供的后台执行器，用于在后台线程中处理记忆操作。

classDiagram
    class ReflectionExecutor {
        +execute(input, config)
        +ainvoke(input, config)
    }
    class MemoryManager {
        +ainvoke(input)
        +search()
    }
    class BaseStore {
        +asearch()
        +aput()
        +adelete()
    }
    ReflectionExecutor --> MemoryManager
    ReflectionExecutor --> BaseStore

导入路径：

from langmem import ReflectionExecutor

工作流程

异步处理架构

sequenceDiagram
    participant Agent as 代理
    participant Background as 后台执行器
    participant Store as 存储
    participant Manager as 记忆管理器

    Agent->>Agent: 处理消息
    Agent-->>User: 返回响应
    Agent->>Background: 调度后台任务 (after_seconds=0)
    Note over Background,Store: 记忆处理在后台线程执行
    Background->>Manager: 提取/更新记忆
    Manager->>Store: 持久化存储

消息流处理

当使用后台处理时，消息的处理流程如下：

1. 用户消息进入主线程
2. 代理生成响应
3. 响应与消息一起被发送到后台执行器
4. 后台执行器调度记忆管理任务
5. 记忆被提取并存储，不阻塞主线程

快速开始指南

环境准备

确保已安装必要的依赖：

uv venv
source .venv/bin/activate
uv sync

基本配置

from langmem import create_memory_store_manager, ReflectionExecutor
from langgraph.store.memory import InMemoryStore
from langgraph.prebuilt import create_react_agent

# 创建向量存储
store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)

# 创建记忆管理器
manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",
    namespace=("memories", "{user_id}")
)

# 创建后台执行器
reflection = ReflectionExecutor(manager, store=store)

创建带后台处理的代理

agent = create_react_agent(
    "anthropic:claude-3-5-sonnet-latest",
    tools=[create_manage_memory_tool(namespace=("memories", "{user_id}"))],
    store=store,
)

config = {"configurable": {"user_id": "user123"}}

# 处理用户消息
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我偏好深色模式"}]},
    config=config,
)

后台执行配置

命名空间配置

后台处理支持动态命名空间，允许为不同用户维护独立的记忆空间：

namespace = ("memories", "{langgraph_user_id}")

配置参数说明：

存储配置

InMemoryStore 适用于开发和测试环境：

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)

独立使用示例

对于不依赖 LangGraph 的应用，可以使用独立的内存管理器：

from langmem import create_memory_manager
from pydantic import BaseModel
from typing import Optional

class PreferenceMemory(BaseModel):
    preference: str
    context: Optional[str] = None

manager = create_memory_manager(
    "anthropic:claude-3-5-sonnet-latest",
    schemas=[PreferenceMemory],
)

conversation = [
    {"role": "user", "content": "I prefer dark mode in all my apps"},
    {"role": "assistant", "content": "I'll remember that preference"},
]

# 异步执行记忆提取
updated_memories = await manager.ainvoke(
    {"messages": conversation}
)

架构设计

组件关系

graph TD
    A[用户消息] --> B[主线程代理]
    B --> C[用户响应]
    B --> D[后台任务调度]
    D --> E[ReflectionExecutor]
    E --> F[MemoryManager]
    F --> G[记忆提取]
    G --> H[BaseStore]
    H --> I[持久化存储]

处理阶段

后台处理包含以下阶段：

最佳实践

性能优化

1. 设置适当的延迟：使用 after_seconds=0 实现即时后台处理
2. 分离主模型和查询模型：使用轻量级模型处理记忆查询
3. 限制检索数量：设置合理的 query_limit

配置示例

manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",  # 主模型
    query_model="anthropic:claude-3-5-haiku-latest",  # 查询模型
    query_limit=10,  # 检索限制
    namespace=("memories", "{langgraph_user_id}"),
)

总结

LangMem 的后台处理功能通过 ReflectionExecutor 提供了高效的异步记忆管理能力。开发者可以利用这一功能在不影响主应用响应速度的情况下，自动提取和管理用户记忆。

概述

LangMem 是由 LangChain AI 开发的大语言模型（LLM）记忆管理库，专注于为 AI 应用提供持久化、可检索的记忆存储能力。该库采用模块化架构设计，将记忆管理、提示工程、知识存储等功能解耦，同时通过 LangGraph 实现复杂的状态管理工作流。LangMem 的核心目标是让 AI 应用能够跨会话持久化记忆，并支持多种向量存储后端以实现高效的语义检索。

LangMem 架构遵循以下设计原则：低耦合度、高可扩展性、多后端兼容性以及类型安全。整个系统由多个功能模块组成，包括记忆模块（memory）、知识模块（knowledge）、提示模块（prompts）以及评估模块（evaluation），每个模块都暴露统一的公共接口供外部调用。

整体架构图

graph TB
    subgraph "表现层"
        API[公共 API 入口]
    end
    
    subgraph "核心模块"
        MEM[记忆模块<br/>memory]
        KNOW[知识模块<br/>knowledge]
        PROM[提示模块<br/>prompts]
        EVAL[评估模块<br/>evaluation]
    end
    
    subgraph "基础设施层"
        LC[LangChain 集成]
        LG[LangGraph 状态机]
        VS[向量存储后端]
    end
    
    API --> MEM
    API --> KNOW
    API --> PROM
    API --> EVAL
    MEM --> LG
    MEM --> VS
    KNOW --> LC
    KNOW --> VS
    PROM --> LG
    EVAL --> LG

核心模块架构

记忆模块（memory）

记忆模块是 LangMem 的核心组件，负责管理和持久化对话历史与用户偏好信息。该模块实现了多种记忆类，支持不同的存储策略和使用场景。

记忆模块通过 LangGraph 实现状态的序列化和反序列化，支持将对话状态持久化到多种后端存储系统。模块内部维护了一个状态图定义，允许开发者定义复杂的状态转换逻辑。

资料来源：src/langmem/memory/__init__.py:1-50

知识模块（knowledge）

知识模块提供了结构化知识存储和检索能力，支持将文档、实体和关系持久化到向量数据库中。该模块利用 LangChain 的文档加载器和文本分割工具，实现大规模知识库的管理。

graph LR
    DOC[文档输入] --> LOAD[文档加载]
    LOAD --> SPLIT[文本分割]
    SPLIT --> EMBED[向量嵌入]
    EMBED --> STORE[向量存储]
    STORE --> RETRIEVE[相似度检索]
    RETRIEVE --> CONTEXT[上下文输出]

知识模块的核心组件包括：

• 文档处理器：负责加载和预处理各类文档格式
• 嵌入引擎：将文本转换为高维向量表示
• 向量存储适配器：支持多种向量数据库后端
• 检索器：实现语义相似度匹配和结果排序

资料来源：src/langmem/knowledge/__init__.py:1-45

提示模块（prompts）

提示模块封装了常用的提示模板和提示优化逻辑，帮助开发者构建高效的提示词。该模块提供了模板化提示、动态参数注入和提示链组合等功能。

提示模块的主要类包括：

• PromptTemplate：基础提示模板类
• MessagePromptTemplate：消息格式提示模板
• ChatPromptTemplate：聊天场景专用提示模板

模块支持 Jinja2 风格的模板语法，允许开发者在提示中嵌入变量、控制流和条件逻辑。

资料来源：src/langmem/prompts/__init__.py:1-38

评估模块（evaluation）

评估模块提供了记忆系统效果评估的工具集，包括记忆召回率、准确率和相关性等指标的测量。该模块通常与测试框架集成，用于验证记忆系统的性能。

资料来源：src/langmem/evaluation/__init__.py:1-30

LangGraph 集成架构

LangMem 通过 langgraph.json 配置文件定义状态机工作流，实现复杂的状态管理和多轮对话逻辑。

graph TD
    START[开始] --> STATE{状态检查}
    STATE -->|新对话| INIT[初始化状态]
    STATE -->|继续对话| LOAD[加载记忆]
    INIT --> STORE[存储状态]
    LOAD --> MERGE[合并历史]
    MERGE --> PROCESS[处理输入]
    STORE --> PROCESS
    PROCESS --> UPDATE[更新记忆]
    UPDATE --> RESPONSE[生成响应]
    RESPONSE --> CHECK{对话结束?}
    CHECK -->|否| STATE
    CHECK -->|是| END[结束]

状态定义

LangGraph 配置文件定义了应用的状态结构，包括以下关键字段：

节点与边

工作流由多个节点组成：

• retrieve_node：从向量存储检索相关记忆
• update_node：更新记忆状态
• generate_node：调用 LLM 生成响应
• evaluate_node：评估当前状态质量

资料来源：langgraph.json:1-100

公共 API 入口

src/langmem/__init__.py 文件作为库的公共入口，导出所有公共 API 和核心类。该模块遵循 LangChain 的导出规范，提供清晰的命名空间组织和类型提示。

# 主要导出内容
from langmem.memory import (
    SummaryMemory,
    ConversationMemory,
    EntityMemory,
)
from langmem.knowledge import (
    KnowledgeBase,
    DocumentProcessor,
)
from langmem.prompts import (
    PromptTemplate,
    ChatPromptTemplate,
)

所有导出的类和函数都遵循以下规范：

• 完整的类型注解
• 详细的文档字符串
• 统一的错误处理机制

资料来源：src/langmem/__init__.py:1-60

数据流架构

写入流程

sequenceDiagram
    participant U as 用户
    participant M as 记忆模块
    participant LG as LangGraph
    participant VS as 向量存储
    participant KB as 知识库
    
    U->>M: 输入消息
    M->>LG: 创建状态
    LG->>LG: 处理状态转换
    LG->>KB: 存储知识
    LG->>VS: 存储向量
    LG->>M: 更新记忆
    M->>U: 确认写入

读取流程

sequenceDiagram
    participant U as 用户
    participant M as 记忆模块
    participant LG as LangGraph
    participant VS as 向量存储
    participant KB as 知识库
    
    U->>M: 查询请求
    M->>LG: 创建检索状态
    LG->>VS: 相似度搜索
    LG->>KB: 知识检索
    VS->>LG: 返回结果
    KB->>LG: 返回上下文
    LG->>M: 合并结果
    M->>U: 返回上下文

后端存储架构

LangMem 支持多种向量存储后端，提供了统一的上层抽象接口。

存储层采用适配器模式，每种后端都实现了统一的接口：

• add_documents()：添加文档
• similarity_search()：相似度搜索
• delete()：删除文档
• update()：更新文档

扩展性设计

插件接口

LangMem 通过以下方式支持扩展：

1. 自定义记忆类：继承基础 BaseMemory 类并实现必需方法
2. 自定义检索器：实现 BaseRetriever 接口
3. 自定义后端：实现存储适配器接口
4. 自定义评估器：扩展评估指标集

配置机制

系统通过以下方式支持灵活配置：

• 环境变量配置
• YAML/JSON 配置文件
• 运行时参数注入
• 依赖注入容器

技术栈总结

LangMem 的架构设计充分考虑了模块化、可扩展性和易用性，通过清晰的层次划分和标准化的接口定义，为开发者提供了构建智能记忆系统的基础设施。

概述与核心定位

记忆工具是 LangMem 知识管理系统的两大核心组件之一（另一个是记忆管理器）。与自动化的 MemoryStoreManager 不同，记忆工具将记忆的管理权直接交给 AI 代理，使其能够根据对话上下文主动决定何时以及如何操作记忆存储。

主要功能特性

资料来源：src/langmem/knowledge/tools.py:1-50

架构设计

工具类型

LangMem 提供两种主要的记忆工具：

graph TD
    A[记忆工具模块] --> B[create_manage_memory_tool]
    A --> C[create_search_memory_tool]
    B --> D[创建/更新/删除]
    C --> E[语义搜索/精确匹配]

• create_manage_memory_tool：管理工具，用于创建、更新、删除记忆
• create_search_memory_tool：搜索工具，用于检索已存储的记忆

资料来源：src/langmem/knowledge/tools.py:180-250

与 BaseStore 的集成

记忆工具通过 LangGraph 的 BaseStore 接口与底层存储层交互。工具内部会从运行时配置中获取 store 实例：

sequenceDiagram
    participant Agent as AI代理
    participant Tool as 记忆工具
    participant Store as BaseStore
    participant Config as 运行时配置

    Agent->>Tool: 调用工具函数
    Tool->>Config: 获取 store 实例
    Config-->>Tool: BaseStore 实例
    Tool->>Store: 执行存储操作
    Store-->>Tool: 操作结果
    Tool-->>Agent: 返回结果

资料来源：src/langmem/knowledge/tools.py:230-280

create_manage_memory_tool

create_manage_memory_tool 函数创建一个允许 AI 代理管理记忆的 LangChain 工具。

函数签名

def create_manage_memory_tool(
    namespace: tuple[str, ...] | str,
    *,
    schema: type[BaseModel] | None = None,
    actions_permitted: list[Literal["create", "update", "delete"]] | None = None,
    instructions: str = "",
    store: BaseStore | None = None,
    name: str = "manage_memory",
)

参数说明

资料来源：src/langmem/knowledge/tools.py:120-175

命名空间配置

命名空间支持运行时配置替换，通过 {config_key} 占位符实现：

memory_tool = create_manage_memory_tool(
    namespace=("memories", "{langgraph_user_id}", "user_profile"),
)

运行时配置示例：

config = {"configurable": {"langgraph_user_id": "user-123"}}
# 实际命名空间: ("memories", "user-123", "user_profile")

资料来源：src/langmem/knowledge/tools.py:140-160

自定义 Schema

可以通过 Pydantic 模型定义记忆的数据结构：

from pydantic import BaseModel

class UserProfile(BaseModel):
    name: str
    age: int | None = None
    recent_memories: list[str] = []
    preferences: dict | None = None

memory_tool = create_manage_memory_tool(
    namespace=("memories", "{langgraph_user_id}"),
    schema=UserProfile,
    actions_permitted=["create", "update"],
)

资料来源：src/langmem/knowledge/tools.py:220-245

操作限制

通过 actions_permitted 参数限制 AI 可执行的操作：

# 仅允许创建和更新，不允许删除
memory_tool = create_manage_memory_tool(
    namespace=("memories", "{langgraph_user_id}"),
    actions_permitted=["create", "update"],
)

工具描述会根据允许的操作自动生成：

• 单操作："{操作名} a memory"
• 双操作："{操作1} or {操作2} a memory"
• 多操作："{操作1}, {操作2}, or {操作3} a memory"

资料来源：src/langmem/knowledge/tools.py:265-285

create_search_memory_tool

create_search_memory_tool 函数创建一个用于搜索记忆的 LangChain 工具。

函数签名

def create_search_memory_tool(
    namespace: tuple[str, ...] | str,
    *,
    instructions: str = _MEMORY_SEARCH_INSTRUCTIONS,
    store: BaseStore | None = None,
    response_format: Literal["content", "content_and_artifact"] = "content",
    name: str = "search_memory",
)

参数说明

资料来源：src/langmem/knowledge/tools.py:295-340

返回格式

• content：仅返回记忆内容（字符串）
• content_and_artifact：返回元组 (serialized_memories, raw_memories)

search_tool = create_search_memory_tool(
    namespace=("memories", "{langgraph_user_id}"),
    response_format="content_and_artifact",
)

资料来源：src/langmem/knowledge/tools.py:320-325

使用示例

在 create_react_agent 中使用

将记忆工具集成到 LangGraph 的预构建 ReAct 代理中：

from langgraph.prebuilt import create_react_agent
from langgraph.config import get_config, get_store
from langmem import create_manage_memory_tool

def prompt(state):
    config = get_config()
    memories = get_store().search(
        ("memories", config["configurable"]["langgraph_user_id"]),
    )
    system_prompt = f"""You are a helpful assistant.
<memories>
{memories}
</memories>
"""
    return [{"role": "system", "content": system_prompt}, *state["messages"]]

agent = create_react_agent(
    "anthropic:claude-3-5-sonnet-latest",
    tools=[create_manage_memory_tool(
        namespace=("memories", "{langgraph_user_id}")
    )],
    store=store,
)

agent.invoke(
    {"messages": [{"role": "user", "content": "我喜欢用 Python 写后端"}]},
    config=config,
)

资料来源：src/langmem/knowledge/tools.py:45-75

完整的工作流示例

sequenceDiagram
    participant User as 用户
    participant Agent as ReAct代理
    participant Tool as 记忆工具
    participant Store as InMemoryStore

    User->>Agent: "我偏好深色模式"
    Agent->>Tool: manage_memory(content, action="create")
    Tool->>Store: put(namespace, key, value)
    Store-->>Tool: MemoryItem
    Tool-->>Agent: "已记住您的偏好"
    Agent-->>User: 响应

    User->>Agent: "我改变了主意"
    Agent->>Tool: manage_memory(id, action="update")
    Tool->>Store: put(namespace, id, new_value)
    Store-->>Tool: Updated
    Agent-->>User: "已更新您的偏好"

资料来源：src/langmem/knowledge/tools.py:75-110

在独立应用中使用

记忆工具也可以在 LangGraph 外部的独立应用中使用：

from langmem import create_memory_manager
from langgraph.store.memory import InMemoryStore
from pydantic import BaseModel

class PreferenceMemory(BaseModel):
    category: str
    preference: str
    context: str

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)

# 独立使用管理器
manager = create_memory_manager(
    "anthropic:claude-3-5-sonnet-latest",
    schemas=[PreferenceMemory],
)

conversation = [
    {"role": "user", "content": "I prefer dark mode in all my apps"},
    {"role": "assistant", "content": "I'll remember that preference"},
]

memories = manager.invoke({"messages": conversation})
print(memories)

资料来源：examples/standalone_examples/README.md:1-45

工具与运行时的绑定

store 参数

store 参数用于指定底层存储。如果为 None，工具会在运行时从 LangGraph 的配置上下文中获取 store：

# 方式一：创建时指定 store
store = InMemoryStore(index={"dims": 1536, "embed": "openai:text-embedding-3-small"})
manage_tool = create_manage_memory_tool(namespace=("mem", "{uid}"), store=store)

# 方式二：运行时从 @entrypoint 获取
# 在 @entrypoint(store=store) 装饰的函数中，store 会自动可用

资料来源：src/langmem/knowledge/tools.py:225-240

命名空间的动态解析

命名空间中的占位符在运行时通过 configurable 配置解析：

config = {
    "configurable": {
        "langgraph_user_id": "user123",
        "team_id": "engineering"
    }
}

# namespace = ("memories", "{langgraph_user_id}", "{team_id}")
# 解析为: ("memories", "user123", "engineering")

资料来源：src/langmem/knowledge/tools.py:145-155

与记忆管理器的对比

记忆工具适合需要 AI 自主管理记忆的场景，而记忆管理器（如 create_memory_manager、create_memory_store_manager）适合需要自动从对话中提取和组织记忆的场景。

资料来源：src/langmem/knowledge/extraction.py:50-150

最佳实践

1. 限制操作权限

始终通过 actions_permitted 限制 AI 可执行的操作，防止意外删除重要记忆：

manage_tool = create_manage_memory_tool(
    namespace=("memories", "{user_id}"),
    actions_permitted=["create", "update"],  # 禁止删除
)

2. 使用 Schema 验证

通过 Pydantic schema 确保记忆数据结构的一致性：

class Memory(BaseModel):
    content: str
    tags: list[str] = []
    priority: int = 0

manage_tool = create_manage_memory_tool(
    namespace=("memories", "{user_id}"),
    schema=Memory,
)

3. 分离搜索和管理

在复杂应用中，建议分别创建搜索和管理工具：

from langmem import create_manage_memory_tool, create_search_memory_tool

manage = create_manage_memory_tool(
    namespace=("memories", "{user_id}"),
    actions_permitted=["create", "update", "delete"],
)

search = create_search_memory_tool(
    namespace=("memories", "{user_id}"),
)

agent = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[manage, search],
    store=store,
)

4. 提供清晰的说明

通过 instructions 参数提供工具使用指导：

manage_tool = create_manage_memory_tool(
    namespace=("preferences", "{user_id}"),
    instructions="仅在用户明确表达偏好时创建记忆，避免过度记录。",
)

注意事项

1. store 必须可用：工具依赖 LangGraph 的 BaseStore，在 @entrypoint 或 create_react_agent 中使用时会自动注入。

1. 命名空间唯一性：每个用户的记忆应使用独立的命名空间路径。

1. 异步支持：工具同时支持同步（invoke）和异步（ainvoke）调用。

1. Schema 兼容性：自定义 schema 必须与已存储的记忆数据结构兼容，否则会导致验证失败。

1. ID 必填规则：更新和删除操作需要提供记忆的 UUID，AI 在调用工具时必须包含此参数。

资料来源：src/langmem/knowledge/tools.py:180-220

导出清单

__all__ = [
    "create_manage_memory_tool",
    "create_search_memory_tool",
]

资料来源：src/langmem/knowledge/tools.py:350-355

核心概念

什么是记忆提取

记忆提取是一种自动化的信息处理流程，它从非结构化的对话数据中识别并提取结构化的记忆信息。这些记忆信息经过处理后可以用于：

• 个性化响应生成
• 上下文连续性维护
• 用户偏好跟踪
• 对话状态管理

资料来源：src/langmem/knowledge/extraction.py:1-50

记忆类型

langmem 支持多种类型的记忆提取模式：

资料来源：src/langmem/prompts/types.py:1-80

架构设计

整体架构

graph TD
    A[对话输入] --> B[记忆管理器<br/>MemoryManager]
    B --> C{提取模式}
    C -->|LLM分析| D[模式匹配]
    C -->|向量搜索| E[语义检索]
    D --> F[记忆存储]
    E --> F
    F --> G[结构化记忆]
    G --> H[长期存储<br/>MemoryStore]
    G --> I[短期记忆<br/>Summarization]

核心组件

资料来源：src/langmem/knowledge/extraction.py:100-200

MemoryManager 详解

初始化配置

create_memory_manager 函数用于创建记忆管理器实例：

from langmem import create_memory_manager

manager = create_memory_manager(
    "anthropic:claude-3-5-sonnet-latest",
    schemas=[PreferenceMemory],  # Pydantic schema 定义
    instructions="从对话中提取用户偏好信息",
    enable_inserts=True,
    enable_updates=True,
    enable_deletes=True,
)

配置参数表

资料来源：src/langmem/knowledge/extraction.py:200-300

调用方式

#### 异步调用

conversation = [
    {"role": "user", "content": "我偏好深色模式"},
    {"role": "assistant", "content": "好的，已记住您的偏好"},
]

memories = await manager.ainvoke(
    {"messages": conversation, "max_steps": 3}
)

#### 同步调用

memories = manager.invoke({"messages": conversation})

资料来源：src/langmem/knowledge/extraction.py:250-280

MemoryStoreManager 详解

概述

MemoryStoreManager 是支持向量存储的增强版记忆管理器，集成了语义搜索能力：

from langmem import create_memory_store_manager
from langgraph.store.memory import InMemoryStore

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)

manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",
    query_model="anthropic:claude-3-5-haiku-latest",
    query_limit=10,
    namespace=("memories", "{langgraph_user_id}"),
    store=store,
)

与 MemoryManager 的区别

资料来源：src/langmem/knowledge/extraction.py:300-400

工作流程

sequenceDiagram
    participant Client
    participant Manager as MemoryStoreManager
    participant QueryLLM
    participant Store as VectorStore
    participant MainLLM

    Client->>Manager: 消息输入
    Manager->>QueryLLM: 生成搜索查询
    QueryLLM-->>Manager: 优化后的查询
    Manager->>Store: 语义检索
    Store-->>Manager: 相关记忆
    Manager->>MainLLM: 分析并提取
    MainLLM-->>Manager: 更新后的记忆
    Manager->>Store: 持久化存储
    Manager-->>Client: 提取结果

搜索功能

# 自然语言搜索
results = await manager.asearch(
    query="用户在医疗领域的偏好",
    filter={"type": "research_paper"},
    limit=5,
)

# 同步搜索
results = manager.search(query="app preferences")

资料来源：src/langmem/knowledge/extraction.py:80-120

短期记忆压缩

SummarizationNode

SummarizationNode 用于在对话过程中压缩和摘要消息，保持上下文的同时减少 token 消耗：

from langmem.short_term import SummarizationNode, RunningSummary

summarization_node = SummarizationNode(
    model=summarization_model,
    max_tokens=256,
    max_tokens_before_summary=256,
    max_summary_tokens=128,
)

配置参数

资料来源：src/langmem/short_term/summarization.py:1-100

状态更新格式

{
    "output_messages_key": [
        "<更新的消息列表，包含摘要消息>"
    ],
    "context": {
        "running_summary": "<RunningSummary 对象>"
    }
}

语义记忆图

SemanticMemoryGraph

SemanticMemoryGraph 提供基于语义图的记忆组织方式：

graph LR
    A[用户输入] --> B[语义解析]
    B --> C[实体识别]
    C --> D[关系抽取]
    D --> E[图结构更新]
    E --> F[记忆节点]
    F --> G[关系边]
    G --> H[语义记忆图]

核心功能

资料来源：src/langmem/graphs/semantic.py:1-50

工具函数

记忆管理工具

create_manage_memory_tool 创建一个 LangGraph 可用的记忆管理工具：

from langmem import create_manage_memory_tool

manage_tool = create_manage_memory_tool(
    namespace=("memories", "{langgraph_user_id}"),
    schema=UserMemory,
    actions_permitted=["create", "update", "delete"],
)

参数说明

资料来源：src/langmem/knowledge/tools.py:1-100

记忆搜索工具

create_search_memory_tool 创建一个语义搜索工具：

from langmem import create_search_memory_tool

search_tool = create_search_memory_tool(
    namespace=("project_memories", "{langgraph_user_id}"),
)

# 在 LangGraph 中使用
memories, _ = await search_tool.ainvoke(
    {"query": "Python preferences", "limit": 5}
)

使用示例

基础示例

from langmem import create_memory_manager
from pydantic import BaseModel

class PreferenceMemory(BaseModel):
    category: str
    preference: str
    context: str

manager = create_memory_manager(
    "anthropic:claude-3-5-sonnet-latest",
    schemas=[PreferenceMemory],
)

conversation = [
    {"role": "user", "content": "我偏好所有应用都使用深色模式"},
    {"role": "assistant", "content": "好的，我会记住这个偏好"},
]

memories = await manager.ainvoke({"messages": conversation})
print(memories)

带配置的完整示例

from langmem import create_memory_store_manager, ReflectionExecutor
from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore
from langgraph.func import entrypoint

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)

manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",
    namespace=("memories", "{langgraph_user_id}"),
)

reflection = ReflectionExecutor(manager, store=store)
agent = create_react_agent(
    "anthropic:claude-3-5-sonnet-latest",
    tools=[create_manage_memory_tool(namespace=("memories", "{langgraph_user_id}"))],
    store=store,
)

在 LangGraph 中使用

from langgraph.prebuilt import create_react_agent
from langgraph.config import get_config, get_store

def prompt(state):
    config = get_config()
    memories = get_store().search(
        ("memories", config["configurable"]["langgraph_user_id"]),
    )
    system_prompt = f"""你是一个有帮助的助手。
    <记忆>
    {memories}
    </记忆>
    """
    return [{"role": "system", "content": system_prompt}, *state["messages"]]

agent = create_react_agent(
    "anthropic:claude-3-5-sonnet-latest",
    tools=[create_manage_memory_tool(namespace=("memories", "{langgraph_user_id}"))],
)

agent.invoke(
    {"messages": [{"role": "user", "content": "我们决定用 golang 代替 python 做后端"}]},
    config=config,
)

资料来源：src/langmem/knowledge/tools.py:100-200

数据结构

AnnotatedTrajectory

用于表示带反馈的对话轨迹：

trajectory = AnnotatedTrajectory(
    messages=[
        {"role": "user", "content": "有什么好的披萨推荐？"},
        {"role": "assistant", "content": "推荐尝试 LangPizza™️"},
        {"role": "user", "content": "别给我打广告"},
        {"role": "assistant", "content": "但你会喜欢的！"},
    ],
    feedback={
        "developer_feedback": "太激进了",
        "score": 0,
    },
)

OptimizerInput

用于单提示优化场景的输入结构：

input_data = OptimizerInput(
    trajectories=[
        AnnotatedTrajectory(
            messages=[...],
            feedback="应该使用搜索工具",
        )
    ],
    prompt=Prompt(
        name="main_assistant",
        prompt="你是一个有帮助的助手，有搜索工具可用。",
        update_instructions="仅做最小更改，专注于修复错误的地方。",
    ),
)

资料来源：src/langmem/prompts/types.py:50-120

配置最佳实践

命名空间配置

性能优化建议

1. 批量处理：将多个相关对话合并处理
2. 增量更新：使用 enable_updates=True 避免重复存储
3. 搜索限制：设置合理的 query_limit 减少检索开销
4. 摘要压缩：对于长对话使用 SummarizationNode 减少 token

安全考虑

• 命名空间隔离：不同用户使用不同命名空间
• 操作权限：仅启用必要的 CRUD 操作
• Schema 验证：使用 Pydantic 模型确保数据一致性

相关模块

总结

记忆提取是 langmem 提供的核心能力之一，它通过结构化的方式从对话中提取有价值的信息，并通过 MemoryManager 和 MemoryStoreManager 两个主要接口提供灵活的存储和检索能力。结合短期记忆压缩和语义搜索功能，langmem 能够构建完整的记忆系统，支持各种需要上下文持久化的 AI 应用场景。

概述

提示优化（Prompt Optimization）是 langmem 库中用于自动改进大型语言模型提示词的核心模块。该模块通过分析对话轨迹和反馈信息，自动调整和优化系统提示词，使 AI 代理能够更好地完成特定任务。

langmem 提供了三种主要的提示优化策略：

核心组件

类型定义

提示优化模块使用强类型定义确保输入输出的准确性。

#### OptimizerInput

class OptimizerInput(TypedDict):
    """单提示优化的输入数据结构"""
    trajectories: typing.Sequence[AnnotatedTrajectory] | str
    prompt: str | Prompt

#### MultiPromptOptimizerInput

class MultiPromptOptimizerInput(TypedDict):
    """多提示优化的输入数据结构"""
    trajectories: typing.Sequence[AnnotatedTrajectory] | str
    prompts: list[Prompt]

#### Prompt 结构

class Prompt(TypedDict, total=False):
    """结构化提示管理"""
    name: Required[str]                    # 提示名称
    prompt: Required[str]                   # 提示内容
    update_instructions: str | None        # 更新指令
    when_to_update: str | None             # 更新时机

#### AnnotatedTrajectory 结构

class AnnotatedTrajectory(typing.NamedTuple):
    """带标注的对话轨迹"""
    messages: typing.Sequence[AnyMessage]  # 消息列表
    feedback: dict[str, str | int | bool] | str | None = None  # 反馈信息

资料来源：src/langmem/prompts/types.py:1-80

API 接口

创建单提示优化器

def create_prompt_optimizer(
    model: str | BaseChatModel,
    kind: typing.Literal["gradient", "metaprompt", "prompt_memory"] = "gradient",
    *,
    config: typing.Optional[dict] = None,
) -> Runnable[OptimizerInput, str]

创建多提示优化器

def create_multi_prompt_optimizer(
    model: str | BaseChatModel,
    kind: typing.Literal["gradient", "metaprompt", "prompt_memory"] = "gradient",
    *,
    config: typing.Optional[dict] = None,
) -> Runnable[MultiPromptOptimizerInput, list[Prompt]]

资料来源：src/langmem/prompts/optimization.py

优化策略详解

1. 梯度优化（Gradient Optimization）

梯度优化是最基本的优化策略，通过分析对话轨迹中的假设和推荐进行提示词调整。

graph TD
    A[输入轨迹和反馈] --> B[生成假设]
    B --> C[提出建议]
    C --> D[应用最小侵入性更改]
    D --> E[输出优化后的提示]

#### 工作流程

1. 假设生成：分析当前提示在处理对话轨迹时的潜在问题
2. 建议生成：基于假设提出具体的调整建议
3. 应用更改：仅实施必要的最小更改

#### 配置选项

资料来源：src/langmem/prompts/gradient.py

#### 使用示例

from langmem import create_prompt_optimizer

optimizer = create_prompt_optimizer(
    "anthropic:claude-3-5-sonnet-latest",
    kind="gradient",
    config={"max_reflection_steps": 3}
)

conversation = [
    {"role": "user", "content": "Tell me about the solar system"},
    {"role": "assistant", "content": "The solar system consists of..."},
]
feedback = {"clarity": "needs more structure"}

trajectories = [(conversation, feedback)]
better_prompt = await optimizer.ainvoke(
    {"trajectories": trajectories, "prompt": "You are an astronomy expert"}
)

2. 元提示优化（Metaprompt Optimization）

元提示优化通过多步元学习过程进行深度反思，适合处理复杂的优化任务。

graph TD
    A[输入当前提示] --> B[元级反思]
    B --> C{反思步数检查}
    C -->|未达到最小步数| B
    C -->|已达到最小步数| D[生成优化建议]
    D --> E[应用最终优化]
    E --> F[输出优化后的提示]

#### 配置选项

#### 使用示例

from langmem import create_prompt_optimizer

optimizer = create_prompt_optimizer(
    "anthropic:claude-3-5-sonnet-latest",
    kind="metaprompt",
    config={"max_reflection_steps": 3, "min_reflection_steps": 1}
)

conversation = [
    {"role": "user", "content": "Explain quantum computing"},
    {"role": "assistant", "content": "Quantum computing uses..."},
]
feedback = "Should include mathematical notation"

trajectories = [(conversation, {"feedback": feedback})]
better_prompt = await optimizer.ainvoke(
    {"trajectories": trajectories, "prompt": "You are a physics tutor"}
)

资料来源：src/langmem/prompts/metaprompt.py

3. 提示记忆优化（Prompt Memory Optimization）

提示记忆优化从历史交互中学习成功模式，识别改进领域，并将学习到的模式应用到新提示中。

graph TD
    A[历史轨迹数据] --> B[提取成功模式]
    A --> C[识别改进领域]
    B --> D[应用模式到新提示]
    C --> D
    D --> E[输出优化后的提示]

#### 核心特点

• 模式提取：从过去的成功交互中提取可复用的模式
• 反馈分析：识别反馈中指出的改进机会
• 知识迁移：将学习到的知识应用到新的提示优化中

#### 使用示例

from langmem import create_multi_prompt_optimizer

optimizer = create_multi_prompt_optimizer(
    "anthropic:claude-3-5-sonnet-latest",
    kind="prompt_memory"
)

conversation = [
    {"role": "user", "content": "How do I write a bash script?"},
    {"role": "assistant", "content": "Let me explain bash scripting..."},
]
feedback = "Response should include a code example"

trajectories = [(conversation, {"feedback": feedback})]
prompts = [
    {"name": "explain", "prompt": "Explain the concept"},
    {"name": "example", "prompt": "Provide a practical example"},
]

better_prompts = await optimizer.ainvoke(
    {"trajectories": trajectories, "prompts": prompts}
)

提示反射模板

INSTRUCTION_REFLECTION_PROMPT

用于单提示优化的反射模板：

INSTRUCTION_REFLECTION_PROMPT = """You are helping an AI agent improve. You can do this by changing their system prompt.

These is their current prompt:
<current_prompt>
{current_prompt}
</current_prompt>

Here was the agent's trajectory:
<trajectory>
{trajectory}
</trajectory>

Here is the user's feedback:

<feedback>
{feedback}
</feedback>

Here are instructions for updating the agent's prompt:

<instructions>
{instructions}
</instructions>


Based on this, return an updated prompt"""

INSTRUCTION_REFLECTION_MULTIPLE_PROMPT

用于多提示优化的反射模板，支持批量处理多个轨迹：

INSTRUCTION_REFLECTION_MULTIPLE_PROMPT = """You are helping an AI agent improve. You can do this by changing their system prompt.

These is their current prompt:
<current_prompt>
{current_prompt}
</current_prompt>

Here are examples of various agent trajectories and associated feedback:
<data>
{data}
</data>

Here are instructions for updating the agent's prompt:

<instructions>
{instructions}
</instructions>


Based on this, return an updated prompt"""

GeneralResponse 结构

优化器的返回响应结构：

class GeneralResponse(TypedDict):
    logic: str                    # 推理逻辑
    update_prompt: bool           # 是否需要更新
    new_prompt: str               # 新提示内容

资料来源：src/langmem/prompts/prompt.py

梯度元提示模板

梯度优化使用的默认元提示：

DEFAULT_GRADIENT_METAPROMPT = """You are optimizing a prompt to handle its target task more effectively.

<current_prompt>
{current_prompt}
</current_prompt>

We hypothesize the current prompt underperforms for these reasons:

<hypotheses>
{hypotheses}
</hypotheses>

Based on these hypotheses, we recommend the following adjustments:

<recommendations>
{recommendations}
</recommendations>

Respond with the updated prompt. Remember to ONLY make changes that are clearly necessary. Aim to be minimally invasive:"""

资料来源：src/langmem/prompts/gradient.py

GradientPromptOptimizer 类

基于类的梯度提示优化器实现，同时支持同步和异步调用：

class GradientPromptOptimizer(Runnable[GradientOptimizerInput, str]):
    """
    Class-based Gradient Prompt Optimizer with both sync and async entry points.
    """
    
    def __init__(
        self,
        model: Union[str, BaseChatModel],
        config: Optional[GradientOptimizerConfig] = None,
    ):
        self.model = model
        self.config = config or {}

GradientOptimizerConfig 配置结构

class GradientOptimizerConfig(TypedDict, total=False):
    """梯度优化器配置"""
    gradient_prompt: str           # 梯度优化提示模板
    metaprompt: str                # 元提示模板
    max_reflection_steps: int      # 最大反思步数
    min_reflection_steps: int      # 最小反思步数

GradientOptimizerInput 输入结构

class GradientOptimizerInput(TypedDict, total=False):
    """梯度优化器输入"""
    trajectories: prompt_types.OptimizerInput | str
    prompt: str | prompt_types.Prompt

最佳实践

1. 选择合适的优化策略

2. 配置反思步数

# 简单任务 - 快速优化
config = {"max_reflection_steps": 1, "min_reflection_steps": 1}

# 复杂任务 - 深度优化
config = {"max_reflection_steps": 5, "min_reflection_steps": 2}

3. 提供有效反馈

反馈应该包含：

• 具体的问题描述
• 期望的行为
• 评分或等级（可选）

4. 多提示协同优化

当多个提示需要保持一致性时，使用 create_multi_prompt_optimizer：

prompts = [
    Prompt(
        name="vision_extract",
        prompt="Extract visual details from the image",
        update_instructions="Focus on using vision model capabilities",
    ),
    Prompt(
        name="vision_classify",
        prompt="Classify specific attributes in the image",
        when_to_update="After vision_extract is updated",
    ),
]

导出接口

提示优化模块的公共 API：

__all__ = [
    "create_prompt_optimizer",
    "create_multi_prompt_optimizer",
    "Prompt",
    "AnnotatedTrajectory",
    "OptimizerInput",
    "MultiPromptOptimizerInput",
]

资料来源：src/langmem/prompts/optimization.py

核心概念

什么是语义记忆

语义记忆是一种结构化的知识表示形式，它从原始对话内容中提取关键信息，并以预定义的数据模式（Schema）进行存储。与简单的文本存储不同，语义记忆具有以下特征：

语义记忆 vs 其他记忆类型

资料来源：src/langmem/knowledge/extraction.py:1-50

架构概述

系统组件

LangMem 的语义记忆管理由以下核心组件构成：

graph TD
    subgraph "语义记忆管理系统"
        A[对话消息] --> B[MemoryStoreManager]
        B --> C[提取器 Extractor]
        C --> D[Schema 验证]
        D --> E[记忆存储 Store]
        F[搜索查询] --> G[QueryLLM]
        G --> H[向量相似度匹配]
        H --> E
    end
    
    subgraph "LangGraph 集成"
        E --> I[InMemoryStore]
        I --> J[Agent 上下文]
        J --> A
    end

工作流程

sequenceDiagram
    participant User as 用户
    participant Agent as Agent
    participant Manager as MemoryStoreManager
    participant Extractor as 提取器
    participant Store as 记忆存储
    participant QueryLLM as 查询模型

    User->>Agent: 发送消息
    Agent-->>User: 生成响应
    Agent->>Manager: 提交对话历史
    Manager->>Extractor: 提取语义记忆
    Extractor->>Store: 存储/更新记忆
    Note over Store: 记忆持久化
    
    User->>Agent: 新请求
    Agent->>Manager: 搜索相关记忆
    Manager->>QueryLLM: 生成优化查询
    QueryLLM-->>Manager: 相似记忆
    Manager-->>Agent: 上下文记忆
    Agent-->>User: 个性化响应

资料来源：src/langmem/knowledge/extraction.py:80-120

核心 API

创建语义记忆管理器

from langmem import create_memory_store_manager
from langgraph.store.memory import InMemoryStore

manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",  # 主模型
    query_model="anthropic:claude-3-5-haiku-latest",  # 查询模型
    query_limit=10,
    namespace=("memories", "{langgraph_user_id}"),
    store=InMemoryStore(),
)

参数说明

资料来源：src/langmem/knowledge/extraction.py:150-200

Schema 定义

语义记忆使用 Pydantic 模型定义数据结构。以下是典型的 Schema 定义示例：

from pydantic import BaseModel
from typing import Literal

class PreferenceMemory(BaseModel):
    """用户偏好记忆"""
    category: str  # 偏好类别
    preference: str  # 具体偏好
    context: str  # 上下文信息

class UserProfile(BaseModel):
    """用户档案记忆"""
    name: str
    age: int | None = None
    recent_memories: list[str] = []
    preferences: dict | None = None

Schema 配置参数

资料来源：src/langmem/prompts/types.py:1-40

集成方式

LangGraph 集成

LangMem 与 LangGraph 的 @entrypoint 装饰器无缝集成：

from langmem import create_memory_store_manager
from langgraph.store.memory import InMemoryStore
from langgraph.func import entrypoint

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)

manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",
    namespace=("memories", "{langgraph_user_id}"),
    store=store,
)

@entrypoint(store=store)
async def my_agent(message: str):
    response = {"role": "assistant", "content": "I'll remember that"}
    await manager.ainvoke(
        {"messages": [{"role": "user", "content": message}, response]}
    )
    return response

后台执行模式

对于需要异步处理的场景，可以使用 ReflectionExecutor：

graph LR
    A[Agent] --> B[处理消息]
    B --> C[返回响应]
    B --> D[调度后台处理]
    D --> E[ReflectionExecutor]
    E --> F[记忆提取]
    F --> G[存储更新]

from langmem import create_memory_store_manager, ReflectionExecutor

reflection = ReflectionExecutor(manager, store=store)

# 在消息处理后调度后台执行
reflection.schedule_after(agent_response, after_seconds=0)

资料来源：src/langmem/knowledge/extraction.py:200-250

搜索和检索

默认搜索行为

默认情况下，记忆检索通过直接嵌入新消息进行相似度匹配：

# 直接使用消息嵌入搜索
results = manager.search(config=config)

LLM 优化查询搜索

使用专用查询模型生成更精确的搜索查询：

manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",
    query_model="anthropic:claude-3-5-haiku-latest",  # 更快的小模型
    query_limit=10,
    namespace=("memories", "{langgraph_user_id}"),
)

# 使用自然语言查询
results = manager.search(query="app preferences", config=config)

过滤和分页

# 带过滤条件的搜索
results = await manager.asearch(
    query="技术偏好",
    filter={"category": "programming"},
    limit=10,
    offset=0,
)

搜索结果格式

资料来源：src/langmem/knowledge/extraction.py:250-300

默认值机制

当没有找到相关记忆时，可以使用默认值工厂提供回退记忆：

manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",
    namespace=("memories", "{langgraph_user_id}"),
    default="Use a concise and professional tone in all responses.",
    default_factory=UserPreferencesFactory,
)

默认值配置

资料来源：src/langmem/knowledge/extraction.py:300-350

操作模式

完整模式（默认值）

允许插入、更新和删除记忆：

manager = create_memory_manager(
    "anthropic:claude-3-5-sonnet-latest",
    schemas=[PreferenceMemory],
    enable_inserts=True,
    enable_updates=True,
    enable_deletes=True,  # 完整功能
)

仅插入模式

仅创建新记忆，不修改现有记忆：

manager = create_memory_manager(
    "anthropic:claude-3-5-sonnet-latest",
    schemas=[PreferenceMemory],
    enable_inserts=True,
    enable_updates=False,
    enable_deletes=False,  # 仅插入
)

操作控制参数

多步骤提取和综合

对于复杂对话场景，可以配置多步骤处理：

conversation = [
    {"role": "user", "content": "I prefer dark mode in all my apps"},
    {"role": "assistant", "content": "I'll remember that preference"},
]

# 设置最大处理步骤数
max_steps = 3
memories = await manager.ainvoke(
    {"messages": conversation, "max_steps": max_steps}
)

处理流程

graph TD
    A[对话历史] --> B[步骤 1: 提取]
    B --> C{是否完成?}
    C -->|否| D[生成工具调用]
    D --> E[步骤 2: 执行]
    E --> B
    C -->|是| F[返回记忆结果]
    
    B --> G[保留未修改的现有记忆]
    G --> F

命名空间管理

运行时命名空间配置

命名空间支持运行时配置变量替换：

namespace=("memories", "{langgraph_user_id}")

# 配置
config = {"configurable": {"langgraph_user_id": "user123"}}
# 结果命名空间: ("memories", "user123")

命名空间结构

最佳实践

1. Schema 设计

• 为不同类型的记忆创建独立的 Schema
• 使用描述性的字段名称和类型注解
• 提供合理的默认值减少空值处理

2. 模型选择

3. 性能考虑

• 使用后台执行处理记忆管理，避免阻塞主流程
• 合理设置 query_limit 避免返回过多不相关记忆
• 对高频访问的记忆使用短期缓存

4. 错误处理

try:
    await manager.ainvoke({"messages": conversation})
except ValueError as e:
    # Schema 验证失败
    logger.error(f"记忆验证失败: {e}")
except Exception as e:
    # 存储操作失败
    logger.error(f"存储错误: {e}")

完整使用示例

from langmem import create_memory_store_manager
from langgraph.store.memory import InMemoryStore
from langgraph.func import entrypoint
from pydantic import BaseModel

class PreferenceMemory(BaseModel):
    """用户偏好记忆"""
    category: str
    preference: str
    context: str

# 初始化存储
store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)

# 创建管理器
manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",
    schemas=[PreferenceMemory],
    query_model="anthropic:claude-3-5-haiku-latest",
    query_limit=10,
    namespace=("memories", "{langgraph_user_id}"),
    store=store,
)

# 定义 Agent
@entrypoint(store=store)
async def my_agent(message: str):
    response = {"role": "assistant", "content": "I'll remember that preference"}
    await manager.ainvoke(
        {"messages": [{"role": "user", "content": message}, response]}
    )
    return response

# 执行对话
config = {"configurable": {"langgraph_user_id": "user123"}}
await my_agent.ainvoke(
    "I prefer dark mode in all my apps",
    config=config,
)

# 检索记忆
print(manager.search(query="app preferences", config=config))

资料来源：src/langmem/knowledge/extraction.py:50-100

相关模块

概述

情景记忆提取（Episodic Memory Extraction）是 LangMem 库的核心功能之一，用于从对话历史中自动识别、提取和管理结构化记忆。该功能使 AI 系统能够持续学习用户偏好、对话模式和环境信息，并在后续交互中智能调用相关记忆来提升个性化服务质量。

LangMem 的情景记忆提取系统基于以下核心设计理念：

• 结构化存储：记忆以预定义的 Pydantic Schema 形式存储，确保类型安全和数据一致性
• 增量更新：支持记忆的创建、更新和删除操作，适应用户偏好的动态变化
• 智能检索：提供基于向量相似度和 LLM 查询的混合检索能力
• 异步处理：支持后台执行记忆处理，不阻塞主对话流程

资料来源：src/langmem/knowledge/extraction.py:1-50

核心组件

MemoryManager

MemoryManager 是情景记忆提取的主控制器，负责协调整个记忆生命周期。

from langmem import create_memory_manager

manager = create_memory_manager(
    "anthropic:claude-3-5-sonnet-latest",
    schemas=[PreferenceMemory],
)

关键参数说明：

资料来源：src/langmem/knowledge/extraction.py:100-150

MemoryStoreManager

MemoryStoreManager 扩展了基础记忆管理能力，集成了 LangGraph 的持久化存储层。

from langmem import create_memory_store_manager
from langgraph.store.memory import InMemoryStore

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)
manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",
    query_model="anthropic:claude-3-5-haiku-latest",
    query_limit=10,
    namespace=("memories", "{langgraph_user_id}"),
    store=store,
)

资料来源：src/langmem/knowledge/extraction.py:200-280

工作流程

基础提取流程

sequenceDiagram
    participant Client
    participant Manager
    participant LLM
    participant Store

    Client->>Manager: messages + existing memories
    Manager->>LLM: analyze conversation
    LLM-->>Manager: extraction decisions
    Manager->>Store: apply memory changes
    Store-->>Manager: updated memories
    Manager-->>Client: memory updates

带 LLM 查询的检索流程

当需要更精确的记忆检索时，系统会先生成优化查询：

sequenceDiagram
    participant Client
    participant Manager
    participant QueryLLM
    participant Store
    participant MainLLM

    Client->>Manager: messages
    Manager->>QueryLLM: generate search query
    QueryLLM-->>Manager: optimized query
    Manager->>Store: find memories
    Store-->>Manager: memories
    Manager->>MainLLM: analyze & extract
    MainLLM-->>Manager: memory updates
    Manager->>Store: apply changes
    Manager-->>Client: result

资料来源：src/langmem/knowledge/extraction.py:50-100

记忆 Schema 定义

基本结构

LangMem 使用 Pydantic BaseModel 定义记忆结构：

from pydantic import BaseModel

class PreferenceMemory(BaseModel):
    """用户偏好记忆"""
    preference: str
    context: str
    category: str = "general"

class UserProfile(BaseModel):
    """用户档案记忆"""
    name: str
    age: int | None = None
    recent_memories: list[str] = []
    preferences: dict | None = None

资料来源：src/langmem/knowledge/tools.py:1-50

嵌套 Schema 示例

class Memory(BaseModel):
    """带类型标签的统一记忆格式"""
    kind: str
    content: dict[str, Any]

存储后的记忆项包含以下元数据：

资料来源：src/langmem/knowledge/tools.py:50-100

使用方式

独立使用

在 LangGraph 外部独立使用记忆管理功能：

from langmem import create_memory_manager
from pydantic import BaseModel

class PreferenceMemory(BaseModel):
    preference: str
    context: str

manager = create_memory_manager(
    "anthropic:claude-3-5-sonnet-latest",
    schemas=[PreferenceMemory],
)

conversation = [
    {"role": "user", "content": "I prefer dark mode in all my apps"},
    {"role": "assistant", "content": "I'll remember that preference"},
]

# 提取记忆
memories = await manager.ainvoke({"messages": conversation})
print(memories)

资料来源：examples/standalone_examples/README.md:1-50

与 LangGraph 集成

在 LangGraph 应用中后台执行记忆处理：

from langmem import create_memory_store_manager, ReflectionExecutor
from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore
from langgraph.func import entrypoint

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": "openai:text-embedding-3-small",
    }
)
manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",
    namespace=("memories", "{user_id}"),
)
reflection = ReflectionExecutor(manager, store=store)

@entrypoint(store=store)
async def my_agent(message: str):
    response = {"role": "assistant", "content": "I'll remember that"}
    await reflection.ainvoke(
        {"messages": [{"role": "user", "content": message}, response]}
    )
    return response

资料来源：src/langmem/knowledge/extraction.py:280-350

检索已存储记忆

# 查看提取的记忆
print(manager.search(config=config))

# 自然语言搜索
results = await manager.asearch(
    query="app preferences",
    filter={"type": "PreferenceMemory"},
    limit=5,
)

高级配置

仅插入模式

适用于只需要记录新信息、不修改现有记忆的场景：

manager = create_memory_manager(
    "anthropic:claude-3-5-sonnet-latest",
    schemas=[PreferenceMemory],
    enable_updates=False,
    enable_deletes=False,
)

自定义提取步数

控制 LLM 推理和综合的迭代次数：

# 设置最大步数
max_steps = 3
memories = await manager.ainvoke(
    {"messages": conversation, "max_steps": max_steps}
)

默认记忆值

当没有找到记忆时返回预设默认值：

manager = create_memory_store_manager(
    "anthropic:claude-3-5-sonnet-latest",
    namespace=("memories", "{langgraph_user_id}"),
    default="Use a concise and professional tone in all responses.",
)

资料来源：src/langmem/knowledge/extraction.py:350-400

工具函数

create_manage_memory_tool

创建可注入 LangGraph Agent 的记忆管理工具：

from langmem import create_manage_memory_tool
from langgraph.prebuilt import create_react_agent

memory_tool = create_manage_memory_tool(
    namespace=("memories", "{langgraph_user_id}"),
    schema=PreferenceMemory,
    actions_permitted=["create", "update", "delete"],
    instructions="Update the existing user profile based on shared information.",
)

agent = create_react_agent(
    "anthropic:claude-3-5-sonnet-latest",
    tools=[memory_tool],
    store=store,
)

资料来源：src/langmem/knowledge/tools.py:100-200

create_search_memory_tool

创建专门用于记忆检索的工具：

from langmem import create_search_memory_tool

search_tool = create_search_memory_tool(
    namespace=("project_memories", "{langgraph_user_id}"),
)

# 在工作流中使用
memories, _ = await search_tool.ainvoke(
    {"query": "Python preferences", "limit": 5}
)

create_memory_searcher

创建独立的记忆搜索流水线：

from langmem import create_memory_searcher

searcher = create_memory_searcher(
    "anthropic:claude-3-5-sonnet-latest",
    prompt="Search for distinct memories relevant to different aspects.",
    namespace=("memories", "{langgraph_user_id}"),
)

资料来源：src/langmem/knowledge/extraction.py:400-500

命名空间设计

命名空间结构

记忆按照层级化的命名空间组织：

namespace = ("memories", "{langgraph_user_id}")
# 运行时解析为：("memories", "user123")

常见命名空间模式

多 Schema 命名空间

memory_tool = create_manage_memory_tool(
    namespace=("memories", "{langgraph_user_id}", "user_profile"),
    schema=UserProfile,
)

资料来源：src/langmem/knowledge/tools.py:200-300

异步与同步 API

异步调用

# 异步提取
updated_memories = await manager.ainvoke(
    {"messages": conversation, "existing": memories}
)

# 异步搜索
results = await manager.asearch(query="preferences", limit=10)

# 异步存储
await manager.aput("key", {"content": "value"})

同步调用

# 同步提取
updated_memories = manager.invoke({"messages": conversation})

# 同步搜索
results = manager.search(query="preferences", limit=10)

与短时记忆的集成

LangMem 同时提供短时记忆（摘要）功能，可与情景记忆协同工作：

from langmem.short_term import SummarizationNode, RunningSummary

summarization_node = SummarizationNode(
    model=summarization_model,
    max_tokens=256,
    max_tokens_before_summary=256,
    max_summary_tokens=128,
)

工作流程：

graph TD
    A[对话历史] --> B[短时记忆摘要]
    B --> C[情景记忆提取]
    C --> D[结构化记忆存储]
    D --> E[后续对话检索]

资料来源：src/langmem/short_term/summarization.py:1-100

最佳实践

Schema 设计原则

1. 原子性：每个记忆项应包含最小独立信息单元
2. 可演进性：使用可选字段和默认值支持功能扩展
3. 可检索性：在 Schema 中包含可用于过滤的分类字段

性能优化

1. 选择合适的查询模型：使用轻量级模型处理检索查询
2. 设置合理的查询限制：根据实际需求调整 query_limit
3. 使用命名空间隔离：避免大规模检索时的性能瓶颈

错误处理

try:
    result = await manager.ainvoke({"messages": conversation})
except Exception as e:
    logger.error(f"Memory extraction failed: {e}")
    # 降级策略：使用空记忆继续
    result = []

总结

情景记忆提取是 LangMem 库实现持久化上下文管理的核心能力。通过结构化的 Schema 定义、灵活的存储层集成和智能的提取算法，开发者可以轻松构建具有持续学习能力的 AI 应用。该系统既支持独立使用，也与 LangGraph 框架深度集成，满足从简单脚本到复杂生产环境的各种需求。

Pitfall Log / 踩坑日志

项目：langchain-ai/langmem

摘要：发现 11 个潜在踩坑项，其中 3 个为 high/blocking；最高优先级：安装坑 - 来源证据：GRAPH_RECURSION_LIMIT。

1. 安装坑 · 来源证据：GRAPH_RECURSION_LIMIT

• 严重度：high
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：GRAPH_RECURSION_LIMIT
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_82b923de03f34d5a9fa8530916b40d14 | https://github.com/langchain-ai/langmem/issues/133 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

2. 安装坑 · 来源证据：Persistence?

• 严重度：high
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安装相关的待验证问题：Persistence?
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_b9688483fa44468a96801d46825b040f | https://github.com/langchain-ai/langmem/issues/154 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

3. 安全/权限坑 · 来源证据：Enhance error message when summarization fails due to missing HumanMessage in trimmed window

• 严重度：high
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Enhance error message when summarization fails due to missing HumanMessage in trimmed window
• 对用户的影响：可能影响授权、密钥配置或安全边界。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_15b3f9c4829745339b470171f005b3df | https://github.com/langchain-ai/langmem/issues/156 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

4. 配置坑 · 可能修改宿主 AI 配置

• 严重度：medium
• 证据强度：source_linked
• 发现：项目面向 Claude/Cursor/Codex/Gemini/OpenCode 等宿主，或安装命令涉及用户配置目录。
• 对用户的影响：安装可能改变本机 AI 工具行为，用户需要知道写入位置和回滚方法。
• 建议检查：列出会写入的配置文件、目录和卸载/回滚步骤。
• 防护动作: 涉及宿主配置目录时必须给回滚路径，不能只给安装命令。
• 证据：capability.host_targets | github_repo:920242883 | https://github.com/langchain-ai/langmem | host_targets=claude, chatgpt

5. 能力坑 · 能力判断依赖假设

• 严重度：medium
• 证据强度：source_linked
• 发现：README/documentation is current enough for a first validation pass.
• 对用户的影响：假设不成立时，用户拿不到承诺的能力。
• 建议检查：将假设转成下游验证清单。
• 防护动作: 假设必须转成验证项；没有验证结果前不能写成事实。
• 证据：capability.assumptions | github_repo:920242883 | https://github.com/langchain-ai/langmem | README/documentation is current enough for a first validation pass.

6. 维护坑 · 维护活跃度未知

• 严重度：medium
• 证据强度：source_linked
• 发现：未记录 last_activity_observed。
• 对用户的影响：新项目、停更项目和活跃项目会被混在一起，推荐信任度下降。
• 建议检查：补 GitHub 最近 commit、release、issue/PR 响应信号。
• 防护动作: 维护活跃度未知时，推荐强度不能标为高信任。
• 证据：evidence.maintainer_signals | github_repo:920242883 | https://github.com/langchain-ai/langmem | last_activity_observed missing

7. 安全/权限坑 · 下游验证发现风险项

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：下游已经要求复核，不能在页面中弱化。
• 建议检查：进入安全/权限治理复核队列。
• 防护动作: 下游风险存在时必须保持 review/recommendation 降级。
• 证据：downstream_validation.risk_items | github_repo:920242883 | https://github.com/langchain-ai/langmem | no_demo; severity=medium

8. 安全/权限坑 · 存在评分风险

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：风险会影响是否适合普通用户安装。
• 建议检查：把风险写入边界卡，并确认是否需要人工复核。
• 防护动作: 评分风险必须进入边界卡，不能只作为内部分数。
• 证据：risks.scoring_risks | github_repo:920242883 | https://github.com/langchain-ai/langmem | no_demo; severity=medium

9. 安全/权限坑 · 来源证据：Security: OWASP Agent Memory Guard for memory poisoning defense (ASI06)

• 严重度：medium
• 证据强度：source_linked
• 发现：GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题：Security: OWASP Agent Memory Guard for memory poisoning defense (ASI06)
• 对用户的影响：可能增加新用户试用和生产接入成本。
• 建议检查：来源问题仍为 open，Pack Agent 需要复核是否仍影响当前版本。
• 防护动作: 不得脱离来源链接放大为确定性结论；需要标注适用版本和复核状态。
• 证据：community_evidence:github | cevd_760d72a9519c42a8b1e31d8932383bc2 | https://github.com/langchain-ai/langmem/issues/164 | 来源讨论提到 python 相关条件，需在安装/试用前复核。

10. 维护坑 · issue/PR 响应质量未知

• 严重度：low
• 证据强度：source_linked
• 发现：issue_or_pr_quality=unknown。
• 对用户的影响：用户无法判断遇到问题后是否有人维护。
• 建议检查：抽样最近 issue/PR，判断是否长期无人处理。
• 防护动作: issue/PR 响应未知时，必须提示维护风险。
• 证据：evidence.maintainer_signals | github_repo:920242883 | https://github.com/langchain-ai/langmem | issue_or_pr_quality=unknown

11. 维护坑 · 发布节奏不明确

• 严重度：low
• 证据强度：source_linked
• 发现：release_recency=unknown。
• 对用户的影响：安装命令和文档可能落后于代码，用户踩坑概率升高。
• 建议检查：确认最近 release/tag 和 README 安装命令是否一致。
• 防护动作: 发布节奏未知或过期时，安装说明必须标注可能漂移。
• 证据：evidence.maintainer_signals | github_repo:920242883 | https://github.com/langchain-ai/langmem | release_recency=unknown