DeerFlow 记忆系统详解

概述

记忆系统负责从对话中提取用户偏好、习惯、背景信息，跨会话持久化，并在未来的对话中注入系统提示词。它通过去抖排队 + LLM 提取 + 文件存储三重机制，在不阻塞主对话的前提下，渐进式构建用户画像。

对话完成
  │
  ├── MemoryMiddleware.after_agent → 过滤消息 → queue.add() (30秒去抖)
  │
  └── SummarizationMiddleware 摘要前 → memory_flush_hook → queue.add_nowait() (立即)
         │
         ▼
  MemoryUpdateQueue → 去抖定时器触发 → 批量处理
         │
         ▼
  MemoryUpdater → LLM 分析对话 → 提取事实 → 保存到 memory.json
         │
         ▼
  下次对话 → DynamicContextMiddleware → format_memory_for_injection
         → 注入 system prompt

一、架构全景

┌─────────────────────────────────────────────────────────────────┐
│ MemoryMiddleware (middlewares/memory_middleware.py)              │
│   after_agent 钩子：过滤消息 → 信号检测 → 入队                    │
├─────────────────────────────────────────────────────────────────┤
│ MemoryUpdateQueue (memory/queue.py)                              │
│   去抖排队：30秒内连续对话合并处理，同 key 替换旧条目              │
├─────────────────────────────────────────────────────────────────┤
│ MemoryUpdater (memory/updater.py)                                │
│   核心更新器：构建 prompt → LLM 调用 → 解析 → 应用 → 保存        │
├─────────────────────────────────────────────────────────────────┤
│ MemoryStorage (memory/storage.py)                                │
│   存储层：原子写入 JSON 文件，mtime 缓存                          │
├─────────────────────────────────────────────────────────────────┤
│ prompt.py                                                        │
│   提示词模板 + memory 注入格式化（token 预算控制）                 │
├─────────────────────────────────────────────────────────────────┤
│ message_processing.py                                            │
│   消息过滤 + 纠错/强化信号检测                                     │
├─────────────────────────────────────────────────────────────────┤
│ summarization_hook.py                                            │
│   摘要钩子：在消息被删除前立即刷入记忆                             │
└─────────────────────────────────────────────────────────────────┘

二、触发路径（双重入口）

记忆更新有两条独立的触发路径，确保消息无论是否被摘要都不会丢失：

路径 1：MemoryMiddleware.after_agent（正常路径）

对话结束 → after_agent 钩子
  │
  ├── 过滤 messages → filter_messages_for_memory()
  ├── 检测纠错/强化信号 → detect_correction/reinforcement()
  ├── 捕获 user_id → get_effective_user_id()
  └── 入队 → queue.add(thread_id, messages, agent_name, user_id, ...)
       └── 重置 30 秒去抖定时器

路径 2：memory_flush_hook（紧急路径）

SummarizationMiddleware 即将删除消息
  │
  └── before_summarization hooks → memory_flush_hook()
       └── queue.add_nowait()  ← 延迟设为 0，立即处理

为什么需要路径 2：如果消息在 after_agent 入队后、去抖定时器触发前被 SummarizationMiddleware 删除，MemoryUpdater 处理时会发现消息内容为空（因为已被 RemoveMessage 标记删除）。路径 2 在删除前立即触发处理。

三、MemoryMiddleware 详解

3.1 after_agent 钩子

defafter_agent(self, state, runtime) -> dict | None:
# 1. 检查配置
ifnotself._config.enabled:
returnNone

# 2. 获取 thread_id
    thread_id = runtime.context.get("thread_id") or ...

# 3. 获取并过滤消息
    messages = state.get("messages") or []
    filtered = filter_messages_for_memory(messages)

# 4. 验证至少 1 条 user + 1 条 assistant
ifnot has_minimum_dialogue(filtered):
returnNone

# 5. 信号检测
    correction = detect_correction(filtered)
    reinforcement = detect_reinforcement(filtered) ifnot correction elseFalse

# 6. 捕获 user_id（threading.Timer 不传播 ContextVar）
    user_id = get_effective_user_id()

# 7. 入队
    queue.add(thread_id, filtered, self._agent_name, user_id,
              correction_detected=correction,
              reinforcement_detected=reinforcement)

returnNone# 不修改 state

3.2 消息过滤规则

filter_messages_for_memory() 决定哪些消息进入记忆：

设计要点：只保留"用户说了什么"和"助手最终回复了什么"，中间的思考过程、工具调用、内部消息全部过滤。

3.3 信号检测

# 纠错信号（11 种中英文模式）
"不对", "你理解错了", "重试", "换一种方式"
"that's wrong", "you misunderstood", "try again"

# 正向强化信号（17 种中英文模式）
"对，就是这样", "完全正确", "这个很好", "继续保持"
"yes, exactly", "perfect", "that's right", "keep doing that"

