微信自动回复机器人完整代码：从回调接收到消息回写

WTAPI｜微信机器人接口二次开发平台

234人浏览 · 2026-10-16 15:20:04

WTAPI｜微信机器人接口二次开发平台 · 2026-10-16 15:20:04 发布

不是所有自动回复都需要大模型。大量真实需求其实非常朴素：客户发"价格"回价目表、发"地址"回门店定位、进群后 @ 机器人问规则回群公告——这类固定问答用关键词匹配就够了，零模型成本、响应毫秒级、话术 100% 可控。

本文给一份最简但完整的关键词自动回复机器人成品：一个文件、四段逻辑（接消息、判场景、匹配回复、限速发回），私聊和群聊都覆盖。微信侧以 WTAPI 的托管 API 形态为例，接口参数以官方文档为准。

一、先看清要处理的四种情况

消息来源	Content 形态	处理策略
自己发出的消息	`FromUserName == Wxid`	直接忽略
私聊文本	纯文本内容	关键词匹配，命中即回
群聊普通文本	`wxid:\n实际内容`（发送者与内容用换行分隔）	默认不回，避免插话
群聊 @ 机器人	内容中带 @ 昵称	解析实际内容后匹配回复

群消息的 FromUserName 以 @chatroom 结尾；其 Content.string 的结构是"发送者 wxid + 换行 + 消息正文"，所以群聊里要拿到真正的正文，必须先做一次切割。

二、完整代码

import json
import time
import queue
import threading
import requests
from flask import Flask, request

# ── 配置 ───────────────────────────────────────────
API_BASE = "https://wx.chuapi.com"
APP_ID = "YOUR_APPID"
BOT_NICKNAME = "小助手"                 # 机器人在群里的昵称，用于 @ 判定
HEADERS = {
    "X-finder-TOKEN": "YOUR_TOKEN",
    "Authorization": "Bearer YOUR_JWT",     # 部分环境需要
    "Content-Type": "application/json",
}

# ── 关键词回复表：命中任一关键词即回复（顺序即优先级）──
REPLY_RULES = [
    (["价格", "多少钱", "费用"], "我们的价目表如下：基础版 XX 元/月……"),
    (["地址", "在哪", "门店"], "我们在 XX 市有 3 家门店，发您最近的定位～"),
    (["营业时间", "几点", "下班"], "服务时间：每天 9:00-21:00。"),
    (["规则", "群规", "公告"], "本群规则：① 禁止广告 ② 问题请直接描述，管理员会统一回复。"),
]
DEFAULT_REPLY = "已收到您的消息，如需人工服务请回复【人工】。"
HUMAN_KEYWORDS = ["人工", "真人", "客服"]

# ── 发送限速：同实例串行 + 固定间隔，防止短时间刷屏 ──
SEND_INTERVAL = 1.0
_send_lock = threading.Lock()
_last_sent = [0.0]

def post_text(to_wxid, content):
    """发文本：参数仅 appId / toWxid / content"""
    with _send_lock:
        wait = SEND_INTERVAL - (time.time() - _last_sent[0])
        if wait > 0:
            time.sleep(wait)
        requests.post(
            f"{API_BASE}/finder/v2/api/message/postText",
            headers=HEADERS,
            data=json.dumps({"appId": APP_ID, "toWxid": to_wxid, "content": content}),
            timeout=10,
        )
        _last_sent[0] = time.time()

def match_reply(text):
    """关键词匹配：命中规则返回复话术，命中人工词返回 None（交业务转人工）"""
    if any(k in text for k in HUMAN_KEYWORDS):
        return "__HUMAN__"
    for keywords, answer in REPLY_RULES:
        if any(k in text for k in keywords):
            return answer
    return None     # 没命中：私聊给兜底，群聊静默

def parse_group_content(raw):
    """群消息：'wxid:\n正文' → (发送者wxid, 正文)；具体格式以回调文档为准"""
    if ":\n" in raw:
        sender, _, content = raw.partition(":\n")
        return sender, content
    return "", raw

# ── 回调接收：去重 → 入队 → 3秒内返回 ────────────────
task_queue = queue.Queue()
_seen = set()
app = Flask(__name__)

@app.route("/callback", methods=["POST"])
def callback():
    msg = request.get_json(silent=True) or {}
    if msg.get("TypeName") == "AddMsg":
        d = msg.get("Data", {})
        key = f"{msg.get('Appid')}:{d.get('NewMsgId')}"    # 官方去重键
        if key not in _seen:
            _seen.add(key)
            task_queue.put(msg)
    return "", 200                            # 3 秒内必须返回

# ── 业务处理 ───────────────────────────────────────
def worker():
    while True:
        msg = task_queue.get()
        try:
            handle(msg)
        except Exception as e:
            print(f"处理失败: {e}")
        finally:
            task_queue.task_done()

def handle(msg):
    d = msg.get("Data", {})
    from_user = d.get("FromUserName", {}).get("string", "")
    raw_content = d.get("Content", {}).get("string", "")

    if d.get("MsgType") != 1:                 # 只处理文本（1）
        return
    if from_user == msg.get("Wxid"):          # 忽略自己发的消息
        return

    # ── 群聊分支 ──
    if from_user.endswith("@chatroom"):
        sender, content = parse_group_content(raw_content)
        # 只响应 @ 机器人的消息（@ 的具体文本格式以实际回调内容为准）
        if f"@{BOT_NICKNAME}" not in content:
            return
        content = content.replace(f"@{BOT_NICKNAME}", "").strip()
        reply = match_reply(content)
        if reply and reply != "__HUMAN__":
            # 群里回复可带上@发送者，具体格式按群消息规范调整
            post_text(from_user, reply)
        return

    # ── 私聊分支 ──
    reply = match_reply(raw_content)
    if reply == "__HUMAN__":
        post_text(from_user, "正在为您转接人工客服，请稍候。")
        # TODO: 推送顾问群（from_user + raw_content）
    elif reply:
        post_text(from_user, reply)
    else:
        post_text(from_user, DEFAULT_REPLY)

threading.Thread(target=worker, daemon=True).start()

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

三、上线前必须知道的五个细节

群聊默认静默是安全底线。 代码里群消息只有被 @ 才响应——对每条群消息都自动回复，机器人活不过一天。@ 的具体文本格式建议先用真实群消息抓一条回调看 Content.string 原文，再校准匹配规则；
群回复的对象是群，不是发送者。 群消息的 toWxid 填群 ID（即 FromUserName 里那个 @chatroom 地址），消息才发回群里；想精准 @ 某人，按群消息的 @ 格式拼接，别把 wxid 当私聊目标；
回复表用配置/数据库，别写死。 运营改价目表的频率远高于发版频率，把 REPLY_RULES 挪到配置文件或数据库，机器人不用重启就能更新话术；
限速参数按账号状态调。 新实例把 SEND_INTERVAL 调大（3-5 秒）、先只开私聊，稳定后再开群聊和提速。新设备登录有 1-3 天功能适应期，低频起步；
非文本消息要有去向。 图片（MsgType=3）、语音（34）等当前直接忽略，生产环境建议回一句"暂不支持该消息类型，请用文字描述"，而不是石沉大海。