6.记忆系统深度解析

记忆是智能系统的核心能力之一。对于 LangGraph 构建的 AI 应用来说，记忆系统决定了代理能否在多次交互中保持上下文、学习用户偏好，以及处理复杂的多步骤任务。本章将深入解析 LangGraph 的记忆机制，从底层的检查点持久化到高层的语义搜索，帮助构建真正"有记忆"的智能应用。

检查点持久化机制

检查点（Checkpoint）是 LangGraph 记忆系统的基石。本质上，检查点是图状态在特定时间点的快照，由检查点器（Checkpointer）在图的每个超级步骤（super-step）自动保存。这种设计不仅实现了对话记忆，还为人工介入、时间旅行和错误恢复提供了可能。

检查点的核心概念

当编译图时传入检查点器，LangGraph 会在每次节点执行后自动保存状态。检查点包含以下关键信息：

• values：状态通道的当前值，包括消息历史、变量等
• next：下一个待执行的节点名称
• config：与检查点关联的配置信息
• metadata：元数据，如时间戳、来源等
• tasks：待执行任务的信息

检查点通过线程（Thread）组织。每个线程代表一个独立的会话，通过 thread_id 标识。当使用相同的 thread_id 调用图时，LangGraph 会自动加载该线程的最新检查点，实现状态恢复。

线程与配置

线程是 LangGraph 中隔离不同会话的机制。调用图时，必须在 config 中指定 thread_id：

config = {"configurable": {"thread_id": "1"}}

这个简单的配置开启了持久化能力。第一次调用时，LangGraph 创建新线程；后续调用则恢复已有线程的状态。thread_id 可以是任意字符串，通常使用用户 ID、会话 ID 或其他业务标识。

检查点的工作流程

看一个具体例子。假设有一个简单的两节点图：

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from typing import Annotated
from typing_extensions import TypedDict
from operator import add

class State(TypedDict):
    foo: str
    bar: Annotated[list[str], add]

def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}

def node_b(state: State):
    return {"foo": "b", "bar": ["b"]}

workflow = StateGraph(State)
workflow.add_node(node_a)
workflow.add_node(node_b)
workflow.add_edge(START, "node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", END)

checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": ""}, config)

这段代码展示了检查点的基本用法。InMemorySaver 是内存型检查点器，适合开发和测试。生产环境应使用 PostgresSaver 或 SqliteSaver。

代码中，我们定义了一个包含 foo 和 bar 两个字段的状态。bar 使用 add 函数作为 Reducer，确保多次执行时数据能正确累加。编译图时传入 checkpointer，LangGraph 就会在每次节点执行后保存状态。

调用 graph.invoke 时，第二个参数是配置，包含 thread_id。执行完成后，可以通过 graph.get_state(config) 查看当前状态快照，或通过 graph.get_state_history(config) 获取历史检查点列表。

生产级检查点器

内存检查点器简单易用，但数据随进程退出而丢失。生产环境需要持久化存储：

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 首次使用需要初始化数据库
    # checkpointer.setup()
    
    graph = builder.compile(checkpointer=checkpointer)
    
    config = {"configurable": {"thread_id": "1"}}
    graph.invoke({"messages": [...]}, config)

PostgresSaver 将检查点存储在 PostgreSQL 数据库中，支持高并发和大数据量。首次使用时调用 setup() 方法创建必要的表结构。异步版本 AsyncPostgresSaver 支持异步图执行。

对于轻量级应用，SqliteSaver 是不错的选择：

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)

SQLite 文件型数据库无需独立服务，适合单机部署。

长期记忆存储系统

短期记忆解决了单一会话内的上下文问题，但跨会话的记忆需要长期记忆系统。LangGraph 通过 Store 接口实现长期记忆，允许在任意线程间共享数据。

Store 与 Checkpointer 的区别

Checkpointer 是线程隔离的，每个 thread_id 有独立的状态历史。Store 则是跨线程的，数据按命名空间（Namespace）组织，可以在不同会话间共享。

Store 适合存储：

• 用户画像和偏好
• 知识库和文档
• 应用级配置
• 跨会话的累积信息

Store 的基本使用

Store 使用键值对存储数据，支持层次化的命名空间：

from langgraph.store.memory import InMemoryStore

# 创建内存型 Store，生产环境使用 PostgresStore 或 RedisStore
store = InMemoryStore()

# 定义命名空间，通常包含用户 ID 和应用场景
user_id = "user_123"
namespace = (user_id, "memories")

# 存储记忆
store.put(
    namespace,
    "food_preference",
    {
        "text": "I love pizza and Italian cuisine",
        "tags": ["food", "preference"]
    }
)

# 检索记忆
items = store.search(namespace, query="food")

Store 的 put 方法接收三个参数：命名空间、键和值。命名空间是元组，可以包含多个层级，如 (user_id, "memories", "personal")。键是字符串，用于唯一标识该命名空间下的数据。值可以是任意 JSON 可序列化的数据结构。

