DeerFlow 端到端流式输出架构解析LangGraph 事件如何同时送达 HTTP SSE 与嵌入式 Python 客户端【免费下载链接】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流式输出是任何长时运行 Agent 产品体验的命脉浏览器、IM 渠道要看到 token 级打字机效果Jupyter / 脚本 / 测试需要拿到逐步到达的原生对象。DeerFlow 为此维护了两条并行且刻意不复用的流式管道——Gatewayasync HTTP SSE JSON与DeerFlowClientsync in-process 原生 LangChain 对象。本文完整梳理这两条路径的分工与契约、stream_mode三层语义、有界历史与gap恢复、客户端三个 id 去重集合的微妙不变式以及防止漏订messages模式这类回归的测试策略。阅读完本文你将掌握DeerFlow 为什么需要两条流式路径而非让客户端复用 Gateway、values / messages / custom三种流模式各自的发射时机与事件契约、messages-tuple与messages两种命名在协议层间的翻译关系、SSE 断线重连与gap恢复边界以及嵌入式客户端三套set[str]各自守护的不变式。TL;DR先建立整体心智模型DeerFlow 存在两条并行的流式路径服务两种截然不同的消费者模型Gateway 路径async def/ HTTP SSE / JSON 序列化服务浏览器前端与飞书 / Slack / Telegram 等 IM 渠道DeerFlowClient 路径sync generator / in-process / 原生 LangChain 对象服务 Jupyter、集成脚本与测试。 两条路径无法合并——消费者模型根本不同详见 为什么有两条流式路径。两条路径都从create_agent()工厂出发核心订阅 LangGraph 的stream_mode[values, messages, custom]。其中values是节点级 state 快照messages是 LLM token 级 deltacustom是显式StreamWriter事件DeerFlow 内置 custom 事件同时通过 callback dispatch 暴露为astream_events(versionv2)的on_custom_event。嵌入式客户端为每次stream()调用维护三个set[str]seen_ids/streamed_ids/counted_usage_ids。三者看似相似实际各管理一个独立不变式不能合并详见 三个 id set 为什么不能合并。为什么有两条流式路径两条路径服务的消费者模型根本不同其差异可用一张对照表概括维度Gateway 路径DeerFlowClient 路径入口FastAPI/runs/streamendpointDeerFlowClient.stream(message)触发层worker.pyrun_agentclient.pyDeerFlowClient.stream执行模型async defagent.astream()sync generator agent.stream()事件传输StreamBridgeasyncio Queuesse_consumer直接yield序列化serialize(chunk)→ 纯 JSON dict匹配 LangGraph Platform wire 格式StreamEvent.data携带原生 LangChain 对象消费者前端useStreamReact hook、飞书/Slack/Telegram channel、LangGraph SDK 客户端Jupyter notebook、集成测试、内部 Python 脚本生命周期管理RunManagerrun_id 跟踪、disconnect 语义、multitask 策略、heartbeat每次stream()生成一个轻量 run_id 供 runtime context / tracing / per-run middleware 使用函数返回即结束断连恢复Last-Event-IDSSE 重连无需要两条路径的存在是 DRY 的刻意妥协Gateway 的全部基础设施async Queue JSON RunManager都是为了跨网络边界把事件送给 HTTP 消费者。当生产者agent和消费者Python 调用栈处于同一进程时整套机制都是纯开销。为什么不能让 DeerFlowClient 复用 Gateway设计上曾经考虑过三种复用方案均被否决让client.stream()变成async def client.astream()——breaking change。Jupyter notebook 和同步脚本需要硬塞用不上的async for/asyncio.run()DeerFlowClient把 agent 当普通函数调用的核心卖点直接消失。在client.stream()内部启动独立事件循环线程用StreamBridge在 sync/async 之间做桥接——会引入线程池、队列、信号量。为了消除重复却把复杂度带进来属于典型的 wrong abstraction开销高于复用收益。让run_agent自己兼容 sync mode——给 Gateway 增加一条用不到的死分支污染 worker 的职责焦点。因此两条路径的事件处理逻辑相似但不共享这是刻意设计不是疏忽。源码佐证这段取舍在 client.py_stream_turn的 docstring 中有同样明确的阐述——run_agent是async def且使用agent.astream()而 client 是 sync generator 使用agent.stream()两者桥接需要每调用启动一个 event loop thread同时 Gateway 事件需要serialize()做 JSON/SSE wire 传输in-process 调用方拿到的是可直接使用的StreamEvent结构。两条路径订阅哪些 LangGraph stream mode 的同步依赖行为断言而非共享常量来保证。LangGraphstream_mode三层语义LangGraph 的agent.stream(stream_mode[...])是多路复用接口一次订阅多个 mode每个 mode 都是一个独立的事件源。三个核心 mode 的语义对照Mode发射时机Payload粒度values每个 graph 节点完成后完整 state dicttitle、messages、artifacts节点级messagesLLM 每次 yield 一个 chunktool 节点完成时(AIMessageChunk \| ToolMessage, metadata_dict)token 级custom用户代码显式调用StreamWriter.write()任意 dict应用定义on_custom_event用户代码调用dispatch_custom_event()通过astream_events(versionv2)消费name 任意data应用定义关键事实这些 mode 不是详细程度递增的梯度而是互相独立的平行事件源。消费者必须显式订阅自己需要的那一个缺订任何一个尤其是messages都会静默丢失一整类事件——这正是历史回归的温床见 为什么这个设计容易出 bug。DeerFlow 对 custom 事件的双通道约束DeerFlow 自身产生的事件必须通过 custom_events.py 中定义的同步 / 异步 helper 发送。其约束为每个内置 payload 必须携带非空字符串type缺少合法type的 payload 只进入customstream不会出现在astream_events。helper 的执行顺序是先写入customstream再做 best-effort callback dispatchcallback 名称取 payload 的typedata保留完整 payload。这样原生 Gateway / Web UI /DeerFlowClient的 custom 事件保持不变astream_events消费者也能观察到同一事件。callback dispatch 的普通异常只记录 debug 日志不允许打断原有 writer 链路writer 自身的异常语义保持不变。从源码看emit_custom_event / aemit_custom_event 先调用writer(payload)再经_event_name校验后走dispatch_custom_event(event_name, payload)异步版走adispatch_custom_eventtype缺失时只记一条 debug 日志并跳过 callback。GraphBubbleUp会被特殊透传其余异常全部吞掉记为 debug——这是writer 为主、callback 为 optional 观察通道这一设计意图的直接实现。两套命名的由来同一件事在三个协议层有完全不同的名字Application HTTP / SSE LangGraph Graph ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ frontend │ │ LangGraph │ │ agent.astream│ │ useStream │──messages- │ Platform SDK │──messages──│ graph.astream│ │ Feishu IM │ tuple──────│ HTTP wire │ │ │ └──────────────┘ └──────────────┘ └──────────────┘Graph 层agent.stream/agent.astreamLangGraph Python 直接 APImode 叫messages。Platform SDK 层langgraph-sdkHTTP client跨进程 HTTP 契约mode 叫messages-tuple。Gateway worker负责在两个命名体系之间显式翻译。后果DeerFlowClient.stream()直接调用agent.stream()Graph 层必须传messageschannels/manager.py 通过langgraph-sdk走 HTTP SDK所以传messages-tuple。这两个字符串不能互相替代也不能被抽成一个共享常量——它们是不同协议层的 type alias共享只会让某一层说不是它母语的话。源码佐证翻译逻辑现在集中在 runtime/stream_modes.py。RunStreamMode定义了 DeerFlow runtime 边界完整支持的公开 mode 集合values、messages-tuple、updates、debug、tasks、checkpoints、customnormalize_stream_modes()负责归一化与白名单校验不支持的 mode 直接抛UnsupportedStreamModeError而真正的翻译由to_langgraph_stream_modes()完成——mapped [messages if mode messages-tuple else mode for mode in modes]随后去重并返回给agent.astream。Gateway 侧的 worker.pyrun_agent通过to_langgraph_stream_modes消费这份翻译并在内部根据是否包含messages-tuple决定是否启用_LargeFileToolChunkBatcher大文件工具 chunk 分批等逻辑。Gateway 路径async HTTP SSEGateway 路径是完整的多组件协作流水线参与方与调用时序如下各关键组件及其职责worker.pyrun_agent在asyncio.Task中运行agent.astream()把每个(mode, chunk)经serialize(chunk, modemode)转成 JSON 后调用bridge.publish()结束时调用bridge.publish_end()。runtime/stream_bridge抽象 Queue 层的完整实现位于 runtime/stream_bridge/含base.py抽象基类、memory.py与redis.py两种 backend。publish/subscribe解耦生产者和消费者支持Last-Event-ID重连、心跳、多订阅者 fan-out。从 base.py 可以看到StreamGap与END_SENTINELStreamEvent(id, event__end__, dataNone)是 stream 结束 / 出现断档的两种显式信号。services.pysse_consumer从 bridge 订阅事件并配合format_sse()格式化成标准 SSE wire 帧event: name\ndata: json\n\nevent_id可选。runtime/serialization.pyserializemode-aware 序列化在messagesmode 下serialize_messages_tuple把(chunk, metadata)转成[chunk.model_dump(), metadata]在valuesmode 下额外剥离__pregel_*内部键并丢弃 hide_from_ui 消息中的 base64 data: 图片块。StreamBridge的存在价值当生产者run_agent任务和消费者HTTP 连接运行在不同的 asyncio task 里时需要一个可以跨 task 传递事件的中介。Queue 同时承担断连重连的缓冲区和多订阅者的 fan-out。有界历史与gap恢复Last-Event-ID只保证在保留窗口内完整重放并不代表无限历史。契约可以总结为有效游标仍在窗口内时bridge 从该 ID 的下一条事件正常恢复。有效游标已被queue_maxsize裁剪或在线消费者慢到落后水位线时Gateway 会在任何部分重放之前先发送一帧gap而不是从当前最早事件静默地部分重放event: gap data: {code:stream_replay_gap,run_id:...,requested_event_id:...,earliest_available_event_id:...,latest_available_event_id:...,recovery:reload_durable_state}关于gap帧需要明确几个语义gap帧没有SSEid:其后也没有正常的end当前订阅随即关闭。它是恢复边界而不是客户端断开因此不会触发on_disconnectcancel。当缓冲区无任何保留事件时earliest_available_event_id与latest_available_event_id均为null。客户端必须丢弃不再可信的瞬时状态重新读取 thread checkpoint 与持久化的 run-event / message history然后以latest_available_event_id为游标跟随新事件若缓冲区为空即latest_available_event_id为null则不带游标重新加入流。DeerFlow Web UI 会自动执行这一整套流程且最多连续恢复五次。Memory 与 Redis 两种 backend 在这一契约上的细微差异来自原设计文档的补充说明Redis对无游标、空 stream 上已建立的阻塞等待遵循相同契约——第一次XREAD唤醒的数据在交付前仍是 provisional baselinebridge 会用下一次事务快照确认其尾 ID 仍在保留窗口若生产者已裁剪该基线订阅直接返回requested_event_id: null的gap。该检查有明显的性能代价每轮订阅需要一个包含XRANGE、XREVRANGE、非阻塞XREAD的事务快照空闲时还需要单独的阻塞XREAD来唤醒。有效但已淘汰 vs malformed cursor 是不同策略该契约只要求前者产生gap。Redis 对 malformed ID 仍从 live tail 等待Memory 对 malformed ID 及序号不低于水位线的未知 ID 仍采用最早保留事件策略。对序号已低于水位线的数字格式 foreign IDMemory 无法再校验已淘汰的 timestamp因此保守返回gap优先保证客户端不会把不完整重放误认为完整。孤儿 run 恢复Memory 与 Redis 都只保留queue_maxsize条数据事件。Redis backend 在每次publish()/publish_end()刷新 retained stream key 的 TTL启动恢复与基于 worker lease 的周期恢复共用 Gateway stream terminalization 路径——RunManager先把 orphan run 持久化为error并写入显式stop_reasonorphan_recovered随后 Gateway 发布END_SENTINEL并安排 stream cleanup。这里隐含一个判断store-only SSE 与/waitconsumer不能把普通 durable terminal status 当成流已完成否则可能跳过延迟发布的 error 等尾部事件只有orphan_recovered信号能在 heartbeat 时触发 END fallback——因为此时 producer 已被确认失联。TTL 是 Redis 内存和故障安全网而不是正常的 subscriber 终止机制。StreamBridge 可配置项Bridge 的关键参数集中在 config/stream_bridge_config.py默认值与边界如下参数默认值说明typememorybackend 类型memory为进程内事件日志仅单进程可用redis用于多 worker Docker 部署Redis Streams。redis_urlNoneRedis URL省略时按DEER_FLOW_STREAM_BRIDGE_REDIS_URL→REDIS_URL→redis://localhost:6379/0的顺序回退。queue_maxsize256每个 run 保留的最大事件数memory 队列长度 / redis stream MAXLEN最小为 1。heartbeat_interval_seconds15.0空闲时两次 heartbeat 之间的秒数0 v ≤ 86400且拒绝布尔值统一作用于 SSE 客户端、非流式 wait 请求与内部 stream 订阅者显式传给subscribe()的值可覆盖单次订阅。max_connectionsNoneRedis 连接池上限。每个活跃 SSE 客户端会持有一条阻塞在XREAD ... BLOCK最长heartbeat_interval_seconds的连接数百并发客户端即数百连接。仅 redis bridge 生效。stream_ttl_seconds86400Redis stream key 滚动 TTL每次publish/publish_end刷新为 0 时关闭。是故障兜底而非正常终止机制。recovered_stream_cleanup_delay_seconds60.0孤儿 run 被恢复并发布 END marker 后删除 stream key 前等待的秒数给重连 SSE 客户端留出排空尾部信号的时间。仅 redis bridge 生效。周期扫描、逐行状态写入和 Gateway callback 作为一个受监督的 single-flight 后台 task 执行慢任务不会堆积也不会阻塞唯一的 lease heartbeatshutdown 优先收敛活跃 run 再处理恢复 task尚未执行的延迟 stream cleanup 会改为立即删除。只有runtime yield 前、无并发请求的启动恢复会把最新受影响 thread 标记为 error周期恢复不做非原子的 thread 投影。DeerFlowClient 路径sync in-process对比之下sync 路径每个环节的移动部件都显著更少没有RunManager一次stream()调用对应一次生命周期只生成轻量run_id供 runtime context、tracing 与 per-run middleware 使用函数返回即结束。从源码看stream()还会基于run_id绑定 trace context——每次next()步进前后bind_trace_id/reset_trace_id成对执行从不跨越yield避免把 id 泄漏进调用方 Context 或触发跨 Context 的 GC 清理异常。没有StreamBridge直接yield生产与消费发生在同一个 Python 调用栈不需要跨 task 中介。没有 JSON 序列化StreamEvent.data直接携带原生 LangChain 值——AIMessage.content、usage_metadata的UsageMetadataTypedDict以及非None的ToolMessage.artifact。messages-tuple工具结果与values快照里的工具消息都会保留 artifact没有 artifact 的工具消息维持原有字段形状。Jupyter 用户拿到的是真正的类型而不是经过网络序列化后的匿名值。没有 asyncio调用者可以直接写for event in ...不必写async for。在 client.py 中事件类型以StreamEvent数据类承载type为values | messages-tuple | custom | end与 LangGraph SSE 协议保持一致因此消费者可以在 HTTP 流与嵌入式模式间切换而无需改写事件处理逻辑。_stream_turn的 docstring 明确列出各事件 payload 的 shape- typevalues data{title: str|None, messages: [...], artifacts: [...]} - typecustom data{...} - typemessages-tuple data{type: ai, content: delta, id: str} - typemessages-tuple data{type: ai, content: delta, id: str, usage_metadata: {...}} - typemessages-tuple data{type: ai, content: , id: str, tool_calls: [...]} - typemessages-tuple data{type: ai, content: , id: str, additional_kwargs: {...}} - typemessages-tuple data{type: tool, content: str, name: str, tool_call_id: str, id: str} # Tool results also include artifact when the source ToolMessage has a non-None artifact. - typeend data{usage: {input_tokens: int, output_tokens: int, total_tokens: int}}一个值得注意的边界多轮对话需要 checkpointer。构造DeerFlowClient时传入checkpointer后thread_id才能保留跨轮上下文否则每次stream()/chat()都是无状态调用thread_id仅用于文件隔离uploads / artifacts。另外系统提示词含日期、memory 与 skills 上下文在 agent 首次创建时生成并按配置键缓存长驻进程中若外部修改了 memory / skills 需调用reset_agent()强制重建。消费语义delta vs cumulativeLangGraphmessagesmode 给出的是delta每个AIMessageChunk.content只包含这一次新 yield 的 token不是从头累计的完整文本。这与 LangChain 的fs2 Stream风格一致——上游发增量下游负责累加Gateway 路径前端useStreamReact hook 自己维护累加器。DeerFlowClient 路径chat()方法替调用者做累加。DeerFlowClient.chat()的 O(n) 累加器client.pychat()的实现是逐 id 收集 delta 列表、最后一次性 joinchunks: dict[str, list[str]] {} last_id: str for event in self.stream(message, thread_idthread_id, **kwargs): if event.type messages-tuple and event.data.get(type) ai: msg_id event.data.get(id) or delta event.data.get(content, ) if delta: chunks.setdefault(msg_id, []).append(delta) last_id msg_id return .join(chunks.get(last_id, ()))为什么不用buffers[id] buffers.get(id, ) deltaCPython 的字符串 in-place concat 优化仅在refcount1且 LHS 是 local name 时才生效这里字符串存放在 dict 中又被 reassign优化失效每次都是 O(n) 拷贝 → 总体 O(n²)。实测对 50 KB / 5000 chunk 的回复纯拷贝开销就要 100–300ms。用list.join()是真正的 O(n)。需要补充的一点细节chat()返回的是最后一条完整 AI 消息累加出的文本——中间 AI 消息如 planner 草稿会被丢弃。需要逐 delta 拿每一条消息时应直接使用stream()。三个 id set 为什么不能合并DeerFlowClient.stream()在一次调用生命周期内维护三个set[str]源码位于 client.py_stream_turnseen_ids: set[str] set() # values 路径内部 dedup streamed_ids: set[str] set() # messages → values 跨模式 dedup counted_usage_ids: set[str] set() # usage_metadata 幂等计数乍看像是三份几乎一样的东西实际每个集合守护不同的不变式Set负责的不变式被谁填充被谁查询seen_ids连续两个values快照里的同一条 message 只生成一个messages-tuple事件values 分支每处理一条消息就加入values 分支处理下一条消息前检查streamed_ids一条消息若已通过messages模式 token 级流过values快照到达时不要再合成一次完整messages-tuplemessages 分支每发一个 AI/tool 事件就加入values 分支看到消息时检查counted_usage_ids同一usage_metadata在 messages 末尾 chunk 与 values 快照的 final AIMessage 中各出现一份累计总量只算一次_account_usage()每次接受 usage 就加入_account_usage()每次调用时检查为什么不能只用一个 set关键观察同一个 message id 在这三个 set 里的加入时机完全不同seen_ids永远在 values 快照到达时加入所以它是 values 已处理 的标记。一条只出现在 messages 流里的消息罕见但可能在seen_ids里永远不存在。streamed_ids在 messages 流的第一个有效事件时加入。一条只通过 values 快照到达的非 AI 消息HumanMessage、被 truncate 的 tool 消息在streamed_ids里永远不存在。counted_usage_ids只在看到非空usage_metadata时加入。一条完全没有 usage 的消息tool message、错误消息永远不会进入。关于集合包含关系counted_usage_ids ⊆ (streamed_ids ∪ seen_ids)大致成立但不是严格子集——一条消息可能在 messages 模式流完 text 后、但在最后一个带 usage 的 chunk 之前就被 values snapshot 赶上此时它已在streamed_ids却还不在counted_usage_ids。若把它们合并成一个 dict-of-flags这个微妙的时序依赖会从类型系统里消失退化成注释里的一句话。三个独立的 set 则把不变式显式化了每个 set 名都对应一个可以口头回答的问题。从源码还可以看到messages-tuple分支中带 usage 的元数据型 follow-up约定若某 AI 消息的additional_kwargs在此前的 chunk 事件中从未发送过values分支会补发一条content 为空的 AI 事件仅携带增量additional_kwargs客户端应按 message id 合并并忽略其文本渲染。_unsent_additional_kwargs用sent_additional_kwargs_by_id按 id 追踪已发送键只发送真正的新增键值。端到端一次真实对话的事件时序假设调用client.stream(Count from 1 to 15)LLM 给出one\ntwo\n...\nfifteen88 字符tokenizer 把它拆成约 35 个 BPE chunk。事件到达序列的精简版如下四个关键观察用户看到35 个 messages-tuple 事件跨越约 476ms每个事件携带一个 token delta 和同一个idai-1。最后那个values快照里的AIMessage不会再触发一个完整的messages-tuple事件——因为ai-1 in streamed_ids跳过了合成。end事件里的usage正好等于那一份 cumulative usage不是它的两倍——counted_usage_ids在 messages 末尾 chunk 上已经吸收了 usagevalues 分支对同一 id 的重复访问是 no-op。消费者拿到的content是增量ele只包含 3 个字符不是one\ntwo\n...ele。要得到完整文本必须按id累加——chat()已经帮你做了。为什么这个设计容易出 bug以及测试策略本文档的直接起因是 DeerFlow 历史上的一个真实回归issue #1969DeerFlowClient.stream()原本只订阅[values, custom]漏掉了messages。结果是client.stream(hello)退化为一次性返回视觉上和chat()没区别——流式能力静默消失而 CI 依然全绿。Custom 事件还有一条独立的回归边界get_stream_writer()产生的 chunk不会自动成为astream_events(versionv2)的on_custom_event而 callback dispatch 也不会自动进入stream_modecustom。因此测试必须使用真实最小 LangGraph 同时锁定两种 API断言每个消费者各收到一次且 payload 相同仅 mock 任一函数无法证明协议互操作。这类 bug 有三个结构性原因多协议层命名messages/messages-tuple/ HTTP SSEmessages是同一概念的三个名字。在其中一层出错不会在另外两层报错。多消费者模型Gateway 与 DeerFlowClient 是两套独立实现没有单一的 订阅哪些 mode 的 single source of truth。前者订阅对了不代表后者也订阅对了。mock 测试绕开了真实路径老测试用agent.stream.return_value iter([dict_chunk, ...])喂 values 形状的 dict 来模拟 state 快照。这类输入永远不会进入messagesmode 分支所以即便stream_mode少一个元素CI 依然全绿。防御手段行为断言 真实 chunk shape mock真正的防线是显式断言messagesmode 被订阅 用真实 chunk shape mock。回归测试位于 backend/tests/test_client.py 的test_messages_mode_emits_token_deltas# backend/tests/test_client.py::test_messages_mode_emits_token_deltas agent.stream.return_value iter([ (messages, (AIMessageChunk(contentHel, idai-1), {})), (messages, (AIMessageChunk(contentlo , idai-1), {})), (messages, (AIMessageChunk(contentworld!, idai-1), {})), (values, {messages: [HumanMessage(...), AIMessage(contentHello world!, idai-1)]}), ]) # ... assert [e.data[content] for e in ai_text_events] [Hel, lo , world!] assert len(ai_text_events) 3 # values snapshot must NOT re-synthesize assert messages in agent.stream.call_args.kwargs[stream_mode]为什么这比抽一个共享常量更有效共享常量只能保证用它的人写对字符串但新增消费者的人可能根本不知道常量在哪。行为断言强制任何改动都要穿过实际执行路径——把订阅改回[values, custom]会立刻让assert messages in ...失败。活体信号BPE 子词边界回归的最终验证是让真实 LLM 数 1–15然后确认输出里能看到 tokenizer 的子词切分[5.460s] ele / ven eleven 被拆成两个 token [5.508s] tw / elve twelve 拆两个 [5.568s] th / irteen thirteen 拆两个 [5.623s] four/ teen fourteen 拆两个 [5.677s] f / if / teen fifteen 拆三个子词切分是 tokenizer 的外部事实无法伪造。能看到它就说明数据流逐 chunk地穿过了整条管道没有被任何中间层缓冲成整段。这种活体信号在流式系统里是比单元测试置信度更高的证据。相关源码定位关心什么看这里DeerFlowClient 嵌入式流client.pyDeerFlowClient.streamEmbedded ToolMessage artifact 序列化client.py_tool_message_event/_serialize_messagechat()的 delta 累加器client.pyDeerFlowClient.chatGateway async 流worker.pyrun_agentStream mode 公开集合与命名翻译runtime/stream_modes.pyStreamBridge 抽象与两种 backendruntime/stream_bridge/Bridge 心跳 / 队列 / Redis TTL 配置config/stream_bridge_config.pyHTTP SSE 帧输出services.pysse_consumer/format_sse序列化到 wire 格式runtime/serialization.pyDeerFlow custom 事件双通道 helperutils/custom_events.py飞书渠道的增量卡片更新channels/manager.py_handle_streaming_chatChannels 自带的 delta/cumulative 防御性累加channels/manager.py_merge_stream_textFrontend 支持的 mode 集合与 chat 流裁剪frontend/src/core/api/stream-mode.ts前端useStream消费入口frontend/src/core/threads/hooks.ts核心回归测试backend/tests/test_client.pytest_messages_mode_emits_token_deltas补充一句前端的契约细节前端在 stream-mode.ts 中维护与后端一致的SUPPORTED_RUN_STREAM_MODES集合含values/messages-tuple/custom等并对不支持的 mode 抛错或告警forceChatRunStreamOptions会主动剔除values避免 SDK 惰性消息追踪顺带请求values、在 graph 每一步后重传整段 thread state——这是消费者必须只订阅自己需要的接口这一原则在前端侧的对应实现。【免费下载链接】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），仅供参考