最近公司内部做了一个“AI 员工”项目核心需求很朴素让同一个大模型大脑能同时在微信、企业微信、飞书三套 IM 里回应员工的消息。听起来像接三个机器人 SDK真正动手后才发现这三条通道的协议差异比想象中大得多——微信公众号是 XML 报文加 AES企业微信是加密回调加 access_token飞书则是事件订阅加长连接。这篇笔记就是围绕“三条 IM 通道的协议实现”整理出来的实战记录含接入形态对比、回调服务骨架、加解密细节、消息去重与异步回复方案、多通道抽象层设计以及上线后我真实踩过的坑。适合准备在公司内部做 IM 机器人、统一客服助手或者想给现有 Agent 套 IM 壳的开发者参考。1. 先回答一个问题为什么要把 AI 员工养在 IM 里大部分团队第一次做 AI 助手第一反应是做个网页版聊天窗口。页面做好了上线两周发现日活全靠自己人测试。问题不在模型效果而在用户习惯——员工一天八小时泡在 IM 里你让他专门打开一个网页去和大模型对话这个动作本身就构成了使用门槛。把 AI 员工养在 IM 里本质上是在复用用户已经建立的工作流消息来了就在对话框里回不需要切换上下文不需要记网址更不需要学一套新交互。从技术实现上看这三条通道虽然协议各不相同但抽象之后核心链路是一致的接收用户消息解析出文本和发送者身份调用大模型再把回复推回对应的会话。难点不在 AI而在每条通道各自的接入规范、加解密规则和消息时限。我当时的选型标准有三个一是必须走官方支持的接入方式不做 Hook、不做逆向、不碰任何灰色手段二是服务端要能部署在公司内网消息链路要可审计三是同一套 AI 逻辑必须能被三条通道复用而不是每个通道各写一套 prompt。基于这三条标准最终方案定为微信侧用公众号后台的服务器配置接收普通消息企业微信侧用自建应用的接收消息回调飞书侧用应用机器人的事件订阅。三套通道都指向同一个回调服务回调服务内部做协议适配真正的 AI 处理放到统一的 Agent 层。为什么不用微信个人号方案虽然个人号在开发者圈子里一直有自动化方案但它违反平台规则存在封号风险而且消息收发依赖逆向协议不稳定。公司内部用还好一旦涉及外部客户或生产数据合规上完全过不了。所以微信通道老老实实走了公众号这条正规军路线。同样企微和飞书也都只用官方开放平台的接口能力这样后续无论是审核、权限控制还是消息审计都能在可控范围内。2. 三条通道的接入姿势微信公众号、企业微信自建应用、飞书应用机器人三条通道放在一起对比最容易混淆的就是它们各自的接入形态和消息承载格式。我在最开始设计时做了一张对照表后面所有适配代码都围绕这张表展开。通道接入实体消息到达方式报文格式主动推送方式微信公众号服务号服务器配置回调XML内嵌密文客服消息接口 / 模板消息企业微信自建应用接收消息回调XML内嵌密文应用消息推送 message/send飞书应用机器人事件订阅Webhook 或长连接JSON加密后机器人发送消息 API2.1 微信侧公众号回调的 XML 与被动回复微信公众号接入的核心是“服务器配置”里的 URL、Token 和 EncodingAESKey。用户在公众号对话框里发消息时微信服务器会往你的 URL 发起 GET 请求做签名校验校验通过后后续消息以 POST 方式推过来body 是 XML 格式。明文模式下的 XML 大致长这样xml ToUserName![CDATA[gh_xxx]]/ToUserName FromUserName![CDATA[oXXXX]]/FromUserName CreateTime1700000000/CreateTime MsgType![CDATA[text]]/MsgType Content![CDATA[你好]]/Content MsgId1234567890/MsgId /xml实际生产环境中我不会用明文模式而是开启安全模式。安全模式下 POST 过来的 body 是一段密文需要先用 EncodingAESKey 解密才能拿到上面的 XML 明文结构。解密之后解析MsgType和Content再按业务逻辑处理后返回 XML 应答。这里有一个非常容易被忽略的点公众号被动回复有严格的时间限制收到消息后必须在5 秒内返回响应报文否则微信会报“该公众号暂时无法提供服务”。如果 AI 处理耗时超过 5 秒你就不能直接同步等待大模型输出后再组装回复。解法我在后面会专门讲核心是“先回一个空应答或特定状态码再通过客服消息接口主动推回复”。微信通道最麻烦的地方在于FromUserName是用户的 openid你无法直接拿到用户的微信号、手机号等身份信息。如果要把消息对应用到内部员工体系必须在公众号网页授权时把 openid 和公司账号绑定建一张映射表。2.2 企业微信侧自建应用的回调与主动推送企业微信的接入逻辑和微信公众号很像但也有几个明显的差异。首先是凭证体系。企微的主动消息推送依赖access_token获取方式是GET /cgi-bin/gettoken?corpidxxxcorpsecretyyyToken 有效期 7200 秒需要缓存并定时刷新。这里我踩过一个高频坑企微的 access_token 在并发刷新时会导致旧 token 失效如果多个 worker 同时发现自己缓存过期同时去刷新先刷新的那个会把后面刷新成功前的 token 挤掉造成“反复失效”。解决方法是加 jitter 和进程内锁把过期时间提前 200 秒刷新。其次企微回调的 POST body 也是加密后的 XML解密后的核心结构大致是xml ToUserName![CDATA[wwXXXX]]/ToUserName FromUserName![CDATA[ZhangSan]]/FromUserName CreateTime1700000000/CreateTime MsgType![CDATA[text]]/MsgType Content![CDATA[你好]]/Content MsgId1234567890/MsgId AgentID1000002/AgentID /xml注意FromUserName在企微里是发消息员工的 UserID如果是从外部联系人发来的消息则是外部联系人 ID这意味着你可以直接用它关联到企业通讯录不需要额外绑定。企微主动推消息的方式也更多样文本、图片、图文、markdown 都支持。我最常用的是 markdown 消息因为它可以直接展示表格、加粗、标题用来回复带有简单结构化内容的 AI 结果非常合适。唯一要注意的是 markdown 消息也需要通过应用的agentid发送而且部分字段在企微客户端里渲染时不完全兼容标准 markdown比如表格的渲染在不同版本上表现不一致。2.3 飞书侧事件订阅的两种模式和卡片消息飞书是三条通道里对开发者最友好的一个协议上也是和传统 Webhook 差别最大的一条。飞书的事件订阅有两种模式一种和微信企微类似需要配置公网回调地址飞书服务器把事件 POST 到你的地址另一种是长连接模式由飞书开放平台维持一个长连接通道消息到达后直接通过该连接推给本地服务完全不需要公网 IP也不必做 frp 之类的内网映射。这个设计我非常喜欢它解决了内网部署的痛点回调服务可以直接放在公司内网不暴露到公网。事件回调的 body 是 JSON明文模式下大致这样{ schema: 2.0, header: { event_id: xxxx, event_type: im.message.receive_v1, create_time: 1700000000, token: xxxx, app_id: cli_xxx }, event: { message: { chat_id: oc_xxx, message_type: text, content: {\text\:\你好\}, message_id: om_xxx }, sender: { sender_id: { open_id: ou_xxx } } } }飞书的content是 JSON 字符串text 消息需要JSON.parse(message.content).text才能拿到真正的用户输入。因为 content 是字符串嵌套初学者容易直接当对象取属性取到 undefined后面我会单独写避坑。飞书主动推送消息也很方便应用机器人可以通过im/v1/messages接口直接发送文本、富文本、图片、卡片等消息。和企微应用消息不同飞书的发消息接口使用receive_id_typeopen_id或chat_id不需要预先在会话里触发什么绑定关系服务端只要拿着用户的 open_id 就能给他发消息这在实现“异步回复”时非常省事。3. 统一回调基座路由设计、签名校验与加解密封装三条通道虽然报文格式不同但整体流程可以抽象成一条链接收请求 - 验证来源合法性 - 解密 - 解析消息 - 交给业务层 - 决定是否主动回推。这套链路的底座我统称“回调基座”是所有协议实现的入口。3.1 路由层一个 URL 还是三个 URL我在设计时给三条通道各保留了一个独立路径这样方便排查和权限控制POST /api/webhook/wechat POST /api/webhook/wecom POST /api/webhook/feishu为什么不直接用一个统一入口因为三条通道的签名逻辑、加密逻辑、握手验证方式都不一样。放在独立 path 里每个 handler 只处理一种协议出错时按 channel 加日志 tag 就能快速定位。如果非要在一个入口里做 switch代码会在版本迭代中越来越臃肿。用 FastAPI 写的话大致骨架是这样from fastapi import FastAPI, Request app FastAPI() app.post(/api/webhook/wechat) async def wechat_webhook(request: Request): # 微信的 URL 验证是 GET消息推送是 POST # 解密、解析 XML返回 XML 或空串 ... app.post(/api/webhook/wecom) async def wecom_webhook(request: Request): ... app.post(/api/webhook/feishu) async def feishu_webhook(request: Request): ...路由这层最重要的是把“通信协议处理”和“业务逻辑处理”拆开。通信协议处理包括签名校验、加解密、报文解析业务逻辑处理包括调用大模型、查知识库、决定回复内容。两者千万不要混在同一个函数里否则后面任何一个加密参数变动都会导致业务代码跟着遭殃。3.2 签名校验防的不是黑客是“来源非法”签名校验的作用是确认请求确实来自微信/企微/飞书的服务器而不是有人伪造了一个 POST 请求往你的回调地址里灌假消息。微信公众号和企微的 URL 验证都依赖msg_signature或signature这个参数它的计算逻辑大同小异把 token、timestamp、nonce以及加密后的报文按字典序排序后拼成字符串做 SHA1再比对签名。企微的WXBizMsgCrypt官方库帮你把这些都封装好了但基于安全考虑我用的是自己维护的 Python 版本只依赖标准库和 cryptography。飞书的新版事件订阅校验方式和微信不同它会校验请求头里的X-Lark-Signature同时要求你在配置事件订阅时填一个 Verification Token。我强烈建议同时开启“加密”选项这样 body 里的encrypt字段需要解密后才能看到真实 JSON。这里有一条我实际排障得出的经验签名校验失败的原因十有八九不是加密算法写错而是服务器时间和真实时间偏差太大。微信和企微的签名机制都包含 timestamp本地时间比标准时间差一分钟以上就会导致校验失败。所以部署回调服务前先确认服务器 NTP 同步正常。3.3 加解密封装AES 的细节决定了成败微信和企微用的是同一种消息加密方案AES-256-CBC 加 PKCS7 padding。EncodingAESKey 是 Base64 编码后的 43 位字符串解码后是 32 字节的 AES 密钥这一点很容易搞错——很多人直接把 43 位字符串当密钥用导致解密失败。解密后的明文结构里有一个 16 字节的随机串前缀紧接着是 4 字节的网络字节序长度最后才是真正的 XML 报文。如果直接用 AES 解密然后decode(utf-8)你会得到一串包含乱码的字符串必须按规范切掉前 20 字节才能拿到干净的 XML。飞书的加密则不同使用的是 AES-256-GCM并且加密结果是 Base64 编码的 JSON 字段encrypt解密后直接得到明文 JSON。实现时要注意飞书的 AES 密钥Encrypt Key是在开发者后台单独配置的和 Verification Token 不是同一个东西。加密这块我给团队留了一条硬规范加密逻辑全部收敛到crypto.py里每个通道给一个独立的编解码函数并针对明文和密文分别写单元测试。测试用例里直接放微信官方文档里的样例密文和期望明文防止后续改库导致隐性破坏。4. 消息去重与异步回复绕过“5秒被动回复”的硬限制这一章是我觉得最有实操价值的部分因为协议文档通常只讲格式不讲业务层的坑。真正让机器人“像个员工”而不是“像个 Echo”的是处理时限和重复消息两道关。4.1 重复推送IM 平台的“消息兜底”机制微信、企微、飞书都会在回调失败或超时后重试推送事件。飞书的重试策略尤其积极如果回调地址没有在限定时间内返回成功状态它会在几秒内重新推送同一event_id。企业微信也会在多长时间内重试同一消息。如果你不做去重AI 会针对同一条用户消息生成两次回复用户体验极差。去重方案我用的是 Redis 布式锁 消息 ID 缓存键的格式是{channel}:{message_id}value 存处理状态过期时间设为 10 分钟——这个时间足够覆盖平台可能的重试窗口。伪代码如下import redis r redis.Redis.from_url(settings.redis_url) def is_duplicate(channel: str, message_id: str) - bool: key fduplicate:{channel}:{message_id} # 第一次处理时返回 False并设置键 return not r.set(key, 1, nxTrue, ex600)这里nxTrue是关键参数表示只有当键不存在时才写入。并发场景下只有一个 worker 能抢到其他 worker 拿到 False 直接丢弃消息。如果你们的消息量不大放在进程内 LRU 缓存也能顶住但一旦回调服务扩到多副本就必须用 Redis。4.2 异步处理先应答再慢慢算前面提到微信被动回复只有 5 秒飞书和企微虽然时限更长一些但大模型完整回答一次往往要 5 到 15 秒尤其是接 Agent 流程、查飞书文档、调内部 API 的场景时间完全不可控。所以我的策略是所有耗时逻辑都不放在回调函数里同步执行。具体流程是收到消息校验签名、解密、解析。记录消息主动或被动返回“空响应”告诉平台我已经收到了微信直接返回空串或 success企微返回空串飞书返回{code: 0}。把处理任务丢进消息队列我用 Redis Stream简单可靠。消费者从队列里取出任务调用 LLM/Agent 生成回复。生成回复后通过各通道的主动推送 API把消息发给用户。这一步需要特别注意微信的客服消息接口有 48 小时有效期限制而且用户主动发消息后客服接口才可用。飞书和企微则没有这个限制只要拿到了 open_id/UserID 就能主动发。实际跑下来异步方案的一个隐藏收益是削峰填谷。白天大家高频使用时大量消息同时到达如果同步等待 LLM回调服务会被拖垮。异步队列天然缓冲了消息洪峰LLM 并发控制在 5 到 10多余的消息排队等待反而比全部乱打要稳定得多。4.3 队列消费的可靠性队列消费最怕的是任务丢失。Redis Stream 的XADD和消费者组的XACK机制可以做到至少一次消费。我的消费逻辑是先XREADGROUP拉消息处理完成后XACK确认。如果消费者在处理中崩溃消息会一直留在 Pending 列表里重新拉起后由另一个消费者接手处理。但这会导致重复处理所以又回到刚才的去重键。一个完整的任务链路应该是消息去重判定放在入队前而消费端的message_id去重作为兜底。两道关都过了基本不会出现对用户可见的重复回复。5. 多通道抽象层让一个 AI 大脑听懂三种 IM 的方言三条通道的协议差异如果直接渗透到业务代码里AI Agent 层就会到处写 if channel wechat 这种面条代码。所以我在回调基座之上又加了一层适配器这是整个项目里收益最高的设计。5.1 定义统一消息模型所有通道在解析完成后都转换为内部统一的IMMessage对象dataclass class IMTextMessage: channel: str # wechat / wecom / feishu sender_id: str # openid / userid / open_id thread_id: str # 会话唯一标识 text: str # 纯文本内容 raw: dict # 原始消息便于排查thread_id在三通道里对应关系如下微信用户的 openid因为公众号场景下基本是一对一对话。企业微信如果是单聊可以用FromUserName如果是群聊需要从会话里解析chat_id再映射到某个唯一的会话键。飞书message.chat_id最合适它既标识了单聊也标识了群聊。统一模型的价值在于AI 层的记忆管理、多轮上下文存储、权限判断都可以只依赖sender_id和thread_id不需要关心消息到底来自哪个平台。比如一个员工先在企业微信里问“昨天数据日报怎么没发”然后再去飞书里问同样的问题这两条消息如果sender_id映射到了同一内部用户AI 就能复用上下文记忆。当然前提是先做账号绑定。5.2 适配器接口与实现每个通道对应一个 Adapter实现以下方法class MessageAdapter(ABC): abstractmethod def parse(self, request) - Optional[IMTextMessage]: ... abstractmethod def send_text(self, recipient, text: str, thread_id: str): ... abstractmethod def send_card(self, recipient, card: dict): ...parse负责把通道特有的请求数据转成统一消息模型send_text和send_card负责把通用回复转成各通道的 API 调用。真正的大模型调用逻辑只和IMTextMessage打交道不接触任何通道协议。举个简单的例子在飞书 Adapter 里parse可能需要这样处理class FeishuAdapter(MessageAdapter): def parse(self, request: dict) - Optional[IMTextMessage]: event request.get(event, {}) message event.get(message, {}) content json.loads(message.get(content, {})) text content.get(text, ) if not text: return None return IMTextMessage( channelfeishu, sender_idevent[sender][sender_id][open_id], thread_idmessage[chat_id], texttext, rawrequest, )class WecomAdapter(MessageAdapter): def parse(self, decrypted_xml: str) - Optional[IMTextMessage]: # 用 XML 解析库提取 Content、FromUserName 等字段 ...这样在新增通道时只需要新写一个 Adapter注册到工厂类里AI 层完全无感知。5.3 路由中心CallbackDispatcher回调请求进来后我先经过一个 Dispatcher 做统一分拣。Dispatcher 的核心逻辑是根据请求路径确定 channel。调用对应 Adapter 的parse。如果解析不到有效文本直接返回成功。对message_id做去重判断。把统一消息对象丢入队列。Dispatcher 本身不处理任何业务逻辑它是一个协议的“翻译官”。业务代码只依赖IMTextMessage这让我可以在不触碰 AI 逻辑的情况下单独替换某一条通道的实现。比如飞书换新版 API 时只需要改 FeishuAdapter其他代码不用动。6. 上线之后的排障经验从“消息丢失”到“卡片乱码”项目上线后最耗时间的往往不是初期开发而是边角问题。我整理了几个真实遇到的故障按出现频率排序。6.1 事件到了但 AI 没回复先查队列第一个周末我收到反馈说企微里有人发消息没反应。排查链路是看回调服务的访问日志发现请求进来了看消息队列发现任务也在看 Worker 日志发现 LLM 调用超时。根因是企微回调在本地开发环境测试时毫秒级返回生产环境里我往队列里塞了大量任务而 Worker 的并发数设成了 1导致排队严重。这个问题的教训是消息队列消费端一定要配置合理的并发数并且要监控积压长度。我后来给 Worker 加了一个简单的指标队列长度超过 100 时自动把并发数上调到 10超过 1000 时发告警到飞书群。对于内部 AI 员工这种场景用户能接受 1 到 2 秒的延迟但超过 10 秒就会反复点击重发反而加剧消息重复。6.2 飞书 content 字段解析返回 undefined飞书 text 消息的content是一个 JSON 字符串这在官方文档里有写但我团队里一位同事第一次接飞书时直接event.message.content.text取到 undefined回调却返回成功导致用户没收到任何回复。这种故障特别隐蔽因为没有报错。后来我在 Adapter 的 parse 里加了解析后的非空校验解析不到文本时主动返回None由 Dispatcher 落下日志并忽略。6.3 企业微信 access_token 并发失效这个前面已经提到过表现是白天消息量上来后莫名出现大量60020或者40014错误码。根因就是多 worker 同时刷新 access_token后刷新的把先刷新的踢下线。解决方式是封装一个带锁的 token provider用 Redis 分布式锁保证全局只有一个刷新任务其他进程阻塞等待新 token。同时把缓存过期时间从 7200 秒提前到 7000 秒刷新留出安全余量。6.4 企微 markdown 卡片在移动端显示不完整企微的 markdown 消息在桌面端支持得不错但在移动端有字符数限制超长内容会被截断。另外部分 markdown 语法比如表格嵌套在移动端渲染会退化成纯文本看起来像乱码。我把所有原本用 markdown 表格输出的内容改成了纯文本列表每条前面加1.、2.这种编号兼容性最好。飞书则没有这个问题卡片消息的 JSON 结构很稳定富文本和交互按钮都能跨端一致渲染。6.5 安全边界回调地址的 HTTPS 与 IP 白名单三条通道都强制要求回调地址为 HTTPS。我在内网环境用的是自建网关证书没有盲目信任所有证书链。同时在企业微信后台配置了可信 IP只有企业出口 IP 能调用回调接口飞书侧则靠事件订阅的签名校验与加密请求体兜底。微信公众号在安全模式下加密后的报文即使被截获也无法解出明文基本可以放心。最后再分享一个我在实际使用中很有效的小技巧每条通道在回复内容末尾都自动附带一个不可见的调试标记比如[w:1.2.3]、[wc:1.2.3]、[f:1.2.3]表示版本号和通道。上线初期这个标记帮我快速判断用户反馈的问题是哪条通道、哪个代码版本引入的。等系统稳定后再去掉。这个小习惯帮我少开了很多次“这个 bug 到底在哪条链路”的会议。