Deer-Flow 的 IM 渠道消息流

Deer-Flow 的 IM 渠道消息流

学习目标

• 理解 Deer-Flow 的 IM 渠道架构：MessageBus、ChannelManager 和平台适配器
• 掌握不同 IM 平台的差异化消息流策略
• 学会分析线程到 thread_id 的映射机制

---

项目实践

渠道架构

核心组件

组件	文件	职责
MessageBus	message_bus.py	异步发布/订阅中心：InboundMessage → 队列 → 分发；OutboundMessage → 回调 → 渠道
ChannelManager	manager.py	核心调度器：通过 client.threads.create() 创建线程、路由命令
store.py	store.py	JSON 文件持久化：channel:chat → thread_id 映射
base.py	base.py	抽象 Channel 基类（start/stop/send 生命周期）
service.py	service.py	管理所有配置渠道的生命周期

差异化消息流策略

不同 IM 平台使用不同的消息流策略：

渠道	策略	原因
Slack	runs.wait() — 阻塞等待完整响应	Slack API 不支持增量更新
Telegram	runs.wait() — 阻塞等待完整响应	简单模型，一次性发送
飞书	runs.stream() — 渐进式卡片更新	飞书支持卡片原地补丁
钉钉	runs.stream() — AI Card 流式打字机	可选，需要配置 card_template_id

飞书渐进式更新

飞书的特殊流程：

1. 发送一个初始卡片（“正在思考…”）
2. 通过 runs.stream() 接收 AI 回复增量
3. 每次收到增量 → 补丁同一卡片（config.update_multi=true）
4. 最终响应时标记卡片为完成

钉钉 AI Card 流式打字机

钉钉可选启用 AI Card 流式效果：

1. 创建初始卡片
2. 通过 PUT /v1.0/card/streaming 流式更新
3. 完成时调用 PUT /v1.0/card/streaming/finish
4. 如果卡片创建或流式失败，降级为 sampleMarkdown

线程映射

store.py 使用 JSON 文件持久化渠道对话到 Deer-Flow 线程的映射：

• 根对话：channel:chat → thread_id
• 线程对话：channel:chat:topic → thread_id

这使得 IM 平台的对话上下文能够跨重启保持。

---

问题与规避

渠道认证过期

IM 平台的 Bot Token 可能过期或失效。

规避：service.py 在启动时检测渠道配置，失败时跳过该渠道而非阻止整个服务启动。

Docker 部署的 localhost 问题

Docker Compose 中 IM 渠道在 gateway 容器内运行，localhost 指向容器自身。

规避：使用容器服务名（如 http://gateway:8001/api）替代 localhost。

---

参考来源

• Deer-Flow CLAUDE.md — IM Channels System 章节
• app/channels/ — IM 渠道源码