DeerFlow Telegram 通道流式输出实现原地编辑占位消息的完整工程方案【免费下载链接】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本文基于 DeerFlow 仓库中的实施计划文档docs/superpowers/plans/2026-06-12-telegram-streaming.md及其设计规格docs/superpowers/specs/2026-06-12-telegram-streaming-design.md展开完整讲解如何把 Telegram 通道从等 agent 跑完再一次性发送改造为边生成边原地编辑同一条消息的流式输出。读完后你会掌握 DeerFlow 消息网关中CHANNEL_CAPABILITIES能力开关、is_final增量/最终消息协议、流式状态登记、编辑节流与 Telegram 限速容错等一套可复用的 IM 通道流式输出工程方案并能对照 测试文件 中的TestTelegramStreaming用例验证每个行为。一、背景与目标Telegram 通道在改造前完全不流式ChannelManager._handle_chat()走client.runs.wait()阻塞路径agent 跑完后一次性send_message发出最终文本。用户先看到 Working on it... 占位回复然后长时间无任何反馈。改造目标是让 Telegram 与飞书行为一致——通过编辑同一条消息的方式流式展示所有 AI 文本增量manager 现有流式管线产出的累积文本最终以is_finalTrue的完整结果收尾。技术栈为 Python 3.12、python-telegram-bot测试中全部 mock、pytest。设计规格中给出了明确的方案选型方案 A采纳channel 侧自适配。只改 telegram.py CHANNEL_CAPABILITIES一行Telegram 通道自己做编辑节流与限速容错不触碰飞书/微信/钉钉共享的 manager 流式代码路径。方案 B否决manager 支持 per-channelstream_min_interval节流。语义更统一但改动共享路径回归面大。整个方案落在分支feat/telegram-streaming上按 6 个 Task 拆解实施。二、架构总览与既有基础事实计划文档开篇给出了对照代码库验证过的关键既有事实Key existing facts这是方案可行性的基础。这些事实解释了为什么只需改动 Telegram 通道一侧即可打通整条链路OutboundMessage.is_final默认为True见 message_bus.py 中is_final: bool True字段定义因此错误回复、命令回复等直发路径天然保持 final 语义无需改动ChannelManager._channel_supports_streaming()当前位于 manager.py优先读取存活通道实例的supports_streaming属性找不到实例时才回退到CHANNEL_CAPABILITIES表——所以两处都要更新流式管线在流结束含出错时必定发布一条is_finalTrue的完整结果_handle_streaming_chat()的finally块保证这是中间帧可以丢、最终完整性不丢的兜底依据_send_running_reply()在 inbound 消息发布之前被await当前实现见 telegram.py 及调用链_process_incoming_with_reply()因此占位消息在任何 outbound 到达之前必然已经存在outbound 的thread_ts等于 inbound 的thread_ts而 Telegram 通道把它设置为用户消息 id因此流式键f{chat_id}:{thread_ts}与占位消息登记时使用的键完全一致既有测试tests/test_channels.py::TestTelegramSendRetry发送重试语义、_max_retries0时抛RuntimeError必须保持绿。一个值得注意的有意行为变化命令回复如/help与错误回复现在会编辑Working on it... 占位消息而不是再发一条新消息因为键匹配且is_finalTrue。这是 UX 改进并有专门测试覆盖。测试运行方式统一为在backend/目录下PYTHONPATH. uv run pytest tests/test_channels.py -v三、Task 1能力开关——让 Telegram 上报流式能力涉及文件backend/app/channels/manager.pyCHANNEL_CAPABILITIES表、backend/app/channels/telegram.py新增supports_streaming属性、test_channels.py新增TestTelegramStreaming测试类。Step 1先写失败测试追加到测试文件末尾class TestTelegramStreaming: def test_telegram_reports_streaming_support(self): from app.channels.manager import CHANNEL_CAPABILITIES from app.channels.telegram import TelegramChannel bus MessageBus() ch TelegramChannel(busbus, config{bot_token: test-token}) assert ch.supports_streaming is True assert CHANNEL_CAPABILITIES[telegram][supports_streaming] is TrueStep 2运行PYTHONPATH. uv run pytest tests/test_channels.py::TestTelegramStreaming::test_telegram_reports_streaming_support -v预期失败assert False is True基类属性返回False。Step 3实现。在CHANNEL_CAPABILITIES中将telegram: {supports_streaming: False}改为True并在TelegramChannel.__init__之后、async def start之前添加属性property def supports_streaming(self) - bool: return True当前仓库中该表位于 manager.py#L132-L142telegram已为True实例属性位于 telegram.py#L81-L83。两处同时生效的原因正是第二节第 2 条事实manager 优先问实例CHANNEL_CAPABILITIES只是实例未注册时的回退表。Step 4再跑测试预期通过。开关打开后manager 会自动为 Telegram 走_handle_streaming_chat()分支当前实现见 manager.py#L2309该分支会通过client.runs.stream([messages-tuple, values])持续消费 agent 输出把累积文本发布为多条is_finalFalse的OutboundMessage发布时带 ▉ 光标后缀并在 manager 层做节流时间间隔与新增字符数的 OR 条件设计规格注明间隔为 0.35s流结束或出错时在finally块中必发一条is_finalTrue的完整结果含 artifacts/attachments。没有其他 manager 改动——这正是方案 A 的价值。四、Task 2流式状态基础设施与占位消息登记涉及文件backend/app/channels/telegram.py常量、__init__、helper 方法、_send_running_reply。Step 1失败测试——验证_send_running_reply发送占位消息后会把消息登记进_stream_messagesdef test_running_reply_registers_stream_placeholder(self): async def go(): bus MessageBus() ch TelegramChannel(busbus, config{bot_token: test-token}) mock_app MagicMock() mock_bot AsyncMock() sent MagicMock() sent.message_id 777 mock_bot.send_message AsyncMock(return_valuesent) mock_app.bot mock_bot ch._application mock_app await ch._send_running_reply(12345, 42) state ch._stream_messages[12345:42] assert state[message_id] 777 assert state[last_text] Working on it... mock_bot.send_message.assert_awaited_once_with( chat_id12345, textWorking on it..., reply_to_message_id42, ) _run(go())Step 3实现分四小步a) 顶部导入time并在logger ...之后添加模块常量TELEGRAM_MAX_MESSAGE_LENGTH 4096 STREAM_EDIT_MIN_INTERVAL_SECONDS 1.0 # Indirection so tests can patch the clock without touching the global time module. _monotonic time.monotonic其中_monotonic的间接引用是刻意为之测试用monkeypatch.setattr(app.channels.telegram._monotonic, ...)注入假时钟无需触碰全局time模块。b) 在__init__中新增流式状态字典# stream_key (chat_id:thread_ts) - state of the in-flight streamed # bot message being edited in place: {message_id, last_edit_at, last_text} self._stream_messages: dict[str, dict[str, Any]] {}c) 在 helpers 区添加一组静态方法staticmethod def _stream_key(chat_id: str, thread_ts: str | None) - str: return f{chat_id}:{thread_ts or } staticmethod def _is_retry_after(exc: Exception) - bool: return getattr(exc, retry_after, None) is not None staticmethod def _retry_after_seconds(exc: Exception) - float: value getattr(exc, retry_after, 0) if hasattr(value, total_seconds): return float(value.total_seconds()) return float(value) staticmethod def _is_not_modified(exc: Exception) - bool: return message is not modified in str(exc).lower() staticmethod def _split_message(text: str) - list[str]: return [text[i : i TELEGRAM_MAX_MESSAGE_LENGTH] for i in range(0, len(text), TELEGRAM_MAX_MESSAGE_LENGTH)] or [text]d) 重写_send_running_reply把占位消息登记为流式目标async def _send_running_reply(self, chat_id: str, reply_to_message_id: int) - None: Send a Working on it... reply and register it as the stream target. if not self._application: return try: bot self._application.bot sent await bot.send_message( chat_idint(chat_id), textWorking on it..., reply_to_message_idreply_to_message_id, ) self._stream_messages[self._stream_key(chat_id, str(reply_to_message_id))] { message_id: sent.message_id, last_edit_at: 0.0, last_text: Working on it..., } logger.info([Telegram] Working on it... reply sent in chat%s, chat_id) except Exception: logger.exception([Telegram] failed to send running reply in chat%s, chat_id)last_edit_at初值设为0.0意味着第一条流式更新无需等待节流窗口即可编辑占位消息。占位消息复用是设计规格中占位消息复用一节的核心不额外发一条流式起始消息用户看到的始终是同一条消息从 Working on it... 变成最终答案。当前仓库中占位消息登记改经_register_stream_message()统一收口见 telegram.py#L633-L641除写入三个键外还负责容量上限保护后文详述。五、Task 3重构send()——抽出_send_new_message纯重构无行为变化涉及文件backend/app/channels/telegram.py的send()既有TestTelegramSendRetry必须保持绿。把原send()整体替换为分发版 抽取的 helperasync def send(self, msg: OutboundMessage, *, _max_retries: int 3) - None: if not self._application: return try: chat_id int(msg.chat_id) except (ValueError, TypeError): logger.error(Invalid Telegram chat_id: %s, msg.chat_id) return await self._send_new_message(chat_id, msg.chat_id, msg.text, _max_retries_max_retries) async def _send_new_message(self, chat_id: int, chat_key: str, text: str, *, _max_retries: int 3) - int | None: Send a fresh message with retry/backoff. Returns the sent message_id. kwargs: dict[str, Any] {chat_id: chat_id, text: text} # Reply to the last bot message in this chat for threading reply_to self._last_bot_message.get(chat_key) if reply_to: kwargs[reply_to_message_id] reply_to bot self._application.bot last_exc: Exception | None None for attempt in range(_max_retries): try: sent await bot.send_message(**kwargs) self._last_bot_message[chat_key] sent.message_id return sent.message_id except Exception as exc: last_exc exc if attempt _max_retries - 1: delay 2**attempt # 1s, 2s logger.warning( [Telegram] send failed (attempt %d/%d), retrying in %ds: %s, attempt 1, _max_retries, delay, exc, ) await asyncio.sleep(delay) logger.error([Telegram] send failed after %d attempts: %s, _max_retries, last_exc) if last_exc is None: raise RuntimeError(Telegram send failed without an exception from any attempt) raise last_exc这一步的意义把带 1s/2s 指数退避重试的新消息发送独立出来返回发送后的message_id。Task 4/5 的流式回退路径编辑失败 → 发新消息将直接复用该 helper保证回退消息也享有与常规发送一致的重试语义。_last_bot_message继续维护线程化回复每条 bot 消息 reply-to 上一条 bot 消息所需的状态。六、Task 4非最终流式更新——原地编辑 节流 截断 限速容错这是整个方案的行为核心。涉及文件telegram.py改造send并新增_send_stream_update。失败测试使用共享 fake-bot 工厂记录所有sent/edited调用核心用例有 5 个test_stream_updates_edit_placeholder_in_place两条is_finalFalse更新间隔 2 秒应编辑同一message_id且bot.sent里只有占位消息一条test_stream_updates_throttled_within_interval1 秒窗口内的更新被丢弃跨过窗口的更新生效edited文本为[a, abc]test_stream_update_without_placeholder_sends_new_message占位消息缺失如发送失败时首条流式更新退化为send_message新建并登记test_stream_update_truncates_long_text5000 字符文本被截断为 4096 字符且以…结尾test_stream_update_retry_after_is_dropped编辑抛出带retry_after的异常429 Flood control时不抛错、不补发新消息静默丢帧。实现send()按is_final分流新增_send_stream_update()async def send(self, msg: OutboundMessage, *, _max_retries: int 3) - None: if not self._application: return try: chat_id int(msg.chat_id) except (ValueError, TypeError): logger.error(Invalid Telegram chat_id: %s, msg.chat_id) return key self._stream_key(msg.chat_id, msg.thread_ts) if not msg.is_final: await self._send_stream_update(chat_id, key, msg.text) return await self._send_new_message(chat_id, msg.chat_id, msg.text, _max_retries_max_retries) async def _send_stream_update(self, chat_id: int, key: str, text: str) - None: Edit the in-flight streamed message with accumulated text. Updates are best-effort: throttled, rate-limit drops are silent. The manager always publishes a final message afterwards, which guarantees delivery of the complete text. if not text: return display text if len(display) TELEGRAM_MAX_MESSAGE_LENGTH: display display[: TELEGRAM_MAX_MESSAGE_LENGTH - 1] … bot self._application.bot state self._stream_messages.get(key) if state is None: try: sent await bot.send_message(chat_idchat_id, textdisplay) except Exception: logger.exception([Telegram] failed to start stream message in chat%s, chat_id) return self._stream_messages[key] { message_id: sent.message_id, last_edit_at: _monotonic(), last_text: display, } return now _monotonic() if now - state[last_edit_at] STREAM_EDIT_MIN_INTERVAL_SECONDS: return if display state[last_text]: return try: await bot.edit_message_text(chat_idchat_id, message_idstate[message_id], textdisplay) except Exception as exc: if self._is_not_modified(exc): state[last_text] display return if self._is_retry_after(exc): logger.debug([Telegram] stream edit rate-limited in chat%s, dropping update, chat_id) return logger.warning([Telegram] stream edit failed in chat%s, sending new message: %s, chat_id, exc) try: sent await bot.send_message(chat_idchat_id, textdisplay) except Exception: logger.exception([Telegram] failed to send fallback stream message in chat%s, chat_id) return state[message_id] sent.message_id state[last_edit_at] _monotonic() state[last_text] display逐条拆解非最终更新的处理规则节流距同 key 上次成功编辑不足 1.0 秒 → 直接丢弃本次更新。这是安全的因为每条更新都是全量累积文本丢掉的只是中间帧最终完整性由 manager 必发的is_finalTrue消息兜底无变化跳过文本与last_text相同 → 跳过避免触发message is not modified错误4096 字符截断超过TELEGRAM_MAX_MESSAGE_LENGTH4096的文本截到 4095 字符并追加…后再编辑三种异常分流BadRequest: message is not modified→ 静默忽略仅同步last_textfinal 文本与最后一帧相同时必然出现RetryAfter(429) →丢弃本次更新不重试不等待下一帧自带全量文本其他编辑失败如消息被用户删除→ 回退send_message发新消息并更新登记的message_id保证流式状态始终指向一条真实存在的消息。设计规格把这套策略概括为1s channel 侧节流 429 丢帧是飞书 0.35s 发布间隔在 Telegram 上的等价物最坏情况是中间帧丢失最终完整性由is_finalTrue保证。七、Task 5最终消息——最后一次编辑、超长分段补发、状态清理涉及文件telegram.py更新send的 final 分支、新增_finalize_stream_message。新增 4 个测试test_final_message_edits_stream_message_and_clears_statefinal 编辑同一条流式消息、清理状态、_last_bot_message指向该消息、test_final_message_splits_long_text4096100 字符首段编辑、余段补发、_last_bot_message指向最后一段、test_final_message_not_modified_error_is_ignorednot-modified 静默忽略、test_final_without_stream_state_sends_plain_message无流式状态时直发回归保护。实现# send() 的 final 分支 key self._stream_key(msg.chat_id, msg.thread_ts) if not msg.is_final: await self._send_stream_update(chat_id, key, msg.text) return state self._stream_messages.pop(key, None) if state is not None: await self._finalize_stream_message(chat_id, msg.chat_id, state, msg.text) return await self._send_new_message(chat_id, msg.chat_id, msg.text, _max_retries_max_retries) async def _finalize_stream_message(self, chat_id: int, chat_key: str, state: dict[str, Any], text: str) - None: Apply the final text: edit the streamed message, splitting overflow into follow-ups. bot self._application.bot chunks self._split_message(text or ) last_message_id state[message_id] if chunks[0] ! state[last_text]: try: await bot.edit_message_text(chat_idchat_id, message_idstate[message_id], textchunks[0]) except Exception as exc: if self._is_not_modified(exc): pass elif self._is_retry_after(exc): await asyncio.sleep(self._retry_after_seconds(exc)) await bot.edit_message_text(chat_idchat_id, message_idstate[message_id], textchunks[0]) else: logger.warning([Telegram] final edit failed in chat%s, sending new message: %s, chat_id, exc) sent await bot.send_message(chat_idchat_id, textchunks[0]) last_message_id sent.message_id for chunk in chunks[1:]: sent await bot.send_message(chat_idchat_id, textchunk) last_message_id sent.message_id self._last_bot_message[chat_key] last_message_id最终消息的处理规则state self._stream_messages.pop(key, None)pop同时完成取出 清理保证每轮对话的流式状态不泄漏文本 ≤ 4096对登记的流式消息做最后一次edit_message_text若首段与last_text相同则跳过编辑避免无谓请求文本 4096第一段4096 内编辑流式消息剩余部分按 4096 分段send_message补发429 在 final 路径的策略与中间帧不同final 必须送达因此按retry_after等待后重试一次编辑更新_last_bot_message[chat_id]指向最后一条消息 id保持现有 threaded-reply 行为后续附件send_file仍能正确 reply-to无登记状态时退回直发走标准_send_new_message含 3 次重试。注意命令回复与_send_error错误回复带有匹配的thread_ts且占位消息已登记因此它们同样走编辑占位消息路径——即第二节提到的有意行为变化。验证顺序为先跑TestTelegramStreaming与TestTelegramSendRetry再跑整个 test_channels.py含 Feishu/WeCom/manager 用例——其代码路径未动以及tests/test_telegram_channel_connections.py。八、Task 6文档同步与全量验证涉及文件backend/CLAUDE.mdIM Channels 章节、README.md仅当其中提及 Telegram 非流式时。文档同步要点backend/CLAUDE.md的 IM Channels System 小节manager 组件描述由keeps Slack/Telegram onclient.runs.wait()改为keeps Slack/Discord onclient.runs.wait(), and usesclient.runs.stream([messages-tuple, values])for Feishu/Telegram incremental outbound updatesMessage Flow 条目由5. Feishu chat:runs.stream()… 6. Slack/Telegram chat:runs.wait()…改为5. Feishu/Telegram chat:runs.stream()… 6. Slack/Discord chat:runs.wait()…在飞书 card-patching 条目后新增一条 Telegram 流式描述占位消息登记为流式目标非最终更新原地editMessageText1s channel 侧节流、4096 字符截断、429 丢帧final 更新做最后一次编辑并把 4096 的文本分段补发。另外用grep -rn Telegram README.md docs/ --include*.md -l | head检查其他文档是否声明了 Telegram 非流式若有则同步更新。最后在backend/下执行make test全量通过与make lint干净。九、自审笔记规格覆盖、类型一致性与已知取舍计划文档末尾的 Self-Review Notes 值得保留为工程复盘规格覆盖能力开关Task 1、占位消息复用Task 2、节流/截断/429 丢帧/回退新消息Task 4、final 编辑/分段/清理/not-modified/RetryAfter 等待Task 5、直发回归保护Task 5 的test_final_without_stream_state_sends_plain_message 既有TestTelegramSendRetry、文档Task 6。设计规格中列出的 6 项测试要求全部映射到具体测试。类型一致性_stream_messages: dict[str, dict[str, Any]]的三个键message_id/last_edit_at/last_text在 Task 2、4、5 中用法一致_send_new_message(chat_id: int, chat_key: str, text: str)签名在 Task 3 与 5 之间一致。已知取舍final 路径的回退send_message当时没有重试循环单次尝试异常向上抛到_on_outbound记日志并跳过附件上传——与当时send()失败契约一致。十、仓库当前实现计划落地后的进一步演进从源码结构看当前仓库中的 telegram.py 已完整实现该计划并在其后叠加了几处演进对照阅读可以验证方案与实现的差异群聊独立节流新增STREAM_EDIT_GROUP_MIN_INTERVAL_SECONDS 3.0。Telegram 对群chat_id为负数限速约 20 条/分钟因此_send_stream_update()中按chat_id 0选择 3s 还是 1s 的最小编辑间隔见 telegram.py#L34-L43 与 #L249-L252。流式状态容量上限MAX_TRACKED_STREAM_MESSAGES 256。条目正常情况下随 final 更新清理该上限只是防止final 永远不到达时的状态泄漏登记统一收口到_register_stream_message()telegram.py#L633-L641超限前弹出最早条目。final 编辑抽为_edit_final_chunk()返回bool表示编辑是否生效message is not modified 视为生效编辑彻底失败如消息被删时回退到带标准重试策略的_send_new_message()补发首段而不是像计划初版那样裸调send_message见 telegram.py#L276-L309。这恰好补齐了自审笔记中的已知取舍。Rich Messages 扩展_can_send_rich()/_edit_rich_message()/_send_new_rich_message()支持在配置rich_messages开启时把流式预览替换为 Bot API 10.1 Rich Message上限 32768 字符TELEGRAM_MAX_RICH_MESSAGE_LENGTH被BadRequest/EndPointNotFound拒绝时回退纯文本见 telegram.py#L311-L354。这属于计划之后的扩展能力不属于本计划范围。发送路径统一重试_send_new_message()的退避重试抽为通用_send_with_retry()流式回退、final 分段补发、富文本发送共用同一重试设施。十一、测试落点与验证清单所有流式行为测试集中在 test_channels.py#L9359 的TestTelegramStreaming类采用 fake-bot 工厂模式SimpleNamespace记录sent/edited调用序列 monkeypatch注入_monotonic假时钟无需真实 Telegram 凭据。验证矩阵如下行为测试用例能力开关实例属性 能力表test_telegram_reports_streaming_support占位消息登记为流式目标test_running_reply_registers_stream_placeholder多条增量编辑同一message_idtest_stream_updates_edit_placeholder_in_place1s 窗口内更新被节流丢弃test_stream_updates_throttled_within_interval占位缺失时首帧退化为send_messagetest_stream_update_without_placeholder_sends_new_message4096 截断并以…结尾test_stream_update_truncates_long_text429 丢帧不抛错不补发test_stream_update_retry_after_is_droppedfinal 编辑 状态清理 线程化指针更新test_final_message_edits_stream_message_and_clears_statefinal 超长首段编辑 分段补发test_final_message_splits_long_textnot-modified 静默忽略test_final_message_not_modified_error_is_ignored无流式状态直发回归保护test_final_without_stream_state_sends_plain_message直发重试语义不回归TestTelegramSendRetrytest_channels.py#L8281常用验证命令均在backend/目录下执行PYTHONPATH. uv run pytest tests/test_channels.py::TestTelegramStreaming -v PYTHONPATH. uv run pytest tests/test_channels.py::TestTelegramStreaming tests/test_channels.py::TestTelegramSendRetry -v PYTHONPATH. uv run pytest tests/test_channels.py -v PYTHONPATH. uv run pytest tests/test_telegram_channel_connections.py -v make test make lint十二、小结这套方案的核心设计取舍可以概括为三点能力开关 通道自适配manager 流式管线对 Telegram 零改动is_finalFalse/True的消息协议保持不变所有 Telegram 特有的节流与限速策略封装在TelegramChannel内部其他通道Feishu/WeCom 等完全不受影响占位消息复用用户全程只看到一条消息从 Working on it... 演变为最终答案避免了占位 流式 最终三条消息的体验割裂命令/错误回复也顺势改为编辑占位消息全量帧 final 兜底每条流式更新都携带全量累积文本因此节流丢帧、429 限速、中间帧丢失都不影响正确性is_finalTrue必达且超长文本自动分段补发完整性由 manager 与通道双层保证。如需继续深入建议按 计划文档 → 设计规格 → 通道实现 → manager 流式分发 → 测试用例 的顺序阅读即可完整复现从设计到落地的全链路。【免费下载链接】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），仅供参考