检测范围：最近 6 轮对话。信号会影响后续 MemoryUpdater 的行为——纠错信号让 LLM 更积极地删除或修改旧事实，强化信号让 LLM 更确信地保留。

四、去抖队列（MemoryUpdateQueue）

4.1 数据结构

classConversationContext:
    thread_id: str
    messages: list
    agent_name: str | None
    user_id: str | None
    correction_detected: bool
    reinforcement_detected: bool

classMemoryUpdateQueue:
    _queue: dict[tuple, ConversationContext]  # key = (thread_id, user_id, agent_name)
    _timer: threading.Timer | None
    _processing: bool# 门锁，防并发处理
    debounce_seconds: int = 30

4.2 去抖机制时序

对话 A 结束 → queue.add() → 定时器启动(30s)
  │
  ├─ 10秒后，对话 B 结束 → queue.add() → 定时器重置(30s)
  │
  ├─ 15秒后，对话 C 结束 → queue.add() → 定时器重置(30s)
  │   同 key 的对话 A 被替换（只保留最新）
  │
  └─ 45秒后（距对话 C 完成 30 秒），定时器触发
      → _process_queue() → 批量处理对话 B 和 C

合并规则：同一 (thread_id, user_id, agent_name) 的新对话会替换旧条目，correction_detected 做 OR 合并。

4.3 线程安全

• • 锁：threading.Lock 保护队列读写
• • 门锁：_processing 标志防止并发执行 _process_queue()
• • 网关保护：若定时器触发时 _processing=True，重新调度 0 秒延迟
• • 守护线程：threading.Timer 使用 daemon 线程，进程退出不阻塞

五、MemoryUpdater — 核心更新器

5.1 更新流程

classMemoryUpdater:
defupdate_memory(self, messages, thread_id, agent_name, ...):
# 1. 构建 prompt
        current_memory = storage.load(user_id, agent_name)
        prompt = self._prepare_update_prompt(current_memory, messages, ...)

# 2. 调用 LLM
        response = model.invoke(prompt)  # 同步调用，避免 httpx 连接池冲突

# 3. 解析 LLM 响应
        update_data = json.loads(response.content)

# 4. 应用更新
        updated = self._apply_updates(current_memory, update_data, thread_id)

# 5. 清理上传提及
        updated = _strip_upload_mentions_from_memory(updated)

# 6. 保存
        storage.save(user_id, agent_name, updated)

5.2 LLM Prompt 结构

MEMORY_UPDATE_PROMPT 包含以下部分：

1. 当前记忆状态（JSON 格式）
   - user 段：工作上下文、个人上下文、最近关注
   - history 段：近期月份、早期上下文、长期背景
   - facts 段：结构化事实列表

2. 本次对话内容（过滤后的消息）
   - 对话格式化文本

3. 更新指令
   - 纠错模式 vs 常规模式
   - 添加/修改/删除/标记为不再相关
   - 置信度评分（0.0-1.0）
   - 事实去重检查

5.3 _apply_updates — 事实管理

def_apply_updates(self, current_memory, update_data, thread_id):
# 1. 更新 user 段（合并新信息）
# 2. 更新 history 段
# 3. 移除标记为 "remove" 的 facts
# 4. 添加新 facts
#    - content 去重（casefold 比较）
#    - 置信度门槛（>= min_confidence）
#    - 强制 max_facts 限制
# 5. 标记新 facts 的 source_thread_id

facts 结构：

{
"id":"uuid",
"content":"用户偏好简洁沟通",
"category":"preference",
"confidence":0.85,
"source_thread_id":"thread_abc123",
"created_at":"2026-06-30T12:00:00Z"
}

5.4 同步调用设计

MemoryUpdater 使用同步 model.invoke() 而非异步版本。原因是主 agent 的异步 httpx 连接池与 Memory 的异步调用可能产生冲突（issue #2615）。同步调用通过 ThreadPoolExecutor 隔离。

六、MemoryStorage — 存储层

6.1 FileMemoryStorage

classFileMemoryStorage(MemoryStorage):
    _memory_cache: dict[tuple, tuple[dict, float]]  # (user_id, agent_name) → (data, mtime)

defload(self, user_id, agent_name):
# 1. 检查缓存：mtime 是否匹配
# 2. 缓存命中 → 直接返回
# 3. 缓存未命中 → 读文件 → 更新缓存

defsave(self, user_id, agent_name, data):
# 1. 写临时文件 memory.{uuid}.tmp
# 2. Path.replace() 原子替换
# 3. 更新缓存 mtime

6.2 文件路径隔离

agent_name 的作用贯穿整个链路：

1. 1. 存储隔离：不同 agent 有独立的 memory 文件
2. 2. 队列去重：queue_key = (thread_id, user_id, agent_name)
3. 3. 事实归属：每个 agent 维护独立的 facts 列表
4. 4. 注入隔离：不同 agent 看到不同的用户画像

6.3 安全

