Memory cn

Jianmu 的记忆系统由三大支柱构成：Checkpoint（状态快照） 确保运行时可恢复性，Long-Term Memory（长期记忆） 持久化跨会话的知识，Memory Tools（记忆工具） 则将这些能力暴露给 LLM Agent 直接调用。此外还有一套上下文构建体系，负责将记忆内容装配为模型可消费的 Prompt。本文聚焦于 Checkpoint 与长期记忆的设计、实现及协作模式。

架构分层：Checkpoint、长期记忆与上下文构建的关系

整个记忆模块位于 jianmu/memory/ 下，呈现清晰的三层子包划分，各层职责边界明确且互不侵入。

graph TB
    subgraph Agent Layer
        AGENT[LLM Agent]
    end

    subgraph "Memory Tools"
        ADD[MemoryAddTool]
        SEARCH[MemorySearchTool]
    end

    subgraph "Long-Term Memory"
        MS[MemoryService]
        MEM0[Mem0LongTermMemory]
        SIMPLE[SimpleStore]
    end

    subgraph "Checkpoint"
        FILE[FileCheckpointer]
        INMEM[InMemoryCheckpointer]
        NULL[NullCheckpointer]
    end

    subgraph "Context Building"
        CB[ContextBuilder]
        PROVIDERS[Message Providers]
        FILTERS[Context Filters]
    end

    subgraph "Runtime Engine"
        RR[ReactiveRunner]
        SM[StateManager]
    end

    AGENT -->|调用| ADD
    AGENT -->|调用| SEARCH
    ADD --> MS
    SEARCH --> MS
    MS --> MEM0
    MS --> SIMPLE
    RR -->|tick 级持久化| FILE
    RR -->|tick 级持久化| INMEM
    SM -->|dump_checkpoint| RR
    CB --> PROVIDERS
    CB --> FILTERS
    RR -->|注入上下文| CB

Checkpoint 层面向 ReactiveRunner，以 tick 为粒度保存行为树状态、用户状态和运行时元数据。Long-Term Memory 层面向 Agent，通过 MemoryService 外观统一 namespace 作用域下的 record / retrieve 操作。二者在运行时通过 StateManager.dump_checkpoint() 间接交汇——Checkpoint 负责"我在哪里"，长期记忆负责"我知道什么"。

Checkpoint 协议与 Runner 生命周期

CheckpointerProtocol：统一接口

CheckpointerProtocol 定义了所有 Checkpoint 后端的契约，仅两个方法：

方法	参数	职责
save_checkpoint	thread_id, step, state_dump, tree_state?, runtime_metadata?	持久化一个快照
get_checkpoint	thread_id	加载最新快照，无则返回 None

thread_id 是跨会话的稳定标识符，使得同一逻辑线程可以跨进程恢复。step 是单调递增的 tick 序号。state_dump 是 StateManager.dump_checkpoint() 产出的可序列化字典，tree_state 则映射 node_id→path 到 Status 名称（如 "RUNNING"）。

信封模式：序列化与验证

所有 Checkpoint 后端共享一套信封序列化工具。make_checkpoint_envelope() 将四个字段打包为统一字典格式，extract_checkpoint_payload() 读取时进行 schema 验证——必须同时包含 "step" 和 "state" 键才视为有效。

# 信封结构
{
    "step": int,          # 单调 tick 序号
    "state": dict,        # StateManager 的状态 dump
    "tree": dict | None,  # 行为树节点状态映射
    "runtime_metadata": dict  # 运行器内部元数据
}

这一设计实现了向前兼容：未来可在信封中添加字段而不破坏现有读取逻辑。测试中明确验证了非标准布局（如嵌套 payload 键或缺少 step/state）会被静默拒绝，返回 None。

三种实现对比

特性	FileCheckpointer	InMemoryCheckpointer	NullCheckpointer
持久化	JSONL 文件追加写入	纯内存字典	无操作
恢复策略	读取最后一行 JSON	deepcopy 返回	始终返回 None
线程安全	依赖文件系统	dict 天然线程安全	无状态
适用场景	本地应用、Demo、可恢复工作流	单元测试、临时会话	测试、短暂运行
默认路径	.outputs/jianmu/checkpoints/	N/A	N/A

