AutoGPT CoPilot Chat Bridge：连接 Discord、Slack、Telegram 的多平台聊天机器人架构与实战

本篇基于 AutoGPT 平台 copilot/bot 模块的官方说明文档与源码实现，系统讲解 CoPilot Chat Bridge 这一多平台聊天机器人的运行机制：它如何以“出站契约 + 入站两分法”的适配器模型桥接 AutoGPT 与 Discord、Slack、Telegram，如何在生产环境中完成启动、配置、账号关联（Linking）、消息批处理与会话线程管理，以及如何按仓库既有规范新增一个聊天平台适配器。读完本文，你能独立部署该服务、配置各平台凭据、理解其消息处理编排流程，并可照此流程扩展新的聊天平台。

一、CoPilot Bot 是什么

CoPilot Bot（内部服务名为 CoPilotChatBridge）是一个多平台聊天机器人，将 AutoGPT 桥接到 Discord、Slack 与 Telegram（Teams/WhatsApp 在规划中）。用户在聊天平台中 @ 机器人或私聊它，即可与 AutoGPT 的 CoPilot 会话进行对话；反过来，AutoPilot 也可以主动向频道或用户 DM 推送消息。

服务入口由 main.py 提供，核心是 backend/app.py 中的 run_processes 驱动一个 AppService 实例：

# usage: poetry run copilot-bot  或  python -m backend.copilot.bot
from backend.app import run_processes
from .app import CoPilotChatBridge

def main():
    """Run the CoPilot Chat Bridge service."""
    run_processes(CoPilotChatBridge())

服务主体 app.py 中的 CoPilotChatBridge 继承自 AppService，职责有三：

1. 运行已配置的 socket 适配器（当前仅 Discord Gateway），并构建 webhook 适配器的出站部分；
2. 暴露出站 RPC——通过 @expose 装饰器对外提供 list_channels、send_message_to_channel、send_dm_to_user、create_thread_in_channel 四个接口，供其他服务（如 copilot tool）以用户名义向聊天平台主动发消息；
3. 健康检查——_adapters_healthy 标志位会在适配器任务退出时被翻转为 False，health_check 随之抛出 UnhealthyServiceError，让编排器重启 Pod。值得注意的是，即使没有任何 socket 适配器配置（例如部署时没有设置 Discord token），服务也会进入空闲循环（每小时休眠）以保持存活，以便健康检查和运行时重配。

二、启动方式与本地开发

官方文档给出的标准运行方式有两种：

# 方式一：作为独立服务运行
poetry run copilot-bot

# 方式二：随整个平台自动启动（设置了 AUTOPILOT_BOT_DISCORD_TOKEN 时会自动拉起 bot）
poetry run app

Docker 本地开发：为什么需要 platform_linking_manager

Bot 的 link/unlink 流程通过集群内部 RPC 调用 platform_linking_manager。该服务在 dev/prod 中是独立 Pod，但本地是可选的——通过 Compose 的 bot profile 显式启用，避免拖慢普通的 docker compose up：

# 启动 linking manager（与现有栈并存）
docker compose up -d platform_linking_manager

# 或按 profile 一键拉起（同时带上其他打了 bot profile 标签的服务）
docker compose --profile bot up -d

文档特别强调了一个典型故障现象：如果 platform_linking_manager 没有运行，/setup 等签发 token 的流程会因后端尝试连接一个不存在的 host 而失败，抛出 httpx.ConnectError。这是本地联调时最常见的报错来源。

三、环境变量配置

完整清单带注释见 backend/.env.default（第 260 行起为各平台 token 区段）。最小化配置如下表：

变量	作用
AUTOPILOT_BOT_DISCORD_TOKEN	Discord bot token，设置后即启用 Discord（socket）适配器
AUTOPILOT_BOT_SLACK_TOKEN + AUTOPILOT_BOT_SLACK_SIGNING_SECRET	Slack bot token + 签名密钥，两者都设置才会把 Slack（webhook / Events API）适配器挂载到主后端 API；.env.default 中另有 AUTOPILOT_BOT_SLACK_CLIENT_ID / AUTOPILOT_BOT_SLACK_CLIENT_SECRET 用于 OAuth 多工作区授权
AUTOPILOT_BOT_TELEGRAM_TOKEN + AUTOPILOT_BOT_TELEGRAM_WEBHOOK_SECRET	Telegram BotFather token + webhook secret，两者都设置才会挂载 Telegram（webhook / Bot API）适配器；之后需用 setWebhook 注册一次 webhook（见 .env.default 中的 curl 示例），并通过 BotFather 的 /setprivacy 关闭群隐私模式，否则机器人看不到群内 @ 提及
FRONTEND_BASE_URL	关联确认页所用的前端 base URL（与后端其余部分共享）
REDIS_HOST / REDIS_PORT	会话与线程订阅状态、copilot 流订阅（继承自共享后端配置）
PLATFORMLINKINGMANAGER_HOST	PlatformLinkingManager 服务 Pod 的 DNS 名（集群内部 RPC）

