微信关键词自动回复机器人（轻量版，不接AI）

前言

许多运营场景需要对微信消息做出即时反应：用户发来"价格"就回一段报价、发来"下载"就回链接、发来"人工"就转接客服提示。如果每条都靠人盯着手机来回复，效率极低，也难以做到 7×24 小时覆盖。

接 AI 大模型固然强大，但 Token 费用、冷启动延迟、回复不可控等问题让很多小团队望而却步。其实绝大多数业务场景只需要"命中关键词→返回固定内容"这一条规则——既快、又稳、零 AI 成本。

本文从需求拆解入手，给出一套完整的轻量关键词回复机器人方案：回调接收消息、规则引擎匹配、接口发送回复，全程 Python 实现，不依赖 AI，不依赖企业微信生态，用个人微信号即可跑起来。

一、整体架构与核心思路

关键词自动回复本质上是一个事件驱动的三段流程：

微信消息 → 回调推送到你的服务器 → 规则匹配 → 调接口回复

每一段拆开看：

三段都是纯同步逻辑，没有状态机、没有会话管理，代码量极少，部署一台有公网 IP 的轻量云服务器即可。

二、消息回调：把微信消息接到你的服务器

2.1 回调地址要求

回调地址必须公网可访问，且接收到请求后需在 3 秒内返回 HTTP 200，否则平台会重试或判断超时。本地开发阶段可用 ngrok / frp 做内网穿透，生产环境直接部署到云服务器。

需要注意以下几点：首先，服务器的安全组/防火墙要放行对应端口；其次，如果使用 HTTPS，证书必须有效，否则平台推送时可能报 SSL 错误；另外，回调路由建议加简单的签名验证（如校验请求头里的 token），防止他人伪造请求打入服务器。

2.2 回调数据结构

平台将消息以 POST JSON 的形式推送到你设置的回调地址，典型字段如下（具体以官方文档为准）：

json{
  "appId": "你的appId",
  "fromWxid": "发送者的wxid",
  "toWxid": "接收者的wxid（通常是你自己）",
  "type": 1,
  "content": "用户发送的文字内容",
  "msgId": "消息唯一ID",
  "createTime": 1718000000
}

其中 type 字段区分消息类型，常见值：1=文字，3=图片，49=链接/小程序，43=视频，34=语音。关键词回复只处理 type==1 的文字消息即可。

2.3 用 Flask 搭建回调服务

pythonfrom flask import Flask, request, jsonify
import json

app = Flask(__name__)

@app.route("/callback", methods=["POST"])
def callback():
    data = request.get_json(force=True)
    # 只处理文字消息
    if data.get("type") == 1:
        handle_text(data)
    return jsonify({"code": 200})   # 必须返回 200

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

注意：handle_text 里的逻辑要尽量快，耗时操作（如发请求）建议放进线程或异步队列，避免超时。

三、规则引擎：关键词匹配逻辑

关键词规则通常分三类：精确匹配、包含匹配、正则匹配。从最简单的开始，按需叠加。

3.1 规则配置文件

把规则写到 JSON 或 YAML 文件，和代码解耦，方便运营人员直接修改而无需改代码。

json{
  "rules": [
    {
      "match_type": "exact",
      "keywords": ["价格", "报价", "多少钱"],
      "reply": "您好，我们的定价请参考官网产品页，或直接联系客服为您出方案。"
    },
    {
      "match_type": "contains",
      "keywords": ["下载", "安装包", "客户端"],
      "reply": "客户端下载地址已通过官网公告发布，搜索「产品名+下载」即可找到。"
    },
    {
      "match_type": "regex",
      "keywords": ["订单[号码编]?\\s*\\d+"],
      "reply": "您的订单信息已收到，稍后由专属客服跟进，请保持微信畅通。"
    },
    {
      "match_type": "contains",
      "keywords": ["人工", "转人工", "客服"],
      "reply": "人工客服工作时间为 09:00-18:00，请稍候，我们会尽快回复您。"
    }
  ],
  "default_reply": ""
}

default_reply 为空字符串表示未命中时不回复，避免骚扰用户；也可以设置一段通用兜底文案。

规则维护上有几个实用建议：关键词表应由运营人员定期检视，及时增删过时词条；同一类业务建议归入同一条规则（共用同一个 keywords 列表），避免规则条数膨胀；规则文件改动后无需重启服务，可以在 handle_text 里每次调用前重新加载配置，或者设置文件 mtime 检测，只在变更时重载，兼顾性能与灵活性。

3.2 匹配引擎代码

pythonimport re
import json

def load_rules(path="rules.json"):
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)

def match_rule(content: str, rules_config: dict) -> str:
    """
    返回命中规则的回复文案，未命中返回 default_reply（可能为空字符串）
    """
    content_lower = content.strip().lower()
    for rule in rules_config.get("rules", []):
        match_type = rule["match_type"]
        for kw in rule["keywords"]:
            if match_type == "exact" and content_lower == kw.lower():
                return rule["reply"]
            elif match_type == "contains" and kw.lower() in content_lower:
                return rule["reply"]
            elif match_type == "regex" and re.search(kw, content, re.IGNORECASE):
                return rule["reply"]
    return rules_config.get("default_reply", "")

