把 AI 小助手接入企业微信：用一个回调接口做群聊机器人实战篇

优质文章学习记录

— 9 min read

你也许已经有了一个「看起来还挺像样」的 AI 小助手服务，比如：

  • 有 HTTP 接口 /v1/chat；
  • 能识别不同 Skill（待办、日报、FAQ 等）；
  • 甚至已经有网页版前端。

但现实是：同事们每天真正打开的是企业微信，很少会专门去打开一个新网页跟机器人聊天。

这篇文章就做一件很实用的小事：

在不动你现有 AI 服务核心逻辑的前提下，
用一个企业微信“回调接口”，
把它变成「群聊里的 @ 机器人」。

一、整体思路：后端不重写，只加一层「翻译器」

假设你现在的 AI 服务长这样：

  • 接口：POST /v1/chat

返回：

{ "answer": "上午开会，下午写代码……", "skill": "daily_plan", "duration_ms": 1234 }

请求体：

{ "question": "今天帮我规划下工作", "session_id": "xxx", "user_id": "u_123", "user_level": "normal" }

我们要做的只是：

  1. 在企业微信后台配置一个消息回调 URL（比如 /wechat/callback）；
  2. 在后端写一个超薄适配层：
    • 收到企业微信推来的加密 XML；
    • 解密 → 提取出「用户 ID + 文本内容」；
    • 把「文本内容」转成 question，连同 user_id 一起转发给 /v1/chat；
    • 拿到 answer，再按照企业微信要求的格式加密返回。

可以简单理解为：

企业微信：提供 UI 和消息推送
你的 AI 小助手：负责“想答案”
中间这个回调接口：负责「翻译 + 转发 + 回填」

二、企业微信端：你必须准备的 3 个关键参数

在写代码前，先在企业微信后台把几个关键信息准备好。

2.1 创建自建应用 / 机器人

  1. 管理员登录企业微信后台；
  2. 进入【应用管理】 →【自建】→【创建应用】；
  3. 按提示填写名称、图标即可。

创建完成后，记下：

  • CorpID（企业 ID）
  • AgentID（应用 ID）
  • Secret（应用密钥）
其中 CorpID 稍后会和 Token、EncodingAESKey 一起用于解密；
Secret 一般用于你「主动调用企业微信接口」，本篇只做回调，不用到。

2.2 配置消息回调

在刚刚创建的自建应用里，找到：

  • 「接收消息」或「回调配置」模块

配置三项：

  1. URL：你后端可公网访问的地址，比如
    https://your-domain.com/wechat/callback
  2. Token：自己随便写一个，比如 my_wechat_token_2024
  3. EncodingAESKey：企业微信自动生成的一串 43 位字符串

这三项记下来，对应到代码里的环境变量：

环境变量名对应企业微信配置
WECHAT_TOKENToken
WECHAT_ENCODING_AES_KEYEncodingAESKey
WECHAT_CORP_ID企业 CorpID

三、项目结构：只新增一个 wechat_adapter.py

假设你当前项目结构：

project/ ├── app/ │ ├── server.py # 已有：包含 /v1/chat │ ├── ... ├── docker-compose.yml └── Dockerfile

现在只需要：

  • 在 app/ 下加一个 wechat_adapter.py，做企业微信适配；
  • 在 server.py 里把这个 router 挂起来。

四、后端实现：FastAPI + wechatpy

这里用 FastAPI 举例（Flask 思路完全一样），并使用 wechatpy 来处理消息加解密，避免自己手写 AES。

4.1 安装依赖

pip install fastapi uvicorn wechatpy httpx

（如果你已经有 FastAPI，只需加 wechatpy 和 httpx 即可。）

4.2 编写 wechat_adapter.py

