ai微信机器人实战：会话隔离、敏感词预审、转人工完整实现

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

230人浏览 · 2026-10-19 20:57:15

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

ChatGPT微信机器人实战：会话隔离、敏感词预审、转人工完整实现

单轮版 AI 自动回复上线第一天，群里就开始出事故：

客户问「那这个多少钱」——AI 答「您指的是哪款产品呢？」，因为它没有上下文，上一句聊的型号早忘了；
客户问「能保本吗」——AI 张口就是「这款产品基本稳赚不赔」，差点构成收益承诺，这是要命的；
客户打了三遍「转人工」——AI 还在温柔地推荐产品，客户直接拉黑。

三个事故对应三件事：会话隔离、敏感词预审、转人工单向门。它们不是"锦上添花的功能"，而是 AI 自动回复能不能上线的准入条件。这篇给完整实现。

微信侧基于托管 API 形态（HTTP 主动调用 + Webhook 回调），代码以 WTAPI 的接口为例；模型侧以 OpenAI 兼容的 /chat/completions 为例（ChatGPT 与 DeepSeek 均遵循该格式，仅需替换 base_url 与模型名），字段以各自官方文档为准。

一、三个组件各自解决什么

组件	不做会怎样	做了之后
会话隔离	上下文串号：A 客户的话被当成 B 的上下文；AI 记不住上句	每个客户独立上下文窗口，多轮对话连贯
敏感词预审	AI 生成违规/承诺性表述直接发给客户，业务事故	双向拦截：客户输入 + AI 输出各过一道闸
转人工单向门	客户明确要人工，AI 还自说自话，体验崩坏	一旦转人工，机器人对该客户永久闭嘴

三者的关系是层层设防：预审决定"这句话能不能进模型"，隔离保证"模型看到的是对的上下文"，转人工是"兜不住时的最后出路"。

二、会话隔离：appId + wxid 双键

为什么必须是双键？ 一个后台往往挂着多个微信号实例（多账号运营），同一个客户可能同时加了你们两个号。只用 wxid 做键，两个实例的上下文会混在一起；只用 appId，不同客户的对话会串。所以键是：

session_key = f"{appId}:{wxid}"

上下文管理的三个必备规则：

规则	做法	为什么
滑窗截断	只保留最近 N 轮（如 6 轮 = 12 条消息）	无脑全量追加，token 成本会线性暴涨，且超上下文上限会直接报错
过期清理	2 小时无新消息，清空该会话	昨天的对话今天续上，模型会答非所问；也避免内存无限增长
只存必要消息	系统提示词不放历史里，每次单独拼	每次请求都塞一遍 system，浪费 token

import time

SESSIONS = {}                 # key: "appId:wxid" -> {"msgs": [...], "last": timestamp}
MAX_MESSAGES = 12             # 保留最近 6 轮
SESSION_TTL  = 2 * 3600       # 2 小时无互动即清空

def skey(app_id, wx_id):
    return f"{app_id}:{wx_id}"

def get_history(app_id, wx_id):
    k = skey(app_id, wx_id)
    s = SESSIONS.get(k)
    if not s:
        return []
    if time.time() - s["last"] > SESSION_TTL:      # 过期清空
        SESSIONS.pop(k, None)
        return []
    return s["msgs"]

def push_history(app_id, wx_id, role, content):
    k = skey(app_id, wx_id)
    s = SESSIONS.setdefault(k, {"msgs": [], "last": 0})
    s["msgs"].append({"role": role, "content": content})
    s["msgs"] = s["msgs"][-MAX_MESSAGES:]          # 滑窗截断
    s["last"] = time.time()

生产环境把 SESSIONS 换成 Redis 并给 key 设 TTL，多进程/多实例部署时才不会各存一份互相看不到。用 sqlite/MySQL 同理，结构不变：一张表按 app_id + wx_id 存消息，定期删过期行。

三、双向内容预审：两道闸都不许省

大多数人的预审只做了一半——只拦 AI 输出的，不拦客户输入的。但客户输入同样要过闸：一是客诉、涉敏内容不该喂给模型（模型可能顺着发挥），二是这类消息本身就该立刻转人工。

# 入站（客户消息）：命中即不入模型，直接转人工
INBOUND_TRIGGER = ["投诉", "举报", "起诉", "退款", "曝光", "找媒体", "12315",
                   "人工", "客服", "转人"]

# 出站（AI 回复）：命中即不可直接发送，转人工并留痕
OUTBOUND_BLOCK = ["保本", "稳赚", "包赚", "保证收益", "百分百", "绝对安全",
                  "收益率", "零风险", "不会亏"]

def inbound_check(text):
    return [w for w in INBOUND_TRIGGER if w in text]