从 webhook_routes.py 的 build_webhook_adapters 源码可以精确看到“哪些凭据组合才启用哪个适配器”的判定逻辑：

# Slack：signing secret 总是必需（入站验签），凭据二选一
# —— OAuth 应用凭据（多工作区 “Add to Slack”）或单个静态 token（单工作区兼容）
has_credentials = (
    slack_config.get_client_id() and slack_config.get_client_secret()
) or slack_config.get_bot_token()
if slack_config.get_signing_secret() and has_credentials:
    adapters.append(SlackAdapter(api))

# Telegram：一个 BotFather token 服务所有聊天，webhook secret 用于
# 入站鉴权，因此两者都必须设置
if telegram_config.get_bot_token() and telegram_config.get_webhook_secret():
    adapters.append(TelegramAdapter(api))

而 app.py 中的 _build_socket_adapters 目前只处理 Discord：

if discord_config.get_bot_token():
    adapters.append(DiscordAdapter(api))
    logger.info("Discord adapter enabled")

两个工厂函数刻意把“平台名”限定在这两处之内，这是后文“局部性规则”的落点。

四、目录结构与模块职责

官方文档给出的目录树与各文件职责（结合源码印证）：

bot/
├── app.py              # CoPilotChatBridge(AppService)，适配器工厂，出站 @expose RPC
├── config.py           # 共享（平台无关）配置
├── handler.py          # 编排器：路由、关联、附件摄取、批处理
├── turn_stream.py      # 单批次 turn 的流式发送：分块、产物、线程改名
├── prompt.py           # Prompt + 线程命名组装
├── attachments.py      # 附件上传 + 失败说明
├── command_core.py     # 共享的 /setup + /unlink 策略（由各适配器渲染）
├── bot_backend.py      # PlatformLinkingManagerClient + stream_registry 的薄门面
├── text.py             # 文本切分 + 批量格式化
├── threads.py          # Redis 支撑的线程订阅跟踪
├── webhook_routes.py   # 把 webhook 适配器的入站路由挂到主 API 上
└── adapters/
    ├── base.py         # PlatformAdapter（出站契约）+ SocketAdapter / WebhookAdapter + MessageContext
    ├── shared.py       # 平台无关的适配器辅助（附件、历史预算、bot 自回环防护）
    ├── discord/        # SocketAdapter — Discord Gateway
    │   ├── adapter.py  # Gateway 连接、事件、发送、创建线程
    │   ├── commands.py # 斜杠命令（/setup、/help、/unlink）
    │   └── config.py   # Discord token + 平台限制参数
    ├── slack/          # WebhookAdapter — Slack Events API
    │   ├── adapter.py       # 入站事件/命令路由、发送、mrkdwn、附件
    │   ├── commands.py      # 斜杠命令（/setup、/help、/unlink）
    │   ├── config.py        # Slack token + signing secret + 平台限制
    │   ├── signing.py       # HMAC-SHA256 请求签名校验
    │   ├── text.py          # CommonMark → Slack mrkdwn
    │   └── app-manifest.yaml # 可导入的 Slack 应用定义（scopes、events、commands）
    └── telegram/       # WebhookAdapter — Telegram Bot API
        ├── adapter.py       # 入站 update 路由、发送、chat 模型映射
        ├── api_client.py    # 薄 httpx Bot API 客户端（JSON + multipart + getFile）
        ├── commands.py      # Bot 命令（/setup、/help、/unlink）
        ├── config.py         # BotFather token + webhook secret + 平台限制
        └── text.py          # CommonMark → Telegram HTML

几个值得注意的实现细节：

