deer-flow 线程历史无丢失方案：用 Append-Only 事件存储替代 Checkpoint 消息源的完整设计剖析

deer-flow 线程历史无丢失方案用 Append-Only 事件存储替代 Checkpoint 消息源的完整设计剖析【免费下载链接】deer-flowAn open-source long-horizon SuperAgent harness that researches, codes, and creates. With the help of sandboxes, memories, tools, skill, subagents and message gateway, it handles different levels of tasks that could take minutes to hours.项目地址: https://gitcode.com/GitHub_Trending/de/deer-flow本文基于 deer-flow 仓库中的设计计划文档 event-store-history 计划深入解析如何用 append-only 的 RunEventStore 取代 checkpoint 状态作为线程 state/history 接口的消息来源从根本上解决上下文压缩summarization导致历史消息永久丢失的问题读完后你将掌握该兼容层的完整设计约束、核心辅助函数实现、真实数据对齐分析、前端反馈分页联动与端到端回归测试策略。一、背景为什么 checkpoint 里的消息会“消失”deer-flow 的 Gatewaymake dev-pro启动的 Gateway 模式通过两个端点向外提供线程的对话内容GET /api/threads/{thread_id}/state— 获取线程最新的物化图状态POST /api/threads/{thread_id}/history— 获取线程的物化图状态历史。这两个端点位于 threads.py当前实现中get_thread_state定义在 threads.py#L1311-L1357get_thread_history定义在 threads.py#L1558-L1576消息正文均取自 LangGraph checkpoint 的channel_values[messages]通道。问题在于当会话触发 summarization上下文压缩后checkpoint 中的原始消息会被替换成一条“以 human 消息形态出现的合成摘要”压缩之前的所有消息在 checkpoint 中彻底消失。于是/state与/history返回的对话内容也随之截断——用户看到的是摘要而不是完整历史。计划文档给出的修复目标非常明确Goal:Replace checkpoint state with the append-only event store as the message source in the thread state/history endpoints, so summarization never causes message loss.其核心思路是RunEventStore运行事件库是**只追加append-only**的存储summarization 不会改写它因此它天然适合作为“完整消息流”的权威来源。改造后的约束是每条消息的响应结构保持不变前端聊天渲染路径无需修改但前端的反馈feedbackHook 必须对齐到同一份全量历史视图对应计划中的 Task 3。适用范围与前提条件计划中明确标注Scope仅 Gateway 模式make dev-pro。Standard 模式make dev直接走 LangGraph Server不经过这些 Gateway 端点summarize 丢消息的问题在该模式下仍然存在计划将其作为独立 follow-up 跟踪Task 5。Tech StackPythonFastAPI、SQLAlchemy、pytest、TypeScriptReact Query。已落地前提journal.py 的on_tool_end已修复为解包Command(update{messages: [ToolMessage(...)]})使返回状态更新类工具如present_files的新运行把内部ToolMessage内容写入事件库而不是str(Command(...))。该修复在仓库中可以确认journal.py#L540-L573 中对isinstance(output, Command)的分支会取出update中的 message 逐个写入事件库。更早落库的历史脏数据则由新辅助函数在读取时防御性清洗见下文_sanitize_legacy_command_repr。二、真实数据对齐分析事件库与 checkpoint 差在哪计划并非拍脑袋设计而是先用真实数据做了逐字段对齐对比真实POST /history响应基于 checkpoint与线程6d30913e-dcd4-41c8-8941-f66c716cf359的run_events表证据来自docs/resp.json与本地deerflow.db完整证据链见 runjournal-history-evaluation 规格文档。对齐结果如下表原文完整继承消息类型对比字段差异human_message全部字段事件库中id为Nonecheckpoint 中是 UUIDai_message (tool_call)全部字段6 个重叠字段完全一致0 差异ai_message (final)全部字段完全一致tool_result (normal)全部字段仅id不同Nonevs UUIDtool_result来自返回Command的工具content历史数据存的是str(Command(...))的 repr 而非内部 ToolMessage—— 新运行已在 journal.py 修复历史行由辅助函数清洗id差异的根因LangGraph 的 checkpoint 在图执行过程中为 HumanMessage 和 ToolMessage 分配id而事件库的写入发生在更早的时刻此时这些id还是None。AI 消息的id来自 LLM 响应lc_run--*前缀因此不受影响。id的修复方案对idNone的消息在读取时用uuid5(NAMESPACE_URL, f{thread_id}:{seq})生成确定性 UUID并只修补内容字典的副本绝不改动存储中的活动对象。确定性same input same output保证了同一消息在多次读取中id稳定前端渲染、反馈定位等依赖id的逻辑不会漂移。summarize 影响在该复现线程上的量化事件库有 16 条消息7 条 AI 9 条其他checkpoint 在 summarize 后只剩 12 条5 条 AI 7 条其他。AI 消息id重叠 5/7——缺失的 2 条正是被压缩掉的 pre-summarize 消息。这个数字直接证明了“checkpoint 作为消息源”在长会话下不可靠。三、文件结构与改造范围计划的文件改动面很小且职责单一原文完整继承文件动作职责backend/app/gateway/routers/threads.pyModify在get_thread_state与get_thread_history中用事件库消息替换 checkpoint 消息backend/tests/test_thread_state_event_store.pyCreate针对改造后端点的测试四、Task 1核心辅助函数_get_event_store_messages的设计与实现这是整个兼容层的心脏一个共享辅助函数负责从事件库加载全量消息流、为idNone的消息打上确定性 UUID、并防御性清洗历史遗留的Command(update...)repr。4.1 四条设计约束源自评估文档 §3/§4/§5全量分页而不是limit1000。RunEventStore.list_messages返回的是“最近 N 条记录”——固定 limit 会静默截断更早的消息。正确做法是先count_messages()确定总量或用after_seq游标循环翻页。先拷贝再修改。MemoryRunEventStore返回的是活动字典引用JSONL/DB 存储虽可能返回已分离的行但不能依赖这一点。打补丁前必须content dict(evt[content])。历史 Command repr 清洗。历史数据中存在content[content] Command(update{artifacts: [...], messages: [ToolMessage(contentX, ...)]})。用正则抽取内部 ToolMessage 的内容字符串并替换抽取失败则保持原样对已 summarize 的线程checkpoint 回退路径同样是错的所以保持原样仍然严格更优。用户上下文。DbRunEventStore.list_messages通过resolve_user_id(AUTO)做用户级隔离依赖require_permission设置的用户 contextvar。两个端点均已带该装饰器——这一点在当前仓库中可以确认threads.py#L1311-L1313 的require_permission(threads, read, owner_checkTrue)与 threads.py#L1558-L1560 上的相同装饰器正是该依赖的落点。辅助函数 docstring 需记录这一依赖。4.2 辅助函数与清洗器实现原文完整继承_LEGACY_CMD_INNER_CONTENT_RE re.compile( rToolMessage\(content(?Pq[\])(?Pinner.*?)(?Pq), re.DOTALL, ) def _sanitize_legacy_command_repr(content_field: Any) - Any: Recover the inner ToolMessage text from a legacy str(Command(...)) repr. Runs that pre-date the on_tool_end fix in journal.py stored str(Command(update{messages:[ToolMessage(contentX, ...)]})) as the tool_result content. New runs store X directly. For old threads, try to extract X defensively; return the original string if extraction fails (still no worse than the current checkpoint-based fallback, which is broken for summarized threads anyway). if not isinstance(content_field, str) or not content_field.startswith(Command(update): return content_field match _LEGACY_CMD_INNER_CONTENT_RE.search(content_field) return match.group(inner) if match else content_field async def _get_event_store_messages(request: Request, thread_id: str) - list[dict] | None: Load messages from the event store, returning None if unavailable. The event store is append-only and immune to summarization. Each message events content field contains a model_dump()d LangChain Message dict that is already JSON-serialisable. **Full pagination, not a fixed limit.** RunEventStore.list_messages returns the newest limit records when no cursor is given, which silently drops older messages. We call count_messages() first and request that many records. For stores that may return fewer (e.g. filtered by user), we also fall back to after_seq-cursor pagination. **Copy-on-read.** Each content dict is copied before id is patched so the live store object is never mutated; MemoryRunEventStore returns live references. **Legacy Command repr sanitization.** See _sanitize_legacy_command_repr. **User context.** DbRunEventStore is user-scoped by default via resolve_user_id(AUTO) (see runtime/user_context.py). Callers of this helper must be inside a request where require_permission has populated the user contextvar. Both get_thread_history and get_thread_state satisfy that. Do not call this helper from CLI or migration scripts without passing user_idNone explicitly. Returns None when the event store is not configured or contains no messages for this thread, so callers can fall back to checkpoint messages. try: event_store get_run_event_store(request) except Exception: return None try: total await event_store.count_messages(thread_id) except Exception: logger.exception(count_messages failed for thread %s, sanitize_log_param(thread_id)) return None if not total: return None # Batch by page_size to keep memory bounded for very long threads. page_size 500 collected: list[dict] [] after_seq: int | None None while True: page await event_store.list_messages(thread_id, limitpage_size, after_seqafter_seq) if not page: break collected.extend(page) if len(page) page_size: break after_seq page[-1].get(seq) if after_seq is None: break messages: list[dict] [] for evt in collected: raw evt.get(content) if not isinstance(raw, dict) or type not in raw: continue # Copy to avoid mutating the store-owned dict. content dict(raw) if content.get(id) is None: content[id] str(uuid.uuid5(uuid.NAMESPACE_URL, f{thread_id}:{evt[seq]})) # Sanitize legacy Command reprs on tool_result messages only. if content.get(type) tool: content[content] _sanitize_legacy_command_repr(content.get(content)) messages.append(content) return messages if messages else None实现要点解读失败即回退degrade, not failget_run_event_store(request)失败、count_messages失败、或该线程没有任何消息时一律返回None由端点回退到 checkpoint 消息——这保证事件库不可用不会让 state/history 接口 500。仓库中也有同款防御风格的现成例子threads.py#L83-L91 的_optional_run_event_store明确写道“读路径不能依赖 feedseq 只是定位元数据缺失时退化为客户端自己的排序规则而不是失败”。分页取全量 内存有界page_size 500逐页拉取after_seq游标推进页长不足page_size即停止。长线程数千条消息不会一次性撑爆内存。只处理“消息类”事件过滤条件是isinstance(raw, dict) and type in rawtrace 类事件如llm_request自然被排除——事件库中的消息与轨迹混存读取端必须自己筛。uuid5 打补丁确定性 id 以{thread_id}:{seq}为命名空间输入seq是消息在 feed 中的绝对位置因此 id 与线程、位置强绑定且跨请求稳定。4.3 事件库从哪里来make_run_event_store工厂辅助函数拿到的event_store由 Gateway 启动时注入。仓库中 store/init.py 的make_run_event_store(config)按run_events.backend配置选择实现memory默认/未配置MemoryRunEventStore进程内存储测试与开发场景使用dbDbRunEventStore基于 SQLAlchemy 会话工厂的持久化存储若database.backendmemory但run_events.backenddb会回退到内存实现jsonlJsonlRunEventStore按行追加的 JSONL 文件存储。这与计划中“JSONL/DB stores may return detached rows”的表述相呼应——正因为存在三种异构后端“先拷贝再改”才成为硬性约束而非风格偏好。事件库的工厂行为有专门测试覆盖见 test_run_event_store.py含make_run_event_store各 backend 分支的用例。4.4 配套测试原文完整继承计划要求先写测试再写实现TDD。测试文件backend/tests/test_thread_state_event_store.py用MemoryRunEventStore造数据覆盖消息提取、id 补丁、空线程、工具调用字段保留四类断言Tests for event-store-backed message loading in thread state/history endpoints. from __future__ import annotations import uuid import pytest from deerflow.runtime.events.store.memory import MemoryRunEventStore pytest.fixture() def event_store(): return MemoryRunEventStore() async def _seed_conversation(event_store: MemoryRunEventStore, thread_id: str t1): Seed a realistic multi-turn conversation matching real checkpoint format. # human_message: id is None (same as real data) await event_store.put( thread_idthread_id, run_idr1, event_typehuman_message, categorymessage, content{ type: human, id: None, content: [{type: text, text: Hello}], additional_kwargs: {}, response_metadata: {}, name: None, }, ) # ai_tool_call: id is set by LLM await event_store.put( thread_idthread_id, run_idr1, event_typeai_tool_call, categorymessage, content{ type: ai, id: lc_run--abc123, content: , tool_calls: [{name: search, args: {q: cats}, id: call_1, type: tool_call}], invalid_tool_calls: [], additional_kwargs: {}, response_metadata: {}, name: None, usage_metadata: {input_tokens: 100, output_tokens: 50, total_tokens: 150}, }, ) # tool_result: id is None (same as real data) await event_store.put( thread_idthread_id, run_idr1, event_typetool_result, categorymessage, content{ type: tool, id: None, content: Found 10 results, tool_call_id: call_1, name: search, artifact: None, status: success, additional_kwargs: {}, response_metadata: {}, }, ) # ai_message: id is set by LLM await event_store.put( thread_idthread_id, run_idr1, event_typeai_message, categorymessage, content{ type: ai, id: lc_run--def456, content: I found 10 results about cats., tool_calls: [], invalid_tool_calls: [], additional_kwargs: {}, response_metadata: {finish_reason: stop}, name: None, usage_metadata: {input_tokens: 200, output_tokens: 100, total_tokens: 300}, }, ) # Also add a trace event — should NOT appear await event_store.put( thread_idthread_id, run_idr1, event_typellm_request, categorytrace, content{model: gpt-4}, ) class TestGetEventStoreMessages: Verify event store message extraction with id patching. pytest.mark.asyncio async def test_extracts_all_message_types(self, event_store): await _seed_conversation(event_store) events await event_store.list_messages(t1, limit500) messages [evt[content] for evt in events if isinstance(evt.get(content), dict) and type in evt[content]] assert len(messages) 4 assert [m[type] for m in messages] [human, ai, tool, ai] pytest.mark.asyncio async def test_null_ids_get_patched(self, event_store): Messages with idNone should get deterministic UUIDs. await _seed_conversation(event_store) events await event_store.list_messages(t1, limit500) messages [] for evt in events: content evt.get(content) if isinstance(content, dict) and type in content: if content.get(id) is None: content[id] str(uuid.uuid5(uuid.NAMESPACE_URL, ft1:{evt[seq]})) messages.append(content) # All messages now have an id for m in messages: assert m[id] is not None assert isinstance(m[id], str) assert len(m[id]) 0 # AI messages keep their original id assert messages[1][id] lc_run--abc123 assert messages[3][id] lc_run--def456 # Human and tool messages get deterministic ids (same input same output) human_id_1 str(uuid.uuid5(uuid.NAMESPACE_URL, t1:1)) assert messages[0][id] human_id_1 pytest.mark.asyncio async def test_empty_thread(self, event_store): events await event_store.list_messages(nonexistent, limit500) messages [evt[content] for evt in events if isinstance(evt.get(content), dict)] assert messages [] pytest.mark.asyncio async def test_tool_call_fields_preserved(self, event_store): await _seed_conversation(event_store) events await event_store.list_messages(t1, limit500) messages [evt[content] for evt in events if isinstance(evt.get(content), dict) and type in evt[content]] # AI tool_call message ai_tc messages[1] assert ai_tc[tool_calls][0][name] search assert ai_tc[tool_calls][0][id] call_1 # Tool result tool messages[2] assert tool[tool_call_id] call_1 assert tool[status] success测试数据的设计刻意复刻了真实数据的形态human 与 tool 消息idNone、AI 消息idlc_run--*并额外塞入一条llm_requesttrace 事件验证过滤逻辑。运行方式cd backend PYTHONPATH. uv run pytest tests/test_thread_state_event_store.py -v4.5 两个端点的 Before/Afterget_thread_history约 585-590 行处的消息注入段Before# Attach messages from checkpointer only for the latest checkpoint if is_latest_checkpoint: messages channel_values.get(messages) if messages: values[messages] serialize_channel_values({messages: messages}).get(messages, []) is_latest_checkpoint FalseAfter# Attach messages: prefer event store (immune to summarization), # fall back to checkpoint messages when event store is unavailable. if is_latest_checkpoint: es_messages await _get_event_store_messages(request, thread_id) if es_messages is not None: values[messages] es_messages else: messages channel_values.get(messages) if messages: values[messages] serialize_channel_values({messages: messages}).get(messages, []) is_latest_checkpoint Falseget_thread_state约 443-444 行处Beforereturn ThreadStateResponse( valuesserialize_channel_values(channel_values),Aftervalues serialize_channel_values(channel_values) # Override messages with event store data (immune to summarization) es_messages await _get_event_store_messages(request, thread_id) if es_messages is not None: values[messages] es_messages return ThreadStateResponse( valuesvalues,注意history端点只在最新 checkpoint上附带messages历史条目不重复携带完整对话避免每个条目都复制整段会话——这一“最新检查点独占 messages”的策略在改造前后保持一致响应形状不变前端渲染路径因此无需改动。验证与提交原文继承# 全量后端测试 cd backend PYTHONPATH. uv run pytest tests/ -v --timeout30 -x # 提交 git add backend/app/gateway/routers/threads.py backend/tests/test_thread_state_event_store.py git commit -m feat(threads): load messages from event store instead of checkpoint state Event store is append-only and immune to summarization. Messages with null ids (human, tool) get deterministic UUIDs based on thread_id:seq for stable frontend rendering.五、Task 2可选项已推迟调低flush_threshold缩短流中段空窗计划明确将其标注为“不是正确性修复”复评见规格文档发现RunJournal已在run_end、run_error、cancel 以及 workerfinally路径上强制 flush该调参唯一能收窄的窗口是进程硬崩溃或运行中 reload。因此决定单独决策、不与 Task 1 的合并耦合。若日后推进把 journal.py#L229 中flush_threshold的默认值从 20 改为 5当前仓库中该默认值确为 20重跑tests/test_run_journal.py并以独立的perf(journal): …提交。从源码结构看阈值触发点在缓冲区达到self._flush_threshold时批量写库journal.py#L675 一带的put_batch逻辑调低它意味着更频繁的落盘需要自行权衡 IO 开销。六、Task 3前端useThreadFeedback分页对齐当/history开始返回事件库支撑的全量消息流后前端的runIdByAiIndex映射“第 N 条 AI 消息属于哪个 run”也必须覆盖全量流否则按位置映射的 AI 索引会漂移feedback 点击会落到错误的run_id上。原 Hook 硬编码limit200计划要求改造 hooks.ts约 679 行处Step 1把固定?limit200换成after_seq全量翻页。改动前const res await fetchWithAuth( ${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}/messages?limit200, );改为用after_seq翻页的循环累积messages直到某页返回数量小于页大小。这里仓库提供了可直接引用的参数名依据/messages端点定义在 thread_runs.py#L1160-L1174签名正是limit: int Query(default50, ge1, le200)与after_seq: int | None Query(defaultNone, ge1)并原样转发给event_store.list_messages(...)。计划特别提醒写 TS 前先核对thread_runs.py:285-323的实际参数名不同端点分页参数可能不同。Step 2防御性索引保护。runIdByAiIndex[aiMessageIndex]在前端先渲染乐观状态、消息查询尚未刷新时仍可能为undefinedmessage-list.tsx:71现有的?? undefined已处理该情况不得移除。Step 3流结束后失效缓存。在useThreadStream或流结束处理处于流关闭时调用queryClient.invalidateQueries({ queryKey: [thread-feedback, threadId] })让runIdByAiIndex立刻纳入新 run 的 AI 消息。Step 4 / Step 5运行cd frontend pnpm check然后git add frontend/src/core/threads/hooks.ts git commit -m fix(feedback): paginate useThreadFeedback and invalidate after stream七、Task 4端到端回归测试——summarize 多 run 反馈计划要求在backend/tests/test_thread_state_event_store.py中追加一个直击本 bug 类别的回归测试一个被 summarize 过的、至少包含两个 run 的线程断言 feedback 点击必须命中正确的run_id。测试步骤向MemoryRunEventStore灌入两个 run 的消息r1human ai human air2human ai再模拟一个“丢掉r1消息”的被 summarize 的 checkpoint 状态调用_get_event_store_messages后断言消息长度等于事件库的数量而不是 checkpoint 的数量第一条消息是r1的原始 human而不是摘要AI 消息按顺序保留其lc_run--*id任何idNone消息获得稳定的uuid5(...)idtool_result 中遗留的str(Command(update...))content 字段被清洗为内部文本。运行并随 Task 1、Task 3 的改动一起提交保证测试与实现同批落地cd backend PYTHONPATH. uv run pytest tests/test_thread_state_event_store.py -v八、Task 5Standard 模式的遗留缺口文档化Standard 模式make dev下/threads/{id}/history直接打到 LangGraph Server不经过刚改造的 Gateway 路由summarize 丢消息的症状在该模式下仍可复现。计划要求把缺口记录到文档或独立 issue 中原文完整继承Follow-up — Standard mode summarize bugget_thread_historyinbackend/app/gateway/routers/threads.pyis only hit in Gateway mode. Standard mode proxies/api/langgraph/*directly to the LangGraph Server. The summarize-message-loss symptom is still reproducible there. Options: (a) teach the LangGraph Server checkpointer to branch on an override, (b) move/historybehind Gateway in Standard mode as well, (c) accept as known limitation for Standard mode. Decide before GA.九、从当前仓库源码看这一设计的落点与演进以本仓库当前代码为证据可以对上述设计的工程价值做三点印证以下均为从源码结构可确认的事实或可推断的结论事件库已是 Gateway 的多处权威数据源而非新引入的孤立组件。get_run_event_store依赖在 deps.py#L639 定义为“无则抛错”的强依赖_require(run_event_store, Run event store)启动时由make_run_event_store构建并挂在app.statedeps.py#L535而history端点在需要精确归属 run 时也会调用event_store.find_latest_ai_message_run_idsthreads.py#L1656-L1671——“checkpoint 存状态、事件库存事实流”的分工在现有代码中已经成立。seq 定位元数据与本文计划同源。当前get_thread_state与get_thread_history都会对 checkpoint 消息执行stamp_messages_with_seq(_optional_run_event_store(request), ...)threads.py#L1346、threads.py#L1744-L1748即消息在事件流 feed 中的绝对位置是从事件库补打到响应里的——“没有 seq被救回的早期轮次就没有可放置的绝对位置”。从源码结构看计划中“uuid5 依赖 seq 生成稳定 id”的设计正是建立在这套 seq 体系之上。防御式回退是 Gateway 的既定风格。_optional_run_event_store的 docstringthreads.py#L83-L91说明“读路径不得依赖 feed 存在”与_get_event_store_messages返回None触发 checkpoint 回退的设计完全同构。十、小结这篇计划文档给出了 deer-flow 解决“summarization 吞历史”问题的完整工程方案以真实数据逐字段对齐确定差异idNone、遗留Commandrepr用四条硬约束全量分页、拷贝再改、防御清洗、用户上下文设计_get_event_store_messages兼容层保持响应形状不变使前端渲染零改动再以前端after_seq分页、流结束缓存失效、多 run 回归测试补齐联动面并把 Standard 模式的遗留缺口显式记录为 GA 前决策项。整个方案“事件库 append-only、checkpoint 仅存状态”的分层原则与当前仓库中make_run_event_store三后端工厂、seq 打戳、run 精确归属等实现相互印证是长会话 Agent 系统中“消息历史不丢失”这一问题的一个可参考范本。【免费下载链接】deer-flowAn open-source long-horizon SuperAgent harness that researches, codes, and creates. With the help of sandboxes, memories, tools, skill, subagents and message gateway, it handles different levels of tasks that could take minutes to hours.项目地址: https://gitcode.com/GitHub_Trending/de/deer-flow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考