需求

• 本地部署好了一个大模型，需要让它对外提供服务，此时就需要一个前端了，可以是网页也可以是微信小程序，但这些都比较麻烦。而微信AI聊天助手不需要前端页面，对开发者友好；用户可以在私聊或者群聊中跟 AI 助手对话，助手可根据用户输入的内容做出响应回复，对用户也很友好，不需要点开网页或小程序就能随地使用。
• 问题：微信对插件的管控非常严格，github 仓库chatgpt-on-wechat 就可以提供一个外挂，运行之后会弹出二维码，扫描这个二维码后，当前账号就会被 AI 助手夺舍，当有人向该账号发送消息后，AI 助手会自动回复。但是自从 25 年 1 月，微信就能扫描到这种类型的外挂，并且会封禁微信账号。
• 可行的办法：企业微信允许创建机器人，该机器人默认使用官方提供的 deepseek，但是也允许自定义 API。这种方法比较合法，没有封禁的隐患，而且更加稳定。
• 不足之处：聊天机器人只能在企业微信中使用，虽然可以设置成允许企业成员在微信首发消息 （设置方法如下图，先登录管理者后台） ，但是在微信端只能与人类账号聊天，也只能看到人类账号间的聊天记录，机器人的发言是完全不会显示的。
  

企业微信开发者文档

• 智能机器人 部分
• 机器人配置 界面
• 企业微信用的加密方案
• python 版本的加密、解密文件下载路径

实现方法

说明：本示例采用的后端框架是 Python 的 FastAPI

1. 先打开如上图所示的机器人配置界面，然后随机生成 Token 和 Encoding-AESKey ，并把这个写到后端代码里。除了这两个外，还需要一个 ReceiveId ，官方文档关于它的说明是：

加解密库里，ReceiveId 在各个场景的含义不同：
企业应用的回调，表示corpid
第三方事件的回调，表示suiteid
个人主体的第三方应用的回调，ReceiveId是一个空字符串。

经过测试，ReceiveId 应该填空字符串！！！，不是企业 id，也不是机器人的 bot_id，最终写法如下：

sToken = "g6iXBXmwcXRZeGOMdJ67k"
sEncodingAESKey = "PeKaniWM88RZTZM7ilQTE7ZmYDGMMo0gXV1eEOyg6N4"
ReceiveId = ""

2. 下载python 版本的加密、解密文件下载（下载路径在这），解压缩之后把以下文件放到项目后台对应位置（位置可以随意，反正只要能 import 就行）

3. 如上图所示，我们需要为聊天机器人填一个 URL，它就是我们后端的 API 接口，填好它后，在点击保存的那一刻，企业微信会向我们所填的 URL 发送一个 get 请求，里面是一些加密的内容，它要求我们的 URL 能够解密然后返回明文，企业微信将校验明文，只有通过才能绑定成功。因此我们需要一个接口处理该 get 请求。

@router.get("/qywx/callback", summary="企业微信 URL 验证")
async def verify_wechat_url(msg_signature: str, timestamp: str, nonce: str, echostr: str):
    """
    这个接口只会管理员在企业微信后台点击【保存】时被调用一次。
    作用：验证你的服务器是否拥有正确的解密能力。
    """
    # 实例化官方提供的加解密类
    wxcpt = WXBizMsgCrypt(sToken, sEncodingAESKey, ReceiveId)
    # 调用 VerifyURL 方法进行解密，它会校验签名，并把 echostr(密文) 解密成 sEchoStr(明文)
    ret, sEchoStr = wxcpt.VerifyURL(msg_signature, timestamp, nonce, echostr)
    if ret != 0:
        print(f"验证失败，错误代码: {ret}")
        # 验证失败必须中断请求，企业微信会显示“请求不通过”
        raise HTTPException(status_code=403, detail="VerifyURL failed")
    # 返回解密后的明文，且不能带引号或其他JSON格式
    return Response(content=sEchoStr, media_type="text/plain")

4. 在校验成功之后，聊天机器人的自定义 API 就设置好了，如下图所示：
   
5. 随后按照企业微信官方文档的说法，当有用户询问或者艾特聊天助手的时候，企业微信会向我们的后台 API 发送一个 post 请求，并要求在 5s 之内收到回应。于是我们需要在收到请求之后，对请求的内容进行解密，得到用户输入内容后立刻返回响应，返回前需要创建一个异步执行的后台任务，让大模型根据用户输入去生成文本，当生成完了之后调用回调链接向企业微信发送请求把生成内容发送出去。原理图是：
   
6. 对应代码是：

bot_name="@农标通AI助手 "
@router.post("/qywx/callback", summary="接收消息")
async def receive_wechat_msg(
        request: Request,
        background_tasks: BackgroundTasks,
        msg_signature: str,
        timestamp: str,
        nonce: str
):
    """
    接收企业微信智能机器人的 JSON 格式回调
    """
    body_bytes = await request.body()
    sReqData = body_bytes.decode('utf-8')
    # print(f"【DEBUG】收到原始密文: {sReqData}")
    wxcpt = WXBizJsonMsgCrypt(sToken, sEncodingAESKey, ReceiveId)
    ret, sMsg = wxcpt.DecryptMsg(sReqData, msg_signature, timestamp, nonce)
    if ret != 0:
        print(f"【ERROR】解密失败，错误码: {ret}")
        # 即使失败也返回 success，避免微信无限重试，我们这边后台打印错误就好
        return Response(content="success")
    try:
        # print(f"【DEBUG】解密后的明文: {sMsg}")
        msg_data = json.loads(sMsg)  # 将 json 字符串转为 Python 字典
        msg_type = msg_data.get("msgtype")
        if msg_type == "text":
            text_dict = msg_data.get("text", {})
            content = text_dict.get("content")
            # 智能机器人特有的 ResponseUrl，用于异步回复
            response_url = msg_data.get("response_url")
            # 如果是群聊的话，一定会自带一个【"@农标通AI助手 "】，需要把它去除
            if msg_data.get("chattype")=="group":
                content = content.replace(bot_name, "").strip()
            print(f"【微信AI助手】收到用户内容: {content}")
            if content and response_url:
                # 参数 1 是回调函数名，参数 2 是回调地址，参数 3 是用户输入内容
                background_tasks.add_task(background_reply, response_url, content)
            else:
                print("内容或 URL 为空，无法回复")
    except Exception as e:
        print("【CRASH】捕获到异常！准备打印堆栈...", f"错误信息: {str(e)}")
    return Response(content="success")

其中回调函数 background_reply 负责将用户输入传入大模型中，得到生成文本，该函数定义如下：

async def background_reply(response_url: str, user_content: str):
    user_name = "企业微信用户"
    content, history = wechat_predict(get_wechat_model(), None, None, user_name, user_content, None, "通用模式",
                                      "../../prompty/wechat.prompty")
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "content": content
        }
    }
    async with httpx.AsyncClient() as client:
        try:
            # 主动发送请求
            await client.post(response_url, json=payload)
        except Exception as e:
            print(f"发送失败: {e}")

注意：需要下载一个库，它用于加密解密

pip intsall pycryptodome