199 lines
13 KiB
Markdown
199 lines
13 KiB
Markdown
# 创建线程延迟优化与业务库 / Checkpoint 分离
|
||
|
||
日期：2026-10-22。
|
||
|
||
## 本次实现
|
||
|
||
适配官方 [#2865](https://github.com/bytedance/deer-flow/pull/2865) 与
|
||
[#4230](https://github.com/bytedance/deer-flow/pull/4230)，不切换部署数据库。
|
||
|
||
### 会话检索与总数
|
||
|
||
`ThreadMetaRepository.search/count` 共享 SQL 条件构建。字符串 / 空值 metadata 过滤和系统线程
|
||
排除在 SQLite / PostgreSQL 中先于分页执行；常见 MySQL 字符串任务、智能体和行动过滤
|
||
同样下推。无剩余 Python 条件时，search 只发一条分页 SELECT，count 只发一条 COUNT。
|
||
这会减少向应用搬运的行数；没有为 JSON 字段自动创建索引，是否还需要表达式索引应由线上
|
||
执行计划确定，不能由“已下推 SQL”推断“数据库不再扫描候选行”。
|
||
|
||
保持原有规则：
|
||
|
||
- 普通会话仍由 `resolve_user_id` 隔离，get / check_access / 更新 / 删除授权不变。
|
||
- 非空 `metadata.taskId` 过滤仍开启任务共享；taskId、agent_id、actionId 全部条件同时生效。
|
||
- 默认列表继续隐藏 truthy `system`、非空白 `task_analysis_seat` 和既有系统 thread_type。
|
||
- search/count 使用相同条件；先过滤再分页，updated_at 相同按 thread_id 倒序稳定排序。
|
||
- 保留本地 `md.get(key) == None` 同时命中缺失键和 JSON null 的语义；数值过滤
|
||
保留 Python 的 `True == 1 == 1.0`，字符串数字不与数值混用。
|
||
- 嵌套 dict/list、特殊键名、超出 int64 的过滤值继续精确做剩余 Python 过滤，绝不忽略。
|
||
独立 openGauss 方言暂留原扫描路径。PG / MySQL 数值（含布尔）过滤也留在兼容路径，避免
|
||
DECIMAL / NUMERIC 与 Python 浮点数语义不同或 DOUBLE 舍入大整数改变任务匹配结果；
|
||
同一次查询的字符串条件仍可下推。SQLite 数值比较可直接下推。
|
||
- MySQL 字符串 id 按二进制比较，避免数据库默认不区分大小写 / 尾随空格的排序规则扩大命中。
|
||
使用 MySQL 5.7 可用的 JSON 函数；Python 空白判定通过完整字符替换实现，不依赖 5.7 的
|
||
字节正则去判定中文。
|
||
|
||
### PG ORM 连接池
|
||
|
||
`database.command_timeout` 默认 30 秒，可设有限正数或 null（关闭命令 deadline）。
|
||
只有 asyncpg 驱动接收该参数，MySQL / SQLite / psycopg 不接收。
|
||
PG 的 `pool_recycle` 未配置时默认 300 秒；MySQL / openGauss 保留 1800 秒，显式值不变。
|
||
初始建连与“数据库不存在、建库后重建 engine”使用相同参数。
|
||
|
||
```yaml
|
||
database:
|
||
backend: postgres
|
||
postgres_url: $DATABASE_URL
|
||
pool_size: 5
|
||
max_overflow: 10
|
||
pool_timeout: 30
|
||
pool_recycle: 300
|
||
command_timeout: 30
|
||
```
|
||
|
||
`pool_timeout` 是等待 ORM 池连接的期限，`command_timeout` 是 asyncpg 命令期限。
|
||
它们不限制 Python saver 锁排队，也不作用于使用 psycopg_pool 的 LangGraph checkpoint。
|
||
业务库已经是 MySQL 时，这个 PG ORM 补丁不会改善独立 PG checkpoint 的等待。
|
||
|
||
## 先确定部署真实使用的配置
|
||
|
||
仓库中存在不同配置文件，不能把源码模板当成生产进程最终配置：
|
||
|
||
- `offline-backend-20260512/config.yaml`：database.backend=mysql，checkpointer.type=postgres，
|
||
但 use_postgres=false；未被环境变量覆盖时，实际 checkpoint 回退 SQLite。
|
||
- `offline-backend-20260512/backend/config.yaml`：database/checkpointer 均为 SQLite。
|
||
- legacy `checkpointer:` 优先于 `database:`，`DEER_FLOW_USE_PG_CHECKPOINTER` 又可覆盖
|
||
legacy use_postgres。还应核对启动器选择的配置路径与已发布的后端版本。
|
||
|
||
现有 `GATEWAY_OPERATION_TIMING operation=thread_create` 日志会输出 metadata_store、
|
||
checkpointer 类型和各阶段耗时。默认延迟初始化时 checkpointer=deferred，只有
|
||
metadata_lookup / metadata_create；显式立即初始化时才有 checkpoint_create。
|
||
先看实际类型是 AsyncPostgresSaver、AsyncSqliteSaver 还是迁移代理，再判断瓶颈属于哪个库。
|
||
业务 MySQL 与 checkpoint PG 即使用同一应用，也有独立连接池。
|
||
|
||
## 创建线程能否只写 MySQL
|
||
|
||
**仅在创建阶段跳过 PG checkpoint，技术上可行；完整问答只使用 MySQL，当前实现不支持。**
|
||
|
||
已经实现延迟初始化：`POST /api/threads` 默认只创建业务元数据，不获取/查询/写入
|
||
checkpointer。业务库为 MySQL 时，创建阶段只使用 MySQL。metadata 保留 taskId、agent_id、
|
||
actionId 等既有分区信息，owner/user_id 仍由服务端控制；幂等重试保持既有记录。
|
||
|
||
已授权、尚无 checkpoint 的会话读取返回空 state/history；首次 run 由图运行时正常创建
|
||
checkpoint，继续使用原持久化、恢复和审计路径。需要旧行为的集成调用方可显式传入
|
||
`initialize_checkpoint: true`，等待立即写入空 checkpoint。
|
||
|
||
`GET /api/threads/{id}` 有业务记录时默认只读业务库，values 为空，running 状态映射 busy。
|
||
需要完整图状态请调用 `/state`，或显式 `?include_state=true`。仅有 checkpoint 的旧会话
|
||
保留读取兜底；所有路径仍先通过原 owner/project 权限装饰器，任务共享语义不变。
|
||
|
||
单纯去掉 await 并后台 `aput(empty_checkpoint)` 会与首次 run 竞态，晚到的空 checkpoint
|
||
可能成为新 head，不能这样处理。延迟初始化也只是把一部分 checkpoint 成本移到首次 run，
|
||
并不保证更快出现首字，必须分别测创建延迟和首字延迟。
|
||
|
||
完整移除 PG 还需要实现或验证 MySQL checkpointer（当前只有 memory/sqlite/postgres），并迁移
|
||
checkpoint/blob/write、兼容父链、恢复、回滚和子图。直接改 backend=mysql 不是此功能。
|
||
改用 SQLite 也保留 saver 串行 / 单写者竞争，不能据此承诺多人使用更快。
|
||
|
||
## 最终 History 能否后台刷新
|
||
|
||
**已实现不阻塞输入的后台状态同步，使用 SDK 的外部 history 接口。**
|
||
|
||
原 hook `fetchStateHistory: { limit: 1 }`，`onFinish(state, runMeta)` 使用最终 state。
|
||
锁定版本 `@langchain/langgraph-sdk@1.9.28` 的 `stream.lgp.js` 在 onSuccess 中先 await
|
||
`history.mutate`，再调用 onFinish，随后才结束 loading。`fetchStateHistory:false` 时，
|
||
有参数的 onFinish 仍会要求刷新，并把 getHistory 换成 getState，后者也读取 checkpointer。
|
||
因此简单关 history 或给服务端加 BackgroundTasks，不能消除这段等待。
|
||
|
||
实现细节：
|
||
|
||
1. 后端主图明确 checkpoint_ns=''，完整消费并关闭 alist 后再加工响应，避免序列化继续
|
||
占用 saver 实例锁；最新 head 返回完整 channel values，其余条目不重复消息。
|
||
2. 前端 `BackgroundHistory` 提供 SDK 外部 history。post-stream mutate 立即提交流式快照，
|
||
不等待网络；快照不携带旧 checkpoint id。服务器明确 end 后解除运行中，允许继续输入，
|
||
再在后台读取 limit=1 的最终历史；网络 EOF/错误仍使用原终态确认和重连流程。
|
||
3. thread id + run id + epoch + request id 共同保护写回。新一轮发送、切线程和卸载会使旧
|
||
请求失效，即使 fetch 忽略 abort 也不能回写；新线程采用服务器 id 不会误清空首轮流。
|
||
同一轮历史若缺少最新消息或内容尚为早期前缀，也不会覆盖已显示回答。
|
||
4. 每次后台读限制 15 秒，完成同步失败最多重试一次（间隔 500ms；401/403/404 不重试）。
|
||
失败保留流式正文并提示可继续提问；已有消息不会被清空。正常 end 的业务完成回调立即
|
||
使用本轮完整流快照，避免下一轮改变页面上下文而漏掉日志/归档；后台同步不重复回调。
|
||
断流恢复可能遗漏最终 SSE，其回调仍等待有效同步，失败时使用保留快照；下一轮正常
|
||
发送使用 checkpoint:null 跟随服务端最新 head，避免以旧快照分叉。
|
||
`isStateSyncing` 与运行中 loading 分开。EOF 若发生在完整 values 帧之间，保留 SDK
|
||
已积累的更新 token 消息，直到终态确认/恢复完成。
|
||
|
||
后台 history 仍消耗数据库和 saver 锁容量，能改善交互等待，但不能代替解决后端串行瓶颈。
|
||
|
||
## 登录与业务接口是否查询 PG
|
||
|
||
登录链路为 `routers/auth.py` → `get_local_provider()` → `LocalAuthProvider` →
|
||
`SQLiteUserRepository(sf)`。仓储类名为历史命名，实际 sf 来自
|
||
`init_engine_from_config(config.database)`，与 checkpoint 的独立连接池无关。
|
||
密码、用户名/口令、token 兑换、本地用户读取、session/token_version 和岗位授权同步
|
||
均使用业务仓储；token 兑换另外会请求上游身份接口，岗位同步也可能增加 MySQL 等待，
|
||
这些等待不能归因为 PG。CSRF、口令验证、白名单、单会话策略均未更改。
|
||
|
||
| 路径 | database.backend=mysql 时的数据来源 |
|
||
| --- | --- |
|
||
| 登录配置中的账号统计、登录/注册、身份查询、用户/岗位权限 | MySQL（策略配置仍可能来自配置文件） |
|
||
| 会话创建、搜索/count、已有元数据的线程信息 | MySQL；显式 eager/include_state 除外 |
|
||
| 智能体、任务、轻应用、菜单、业务映射等 SQL 仓储 | 共享业务 session factory → MySQL |
|
||
| state/history/messages、聊天执行/恢复/删除、从聊天提取资料 | checkpoint（配置为 PG 时访问 PG） |
|
||
| AI 写作等图执行的暂停/继续、管理员查看聊天问答明细 | 同属图状态或聊天内容，仍需要 checkpoint |
|
||
|
||
分库配置示例（不包含实际连接信息）：
|
||
|
||
```yaml
|
||
database:
|
||
backend: mysql
|
||
mysql_url: $MYSQL_DATABASE_URL
|
||
checkpointer:
|
||
type: postgres
|
||
use_postgres: true
|
||
connection_string: $CHECKPOINT_POSTGRES_URL
|
||
```
|
||
|
||
本次未修改现有部署连接串、迁移或删除旧表。已有数据库不会因改配置自动迁移，
|
||
不能在未迁移业务数据时直接把线上 database.backend 从 postgres 改成 mysql。
|
||
|
||
尚未完全拆开的边界：LangGraph BaseStore 目前跟随 legacy checkpointer 配置，
|
||
旧线程元数据迁移和部分运行时 store 使用它；Gateway 启动也仍等待 checkpoint/store
|
||
初始化与历史迁移。因此“已运行服务的业务请求不查询 checkpoint”不等于“PG 故障不影响
|
||
Gateway 冷启动”。若需要故障域完全独立，应后续拆运行时初始化/部署并迁移旧 Store，
|
||
不能以切换到易失内存或跳过旧数据的方式冒充完成。配置文件与文件型资料也未迁入 MySQL。
|
||
|
||
## 验证范围
|
||
|
||
后端相关回归：**133 passed**；前端 `pnpm test:thread-history`：**35 passed**。
|
||
从 backend 目录执行（Windows 将 PYTHONPATH 分隔符换为分号）：
|
||
|
||
```bash
|
||
PYTHONPATH=packages/harness:. .venv/bin/python -m pytest \
|
||
packages/harness/tests/test_thread_metadata_sql_filters.py \
|
||
packages/harness/tests/test_database_pool_deadlines.py \
|
||
packages/harness/tests/test_thread_operation_timing.py \
|
||
packages/harness/tests/test_thread_lazy_checkpoint.py \
|
||
packages/harness/tests/test_run_completion_timeout.py \
|
||
packages/harness/tests/test_run_stream_response_headers.py \
|
||
tests/test_account_authentication.py \
|
||
tests/test_gauss_database.py tests/test_ensure_orm_columns_sync.py -q
|
||
```
|
||
|
||
- 新增 SQLite 实际查询回归：标量 / 缺失值 / 嵌套兼容、系统排除、共享任务、agent/action
|
||
分区、普通用户隔离与修改权限、分页、标题/状态过滤，以及 1200 条噪声下只发单条查询。
|
||
- 新增 PG/MySQL SQL 编译与参数绑定、MySQL 二进制 id 比较、Gauss/服务端数值兼容回归。
|
||
- 新增 ORM 真实初始化路径的参数捕获：初次 / 建库重试、null 和自定义命令超时、MySQL /
|
||
SQLite 参数隔离、配置校验。
|
||
- HTTP + SQL + SQLite saver 验证持锁时 20 个创建请求完成、首轮真实 LangGraph 初始化与
|
||
第二轮续聊、元数据空线程读写、私有/任务共享权限、history 释放锁和取消关闭迭代器。
|
||
- 登录配置/密码登录/用户名口令登录/token 兑换/身份/用户列表使用实际 SQL 用户仓储，并安装
|
||
禁止访问 checkpoint 的桩，验证这些请求不会走聊天后端。
|
||
- 前端含实际安装 SDK 的 SSE 解码和 StreamManager 集成，验证 history 挂起不阻塞结束、
|
||
旧响应/新轮次/切线程/卸载隔离、异常/EOF恢复、超时/重试、旧副本保护和流式状态保留。
|
||
- 修改文件 Ruff 与 `git diff --check` 通过。前端全量 `pnpm typecheck` 仍有 81 条既有
|
||
类型错误；与 HEAD 基线比对并修正新增测试的类型后，本次改动文件无新增报错。
|
||
- 本机 Docker Linux daemon 未运行，未以真实 PG/MySQL 服务做集成或负载验证；不把 SQLite
|
||
查询与驱动配置测试当作线上延迟改善比例。发布前仍应在对应版本数据库做隔离集成验证。
|
||
- 额外尝试收集本机原有 `tests/test_gauss_mysql_compat.py` 时，因仓库缺少其导入的
|
||
`deerflow.persistence.gauss_mysql_compat` 模块失败；该本机测试文件未列入上述 133 项，
|
||
未为其添加与本次补丁无关的兼容模块。
|