• MessageContext 数据结构（adapters/base.py）：任何一条入站消息被适配器包装成统一的上下文后交给共享 MessageHandler，字段包括 channel_type（"dm" / "channel" / "thread" 三种字面量）、bot_mentioned、thread_history、mentionable_users（机器人本轮被允许 @ 回的用户白名单，防止 LLM 幻觉提及无关用户）、referenced_conversations（消息中链接/@ 引用的其他线程内容，由机器人经 Gateway 预先抓取，让模型直接读取而无需 web 抓取 JS 渲染页）、attachments 与 skipped_attachments（下载失败的附件会作为 (文件名, 原因) 对同时告知用户与模型，确保任何一方都不会误以为文件已被读取）。
• localize_markup：核心处理器与模型统一使用 CommonMark 作为“规范标记”，各适配器负责翻译成平台方言——Discord 近似 CommonMark 故默认恒等转换，Slack 覆写为 mrkdwn（slack/text.py），Telegram 覆写为 HTML（telegram/text.py）。

五、连接器分类学（Connector Taxonomy）

这是整个模块架构的核心设计，定义在 adapters/base.py 中：

PlatformAdapter 是核心处理器面对的出站契约——它从不指名任何具体平台。具体适配器按入站事件的到达方式扩展以下两个子类型之一：

SocketAdapter — 拥有长连接

• 拥有长生命周期连接（Discord Gateway、Slack Socket Mode），由 start/stop 驱动，运行在独立的 copilot-bot Pod 中；
• 源码中定义为抽象 start()/stop() 两个方法，app.py 的 _build_socket_adapters 负责构建；
• CoPilotChatBridge._run_adapters 用 asyncio.gather(*(a.start() for a in socket_adapters)) 并发运行所有 socket 适配器，退出时统一 stop() 并关闭 BotBackend。

WebhookAdapter — 无状态入站 POST

• 接收入站 HTTPS POST（Slack Events API、Telegram、未来的 Teams/WhatsApp），本身无状态；
• 其 register_routes(app) 把路由挂载到主后端 API 上（经 webhook_routes.py 的 register_webhook_adapters），从而直接复用既有的 N 副本部署——不需要独立 Pod；
• 源码注释明确约束：适配器必须自己负责请求签名校验，并须在平台超时前 ACK，把真正的工作调度到请求之外。base.py 还为此提供了共享工具 read_verified_webhook_body（先对平台签名的原始字节验签再解析，防止签名与解析字节不一致）和统一的 401 响应 unauthorized_webhook_response。

一个有意思的细节：copilot-bot Pod 里也会构建 webhook 适配器的出站部分（app.py 中 outbound_only = build_webhook_adapters(api)）——不挂路由、不接 handler，只为让主动 RPC（proactive posts）能触达所有已配置平台。因此 build_webhook_adapters 工厂被两个消费方共享。

局部性规则（Locality Rule）

一切平台专属代码都放在 adapters/<platform>/ 下。只有上述两个工厂函数会指名具体平台，它们根据哪些 token 被设置来决定实例化哪些适配器。核心处理器、文本工具、线程跟踪与平台 API 均保持平台无关。

六、消息是如何流转的

官方文档描述的端到端流程如下，每一步都能在 handler.py 源码中找到对应实现：

1. 用户在频道中 @ 机器人；
2. 适配器的 on_message 回调触发，构造 MessageContext，交给共享 MessageHandler（app.py 中 adapter.on_message(handler.handle) 完成绑定）；
3. Handler 依次处理：
   • 经 bot_backend 检查用户/服务器是否已关联（_ensure_linked：DM 走 resolve_user，非 DM 走 resolve_server；未关联时 DM 用户会收到 “Link Account” 按钮提示，服务器则会收到“请管理员先运行 /setup”的提示）；
   • 已关联时：对频道消息创建线程（_resolve_target 用 build_thread_name 生成名称、调 adapter.create_thread，创建失败则回退到直接在频道回复），DM 与既有线程则原地回复；
   • 在 Redis 中把该线程标记为已订阅（7 天 TTL，见下文）；
   • 把 AutoPilot 的响应以流式返回，按适配器的 chunk_flush_at 边界分块发送；
4. 批处理：流式运行期间到达的新消息会进入该目标（thread/DM）的 pending 队列，当前流结束后作为一个批量 follow-up turn 一起处理。

批处理与状态管理的源码细节