优先级说明：列表靠前的规则优先命中。如果业务需要"精确匹配优先于包含匹配"，可以在加载时先把 exact 规则排到前面。

3.3 防回环：不回复自己发的消息

pythonMY_WXID = "你自己的wxid"   # 登录后通过接口获取

def handle_text(data: dict):
    from_wxid = data.get("fromWxid", "")
    # 自己发的消息不处理，防止机器人回复自己造成死循环
    if from_wxid == MY_WXID:
        return
    content = data.get("content", "")
    reply = match_rule(content, RULES_CONFIG)
    if reply:
        send_text(from_wxid, reply)

这一步非常关键，否则机器人发出的消息会触发自己的回调，无限循环。

四、消息发送：调接口把回复送出去

4.1 配置区

pythonBASE  = "https://你的接口域名"   # 注册后在官方文档获取
TOKEN = "你的Token"
APPID = "你的appId"
HEADERS = {"token": TOKEN}       # 鉴权字段名以官方文档为准

以上为示例占位符，具体域名、鉴权方式以官方文档为准。

4.2 发送文字消息

pythonimport requests

def send_text(to_wxid: str, content: str):
    """
    发送文字回复
    代码为示例，具体接口/字段以官方文档为准
    """
    url = f"{BASE}/message/postText"
    payload = {
        "appId": APPID,
        "toWxid": to_wxid,
        "content": content
    }
    try:
        resp = requests.post(url, json=payload, headers=HEADERS, timeout=5)
        result = resp.json()
        if result.get("ret") != 200:
            print(f"发送失败: {result}")
    except Exception as e:
        print(f"发送异常: {e}")

4.3 发送图片回复（进阶）

部分关键词希望直接回一张图片（如"二维码"→回群二维码图片）：

pythondef send_image(to_wxid: str, img_url: str):
    """
    发送图片消息
    代码为示例，具体接口/字段以官方文档为准
    """
    url = f"{BASE}/message/postImage"
    payload = {
        "appId": APPID,
        "toWxid": to_wxid,
        "imgUrl": img_url   # 填图片的公网URL
    }
    try:
        resp = requests.post(url, json=payload, headers=HEADERS, timeout=10)
        result = resp.json()
        if result.get("ret") != 200:
            print(f"图片发送失败: {result}")
    except Exception as e:
        print(f"图片发送异常: {e}")

规则配置中可以增加 reply_type 字段区分文字和图片，根据类型调用不同发送函数。

需要注意的是，图片 URL 必须是公网可访问的直链，不能是本地路径或需要鉴权才能访问的私有链接。如果图片存储在对象存储（如七牛云、阿里云 OSS）上，使用公开读取权限的地址即可；如果是自建文件服务器，需确保服务器带宽足够，避免平台拉取超时导致图片发送失败。

五、完整整合：把三段拼在一起

python# bot.py  完整示例
# 代码为示例，具体接口/字段以官方文档为准

import re
import json
import threading
import requests
from flask import Flask, request, jsonify

# ===== 配置 =====
BASE    = "https://你的接口域名"
TOKEN   = "你的Token"
APPID   = "你的appId"
HEADERS = {"token": TOKEN}
MY_WXID = "你自己的wxid"

# ===== 规则加载 =====
with open("rules.json", "r", encoding="utf-8") as f:
    RULES_CONFIG = json.load(f)

# ===== 规则匹配 =====
def match_rule(content: str) -> str:
    content_lower = content.strip().lower()
    for rule in RULES_CONFIG.get("rules", []):
        mt = rule["match_type"]
        for kw in rule["keywords"]:
            if mt == "exact" and content_lower == kw.lower():
                return rule["reply"]
            elif mt == "contains" and kw.lower() in content_lower:
                return rule["reply"]
            elif mt == "regex" and re.search(kw, content, re.IGNORECASE):
                return rule["reply"]
    return RULES_CONFIG.get("default_reply", "")

# ===== 发送函数 =====
def send_text(to_wxid: str, content: str):
    url = f"{BASE}/message/postText"
    payload = {"appId": APPID, "toWxid": to_wxid, "content": content}
    try:
        resp = requests.post(url, json=payload, headers=HEADERS, timeout=5)
        result = resp.json()
        if result.get("ret") != 200:
            print(f"[WARN] 发送失败: {result}")
    except Exception as e:
        print(f"[ERROR] 发送异常: {e}")

# ===== 回调处理 =====
def handle_text(data: dict):
    from_wxid = data.get("fromWxid", "")
    if from_wxid == MY_WXID:
        return
    content = data.get("content", "")
    reply = match_rule(content)
    if reply:
        # 放线程避免阻塞回调响应
        threading.Thread(target=send_text, args=(from_wxid, reply), daemon=True).start()