def outbound_check(reply):
    return [w for w in OUTBOUND_BLOCK if w in reply]

出站拦截为什么不能"替换成安全词"再发：你替换掉的是词，不是模型想承诺的意思——删掉「稳赚不赔」还剩「这款产品……」的语境仍在暗示。命中即不发、转人工，最稳妥。

词库怎么维护：行业敏感词由业务/合规定，代码里做成配置文件（JSON）热加载，别硬编码在逻辑里——合规要加词的时候，你需要的是改配置，不是改代码重新发布。

四、转人工：单向门，触发条件有三个

单向门的意思是：一旦某个客户被标记为"转人工"，机器人对这个客户的所有后续自动回复永久停止（除非人工介入后手动解除）。最忌讳的是顾问已经在跟进，机器人还在按流程推资料——客户视角就是精神分裂。

三个触发条件：

触发	判定	动作
明确要人工	入站命中「人工」「客服」	立即转，回一句承接话术
客诉/涉敏	入站命中「投诉」「起诉」等	立即转，不作任何解释性回复
连续兜底	模型连续失败/出站被拦 ≥ 2 次	转人工，避免客户干等

HANDOFF = set()                 # 已转人工的会话 key
_fallback_count = {}            # 连续兜底计数

def is_handoff(app_id, wx_id):
    return skey(app_id, wx_id) in HANDOFF

def mark_handoff(app_id, wx_id, reason):
    HANDOFF.add(skey(app_id, wx_id))
    notify_staff(wx_id, reason)         # 通知顾问接手（自行实现）

关键点：转人工后不是"AI 再答一句"，而是"AI 只说一句承接、然后闭嘴"。承接话术固定、不含任何业务信息：

"收到，我请专属顾问来跟您对接，马上联系您～"

五、完整代码：三个组件组装起来

# bot_ai_pro.py —— 会话隔离 + 双向预审 + 转人工
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"
WX_HEADERS = {
    "X-finder-TOKEN": "YOUR_TOKEN",
    "Authorization": "Bearer YOUR_JWT",   # 部分环境需要
    "Content-Type": "application/json",
}
# ---------- 模型侧（OpenAI 兼容；换 DeepSeek 只需改这两个值）----------
LLM_URL   = "https://api.openai.com/v1/chat/completions"
LLM_KEY   = "sk-你的Key"
LLM_MODEL = "gpt-4o-mini"                 # DeepSeek 填 deepseek-chat

SYSTEM_PROMPT = (
    "你是一名微信客服助理。回答简短口语化（不超过150字），不使用 Markdown。"
    "严禁承诺收益、保本、稳赚等表述。"
    "遇到价格、退款、投诉类问题，回复：这个问题我请人工顾问为您处理。"
    "不确定的事直说不知道，不要编造。"
)

# ---------- 会话隔离 ----------
SESSIONS = {}
MAX_MESSAGES, SESSION_TTL = 12, 2 * 3600
def skey(a, w): return f"{a}:{w}"

def get_history(a, w):
    k, s = skey(a, w), SESSIONS.get(skey(a, w))
    if not s: return []
    if time.time() - s["last"] > SESSION_TTL:
        SESSIONS.pop(k, None); return []
    return s["msgs"]

def push_history(a, w, role, content):
    s = SESSIONS.setdefault(skey(a, w), {"msgs": [], "last": 0})
    s["msgs"].append({"role": role, "content": content})
    s["msgs"] = s["msgs"][-MAX_MESSAGES:]
    s["last"] = time.time()

# ---------- 预审词库（建议外置为 JSON 配置热加载）----------
INBOUND_TRIGGER = ["投诉", "举报", "起诉", "退款", "曝光", "找媒体", "12315", "人工", "客服"]
OUTBOUND_BLOCK  = ["保本", "稳赚", "包赚", "保证收益", "百分百", "绝对安全", "收益率", "零风险"]

# ---------- 转人工 ----------
HANDOFF, _fallback_count = set(), {}
def is_handoff(a, w): return skey(a, w) in HANDOFF

def mark_handoff(a, w, reason):
    HANDOFF.add(skey(a, w))
    notify_staff(w, reason)

def notify_staff(wx_id, reason):
    """示例：用文件传输助手给运营提醒；正式环境换成内部系统推送"""
    requests.post(f"{API_BASE}/finder/v2/api/message/postText",
        headers=WX_HEADERS,
        data=json.dumps({"appId": APP_ID, "toWxid": "filehelper",
                         "content": f"【转人工】客户 {wx_id}｜原因：{reason}"}),
        timeout=10)

