25 KiB
Per-Thread Memory Brainstorm
Date: 2026-05-07
Background
Deerflow 现有的记忆功能是单租户的——不同会话都属于同一个用户,所有对话共享一份全局 memory.json。
要做一个新的记忆功能:不同对话属于不同用户,每个会话都有一个长期记忆,内容包括用户的使用习惯、个人信息、个人喜好和偏好语气。
现有记忆系统
- 存储:单一全局
backend/.deer-flow/memory.json,所有会话共享 - 认证:没有用户认证,没有用户隔离(better-auth 已搭建但未启用)
- 结构:
user: workContext / personalContext / topOfMindhistory: recentMonths / earlierContext / longTermBackgroundfacts[]: id, content, category, confidence, source
- 读路径:system prompt 生成时注入
<memory>...</memory>XML 标签 - 写路径:MemoryMiddleware 在对话后过滤消息 → MemoryUpdateQueue debounce 30s → MemoryUpdater 调 LLM 提取更新 → 原子写入
- 配置:
config.yaml > memory(enabled, debounce_seconds, max_facts, max_injection_tokens 等)
决策记录
存储方式: 数据库
文件存储 → 改为数据库表,通过 threads/{thread_id}/profile-memory.jsonthread_id 区分用户。
数据库: SQLite(本地/测试) + MySQL(生产环境)
表结构: 单表 + JSON 列(Option A)
依赖: 最小化,不引入 SQLAlchemy
SQLite 用标准库 sqlite3,MySQL 用 pymysql(纯 Python,轻量)。
与全局记忆关系: 策略 B(fallback)
Per-thread 有记忆就用 per-thread 的,没有就 fallback 到全局记忆。
首次对话: 不主动询问用户偏好
1. 数据库表设计
-- SQLite
CREATE TABLE IF NOT EXISTS thread_memory (
thread_id TEXT PRIMARY KEY,
profile TEXT NOT NULL DEFAULT '{}',
preferences TEXT NOT NULL DEFAULT '{}',
facts TEXT NOT NULL DEFAULT '[]',
last_updated TEXT NOT NULL DEFAULT (datetime('now'))
);
-- MySQL
CREATE TABLE IF NOT EXISTS thread_memory (
thread_id VARCHAR(64) PRIMARY KEY,
profile JSON NOT NULL,
preferences JSON NOT NULL,
facts JSON NOT NULL,
last_updated TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);
profile ({}):
| 字段 | 类型 | 说明 |
|---|---|---|
name |
string | null |
用户称呼 |
role |
string | null |
职业/角色 |
expertise |
string[] |
技术栈/专业领域 |
language |
"zh-CN" | "en-US" | null |
使用的语言 |
context |
string | null |
其他上下文(自由文本) |
preferences ({}):
| 字段 | 类型 | 说明 |
|---|---|---|
tone |
"casual" | "formal" | "technical" | "friendly" | null |
语气偏好 |
verbosity |
"concise" | "detailed" | null |
回答详细程度 |
codeStyle |
string | null |
代码风格偏好 |
other |
string | null |
其他偏好(自由文本) |
facts ([]):复用现有全局记忆的 fact 结构
{
"id": "fact_abc123",
"content": "用户在使用 React + TypeScript",
"category": "tech_stack | preference | personal | context | goal",
"confidence": 0.9,
"createdAt": "2026-05-07T...",
"source": "thread_id"
}
说明:三个 JSON 字段在 SQLite 中存为 TEXT(sqlite3 标准库没有原生 JSON 类型),在 MySQL 中存为 JSON。代码层面读写时做 json.dumps / json.loads,对上层透明。
2. config.yaml 新增配置段
thread_memory:
enabled: true
debounce_seconds: 30
model_name: null # null = 使用默认模型
max_facts: 100
fact_confidence_threshold: 0.7
injection_enabled: true
max_injection_tokens: 2000
database:
type: sqlite # sqlite | mysql
sqlite:
path: "thread_memory.db"
mysql:
host: "localhost"
port: 3306
user: "root"
password: "$MYSQL_PASSWORD"
database: "deerflow"
大部分字段和现有 memory 配置段语义相同,可以在两个配置段之间复用。database 段按 type 取子段,工厂函数只读自己需要的部分。
3. 存储层设计
3.1 抽象接口
# deerflow/agents/memory/thread_storage.py
import abc
import json
import sqlite3
from datetime import datetime
from typing import Any
class ThreadMemoryStorage(abc.ABC):
@abc.abstractmethod
def load(self, thread_id: str) -> dict[str, Any] | None:
"""加载指定 thread 的记忆,不存在返回 None。"""
...
@abc.abstractmethod
def save(self, thread_id: str, data: dict[str, Any]) -> bool:
"""保存指定 thread 的记忆(upsert)。"""
...
@abc.abstractmethod
def delete(self, thread_id: str) -> bool:
"""删除指定 thread 的记忆(thread 被删除时联动)。"""
...
def _create_empty_memory() -> dict[str, Any]:
"""Per-thread 记忆的初始空结构。"""
return {
"profile": {
"name": None,
"role": None,
"expertise": [],
"language": None,
"context": None,
},
"preferences": {
"tone": None,
"verbosity": None,
"codeStyle": None,
"other": None,
},
"facts": [],
}
def _row_to_memory(row: tuple) -> dict[str, Any]:
"""将数据库行转为 memory dict。SQLite 的 JSON 列存的是 TEXT,需要 parse。"""
return {
"threadId": row[0],
"profile": json.loads(row[1]),
"preferences": json.loads(row[2]),
"facts": json.loads(row[3]),
"lastUpdated": row[4],
}
3.2 SQLite 实现(本地测试)
class SqliteThreadMemoryStorage(ThreadMemoryStorage):
def __init__(self, db_path: str):
self._conn = sqlite3.connect(db_path)
self._conn.execute("""
CREATE TABLE IF NOT EXISTS thread_memory (
thread_id TEXT PRIMARY KEY,
profile TEXT NOT NULL DEFAULT '{}',
preferences TEXT NOT NULL DEFAULT '{}',
facts TEXT NOT NULL DEFAULT '[]',
last_updated TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
self._conn.commit()
def load(self, thread_id: str) -> dict | None:
row = self._conn.execute(
"SELECT thread_id, profile, preferences, facts, last_updated "
"FROM thread_memory WHERE thread_id = ?",
(thread_id,)
).fetchone()
return _row_to_memory(row) if row else None
def save(self, thread_id: str, data: dict) -> bool:
now = datetime.utcnow().isoformat() + "Z"
self._conn.execute("""
INSERT INTO thread_memory (thread_id, profile, preferences, facts, last_updated)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(thread_id) DO UPDATE SET
profile = excluded.profile,
preferences = excluded.preferences,
facts = excluded.facts,
last_updated = excluded.last_updated
""", (
thread_id,
json.dumps(data["profile"], ensure_ascii=False),
json.dumps(data["preferences"], ensure_ascii=False),
json.dumps(data["facts"], ensure_ascii=False),
now,
))
self._conn.commit()
return True
def delete(self, thread_id: str) -> bool:
self._conn.execute("DELETE FROM thread_memory WHERE thread_id = ?", (thread_id,))
self._conn.commit()
return True
3.3 MySQL 实现(生产环境)
class MysqlThreadMemoryStorage(ThreadMemoryStorage):
def __init__(self, host: str, port: int, user: str, password: str, database: str):
import pymysql
self._conn = pymysql.connect(
host=host, port=port, user=user, password=password, database=database,
charset="utf8mb4",
)
with self._conn.cursor() as cur:
cur.execute("""
CREATE TABLE IF NOT EXISTS thread_memory (
thread_id VARCHAR(64) PRIMARY KEY,
profile JSON NOT NULL,
preferences JSON NOT NULL,
facts JSON NOT NULL,
last_updated TIMESTAMP NOT NULL
DEFAULT CURRENT_TIMESTAMP
ON UPDATE CURRENT_TIMESTAMP
)
""")
self._conn.commit()
def load(self, thread_id: str) -> dict | None:
with self._conn.cursor() as cur:
cur.execute(
"SELECT thread_id, profile, preferences, facts, last_updated "
"FROM thread_memory WHERE thread_id = %s",
(thread_id,)
)
row = cur.fetchone()
return _row_to_memory(row) if row else None
def save(self, thread_id: str, data: dict) -> bool:
now = datetime.utcnow()
with self._conn.cursor() as cur:
cur.execute("""
INSERT INTO thread_memory (thread_id, profile, preferences, facts, last_updated)
VALUES (%s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
profile = VALUES(profile),
preferences = VALUES(preferences),
facts = VALUES(facts),
last_updated = VALUES(last_updated)
""", (
thread_id,
json.dumps(data["profile"], ensure_ascii=False),
json.dumps(data["preferences"], ensure_ascii=False),
json.dumps(data["facts"], ensure_ascii=False),
now,
))
self._conn.commit()
return True
def delete(self, thread_id: str) -> bool:
with self._conn.cursor() as cur:
cur.execute("DELETE FROM thread_memory WHERE thread_id = %s", (thread_id,))
self._conn.commit()
return True
3.4 工厂函数
def get_thread_memory_storage() -> ThreadMemoryStorage:
"""从 config 读取 database 配置,构建对应的 storage 实例(单例)。"""
config = get_thread_memory_config()
db = config.database
if db.type == "sqlite":
return SqliteThreadMemoryStorage(db.sqlite.path)
elif db.type == "mysql":
return MysqlThreadMemoryStorage(
host=db.mysql.host,
port=db.mysql.port,
user=db.mysql.user,
password=db.mysql.password,
database=db.mysql.database,
)
else:
raise ValueError(f"Unknown thread_memory database type: {db.type}")
3.5 注意事项
- JSON 在 SQLite 中存为 TEXT:
sqlite3标准库没有 JSON 类型,用 TEXT 存储json.dumps的结果。读写时做序列化/反序列化。MySQL 用原生 JSON 列,pymysql自动处理。 - upsert 语法差异:SQLite 用
ON CONFLICT ... DO UPDATE SET,MySQL 用ON DUPLICATE KEY UPDATE,语义等价。 - 连接管理:两个实现都在
__init__创建连接并持有。单线程场景没问题。如果将来需要并发,可以加连接池或改为每次操作创建连接。
4. upsert 语义:全量替换 vs 合并更新
两种模式
模式 A — 增量合并(LLM 出 delta,应用层合并):
LLM 输入: 现有记忆 + 新对话
LLM 输出: { profile: { name: "新值", shouldUpdate: true }, newFacts: [...], factsToRemove: [...] }
应用层: 读取现有记忆 → 按 delta 逐字段合并 → 写入
现有全局记忆用的就是这个模式。LLM 输出里带 shouldUpdate 标记和 factsToRemove 列表,应用代码做合并。
模式 B — 全量替换(LLM 出完整状态,应用层直接覆盖):
LLM 输入: 现有记忆 + 新对话
LLM 输出: { profile: { name: "...", role: "...", ... }, preferences: {...}, facts: [...] }
应用层: INSERT ... ON CONFLICT DO UPDATE(整行覆盖)
选择模式 B 的理由
-
profile 和 preferences 本身很小。每个对象 5-6 个字段,全部输出最多几十个 token,增量节省的 token 可以忽略。
-
去重和淘汰由 LLM 负责,应用层零逻辑。LLM 看到了完整的现有记忆,在 prompt 中就能决定哪些 facts 要保留、哪些过时了要删、哪些要合并。应用代码只需要
json.dumps+ upsert。 -
避免字段删除的尴尬。如果 LLM 想把
profile.context从"前端开发者"改成null(表示不再确定这个信息),增量模式需要额外表达"显式置 null"还是"不变",全量替换没有歧义。 -
和现有全局记忆的模式不同是合理的。全局记忆的
history有大量的对话摘要文本,不适合全量替换。Per-thread 记忆的 profile/preferences 是结构化的元数据,全量输出成本低。
具体流程
用户对话结束
↓
MemoryMiddleware.after_agent() 提取 user + final AI 消息
↓
queue.add(thread_id, messages) # debounce 30s
↓
ThreadMemoryUpdater.update()
1. 从 DB 读取现有记忆(不存在就用 _create_empty_memory())
2. 构建 prompt: "以下是用户的现有画像和偏好:{existing_memory},以下是新的对话:{conversation},请更新用户画像。"
3. LLM 返回完整的 profile + preferences + facts
4. storage.save(thread_id, data) # upsert 整行覆盖
关键点:LLM 的 prompt 里放了现有记忆,LLM 看到之后自己决定:
- 保留哪些 facts
- 更新哪些 profile 字段
- 新增什么偏好
- 删除过时的信息(不输出就是删除)
应用代码不做任何合并判断,只负责把 LLM 输出写入数据库。
5. 更新路径
5.1 MemoryMiddleware 改造(最小改动)
在现有 MemoryMiddleware.after_agent() 中加一段逻辑,当 thread_id 存在时,同时向 per-thread 记忆的 queue 推一条:
# 现有逻辑:全局记忆
queue = get_memory_queue()
queue.add(thread_id=thread_id, messages=filtered_messages, ...)
# 新增:per-thread 记忆
if thread_id:
thread_queue = get_thread_memory_queue()
thread_queue.add(thread_id=thread_id, messages=filtered_messages)
5.2 ThreadMemoryUpdater
新类,结构类似现有的 MemoryUpdater,但使用不同的 prompt 和存储后端:
class ThreadMemoryUpdater:
def update(self, messages, thread_id):
storage = get_thread_memory_storage()
existing = storage.load(thread_id) or _create_empty_memory()
prompt = THREAD_MEMORY_UPDATE_PROMPT.format(
existing_memory=json.dumps(existing, ensure_ascii=False),
conversation=format_conversation(messages),
)
response = model.invoke(prompt)
new_memory = parse_llm_output(response) # { profile, preferences, facts }
storage.save(thread_id, new_memory)
5.3 Prompt 设计要点
与全局记忆 prompt 的关键区别:
| 全局记忆 prompt | Per-thread 记忆 prompt | |
|---|---|---|
| 目标 | "对话中发生了什么" | "这个人是谁、喜欢什么" |
| 输出 | user context 摘要 + history 摘要 + facts | profile + preferences + facts |
| 侧重 | 保留对话内容的事实性信息 | 推断用户的身份、偏好、风格 |
| 语气影响 | 无 | 输出 preferences.tone 直接影响后续回复风格 |
6. 读取路径(注入 System Prompt)
def inject_thread_memory(system_prompt: str, thread_id: str) -> str:
storage = get_thread_memory_storage()
memory = storage.load(thread_id)
if memory is None:
# fallback 到全局记忆
return inject_global_memory(system_prompt)
# 生成 <memory profile="..."> 标签注入 system prompt
profile_xml = _format_profile_xml(memory)
return system_prompt + "\n" + profile_xml
注入内容的 XML 结构示例:
<memory>
<profile>
<name>张三</name>
<role>全栈工程师</role>
<expertise>React, TypeScript, Python</expertise>
<language>zh-CN</language>
<context>在做一个电商项目</context>
</profile>
<preferences>
<tone>casual</tone>
<verbosity>detailed</verbosity>
<codeStyle>prefers functional components with hooks</codeStyle>
</preferences>
</memory>
语气偏好(preferences.tone)不直接改 system prompt 模板,而是放在 <preferences> XML 里让 LLM 自己理解。方式简单,不用维护 prompt 模板的分支逻辑。如果发现 LLM 不遵循,再考虑动态改写 prompt 模板。
7. Thread 删除时的联动
Gateway 已有 DELETE /api/threads/{id}。在现有 handler 中加一行:
# app/gateway/routers/threads.py
@router.delete("/api/threads/{thread_id}")
async def delete_thread(thread_id: str):
# ... 现有清理逻辑 ...
# 新增:删除 per-thread 记忆
get_thread_memory_storage().delete(thread_id)
8. 实施步骤
- 新增配置模型 —
thread_memory_config.py(参考现有memory_config.py) - 新增存储层 —
thread_storage.py(ThreadMemoryStorage+SqliteThreadMemoryStorage+MysqlThreadMemoryStorage) - 新增 prompt —
thread_memory_prompt.py(用于 LLM 提取用户画像) - 新增 updater — 或扩展现有
MemoryUpdater,根据thread_id参数路由到不同逻辑 - 改造 middleware —
MemoryMiddleware中加 per-thread 记忆的 queue 逻辑 - 改造注入 — system prompt 生成时注入
<memory>标签 - 扩展 thread 删除 handler — 联动删除 DB 记录
- 写入测试 —
test_thread_memory_storage.py,test_thread_memory_updater.py
9. 待确认事项
- pymysql 作为新依赖是否 OK?
database配置段结构是否合适?- upsert 使用全量替换模式(模式 B)是否认同?
10. 第二轮脑暴(风险前置)
下面这轮不是改大方向,而是把容易在落地时踩坑的点先钉住。
10.1 隔离键:thread_id 是否足够?
当前设计用 thread_id 作为主键隔离用户记忆,简单可行。但有一个隐含前提:
- 一个 thread 永远只对应一个真实用户
如果未来支持“同一用户多 thread 共享画像”或“thread 可能转移 owner”,只用 thread_id 会限制扩展。
可选路径:
- 路径 A(维持现状,推荐短期):主键
thread_id,最快上线。 - 路径 B(兼容未来):增加
owner_id(可空),并加索引(owner_id, thread_id)。
建议:
- 第一版继续
thread_id,但在表里预留owner_idnullable 字段,避免后续大迁移。
10.2 并发一致性:同一 thread 的并发写覆盖问题
场景:同一 thread 在短时间内触发多次 update,后到达的旧结果可能覆盖先到达的新结果。
可选保护:
- 方案 A:
last_updated乐观锁(更新时带 where 条件) - 方案 B:
memory_version整数版本号(推荐) - 方案 C:严格串行队列(单 thread 单 worker)
建议:
- 加
memory_version(默认 0)。save时做 compare-and-swap 语义:- 读取 version = n
- 写入时要求 version 仍为 n,成功后 version = n+1
- 失败则重试一次(重新 load + merge prompt 再写)
这样不需要分布式锁,也能规避“旧结果回写”。
10.3 记忆质量控制:防止噪声和幻觉固化
LLM 抽取用户画像时,最大风险是把一次性表达当长期偏好。
建议加三道门:
- 事实类别阈值
preference类阈值可略低(如 0.7)personal类阈值更高(如 0.85)
- 稳定性规则
- 同类偏好至少被 2 次独立对话支持,才提升为 profile/preference 的强字段
- 冲突降级
- 新旧事实冲突时,不立刻删旧值
- 先把旧值降权并标记
supersededBy,下一轮再淘汰
10.4 隐私与合规:先定义“不能记”的边界
建议在 prompt 与代码都加 denylist(双保险):
- 默认不写入:身份证号、手机号、邮箱、住址、银行卡、密码/API Key 等敏感信息
- 允许写入:技术偏好、工作语境、沟通风格、项目目标
实现上:
- 在
ThreadMemoryUpdaterparse 后做一次 server-side scrub - 命中敏感模式就丢弃并打审计日志(不落库原文)
10.5 注入预算:避免 memory 挤爆上下文
当前有 max_injection_tokens,但还缺“裁剪策略”。
建议固定优先级:
- profile(最高)
- preferences
- facts(按 confidence + recency 排序后截断)
当超预算时:
- 永远保留 profile/preference
- 只裁剪 facts
10.6 可观测性:上线后如何判断有效
建议最小指标集:
thread_memory_update_total{status=ok|error}thread_memory_injection_tokensthread_memory_fact_countthread_memory_update_latency_msthread_memory_conflict_retry_total
加两条抽样日志:
- 更新前后摘要 diff(脱敏后)
- 注入片段长度与截断原因
10.7 迁移与回滚策略(从全局记忆过渡)
你已选 fallback 策略,这很好。建议再补两个机制:
-
冷启动导入(可选)
- 首次访问 thread 且无 per-thread 记录时,从全局记忆抽取一份“弱画像”写入
- 打
bootstrapped_from_global=true
-
一键回滚
- 配置开关
thread_memory.injection_enabled=false时,立刻只走全局注入 - 更新链路可继续跑,便于回滚期间保留数据
- 配置开关
10.8 API 语义建议(便于后续运维)
即使第一版 UI 不暴露,也建议预留内部接口:
GET /internal/thread-memory/{thread_id}(脱敏视图)DELETE /internal/thread-memory/{thread_id}POST /internal/thread-memory/{thread_id}/rebuild
这样排障时不用直接查库。
11. 第三轮决策清单(进入实现前最后拍板)
- 表结构是否预留
owner_id与memory_version? - 是否采用
memory_version方案处理并发覆盖? - 敏感信息 denylist 范围是否按 10.4 执行?
- 注入裁剪优先级是否固定为 profile > preferences > facts?
- 是否需要“冷启动导入”全局记忆到 per-thread?
- 是否要在首版就加内部运维接口?
如果以上 6 项确定,基本就能把实现风险压到可控范围内。
12. 默认拍板方案(建议直接采用)
目标:在不显著增加复杂度的前提下,拿到“可上线 + 可回滚 + 可演进”的第一版。
12.1 表结构默认值
采用:预留 owner_id + 引入 memory_version。
SQLite:
CREATE TABLE IF NOT EXISTS thread_memory (
thread_id TEXT PRIMARY KEY,
owner_id TEXT NULL,
profile TEXT NOT NULL DEFAULT '{}',
preferences TEXT NOT NULL DEFAULT '{}',
facts TEXT NOT NULL DEFAULT '[]',
memory_version INTEGER NOT NULL DEFAULT 0,
last_updated TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_thread_memory_owner_id ON thread_memory(owner_id);
MySQL:
CREATE TABLE IF NOT EXISTS thread_memory (
thread_id VARCHAR(64) PRIMARY KEY,
owner_id VARCHAR(64) NULL,
profile JSON NOT NULL,
preferences JSON NOT NULL,
facts JSON NOT NULL,
memory_version INT NOT NULL DEFAULT 0,
last_updated TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
INDEX idx_owner_id (owner_id)
);
12.2 并发一致性默认值
采用:memory_version 乐观并发控制 + 失败重试 1 次。
保存逻辑:
load()读出memory_version=nsave()时执行条件更新(WHERE thread_id=? AND memory_version=n)- 成功则
memory_version=n+1 - 如果受影响行数为 0,说明被并发写抢先,重读并重试一次
这能防止“旧更新覆盖新更新”,同时实现复杂度可控。
12.3 隐私策略默认值
采用:默认拒绝敏感信息入库(代码层 hard filter)。
默认 denylist:
- 手机号
- 邮箱
- 身份证号/护照号
- 银行卡号
- 密码/API Key/Token
- 详细住址
规则:
- 命中则从
profile/preferences/facts中删除该片段 - 仅记录脱敏审计信息(类型 + 时间 + thread_id),不记录原文
12.4 注入裁剪默认值
采用固定优先级:profile > preferences > facts。
当超过 max_injection_tokens:
- 必保留:
profile、preferences - 裁剪:
facts(按confidence DESC, createdAt DESC排序后截断)
这能保证人格与风格信息稳定注入,不被历史 facts 挤掉。
12.5 冷启动策略默认值
采用:首版不开启自动冷启动导入(bootstrap_from_global=false)。
理由:
- 降低“全局脏数据复制到 thread”风险
- 逻辑更清晰,便于观察 per-thread 记忆真实质量
补充:
- 保留 fallback(你当前已定)
- 后续若需要可加后台任务做可控回填
12.6 内部运维接口默认值
采用:首版只加读接口,写接口延后。
第一版建议:
GET /internal/thread-memory/{thread_id}(脱敏后返回)
暂不做:
DELETE /internal/thread-memory/{thread_id}(已有 thread delete 联动可覆盖主场景)POST /internal/thread-memory/{thread_id}/rebuild(二期再加)
这样可以先满足排障可见性,避免过早扩大运维面。
13. 实施前冻结版 Checklist(可直接转开发)
- DDL 按 12.1 落地(含
owner_id,memory_version, index) - Storage
save()改为 compare-and-swap 语义 - Updater 增加一次冲突重试
- parse 后执行敏感信息 scrub
- 注入模块按
profile > preferences > facts裁剪 - fallback 保持开启,冷启动导入保持关闭
- 增加最小指标与脱敏 diff 日志
- 增加内部只读排障接口
到这一步,方案已经可以进入实现,不需要再做大改。