# app/wechat_adapter.py import os from fastapi import APIRouter, Request from fastapi.responses import PlainTextResponse from wechatpy.enterprise.crypto import WeChatCrypto from wechatpy.enterprise import parse_message, create_reply import httpx # 从环境变量读取配置 WECHAT_TOKEN = os.getenv("WECHAT_TOKEN") WECHAT_ENCODING_AES_KEY = os.getenv("WECHAT_ENCODING_AES_KEY") WECHAT_CORP_ID = os.getenv("WECHAT_CORP_ID") # 你自己的 AI 服务接口 AGENT_BACKEND_URL = os.getenv("AGENT_BACKEND_URL", "http://localhost:8000/v1/chat") router = APIRouter() @router.get("/callback") async def wechat_verify( msg_signature: str, timestamp: str, nonce: str, echostr: str ): """ 企业微信在配置回调 URL 时，会发一个 GET 请求校验。 要做的就是：用 WeChatCrypto 解密 echostr，然后原样返回。 """ crypto = WeChatCrypto(WECHAT_TOKEN, WECHAT_ENCODING_AES_KEY, WECHAT_CORP_ID) try: echostr_decoded = crypto.decrypt_message(echostr, msg_signature, timestamp, nonce) return PlainTextResponse(content=echostr_decoded) except Exception: # 验签失败就返回空，企业微信会认为不通过 return PlainTextResponse(content="") @router.post("/callback") async def wechat_callback(request: Request): """ 真正收消息的入口。 1. 解密 XML 2. 解析文本内容 3. 调用内部 AI 服务 4. 构造并加密回复 """ query = request.query_params msg_signature = query.get("msg_signature") timestamp = query.get("timestamp") nonce = query.get("nonce") raw_xml = await request.body() crypto = WeChatCrypto(WECHAT_TOKEN, WECHAT_ENCODING_AES_KEY, WECHAT_CORP_ID) try: msg_xml = crypto.decrypt_message( raw_xml, msg_signature, timestamp, nonce ) except Exception: return PlainTextResponse(content="") # 解析企业微信 XML msg = parse_message(msg_xml) # 只处理文本消息 if msg.type != "text": reply = create_reply("暂时只支持文本消息哦～", msg) encrypted = crypto.encrypt_message(reply.render(), nonce, timestamp) return PlainTextResponse(content=encrypted) # 用户ID（企业微信里的 userid） user_id = msg.source content = msg.content.strip() # [可选] 去掉前面的 @机器人 文本，只保留真正问题 # 例如："@AI小助手 帮我写日报" → "帮我写日报" if content.startswith("@"): idx = content.find(" ") if idx != -1: content = content[idx + 1 :].strip() # 调用你自己的 AI 接口 async with httpx.AsyncClient(timeout=30) as client: try: resp = await client.post( AGENT_BACKEND_URL, json={ "question": content, "session_id": f"wx_{user_id}", "user_id": f"wx_{user_id}", "user_level": "normal", } ) data = resp.json() answer = data.get("answer", "后端没有返回 answer 字段，请检查。") except Exception as e: answer = f"后端服务异常：{e}" # 构造回复 XML 并加密 reply = create_reply(answer, msg) encrypted = crypto.encrypt_message(reply.render(), nonce, timestamp) return PlainTextResponse(content=encrypted)

4.3 在 server.py 中挂载 router

# app/server.py from fastapi import FastAPI from app.wechat_adapter import router as wechat_router app = FastAPI(title="My AI Agent Service") # 原有 /v1/chat 等接口 # @app.post("/v1/chat") # async def chat(...): # ... # 新增企业微信路由 app.include_router(wechat_router, prefix="/wechat")

本地启动（开发阶段）：

uvicorn app.server:app --host 0.0.0.0 --port 8000 --reload

现在，你有了如下几个接口：

  • POST /v1/chat：原来的 AI 接口
  • GET /wechat/callback：用于企业微信验证 URL
  • POST /wechat/callback：用于接收真实消息并回复

五、让企业微信找到你的服务：Nginx / 域名配置

企业微信需要访问你服务器的公网地址，所以一般会放在 Nginx 之后。

假设你有一个域名 https://bot.example.com，Nginx 配置大致如下（核心是 /wechat/ 这段）：