def post_text(to_wxid, content):
    requests.post(f"{API_BASE}/finder/v2/api/message/postText",
        headers=WX_HEADERS,
        data=json.dumps({"appId": APP_ID, "toWxid": to_wxid, "content": content}),
        timeout=10)

# ---------- 模型调用 ----------
def call_llm(messages):
    try:
        r = requests.post(LLM_URL,
            headers={"Authorization": f"Bearer {LLM_KEY}",
                     "Content-Type": "application/json"},
            json={"model": LLM_MODEL, "messages": messages,
                  "temperature": 0.7, "max_tokens": 500},
            timeout=30)
        d = r.json()
        return d["choices"][0]["message"]["content"], d.get("usage", {})
    except Exception as e:
        print("[LLM异常]", e)
        return None, {}

# ---------- 业务主流程 ----------
def handle_message(app_id, wx_id, text):
    # ① 单向门：已转人工 → 永久不自动回复
    if is_handoff(app_id, wx_id):
        return

    # ② 入站预审
    hits = [w for w in INBOUND_TRIGGER if w in text]
    if hits:
        post_text(wx_id, "收到，我请专属顾问来跟您对接，马上联系您～")
        mark_handoff(app_id, wx_id, f"客户输入命中:{','.join(hits)}")
        return

    # ③ 带上下文调模型
    history = get_history(app_id, wx_id)
    messages = [{"role": "system", "content": SYSTEM_PROMPT}] + \
               history + [{"role": "user", "content": text}]
    reply, usage = call_llm(messages)

    # ④ 兜底与熔断
    if not reply:
        n = _fallback_count.get(skey(app_id, wx_id), 0) + 1
        _fallback_count[skey(app_id, wx_id)] = n
        if n >= 2:                                   # 连续失败 2 次 → 转人工
            post_text(wx_id, "抱歉，我这会儿有点慢，已为您转接人工顾问～")
            mark_handoff(app_id, wx_id, "模型连续失败")
        else:
            post_text(wx_id, "抱歉，我这边有点小状况，请再说一次或回复【人工】～")
        return
    _fallback_count[skey(app_id, wx_id)] = 0

    # ⑤ 出站预审：命中即不发，转人工留痕
    bad = [w for w in OUTBOUND_BLOCK if w in reply]
    if bad:
        mark_handoff(app_id, wx_id, f"AI输出命中敏感词:{','.join(bad)}｜原文:{reply[:60]}")
        post_text(wx_id, "这个问题我请人工顾问为您详细说明，马上联系您～")
        return

    # ⑥ 通过全部闸门，发送并写入历史
    post_text(wx_id, reply)
    push_history(app_id, wx_id, "user", text)
    push_history(app_id, wx_id, "assistant", reply)
    print(f"[用量] {usage}")

# ---------- 回调接入 ----------
app = Flask(__name__)
_seen, task_queue = set(), queue.Queue()

def consumer():
    while True:
        msg = task_queue.get()
        d = msg.get("Data", {})
        if d.get("MsgType") != 1:
            continue
        from_user = d.get("FromUserName", {}).get("string", "")
        to_user   = d.get("ToUserName", {}).get("string", "")
        if from_user.endswith("@chatroom") or from_user == to_user:
            continue
        text = d.get("Content", {}).get("string", "").strip()
        handle_message(msg.get("Appid", APP_ID), from_user, text)

@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 秒内必须返回

if __name__ == "__main__":
    threading.Thread(target=consumer, daemon=True).start()
    app.run(host="0.0.0.0", port=8080)

六、上线前的六项验证（照着打勾）

#	验证动作	期望结果
1	问「你们几点上班」，再问「那周末呢」	AI 第二次回答知道在聊上班时间（上下文生效）
2	用两个微信号交替提问同一问题	两边回复互不串号（隔离生效）
3	发「转人工」	收到承接话术 + 文件传输助手收到转人工提醒
4	转人工后再发一条普通问题	机器人不再自动回复（单向门生效）
5	临时往 `OUTBOUND_BLOCK` 加一个高频词，触发它	AI 回复被拦下、改发承接话术、转人工留痕
6	把 `LLM_KEY` 改错，连发两条	第一条兜底、第二条转人工（熔断生效）

第 4 项是最容易漏测的——很多人实现了转人工通知，却忘了"闭嘴"这一步。

合规声明：AI 自动回复必须配置内容审核机制（至少包括输入与输出双向敏感词过滤），禁止生成收益承诺、保本保息等违规表述；金融、保险、医疗等强监管行业，自动回复不得涉及产品销售与建议。请控制发送频次、避免骚扰，遵守微信平台规则及相关法律法规。模型接口与微信接口的能力、参数与限制以各自官方文档实时内容为准。