# ===== Flask 入口 =====
app = Flask(__name__)

@app.route("/callback", methods=["POST"])
def callback():
    data = request.get_json(force=True)
    if data.get("type") == 1:
        handle_text(data)
    return jsonify({"code": 200})

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

依赖清单：

flask>=2.0
requests>=2.28

一条命令启动：

bashpip install flask requests && python bot.py

六、进阶功能：群消息过滤与@处理

如果机器人也在群里，回调会收到大量群消息。通常只希望处理：

1. 有人@了机器人的消息
2. 群内触发特定关键词（如"!帮助"前缀命令）

pythondef handle_text(data: dict):
    from_wxid = data.get("fromWxid", "")
    if from_wxid == MY_WXID:
        return
    content = data.get("content", "")
    to_wxid  = data.get("toWxid", "")

    # 判断是否群消息（群wxid通常以 @ 结尾，如 xxx@chatroom）
    is_group = to_wxid.endswith("@chatroom")

    if is_group:
        # 群消息只处理以 "!" 开头的命令，或包含@机器人标记的
        if not content.startswith("!") and MY_WXID not in content:
            return
        # 去掉前缀再匹配
        content = content.lstrip("!").strip()

    reply = match_rule(content)
    if reply:
        # 群消息回复要带上发消息者的@标记
        if is_group:
            reply = f"@{from_wxid} {reply}"
        threading.Thread(target=send_text, args=(to_wxid if is_group else from_wxid, reply), daemon=True).start()

群消息的 to_wxid 是群的 roomId，私聊时 to_wxid 是自己的 wxid。发送群回复时 toWxid 填群 roomId 即可。

群消息场景还有一些实操细节值得注意：群成员较多时，短时间内多人触发同一关键词会产生消息风暴，建议在群维度也加冷却时间（即同一个群 1 分钟内同一关键词只回复一次）；另外，@机器人的消息内容通常会在文本开头带上"@昵称"字样，正式匹配前需要把这段前缀去掉，否则会影响规则命中率。

七、托管 API 方案说明

上述代码涉及"扫码登录后获取 appId、设置回调地址、调接口发消息"三个核心动作。WechatApi 提供扫码登录、消息收发、好友与群管理等 REST 接口，HTTP 调用即可，后台支持在线设置回调地址和查看消息日志，适合快速接入本文的轻量机器人方案。

八、防封与频率控制建议

关键词回复场景下发送量可能较大，以下几点务必注意：

简单的冷却实现：

pythonimport time
_last_reply: dict[str, float] = {}
COOLDOWN = 60  # 秒

def can_reply(wxid: str) -> bool:
    now = time.time()
    if now - _last_reply.get(wxid, 0) < COOLDOWN:
        return False
    _last_reply[wxid] = now
    return True

在 handle_text 开头调用 can_reply(from_wxid) 做门控即可。

除频率控制外，回复内容本身也需要审慎设计。避免在自动回复中包含营销广告链接、导流话术或违规词汇；回复文案应保持简洁克制，让用户感知到这是机器人在服务而非真人冒充；对于敏感词命中的用户消息，可选择静默不回复，而不是给出可能引发误解的通用兜底语。

九、常见问题与排查思路

Q：回调地址配置后，平台一直提示连接失败，怎么排查？

首先确认服务已启动且端口正常监听（用 curl http://本机IP:8080/callback -X POST -d '{}' 在服务器本地测试）；其次检查云服务器安全组是否开放了对应端口；最后确认回调地址填写的是公网 IP 或域名，而非 127.0.0.1。

Q：消息明明发来了，服务器日志没有收到，是怎么回事？

可能是平台推送超时重试机制的问题：若你的服务在 3 秒内没有返回 200，平台会判定超时并重试，同时日志中可能不记录失败推送。建议先把 handle_text 的内容全部注释掉，仅保留返回 200 的逻辑，确认基础通路正常后再逐步加入业务逻辑。

Q：同一条消息被回调推送了好几次，重复处理怎么办？

根据消息体中的 msgId 字段做幂等去重。用 Python 的 set 或 Redis 的 setnx 记录已处理过的 msgId，进入处理逻辑前先判断是否已见过该 ID，若是则直接跳过。

Q：关键词太多，文件越来越大，维护很麻烦，有什么优化方案？

规则条数超过 50 条后，建议把 JSON 配置迁移到 SQLite 或轻量数据库，配合一个简单的后台页面（如用 Flask-Admin 快速搭建）让运营人员通过界面管理规则，彻底告别手改 JSON 文件。

总结

不接 AI 的关键词回复机器人，代码结构清晰、部署简单、运行稳定，适合绝大多数运营场景。核心就三步：搭回调服务接消息、写规则表匹配内容、调发送接口回复——加上防回环和冷却控制，整个方案就完备了。群消息场景额外注意过滤规则和消息风暴防护；实际运营中要保持回复内容合规、频率克制，才能让机器人长期稳定运行。遇到回调不通、重复推送等问题，按本文排查思路逐步定位，通常都能快速解决。