FileCheckpointer 采用 JSONL 格式：每个 thread_id 对应一个 .jsonl 文件，每次 save_checkpoint 追加一行。get_checkpoint 仅读取最后一行，因此恢复的是最新快照。这种设计简单可靠，适合轻量级本地持久化场景。

InMemoryCheckpointer 在保存和读取时都通过 copy.deepcopy 创建副本，防止外部修改污染内部状态——这对测试环境的隔离性至关重要。

NullCheckpointer 实现了相同接口但所有方法体为空，用于无需持久化的场景（如一次性脚本）。如果 RuntimeCheckpointConfig.enabled 为 False（默认值），Runner 在未显式传入 Checkpointer 时不会自动创建，执行即为一次性模式。

Runner 中的 Checkpoint 生命周期

ReactiveRunner 在三种时机触发 Checkpoint 操作：启动恢复、周期性保存和即时保存。

sequenceDiagram
    participant User
    participant RR as ReactiveRunner
    participant CP as Checkpointer
    participant SM as StateManager
    participant Tree as BehaviourTree

    User->>RR: run(checkpointer, thread_id)
    RR->>CP: get_checkpoint(thread_id)
    alt checkpoint 存在
        CP-->>RR: envelope
        RR->>SM: initialize(state_dump)
        RR->>SM: restore_runtime_metadata(meta)
        RR->>Tree: _restore_tree_status(tree_state)
        RR->>Tree: _repair_running_composite_pointers()
        Note over RR: emit "checkpoint.restored"
    else 无 checkpoint
        RR->>SM: initialize(input_data)
        Note over RR: 全新启动
    end

    loop 每 tick
        RR->>Tree: tick_once()
        alt tick_count % interval == 0
            RR->>SM: dump_checkpoint()
            RR->>CP: save_checkpoint(...)
            Note over RR: emit "checkpoint.saved"
        end
        alt 挂起 (YIELD 模式)
            RR->>SM: dump_checkpoint()
            RR->>CP: save_checkpoint(...)
            Note over RR: 立即保存后返回
        end
    end

启动恢复 (_try_restore_from_checkpoint)：在 run() 入口处，如果传入了 checkpointer 且 restore_policy != NEVER，Runner 调用 checkpointer.get_checkpoint(thread_id)。若存在，则将 state_dump 写入 StateManager，runtime_metadata 恢复，tree_state 用于重建行为树各节点的状态。关键一步是 _repair_running_composite_pointers()——恢复 RUNNING 状态的 Composite 节点的 current_child 指针，确保下次 tick 从正确子节点继续。

周期性保存 (_maybe_save_checkpoint)：在主循环每个 tick 结束后检查 total_tick_count % checkpoint_interval == 0，满足条件时调用 StateManager.dump_checkpoint() 获取当前状态并持久化。

即时保存 (_save_checkpoint_now)：在挂起（SuspensionMode.YIELD）或审批确认刷新时立即保存，不等待周期性触发。

配置默认值

RuntimeCheckpointConfig 控制全局 Checkpoint 策略：

配置项	默认值	含义
enabled	False	是否在未显式传入 checkpointer 时自动创建
backend	"file"	默认后端类型
interval	1	tick 间隔（每个 tick 保存一次）

参考实现：project.py

长期记忆：协议、后端与 MemoryService 外观

LongTermMemoryProtocol：能力契约

LongTermMemoryProtocol 是一个 Python Protocol，不假定后端是键值存储，只声明三个能力方法：

方法	签名	语义
record	(inputs, *, namespace?, **kwargs) → LongTermMemoryRecordResult	将信息持久化到长期记忆
retrieve	(query, *, limit=5, namespace?, **kwargs) → LongTermMemoryRetrieved	语义检索相关记忆
format_retrieved	(retrieved, **kwargs) → str	将检索结果渲染为 LLM 可消费文本

namespace 使用 tuple[str, ...] 类型，如 ("agent", "thread")，提供多级隔离。输入类型 LongTermMemoryInput 接受 str | Message | Sequence[Message]，实现负责统一转换为文本。删除操作不是 Protocol 的核心方法——SimpleStore 将其作为可选扩展通过 getattr 鸭子类型检测。

Mem0LongTermMemory：语义记忆引擎

Mem0LongTermMemory 是 Mem0 开源库的适配层，支持两种模式：

本地模式（无 api_key）：