handler.py 中的 TargetState 模型精确定义了“目标”（线程 ID 或 DM 通道 ID）级别的流式状态：

class TargetState(BaseModel):
    processing: bool = False                      # 是否有流正在运行
    pending: list[tuple[str, str, str]] = ...    # (username, user_id, text) 待处理消息
    pending_file_ids: list[str] = ...             # 随 pending 消息一起排空的附件文件 ID
    session_id: str | None = None                # pending 附件上传到的会话，直接带入 turn

_enqueue_and_process 的循环逻辑是：把消息追加到 pending 后，若 processing 已为 True 直接返回（让正在跑的流结束时来取）；否则置 processing = True，在 while state.pending 循环中整批取出、清空、流式发送，finally 中把空状态从字典中弹出以避免长期运行导致内存无限增长。

几个关键约束常量及其设计理由（均取自源码注释）：

• MAX_TURN_FILE_IDS = 20（handler.py）：单个 turn 携带的文件 ID 上限，必须 ≤ BotChatRequest.file_ids 的 max_length。批量多条含文件消息可能累积超过该值，超限部分会被丢弃并记日志——宁可截断也要保证 turn 能通过校验跑起来；
• MAX_INBOUND_ATTACHMENTS = 10 与 SESSION_TTL = 7 * 86400（config.py）：前者是机器人策略（非平台限制），限制单条消息最多摄取 10 个附件，防止一条带几十个文件的消息扇出成等量上传/扫描；后者是 AutoPilot 会话 ID 缓存 TTL（按通道/线程），必须 ≥ 线程自动回复订阅 TTL——若更短，会话缓存过期后机器人会在旧线程里悄悄开一个“丢失全部对话历史”的全新 copilot 会话，转而基于跨会话的陈旧记忆行事，属于危险行为。每个 turn 都会刷新该 TTL，因此活跃对话在订阅窗口内不会丢失上下文；
• 线程订阅 TTL = 7 天（threads.py）：机器人创建线程后写入 Redis 键 copilot-bot:thread:{platform}:{thread_id}（ex=7*86400），使该线程内的后续消息无需再次 @；过期线程自动老化消失。

另外两条值得理解的行为规则：

• 非自有线程不自动回复：只有在机器人自己拥有的线程（即响应频道 @ 而创建）中才自动回复；被拉入的他人线程要求每轮显式 @，以免“劫持”团队正在进行的对话。首次被 @ 进一个陌生线程时，会拉取最近的线程历史放入 prompt 作为上下文，但不建立订阅；
• 空 turn 回收：仅包含（全部失败/跳过的）附件且无正文的消息，不会入队空 turn，且若该线程是为频道消息刚创建的，会立即 unsubscribe，避免留下一个“孤儿但已订阅 7 天”的线程；
• 附件必须先定会话再上传：附件要写入 turn 所在会话的 /sessions/<id>/ 目录，AutoGPT 才能读取。_resolve_session_for_attachments 用每目标的 asyncio.Lock 串行化会话解析，保证并发附件消息收敛到同一个会话而不是各自建会、把文件拆散到不同会话；turn gate 拒绝用户时在任何文件被扫描/存储之前就渲染拒绝并终止。

七、出站 RPC：让 AutoPilot 主动发言

CoPilotChatBridge 上四个 @expose 方法支撑“主动输出”路径（定时任务 / AutoPilot 发起的发帖），配套 CoPilotChatBridgeClient（AppServiceClient 子类）供其他服务调用：

方法	语义
list_channels(platform, user_id)	列出该用户经机器人可发帖的频道，支撑 copilot tool 里的频道名解析与选择器
send_message_to_channel(platform, user_id, channel, content)	以该 AutoGPT 用户的名义向频道（名字或 ID）发独立消息；授权基于该用户已关联的服务器
send_dm_to_user(platform, user_id, content)	发到用户与机器人的自己的 DM——目标从用户 DM link 解析而非调用方指定，确保用户只能 DM 自己
create_thread_in_channel(platform, user_id, channel, thread_name, content)	在频道创建独立线程并发帖

