简介一套基于大模型的智能对话机器人项目CoW面向需要将 AI 助手接入微信公众号、企业微信、飞书、钉钉等平台的开发者、运维人员或企业应用集成团队解决多渠道消息统一接入与智能应答的落地难题。项目内置多模型切换机制支持 GPT-4o、Claude、Gemini、DeepSeek、文心一言、讯飞星火、通义千问、ChatGLM、Kimi 等主流大模型并涵盖多轮会话上下文记忆、语音识别与合成、图片理解与生成、插件调用外部系统、基于私有知识库定制企业 AI 应用等实用功能。资源包共 199 个文件约 480KB以 141 个 Python 脚本为主体辅以 16 份 Markdown 文档、Shell 脚本、模板文件及 YAML/TOML 配置覆盖 Dockerfile、配置模板、聊天页面等关键模块目录结构清晰便于按需查阅和二次开发。已有 405 人学习下载适合希望快速搭建多端智能对话机器人、深入理解大模型接入与企业级部署的开发者从中可以获得可运行的代码框架、多平台接入配置思路以及语音、图像等扩展能力的完整实现参考。1. 智能对话机器人接入 DeepSeek四个入口一套核心内部做智能对话机器人最尴尬的不是模型不够聪明而是后端调通了 DeepSeek消息却送不到同事手边。公司里有人用企业微信、有人用钉钉对外还要挂一个微信公众号四个入口各有一套回调协议、加解密规则和消息格式。这套资源把这些问题一次解决基于大模型的智能对话机器人以 DeepSeek 为对话核心已打通微信公众号、企业微信应用、飞书、钉钉四个入口统一处理文本、语音和图片消息。适合正在做客服机器人、内部助理、办公自动化工具的开发者和实施工程师。接入细节比想象中多下面按对接顺序把代码、参数和踩过的坑拆开讲。2. 对接 DeepSeek封装统一对话服务再看消息归一化怎么设计2.1 为什么选 DeepSeek协议兼容、成本与边界接大模型做对话核心选型的核心就三个点接口好不好接、成本扛不扛得住、能力边界在哪里。DeepSeek 的接口格式兼容 OpenAI 的 chat completions 协议项目里之前用过其它 OpenAI 兼容服务的改一行 endpoint 和 key 就能切过来这对多平台接入来说是巨大的省事——四个入口共用一个对话服务模型层变动只影响一个函数。成本方面按 token 计费日常答疑和客服场景的量级下开销可控比自建 GPU 推理服务的固定成本要低很多适合中小团队起步。功能上 DeepSeek 支持 function calling做查库存、查订单这类工具调用是够用的但有一点要提醒工具调用结果需要在下一轮消息里立即返回不适合挂很长链路的外部工具设计对话流程时不要把这个当成万能。能力边界也要认清主模型是文本模型图片、语音不能直接喂进去必须先转成文字再处理这个边界决定了后面所有多模态消息的处理路径。2.2 封装统一对话服务chat_with_deepseek 与参数说明四个平台不能各自调模型否则上下文代码、API key 配置、超时处理会各写一遍后面改一个参数要动四个地方。我一般先把对话能力封装成一个函数输入 session_id 和用户消息输出大模型回复平台侧只负责解析消息、调用这个函数、把结果发回去。import requests import uuid from datetime import datetime # 全局 session 池session_id - [{role, content}] SESSION_POOL {} def chat_with_deepseek(session_id: str, user_message: str, system_prompt: str 你是一个友好的智能对话机器人) - str: # 上下文最多保留最近 20 条避免 token 超出窗口 history SESSION_POOL.get(session_id, []) if not history: history.append({role: system, content: system_prompt}) SESSION_POOL[session_id] history history.append({role: user, content: user_message}) # 超过 21 条时丢弃最早的 user/assistant 对保留 system 指令 if len(history) 21: history [history[0]] history[-20:] SESSION_POOL[session_id] history payload { model: deepseek-chat, # 换成你账户可用的模型标识 messages: history, temperature: 0.7, max_tokens: 1024, stream: False } headers { Authorization: Bearer API_KEY, Content-Type: application/json } resp requests.post( https://api.deepseek.com/chat/completions, # 以你服务商提供的 endpoint 为准 jsonpayload, headersheaders, timeout30 ) resp.raise_for_status() data resp.json() reply data[choices][0][message][content] history.append({role: assistant, content: reply}) # 把 token 统计打出来方便观察单轮成本 usage data.get(usage, {}) print(f[{datetime.now()}] session{session_id} fprompt_tokens{usage.get(prompt_tokens)} fcompletion_tokens{usage.get(completion_tokens)}) return reply这段代码把四个核心点集中处理了上下文是从内存池里按 session_id 取的超过 21 条就把最早的用户/助手消息对裁掉、只保留 system 指令请求超时设为 30 秒模型响应慢时不会把回调线程无限拖住每次调用把 usage 打出来跑几轮就能估算单条消息的成本。参数按场景调整model换成当前账户可用的模型标识temperature控制随机性做知识问答调到 0.20.3做闲聊调到 0.70.9max_tokens限制单次回复长度中文场景 1024 上下够用stream参数在回调场景下建议先关掉SSE 流式输出要额外处理协议等核心链路跑通再考虑。如果后续要本地部署大模型替换 DeepSeek改这一个函数的内部实现就行平台适配、消息归一化都不受影响。2.3 上下文管理session_id 拼接、TTL 与重置时机多轮对话的体感全靠上下文。session_id 的拼接规则直接影响会不会串台我一般用 platform 加 sender_id同一个用户在钉钉和企微里各自有独立会话互不干扰。生产环境里 session 对象不要放进程内存多 worker 部署时每个用户可能落到不同进程内存池就失效了常见做法是放 Redis 并设置过期时间。import redis import json r redis.Redis(host127.0.0.1, port6379, db0) SESSION_TTL 1800 # 30 分钟无交互自动清空 def build_session_id(platform: str, sender_id: str) - str: return f{platform}:{sender_id} def load_history(session_id: str): data r.get(session_id) return json.loads(data) if data else None def save_history(session_id: str, history: list) - None: r.set(session_id, json.dumps(history, ensure_asciiFalse), exSESSION_TTL) def reset_session(session_id: str) - None: r.delete(session_id)TTL 设 1800 秒是兼顾体验和存储成本用户半小时没说话再问就是新会话这段历史留在内存里也是浪费。reset_session在收到「清空」「重新开始」「新话题」这类指令时调用直接把 key 删掉下一次对话就是全新上下文。2.4 消息归一化把四类平台消息抽象成同一结构四个平台回调里的字段名、消息格式都不一样如果直接分发到业务代码每个分支都要写一遍解析。统一做法是先做一层归一化把原始消息转成同一结构后面所有逻辑只认这一个结构。def normalize_message(platform: str, raw: dict) - dict: # 各平台回调字段不同统一转换成内部结构 return { platform: platform, msg_id: raw.get(msg_id) or raw.get(MsgId), sender_id: raw.get(sender_id) or raw.get(FromUserName), msg_type: raw.get(msg_type) or raw.get(MsgType, ).lower(), content: raw.get(content) or raw.get(Content) or raw.get(Recognition), media_id: raw.get(media_id) or raw.get(MediaId), media_url: raw.get(media_url) or raw.get(PicUrl), }归一化后的核心字段是platform、sender_id、content、media_idplatform决定回推时走哪个适配器sender_id参与 session_id 拼接content是喂给大模型的文本media_id用于语音和图片的素材下载。之后整个管线就是normalize → 敏感词过滤 → 加载上下文 → chat_with_deepseek → 平台回推新平台接入只需要写一个适配器对话核心一个不动。3. 接入企业微信应用回调验签、消息解密与主动推送的落地细节3.1 企业微信应用与群机器人的区别企业微信接入容易把应用和群机器人搞混。群机器人配置一个 webhook 就能往群里发消息但不具备接收消息能力做不了问答只能单向推送告警。应用是完整形态用户在聊天框给应用发消息后台能收到事件回调也能主动推送消息。这套资源接的是企微应用形态原因很直接对话机器人必须有收有发只发不收只能做通知工具。3.2 回调验签与消息解密AES 和签名的处理顺序企微的难点在于消息是 AES 加密的 XML且签名把加密文本也纳入了计算。回调分两步平台先 GET 请求验证 URL返回 echostr 才能配置成功之后 POST 推送消息每次都要验签加解密。很多接入翻车都发生在这一步下面把处理顺序写清楚。import base64 import hashlib from Crypto.Cipher import AES from Crypto.Util.Padding import unpad def verify_and_decrypt(encrypt, msg_signature, timestamp, nonce): # 1. 签名校验参与字段按字典序拼接后 sha1 sort_list sorted([TOKEN, timestamp, nonce, encrypt]) if hashlib.sha1(.join(sort_list).encode()).hexdigest() ! msg_signature: raise PermissionError(signature mismatch) # 2. AES-CBC 解密EncodingAESKey 末尾补 做 base64 解码 key base64.b64decode(ENCODING_AES_KEY ) cipher AES.new(key, AES.MODE_CBC, key[:16]) plain unpad(cipher.decrypt(base64.b64decode(encrypt)), 16) # 3. 密文结构16 字节随机串 4 字节消息长度 消息体 corpid校验用 raw plain[16:] msg_len int.from_bytes(raw[:4], big) msg_xml raw[4:4 msg_len].decode(utf-8) return msg_xml签名校验时四个字段的字典序不能乱sha1 结果是 hex 字符串和企业微信后台的验签算法一致。解密使用 AES-CBCEncodingAESKey 需要末尾补一个再 base64 解码IV 就是 key 的前 16 字节。密文前 16 字节是随机串第 17 到 20 字节是消息长度大端序取出来截取消息体。尾部还带 corpid生产环境要校验防止伪造来源。URL 验证的接口写法如下注意返回的是解密后的 echostr 原文不是 XML也不是 JSON。app.route(/wecom/callback, methods[GET]) def wecom_verify(): args request.args echostr verify_and_decrypt( args.get(encrypt), args.get(msg_signature), args.get(timestamp), args.get(nonce), ) return echostr3.3 主动推送access_token 缓存与 message/send被动回复有 5 秒限制模型推理一慢就容易超时所以多数场景要靠主动推送。主动推送前要先拿 access_token这个 token 有 7200 秒有效期接口换 token 本身又有限频必须缓存复用。def get_access_token(corp_id: str, secret: str) - str: # access_token 有效期 7200 秒必须缓存复用否则频繁刷新会被限流 cache_key fwecom_token:{corp_id} token cache.get(cache_key) if token: return token resp requests.get( https://qyapi.weixin.qq.com/cgi-bin/gettoken, params{corpid: corp_id, corpsecret: secret}, timeout10 ) data resp.json() token data[access_token] cache.set(cache_key, token, exdata.get(expires_in, 7200) - 300) return token def push_wecom_text(agent_id: str, user_id: str, content: str): token get_access_token(CORP_ID, SECRET) resp requests.post( https://qyapi.weixin.qq.com/cgi-bin/message/send, params{access_token: token}, json{ touser: user_id, msgtype: text, agentid: agent_id, text: {content: content}, safe: 0 }, timeout10 ) print(resp.json()) # errcode 为 0 才算成功expires_in - 300是提前 5 分钟过期留足刷新时间避免临界点失效。touser是企微里的 UserID测试阶段先填自己不要用all避免打扰全公司。返回里errcode为 0 才代表成功非 0 要把errmsg打进日志。如果要支持图片、语音这类富媒体企微要求先上传素材拿到 media_idmessage/send里msgtype换成image、voicetext字段换成media_id。上传素材接口是/cgi-bin/media/upload传参临时文件路径资源包里直接有封装。3.4 回调里先响应再异步提交把超时问题挡在门外企业微信对回调响应时间限制在 5 秒左右DeepSeek 单轮推理经常在 3 秒上下加上网络耗时很容易超时。回调里直接同步调用模型平台会判定失败并重试用户看到的就是一条消息被重复回复。正确做法是回调里只做验签、解析、入队立刻返回固定响应后台 worker 调模型后通过主动推送接口把结果发回。from concurrent.futures import ThreadPoolExecutor executor ThreadPoolExecutor(max_workers8) app.route(/wecom/callback, methods[POST]) def wecom_message(): msg_xml verify_and_decrypt(...) # 参数从请求里取省略 root ET.fromstring(msg_xml) msg normalize_message(wecom, { msg_id: root.findtext(MsgId), sender_id: root.findtext(FromUserName), msg_type: root.findtext(MsgType), content: root.findtext(Content), media_id: root.findtext(MediaId), }) # 先返回成功平台不再重试异步线程去调 DeepSeek 再主动推给用户 executor.submit(handle_message, msg) return successexecutor.submit把消息处理丢到线程池回调线程立刻返回。线程池大小按消息量调整量大就换 Redis 队列加独立 worker 进程这套异步设计在公众号和飞书里同样适用。4. 多平台适配公众号、钉钉、飞书的接入差异与富媒体处理4.1 微信公众号被动回复、验签与素材下载公众号接入是四家里相对规整的签名算法与企业微信接近但两个差异要记牢。一个是 URL 验证只验signatureecho 回echostr即可消息体在明文模式下可以直接解析 XML安全模式下才需要 AES 解密。另一个是公众号的客服消息有 48 小时互动窗口超过时间就不能主动推送所以回复链路尽量在回调触发后的短时间内完成。app.route(/mp/callback, methods[GET]) def mp_verify(): args request.args calc hashlib.sha1(.join( sorted([TOKEN, args.get(timestamp), args.get(nonce)]) ).encode()).hexdigest() if calc args.get(signature): return args.get(echostr) return fail公众号验签参数比企微少一个encrypt因为 URL 验证时还没有消息体。排序拼接用字典序这个不能改。语音消息的原始格式一般是 amr 或 silk图片消息拿到media_id后调用media/get换取临时 URL 下载临时素材有有效期收到就要立刻拉取。4.2 钉钉webhook 推送与 Stream Mode 的选择钉钉有两种接入形态很多人分不清。自定义机器人只需要一个 webhook往群里发消息很顺手但它是单向的收不到用户发的内容而且 webhook 对单条消息大小、调用频率都有隐晦限制不适合做问答。交互式机器人要接企业内部机器人接收消息现在推荐 Stream Mode 长连接服务端把事件推给本地进程不需要公网回调地址本地开发调试最省事。def push_dingtalk_webhook(webhook_url: str, content: str): payload { msgtype: text, text: {content: content}, at: {atMobiles: [], isAtAll: False} } resp requests.post(webhook_url, jsonpayload, timeout10) return resp.json().get(errcode) 0webhook_url由群机器人创建后生成里面带签名参数atMobiles传手机号数组适合 特定值班人。单向告警、定时报告走这个方式足够交互问答必须走 Stream Mode。钉钉有个体验优势语音消息在事件推送里直接带recognition字段是服务端已经识别好的文字不用自己接 ASR。这一点比微信和企微省掉一整段转码链路。4.3 飞书应用机器人事件订阅与图片资源飞书应用机器人支持长连接和 webhook 两种事件订阅长连接免公网同样适合开发阶段。配置事件订阅时有个 challenge 机制首次保存订阅要响应 challenge 值否则后台保存不下来这是个容易忽视的步骤。图片消息给的是image_key需要调飞书 API 把图片二进制拉下来语音格式是 opus大部分 ASR 服务不认转成 wav 再送识别。飞书的富文本卡片功能很强大但第一版还是先接 text把链路跑通再加卡片。4.4 平台差异对照与统一处理管线平台推荐接入形态连接方式需公网回调主要消息类型语音原始格式微信公众号服务号被动回复Webhook 回调是文本/图片/语音/视频amr、silk企业微信应用企业内部应用Webhook 回调是文本/图片/语音/文件amr、silk钉钉企业内部机器人Stream 长连接否文本/图片/互动卡片语音自带识别文本飞书应用机器人长连接/Webhook可选文本/图片/音频/卡片opus归一化之后四个平台的消息收敛到同一结构platform、sender_id、content或解析后的文本。之后的管线完全一致消息进来先归一化敏感词过滤查上下文调chat_with_deepseek再由对应平台适配器把回复推回去。新增平台的时候只写适配器对话核心一个不动。def download_temp_media(platform: str, media_id: str, access_token: str): # 各平台临时素材接口路径不同统一封装后返回本地文件路径 if platform wecom: url https://qyapi.weixin.qq.com/cgi-bin/media/get elif platform mp: url https://api.weixin.qq.com/cgi-bin/media/get elif platform feishu: url https://open.feishu.cn/open-apis/im/v1/messages/{media_id}/resources/... # 微信公众号和企业微信 media/get 都靠 access_token 作为 query 参数 resp requests.get(url, params{access_token: access_token, media_id: media_id}, timeout15) return save_to_local(resp.content)这段代码是示意飞书的具体路径以官方文档为准但核心原则不变收到消息马上拉取素材存到本地对象存储后续 OCR、ASR 都用本地文件不要等用户追问时素材已经过期了。5. 接入避坑五个高频翻车点与排查路径多平台接入的坑不在模型而在协议细节。下面五条是按真实接入顺序整理的踩坑记录按现象、原因、解决三段写每条后面都能直接对应到调试日志里的报错。5.1 回调超时平台重试导致消息重复回复现象用户发一条消息机器人回了三遍聊天框里消息反复出现。原因回调线程里同步调 DeepSeek推理加网络耗时超过平台规定的回调响应时限公众号和企微都是 5 秒左右平台判定失败后自动重试推送同一事件。解决回调里只解密、解析、入队立即返回固定响应对话任务交给后台线程或消息队列处理完成后再走主动推送接口把结果发回去。同时用msg_id做幂等平台重试时直接丢弃重复事件。5.2 签名校验不过URL 验证不是返回 200 就行现象配置回调 URL 时一直报 token 验证失败看服务日志请求明明到了状态码也是 200。原因平台验签用的是 sha1 拼接排序拼接顺序、加密字段参与方式不对都会失败回调地址配在带 query 参数的路径上平台拼接签名时把额外参数也吃进去结果自然对不上。解决把参与签名的字段列出来按字典序排序后逐个比对回调地址单独用一个裸路径不要带任何额外参数。本地调试先绕过验签把解密函数用固定用例单测确认无误再开正式验签。5.3 语音消息转不了文字格式和获取路径都要管现象用户发语音机器人回复「没有听懂」或者下游 ASR 一直报格式错误。原因公众号和企业微信的语音是 amr 或 silk飞书是 opus直接拿原始格式送 ASR 必然翻车钉钉虽然自带识别文本但如果不取Recognition字段而是去下载音频反而绕了远路。解决按平台规则下载临时素材并转码成 wav 或 mp3 再送 ASR钉钉优先用消息体里的Recognition字段省一次转码和调用成本。转码工具用 ffmpeg 就行资源包里有现成的封装函数。5.4 access_token 限流多进程各自刷新带来的偶发失效现象主动推送偶尔报 invalid credential重启应用后恢复正常跑一阵又报。原因access_token 有 7200 秒有效期获取接口有频率限制。代码里 token 缓存放在进程内存时多 worker 每个进程各缓存一份同一 token 被刷新多次服务端认定调用异常。解决token 统一放 Redis加锁刷新避免多个 worker 同时去 gettoken。expires_in - 300提前过期刷新失败时回退到旧 token 继续用不至于直接断服。5.5 上下文串台session_id 没有拼用户维度现象A 用户问的问题B 用户收到了对应回答或者机器人的回复前言不搭后语明显是接上了别人聊到一半的上下文。原因session_id 只用了平台标识没有拼用户标识导致所有用户共用一个会话或者是不同代码路径各自建 session没有走统一的build_session_id。解决session_id 统一用platform:sender_id拼接在消息归一化时就把sender_id解析出来。多机部署时会话对象放 Redis 并设置过期时间不要用进程内字典。6. 端到端联调本地模拟回调、参数整定与上线检查清单6.1 本地模拟回调绕过验签直接测业务逻辑没有公网域名时也能把业务逻辑测完。做法是把验签函数在测试环境短路直接构造明文 XML 或 JSON 喂给解析函数平台回调没配好也能跑通代码。语音、图片分支同样可以模拟把media_id换成测试素材即可。curl -X POST http://127.0.0.1:8000/wecom/callback \ -H Content-Type: text/plain \ -d xmlMsgTypetext/MsgTypeContent你好大模型/ContentFromUserNameu_001/FromUserName/xml测试环境把路由里的验签逻辑先短路直接从 XML 里取字段走归一化、上下文、模型调用生产环境再开正式验签。不要一上来就配公网回调回调配好后出问题反而难定位。6.2 参数整定temperature、max_tokens 与 system prompt场景temperaturemax_tokens备注内部知识问答0.21024回答严谨少闲聊日常闲聊0.71024表现自然文案生成0.92048允许发散还有一个省钱提效的做法system prompt 里把机器人身份和场景说清楚比如「你是 IT 部门的客服助理回复控制在 200 字以内」能明显降低答非所问的概率也减少无效的max_tokens消耗。6.3 上线前检查清单与压测建议回调地址必须是公网 HTTPS路径不带多余 query验签逻辑在生产环境开启access_token 缓存放进 Redis不要停留在进程内存消息幂等用msg_id做去重防止平台重试造成重复回答敏感词过滤放在模型调用之前输出侧也要过一遍日志至少包含platform、sender_id、msg_id、耗时、usage 五个字段排查问题时没有这些全靠猜压测用脚本批量模拟用户消息观察线程池积压情况和 DeepSeek API 的限流错误码整套代码和配置模板都打包在资源里下载后按 README 一个个填四个平台的密钥和回调地址先通一个平台再往下加。我的血泪教训是第一次接企微时 access_token 缓存放在进程内存里线上多个 worker 互相踢掉 token排查了半个晚上最后发现是缓存位置的问题。从那以后我每次接新平台都强制先跑一遍本地模拟脚本把消息生命周期画出来再写适配器。希望帮到你。本文还有配套的精品资源点击获取 DeepSeek智能对话机器人四平台接入：公众号/企微/钉钉/飞书一套核心
简介一套基于大模型的智能对话机器人项目CoW面向需要将 AI 助手接入微信公众号、企业微信、飞书、钉钉等平台的开发者、运维人员或企业应用集成团队解决多渠道消息统一接入与智能应答的落地难题。项目内置多模型切换机制支持 GPT-4o、Claude、Gemini、DeepSeek、文心一言、讯飞星火、通义千问、ChatGLM、Kimi 等主流大模型并涵盖多轮会话上下文记忆、语音识别与合成、图片理解与生成、插件调用外部系统、基于私有知识库定制企业 AI 应用等实用功能。资源包共 199 个文件约 480KB以 141 个 Python 脚本为主体辅以 16 份 Markdown 文档、Shell 脚本、模板文件及 YAML/TOML 配置覆盖 Dockerfile、配置模板、聊天页面等关键模块目录结构清晰便于按需查阅和二次开发。已有 405 人学习下载适合希望快速搭建多端智能对话机器人、深入理解大模型接入与企业级部署的开发者从中可以获得可运行的代码框架、多平台接入配置思路以及语音、图像等扩展能力的完整实现参考。1. 智能对话机器人接入 DeepSeek四个入口一套核心内部做智能对话机器人最尴尬的不是模型不够聪明而是后端调通了 DeepSeek消息却送不到同事手边。公司里有人用企业微信、有人用钉钉对外还要挂一个微信公众号四个入口各有一套回调协议、加解密规则和消息格式。这套资源把这些问题一次解决基于大模型的智能对话机器人以 DeepSeek 为对话核心已打通微信公众号、企业微信应用、飞书、钉钉四个入口统一处理文本、语音和图片消息。适合正在做客服机器人、内部助理、办公自动化工具的开发者和实施工程师。接入细节比想象中多下面按对接顺序把代码、参数和踩过的坑拆开讲。2. 对接 DeepSeek封装统一对话服务再看消息归一化怎么设计2.1 为什么选 DeepSeek协议兼容、成本与边界接大模型做对话核心选型的核心就三个点接口好不好接、成本扛不扛得住、能力边界在哪里。DeepSeek 的接口格式兼容 OpenAI 的 chat completions 协议项目里之前用过其它 OpenAI 兼容服务的改一行 endpoint 和 key 就能切过来这对多平台接入来说是巨大的省事——四个入口共用一个对话服务模型层变动只影响一个函数。成本方面按 token 计费日常答疑和客服场景的量级下开销可控比自建 GPU 推理服务的固定成本要低很多适合中小团队起步。功能上 DeepSeek 支持 function calling做查库存、查订单这类工具调用是够用的但有一点要提醒工具调用结果需要在下一轮消息里立即返回不适合挂很长链路的外部工具设计对话流程时不要把这个当成万能。能力边界也要认清主模型是文本模型图片、语音不能直接喂进去必须先转成文字再处理这个边界决定了后面所有多模态消息的处理路径。2.2 封装统一对话服务chat_with_deepseek 与参数说明四个平台不能各自调模型否则上下文代码、API key 配置、超时处理会各写一遍后面改一个参数要动四个地方。我一般先把对话能力封装成一个函数输入 session_id 和用户消息输出大模型回复平台侧只负责解析消息、调用这个函数、把结果发回去。import requests import uuid from datetime import datetime # 全局 session 池session_id - [{role, content}] SESSION_POOL {} def chat_with_deepseek(session_id: str, user_message: str, system_prompt: str 你是一个友好的智能对话机器人) - str: # 上下文最多保留最近 20 条避免 token 超出窗口 history SESSION_POOL.get(session_id, []) if not history: history.append({role: system, content: system_prompt}) SESSION_POOL[session_id] history history.append({role: user, content: user_message}) # 超过 21 条时丢弃最早的 user/assistant 对保留 system 指令 if len(history) 21: history [history[0]] history[-20:] SESSION_POOL[session_id] history payload { model: deepseek-chat, # 换成你账户可用的模型标识 messages: history, temperature: 0.7, max_tokens: 1024, stream: False } headers { Authorization: Bearer API_KEY, Content-Type: application/json } resp requests.post( https://api.deepseek.com/chat/completions, # 以你服务商提供的 endpoint 为准 jsonpayload, headersheaders, timeout30 ) resp.raise_for_status() data resp.json() reply data[choices][0][message][content] history.append({role: assistant, content: reply}) # 把 token 统计打出来方便观察单轮成本 usage data.get(usage, {}) print(f[{datetime.now()}] session{session_id} fprompt_tokens{usage.get(prompt_tokens)} fcompletion_tokens{usage.get(completion_tokens)}) return reply这段代码把四个核心点集中处理了上下文是从内存池里按 session_id 取的超过 21 条就把最早的用户/助手消息对裁掉、只保留 system 指令请求超时设为 30 秒模型响应慢时不会把回调线程无限拖住每次调用把 usage 打出来跑几轮就能估算单条消息的成本。参数按场景调整model换成当前账户可用的模型标识temperature控制随机性做知识问答调到 0.20.3做闲聊调到 0.70.9max_tokens限制单次回复长度中文场景 1024 上下够用stream参数在回调场景下建议先关掉SSE 流式输出要额外处理协议等核心链路跑通再考虑。如果后续要本地部署大模型替换 DeepSeek改这一个函数的内部实现就行平台适配、消息归一化都不受影响。2.3 上下文管理session_id 拼接、TTL 与重置时机多轮对话的体感全靠上下文。session_id 的拼接规则直接影响会不会串台我一般用 platform 加 sender_id同一个用户在钉钉和企微里各自有独立会话互不干扰。生产环境里 session 对象不要放进程内存多 worker 部署时每个用户可能落到不同进程内存池就失效了常见做法是放 Redis 并设置过期时间。import redis import json r redis.Redis(host127.0.0.1, port6379, db0) SESSION_TTL 1800 # 30 分钟无交互自动清空 def build_session_id(platform: str, sender_id: str) - str: return f{platform}:{sender_id} def load_history(session_id: str): data r.get(session_id) return json.loads(data) if data else None def save_history(session_id: str, history: list) - None: r.set(session_id, json.dumps(history, ensure_asciiFalse), exSESSION_TTL) def reset_session(session_id: str) - None: r.delete(session_id)TTL 设 1800 秒是兼顾体验和存储成本用户半小时没说话再问就是新会话这段历史留在内存里也是浪费。reset_session在收到「清空」「重新开始」「新话题」这类指令时调用直接把 key 删掉下一次对话就是全新上下文。2.4 消息归一化把四类平台消息抽象成同一结构四个平台回调里的字段名、消息格式都不一样如果直接分发到业务代码每个分支都要写一遍解析。统一做法是先做一层归一化把原始消息转成同一结构后面所有逻辑只认这一个结构。def normalize_message(platform: str, raw: dict) - dict: # 各平台回调字段不同统一转换成内部结构 return { platform: platform, msg_id: raw.get(msg_id) or raw.get(MsgId), sender_id: raw.get(sender_id) or raw.get(FromUserName), msg_type: raw.get(msg_type) or raw.get(MsgType, ).lower(), content: raw.get(content) or raw.get(Content) or raw.get(Recognition), media_id: raw.get(media_id) or raw.get(MediaId), media_url: raw.get(media_url) or raw.get(PicUrl), }归一化后的核心字段是platform、sender_id、content、media_idplatform决定回推时走哪个适配器sender_id参与 session_id 拼接content是喂给大模型的文本media_id用于语音和图片的素材下载。之后整个管线就是normalize → 敏感词过滤 → 加载上下文 → chat_with_deepseek → 平台回推新平台接入只需要写一个适配器对话核心一个不动。3. 接入企业微信应用回调验签、消息解密与主动推送的落地细节3.1 企业微信应用与群机器人的区别企业微信接入容易把应用和群机器人搞混。群机器人配置一个 webhook 就能往群里发消息但不具备接收消息能力做不了问答只能单向推送告警。应用是完整形态用户在聊天框给应用发消息后台能收到事件回调也能主动推送消息。这套资源接的是企微应用形态原因很直接对话机器人必须有收有发只发不收只能做通知工具。3.2 回调验签与消息解密AES 和签名的处理顺序企微的难点在于消息是 AES 加密的 XML且签名把加密文本也纳入了计算。回调分两步平台先 GET 请求验证 URL返回 echostr 才能配置成功之后 POST 推送消息每次都要验签加解密。很多接入翻车都发生在这一步下面把处理顺序写清楚。import base64 import hashlib from Crypto.Cipher import AES from Crypto.Util.Padding import unpad def verify_and_decrypt(encrypt, msg_signature, timestamp, nonce): # 1. 签名校验参与字段按字典序拼接后 sha1 sort_list sorted([TOKEN, timestamp, nonce, encrypt]) if hashlib.sha1(.join(sort_list).encode()).hexdigest() ! msg_signature: raise PermissionError(signature mismatch) # 2. AES-CBC 解密EncodingAESKey 末尾补 做 base64 解码 key base64.b64decode(ENCODING_AES_KEY ) cipher AES.new(key, AES.MODE_CBC, key[:16]) plain unpad(cipher.decrypt(base64.b64decode(encrypt)), 16) # 3. 密文结构16 字节随机串 4 字节消息长度 消息体 corpid校验用 raw plain[16:] msg_len int.from_bytes(raw[:4], big) msg_xml raw[4:4 msg_len].decode(utf-8) return msg_xml签名校验时四个字段的字典序不能乱sha1 结果是 hex 字符串和企业微信后台的验签算法一致。解密使用 AES-CBCEncodingAESKey 需要末尾补一个再 base64 解码IV 就是 key 的前 16 字节。密文前 16 字节是随机串第 17 到 20 字节是消息长度大端序取出来截取消息体。尾部还带 corpid生产环境要校验防止伪造来源。URL 验证的接口写法如下注意返回的是解密后的 echostr 原文不是 XML也不是 JSON。app.route(/wecom/callback, methods[GET]) def wecom_verify(): args request.args echostr verify_and_decrypt( args.get(encrypt), args.get(msg_signature), args.get(timestamp), args.get(nonce), ) return echostr3.3 主动推送access_token 缓存与 message/send被动回复有 5 秒限制模型推理一慢就容易超时所以多数场景要靠主动推送。主动推送前要先拿 access_token这个 token 有 7200 秒有效期接口换 token 本身又有限频必须缓存复用。def get_access_token(corp_id: str, secret: str) - str: # access_token 有效期 7200 秒必须缓存复用否则频繁刷新会被限流 cache_key fwecom_token:{corp_id} token cache.get(cache_key) if token: return token resp requests.get( https://qyapi.weixin.qq.com/cgi-bin/gettoken, params{corpid: corp_id, corpsecret: secret}, timeout10 ) data resp.json() token data[access_token] cache.set(cache_key, token, exdata.get(expires_in, 7200) - 300) return token def push_wecom_text(agent_id: str, user_id: str, content: str): token get_access_token(CORP_ID, SECRET) resp requests.post( https://qyapi.weixin.qq.com/cgi-bin/message/send, params{access_token: token}, json{ touser: user_id, msgtype: text, agentid: agent_id, text: {content: content}, safe: 0 }, timeout10 ) print(resp.json()) # errcode 为 0 才算成功expires_in - 300是提前 5 分钟过期留足刷新时间避免临界点失效。touser是企微里的 UserID测试阶段先填自己不要用all避免打扰全公司。返回里errcode为 0 才代表成功非 0 要把errmsg打进日志。如果要支持图片、语音这类富媒体企微要求先上传素材拿到 media_idmessage/send里msgtype换成image、voicetext字段换成media_id。上传素材接口是/cgi-bin/media/upload传参临时文件路径资源包里直接有封装。3.4 回调里先响应再异步提交把超时问题挡在门外企业微信对回调响应时间限制在 5 秒左右DeepSeek 单轮推理经常在 3 秒上下加上网络耗时很容易超时。回调里直接同步调用模型平台会判定失败并重试用户看到的就是一条消息被重复回复。正确做法是回调里只做验签、解析、入队立刻返回固定响应后台 worker 调模型后通过主动推送接口把结果发回。from concurrent.futures import ThreadPoolExecutor executor ThreadPoolExecutor(max_workers8) app.route(/wecom/callback, methods[POST]) def wecom_message(): msg_xml verify_and_decrypt(...) # 参数从请求里取省略 root ET.fromstring(msg_xml) msg normalize_message(wecom, { msg_id: root.findtext(MsgId), sender_id: root.findtext(FromUserName), msg_type: root.findtext(MsgType), content: root.findtext(Content), media_id: root.findtext(MediaId), }) # 先返回成功平台不再重试异步线程去调 DeepSeek 再主动推给用户 executor.submit(handle_message, msg) return successexecutor.submit把消息处理丢到线程池回调线程立刻返回。线程池大小按消息量调整量大就换 Redis 队列加独立 worker 进程这套异步设计在公众号和飞书里同样适用。4. 多平台适配公众号、钉钉、飞书的接入差异与富媒体处理4.1 微信公众号被动回复、验签与素材下载公众号接入是四家里相对规整的签名算法与企业微信接近但两个差异要记牢。一个是 URL 验证只验signatureecho 回echostr即可消息体在明文模式下可以直接解析 XML安全模式下才需要 AES 解密。另一个是公众号的客服消息有 48 小时互动窗口超过时间就不能主动推送所以回复链路尽量在回调触发后的短时间内完成。app.route(/mp/callback, methods[GET]) def mp_verify(): args request.args calc hashlib.sha1(.join( sorted([TOKEN, args.get(timestamp), args.get(nonce)]) ).encode()).hexdigest() if calc args.get(signature): return args.get(echostr) return fail公众号验签参数比企微少一个encrypt因为 URL 验证时还没有消息体。排序拼接用字典序这个不能改。语音消息的原始格式一般是 amr 或 silk图片消息拿到media_id后调用media/get换取临时 URL 下载临时素材有有效期收到就要立刻拉取。4.2 钉钉webhook 推送与 Stream Mode 的选择钉钉有两种接入形态很多人分不清。自定义机器人只需要一个 webhook往群里发消息很顺手但它是单向的收不到用户发的内容而且 webhook 对单条消息大小、调用频率都有隐晦限制不适合做问答。交互式机器人要接企业内部机器人接收消息现在推荐 Stream Mode 长连接服务端把事件推给本地进程不需要公网回调地址本地开发调试最省事。def push_dingtalk_webhook(webhook_url: str, content: str): payload { msgtype: text, text: {content: content}, at: {atMobiles: [], isAtAll: False} } resp requests.post(webhook_url, jsonpayload, timeout10) return resp.json().get(errcode) 0webhook_url由群机器人创建后生成里面带签名参数atMobiles传手机号数组适合 特定值班人。单向告警、定时报告走这个方式足够交互问答必须走 Stream Mode。钉钉有个体验优势语音消息在事件推送里直接带recognition字段是服务端已经识别好的文字不用自己接 ASR。这一点比微信和企微省掉一整段转码链路。4.3 飞书应用机器人事件订阅与图片资源飞书应用机器人支持长连接和 webhook 两种事件订阅长连接免公网同样适合开发阶段。配置事件订阅时有个 challenge 机制首次保存订阅要响应 challenge 值否则后台保存不下来这是个容易忽视的步骤。图片消息给的是image_key需要调飞书 API 把图片二进制拉下来语音格式是 opus大部分 ASR 服务不认转成 wav 再送识别。飞书的富文本卡片功能很强大但第一版还是先接 text把链路跑通再加卡片。4.4 平台差异对照与统一处理管线平台推荐接入形态连接方式需公网回调主要消息类型语音原始格式微信公众号服务号被动回复Webhook 回调是文本/图片/语音/视频amr、silk企业微信应用企业内部应用Webhook 回调是文本/图片/语音/文件amr、silk钉钉企业内部机器人Stream 长连接否文本/图片/互动卡片语音自带识别文本飞书应用机器人长连接/Webhook可选文本/图片/音频/卡片opus归一化之后四个平台的消息收敛到同一结构platform、sender_id、content或解析后的文本。之后的管线完全一致消息进来先归一化敏感词过滤查上下文调chat_with_deepseek再由对应平台适配器把回复推回去。新增平台的时候只写适配器对话核心一个不动。def download_temp_media(platform: str, media_id: str, access_token: str): # 各平台临时素材接口路径不同统一封装后返回本地文件路径 if platform wecom: url https://qyapi.weixin.qq.com/cgi-bin/media/get elif platform mp: url https://api.weixin.qq.com/cgi-bin/media/get elif platform feishu: url https://open.feishu.cn/open-apis/im/v1/messages/{media_id}/resources/... # 微信公众号和企业微信 media/get 都靠 access_token 作为 query 参数 resp requests.get(url, params{access_token: access_token, media_id: media_id}, timeout15) return save_to_local(resp.content)这段代码是示意飞书的具体路径以官方文档为准但核心原则不变收到消息马上拉取素材存到本地对象存储后续 OCR、ASR 都用本地文件不要等用户追问时素材已经过期了。5. 接入避坑五个高频翻车点与排查路径多平台接入的坑不在模型而在协议细节。下面五条是按真实接入顺序整理的踩坑记录按现象、原因、解决三段写每条后面都能直接对应到调试日志里的报错。5.1 回调超时平台重试导致消息重复回复现象用户发一条消息机器人回了三遍聊天框里消息反复出现。原因回调线程里同步调 DeepSeek推理加网络耗时超过平台规定的回调响应时限公众号和企微都是 5 秒左右平台判定失败后自动重试推送同一事件。解决回调里只解密、解析、入队立即返回固定响应对话任务交给后台线程或消息队列处理完成后再走主动推送接口把结果发回去。同时用msg_id做幂等平台重试时直接丢弃重复事件。5.2 签名校验不过URL 验证不是返回 200 就行现象配置回调 URL 时一直报 token 验证失败看服务日志请求明明到了状态码也是 200。原因平台验签用的是 sha1 拼接排序拼接顺序、加密字段参与方式不对都会失败回调地址配在带 query 参数的路径上平台拼接签名时把额外参数也吃进去结果自然对不上。解决把参与签名的字段列出来按字典序排序后逐个比对回调地址单独用一个裸路径不要带任何额外参数。本地调试先绕过验签把解密函数用固定用例单测确认无误再开正式验签。5.3 语音消息转不了文字格式和获取路径都要管现象用户发语音机器人回复「没有听懂」或者下游 ASR 一直报格式错误。原因公众号和企业微信的语音是 amr 或 silk飞书是 opus直接拿原始格式送 ASR 必然翻车钉钉虽然自带识别文本但如果不取Recognition字段而是去下载音频反而绕了远路。解决按平台规则下载临时素材并转码成 wav 或 mp3 再送 ASR钉钉优先用消息体里的Recognition字段省一次转码和调用成本。转码工具用 ffmpeg 就行资源包里有现成的封装函数。5.4 access_token 限流多进程各自刷新带来的偶发失效现象主动推送偶尔报 invalid credential重启应用后恢复正常跑一阵又报。原因access_token 有 7200 秒有效期获取接口有频率限制。代码里 token 缓存放在进程内存时多 worker 每个进程各缓存一份同一 token 被刷新多次服务端认定调用异常。解决token 统一放 Redis加锁刷新避免多个 worker 同时去 gettoken。expires_in - 300提前过期刷新失败时回退到旧 token 继续用不至于直接断服。5.5 上下文串台session_id 没有拼用户维度现象A 用户问的问题B 用户收到了对应回答或者机器人的回复前言不搭后语明显是接上了别人聊到一半的上下文。原因session_id 只用了平台标识没有拼用户标识导致所有用户共用一个会话或者是不同代码路径各自建 session没有走统一的build_session_id。解决session_id 统一用platform:sender_id拼接在消息归一化时就把sender_id解析出来。多机部署时会话对象放 Redis 并设置过期时间不要用进程内字典。6. 端到端联调本地模拟回调、参数整定与上线检查清单6.1 本地模拟回调绕过验签直接测业务逻辑没有公网域名时也能把业务逻辑测完。做法是把验签函数在测试环境短路直接构造明文 XML 或 JSON 喂给解析函数平台回调没配好也能跑通代码。语音、图片分支同样可以模拟把media_id换成测试素材即可。curl -X POST http://127.0.0.1:8000/wecom/callback \ -H Content-Type: text/plain \ -d xmlMsgTypetext/MsgTypeContent你好大模型/ContentFromUserNameu_001/FromUserName/xml测试环境把路由里的验签逻辑先短路直接从 XML 里取字段走归一化、上下文、模型调用生产环境再开正式验签。不要一上来就配公网回调回调配好后出问题反而难定位。6.2 参数整定temperature、max_tokens 与 system prompt场景temperaturemax_tokens备注内部知识问答0.21024回答严谨少闲聊日常闲聊0.71024表现自然文案生成0.92048允许发散还有一个省钱提效的做法system prompt 里把机器人身份和场景说清楚比如「你是 IT 部门的客服助理回复控制在 200 字以内」能明显降低答非所问的概率也减少无效的max_tokens消耗。6.3 上线前检查清单与压测建议回调地址必须是公网 HTTPS路径不带多余 query验签逻辑在生产环境开启access_token 缓存放进 Redis不要停留在进程内存消息幂等用msg_id做去重防止平台重试造成重复回答敏感词过滤放在模型调用之前输出侧也要过一遍日志至少包含platform、sender_id、msg_id、耗时、usage 五个字段排查问题时没有这些全靠猜压测用脚本批量模拟用户消息观察线程池积压情况和 DeepSeek API 的限流错误码整套代码和配置模板都打包在资源里下载后按 README 一个个填四个平台的密钥和回调地址先通一个平台再往下加。我的血泪教训是第一次接企微时 access_token 缓存放在进程内存里线上多个 worker 互相踢掉 token排查了半个晚上最后发现是缓存位置的问题。从那以后我每次接新平台都强制先跑一遍本地模拟脚本把消息生命周期画出来再写适配器。希望帮到你。本文还有配套的精品资源点击获取