server { listen 80; server_name bot.example.com; location / { proxy_pass http://app:8000/; } # 专门给企业微信回调用 location /wechat/ { proxy_pass http://app:8000/wechat/; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } }

重载 Nginx：

nginx -s reload

然后在企业微信后台把回调 URL 填写为：

https://bot.example.com/wechat/callback

保存时，企业微信会向这个 URL 发一个 GET 请求进行签名验证。只要你代码和配置写对，后台就会提示「验证通过」。

六、Docker Compose 中增加环境变量（如在容器里跑）

如果你是用 docker-compose 启动 FastAPI 应用，在 docker-compose.yml 里为 app 服务增加环境变量：

services: app: build: . environment: - OPENAI_API_KEY=sk-xxxx - WECHAT_TOKEN=my_wechat_token_2024 - WECHAT_ENCODING_AES_KEY=你的43位AESKey - WECHAT_CORP_ID=你的企业CorpID - AGENT_BACKEND_URL=http://app:8000/v1/chat ports: - "8000:8000"

重新构建并启动：

docker-compose up -d --build

七、在企业微信群里真实体验一把

  1. 确保：
    • 你的 app 服务已启动；
    • Nginx 反向代理配置正确；
    • 企业微信后台的回调验证已经通过。
  2. 在企业微信中：
    • 找到你刚刚创建的自建应用 / 机器人；
    • 将它添加到某个测试群。
  3. 对同事来说，这就是一个普通的群聊机器人：
    • 不需要任何技术知识；
    • 会记住对话上下文（取决于你 /v1/chat 的实现）；
    • 会调用你现有的多技能 / RAG / 缓存逻辑。

在群里直接 @ 它，例如：

张三：@AI小助手 帮我写一份今天的工作计划： 上午要开会，下午写代码，晚上看两篇技术文章。 AI小助手： 好的，这是你今天的简要工作计划： 1. 上午：准备和参加会议，记录行动项…… 2. 下午：根据会议结论编写代码…… 3. 晚上：选择两篇和当前项目相关的技术文章……

八、简单故障排查 Checklist

1. 企业微信后台提示「URL 验证失败」

重点检查：

  • WECHAT_TOKEN、EncodingAESKey、CorpID 是否和后台一致；
  • URL 是否和你代码里的路由匹配（是否带 /wechat/callback 前缀）；
  • Nginx 是否正确转发到你的 app:8000/wechat/。

可以在回调接口里先打印日志，确认 GET 请求有没有到后端。

2. 群里 @ 机器人没反应

可能原因：

  1. /wechat/callback 配置的是内网地址，企业微信访问不到；
  2. 你的 AI 服务 /v1/chat 报错，导致 answer 为空或超时；
  3. 机器人没有被拉进当前群。

排查方法：

  • 先在后端打印收到的 msg.content，确保消息已经到达你服务；
  • 再单独用 curl 或 Postman 测试 /v1/chat 是否工作正常。

3. 机器人一直回复「只支持文本」

说明你发送的是图片、文件等非文本类型，而代码里只接受 msg.type == "text"。可以根据实际需要扩展对其他类型的支持。

九、进阶扩展思路（可选）

当基础版本跑通后，你可以逐步加一些“工程化”功能：

  1. 按用户配额控制用量
    • 把 user_id = msg.source 传给 /v1/chat；
    • 在 AI 服务侧按人统计 Token 消耗，设置每日上限。
  2. 按部门开放不同 Skill
    • 若能从企业微信接口拿到部门信息，可在内部配置文件中，为不同部门开放不同功能。
  3. 对接钉钉 / 飞书
    • 保持 /v1/chat 不变；
    • 新增 dingtalk_adapter.py、feishu_adapter.py，用同样的“回调 → 转发”模式接入。

十、总结

  • 不动核心，只加一层适配：
    企业微信只是换了一个入口，你的 AI 逻辑仍只暴露 /v1/chat。
  • 成本低，上手快：
    一份 wechat_adapter.py + 一点 Nginx / Docker 配置，就能把你的 AI 小助手真正「塞进」日常企业工作流。
  • 具备可扩展空间：
    后续可以按配额、权限、统计、管理后台等方向继续进化，而这一层群聊回调接口几乎无须改变。

Read more

数字化浪潮下，RPA 财务机器人引领财务变革​

在当今数字化飞速发展的时代，财务领域正经历着从传统手工操作向智能化、自动化转型的深刻变革。RPA（机器人流程自动化）财务机器人作为实现财务流程自动化、推动会计数字化转型的关键技术力量，凭借其高效、精准、可重复的特性，正逐渐打破传统财务管理模式的局限，成为企业提升财务运营效率、降低管理成本的核心工具。 本文将结合全球及中国市场的行业现状，深入探讨 2025 年 RPA 财务机器人的十大核心应用场景，同时剖析其在落地过程中面临的挑战，并对未来发展趋势进行展望，为企业了解和应用 RPA 财务机器人提供参考。 一、行业现状：全球普及加速，中国大企业率先落地 随着数字化技术在财务领域的深度渗透，RPA 财务机器人已在全球范围内实现规模化应用，市场规模持续扩张。根据Gartner（高德纳）2025 年全球 RPA 技术应用报告数据显示，截至 2025 年上半年，全球 RPA 财务领域相关市场规模已突破 80 亿美元，较 2024 年同比增长 18%
低空无人机车辆目标跟踪技术研究

低空无人机车辆目标跟踪技术研究

1. 1 选题依据        在科技创新不断突破的背景下，低空无人机作为新兴军事装备，凭借其在战场上的成功运用，正推动战争形态向以智能化远程攻击武器为主导的“非接触战争”演进。传统监视与侦察任务长期依赖人工，存在成本高、效率低、安全风险大等局限。而低空无人机凭借独特的空中视角与灵活机动能力，显著提升了任务效率与安全性。进一步将其与目标跟踪技术相结合，依托广阔视野和先进视觉算法，无人机不仅在军事领域作用突出，也在交通管理等民用领域中展现出广阔潜力[1-4]，应用场景如图1-1所示。 图1-1 无人机应用场景        低空无人机车辆目标跟踪技术的主要思想是:通过无人机搭载的摄像设备获取视频或图像数据，并借助目标检测、目标跟踪和图像处理等技术，实现对地面车辆目标的实时或准实时识别、定位与持续跟踪。尽管现有目标跟踪技术已相对成熟，但在实际应用环境中，尤其是无人机在低空飞行、复杂背景以及跟踪特定目标情形下，仍面临严峻的技术挑战。 （1）大场景中弱小目标容易漏检        相比于高空平台（如卫星或高空侦察机），低空无人机能提供更优的图像分辨率与更丰富的细节信息，这对精确识别

FPGA实现多协议编码器接口：BISS-C、SSI与多摩川的集成设计

1. 工业编码器接口的统一挑战与FPGA方案 在工业自动化领域，高精度运动控制系统的核心挑战之一是如何高效集成多种编码器协议。不同厂商的编码器采用不同的通信协议，比如BISS-C、SSI和多摩川协议，每种协议都有自己的时序要求、数据格式和校验机制。传统方案往往需要为每种协议设计独立的硬件接口，这不仅增加了系统复杂度，还提高了成本和维护难度。 我在实际项目中多次遇到这样的需求：客户希望用一个控制板卡同时支持多种编码器，但又不愿意增加额外的硬件成本。这时候FPGA的优势就凸显出来了。FPGA的可编程特性允许我们在同一块硬件上实现多种协议接口，通过逻辑资源复用和状态机控制，真正做到"硬件统一、软件定义"。 我记得有一次为数控机床项目设计编码器接口时，就遇到了同时连接BISS-C和多摩川编码器的需求。最初尝试用MCU+多路转换芯片的方案，但实时性总是达不到要求。后来转向FPGA方案，不仅实现了协议兼容，还将响应时间从原来的毫秒级降低到了微秒级。这种性能提升对于高精度运动控制来说是至关重要的。 2. BISS-C协议深度解析与FPGA实现 2.1 BISS-C协议核心机制 BISS

智能家居与物联网项目实战全指南：从架构设计到落地部署

随着物联网（IoT）、边缘计算与AI技术的深度融合，智能家居已从单一设备控制升级为“感知-决策-执行”的全场景智能系统。无论是个人开发者搭建家庭智能环境，还是企业级项目落地，都需要兼顾硬件兼容性、网络稳定性、场景实用性与安全性。本文将从系统架构、硬件选型、软件开发、场景实战、问题排查五个核心模块，提供可直接落地的实战方案，助力开发者快速完成智能家居项目从0到1的搭建。 一、智能家居系统核心架构设计（四层架构+技术选型） 智能家居系统的本质是“设备互联+数据驱动+场景联动”，采用经典的“感知层-网络层-平台层-应用层”四层架构设计，可确保系统的稳定性、可扩展性与兼容性。 1. 感知层：数据采集的“神经末梢” 感知层是系统的数据来源，负责采集环境参数、设备状态与人体行为信息，核心设备包括传感器与执行器，选型需兼顾精度、功耗与兼容性。 - 核心设备分类： - 环境传感器：温湿度传感器（推荐DHT22，精度±0.5℃