# 默认构造
memory = Mem0LongTermMemory()
# 带配置
memory = Mem0LongTermMemory(config={"vector_store": {"provider": "chroma", ...}})
# 带 infer 控制
memory = Mem0LongTermMemory(infer=False)  # 关闭语义提取

托管模式（传入 api_key）：

memory = Mem0LongTermMemory(api_key="m0-xxx...")

namespace 映射为 Mem0 的 user_id：("agent", "thread") → "agent/thread"。record() 方法将 Jianmu Message 序列转换为纯文本后调用 mem0.Memory.add()，并在 metadata 中注入 key。retrieve() 调用 mem0.Memory.search()，通过 filters={"user_id": ...} 实现命名空间隔离。

语义推理控制：infer=True（默认）时，Mem0 在 record 过程中执行事实提取与合并——写入的可能是经过语义整合的记忆而非原始文本。infer=False 则将原始文本直接存入向量存储。这在测试中明确验证：当 infer=False 时，client.add 的参数中 infer=False 被原样传递。

Mem0 结果归一化

Mem0 的 OSS 版本和托管版本返回结构不同，Mem0LongTermMemory 通过 _extract_results() 统一处理：若返回值是 dict 则取 "results" 键，若是 list 则直接使用。每个结果项通过 _normalize_item() 归一化为 Jianmu 标准格式：

{
    "id": str | None,         # Mem0 记忆 ID
    "content": str,           # memory 或 content 字段的文本
    "metadata": dict,         # 元数据
    "score": float | None,    # 可选相关性分数
}

format_retrieved() 将归一化列表渲染为带序号、ID、分数和元数据的可读文本，直接适合注入 Prompt。

SimpleStore：轻量级 JSON 文件后端

SimpleStore 是一个零依赖的本地长期记忆实现，适合 Demo 和不依赖语义检索的场景。核心特点：

• JSON 文件持久化：默认路径 .outputs/jianmu/store.json，内存中维护嵌套字典结构，写入时全量 json.dumps。
• 命名空间即嵌套键：("session", "prefs") 映射为 data["session"]["prefs"]。
• 关键词检索：retrieve() 采用两级策略——先尝试精确子串匹配，失败后回退到基于小写词元的交集评分。内置轻量词干化处理 ies→y、ing/ed/es/s 后缀剥离。
• 可选删除：delete() 通过 memory_id 从命名空间节点移除条目，不在 Protocol 中声明但可通过鸭子类型调用。

SimpleStore.record() 要求必须提供非空 key 参数，否则抛出 ValueError。每条记录以 key 为 ID，保存 {"content": ..., "metadata": ...} 结构。

MemoryService：命名空间作用域的外观

MemoryService 将任意 LongTermMemoryProtocol 后端绑定到固定命名空间，提供简化的 add/search/format_retrieved/delete 方法，并可通过 as_tools() 一键生成 MemoryAddTool + MemorySearchTool 工具对。

service = MemoryService(
    long_term_memory=SimpleStore(db_path="./memory.json"),
    namespace=("session",)
)

# 记录
service.add("用户喜欢 Python", key="pref1", metadata={"type": "language"})

# 检索
results = service.search("Python", limit=3)

# 生成工具
tools = service.as_tools()  # [MemorySearchTool, MemoryAddTool]

delete() 方法通过 getattr(self.memory, "delete", None) 检测后端是否支持删除——这是一种鸭子类型扩展模式，Mem0LongTermMemory 的删除操作可通过 Mem0 客户端原生方法实现，而 SimpleStore 显式实现了 delete()。

记忆工具：Agent 可直接调用的 Memory Tool

MemoryAddTool 和 MemorySearchTool 是连接 Agent 与长期记忆的桥梁。它们继承 Tool 基类，定义了标准的 name、description、input_schema 和 output_schema，使 LLM 能通过 function calling 自主管理记忆。

Tool 规格对比

属性	MemorySearchTool	MemoryAddTool
name	"memory_search"	"memory_add"
核心参数	query (必填), k, namespace	content (必填), metadata, key, namespace
输出类型	格式化文本或 "No results."	状态消息如 "Stored (id: mem_1, key: pref1)"
错误处理	异常时静默降级为 "No results."	异常时返回 "Error storing memory: {e}"