实现要点：_require(platform) 会先检查 _adapters_healthy，未就绪或未配置该平台适配器时抛 UnhealthyServiceError——这是一种“瞬时形态”的错误，让重试方退避而不是当成永久失败。适配器侧对应 looks_like_channel_id、list_text_channels、get_channel_server_id、post_channel_message、create_channel_thread 等主动输出契约方法（见 adapters/base.py 的 “Proactive output” 段）：由于各平台 ID 文法不同（Discord 数字 snowflake、Slack C0123ABCD、Telegram 带符号整数），“这是 ID 还是名字”的判别必须由平台适配器自行回答，无法放在平台无关层。

八、新增一个聊天平台：官方四步流程

文档给出的扩展流程（“核心处理器、文本工具、线程跟踪与平台 API 均保持不动”）：

第 1 步：创建 adapters/<platform>/ 目录，包含 adapter.py、commands.py（若平台支持命令）、config.py。

第 2 步：adapter.py 子类化 SocketAdapter（长连接）或 WebhookAdapter（入站 HTTPS），实现全部出站 PlatformAdapter 方法——max_message_length、chunk_flush_at、send_message、send_link、create_thread、send_file 等——外加其子类型的入站方法（start/stop 或 register_routes）。从 adapters/base.py 的抽象契约看，还需实现的平台限制参数包括：max_attachment_bytes（单文件上传硬上限）、max_thread_name_length（线程命名上限，共享线程命名逻辑会先按此截断）、typing_refresh_interval（打字指示器续期间隔——平台指示器会自动过期，如 Discord 约 10 秒、Telegram 约 5 秒，共享保活循环按此频率重发；无打字指示器的平台可任意取值，start_typing 本就是 no-op）、looks_like_channel_id。可选覆写的增强能力：supports_stream_drafts + send_stream_draft（仅在有原生草稿流 API 的平台启用，如 Telegram sendMessageDraft；返回三态 StreamDraftOutcome——SHOWN/SKIPPED/STOPPED，用于区分“推进节流”“下块重试”“永久停止”）、rename_thread（默认不支持，Slack 这类无名线程直接继承）、open_dm_channel（默认不支持，主动 DM 路径会如实报告平台无法投递）。

第 3 步：config.py 声明平台环境变量与平台专属数值（消息长度限制、token 名等）。

第 4 步：在对应工厂中注册，以其 token 为准入门槛：

# Socket → app.py::_build_socket_adapters
# Webhook → webhook_routes.py::_build_webhook_adapters
if <platform>_config.get_bot_token():
    adapters.append(<Platform>Adapter(api))

若为 Webhook 平台，register_routes 中务必遵循 WebhookAdapter 的契约：先验签（可复用 read_verified_webhook_body 与 unauthorized_webhook_response）、在平台超时前 ACK、真实工作调度到请求之外。

九、测试与验证

该模块每个核心文件都配有同名 *_test.py，可用来自检扩展实现是否走样：

• handler_test.py / command_core_test.py：编排、关联策略与命令行为；
• bot_backend_test.py：BotBackend 门面（linking 客户端 + 流注册表）；
• turn_stream_test.py、outbound_test.py：流式分块与出站投递；
• sessions_test.py、threads_test.py、text_test.py、prompt_test.py：会话缓存、线程订阅 TTL、文本切分、prompt 组装；
• webhook_routes_test.py、app_test.py：工厂挂载逻辑与服务生命周期。

十、小结：这套架构的取舍

回到官方文档的主张，copilot/bot 模块的设计可归纳为三条：

1. 出站统一、入站二分：核心 MessageHandler 只依赖 PlatformAdapter 出站契约，平台差异被压缩进两个入站子类型与 adapters/<platform>/ 目录，唯一指名平台的代码集中在 _build_socket_adapters 与 _build_webhook_adapters 两个工厂；
2. 部署形态由入站方式决定：长连接适配器（Discord）独占一个 Pod（连接是每 token 单例）；webhook 适配器无状态，白嫖主后端 N 副本的容量——这也是为什么 Telegram/Slack 不需要额外进程；
3. 状态收敛在 Redis 与每目标锁：7 天的线程订阅 TTL、与会话缓存 TTL 严格对齐、按目标串行化的会话解析锁，共同保证多副本/长生命周期场景下对话上下文不漂移、文件不拆散、线程不“孤儿订阅”。

对于要在 AutoGPT 上接入新聊天平台的开发者，正确姿势就是文档第八节（本文“新增一个聊天平台”）的四步流程：实现契约 → 声明配置 → 注册工厂，其余全部复用。