简介这是一套开箱即用的多平台智能对话机器人开源实现CoW项目面向AI应用开发者、企业IT集成工程师及大模型落地实践者解决多渠道客服系统快速接入与私有化部署难题。资源包含200个文件以141个Python核心逻辑文件为主辅以16个Markdown说明文档、13个模板配置、6个Shell部署脚本及Dockerfile等基础设施文件整体仅480KB轻量紧凑且结构清晰——config-template.json、roles.json等关键配置体现模块化设计chat.html和contact.jpg佐证前端交互能力。目前已有161人学习下载读者可直接获得支持微信公众号、企业微信、飞书、钉钉四端接入的完整代码基线涵盖多轮对话上下文管理、语音识别/合成Azure/Baidu/Whisper、图片理解、插件扩展机制及12种主流大模型API适配能力并基于自有知识库快速定制行业AI助手。1. 为什么一个“智能对话机器人deepseek多平台接入”的标题会让运维和产品同学同时皱眉又心动这不是又一个套壳聊天框。它直指当前企业级AI落地最痛的断层大模型能力很强但业务入口散在微信公众号、企业微信、飞书、钉钉四个互不联通的渠道每个渠道的鉴权机制、消息格式、事件回调、会话上下文维持方式都不同而 deepseek 这类开源大模型又不像商用API那样自带渠道适配层——你得自己把“模型推理”这个黑匣子稳稳嵌进四个完全不同的企业通讯协议里。某高校实验室做过测算若不做抽象为每个平台单独写一套接入逻辑重复代码量超65%且后续模型升级、提示词迭代、意图识别规则调整要同步改四份。本文讲的就是如何用一套可复用的中间件架构把 deepseek 的对话能力真正“插”进这四个主流办公平台不是演示Demo是能扛住日均5万消息、支持会话状态跨平台延续、具备错误降级与审计追踪的真实部署方案。适合正在推进AI客服、内部知识助手或流程自动化落地的后端工程师、AI Infra 工程师和懂技术的产品负责人。2. 架构选型为什么不用现成SDK而选择自建「协议抽象层 模型调度中心」双核结构2.1 四大平台接入的本质差异决定了不能靠“封装SDK”一劳永逸很多人第一反应是微信有官方Python SDK钉钉有OpenAPI Python Client飞书有lark-sdk……直接 pip install 不香吗香但只香在单点验证阶段。真实生产中你会立刻撞上三个不可回避的硬伤事件生命周期不一致微信公众号依赖POST /callback接收用户消息但企业微信要求先调用get_jsapi_ticket获取临时票据才能解密消息体飞书事件需校验X-Lark-Signature头而钉钉必须用aes_key解密encrypt字段——这些前置动作无法用同一套初始化逻辑完成。会话上下文存储语义冲突微信用FromUserName作为会话ID企业微信用sender_id需配合chatid判断是否群聊飞书用open_idtenant_key组合钉钉用userid但需区分是否开启免登。若强行统一为user_id会导致群聊场景下所有成员共享同一上下文指令错乱。错误重试策略不可控微信要求5秒内响应超时即重发飞书允许30秒但重试间隔呈指数退避钉钉未明确定义实测发现超时后可能静默丢弃。若共用一个HTTP client超时设置必有一方失败。提示所谓“多平台接入”本质是四套独立的网络协议栈 四套独立的状态管理逻辑。任何试图用单一SDK覆盖全部的方案都会在灰度发布时暴露耦合缺陷。2.2 我们最终采用的双核架构协议抽象层Adapter 模型调度中心Orchestrator我们放弃“一个SDK打天下”的思路转而构建两层松耦合模块Adapter 层协议适配器每个平台对应一个独立子模块wechat_adapter.py、ww_adapter.py、feishu_adapter.py、dingtalk_adapter.py职责唯一将平台原始HTTP请求标准化为统一的内部消息对象InternalMessage。该对象字段固定为class InternalMessage(BaseModel): platform: Literal[wechat, ww, feishu, dingtalk] user_id: str # 平台原始用户标识不作归一化 chat_id: Optional[str] # 群聊ID无则None message_id: str content: str timestamp: int # Unix毫秒时间戳 raw_event: Dict # 原始请求体供debug用关键设计Adapter 不做任何业务逻辑不调用模型不查数据库只做“翻译”。哪怕未来新增Telegram接入也只需新增telegram_adapter.py不影响其他模块。Orchestrator 层调度中心接收所有Adapter转换后的InternalMessage统一执行以下流程会话路由根据platform user_id chat_id生成唯一session_key从Redis读取历史上下文含system prompt、最近10轮对话、用户画像标签模型调度将拼接好的prompt送入 deepseek-v2-7b 模型本地vLLM部署设置max_tokens512,temperature0.3,top_p0.85结果分发将模型输出response_text和结构化response_meta含是否触发工单、是否需要人工介入等标记交还对应Adapter由其按平台规范组装响应体并回传。这种分离让各模块可独立演进模型升级只需改Orchestrator的推理调用微信接口变更只需修wechat_adapter.py审计日志只需在Orchestrator入口加一行logger.info(fdispatch: {msg.session_key} - {model_name})。3. 四大平台Adapter实现从微信公众号到钉钉每一步都踩过坑的最小可行代码3.1 微信公众号Adapter别被“明文模式”骗了99%的线上环境必须走AES加密微信公众号提供三种消息加解密模式明文、兼容、安全AES。文档说“明文模式仅用于调试”但很多教程仍用它起步——这是第一个坑。真实环境启用JSAPI支付、获取用户手机号等高级接口时微信强制要求安全模式否则回调失败。# wechat_adapter.py import base64 from Crypto.Cipher import AES from Crypto.Util.Padding import unpad from fastapi import Request, HTTPException class WeChatAdapter: def __init__(self, token: str, appid: str, encoding_aes_key: str): self.token token self.appid appid # encoding_aes_key 是43位base64字符串需转为32字节key self.aes_key base64.b64decode(encoding_aes_key ) async def parse_request(self, request: Request) - InternalMessage: # 1. 校验签名必须否则恶意请求可伪造 signature request.query_params.get(signature) timestamp request.query_params.get(timestamp) nonce request.query_params.get(nonce) echostr request.query_params.get(echostr) # 首次接入验证用 if echostr: if self._check_signature(signature, timestamp, nonce, echostr): return InternalMessage( platformwechat, user_idverify_only, message_idverify, contentechostr, timestampint(timestamp), chat_idNone, raw_event{echostr: echostr} ) else: raise HTTPException(403, Invalid wechat signature) # 2. 解析POST body安全模式下为XML加密包 body await request.body() if not body: raise HTTPException(400, Empty body) # 安全模式body是xmlEncrypt![CDATA[xxx]]/Encrypt/xml # 先提取CDATA内容再AES解密 try: import xml.etree.ElementTree as ET root ET.fromstring(body) encrypt_node root.find(Encrypt) if encrypt_node is None: raise ValueError(No Encrypt node in XML) encrypted_data encrypt_node.text if not encrypted_data: raise ValueError(Encrypt node empty) # AES解密PKCS7填充CBC模式IVapp_id右16位 iv self.appid[-16:].encode() # 注意必须是bytes cipher AES.new(self.aes_key, AES.MODE_CBC, iv) decrypted unpad(cipher.decrypt(base64.b64decode(encrypted_data)), AES.block_size) xml_content decrypted.decode() # 再解析解密后的XML sub_root ET.fromstring(xml_content) from_user sub_root.find(FromUserName).text msg_type sub_root.find(MsgType).text content_node sub_root.find(Content) content content_node.text if content_node is not None else return InternalMessage( platformwechat, user_idfrom_user, message_idsub_root.find(MsgId).text, contentcontent, timestampint(sub_root.find(CreateTime).text), chat_idNone, # 公众号私聊无chat_id raw_event{xml: xml_content} ) except Exception as e: logger.error(fWeChat parse failed: {e}, raw body: {body[:100]}) raise HTTPException(400, fWeChat parsing error: {str(e)}) def _check_signature(self, sig: str, ts: str, nonce: str, echo: str) - bool: tmp_list [self.token, ts, nonce, echo] tmp_list.sort() tmp_str .join(tmp_list) import hashlib return hashlib.sha1(tmp_str.encode()).hexdigest() sig参数说明encoding_aes_key公众号后台“基本配置”页的“消息加解密密钥”43位base64字符串末尾缺自动补appid公众号AppID用于构造AES IVtoken公众号Token用于签名校验。注意echostr验证必须放在解密前否则首次接入时微信发来的明文验证串会被当成加密包去解必然报错。3.2 企业微信Adaptersender_id≠user_id群聊消息必须用chatid做会话锚点企业微信的坑在于sender_id是发送者在当前会话中的ID不是全局用户ID。在群聊中sender_id是群内临时ID换一个群就变而真正的用户身份是userid需调用/user/getuserinfo用code换。但我们的Adapter层绝不主动调用外部API否则违反“只翻译不业务”的原则所以必须用chatidsender_id组合作为会话键。# ww_adapter.py from fastapi import Request, HTTPException import json import hmac import hashlib class WWAdapter: def __init__(self, secret: str, token: str): self.secret secret # 应用Secret self.token token # 应用Token async def parse_request(self, request: Request) - InternalMessage: # 1. 校验消息签名企业微信强制 signature request.headers.get(X-WX-KEY) timestamp request.headers.get(X-WX-TIMESTAMP) nonce request.headers.get(X-WX-NONCE) if not all([signature, timestamp, nonce]): raise HTTPException(400, Missing X-WX-* headers) # 签名算法sha256(secret timestamp nonce) expected_sig hmac.new( self.secret.encode(), f{timestamp}{nonce}.encode(), hashlib.sha256 ).hexdigest() if not hmac.compare_digest(signature, expected_sig): raise HTTPException(403, Invalid WW signature) # 2. 解析JSON body try: body await request.json() except json.JSONDecodeError: raise HTTPException(400, Invalid JSON body) # 企业微信事件类型极多我们只处理文本消息msgtypetext if body.get(MsgType) ! text: # 忽略图片、卡片等非文本事件返回空响应微信不重试 return None # 关键群聊 vs 单聊判断 chat_id body.get(ChatId) # 群聊存在单聊为None sender_id body.get(SenderID) if not sender_id: raise HTTPException(400, Missing SenderID) # 企业微信文本内容在Content字段已自动解码无需base64 content body.get(Content, ).strip() return InternalMessage( platformww, user_idsender_id, # 保留原始sender_idOrchestrator层再决定是否查userid chat_idchat_id, message_idbody.get(MsgId, ), contentcontent, timestampint(body.get(CreateTime, 0)), raw_eventbody )参数说明secret企业微信管理后台 → 应用 → “应用详情”页的“Secret”token同页的“Token”用于签名校验注意不是CorpID。提示企业微信的SenderID在单聊中是员工userid在群聊中是群内临时ID。Orchestrator层需根据chat_id是否为空决定是否调用/user/get接口补全信息——这正是业务逻辑该在的位置Adapter绝不越界。3.3 飞书AdapterX-Lark-Signature校验必须用原始body不能先json.load再校验飞书签名算法要求对原始二进制body计算HMAC-SHA256而很多开发者习惯先await request.json()得到dict再json.dumps()拼接——这会导致空格、键序、转义符不一致签名永远失败。# feishu_adapter.py from fastapi import Request, HTTPException import hmac import hashlib import json class FeiShuAdapter: def __init__(self, app_secret: str): self.app_secret app_secret async def parse_request(self, request: Request) - InternalMessage: # 1. 获取原始body关键 raw_body await request.body() if not raw_body: raise HTTPException(400, Empty body) # 2. 校验X-Lark-Signature必须用raw_body signature request.headers.get(X-Lark-Signature) timestamp request.headers.get(X-Lark-Timestamp) nonce request.headers.get(X-Lark-Nonce) if not all([signature, timestamp, nonce]): raise HTTPException(400, Missing X-Lark-* headers) # 签名算法hmac_sha256(app_secret, timestamp nonce body) # 注意body是原始字节不是json字符串 expected_sig hmac.new( self.app_secret.encode(), f{timestamp}{nonce}.encode() raw_body, hashlib.sha256 ).hexdigest() if not hmac.compare_digest(signature, expected_sig): raise HTTPException(403, Invalid Feishu signature) # 3. 解析body此时才可json.load try: body json.loads(raw_body) except json.JSONDecodeError: raise HTTPException(400, Invalid JSON body) # 飞书事件结构event.type message 且 event.message.chat_type p2p or group event body.get(event, {}) if event.get(type) ! message: return None # 忽略非消息事件 msg event.get(message, {}) chat_type msg.get(chat_type) if chat_type not in [p2p, group]: return None # 用户标识open_id全局唯一 tenant_key租户ID open_id event.get(sender, {}).get(sender_id, {}).get(open_id) tenant_key event.get(tenant_key) if not open_id or not tenant_key: raise HTTPException(400, Missing open_id or tenant_key) # 文本内容msg.content 是JSON字符串需二次解析 try: content_json json.loads(msg.get(content, {})) text_content content_json.get(text, ).strip() except (json.JSONDecodeError, TypeError): text_content return InternalMessage( platformfeishu, user_idopen_id, chat_idmsg.get(chat_id) if chat_type group else None, message_idmsg.get(message_id, ), contenttext_content, timestampint(event.get(created_at, 0)), raw_eventbody )参数说明app_secret飞书开放平台 → 应用 → “凭证与基础信息”页的“App Secret”。注意飞书的content字段是JSON字符串如{text:hello}不是纯文本必须json.loads两次——这是文档里藏得很深的细节。3.4 钉钉Adapterencrypt字段解密前必须先Base64解码且AES IV固定为16个\0钉钉的加密逻辑最反直觉encrypt字段是Base64编码后的密文解密时IV不是动态生成而是16个ASCII 0\x00\x00...。# dingtalk_adapter.py from fastapi import Request, HTTPException import base64 from Crypto.Cipher import AES from Crypto.Util.Padding import unpad import json class DingTalkAdapter: def __init__(self, aes_key: str, token: str): # aes_key 是43位base64字符串转为32字节 self.aes_key base64.b64decode(aes_key ) self.token token async def parse_request(self, request: Request) - InternalMessage: # 1. 校验签名钉钉用SHA256 signature request.query_params.get(signature) timestamp request.query_params.get(timestamp) nonce request.query_params.get(nonce) if not all([signature, timestamp, nonce]): raise HTTPException(400, Missing signature/timestamp/nonce) # 签名算法sha256(token timestamp nonce body) body await request.body() expected_sig hashlib.sha256( (self.token timestamp nonce body.decode()).encode() ).hexdigest() if not hmac.compare_digest(signature, expected_sig): raise HTTPException(403, Invalid DingTalk signature) # 2. 解析JSON body提取encrypt字段 try: body_json json.loads(body) except json.JSONDecodeError: raise HTTPException(400, Invalid JSON body) encrypt_str body_json.get(encrypt) if not encrypt_str: raise HTTPException(400, Missing encrypt field) # 3. AES解密key32字节IV16个\x00modeCBCPKCS7填充 try: encrypted_data base64.b64decode(encrypt_str) iv b\x00 * 16 cipher AES.new(self.aes_key, AES.MODE_CBC, iv) decrypted unpad(cipher.decrypt(encrypted_data), AES.block_size) decrypt_json json.loads(decrypted.decode()) # 钉钉消息体在decrypt_json[msg]中 msg decrypt_json.get(msg, {}) if not isinstance(msg, dict): raise ValueError(msg is not a dict) # 用户IDuserid需企业授权或 unionid跨企业 userid msg.get(userid) if not userid: raise ValueError(Missing userid in decrypted msg) # 文本内容text字段 content msg.get(text, {}).get(content, ).strip() return InternalMessage( platformdingtalk, user_iduserid, chat_idNone, # 钉钉群聊需额外处理此处简化为单聊 message_idmsg.get(msgId, ), contentcontent, timestampint(msg.get(createTime, 0)), raw_eventdecrypt_json ) except Exception as e: logger.error(fDingTalk decrypt failed: {e}) raise HTTPException(400, fDingTalk decrypt error: {str(e)})参数说明aes_key钉钉管理后台 → 应用 → “开发管理” → “加解密密钥”页的密钥43位base64token同页的“Token”。血泪经验钉钉的encrypt解密失败90%是因为忘了base64.b64decode直接拿字符串当字节解密——AES会静默返回乱码然后json.loads报错很难定位。4. 避坑指南四大平台接入中那些让你凌晨三点还在看日志的典型问题4.1 现象微信公众号消息正常但用户发“你好”后无响应日志显示400 Bad Request原因微信回调URL配置时未勾选“消息加密”选项但代码中却按安全模式解析。微信会以明文XML发送而你的代码强行当AES密文解base64.b64decode报错。解决登录公众号后台 → 基本配置 → 消息加解密方式确认与代码中encoding_aes_key是否匹配。若用明文模式parse_request中跳过解密步骤直接解析原始XML。4.2 现象企业微信单聊正常群聊消息全部丢失Orchestrator日志无记录原因企业微信群聊消息的MsgType是text但事件类型字段是EventType值为MSG而你的代码只判断了MsgType忽略了群聊消息的EventType字段。解决检查企业微信文档群聊文本消息的完整路径是event.EventType MSG且event.MsgType text二者需同时满足。修改ww_adapter.py的判断逻辑。4.3 现象飞书消息偶尔收不到重试后又正常监控显示HTTP 429原因飞书对同一租户的API调用有频控默认1000次/小时但你的Orchestrator层在处理飞书消息时调用了user/get接口补全信息而该接口未做缓存高频触发频控。解决在Orchestrator层增加Redis缓存keyfeishu:open_id:{open_id}ttl1小时首次查询后存入后续直接读缓存。避免Adapter层调用任何外部API。4.4 现象钉钉消息解密后json.loads报UnicodeDecodeError: utf-8 codec cant decode byte 0xe4原因钉钉加密时使用UTF-8编码但AES解密后得到的字节流可能包含BOM头或非法序列更常见的是unpad后的字节未用.decode(utf-8)而是直接传给json.loads它期望str。解决确保解密后先decrypted_bytes.decode(utf-8)得到字符串再json.loads。添加异常捕获try: decrypted_str decrypted_bytes.decode(utf-8) decrypt_json json.loads(decrypted_str) except UnicodeDecodeError: # 尝试忽略错误字符 decrypted_str decrypted_bytes.decode(utf-8, errorsignore) decrypt_json json.loads(decrypted_str)4.5 现象所有平台消息都能收到但deepseek回复总是“我正在思考…”无实际内容原因vLLM部署时未正确设置--tensor-parallel-size导致GPU显存不足模型加载失败推理时返回空字符串。vLLM日志中会有CUDA out of memory但你的Adapter层无感知。解决检查vLLM启动日志确认INFO 05-20 10:23:45 llm_engine.py:123] Initialized with ...行是否存在若无说明模型未加载成功。降低--tensor-parallel-size如从4改为2或增加--gpu-memory-utilization 0.9。5. 模型调度中心Orchestrator实战如何让 deepseek 在多平台间保持一致的对话体验5.1 会话状态管理用Redis Hash存储而非单个String避免并发覆盖很多方案用redis.set(fsession:{key}, json.dumps(state))存储会话看似简单但在高并发下极易出错用户连续发两条消息A线程读state→加一轮对话→写回B线程同样操作B的写入会覆盖A的更新导致丢掉一轮对话。正确做法是用Redis Hash对每个字段原子操作# orchestrator.py import redis import json from datetime import timedelta class Orchestrator: def __init__(self, redis_url: str): self.r redis.from_url(redis_url, decode_responsesTrue) def get_session_state(self, session_key: str) - dict: # HGETALL 返回所有field-value对 data self.r.hgetall(fsession:{session_key}) if not data: return {history: [], user_profile: {}, last_active: 0} # 转换value为正确类型Redis存的是str state {} for k, v in data.items(): try: state[k] json.loads(v) if k in [history, user_profile] else v except (json.JSONDecodeError, TypeError): state[k] v return state def update_session_history(self, session_key: str, new_msg: dict, response: str): # 原子性追加LPUSH history但Hash不支持list所以用JSON存整个list state self.get_session_state(session_key) state[history].append({role: user, content: new_msg[content]}) state[history].append({role: assistant, content: response}) # 只保留最近10轮防止prompt过长 state[history] state[history][-20:] # 10轮20条 state[last_active] int(time.time()) # 写回Hash每个字段单独HSET原子 pipe self.r.pipeline() pipe.hset(fsession:{session_key}, history, json.dumps(state[history])) pipe.hset(fsession:{session_key}, last_active, state[last_active]) pipe.expire(fsession:{session_key}, timedelta(hours24)) pipe.execute()关键设计history字段存整个JSON数组每次更新都重写但因是单字段HSET不会与其他字段冲突expire设置24小时过期避免Redis内存无限增长last_active字段用于会话超时清理Orchestrator可起定时任务扫last_active now-3600的key。5.2 Prompt工程为不同平台定制system prompt但核心逻辑统一注入deepseek 对system prompt敏感。我们不为每个平台写不同prompt而是定义一个模板用变量注入平台特性SYSTEM_PROMPT_TEMPLATE 你是一个专业的企业服务助手当前在{platform}平台为用户服务。 - 若用户提问涉及公司内部流程如请假、报销、IT报修请严格依据《{kb_version}版知识库》回答不得编造。 - 当前用户身份{user_role}{user_dept}部门 - 当前会话已进行{turns}轮对话。 - 请用简洁、口语化中文回复避免长段落关键信息加粗如**审批通过**。 - 若问题超出知识库范围请回复“我暂时无法回答这个问题已为您转接人工客服。” def build_prompt(self, session_state: dict, user_input: str, platform: str) - str: # 从session_state中提取变量 user_role session_state.get(user_profile, {}).get(role, 普通员工) user_dept session_state.get(user_profile, {}).get(dept, 未知部门) turns len(session_state.get(history, [])) // 2 # 每轮含userassistant # 版本号从配置中心读取避免硬编码 kb_version self.config.get(kb_version, 2024Q2) system_prompt SYSTEM_PROMPT_TEMPLATE.format( platformplatform, kb_versionkb_version, user_roleuser_role, user_deptuser_dept, turnsturns ) # 拼接历史对话最多5轮避免超长 history session_state.get(history, [])[-10:] # 5轮 messages [{role: system, content: system_prompt}] messages.extend(history) messages.append({role: user, content: user_input}) # deepseek-v2-7b 使用chatml格式 prompt for msg in messages: if msg[role] system: prompt f|im_start|system\n{msg[content]}|im_end|\n elif msg[role] user: prompt f|im_start|user\n{msg[content]}|im_end|\n elif msg[role] assistant: prompt f|im_start|assistant\n{msg[content]}|im_end|\n prompt |im_start|assistant\n return prompt参数说明kb_version从Consul/Etcd等配置中心动态获取热更新无需重启history截取最后10条5轮保证prompt长度可控|im_start|标记是 deepseek-v2 的标准chatml格式必须严格匹配。5.3 模型调用vLLM部署与异步推理吞吐翻倍的关键配置我们用vLLM 0.4.2部署 deepseek-v2-7b关键配置如下# 启动命令8*A10G GPU python -m vllm.entrypoints.api_server \ --model deepseek-ai/deepseek-v2-7b \ --tensor-parallel-size 4 \ --pipeline-parallel-size 2 \ --dtype bfloat16 \ --max-num-seqs 256 \ --max-model-len 4096 \ --gpu-memory-utilization 0.92 \ --enforce-eager \ --port 8000参数详解--tensor-parallel-size 48卡分2组每组4卡做张量并行平衡通信与计算--pipeline-parallel-size 2将模型层切为2段流水线并行提升GPU利用率--gpu-memory-utilization 0.92显存占用率设为92%留8%给KV Cache突发增长--max-num-seqs 256最大并发请求数根据QPS压测结果设定我们实测200 QPS时稳定--enforce-eager禁用CUDA Graph避免某些动态shape报错deepseek-v2偶发。在Orchestrator中调用import aiohttp import asyncio async def call_vllm(self, prompt: str) - str: timeout aiohttp.ClientTimeout(total30) async with aiohttp.ClientSession(timeouttimeout) as session: try: async with session.post( http://vllm-server:8000/generate, json{ prompt: prompt, max_tokens: 512, temperature: 0.3, top_p: 0.85, repetition_penalty: 1.1, stop: [|im_end|] # 强制在结束标记停 } ) as resp: if resp.status ! 200: raise Exception(fvLLM error: {resp.status}) result await resp.json() return result[text].strip() except asyncio.TimeoutError: logger.warning(vLLM timeout, fallback to cached response) return 系统繁忙请稍后再试。 except Exception as e: logger.error(fvLLM call failed: {e}) return 我暂时无法回答这个问题。提示stop[|im_end|]是关键否则deepseek可能生成多余内容如|im_end|用户你好需后处理清洗。5.4 多平台响应组装同一个模型输出如何生成四种格式的HTTP响应Orchestrator产出response_text后交还给对应Adapter由其组装平台专属响应。以微信为例需返回XML# wechat_adapter.py (续) def build_response(self, msg: InternalMessage, response_text: str) - Response: # 微信要求XML响应且必须包含ToUserName/FromUserName/CreateTime/MsgType xml_template xml ToUserName![CDATA[{to_user}]]/ToUserName FromUserName![CDATA[{from_user}]]/FromUserName CreateTime{timestamp}/CreateTime MsgType![CDATA[text]]/MsgType Content![CDATA[{content}]]/Content /xml xml_body xml_template.format( to_usermsg.user_id, # 公众号中用户发消息我们回给他 from_useryour_app_id, # 替换为你的公众号AppID timestampint(time.time()), contentresponse_text[:2048] # 微信限制2048字 ) return Response( contentxml_body, media_typeapplication/xml )企业微信返回JSON# ww_adapter.py (续) def build_response(self, msg: InternalMessage, response_text: str) - JSONResponse: return JSONResponse({ msgtype: text, text: { content: response_text[:2048] } })飞书返回JSON需本文还有配套的精品资源点击获取