search 方法支持多种检索方式。不指定查询时返回命名空间下所有数据；指定 query 时执行语义搜索（需配置嵌入模型）；使用 filter 参数可按字段精确过滤。

在图中集成 Store

Store 需要与 Checkpointer 一起使用，在编译图时传入：

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState, START

# 创建 Checkpointer 和 Store
checkpointer = InMemorySaver()
store = InMemoryStore()

# 定义使用 Store 的节点
def chat_node(state: MessagesState, config: dict, *, store: BaseStore):
    user_id = config["configurable"]["user_id"]
    namespace = (user_id, "memories")
    
    # 搜索相关记忆
    memories = store.search(
        namespace,
        query=state["messages"][-1].content,
        limit=3
    )
    
    # 将记忆加入系统提示
    memory_text = "\n".join([m.value["text"] for m in memories])
    system_msg = f"User memories:\n{memory_text}"
    
    # 调用模型
    response = model.invoke(
        [{"role": "system", "content": system_msg}] + state["messages"]
    )
    
    return {"messages": [response]}

# 编译图
builder = StateGraph(MessagesState)
builder.add_node(chat_node)
builder.add_edge(START, "chat_node")
graph = builder.compile(checkpointer=checkpointer, store=store)

# 调用图
config = {
    "configurable": {
        "thread_id": "1",
        "user_id": "user_123"
    }
}
graph.invoke({"messages": [{"role": "user", "content": "Hi"}]}, config)

关键是在节点函数签名中添加 store: BaseStore 参数，LangGraph 会自动注入 Store 实例。通过 config 获取 user_id，构建命名空间，实现用户级记忆隔离。

生产级 Store

与 Checkpointer 类似，Store 也有多种后端实现：

from langgraph.store.postgres import PostgresStore

DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresStore.from_conn_string(DB_URI) as store:
    # 首次使用需初始化
    # store.setup()
    
    graph = builder.compile(checkpointer=checkpointer, store=store)

PostgresStore 将数据存储在 PostgreSQL 中，支持全文搜索和语义搜索。RedisStore 提供高性能的内存缓存方案。选择哪种实现取决于数据规模、查询性能和运维复杂度。

语义搜索向量实现

传统的关键词搜索难以捕捉语义相似性。LangGraph Store 支持基于向量的语义搜索，通过嵌入模型将文本转换为高维向量，实现按"意义"检索。

配置语义搜索

启用语义搜索需要配置嵌入模型和向量维度：

from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

# 初始化嵌入模型
embeddings = init_embeddings("openai:text-embedding-3-small")

# 创建支持语义搜索的 Store
store = InMemoryStore(
    index={
        "embed": embeddings,  # 嵌入函数
        "dims": 1536,         # 向量维度，需与模型匹配
        "fields": ["text"]    # 要嵌入的字段
    }
)

fields 参数指定哪些字段需要生成嵌入向量。["$"] 表示嵌入整个文档，也可以指定特定字段如 ["text", "title"]。嵌入过程在 put 操作时自动完成，无需手动调用。

执行语义搜索

配置完成后，search 方法自动使用语义匹配：

# 存储一些记忆
store.put(("user_123", "memories"), "1", {"text": "I love pizza"})
store.put(("user_123", "memories"), "2", {"text": "I am a plumber"})
store.put(("user_123", "memories"), "3", {"text": "I enjoy hiking on weekends"})

# 语义搜索，查询与"I'm hungry"最相关的记忆
results = store.search(
    ("user_123", "memories"),
    query="I'm hungry",
    limit=1
)

# 返回结果按相似度排序
print(results[0].value["text"])  # 输出: I love pizza

虽然查询词"I'm hungry"与"I love pizza"没有共同词汇，但语义搜索能理解它们的关联性。这在处理用户模糊查询或同义表达时特别有用。

混合检索策略

Store 支持语义搜索与精确过滤的组合：

# 搜索关于食物的偏好，且标签包含"preference"
results = store.search(
    ("user_123", "memories"),
    query="food I like",
    filter={"tags": ["preference"]},
    limit=2
)

filter 参数使用精确匹配，先过滤再按语义相似度排序。这种混合策略既保证了召回率，又提高了准确率。

控制嵌入行为

有时需要精细控制哪些数据被嵌入：

# 仅嵌入特定字段
store.put(
    namespace,
    "memory_1",
    {
        "text": "Important information",
        "metadata": {"source": "chat", "timestamp": "2024-01-01"}
    },
    index=["text"]  # 只嵌入 text 字段
)

# 不嵌入，仅存储
store.put(
    namespace,
    "system_info",
    {"version": "1.0", "last_updated": "2024-01-01"},
    index=False
)

index 参数在 put 时指定，覆盖 Store 的默认配置。对于不需要搜索的系统信息，禁用嵌入可节省存储空间和计算资源。

工具内记忆读写操作

工具是代理与外部世界交互的接口。在工具中读写记忆，能让代理根据执行结果动态更新知识库，实现真正的自主学习。

在工具中读取短期记忆

短期记忆即图的状态，工具可以通过特殊参数访问：