两个 Tool 的构造函数均接受 long_term_memory 或 memory 参数（二选一），内部自动包裹为 MemoryService。这意味着它们天然支持所有 LongTermMemoryProtocol 后端——无论是 Mem0LongTermMemory、SimpleStore 还是自定义实现。

MemoryAddTool.run() 中如果未提供 key，自动生成 UUID，确保每条记忆有唯一标识。

工厂函数与装配模式

create_memory_tools() 是对 MemoryService.as_tools() 的薄封装，确保向后兼容。二者等价：

# 方式 1：直接工厂
tools = create_memory_tools(
    namespace=("agent",),
    long_term_memory=mem0_memory
)

# 方式 2：通过 MemoryService
service = MemoryService(mem0_memory, namespace=("agent",))
tools = service.as_tools()

集成示例：Checkpoint + 长期记忆在运行时的协作

ReactiveRunner 的 run() 方法展示了 Checkpoint 与长期记忆如何协同工作。以下是一个典型的持久化场景：

from jianmu import ReactiveRunner, StateManager
from jianmu.memory import FileCheckpointer, Mem0LongTermMemory, MemoryService
from jianmu.memory.tools import create_memory_tools

# Checkpoint：状态可恢复性
checkpointer = FileCheckpointer(".outputs/jianmu/checkpoints")

# 长期记忆：跨会话知识
ltm = Mem0LongTermMemory(config={"vector_store": {"provider": "chroma"}})
memory_service = MemoryService(ltm, namespace=("my_agent",))
memory_tools = memory_service.as_tools()

# Runner 同时使用 checkpointer 和长期记忆工具
runner = ReactiveRunner(root=my_tree, state_manager=state_mgr)
await runner.run(
    checkpointer=checkpointer,
    thread_id="session-42",
    checkpoint_interval=1,
)

在这个场景中，Checkpoint 确保即使进程崩溃，Runner 也能从 thread_id="session-42" 的最后一个快照恢复，包括行为树中哪些节点处于 RUNNING 状态。长期记忆 则让 Agent 在下次会话中通过 memory_search 工具检索之前记录的偏好和知识——二者解决完全不同的问题但通过运行时的统一装配协同工作。

Checkpoint 事件与可观测性

ReactiveRunner 在 Checkpoint 操作时发射两类运行时事件：

事件类型	触发时机	Payload 关键字段
checkpoint.saved	每次成功持久化（周期或即时）	thread_id, step, forced
checkpoint.restored	启动时从快照恢复成功	thread_id, has_state, has_tree_state

这些事件通过 RuntimeEventBus 传递，可被遥测 Sink 捕获，为调试和执行审计提供可观测性。测试中明确验证了完整的事件生命周期：Runner 启动后 checkpoint.saved 出现，恢复运行后 checkpoint.restored 出现。

与上下文构建器的关系

Checkpoint 和长期记忆模块不直接参与 ContextBuilder 的 Prompt 装配流程——上下文构建在 上下文构建器：消息过滤、Token 预算控制与多源 Prompt 装配 中详细展开。但三者存在间接协作：

• Checkpoint 保存了 StateManager 中的消息历史，这些消息是 StateHistoryProvider 的数据源。
• 长期记忆工具 (MemoryAddTool / MemorySearchTool) 是 Agent 工具集的一部分，Agent 可在 ReAct 循环中主动调用它们读写记忆。
• 如果长期记忆检索结果需要注入 Prompt，通常由 ContextBuilder.for_react() 或 for_skill() 中的 ToolsDescProvider 提供工具描述，Agent 自行决定何时调用 memory_search。

继续阅读

本章聚焦 Checkpoint 与长期记忆的核心机制。以下页面与之密切相关：

• 上下文构建器：消息过滤、Token 预算控制与多源 Prompt 装配 — 理解记忆内容如何装配为模型上下文
• ReactiveRunner：事件驱动的异步 tick 调度与挂起恢复机制 — Checkpoint 的消费者，Runner 调度全景
• 类型化状态管理：Pydantic Schema、Reducer 合并与 Ephemeral 字段 — StateManager.dump_checkpoint() 的底层状态机制
• 工具抽象：Tool 基类、@tool 装饰器与 ToolSet 工具集装配 — Memory Tool 作为 Tool 体系的具体实例
• RAG 流水线：Embedder、Retriever、Reranker 与 KnowledgeBase — 与长期记忆互为补充的知识检索方案