• • agent_name 必须匹配 AGENT_NAME_PATTERN，防止路径穿越
• • 原子写入保证不出现半截 JSON 文件

七、记忆注入

7.1 format_memory_for_injection()

将存储的记忆格式化后注入系统提示词：

defformat_memory_for_injection(memory_data, max_tokens=2000, ...):
# Phase 1：格式化 User Context 段
    user_context = format_user_context(memory_data["user"])

# Phase 2：格式化 History 段
    history = format_history(memory_data["history"])

# Phase 3：事实分两阶段处理
#   Guaranteed 池（默认 "correction" 类别）：
#     独立 500 token 预算，优先放入
#   Regular 池（其他类别）：
#     在 max_tokens 剩余预算内按置信度降序选择

# Phase 4：Token 溢出保护
#   Facts 块作为受保护后缀，只截断前面的 user/history 段

return formatted_memory

7.2 保底类别机制

correction 类别的事实有独立的 token 预算，即使超出总预算也会被保留。这确保了用户纠错信息永不丢失。

guaranteed_pool = [f for f in facts if f.category in guaranteed_categories]  # 默认 ["correction"]
regular_pool = [f for f in facts if f.category notin guaranteed_categories]

# 先在 guaranteed 预算内选满
# 再在 regular 预算内按置信度降序选

7.3 注入位置

DynamicContextMiddleware.before_agent 调用 format_memory_for_injection() 并注入：

<system-reminder>
当前日期：2026-06-30
</system-reminder>

<user_memory>
用户偏好：简洁沟通，不喜欢废话
项目背景：视频创作，使用 MoviePy/PIL
最近关注：DeerFlow 架构分析
...
</user_memory>

User: 帮我分析一下这个代码

八、上传事件防泄漏

多层防御确保文件上传事件不污染长期记忆：

# 第 3 层的正则模式
_UPLOAD_MENTION_PATTERNS = [
r"(uploaded a?n? file)|(uploaded a?n? image)",
r"/mnt/user-data/uploads/\S+",
    ...
]

九、配置

# config.yaml
memory:
enabled:true# 总开关
debounce_seconds:30# 去抖延迟（默认 30 秒）
max_facts:200# 每个 agent 最大 fact 数量
model_name:gpt-4o-mini# 记忆更新用的 LLM（轻量模型即可）
token_counting:true# 是否用 tiktoken 精确计数
injected_memory_max_tokens:2000# 注入记忆的最大 token 数
guaranteed_categories:# 保底类别
-correction
storage:
use:deerflow.agents.memory.storage:FileMemoryStorage

# MemoryConfig (config/memory_config.py)
classMemoryConfig(BaseModel):
    enabled: bool = False
    debounce_seconds: int = 30
    max_facts: int = 200
    model_name: str | None = None
    token_counting: bool = True
    injected_memory_max_tokens: int = 2000
    guaranteed_categories: list[str] = Field(default_factory=lambda: ["correction"])

十、完整数据流

用户对话完成
  │
  ▼
MemoryMiddleware.after_agent
  │
  ├─ filter_messages_for_memory()
  │   └─ 保留: user 消息 + 最终 assistant 回复
  │   └─ 丢弃: 中间步骤、隐藏消息、纯上传消息
  │
  ├─ detect_correction() / detect_reinforcement()
  │   └─ 中英文正则匹配最近 6 轮
  │
  └─ queue.add(thread_id, messages, agent_name, user_id, signals)
      └─ _enqueue_locked: 同 key 替换旧条目
      └─ _reset_timer: 30 秒去抖

───────────────────────────────────────
  SummarizationMiddleware 触发 → memory_flush_hook
  └─ queue.add_nowait(thread_id, messages, ...)  ← 延迟 0 秒，立即处理
───────────────────────────────────────

定时器触发 (_process_queue)
  │
  ├─ 遍历队列中的 ConversationContext
  │
  └─ 每个 context → MemoryUpdater.update_memory()
      │
      ├─ _prepare_update_prompt(current_memory, messages, signals)
      │   └─ 格式化记忆 + 对话 → LLM prompt
      │
      ├─ model.invoke(prompt)  ← 同步调用，线程池隔离
      │   └─ LLM 提取新事实、标记过时事实
      │
      ├─ _parse_memory_update_response(response)
      │   └─ 解析 JSON → update_data
      │
      ├─ _apply_updates(current_memory, update_data)
      │   └─ 去重 (casefold) → 置信度门槛 → 最大数量限制
      │
      ├─ _strip_upload_mentions_from_memory(updated)
      │
      └─ storage.save(user_id, agent_name, updated)
          └─ 原子写入 memory.json
───────────────────────────────────────

下次对话
  │
  ▼
DynamicContextMiddleware.before_agent
  │
  └─ format_memory_for_injection(storage.load(user_id, agent_name))
      └─ User Context + History + Facts (按置信度 + token 预算)
      └─ 注入为 system prompt 中的 <user_memory> 段

十一、设计哲学