DeerFlow Telegram 通道流式输出实现：原地编辑占位消息的完整工程方案-平芜编程栈

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_final=True的完整结果收尾。技术栈为 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_final=True的完整结果（_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_retries=0时抛RuntimeError）必须保持绿。

一个值得注意的有意行为变化：命令回复（如/help）与错误回复现在会编辑"Working on it..." 占位消息，而不是再发一条新消息（因为键匹配且is_final=True）。这是 UX 改进，并有专门测试覆盖。

测试运行方式统一为（在backend/目录下）：

PYTHONPATH=. uv run pytest tests/test_channels.py -v

三、Task 1：能力开关——让 Telegram 上报流式能力

涉及文件：backend/app/channels/manager.py（CHANNEL_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(bus=bus, config={"bot_token": "test-token"}) assert ch.supports_streaming is True assert CHANNEL_CAPABILITIES["telegram"]["supports_streaming"] is True

Step 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-L142，telegram已为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_final=False的OutboundMessage，发布时带 " ▉" 光标后缀，并在 manager 层做节流（时间间隔与新增字符数的 OR 条件，设计规格注明间隔为 0.35s）；
流结束（或出错）时在finally块中必发一条is_final=True的完整结果（含 artifacts/attachments）。

没有其他 manager 改动——这正是方案 A 的价值。

四、Task 2：流式状态基础设施与占位消息登记

涉及文件：backend/app/channels/telegram.py（常量、__init__、helper 方法、_send_running_reply）。

Step 1：失败测试——验证_send_running_reply发送占位消息后，会把消息登记进_stream_messages：

def test_running_reply_registers_stream_placeholder(self): async def go(): bus = MessageBus() ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"}) mock_app = MagicMock() mock_bot = AsyncMock() sent = MagicMock() sent.message_id = 777 mock_bot.send_message = AsyncMock(return_value=sent) 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_id=12345, text="Working on it...", reply_to_message_id=42, ) _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_id=int(chat_id), text="Working on it...", reply_to_message_id=reply_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()整体替换为"分发版 + 抽取的 helper"：

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 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_final=False更新（间隔 2 秒）应编辑同一message_id，且bot.sent里只有占位消息一条；
test_stream_updates_throttled_within_interval：1 秒窗口内的更新被丢弃，跨过窗口的更新生效，edited文本为["a", "abc"]；
test_stream_update_without_placeholder_sends_new_message：占位消息缺失（如发送失败）时，首条流式更新退化为send_message新建并登记；
test_stream_update_truncates_long_text：5000 字符文本被截断为 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_id=chat_id, text=display) 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_id=chat_id, message_id=state["message_id"], text=display) 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_id=chat_id, text=display) 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_final=True消息兜底；
无变化跳过：文本与last_text相同 → 跳过，避免触发message is not modified错误；
4096 字符截断：超过TELEGRAM_MAX_MESSAGE_LENGTH（4096）的文本截到 4095 字符并追加…后再编辑；
三种异常分流：
- BadRequest: message is not modified→ 静默忽略，仅同步last_text（final 文本与最后一帧相同时必然出现）；
- RetryAfter(429) →丢弃本次更新，不重试不等待（下一帧自带全量文本）；
- 其他编辑失败（如消息被用户删除）→ 回退send_message发新消息并更新登记的message_id，保证流式状态始终指向一条真实存在的消息。

设计规格把这套策略概括为：1s channel 侧节流 + 429 丢帧，是飞书 0.35s 发布间隔在 Telegram 上的等价物；最坏情况是中间帧丢失，最终完整性由is_final=True保证。

七、Task 5：最终消息——最后一次编辑、超长分段补发、状态清理

涉及文件：telegram.py（更新send的 final 分支、新增_finalize_stream_message）。

新增 4 个测试：test_final_message_edits_stream_message_and_clears_state（final 编辑同一条流式消息、清理状态、_last_bot_message指向该消息）、test_final_message_splits_long_text（4096+100 字符：首段编辑、余段补发、_last_bot_message指向最后一段）、test_final_message_not_modified_error_is_ignored（not-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_id=chat_id, message_id=state["message_id"], text=chunks[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_id=chat_id, message_id=state["message_id"], text=chunks[0]) else: logger.warning("[Telegram] final edit failed in chat=%s, sending new message: %s", chat_id, exc) sent = await bot.send_message(chat_id=chat_id, text=chunks[0]) last_message_id = sent.message_id for chunk in chunks[1:]: sent = await bot.send_message(chat_id=chat_id, text=chunk) 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.md（IM 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 updates"；
Message 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 流式描述：占位消息登记为流式目标；非最终更新原地editMessageText（1s 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_id`	`test_stream_updates_edit_placeholder_in_place`
1s 窗口内更新被节流丢弃	`test_stream_updates_throttled_within_interval`
占位缺失时首帧退化为`send_message`	`test_stream_update_without_placeholder_sends_new_message`
>4096 截断并以`…`结尾	`test_stream_update_truncates_long_text`
429 丢帧不抛错不补发	`test_stream_update_retry_after_is_dropped`
final 编辑 + 状态清理 + 线程化指针更新	`test_final_message_edits_stream_message_and_clears_state`
final 超长：首段编辑 + 分段补发	`test_final_message_splits_long_text`
not-modified 静默忽略	`test_final_message_not_modified_error_is_ignored`
无流式状态直发（回归保护）	`test_final_without_stream_state_sends_plain_message`
直发重试语义不回归	`TestTelegramSendRetry`（test_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_final=False/True的消息协议保持不变，所有 Telegram 特有的节流与限速策略封装在TelegramChannel内部，其他通道（Feishu/WeCom 等）完全不受影响；
占位消息复用：用户全程只看到一条消息从 "Working on it..." 演变为最终答案，避免了"占位 + 流式 + 最终"三条消息的体验割裂，命令/错误回复也顺势改为编辑占位消息；
全量帧 + final 兜底：每条流式更新都携带全量累积文本，因此节流丢帧、429 限速、中间帧丢失都不影响正确性；is_final=True必达且超长文本自动分段补发，完整性由 manager 与通道双层保证。

如需继续深入，建议按计划文档 → 设计规格 → 通道实现 → manager 流式分发 → 测试用例的顺序阅读，即可完整复现从设计到落地的全链路。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

DeerFlow Telegram 通道流式输出实现：原地编辑占位消息的完整工程方案