# https://github.com/langchain-ai/langmem 项目说明书
生成时间: 2026-05-21 16:18:01 UTC
## 目录
- [LangMem 概述](#page-overview)
- [安装与配置](#page-installation)
- [热路径快速开始](#page-hot-path-quickstart)
- [后台处理快速开始](#page-background-quickstart)
- [系统架构](#page-system-architecture)
- [记忆工具](#page-memory-tools)
- [记忆提取](#page-memory-extraction)
- [提示优化](#page-prompt-optimization)
- [语义记忆管理](#page-semantic-memory)
- [情景记忆提取](#page-episodic-memory)
## LangMem 概述
### 相关页面
相关主题：[安装与配置](#page-installation), [系统架构](#page-system-architecture)
相关源码文件
以下源码文件用于生成本页说明：
- [src/langmem/prompts/types.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/types.py)
- [src/langmem/knowledge/extraction.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/extraction.py)
- [src/langmem/knowledge/tools.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/tools.py)
- [src/langmem/short_term/summarization.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/short_term/summarization.py)
- [src/langmem/prompts/optimization.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/optimization.py)
- [src/langmem/prompts/gradient.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/gradient.py)
- [src/langmem/prompts/prompt.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/prompt.py)
# LangMem 概述
LangMem 是由 LangChain 团队开发的一个开源库，专注于为大型语言模型（LLM）代理提供**记忆（Memory）管理**和**提示优化（Prompt Optimization）**能力。该库帮助开发者构建能够跨会话学习、适应用户偏好并持续改进的智能代理系统。
## 核心定位
LangMem 填补了 LLM 代理在以下方面的能力空白：
| 能力维度 | 说明 |
|---------|------|
| **长期记忆** | 从对话历史中提取、存储和检索结构化记忆 |
| **短期记忆** | 对话过程中的实时摘要和信息压缩 |
| **提示优化** | 基于反馈和历史轨迹自动优化系统提示 |
资料来源：[src/langmem/prompts/types.py:1-15]()
## 整体架构
LangMem 的架构围绕三大核心模块展开，每个模块负责不同的记忆或优化职责：
```mermaid
graph TD
A[LangMem] --> B[knowledge
长期记忆模块]
A --> C[short_term
短期记忆模块]
A --> D[prompts
提示优化模块]
B --> B1[extraction.py
记忆提取管理]
B --> B2[tools.py
记忆操作工具]
C --> C1[summarization.py
对话摘要]
D --> D1[optimization.py
优化器工厂]
D --> D2[gradient.py
梯度优化器]
D --> D3[prompt.py
提示模板]
D --> D4[types.py
类型定义]
```
## 核心数据类型
LangMem 使用 TypeScript 风格的 TypedDict 定义数据结构，确保类型安全和清晰的接口契约。
资料来源：[src/langmem/prompts/types.py:1-90]()
### Prompt（提示定义）
```python
class Prompt(TypedDict, total=False):
name: Required[str] # 提示名称，唯一标识
prompt: Required[str] # 提示内容
update_instructions: str | None # 更新指南
when_to_update: str | None # 更新依赖条件
```
### AnnotatedTrajectory（标注轨迹）
用于记录对话历史及其反馈信息：
```python
class AnnotatedTrajectory(typing.NamedTuple):
messages: list[AnyMessage] # 对话消息列表
feedback: dict[str, typing.Any] # 反馈信息（如 developer_feedback、score）
```
### 优化器输入类型
| 类型 | 用途 | 关键字段 |
|-----|------|---------|
| `PromptOptimizerInput` | 单提示优化 | `trajectories`, `prompt` |
| `MultiPromptOptimizerInput` | 多提示协同优化 | `trajectories`, `prompts` |
资料来源：[src/langmem/prompts/types.py:50-120]()
## 长期记忆模块
长期记忆模块负责从对话中提取、存储和检索用户偏好及知识信息。
资料来源：[src/langmem/knowledge/extraction.py:1-100]()
### 核心组件
#### MemoryManager（记忆管理器）
`create_memory_manager` 函数创建核心的记忆管理器：
```python
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 存储层集成的管理器：
```python
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` 函数创建语义搜索管道：
```python
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 的工具：
#### 记忆操作工具
```python
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` | 创建新记忆 |
| `update` | 更新现有记忆 |
| `upsert` | 存在则更新，不存在则创建 |
| `delete` | 删除指定记忆 |
#### 记忆搜索工具
```python
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 工作流中的消息摘要：
```python
summarization_node = SummarizationNode(
model=summarization_model, # 专用的轻量摘要模型
max_tokens=256, # 摘要最大 token 数
max_tokens_before_summary=256, # 触发摘要的消息长度阈值
max_summary_tokens=128, # 单次摘要最大 token
)
```
### RunningSummary（运行摘要）
状态管理对象，用于跟踪和管理对话摘要：
```python
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]()
### 优化策略类型
| 策略 | 说明 | 配置参数 |
|-----|------|---------|
| `gradient` | 梯度式优化，基于假设-验证循环 | `max_reflection_steps` |
| `metaprompt` | 元提示式优化，使用结构化推理 | `max_reflection_steps`, `min_reflection_steps` |
| `prompt_memory` | 记忆式优化，从历史中学习模式 | 无特殊配置 |
### 单提示优化器
```python
def create_prompt_optimizer(
model: str | BaseChatModel,
kind: Literal["gradient", "metaprompt", "prompt_memory"] = "gradient",
config: dict | None = None,
) -> PromptOptimizer
```
**输入结构**：
```python
{
"trajectories": [
AnnotatedTrajectory(
messages=[...],
feedback={"clarity": "needs more structure"}
)
],
"prompt": "You are an astronomy expert"
}
```
资料来源：[src/langmem/prompts/optimization.py:80-120]()
### 多提示优化器
```python
def create_multi_prompt_optimizer(
model: str | BaseChatModel,
kind: Literal["gradient", "metaprompt", "prompt_memory"] = "gradient",
config: dict | None = None,
) -> MultiPromptOptimizer
```
**输入结构**：
```python
{
"trajectories": [...], # 多个标注轨迹
"prompts": [
Prompt(name="research", prompt="Research the given topic"),
Prompt(name="summarize", prompt="Summarize findings"),
]
}
```
### 梯度优化器详解
资料来源：[src/langmem/prompts/gradient.py:1-60]()
梯度优化器通过迭代的"假设-验证"循环改进提示：
```python
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 预定义了两类优化提示模板：
#### 单轨迹反思模板
```python
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}
Here was the agent's trajectory:
{trajectory}
Here is the user's feedback:
{feedback}
...
"""
```
#### 多轨迹批量反思模板
```python
INSTRUCTION_REFLECTION_MULTIPLE_PROMPT = """..."""
# 使用 标签批量包含多个轨迹-反馈对
```
**响应格式**：
```python
class GeneralResponse(TypedDict):
logic: str # 优化逻辑说明
update_prompt: bool # 是否需要更新
new_prompt: str # 新提示内容
```
## 与 LangGraph 集成
LangMem 被设计为与 LangGraph 无缝协作，利用 LangGraph 的以下组件：
| LangGraph 组件 | LangMem 集成点 |
|---------------|---------------|
| `BaseStore` | 记忆的持久化存储后端 |
| `InMemoryStore` | 开发/测试用内存存储 |
| `@entrypoint` | 异步工作流装饰器 |
| `create_react_agent` | 集成记忆工具的 ReAct 代理 |
| `RunnableConfig` | 命名空间和配置管理 |
### 命名空间机制
记忆使用层级命名空间组织：
```python
namespace = ("memories", "{langgraph_user_id}")
# 示例: ("memories", "user123")
```
命名空间模板支持运行时占位符替换，确保多用户数据隔离。
### 后台反射执行器
```python
from langmem import ReflectionExecutor
reflection = ReflectionExecutor(manager, store=store)
```
`ReflectionExecutor` 支持将记忆处理任务后台化：
```mermaid
sequenceDiagram
participant Agent
participant Background
participant Store
Agent->>Agent: process message
Agent-->>User: response
Agent->>Background: schedule enrichment
(after_seconds=0)
Note over Background,Store: Memory processing happens
in background thread
```
## API 速查表
### 创建函数一览
| 函数 | 返回类型 | 用途 |
|-----|---------|------|
| `create_memory_manager` | `MemoryManager` | 创建记忆管理器 |
| `create_memory_searcher` | `Runnable` | 创建搜索管道 |
| `create_memory_store_manager` | `MemoryStoreManager` | 创建存储管理器 |
| `create_thread_extractor` | 提取器 | 创建线程提取器 |
| `create_prompt_optimizer` | `PromptOptimizer` | 创建单提示优化器 |
| `create_multi_prompt_optimizer` | `MultiPromptOptimizer` | 创建多提示优化器 |
| `create_manage_memory_tool` | `BaseTool` | 创建记忆操作工具 |
| `create_search_memory_tool` | `BaseTool` | 创建记忆搜索工具 |
## 总结
LangMem 提供了一套完整的记忆和提示优化解决方案：
- **长期记忆**：通过结构化 schema 和语义搜索实现持久化知识管理
- **短期记忆**：通过实时摘要保持对话上下文的高效利用
- **提示优化**：通过多种策略（梯度、元提示、记忆学习）持续改进代理行为
该库与 LangGraph 深度集成，既可以作为独立组件使用，也能作为 LangGraph 代理的能力扩展。
---
## 安装与配置
### 相关页面
相关主题：[LangMem 概述](#page-overview), [记忆工具](#page-memory-tools)
相关源码文件
以下源码文件用于生成本页说明：
- [pyproject.toml](https://github.com/langchain-ai/langmem/blob/main/pyproject.toml)
- [README.md](https://github.com/langchain-ai/langmem/blob/main/README.md)
- [src/langmem/knowledge/extraction.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/extraction.py)
- [src/langmem/prompts/types.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/types.py)
- [src/langmem/knowledge/tools.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/tools.py)
- [src/langmem/short_term/summarization.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/short_term/summarization.py)
- [src/langmem/prompts/optimization.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/optimization.py)
# 安装与配置
## 概述
LangMem 是一个用于构建具有记忆能力的大语言模型应用的 Python 库。该库提供了一套完整的工具，用于管理对话记忆、提示优化和短期记忆总结等功能。安装与配置章节将详细介绍如何正确设置 LangMem 环境、配置依赖项、以及初始化各类管理器组件。
## 环境要求
### Python 版本要求
LangMem 需要 Python 3.10 或更高版本。建议使用 Python 3.11 以获得最佳性能和兼容性。
### 核心依赖
LangMem 的核心依赖包括：
| 依赖包 | 最低版本 | 用途说明 |
|--------|----------|----------|
| langchain-core | 最新稳定版 | 核心语言模型接口 |
| langchain-anthropic | 最新稳定版 | Anthropic 模型支持 |
| langchain-openai | 最新稳定版 | OpenAI 模型支持 |
| langgraph | 最新稳定版 | 图状态管理和工作流 |
| pydantic | 2.x | 数据验证和模型定义 |
## 安装方式
### 使用 uv 安装（推荐）
```bash
uv add langmem
```
### 使用 pip 安装
```bash
pip install langmem
```
### 从源码安装
```bash
cd examples/standalone_examples
uv venv
source .venv/bin/activate
uv sync
```
## 环境变量配置
### API 密钥设置
使用 LangMem 前需要配置相应的 API 密钥：
```bash
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]()
#### 基础配置示例
```python
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],
)
```
#### 配置参数说明
| 参数名 | 类型 | 默认值 | 说明 |
|--------|------|--------|------|
| model | str \| BaseChatModel | 必需 | 用于记忆处理的主模型 |
| schemas | list[type[BaseModel]] | 必需 | 记忆数据的 Pydantic schema 定义 |
| instructions | str \| None | None | 自定义指令模板 |
| enable_inserts | bool | True | 启用创建新记忆 |
| enable_updates | bool | True | 启用更新现有记忆 |
| enable_deletes | bool | True | 启用删除记忆 |
### 记忆存储管理器 (Memory Store Manager)
`create_memory_store_manager` 函数创建支持持久化存储的记忆管理器，支持同步和异步操作。资料来源：[src/langmem/knowledge/extraction.py:1-200]()
#### 配置示例
```python
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}"),
)
```
#### 命名空间配置
命名空间使用模板占位符，支持运行时配置：
```python
namespace = ("memories", "{langgraph_user_id}")
```
在调用时通过配置传入实际值：
```python
config = {"configurable": {"langgraph_user_id": "user123"}}
```
### 记忆搜索器 (Memory Searcher)
`create_memory_searcher` 创建自动生成查询的记忆搜索管道。资料来源：[src/langmem/knowledge/extraction.py:1-200]()
```python
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]()
```python
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]()
#### 单一提示优化
```python
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",
})
```
#### 多提示优化
```python
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},
)
```
#### 优化类型配置
| 类型 | 说明 | 配置参数 |
|------|------|----------|
| metaprompt | 元提示优化 | max_reflection_steps, min_reflection_steps |
| gradient | 梯度下降优化 | gradient_prompt, metaprompt, max_reflection_steps |
| prompt_memory | 记忆驱动优化 | 无特殊配置 |
### 短期记忆总结
`SummarizationNode` 用于 LangGraph 状态图中的消息总结。资料来源：[src/langmem/short_term/summarization.py:1-100]()
```python
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]()
```python
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]()
```python
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},
)
```
## 架构流程图
### 记忆管理流程
```mermaid
graph TD
A[用户消息] --> B[记忆管理器]
B --> C{操作类型}
C -->|创建| D[提取新记忆]
C -->|更新| E[合并现有记忆]
C -->|删除| F[移除过时记忆]
D --> G[存储到向量数据库]
E --> G
F --> G
G --> H[返回结果]
```
### 提示优化流程
```mermaid
graph TD
A[对话历史] --> B[提示优化器]
B --> C[元提示优化]
B --> D[梯度优化]
B --> E[记忆驱动优化]
C --> F[优化后的提示]
D --> F
E --> F
```
## 集成配置
### LangGraph 集成
LangMem 可以与 LangGraph 的 ReAct Agent 无缝集成：
```python
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` 在后台异步执行记忆处理：
```python
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）以提高应用性能
---
## 热路径快速开始
### 相关页面
相关主题：[后台处理快速开始](#page-background-quickstart), [记忆工具](#page-memory-tools)
相关源码文件
以下源码文件用于生成本页说明：
- [docs/docs/hot_path_quickstart.md](https://github.com/langchain-ai/langmem/blob/main/docs/docs/hot_path_quickstart.md)
- [src/langmem/knowledge/tools.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/tools.py)
- [src/langmem/prompts/types.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/types.py)
- [src/langmem/knowledge/extraction.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/extraction.py)
- [src/langmem/short_term/summarization.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/short_term/summarization.py)
# 热路径快速开始
热路径（Hot Path）是指在 LangMem 库中实现的一种快速、简化的方式来为 LangGraph 智能体添加工具和内存管理功能。该功能允许开发者通过最少的配置，快速将内存管理能力集成到现有应用中。
## 功能概述
热路径快速开始模块主要提供以下核心能力：
| 功能 | 描述 |
|------|------|
| 记忆管理工具创建 | 通过 `create_manage_memory_tool` 创建可操作的记忆管理工具 |
| 操作权限控制 | 支持限制允许的操作类型（create、update、delete） |
| 命名空间管理 | 支持层级化命名空间配置 |
| Schema 验证 | 支持自定义数据结构 Schema |
| LangGraph 集成 | 与 LangGraph 的 `create_react_agent` 无缝集成 |
## 核心 API
### create_manage_memory_tool
创建记忆管理工具的主函数，定义在 `src/langmem/knowledge/tools.py` 中。
```python
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:
```
**参数说明：**
| 参数 | 类型 | 默认值 | 描述 |
|------|------|--------|------|
| `namespace` | `tuple[str, ...]` | `("memories", "{langgraph_user_id}")` | 层级化命名空间，支持运行时占位符如 `{langgraph_user_id}` |
| `actions_permitted` | `Sequence[Literal["create", "update", "delete"]]` | `("create", "update", "delete")` | 允许的操作类型列表 |
| `schema` | `type[BaseModel] \| None` | `None` | 可选的数据模型，用于验证记忆内容 |
| `store` | `BaseStore \| None` | `None` | 底层存储后端 |
资料来源：[src/langmem/knowledge/tools.py:36-43]()
## 工作流程
### 基础集成流程
```mermaid
graph TD
A[创建工具] --> B[配置命名空间]
B --> C[设置操作权限]
C --> D[集成到 Agent]
D --> E[运行时执行]
```
### 工具执行流程
当智能体调用记忆管理工具时，内部执行以下流程：
```mermaid
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: 返回结果
```
## 使用示例
### 基础用法
创建支持所有操作的记忆管理工具：
```python
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]()
### 仅创建操作
限制工具只能创建新记忆：
```python
tool = create_manage_memory_tool(
namespace=("memories", "{user_id}"),
actions_permitted=["create"], # 仅允许创建
)
```
### 配合系统提示词使用
在 LangGraph 中获取记忆并注入到系统提示词：
```python
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}
"""
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]()
## 配置选项详解
### 命名空间配置
命名空间支持层级结构和运行时变量替换：
| 配置方式 | 示例 | 说明 |
|----------|------|------|
| 静态命名空间 | `("memories", "固定用户ID")` | 使用固定值 |
| 动态命名空间 | `("memories", "{langgraph_user_id}")` | 从配置中读取 |
| 多级命名空间 | `("app", "users", "{user_id}", "memories")` | 支持任意层级 |
资料来源：[src/langmem/knowledge/tools.py:36]()
### 操作权限矩阵
| 权限组合 | 效果 |
|----------|------|
| `["create"]` | 仅能创建新记忆 |
| `["create", "update"]` | 可创建和更新 |
| `["create", "update", "delete"]` | 完整 CRUD 操作 |
| `["delete"]` | 仅能删除（较少使用） |
### 自定义 Schema
可以使用 Pydantic 模型定义记忆的数据结构：
```python
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` 内部通过组合多个组件实现：
```python
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` 是核心的异步处理函数：
| 参数 | 类型 | 说明 |
|------|------|------|
| `content` | `schema \| None` | 记忆内容数据 |
| `action` | `Literal[actions_permitted]` | 要执行的操作 |
| `id` | `UUID \| None` | 可选的记忆标识符 |
## 与其他模块的关系
```mermaid
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]
```
### 相关模块
| 模块 | 文件路径 | 用途 |
|------|----------|------|
| 记忆管理 | `src/langmem/knowledge/extraction.py` | 提供 `create_memory_manager` 等高级封装 |
| 记忆搜索 | `src/langmem/knowledge/extraction.py` | 提供 `create_memory_searcher` |
| 提示词类型 | `src/langmem/prompts/types.py` | 定义 `Prompt`、`AnnotatedTrajectory` 等数据结构 |
| 短时记忆 | `src/langmem/short_term/summarization.py` | 提供会话总结能力 |
## 最佳实践
### 1. 限制操作权限
根据业务需求限制允许的操作，避免过度权限：
```python
# 推荐：按需授权
tool = create_manage_memory_tool(
actions_permitted=["create"], # 只允许创建
)
```
### 2. 使用 Schema 验证
对记忆内容进行结构化定义：
```python
class PreferenceMemory(BaseModel):
preference: str
context: str
category: str = "general"
tool = create_manage_memory_tool(schema=PreferenceMemory)
```
### 3. 合理设计命名空间
```python
# 推荐：用户隔离
namespace=("app", "users", "{user_id}", "memories")
# 避免：过于扁平
namespace=("memories",)
```
## 错误处理
| 错误情况 | 处理方式 |
|----------|----------|
| `actions_permitted` 为空 | 抛出 `ValueError` |
| 操作不在允许列表中 | 抛出 `ValueError` |
| Schema 验证失败 | 由 Pydantic 处理验证错误 |
| Store 不可用 | 依赖 LangGraph 配置 |
## 总结
热路径快速开始模块为 LangGraph 应用提供了便捷的记忆管理工具创建方式。通过 `create_manage_memory_tool`，开发者可以：
1. **快速集成**：几行代码即可添加工具到智能体
2. **灵活配置**：通过参数控制操作权限和数据结构
3. **类型安全**：支持 Schema 验证和类型提示
4. **无缝衔接**：与 LangGraph 生态完美配合
该模块是 LangMem 库的核心组件之一，适用于需要用户级记忆管理的对话应用场景。
---
## 后台处理快速开始
### 相关页面
相关主题：[热路径快速开始](#page-hot-path-quickstart), [记忆提取](#page-memory-extraction)
相关源码文件
以下源码文件用于生成本页说明：
- [src/langmem/knowledge/extraction.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/extraction.py)
- [src/langmem/prompts/types.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/types.py)
- [src/langmem/short_term/summarization.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/short_term/summarization.py)
- [src/langmem/knowledge/tools.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/tools.py)
- [examples/standalone_examples/README.md](https://github.com/langchain-ai/langmem/blob/main/examples/standalone_examples/README.md)
# 后台处理快速开始
## 概述
LangMem 的后台处理功能允许在独立线程中异步执行内存管理操作，从而避免阻塞主应用程序流程。通过 `ReflectionExecutor` 类，开发者可以将记忆提取和管理任务后台化，提升应用程序的响应速度和整体性能。
## 核心组件
### ReflectionExecutor
`ReflectionExecutor` 是 LangMem 提供的后台执行器，用于在后台线程中处理记忆操作。
```mermaid
classDiagram
class ReflectionExecutor {
+execute(input, config)
+ainvoke(input, config)
}
class MemoryManager {
+ainvoke(input)
+search()
}
class BaseStore {
+asearch()
+aput()
+adelete()
}
ReflectionExecutor --> MemoryManager
ReflectionExecutor --> BaseStore
```
**导入路径**：
```python
from langmem import ReflectionExecutor
```
## 工作流程
### 异步处理架构
```mermaid
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. 记忆被提取并存储，不阻塞主线程
## 快速开始指南
### 环境准备
确保已安装必要的依赖：
```bash
uv venv
source .venv/bin/activate
uv sync
```
### 基本配置
```python
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)
```
### 创建带后台处理的代理
```python
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,
)
```
## 后台执行配置
### 命名空间配置
后台处理支持动态命名空间，允许为不同用户维护独立的记忆空间：
```python
namespace = ("memories", "{langgraph_user_id}")
```
**配置参数说明**：
| 参数 | 类型 | 说明 |
|------|------|------|
| `namespace` | tuple[str, ...] | 记忆存储的命名空间结构 |
| `{langgraph_user_id}` | 占位符 | 运行时替换为用户ID |
### 存储配置
InMemoryStore 适用于开发和测试环境：
```python
store = InMemoryStore(
index={
"dims": 1536,
"embed": "openai:text-embedding-3-small",
}
)
```
## 独立使用示例
对于不依赖 LangGraph 的应用，可以使用独立的内存管理器：
```python
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}
)
```
## 架构设计
### 组件关系
```mermaid
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`
### 配置示例
```python
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 概述](#page-overview), [记忆工具](#page-memory-tools), [提示优化](#page-prompt-optimization)
相关源码文件
以下源码文件用于生成本页说明：
- [src/langmem/__init__.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/__init__.py)
- [src/langmem/knowledge/__init__.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/__init__.py)
- [src/langmem/prompts/__init__.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/__init__.py)
- [langgraph.json](https://github.com/langchain-ai/langmem/blob/main/langgraph.json)
- [src/langmem/memory/__init__.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/memory/__init__.py)
- [src/langmem/evaluation/__init__.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/evaluation/__init__.py)
# 系统架构
## 概述
LangMem 是由 LangChain AI 开发的大语言模型（LLM）记忆管理库，专注于为 AI 应用提供持久化、可检索的记忆存储能力。该库采用模块化架构设计，将记忆管理、提示工程、知识存储等功能解耦，同时通过 LangGraph 实现复杂的状态管理工作流。LangMem 的核心目标是让 AI 应用能够跨会话持久化记忆，并支持多种向量存储后端以实现高效的语义检索。
LangMem 架构遵循以下设计原则：低耦合度、高可扩展性、多后端兼容性以及类型安全。整个系统由多个功能模块组成，包括记忆模块（memory）、知识模块（knowledge）、提示模块（prompts）以及评估模块（evaluation），每个模块都暴露统一的公共接口供外部调用。
## 整体架构图
```mermaid
graph TB
subgraph "表现层"
API[公共 API 入口]
end
subgraph "核心模块"
MEM[记忆模块
memory]
KNOW[知识模块
knowledge]
PROM[提示模块
prompts]
EVAL[评估模块
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 的核心组件，负责管理和持久化对话历史与用户偏好信息。该模块实现了多种记忆类，支持不同的存储策略和使用场景。
| 类名 | 功能描述 | 存储方式 |
|------|----------|----------|
| `SummaryMemory` | 对话摘要记忆，自动生成对话摘要 | 内存/持久化 |
| `ConversationMemory` | 完整对话历史记忆 | 向量存储 |
| `EntityMemory` | 实体信息记忆，提取和管理实体 | 键值存储 |
| `UserPreferenceMemory` | 用户偏好记忆 | 持久化存储 |
记忆模块通过 LangGraph 实现状态的序列化和反序列化，支持将对话状态持久化到多种后端存储系统。模块内部维护了一个状态图定义，允许开发者定义复杂的状态转换逻辑。
资料来源：[src/langmem/memory/__init__.py:1-50]()
### 知识模块（knowledge）
知识模块提供了结构化知识存储和检索能力，支持将文档、实体和关系持久化到向量数据库中。该模块利用 LangChain 的文档加载器和文本分割工具，实现大规模知识库的管理。
```mermaid
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` 配置文件定义状态机工作流，实现复杂的状态管理和多轮对话逻辑。
```mermaid
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 配置文件定义了应用的状态结构，包括以下关键字段：
| 字段名 | 类型 | 描述 |
|--------|------|------|
| `messages` | List[BaseMessage] | 对话消息历史 |
| `memory_state` | Dict | 当前记忆状态 |
| `context` | Dict | 检索到的上下文 |
| `metadata` | Dict | 元数据信息 |
### 节点与边
工作流由多个节点组成：
- **retrieve_node**：从向量存储检索相关记忆
- **update_node**：更新记忆状态
- **generate_node**：调用 LLM 生成响应
- **evaluate_node**：评估当前状态质量
资料来源：[langgraph.json:1-100]()
## 公共 API 入口
`src/langmem/__init__.py` 文件作为库的公共入口，导出所有公共 API 和核心类。该模块遵循 LangChain 的导出规范，提供清晰的命名空间组织和类型提示。
```python
# 主要导出内容
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]()
## 数据流架构
### 写入流程
```mermaid
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: 确认写入
```
### 读取流程
```mermaid
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 支持多种向量存储后端，提供了统一的上层抽象接口。
| 后端类型 | 支持功能 | 适用场景 |
|----------|----------|----------|
| 内存存储 | 快速原型开发 | 开发调试 |
| SQLite | 本地持久化 | 小规模应用 |
| Chroma | 嵌入式向量 | 单机应用 |
| Pinecone | 云端托管 | 生产环境 |
| Weaviate | 混合搜索 | 企业级应用 |
| Qdrant | 高性能检索 | 实时应用 |
存储层采用适配器模式，每种后端都实现了统一的接口：
- `add_documents()`：添加文档
- `similarity_search()`：相似度搜索
- `delete()`：删除文档
- `update()`：更新文档
## 扩展性设计
### 插件接口
LangMem 通过以下方式支持扩展：
1. **自定义记忆类**：继承基础 `BaseMemory` 类并实现必需方法
2. **自定义检索器**：实现 `BaseRetriever` 接口
3. **自定义后端**：实现存储适配器接口
4. **自定义评估器**：扩展评估指标集
### 配置机制
系统通过以下方式支持灵活配置：
- 环境变量配置
- YAML/JSON 配置文件
- 运行时参数注入
- 依赖注入容器
## 技术栈总结
| 层级 | 技术选型 | 作用 |
|------|----------|------|
| 核心框架 | Python 3.10+ | 开发语言 |
| LLM 集成 | LangChain | 模型调用抽象 |
| 状态管理 | LangGraph | 工作流编排 |
| 类型系统 | Pydantic | 数据验证 |
| 向量存储 | 多后端支持 | 持久化存储 |
| 文档处理 | LangChain 生态 | 文本处理 |
LangMem 的架构设计充分考虑了模块化、可扩展性和易用性，通过清晰的层次划分和标准化的接口定义，为开发者提供了构建智能记忆系统的基础设施。
---
## 记忆工具
### 相关页面
相关主题：[系统架构](#page-system-architecture), [记忆提取](#page-memory-extraction)
相关源码文件
以下源码文件用于生成本页说明：
- [src/langmem/knowledge/tools.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/tools.py)
- [src/langmem/knowledge/extraction.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/extraction.py)
- [src/langmem/knowledge/manager.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/manager.py)
- [src/langmem/prompts/types.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/types.py)
- [examples/standalone_examples/README.md](https://github.com/langchain-ai/langmem/blob/main/examples/standalone_examples/README.md)
# 记忆工具
LangMem 的记忆工具（Memory Tools）是一组用于在 LangGraph 应用中管理长期记忆的函数式工具集。这些工具以 LangChain Tool 的形式实现，允许 AI 助手通过函数调用来创建、更新、删除和搜索记忆内容。
## 概述与核心定位
记忆工具是 LangMem 知识管理系统的两大核心组件之一（另一个是记忆管理器）。与自动化的 `MemoryStoreManager` 不同，记忆工具将记忆的管理权直接交给 AI 代理，使其能够根据对话上下文主动决定何时以及如何操作记忆存储。
### 主要功能特性
| 功能 | 说明 |
|------|------|
| 创建记忆 | AI 可主动将重要信息存储为持久记忆 |
| 更新记忆 | 修改已存在的记忆内容 |
| 删除记忆 | 移除不再需要的记忆 |
| 搜索记忆 | 基于语义或精确匹配查找相关记忆 |
资料来源：[src/langmem/knowledge/tools.py:1-50]()
## 架构设计
### 工具类型
LangMem 提供两种主要的记忆工具：
```mermaid
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 实例：
```mermaid
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 工具。
### 函数签名
```python
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",
)
```
### 参数说明
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `namespace` | `tuple[str, ...] \| str` | 必需 | 记忆的命名空间路径，支持占位符如 `{langgraph_user_id}` |
| `schema` | `type[BaseModel] \| None` | `None` | 记忆内容的数据模型（Pydantic BaseModel） |
| `actions_permitted` | `list[Literal["create", "update", "delete"]] \| None` | `None` | 允许的操作类型列表 |
| `instructions` | `str` | `""` | 给 AI 的额外使用说明 |
| `store` | `BaseStore \| None` | `None` | 底层存储实例 |
| `name` | `str` | `"manage_memory"` | 工具名称 |
资料来源：[src/langmem/knowledge/tools.py:120-175]()
### 命名空间配置
命名空间支持运行时配置替换，通过 `{config_key}` 占位符实现：
```python
memory_tool = create_manage_memory_tool(
namespace=("memories", "{langgraph_user_id}", "user_profile"),
)
```
运行时配置示例：
```python
config = {"configurable": {"langgraph_user_id": "user-123"}}
# 实际命名空间: ("memories", "user-123", "user_profile")
```
资料来源：[src/langmem/knowledge/tools.py:140-160]()
### 自定义 Schema
可以通过 Pydantic 模型定义记忆的数据结构：
```python
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 可执行的操作：
```python
# 仅允许创建和更新，不允许删除
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 工具。
### 函数签名
```python
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",
)
```
### 参数说明
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `namespace` | `tuple[str, ...] \| str` | 必需 | 搜索范围的命名空间 |
| `instructions` | `str` | `""` | 搜索工具的使用说明 |
| `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)`
```python
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 代理中：
```python
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}
"""
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]()
### 完整的工作流示例
```mermaid
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 外部的独立应用中使用：
```python
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：
```python
# 方式一：创建时指定 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` 配置解析：
```python
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 主动调用 | 自动处理 |
| 适用场景 | 需要 AI 自主决策 | 需要自动化流程 |
| 灵活性 | 高 | 中 |
| 复杂度 | 低 | 中 |
| 批量操作 | 不支持 | 支持 |
记忆工具适合需要 AI 自主管理记忆的场景，而记忆管理器（如 `create_memory_manager`、`create_memory_store_manager`）适合需要自动从对话中提取和组织记忆的场景。
资料来源：[src/langmem/knowledge/extraction.py:50-150]()
## 最佳实践
### 1. 限制操作权限
始终通过 `actions_permitted` 限制 AI 可执行的操作，防止意外删除重要记忆：
```python
manage_tool = create_manage_memory_tool(
namespace=("memories", "{user_id}"),
actions_permitted=["create", "update"], # 禁止删除
)
```
### 2. 使用 Schema 验证
通过 Pydantic schema 确保记忆数据结构的一致性：
```python
class Memory(BaseModel):
content: str
tags: list[str] = []
priority: int = 0
manage_tool = create_manage_memory_tool(
namespace=("memories", "{user_id}"),
schema=Memory,
)
```
### 3. 分离搜索和管理
在复杂应用中，建议分别创建搜索和管理工具：
```python
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` 参数提供工具使用指导：
```python
manage_tool = create_manage_memory_tool(
namespace=("preferences", "{user_id}"),
instructions="仅在用户明确表达偏好时创建记忆，避免过度记录。",
)
```
## 注意事项
1. **store 必须可用**：工具依赖 LangGraph 的 BaseStore，在 `@entrypoint` 或 `create_react_agent` 中使用时会自动注入。
2. **命名空间唯一性**：每个用户的记忆应使用独立的命名空间路径。
3. **异步支持**：工具同时支持同步（`invoke`）和异步（`ainvoke`）调用。
4. **Schema 兼容性**：自定义 schema 必须与已存储的记忆数据结构兼容，否则会导致验证失败。
5. **ID 必填规则**：更新和删除操作需要提供记忆的 UUID，AI 在调用工具时必须包含此参数。
资料来源：[src/langmem/knowledge/tools.py:180-220]()
## 导出清单
```python
__all__ = [
"create_manage_memory_tool",
"create_search_memory_tool",
]
```
资料来源：[src/langmem/knowledge/tools.py:350-355]()
---
## 记忆提取
### 相关页面
相关主题：[语义记忆管理](#page-semantic-memory), [情景记忆提取](#page-episodic-memory), [后台处理快速开始](#page-background-quickstart)
相关源码文件
以下源码文件用于生成本页说明：
- [src/langmem/knowledge/extraction.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/extraction.py)
- [src/langmem/knowledge/tools.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/tools.py)
- [src/langmem/prompts/types.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/types.py)
- [src/langmem/graphs/semantic.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/graphs/semantic.py)
- [src/langmem/short_term/summarization.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/short_term/summarization.py)
# 记忆提取
记忆提取（Memory Extraction）是 langmem 库中用于从对话或交互轨迹中识别、提取和存储有价值信息的核心功能模块。该模块通过大语言模型（LLM）分析对话内容，自动识别用户偏好、关键事实、上下文信息等结构化记忆，并支持持久化存储以便后续检索和使用。
## 核心概念
### 什么是记忆提取
记忆提取是一种自动化的信息处理流程，它从非结构化的对话数据中识别并提取结构化的记忆信息。这些记忆信息经过处理后可以用于：
- 个性化响应生成
- 上下文连续性维护
- 用户偏好跟踪
- 对话状态管理
资料来源：[src/langmem/knowledge/extraction.py:1-50]()
### 记忆类型
langmem 支持多种类型的记忆提取模式：
| 记忆类型 | 描述 | 适用场景 |
|---------|------|---------|
| 实体记忆 | 提取人名、地点、事件等实体信息 | 事实性信息存储 |
| 偏好记忆 | 记录用户偏好、习惯、偏好设置 | 个性化服务 |
| 上下文记忆 | 保存对话上下文和会话状态 | 多轮对话支持 |
| 关系记忆 | 记录实体之间的关系和关联 | 知识图谱构建 |
资料来源：[src/langmem/prompts/types.py:1-80]()
## 架构设计
### 整体架构
```mermaid
graph TD
A[对话输入] --> B[记忆管理器
MemoryManager]
B --> C{提取模式}
C -->|LLM分析| D[模式匹配]
C -->|向量搜索| E[语义检索]
D --> F[记忆存储]
E --> F
F --> G[结构化记忆]
G --> H[长期存储
MemoryStore]
G --> I[短期记忆
Summarization]
```
### 核心组件
| 组件 | 职责 | 文件位置 |
|------|------|---------|
| MemoryManager | 记忆提取的主控制器 | extraction.py |
| MemoryStoreManager | 支持向量存储的记忆管理 | extraction.py |
| SummarizationNode | 短期记忆压缩节点 | summarization.py |
| SemanticMemoryGraph | 语义记忆图结构 | semantic.py |
| create_manage_memory_tool | 记忆管理工具 | tools.py |
| create_search_memory_tool | 记忆搜索工具 | tools.py |
资料来源：[src/langmem/knowledge/extraction.py:100-200]()
## MemoryManager 详解
### 初始化配置
`create_memory_manager` 函数用于创建记忆管理器实例：
```python
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,
)
```
### 配置参数表
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| model | str \| BaseChatModel | 必填 | 用于记忆提取的 LLM |
| schemas | list[type[BaseModel]] | None | 记忆数据结构定义 |
| instructions | str | None | 提取指令提示 |
| enable_inserts | bool | True | 是否启用插入操作 |
| enable_updates | bool | True | 是否启用更新操作 |
| enable_deletes | bool | True | 是否启用删除操作 |
资料来源：[src/langmem/knowledge/extraction.py:200-300]()
### 调用方式
#### 异步调用
```python
conversation = [
{"role": "user", "content": "我偏好深色模式"},
{"role": "assistant", "content": "好的，已记住您的偏好"},
]
memories = await manager.ainvoke(
{"messages": conversation, "max_steps": 3}
)
```
#### 同步调用
```python
memories = manager.invoke({"messages": conversation})
```
资料来源：[src/langmem/knowledge/extraction.py:250-280]()
## MemoryStoreManager 详解
### 概述
`MemoryStoreManager` 是支持向量存储的增强版记忆管理器，集成了语义搜索能力：
```python
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 的区别
| 特性 | MemoryManager | MemoryStoreManager |
|------|---------------|-------------------|
| 存储后端 | 内存/外部 | 向量存储 |
| 语义搜索 | ✗ | ✓ |
| 查询优化 | 基础 | 高级（LLM生成查询） |
| 适用场景 | 快速原型 | 生产环境 |
资料来源：[src/langmem/knowledge/extraction.py:300-400]()
### 工作流程
```mermaid
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: 提取结果
```
### 搜索功能
```python
# 自然语言搜索
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 消耗：
```python
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,
)
```
### 配置参数
| 参数 | 类型 | 说明 |
|------|------|------|
| model | Runnable | 用于摘要生成的 LLM |
| max_tokens | int | 单次摘要的最大 token 数 |
| max_tokens_before_summary | int | 触发摘要前的最大 token 数 |
| max_summary_tokens | int | 摘要内容本身的最大 token 数 |
| input_messages_key | str | 输入消息的字段名，默认 "messages" |
| output_messages_key | str | 输出消息的字段名，默认 "messages" |
| snapshot_mode | SnapshotMode | 快照模式配置 |
资料来源：[src/langmem/short_term/summarization.py:1-100]()
### 状态更新格式
```json
{
"output_messages_key": [
"<更新的消息列表，包含摘要消息>"
],
"context": {
"running_summary": ""
}
}
```
## 语义记忆图
### SemanticMemoryGraph
`SemanticMemoryGraph` 提供基于语义图的记忆组织方式：
```mermaid
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 可用的记忆管理工具：
```python
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"],
)
```
### 参数说明
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| namespace | tuple[str, ...] | 是 | 记忆的命名空间路径 |
| schema | type[BaseModel] | 否 | 记忆的数据结构 |
| actions_permitted | list | 是 | 允许的操作类型 |
| store | BaseStore | 否 | 底层存储后端 |
资料来源：[src/langmem/knowledge/tools.py:1-100]()
### 记忆搜索工具
`create_search_memory_tool` 创建一个语义搜索工具：
```python
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}
)
```
## 使用示例
### 基础示例
```python
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)
```
### 带配置的完整示例
```python
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 中使用
```python
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
用于表示带反馈的对话轨迹：
```python
trajectory = AnnotatedTrajectory(
messages=[
{"role": "user", "content": "有什么好的披萨推荐？"},
{"role": "assistant", "content": "推荐尝试 LangPizza™️"},
{"role": "user", "content": "别给我打广告"},
{"role": "assistant", "content": "但你会喜欢的！"},
],
feedback={
"developer_feedback": "太激进了",
"score": 0,
},
)
```
### OptimizerInput
用于单提示优化场景的输入结构：
```python
input_data = OptimizerInput(
trajectories=[
AnnotatedTrajectory(
messages=[...],
feedback="应该使用搜索工具",
)
],
prompt=Prompt(
name="main_assistant",
prompt="你是一个有帮助的助手，有搜索工具可用。",
update_instructions="仅做最小更改，专注于修复错误的地方。",
),
)
```
资料来源：[src/langmem/prompts/types.py:50-120]()
## 配置最佳实践
### 命名空间配置
| 占位符 | 说明 | 示例 |
|--------|------|------|
| `{langgraph_user_id}` | 用户唯一标识 | memories/user123 |
| `{session_id}` | 会话标识 | memories/user123/session456 |
| `{custom_key}` | 自定义键 | memories/user123/custom |
### 性能优化建议
1. **批量处理**：将多个相关对话合并处理
2. **增量更新**：使用 `enable_updates=True` 避免重复存储
3. **搜索限制**：设置合理的 `query_limit` 减少检索开销
4. **摘要压缩**：对于长对话使用 `SummarizationNode` 减少 token
### 安全考虑
- 命名空间隔离：不同用户使用不同命名空间
- 操作权限：仅启用必要的 CRUD 操作
- Schema 验证：使用 Pydantic 模型确保数据一致性
## 相关模块
| 模块 | 文件路径 | 用途 |
|------|----------|------|
| prompts.types | prompts/types.py | 类型定义和数据结构 |
| prompts.optimization | prompts/optimization.py | 提示词优化集成 |
| prompts.prompt | prompts/prompt.py | 提示模板 |
| prompts.gradient | prompts/gradient.py | 梯度提示优化器 |
| short_term | short_term/summarization.py | 短期记忆压缩 |
## 总结
记忆提取是 langmem 提供的核心能力之一，它通过结构化的方式从对话中提取有价值的信息，并通过 MemoryManager 和 MemoryStoreManager 两个主要接口提供灵活的存储和检索能力。结合短期记忆压缩和语义搜索功能，langmem 能够构建完整的记忆系统，支持各种需要上下文持久化的 AI 应用场景。
---
## 提示优化
相关源码文件
以下源码文件用于生成本页说明：
- [src/langmem/prompts/optimization.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/optimization.py)
- [src/langmem/prompts/gradient.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/gradient.py)
- [src/langmem/prompts/metaprompt.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/metaprompt.py)
- [src/langmem/prompts/prompt.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/prompt.py)
- [src/langmem/prompts/types.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/types.py)
# 提示优化
## 概述
提示优化（Prompt Optimization）是 langmem 库中用于自动改进大型语言模型提示词的核心模块。该模块通过分析对话轨迹和反馈信息，自动调整和优化系统提示词，使 AI 代理能够更好地完成特定任务。
langmem 提供了三种主要的提示优化策略：
| 优化类型 | 说明 | 使用场景 |
|---------|------|---------|
| **Gradient（梯度优化）** | 基于假设和推荐进行渐进式调整 | 需要分析失败原因并针对性改进 |
| **Metaprompt（元提示优化）** | 通过元学习步骤进行深度优化 | 复杂任务需要多次反思 |
| **Prompt Memory（提示记忆优化）** | 从历史交互中学习成功模式 | 积累经验持续改进 |
## 核心组件
### 类型定义
提示优化模块使用强类型定义确保输入输出的准确性。
#### OptimizerInput
```python
class OptimizerInput(TypedDict):
"""单提示优化的输入数据结构"""
trajectories: typing.Sequence[AnnotatedTrajectory] | str
prompt: str | Prompt
```
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `trajectories` | `Sequence[AnnotatedTrajectory] \| str` | 是 | 对话轨迹列表或字符串描述 |
| `prompt` | `str \| Prompt` | 是 | 需要优化的提示词 |
#### MultiPromptOptimizerInput
```python
class MultiPromptOptimizerInput(TypedDict):
"""多提示优化的输入数据结构"""
trajectories: typing.Sequence[AnnotatedTrajectory] | str
prompts: list[Prompt]
```
| 字段 | 类型 | 必填 | 说明 |
|------|------|------|------|
| `trajectories` | `Sequence[AnnotatedTrajectory] \| str` | 是 | 对话轨迹列表 |
| `prompts` | `list[Prompt]` | 是 | 需要一起优化的多个提示 |
#### Prompt 结构
```python
class Prompt(TypedDict, total=False):
"""结构化提示管理"""
name: Required[str] # 提示名称
prompt: Required[str] # 提示内容
update_instructions: str | None # 更新指令
when_to_update: str | None # 更新时机
```
#### AnnotatedTrajectory 结构
```python
class AnnotatedTrajectory(typing.NamedTuple):
"""带标注的对话轨迹"""
messages: typing.Sequence[AnyMessage] # 消息列表
feedback: dict[str, str | int | bool] | str | None = None # 反馈信息
```
资料来源：[src/langmem/prompts/types.py:1-80](src/langmem/prompts/types.py)
## API 接口
### 创建单提示优化器
```python
def create_prompt_optimizer(
model: str | BaseChatModel,
kind: typing.Literal["gradient", "metaprompt", "prompt_memory"] = "gradient",
*,
config: typing.Optional[dict] = None,
) -> Runnable[OptimizerInput, str]
```
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `model` | `str \| BaseChatModel` | 必填 | 使用的语言模型 |
| `kind` | `Literal["gradient", "metaprompt", "prompt_memory"]` | `"gradient"` | 优化策略类型 |
| `config` | `dict \| None` | `None` | 优化器配置 |
### 创建多提示优化器
```python
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]]
```
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `model` | `str \| BaseChatModel` | 必填 | 使用的语言模型 |
| `kind` | `Literal["gradient", "metaprompt", "prompt_memory"]` | `"gradient"` | 优化策略类型 |
| `config` | `dict \| None` | `None` | 优化器配置 |
资料来源：[src/langmem/prompts/optimization.py](src/langmem/prompts/optimization.py)
## 优化策略详解
### 1. 梯度优化（Gradient Optimization）
梯度优化是最基本的优化策略，通过分析对话轨迹中的假设和推荐进行提示词调整。
```mermaid
graph TD
A[输入轨迹和反馈] --> B[生成假设]
B --> C[提出建议]
C --> D[应用最小侵入性更改]
D --> E[输出优化后的提示]
```
#### 工作流程
1. **假设生成**：分析当前提示在处理对话轨迹时的潜在问题
2. **建议生成**：基于假设提出具体的调整建议
3. **应用更改**：仅实施必要的最小更改
#### 配置选项
| 配置项 | 类型 | 默认值 | 说明 |
|--------|------|--------|------|
| `max_reflection_steps` | `int` | 3 | 最大反思步数 |
| `min_reflection_steps` | `int` | 1 | 最小反思步数 |
| `gradient_prompt` | `str` | 默认模板 | 梯度优化的提示模板 |
| `metaprompt` | `str` | 默认模板 | 元提示模板 |
资料来源：[src/langmem/prompts/gradient.py](src/langmem/prompts/gradient.py)
#### 使用示例
```python
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）
元提示优化通过多步元学习过程进行深度反思，适合处理复杂的优化任务。
```mermaid
graph TD
A[输入当前提示] --> B[元级反思]
B --> C{反思步数检查}
C -->|未达到最小步数| B
C -->|已达到最小步数| D[生成优化建议]
D --> E[应用最终优化]
E --> F[输出优化后的提示]
```
#### 配置选项
| 配置项 | 类型 | 默认值 | 说明 |
|--------|------|--------|------|
| `max_reflection_steps` | `int` | 3 | 最大元学习步数 |
| `min_reflection_steps` | `int` | 1 | 最小元学习步数 |
#### 使用示例
```python
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](src/langmem/prompts/metaprompt.py)
### 3. 提示记忆优化（Prompt Memory Optimization）
提示记忆优化从历史交互中学习成功模式，识别改进领域，并将学习到的模式应用到新提示中。
```mermaid
graph TD
A[历史轨迹数据] --> B[提取成功模式]
A --> C[识别改进领域]
B --> D[应用模式到新提示]
C --> D
D --> E[输出优化后的提示]
```
#### 核心特点
- **模式提取**：从过去的成功交互中提取可复用的模式
- **反馈分析**：识别反馈中指出的改进机会
- **知识迁移**：将学习到的知识应用到新的提示优化中
#### 使用示例
```python
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
用于单提示优化的反射模板：
```python
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}
Here was the agent's trajectory:
{trajectory}
Here is the user's feedback:
{feedback}
Here are instructions for updating the agent's prompt:
{instructions}
Based on this, return an updated prompt"""
```
### INSTRUCTION_REFLECTION_MULTIPLE_PROMPT
用于多提示优化的反射模板，支持批量处理多个轨迹：
```python
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}
Here are examples of various agent trajectories and associated feedback:
{data}
Here are instructions for updating the agent's prompt:
{instructions}
Based on this, return an updated prompt"""
```
### GeneralResponse 结构
优化器的返回响应结构：
```python
class GeneralResponse(TypedDict):
logic: str # 推理逻辑
update_prompt: bool # 是否需要更新
new_prompt: str # 新提示内容
```
资料来源：[src/langmem/prompts/prompt.py](src/langmem/prompts/prompt.py)
## 梯度元提示模板
梯度优化使用的默认元提示：
```python
DEFAULT_GRADIENT_METAPROMPT = """You are optimizing a prompt to handle its target task more effectively.
{current_prompt}
We hypothesize the current prompt underperforms for these reasons:
{hypotheses}
Based on these hypotheses, we recommend the following adjustments:
{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](src/langmem/prompts/gradient.py)
## GradientPromptOptimizer 类
基于类的梯度提示优化器实现，同时支持同步和异步调用：
```python
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 配置结构
```python
class GradientOptimizerConfig(TypedDict, total=False):
"""梯度优化器配置"""
gradient_prompt: str # 梯度优化提示模板
metaprompt: str # 元提示模板
max_reflection_steps: int # 最大反思步数
min_reflection_steps: int # 最小反思步数
```
### GradientOptimizerInput 输入结构
```python
class GradientOptimizerInput(TypedDict, total=False):
"""梯度优化器输入"""
trajectories: prompt_types.OptimizerInput | str
prompt: str | prompt_types.Prompt
```
## 最佳实践
### 1. 选择合适的优化策略
| 场景 | 推荐策略 |
|------|---------|
| 快速迭代 | Gradient（梯度优化） |
| 复杂任务 | Metaprompt（元提示优化） |
| 持续学习 | Prompt Memory（提示记忆优化） |
### 2. 配置反思步数
```python
# 简单任务 - 快速优化
config = {"max_reflection_steps": 1, "min_reflection_steps": 1}
# 复杂任务 - 深度优化
config = {"max_reflection_steps": 5, "min_reflection_steps": 2}
```
### 3. 提供有效反馈
反馈应该包含：
- 具体的问题描述
- 期望的行为
- 评分或等级（可选）
### 4. 多提示协同优化
当多个提示需要保持一致性时，使用 `create_multi_prompt_optimizer`：
```python
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：
```python
__all__ = [
"create_prompt_optimizer",
"create_multi_prompt_optimizer",
"Prompt",
"AnnotatedTrajectory",
"OptimizerInput",
"MultiPromptOptimizerInput",
]
```
资料来源：[src/langmem/prompts/optimization.py](src/langmem/prompts/optimization.py)
---
## 语义记忆管理
### 相关页面
相关主题：[情景记忆提取](#page-episodic-memory), [记忆提取](#page-memory-extraction)
相关源码文件
以下源码文件用于生成本页说明：
- [docs/docs/guides/extract_semantic_memories.md](https://github.com/langchain-ai/langmem/blob/main/docs/docs/guides/extract_semantic_memories.md)
- [src/langmem/graphs/semantic.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/graphs/semantic.py)
# 语义记忆管理
语义记忆管理是 LangMem 框架中用于从对话历史中自动提取、结构化和存储有意义信息的核心模块。它使 AI 系统能够跨会话保留和利用用户偏好、事实知识和上下文信息，从而提供更个性化和连贯的响应体验。
## 核心概念
### 什么是语义记忆
语义记忆是一种结构化的知识表示形式，它从原始对话内容中提取关键信息，并以预定义的数据模式（Schema）进行存储。与简单的文本存储不同，语义记忆具有以下特征：
| 特征 | 描述 |
|------|------|
| **结构化** | 使用 Pydantic 模型定义记忆的数据结构 |
| **可查询** | 支持向量相似性搜索和精确过滤 |
| **可演进** | 能够根据新信息自动更新现有记忆 |
| **上下文相关** | 与特定用户或会话命名空间关联 |
### 语义记忆 vs 其他记忆类型
| 类型 | 用途 | 持久性 | 典型实现 |
|------|------|--------|----------|
| 语义记忆 | 长期偏好和知识 | 持久化 | `create_memory_store_manager` |
| 短时记忆 | 当前会话上下文 | 会话级 | `SummarizationNode` |
| 程序记忆 | 提示优化学习 | 元数据 | `create_prompt_optimizer` |
资料来源：[src/langmem/knowledge/extraction.py:1-50]()
## 架构概述
### 系统组件
LangMem 的语义记忆管理由以下核心组件构成：
```mermaid
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
```
### 工作流程
```mermaid
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
### 创建语义记忆管理器
```python
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(),
)
```
### 参数说明
| 参数 | 类型 | 必需 | 默认值 | 说明 |
|------|------|------|--------|------|
| `model` | `str \| BaseChatModel` | 是 | - | 用于提取和综合的主模型 |
| `schemas` | `list[type[BaseModel]]` | 否 | - | 记忆数据的 Pydantic Schema |
| `query_model` | `str \| BaseChatModel` | 否 | - | 用于生成搜索查询的专用模型 |
| `query_limit` | `int` | 否 | 5 | 检索的记忆数量上限 |
| `namespace` | `tuple[str, ...]` | 否 | `("memories",)` | 记忆存储的命名空间 |
| `store` | `BaseStore` | 否 | - | LangGraph 底层存储 |
| `instructions` | `str` | 否 | - | 提取和更新指令 |
| `enable_inserts` | `bool` | 否 | `True` | 允许创建新记忆 |
| `enable_deletes` | `bool` | 否 | `True` | 允许删除旧记忆 |
资料来源：[src/langmem/knowledge/extraction.py:150-200]()
## Schema 定义
语义记忆使用 Pydantic 模型定义数据结构。以下是典型的 Schema 定义示例：
```python
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 配置参数
| 参数 | 说明 |
|------|------|
| `total=False` | 所有字段默认为可选 |
| 字段类型注解 | 定义记忆内容的结构和验证规则 |
| 默认值 | 为可选字段提供回退值 |
资料来源：[src/langmem/prompts/types.py:1-40]()
## 集成方式
### LangGraph 集成
LangMem 与 LangGraph 的 `@entrypoint` 装饰器无缝集成：
```python
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`：
```mermaid
graph LR
A[Agent] --> B[处理消息]
B --> C[返回响应]
B --> D[调度后台处理]
D --> E[ReflectionExecutor]
E --> F[记忆提取]
F --> G[存储更新]
```
```python
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]()
## 搜索和检索
### 默认搜索行为
默认情况下，记忆检索通过直接嵌入新消息进行相似度匹配：
```python
# 直接使用消息嵌入搜索
results = manager.search(config=config)
```
### LLM 优化查询搜索
使用专用查询模型生成更精确的搜索查询：
```python
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)
```
### 过滤和分页
```python
# 带过滤条件的搜索
results = await manager.asearch(
query="技术偏好",
filter={"category": "programming"},
limit=10,
offset=0,
)
```
### 搜索结果格式
| 字段 | 类型 | 说明 |
|------|------|------|
| `namespace` | `tuple[str, ...]` | 记忆所属命名空间 |
| `key` | `str` | 记忆唯一标识符 |
| `value` | `dict` | 记忆内容数据 |
| `created_at` | `datetime` | 创建时间戳 |
| `updated_at` | `datetime` | 更新时间戳 |
| `score` | `float \| None` | 相似度分数 |
资料来源：[src/langmem/knowledge/extraction.py:250-300]()
## 默认值机制
当没有找到相关记忆时，可以使用默认值工厂提供回退记忆：
```python
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,
)
```
### 默认值配置
| 配置方式 | 说明 |
|----------|------|
| `default` | 静态默认字符串或 Pydantic 实例 |
| `default_factory` | 动态生成默认值的函数 |
资料来源：[src/langmem/knowledge/extraction.py:300-350]()
## 操作模式
### 完整模式（默认值）
允许插入、更新和删除记忆：
```python
manager = create_memory_manager(
"anthropic:claude-3-5-sonnet-latest",
schemas=[PreferenceMemory],
enable_inserts=True,
enable_updates=True,
enable_deletes=True, # 完整功能
)
```
### 仅插入模式
仅创建新记忆，不修改现有记忆：
```python
manager = create_memory_manager(
"anthropic:claude-3-5-sonnet-latest",
schemas=[PreferenceMemory],
enable_inserts=True,
enable_updates=False,
enable_deletes=False, # 仅插入
)
```
### 操作控制参数
| 参数 | 允许的操作 | 典型用例 |
|------|-----------|----------|
| `enable_inserts=True` | 创建新记忆 | 首次学习用户偏好 |
| `enable_updates=True` | 更新现有记忆 | 偏好变化时同步更新 |
| `enable_deletes=True` | 删除过时记忆 | 清理矛盾或错误记忆 |
## 多步骤提取和综合
对于复杂对话场景，可以配置多步骤处理：
```python
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}
)
```
### 处理流程
```mermaid
graph TD
A[对话历史] --> B[步骤 1: 提取]
B --> C{是否完成?}
C -->|否| D[生成工具调用]
D --> E[步骤 2: 执行]
E --> B
C -->|是| F[返回记忆结果]
B --> G[保留未修改的现有记忆]
G --> F
```
## 命名空间管理
### 运行时命名空间配置
命名空间支持运行时配置变量替换：
```python
namespace=("memories", "{langgraph_user_id}")
# 配置
config = {"configurable": {"langgraph_user_id": "user123"}}
# 结果命名空间: ("memories", "user123")
```
### 命名空间结构
| 示例 | 说明 |
|------|------|
| `("memories",)` | 基础命名空间 |
| `("memories", "{user_id}")` | 用户级别隔离 |
| `("memories", "{team_id}", "{user_id}")` | 团队+用户层级 |
## 最佳实践
### 1. Schema 设计
- 为不同类型的记忆创建独立的 Schema
- 使用描述性的字段名称和类型注解
- 提供合理的默认值减少空值处理
### 2. 模型选择
| 场景 | 推荐模型配置 |
|------|-------------|
| 高质量提取 | 主模型使用 Sonnet-3.5，查询使用 Haiku |
| 成本优化 | 主模型和查询模型都使用轻量模型 |
| 快速迭代 | 仅使用主模型，不启用查询优化 |
### 3. 性能考虑
- 使用后台执行处理记忆管理，避免阻塞主流程
- 合理设置 `query_limit` 避免返回过多不相关记忆
- 对高频访问的记忆使用短期缓存
### 4. 错误处理
```python
try:
await manager.ainvoke({"messages": conversation})
except ValueError as e:
# Schema 验证失败
logger.error(f"记忆验证失败: {e}")
except Exception as e:
# 存储操作失败
logger.error(f"存储错误: {e}")
```
## 完整使用示例
```python
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]()
## 相关模块
| 模块 | 路径 | 功能 |
|------|------|------|
| 知识提取 | `src/langmem/knowledge/extraction.py` | 核心记忆管理逻辑 |
| 记忆工具 | `src/langmem/knowledge/tools.py` | LangGraph 工具集成 |
| 提示类型 | `src/langmem/prompts/types.py` | Schema 定义类型 |
| 短时摘要 | `src/langmem/short_term/summarization.py` | 会话级摘要管理 |
| 提示优化 | `src/langmem/prompts/optimization.py` | 元学习和优化 |
---
## 情景记忆提取
### 相关页面
相关主题：[语义记忆管理](#page-semantic-memory), [记忆提取](#page-memory-extraction)
相关源码文件
以下源码文件用于生成本页说明：
- [src/langmem/knowledge/extraction.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/extraction.py)
- [src/langmem/knowledge/tools.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/knowledge/tools.py)
- [src/langmem/prompts/optimization.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/optimization.py)
- [src/langmem/prompts/types.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/prompts/types.py)
- [src/langmem/short_term/summarization.py](https://github.com/langchain-ai/langmem/blob/main/src/langmem/short_term/summarization.py)
- [examples/standalone_examples/README.md](https://github.com/langchain-ai/langmem/blob/main/examples/standalone_examples/README.md)
# 情景记忆提取
## 概述
情景记忆提取（Episodic Memory Extraction）是 LangMem 库的核心功能之一，用于从对话历史中自动识别、提取和管理结构化记忆。该功能使 AI 系统能够持续学习用户偏好、对话模式和环境信息，并在后续交互中智能调用相关记忆来提升个性化服务质量。
LangMem 的情景记忆提取系统基于以下核心设计理念：
- **结构化存储**：记忆以预定义的 Pydantic Schema 形式存储，确保类型安全和数据一致性
- **增量更新**：支持记忆的创建、更新和删除操作，适应用户偏好的动态变化
- **智能检索**：提供基于向量相似度和 LLM 查询的混合检索能力
- **异步处理**：支持后台执行记忆处理，不阻塞主对话流程
资料来源：[src/langmem/knowledge/extraction.py:1-50]()
## 核心组件
### MemoryManager
`MemoryManager` 是情景记忆提取的主控制器，负责协调整个记忆生命周期。
```python
from langmem import create_memory_manager
manager = create_memory_manager(
"anthropic:claude-3-5-sonnet-latest",
schemas=[PreferenceMemory],
)
```
**关键参数说明**：
| 参数 | 类型 | 说明 |
|------|------|------|
| `model` | `str \| BaseChatModel` | 用于分析和提取记忆的主模型 |
| `schemas` | `list[type[BaseModel]]` | 记忆数据的 Pydantic Schema 定义 |
| `instructions` | `str \| None` | 自定义提取指令 |
| `enable_inserts` | `bool` | 启用创建新记忆（默认：True） |
| `enable_updates` | `bool` | 启用更新现有记忆（默认：True） |
| `enable_deletes` | `bool` | 启用删除过时记忆（默认：True） |
| `max_steps` | `int` | 最大提取和综合步数（默认：3） |
资料来源：[src/langmem/knowledge/extraction.py:100-150]()
### MemoryStoreManager
`MemoryStoreManager` 扩展了基础记忆管理能力，集成了 LangGraph 的持久化存储层。
```python
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]()
## 工作流程
### 基础提取流程
```mermaid
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 查询的检索流程
当需要更精确的记忆检索时，系统会先生成优化查询：
```mermaid
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 定义记忆结构：
```python
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 示例
```python
class Memory(BaseModel):
"""带类型标签的统一记忆格式"""
kind: str
content: dict[str, Any]
```
存储后的记忆项包含以下元数据：
| 字段 | 类型 | 说明 |
|------|------|------|
| `namespace` | `tuple[str, ...]` | 记忆的命名空间路径 |
| `key` | `str` | 记忆的唯一标识符 |
| `value` | `dict` | 序列化的记忆内容 |
| `created_at` | `datetime` | 创建时间戳 |
| `updated_at` | `datetime` | 最后更新时间戳 |
| `score` | `float \| None` | 相似度分数（检索时） |
资料来源：[src/langmem/knowledge/tools.py:50-100]()
## 使用方式
### 独立使用
在 LangGraph 外部独立使用记忆管理功能：
```python
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 应用中后台执行记忆处理：
```python
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]()
### 检索已存储记忆
```python
# 查看提取的记忆
print(manager.search(config=config))
# 自然语言搜索
results = await manager.asearch(
query="app preferences",
filter={"type": "PreferenceMemory"},
limit=5,
)
```
## 高级配置
### 仅插入模式
适用于只需要记录新信息、不修改现有记忆的场景：
```python
manager = create_memory_manager(
"anthropic:claude-3-5-sonnet-latest",
schemas=[PreferenceMemory],
enable_updates=False,
enable_deletes=False,
)
```
### 自定义提取步数
控制 LLM 推理和综合的迭代次数：
```python
# 设置最大步数
max_steps = 3
memories = await manager.ainvoke(
{"messages": conversation, "max_steps": max_steps}
)
```
### 默认记忆值
当没有找到记忆时返回预设默认值：
```python
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 的记忆管理工具：
```python
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
创建专门用于记忆检索的工具：
```python
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
创建独立的记忆搜索流水线：
```python
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]()
## 命名空间设计
### 命名空间结构
记忆按照层级化的命名空间组织：
```python
namespace = ("memories", "{langgraph_user_id}")
# 运行时解析为：("memories", "user123")
```
### 常见命名空间模式
| 用途 | 命名空间示例 |
|------|-------------|
| 用户级记忆 | `("memories", "{user_id}")` |
| 线程级记忆 | `("memories", "{user_id}", "threads", "{thread_id}")` |
| 类型隔离 | `("memories", "{user_id}", "preferences")` |
| 项目级记忆 | `("project_memories", "{project_id}")` |
### 多 Schema 命名空间
```python
memory_tool = create_manage_memory_tool(
namespace=("memories", "{langgraph_user_id}", "user_profile"),
schema=UserProfile,
)
```
资料来源：[src/langmem/knowledge/tools.py:200-300]()
## 异步与同步 API
### 异步调用
```python
# 异步提取
updated_memories = await manager.ainvoke(
{"messages": conversation, "existing": memories}
)
# 异步搜索
results = await manager.asearch(query="preferences", limit=10)
# 异步存储
await manager.aput("key", {"content": "value"})
```
### 同步调用
```python
# 同步提取
updated_memories = manager.invoke({"messages": conversation})
# 同步搜索
results = manager.search(query="preferences", limit=10)
```
## 与短时记忆的集成
LangMem 同时提供短时记忆（摘要）功能，可与情景记忆协同工作：
```python
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,
)
```
工作流程：
```mermaid
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. **使用命名空间隔离**：避免大规模检索时的性能瓶颈
### 错误处理
```python
try:
result = await manager.ainvoke({"messages": conversation})
except Exception as e:
logger.error(f"Memory extraction failed: {e}")
# 降级策略：使用空记忆继续
result = []
```
## 总结
情景记忆提取是 LangMem 库实现持久化上下文管理的核心能力。通过结构化的 Schema 定义、灵活的存储层集成和智能的提取算法，开发者可以轻松构建具有持续学习能力的 AI 应用。该系统既支持独立使用，也与 LangGraph 框架深度集成，满足从简单脚本到复杂生产环境的各种需求。
---
---
## Doramagic 踩坑日志
项目：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