接手这个活儿之前先说明一个原则性问题直接操作个人微信账号做自动化是违反微信用户协议的高风险行为轻则功能异常重则账号被限制登录。所以这篇文章里我选择的是更稳妥、更适合技术探索的路线——基于微信官方生态提供的合法通道来构建我们的AI助手。这样既能满足“AI机器人”的核心诉求也能让大家玩得安心。这个项目最适合两类人一是想用Python做自动化应用、但对具体落地场景还比较迷茫的初学者二是天天泡在微信里、想给自己找个“第二大脑”的效率党。文章里我会把环境搭建、权限申请、对话逻辑、后台部署这几个环节全部走一遍代码可复制、步骤可追踪。1. 整体设计与方案选型1.1 为什么不用个人号 Hook 方案网上搜索“微信AI机器人”十有八九会指向hook个人微信的方案——模拟电脑端登录抓取收发消息的接口包再自己发消息回去。这类方案的原理其实是“劫持”了微信客户端的协议用脚本代替人手去点发送按钮。我不推荐这个方向理由很实在首先是封号风险微信的风控对异常登录、批量发送、非官方客户端检测得很严一套自动化脚本跑不了几个月号就可能没了其次是功能极不稳定微信客户端随便更新一个版本加密协议和消息结构就变了你的机器人就“失联”了最后是法律边界模糊私自拦截通信内容、绕过数据加密说严重点已经踩到《网络安全法》和《个人信息保护法》的红线。这锅没必要背。1.2 安全合规的技术栈选型换个思路同样是Python实现我们可以走官方认可的路径。服务端方案企业微信 自建应用企业微信本来就是腾讯推出的办公协作产品它提供了完整的API体系允许开发者自建应用、接收消息回调、回复员工消息。我们要做的就是让企业微信充当“消息中转站”员工向自建应用发消息文字、图片等企业微信服务器把消息内容推送到我们自己写的Web服务Web服务把文本转发给DeepSeek的API拿到AI回复再把回复推回企业微信由企业微信统一分发从用户视角来看这就是“在微信里找一个机器人聊天”从技术视角看所有操作都走官方接口没有中间商赚差价风险极低。接入DeepSeek的原因选DeepSeek而不是其他大模型主要看三点一是API调用成本非常低适合个人开发者反复调试二是文本理解能力在同梯队模型里表现过硬中文语境尤其顺手三是兼容OpenAI的接口格式代码迁移成本几乎为零。这三点决定了它就是个人项目快速上手的“最优解”。1.3 项目架构图纯文字版-------------- ------------------ ---------------- | 企业微信客户端 | ---- | 企业微信服务器 | ---- | 你的Web服务 | | (用户侧) | 回调 | (官方接口) | 推送 | (Python) |-----调用---- -------------- ------------------ ---------------- | v ---------------- | DeepSeek API | ----------------Web服务是整个链条的“心脏”它既是企业微信回调的接收者也是DeepSeek的调用方。消息进来组装上下文请求模型拿到答案再原路返回——这就是一整个闭环。2. 环境准备与依赖安装2.1 Python版本与虚拟环境项目用的是Python 3.10。为什么不建议低于3.10因为新版FastAPI和pydantic对类型注解的支持越来越好3.10的语法写起来也清爽。当然你要是机器上只有3.8也没法强求代码里我尽量不写太高级的特性保证大多数环境能直接跑。安装依赖之前强烈建议先建虚拟环境。试想一下你系统里可能有某个项目依赖Flask 2.0又有另一个项目要Flask 3.0直接装到全局那就是一场灾难。Python官方给出的最佳实践就是每个项目一个独立的虚拟环境互不干扰。python -m venv wechat_ai_env source wechat_ai_env/bin/activate # Windows下使用 wechat_ai_env\Scripts\activate2.2 核心依赖清单库名用途说明安装命令fastapiWeb框架处理回调请求pip install fastapiuvicornASGI服务器跑FastAPIpip install uvicornrequests调用外部HTTP接口pip install requestspython-dotenv读取环境变量文件pip install python-dotenv命令一条条跑pip install fastapi uvicorn requests python-dotenv装完之后建议大家顺手跑一句pip list看看版本是否正常避免后续出现类型不兼容的问题。2.3 DeepSeek API Key的准备这一步需要去DeepSeek的开放平台注册账号然后创建一个API Key。Key长这样sk-xxxxxxxxxxxxxxxxxxxx。请注意,Key属于敏感凭据绝对不能硬编码在源代码里不然传GitHub上等于把自家的门钥匙发给所有人。我的做法是放到.env文件里然后靠.gitignore把它排除掉# .env DEEPSEEK_API_KEYsk-xxxxx DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat在Python代码里用python-dotenv加载from dotenv import load_dotenv import os load_dotenv() DEEPSEEK_API_KEY os.getenv(DEEPSEEK_API_KEY)这里加一句DeepSeek的Base URL官方文档里写的是https://api.deepseek.com如果你用的是兼容模式也可能是https://api.deepseek.com/v1。以官方实时文档为准毕竟API版本更新频率不低。2.4 企业微信应用的基础配置企业微信里要干这么几件事注册一个企业哪怕就你一个人也算一个企业个人用它不需要任何费用。进入“应用管理”创建一个自建应用拿到AgentId应用ID和CorpId企业ID。在“接收消息服务器配置”里填上你的回调URL后面用ngrok或者服务器公网IP来暴露并生成Token和EncodingAESKey。在“权限管理”里给应用开启“接收消息”和“发送应用消息”的权限。前三步是硬性条件第四步比较容易漏——光建了应用却没开消息权限回调就形同虚设。ES密钥和Token也要记好下面代码里要用的。3. 核心逻辑与代码实现3.1 消息回调签名校验企业微信服务器在推送消息到我们的URL之前会先带上一大串签名参数msg_signature、timestamp、nonce、echostr等我们的服务器必须正确计算签名并原样返回echostr才能完成“握手”。这一步的本质是防止陌生人随便拿一个URL冒充企业微信服务器。计算方法将token、timestamp、nonce、echostr四个参数按字典序排序后拼接再做SHA-1加密对比与传过来的msg_signature是否一致。import hashlib def check_signature(token, msg_signature, timestamp, nonce, echostr): sort_list sorted([token, timestamp, nonce, echostr]) raw .join(sort_list).encode(utf-8) sha1 hashlib.sha1(raw).hexdigest() return sha1 msg_signature企业微信的签名校验用的是SHA-1这正是它文档明确写的。校验通过以后我们需要解密消息内容。这部分官方提供了加解密库wechatpy或WXBizMsgCrypt3直接用封装好的类就行不必自己撸加解密逻辑容易踩到字节序、Base64填充这种暗坑。3.2 FastAPI 搭建接收服务FastAPI做这类回调服务非常顺手自带Request校验和异步支持。我们定义一个/wechat/callback路由GET用于验证URLPOST用于接收真实消息。from fastapi import FastAPI, Request from fastapi.responses import PlainTextResponse import hashlib import xml.etree.ElementTree as ET app FastAPI() TOKEN your_token_here app.get(/wechat/callback) async def verify_url(msg_signature: str, timestamp: str, nonce: str, echostr: str): if check_signature(TOKEN, msg_signature, timestamp, nonce, echostr): return PlainTextResponse(echostr) return PlainTextResponse(signature error)细心一点就能发现这里是GET请求参数是Query String携带的。企业微信首次配置URL时就会发这样一次GET来验证你的服务器是否真实存在、是否懂签名规则。3.3 解密消息与解析XML企业微信推送的消息正文是一段加密后的XML结构形如xml ToUserName![CDATA[corpid]]/ToUserName Encrypt![CDATA[加密的密文]]/Encrypt AgentID![CDATA[agentid]]/AgentID /xml这里要用官方给到的WXBizMsgCrypt类。以wechatpy为例from wechatpy import WeChatClient from wechatpy.crypto import WeChatCrypto from wechatpy.exceptions import InvalidSignatureException但更常用的还是直接用WXBizMsgCrypt3这个官方示例类它提供了DecryptMsg方法。解密出来之后拿到的是明文XMLxml ToUserName![CDATA[corpid]]/ToUserName FromUserName![CDATA[userid]]/FromUserName CreateTime123456789/CreateTime MsgType![CDATA[text]]/MsgType Content![CDATA[你好]]/Content MsgId1234567890123456789/MsgId AgentID![CDATA[1000002]]/AgentID /xml拿到Content就是用户对我们的机器人说的话。3.4 调用 DeepSeek API 生成回复把用户消息丢给大模型拿到回复。这里我用requests库来发HTTP请求也可以换成OpenAI官方SDKDeepSeek接口兼容OpenAI格式所以直接import requests import json def call_deepseek(prompt): headers { Authorization: fBearer {DEEPSEEK_API_KEY}, Content-Type: application/json } payload { model: DEEPSEEK_MODEL, messages: [ {role: system, content: 你是一个友好、可靠的微信AI助手。}, {role: user, content: prompt} ], temperature: 0.7 } resp requests.post( f{DEEPSEEK_BASE_URL}/chat/completions, headersheaders, jsonpayload, timeout60 ) data resp.json() return data[choices][0][message][content]这里有个细节timeout60我很建议大家加上。大模型接口偶发慢响应如果不设超时你的Web服务就可能因为一个请求卡死所有后续请求全部排队表现就是“机器人半天不回话”。如果你希望AI具备上下文记忆那就要做会话管理——简单方案是维护一个字典键是用户ID值是消息历史列表请求时把这个列表的最近N条一起发给模型memory {} def get_prompt(user_id): history memory.get(user_id, []) return history[-10:] # 只取最近10轮 def update_memory(user_id, user_msg, ai_msg): if user_id not in memory: memory[user_id] [] memory[user_id].append({role: user, content: user_msg}) memory[user_id].append({role: assistant, content: ai_msg}) memory[user_id] memory[user_id][-20:] # 防止无限增长3.5 被动回复消息企业微信收到我们的回复有两种方式一种是在回调HTTP响应里被动回复直接在XML格式里包裹回复内容另一种是主动调用接口发送。被动回复的好处是即时、无需额外AccessToken但有时间限制——必须在5秒内响应否则必须走主动发送。考虑到DeepSeek接口一般要1-3秒才能返回大多数场景能赶上5秒窗口但网络抖动就可能超时。所以稳妥做法是用主动发送接口服务端先把回调请求的HTTP状态码置为200空响应再异步调用message/send接口把回复发给用户。def send_wechat_message(user_id, content): access_token get_access_token() url fhttps://qyapi.weixin.qq.com/cgi-bin/message/send?access_token{access_token} data { touser: user_id, msgtype: text, agentid: AGENT_ID, text: {content: content} } resp requests.post(url, jsondata) return resp.json()get_access_token这个函数需要定时刷新Token。企业微信的AccessToken有效期是7200秒要拿corpid和corpsecret去换。最好在内存里做缓存cached_token None cached_time 0 def get_access_token(): global cached_token, cached_time if time.time() - cached_time 7000 and cached_token: return cached_token resp requests.get( https://qyapi.weixin.qq.com/cgi-bin/gettoken, params{corpid: CORP_ID, corpsecret: SECRET} ) cached_token resp.json()[access_token] cached_time time.time() return cached_token3.6 完整回调处理流程串起来到这里几个模块都有了把它们组合到一个函数中。入口是FastAPI的POST回调先验签后解密提取消息内容调用DeepSeek再异步推送回去。app.post(/wechat/callback) async def handle_message(request: Request): params await request.query_params() msg_signature params.get(msg_signature, ) timestamp params.get(timestamp, ) nonce params.get(nonce, ) body (await request.body()).decode(utf-8) # 解密 crypt WXBizMsgCrypt(TOKEN, ENCODING_AES_KEY, CORP_ID) ret, decrypted_xml crypt.DecryptMsg(body, msg_signature, timestamp, nonce) if ret ! 0: return PlainTextResponse(decrypt fail) # 解析明文XML root ET.fromstring(decrypted_xml) from_user root.findtext(FromUserName) content root.findtext(Content) msg_type root.findtext(MsgType) # 仅处理文本消息 if msg_type text and content: reply call_deepseek_with_memory(from_user, content) # 主动发送 send_wechat_message(from_user, reply) return PlainTextResponse(success)这里故意不做被动回复而是走HTTP 200 主动发送为的就是稳定。4. 本地联调与外网暴露4.1 跑起服务uvicorn main:app --host 0.0.0.0 --port 8000 --reload加--reload是为了开发期改代码自动重启。确认日志里出现“Uvicorn running”再把http://你的公网域名/wechat/callback填到企业微信后台。4.2 ngrok 临时隧道本地服务默认是跑在localhost上的企业微信服务器无法直接访问你电脑的端口。这时可以借助ngrok或cpolar这类内网穿透工具把你本地的8000端口映射成一个公网HTTPS地址。ngrok http 8000命令跑起来后ngrok会显示一串https://xxxx.ap.ngrok.io复制到企业微信后台的URL填写框中就完事了。注意企业微信后台要求回调URL必须是HTTPSHTTP地址会被拒。ngrok免费版域名每次启动都会变所以这个方案只适合本地调试真正跑长期服务还是需要一台云服务器。4.3 服务器部署思路要让机器人7x24小时在线最省心的方案是买一台便宜的云服务器配置不用太高2核2G就够。把项目代码克隆上去装依赖写个systemd服务单元让进程守护再用Nginx反代到8000端口加上SSL证书。部署过程中最容易踩的坑是防火墙。云服务器安全组出站入站规则都要放行80/443/8000端口不然服务监听了外网还是不通。这个我当年排查了整整一个下午结果就是安全组规则漏了一条。5. 常见问题与排查记录5.1 签名校验失败一旦后台配置URL报“签名校验失败”先看三个地方Token是否抄错了有没有多余空格时间戳是否非正常——服务器时间不准会导致旧的时间戳被签名判断为过期代码里排序规则是否严格字典序。我自己有一次栽在使用f-string拼参数的时候把顺序拼错了排序是排了但拼接顺序没按排完的结果来自然对不上。5.2 回调收到了但机器人不回复比较隐蔽。服务日志里如果能看到回调进来但没有后续动作大概率是主动发送时报了错误码。常见的是60020应用无权限和40001AccessToken无效。对着错误码去看权限配置或者重新获取Token就好。5.3 DeepSeek调用报错或超时看返回是否带401API Key错、429限流、500服务端故障。429时合理安排重试策略可以加指数退避第一次等1秒第二次等2秒第三次4秒最高封顶。不要一股脑疯狂重试限流状态下越试越糟糕。5.4 一个容易被忽略的问题用户ID企业微信里人的userid和微信ID不是一回事。你在家里给自己测试回调里FromUserName是一长串企业微信内部的用户ID发送消息时touser必须填这个ID不是微信号也不是手机号。很多人第一次退信深挖一下才发现填错了对象。5.5 异常外呼的口径控制如果你的AI机器人会接触到多个用户建议在调用模型时加上“内容安全”System Prompt——本质上这跟前面说的合规一个道理让模型自动过滤危险请求。DeepSeek本身自带审核但多一层防护总归是好的。6. 一些经验和扩展想法真正跑通之后你可以做得更多增加语音能力企业微信的语音消息回调配合DeepSeek的文本接口加上一套语音转文字、文字转语音的管道就能让AI开口说话了。多轮上下文增强现在的记忆逻辑还只是简单的列表存最近几轮再接下去可以引入向量数据库做长期记忆让AI记住你的偏好。比如“上周你让我查过的那个项目的进展”下次它就能主动接上话。接入知识库企业内部文档、个人笔记通过RAG的方式喂给模型你的微信AI就能变成“懂公司业务”的助手而不只是一个通用聊天机器人。任务自动化收到特定指令比如“生成日报”“查天气”“创建待办”写一个函数调度器去执行。AI生成的不只是文本还能触发脚本。这些扩展技能说白了都是今天这个基础架构上的加装。核心链路——企业微信回调、签名验证、消息解密、调用模型、异步发送——你已经完整掌握了剩下就是想象力的问题。最后分享一个调试的小技巧刚开始联调你可以在服务里把所有环节的打点都print出来——收到回调时打印请求参数解密后打印明文XMLAI调用前打印prompt发送后打印响应码。这样一遍跑下来就能清楚看到每个环节的数据长什么样排查问题效率翻倍。等全部稳定了再换成日志库统一管理。项目通了之后你会发现自己对“人和AI之间的接口设计”这套东西有了完全不一样的体感。
基于企业微信与DeepSeek API构建合规微信AI机器人的Python实践
接手这个活儿之前先说明一个原则性问题直接操作个人微信账号做自动化是违反微信用户协议的高风险行为轻则功能异常重则账号被限制登录。所以这篇文章里我选择的是更稳妥、更适合技术探索的路线——基于微信官方生态提供的合法通道来构建我们的AI助手。这样既能满足“AI机器人”的核心诉求也能让大家玩得安心。这个项目最适合两类人一是想用Python做自动化应用、但对具体落地场景还比较迷茫的初学者二是天天泡在微信里、想给自己找个“第二大脑”的效率党。文章里我会把环境搭建、权限申请、对话逻辑、后台部署这几个环节全部走一遍代码可复制、步骤可追踪。1. 整体设计与方案选型1.1 为什么不用个人号 Hook 方案网上搜索“微信AI机器人”十有八九会指向hook个人微信的方案——模拟电脑端登录抓取收发消息的接口包再自己发消息回去。这类方案的原理其实是“劫持”了微信客户端的协议用脚本代替人手去点发送按钮。我不推荐这个方向理由很实在首先是封号风险微信的风控对异常登录、批量发送、非官方客户端检测得很严一套自动化脚本跑不了几个月号就可能没了其次是功能极不稳定微信客户端随便更新一个版本加密协议和消息结构就变了你的机器人就“失联”了最后是法律边界模糊私自拦截通信内容、绕过数据加密说严重点已经踩到《网络安全法》和《个人信息保护法》的红线。这锅没必要背。1.2 安全合规的技术栈选型换个思路同样是Python实现我们可以走官方认可的路径。服务端方案企业微信 自建应用企业微信本来就是腾讯推出的办公协作产品它提供了完整的API体系允许开发者自建应用、接收消息回调、回复员工消息。我们要做的就是让企业微信充当“消息中转站”员工向自建应用发消息文字、图片等企业微信服务器把消息内容推送到我们自己写的Web服务Web服务把文本转发给DeepSeek的API拿到AI回复再把回复推回企业微信由企业微信统一分发从用户视角来看这就是“在微信里找一个机器人聊天”从技术视角看所有操作都走官方接口没有中间商赚差价风险极低。接入DeepSeek的原因选DeepSeek而不是其他大模型主要看三点一是API调用成本非常低适合个人开发者反复调试二是文本理解能力在同梯队模型里表现过硬中文语境尤其顺手三是兼容OpenAI的接口格式代码迁移成本几乎为零。这三点决定了它就是个人项目快速上手的“最优解”。1.3 项目架构图纯文字版-------------- ------------------ ---------------- | 企业微信客户端 | ---- | 企业微信服务器 | ---- | 你的Web服务 | | (用户侧) | 回调 | (官方接口) | 推送 | (Python) |-----调用---- -------------- ------------------ ---------------- | v ---------------- | DeepSeek API | ----------------Web服务是整个链条的“心脏”它既是企业微信回调的接收者也是DeepSeek的调用方。消息进来组装上下文请求模型拿到答案再原路返回——这就是一整个闭环。2. 环境准备与依赖安装2.1 Python版本与虚拟环境项目用的是Python 3.10。为什么不建议低于3.10因为新版FastAPI和pydantic对类型注解的支持越来越好3.10的语法写起来也清爽。当然你要是机器上只有3.8也没法强求代码里我尽量不写太高级的特性保证大多数环境能直接跑。安装依赖之前强烈建议先建虚拟环境。试想一下你系统里可能有某个项目依赖Flask 2.0又有另一个项目要Flask 3.0直接装到全局那就是一场灾难。Python官方给出的最佳实践就是每个项目一个独立的虚拟环境互不干扰。python -m venv wechat_ai_env source wechat_ai_env/bin/activate # Windows下使用 wechat_ai_env\Scripts\activate2.2 核心依赖清单库名用途说明安装命令fastapiWeb框架处理回调请求pip install fastapiuvicornASGI服务器跑FastAPIpip install uvicornrequests调用外部HTTP接口pip install requestspython-dotenv读取环境变量文件pip install python-dotenv命令一条条跑pip install fastapi uvicorn requests python-dotenv装完之后建议大家顺手跑一句pip list看看版本是否正常避免后续出现类型不兼容的问题。2.3 DeepSeek API Key的准备这一步需要去DeepSeek的开放平台注册账号然后创建一个API Key。Key长这样sk-xxxxxxxxxxxxxxxxxxxx。请注意,Key属于敏感凭据绝对不能硬编码在源代码里不然传GitHub上等于把自家的门钥匙发给所有人。我的做法是放到.env文件里然后靠.gitignore把它排除掉# .env DEEPSEEK_API_KEYsk-xxxxx DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat在Python代码里用python-dotenv加载from dotenv import load_dotenv import os load_dotenv() DEEPSEEK_API_KEY os.getenv(DEEPSEEK_API_KEY)这里加一句DeepSeek的Base URL官方文档里写的是https://api.deepseek.com如果你用的是兼容模式也可能是https://api.deepseek.com/v1。以官方实时文档为准毕竟API版本更新频率不低。2.4 企业微信应用的基础配置企业微信里要干这么几件事注册一个企业哪怕就你一个人也算一个企业个人用它不需要任何费用。进入“应用管理”创建一个自建应用拿到AgentId应用ID和CorpId企业ID。在“接收消息服务器配置”里填上你的回调URL后面用ngrok或者服务器公网IP来暴露并生成Token和EncodingAESKey。在“权限管理”里给应用开启“接收消息”和“发送应用消息”的权限。前三步是硬性条件第四步比较容易漏——光建了应用却没开消息权限回调就形同虚设。ES密钥和Token也要记好下面代码里要用的。3. 核心逻辑与代码实现3.1 消息回调签名校验企业微信服务器在推送消息到我们的URL之前会先带上一大串签名参数msg_signature、timestamp、nonce、echostr等我们的服务器必须正确计算签名并原样返回echostr才能完成“握手”。这一步的本质是防止陌生人随便拿一个URL冒充企业微信服务器。计算方法将token、timestamp、nonce、echostr四个参数按字典序排序后拼接再做SHA-1加密对比与传过来的msg_signature是否一致。import hashlib def check_signature(token, msg_signature, timestamp, nonce, echostr): sort_list sorted([token, timestamp, nonce, echostr]) raw .join(sort_list).encode(utf-8) sha1 hashlib.sha1(raw).hexdigest() return sha1 msg_signature企业微信的签名校验用的是SHA-1这正是它文档明确写的。校验通过以后我们需要解密消息内容。这部分官方提供了加解密库wechatpy或WXBizMsgCrypt3直接用封装好的类就行不必自己撸加解密逻辑容易踩到字节序、Base64填充这种暗坑。3.2 FastAPI 搭建接收服务FastAPI做这类回调服务非常顺手自带Request校验和异步支持。我们定义一个/wechat/callback路由GET用于验证URLPOST用于接收真实消息。from fastapi import FastAPI, Request from fastapi.responses import PlainTextResponse import hashlib import xml.etree.ElementTree as ET app FastAPI() TOKEN your_token_here app.get(/wechat/callback) async def verify_url(msg_signature: str, timestamp: str, nonce: str, echostr: str): if check_signature(TOKEN, msg_signature, timestamp, nonce, echostr): return PlainTextResponse(echostr) return PlainTextResponse(signature error)细心一点就能发现这里是GET请求参数是Query String携带的。企业微信首次配置URL时就会发这样一次GET来验证你的服务器是否真实存在、是否懂签名规则。3.3 解密消息与解析XML企业微信推送的消息正文是一段加密后的XML结构形如xml ToUserName![CDATA[corpid]]/ToUserName Encrypt![CDATA[加密的密文]]/Encrypt AgentID![CDATA[agentid]]/AgentID /xml这里要用官方给到的WXBizMsgCrypt类。以wechatpy为例from wechatpy import WeChatClient from wechatpy.crypto import WeChatCrypto from wechatpy.exceptions import InvalidSignatureException但更常用的还是直接用WXBizMsgCrypt3这个官方示例类它提供了DecryptMsg方法。解密出来之后拿到的是明文XMLxml ToUserName![CDATA[corpid]]/ToUserName FromUserName![CDATA[userid]]/FromUserName CreateTime123456789/CreateTime MsgType![CDATA[text]]/MsgType Content![CDATA[你好]]/Content MsgId1234567890123456789/MsgId AgentID![CDATA[1000002]]/AgentID /xml拿到Content就是用户对我们的机器人说的话。3.4 调用 DeepSeek API 生成回复把用户消息丢给大模型拿到回复。这里我用requests库来发HTTP请求也可以换成OpenAI官方SDKDeepSeek接口兼容OpenAI格式所以直接import requests import json def call_deepseek(prompt): headers { Authorization: fBearer {DEEPSEEK_API_KEY}, Content-Type: application/json } payload { model: DEEPSEEK_MODEL, messages: [ {role: system, content: 你是一个友好、可靠的微信AI助手。}, {role: user, content: prompt} ], temperature: 0.7 } resp requests.post( f{DEEPSEEK_BASE_URL}/chat/completions, headersheaders, jsonpayload, timeout60 ) data resp.json() return data[choices][0][message][content]这里有个细节timeout60我很建议大家加上。大模型接口偶发慢响应如果不设超时你的Web服务就可能因为一个请求卡死所有后续请求全部排队表现就是“机器人半天不回话”。如果你希望AI具备上下文记忆那就要做会话管理——简单方案是维护一个字典键是用户ID值是消息历史列表请求时把这个列表的最近N条一起发给模型memory {} def get_prompt(user_id): history memory.get(user_id, []) return history[-10:] # 只取最近10轮 def update_memory(user_id, user_msg, ai_msg): if user_id not in memory: memory[user_id] [] memory[user_id].append({role: user, content: user_msg}) memory[user_id].append({role: assistant, content: ai_msg}) memory[user_id] memory[user_id][-20:] # 防止无限增长3.5 被动回复消息企业微信收到我们的回复有两种方式一种是在回调HTTP响应里被动回复直接在XML格式里包裹回复内容另一种是主动调用接口发送。被动回复的好处是即时、无需额外AccessToken但有时间限制——必须在5秒内响应否则必须走主动发送。考虑到DeepSeek接口一般要1-3秒才能返回大多数场景能赶上5秒窗口但网络抖动就可能超时。所以稳妥做法是用主动发送接口服务端先把回调请求的HTTP状态码置为200空响应再异步调用message/send接口把回复发给用户。def send_wechat_message(user_id, content): access_token get_access_token() url fhttps://qyapi.weixin.qq.com/cgi-bin/message/send?access_token{access_token} data { touser: user_id, msgtype: text, agentid: AGENT_ID, text: {content: content} } resp requests.post(url, jsondata) return resp.json()get_access_token这个函数需要定时刷新Token。企业微信的AccessToken有效期是7200秒要拿corpid和corpsecret去换。最好在内存里做缓存cached_token None cached_time 0 def get_access_token(): global cached_token, cached_time if time.time() - cached_time 7000 and cached_token: return cached_token resp requests.get( https://qyapi.weixin.qq.com/cgi-bin/gettoken, params{corpid: CORP_ID, corpsecret: SECRET} ) cached_token resp.json()[access_token] cached_time time.time() return cached_token3.6 完整回调处理流程串起来到这里几个模块都有了把它们组合到一个函数中。入口是FastAPI的POST回调先验签后解密提取消息内容调用DeepSeek再异步推送回去。app.post(/wechat/callback) async def handle_message(request: Request): params await request.query_params() msg_signature params.get(msg_signature, ) timestamp params.get(timestamp, ) nonce params.get(nonce, ) body (await request.body()).decode(utf-8) # 解密 crypt WXBizMsgCrypt(TOKEN, ENCODING_AES_KEY, CORP_ID) ret, decrypted_xml crypt.DecryptMsg(body, msg_signature, timestamp, nonce) if ret ! 0: return PlainTextResponse(decrypt fail) # 解析明文XML root ET.fromstring(decrypted_xml) from_user root.findtext(FromUserName) content root.findtext(Content) msg_type root.findtext(MsgType) # 仅处理文本消息 if msg_type text and content: reply call_deepseek_with_memory(from_user, content) # 主动发送 send_wechat_message(from_user, reply) return PlainTextResponse(success)这里故意不做被动回复而是走HTTP 200 主动发送为的就是稳定。4. 本地联调与外网暴露4.1 跑起服务uvicorn main:app --host 0.0.0.0 --port 8000 --reload加--reload是为了开发期改代码自动重启。确认日志里出现“Uvicorn running”再把http://你的公网域名/wechat/callback填到企业微信后台。4.2 ngrok 临时隧道本地服务默认是跑在localhost上的企业微信服务器无法直接访问你电脑的端口。这时可以借助ngrok或cpolar这类内网穿透工具把你本地的8000端口映射成一个公网HTTPS地址。ngrok http 8000命令跑起来后ngrok会显示一串https://xxxx.ap.ngrok.io复制到企业微信后台的URL填写框中就完事了。注意企业微信后台要求回调URL必须是HTTPSHTTP地址会被拒。ngrok免费版域名每次启动都会变所以这个方案只适合本地调试真正跑长期服务还是需要一台云服务器。4.3 服务器部署思路要让机器人7x24小时在线最省心的方案是买一台便宜的云服务器配置不用太高2核2G就够。把项目代码克隆上去装依赖写个systemd服务单元让进程守护再用Nginx反代到8000端口加上SSL证书。部署过程中最容易踩的坑是防火墙。云服务器安全组出站入站规则都要放行80/443/8000端口不然服务监听了外网还是不通。这个我当年排查了整整一个下午结果就是安全组规则漏了一条。5. 常见问题与排查记录5.1 签名校验失败一旦后台配置URL报“签名校验失败”先看三个地方Token是否抄错了有没有多余空格时间戳是否非正常——服务器时间不准会导致旧的时间戳被签名判断为过期代码里排序规则是否严格字典序。我自己有一次栽在使用f-string拼参数的时候把顺序拼错了排序是排了但拼接顺序没按排完的结果来自然对不上。5.2 回调收到了但机器人不回复比较隐蔽。服务日志里如果能看到回调进来但没有后续动作大概率是主动发送时报了错误码。常见的是60020应用无权限和40001AccessToken无效。对着错误码去看权限配置或者重新获取Token就好。5.3 DeepSeek调用报错或超时看返回是否带401API Key错、429限流、500服务端故障。429时合理安排重试策略可以加指数退避第一次等1秒第二次等2秒第三次4秒最高封顶。不要一股脑疯狂重试限流状态下越试越糟糕。5.4 一个容易被忽略的问题用户ID企业微信里人的userid和微信ID不是一回事。你在家里给自己测试回调里FromUserName是一长串企业微信内部的用户ID发送消息时touser必须填这个ID不是微信号也不是手机号。很多人第一次退信深挖一下才发现填错了对象。5.5 异常外呼的口径控制如果你的AI机器人会接触到多个用户建议在调用模型时加上“内容安全”System Prompt——本质上这跟前面说的合规一个道理让模型自动过滤危险请求。DeepSeek本身自带审核但多一层防护总归是好的。6. 一些经验和扩展想法真正跑通之后你可以做得更多增加语音能力企业微信的语音消息回调配合DeepSeek的文本接口加上一套语音转文字、文字转语音的管道就能让AI开口说话了。多轮上下文增强现在的记忆逻辑还只是简单的列表存最近几轮再接下去可以引入向量数据库做长期记忆让AI记住你的偏好。比如“上周你让我查过的那个项目的进展”下次它就能主动接上话。接入知识库企业内部文档、个人笔记通过RAG的方式喂给模型你的微信AI就能变成“懂公司业务”的助手而不只是一个通用聊天机器人。任务自动化收到特定指令比如“生成日报”“查天气”“创建待办”写一个函数调度器去执行。AI生成的不只是文本还能触发脚本。这些扩展技能说白了都是今天这个基础架构上的加装。核心链路——企业微信回调、签名验证、消息解密、调用模型、异步发送——你已经完整掌握了剩下就是想象力的问题。最后分享一个调试的小技巧刚开始联调你可以在服务里把所有环节的打点都print出来——收到回调时打印请求参数解密后打印明文XMLAI调用前打印prompt发送后打印响应码。这样一遍跑下来就能清楚看到每个环节的数据长什么样排查问题效率翻倍。等全部稳定了再换成日志库统一管理。项目通了之后你会发现自己对“人和AI之间的接口设计”这套东西有了完全不一样的体感。