from typing import Annotated
from langgraph.prebuilt import InjectedState, create_react_agent

class CustomState(MessagesState):
    user_name: str

def get_user_info(
    state: Annotated[CustomState, InjectedState]
) -> str:
    """查找用户信息"""
    user_id = state["user_id"]
    user_name = state.get("user_name", "Unknown")
    return f"User {user_id} is {user_name}"

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_user_info],
    state_schema=CustomState,
)

agent.invoke({
    "messages": [{"role": "user", "content": "look up user"}],
    "user_id": "user_123",
    "user_name": "Alice"
})

InjectedState 注解告诉 LangGraph 将当前状态注入工具函数。工具可以读取状态中的任何数据，包括消息历史、用户 ID、累积结果等。这在需要根据上下文调整行为时非常有用，比如根据用户等级返回不同详细程度的答案。

在工具中写入短期记忆

工具可以通过返回 Command 对象来更新状态：

from langgraph.types import Command
from langchain_core.messages import ToolMessage
from langchain_core.tools import InjectedToolCallId

@tool
def update_user_name(
    new_name: str,
    tool_call_id: Annotated[str, InjectedToolCallId],
    config: RunnableConfig
) -> Command:
    """更新用户名"""
    return Command(update={
        "user_name": new_name,
        "messages": [
            ToolMessage(
                content=f"Updated user name to {new_name}",
                tool_call_id=tool_call_id
            )
        ]
    })

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[update_user_name],
    state_schema=CustomState,
)

agent.invoke({
    "messages": [{"role": "user", "content": "My name is Bob"}],
    "user_id": "user_123"
})

Command 允许工具精确控制状态更新。update 字典中的字段会被合并到状态中，messages 列表用于更新消息历史。这种方式让工具能持久化执行结果，供后续节点使用。

在工具中读写长期记忆

长期记忆的读写通过 Store 实现。工具可以通过 get_store 函数或配置对象访问 Store：

from langgraph.config import get_store
from langchain_core.runnables import RunnableConfig

@tool
def save_user_preference(
    preference: str,
    config: RunnableConfig
) -> str:
    """保存用户偏好"""
    store = get_store()
    user_id = config["configurable"]["user_id"]
    
    # 读取现有偏好
    namespace = (user_id, "preferences")
    existing = store.get(namespace, "food")
    if existing:
        # 更新现有记录
        data = existing.value
        data["items"].append(preference)
        store.put(namespace, "food", data)
    else:
        # 创建新记录
        store.put(namespace, "food", {
            "category": "food",
            "items": [preference]
        })
    
    return "Preference saved"

@tool
def get_user_preferences(
    category: str,
    config: RunnableConfig
) -> str:
    """获取用户偏好"""
    store = get_store()
    user_id = config["configurable"]["user_id"]
    
    # 搜索相关偏好
    namespace = (user_id, "preferences")
    results = store.search(namespace, query=category, limit=5)
    
    preferences = [r.value for r in results]
    return str(preferences)

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[save_user_preference, get_user_preferences],
    store=store  # 在创建代理时传入 Store
)

工具通过 get_store() 获取 Store 实例，然后使用 put、get、search 等方法操作记忆。这种方式将记忆管理封装在工具内部，代理只需调用工具即可，无需关心实现细节。

记忆更新的最佳实践

在工具中更新记忆时，需要考虑并发和一致性：

@tool
def add_to_cart(
    item_id: str,
    quantity: int,
    config: RunnableConfig
) -> Command:
    """添加商品到购物车"""
    store = get_store()
    user_id = config["configurable"]["user_id"]
    namespace = (user_id, "cart")
    
    # 使用唯一键避免冲突
    cart_key = f"cart_{int(time.time())}"
    
    # 读取当前购物车
    current_cart = store.get(namespace, "current") or {"items": []}
    
    # 更新数据
    current_cart["items"].append({
        "item_id": item_id,
        "quantity": quantity,
        "added_at": time.time()
    })
    
    # 保存更新
    store.put(namespace, "current", current_cart)
    
    return Command(update={
        "cart_summary": f"Added {quantity} of {item_id}",
        "messages": [ToolMessage(content="Item added to cart", tool_call_id=tool_call_id)]
    })

使用带时间戳的键或版本号可以避免并发写入冲突。对于关键数据，考虑使用数据库的事务机制。Store 的 put 操作通常是原子的，但复杂的读-改-写序列需要额外注意。

总结

LangGraph 的记忆系统通过检查点和 Store 提供了短期和长期记忆能力。检查点自动保存图状态，实现会话内的上下文保持；Store 提供跨会话的数据存储，支持语义搜索和精确过滤。在工具中读写记忆，让代理能动态更新知识库，实现持续学习。

记忆系统的设计需要考虑数据隔离、并发控制和检索效率。合理使用命名空间组织数据，选择合适的后端存储，配置语义搜索提升召回率，这些都是构建生产级记忆系统的关键。

下一章将探讨多代理架构设计，看看如何利用记忆系统构建协作的代理团队，实现更复杂的任务分解和协调。