微信 AI 机器人多轮对话与上下文管理实战

前言

随着大语言模型（LLM）的普及，越来越多的开发者希望把 AI 能力嵌入微信，做成一个能和用户"聊明白"的智能客服或助手机器人。单轮问答很简单——收到一条消息、调一次 LLM、把结果回复回去就行；但真实场景里用户的意图往往跨越多条消息，比如先问"帮我查一下北京今天天气"，接着说"顺便推荐几家餐厅"——如果机器人失忆了，第二条消息就会变成一句废话。

本文聚焦微信机器人的多轮对话设计与上下文管理，从最基础的消息回调接收开始，逐步讲清楚如何维护每个用户的对话历史、如何控制上下文窗口长度、如何处理超时/切换场景、以及一些实战中常见的坑。全文代码以 Python 为例，接口调用逻辑适用于能提供微信消息收发能力的 HTTP API 方案。

一、多轮对话的核心挑战

1.1 为什么单轮处理不够

LLM 本身是无状态的——每次调用 API 时，你传入什么历史就得到对应的回答，模型本身不会帮你记录上一轮说过什么。微信消息流同样是无状态的：平台把每条消息推送到你的回调地址，消息与消息之间没有显式的"会话 ID"。

因此，多轮对话的状态维护完全由业务侧负责：

• 识别"这条消息属于哪个用户的哪段对话"
• 将历史消息按 LLM 要求的格式组织好，随每次请求一起传入
• 在合适的时机截断历史，避免 token 超限

1.2 上下文的三个维度

二、回调消息接收与会话 Key 设计

要做多轮对话，首先要稳定地接收微信消息。常见方案是通过平台提供的 setCallback 接口，把回调地址注册给平台，平台在收到微信消息时会实时 POST 到你的服务。

2.1 回调数据结构示例

python# 回调字段以官方文档为准，下面是常见结构示意
# {
#   "appId": "设备ID",
#   "fromWxid": "发送方wxid",
#   "toWxid": "接收方wxid（你的机器人）",
#   "type": 1,          # 1=文本, 3=图片, 等
#   "content": "消息正文",
#   "msgId": "消息唯一ID",
#   "createTime": 1718000000
# }

2.2 会话 Key 的设计

最简单的隔离策略是以 fromWxid 作为 Key——每个微信号独立维护一份对话历史。如果你的机器人同时在多个群里工作，还需要把群 ID 纳入 Key：

pythondef make_session_key(msg: dict) -> str:
    """
    私聊：用 fromWxid
    群聊：用 chatroom_id + fromWxid（区分不同群里的同一个人）
    """
    from_wxid = msg.get("fromWxid", "")
    # 判断是否为群消息（群ID通常以@chatroom结尾，具体格式以文档为准）
    if from_wxid.endswith("@chatroom"):
        # 群聊场景：以 群ID 为隔离单元，所有群成员共享同一段对话
        return f"group:{from_wxid}"
    else:
        return f"user:{from_wxid}"

注意：群聊场景下，是"以群为单位共享历史"还是"每个群成员各自独立"，取决于产品定义。客服机器人通常按用户隔离，话题讨论机器人通常按群隔离。

2.3 Flask 回调接入示例

python# 代码为示例，具体接口/字段以官方文档为准
from flask import Flask, request, jsonify
import threading

app = Flask(__name__)
message_queue = []  # 生产环境建议换成 Redis Queue 或 Celery
lock = threading.Lock()

@app.route("/callback", methods=["POST"])
def callback():
    data = request.get_json(force=True)
    msg_type = data.get("type")
    # 只处理文本消息（type=1），其它类型可按需扩展
    if msg_type == 1:
        with lock:
            message_queue.append(data)
    return jsonify({"code": 200})   # 必须立即返回200，勿在此处做耗时处理

回调函数必须快速返回 200，所有耗时操作（调 LLM、发消息）放到异步队列里处理，否则平台会认为推送失败并重试，导致重复处理。

三、对话历史的存储与管理

3.1 内存存储（快速起步）

python# 代码为示例，具体接口/字段以官方文档为准
import time
from collections import defaultdict

# 每个 session_key 对应一个列表，元素格式与 OpenAI / Claude 等 LLM 兼容
# {"role": "user"/"assistant", "content": "..."}
conversation_store: dict[str, list] = defaultdict(list)

# 记录每个会话最后活跃时间，用于超时重置
last_active: dict[str, float] = defaultdict(float)

SESSION_TIMEOUT = 30 * 60  # 30分钟无消息视为新会话

def get_history(session_key: str) -> list:
    now = time.time()
    if now - last_active[session_key] > SESSION_TIMEOUT:
        conversation_store[session_key] = []   # 超时重置
    last_active[session_key] = now
    return conversation_store[session_key]

def append_message(session_key: str, role: str, content: str):
    history = get_history(session_key)
    history.append({"role": role, "content": content})

内存存储的缺点是进程重启后历史清空，适合个人机器人或测试环境。生产环境建议用 Redis（设置 TTL 自动过期）或关系型数据库。

3.2 Redis 持久化方案

python# 代码为示例，具体接口/字段以官方文档为准
import json
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
SESSION_TTL = 1800  # 秒，超时自动删除

def get_history_redis(session_key: str) -> list:
    raw = r.get(f"chat:{session_key}")
    if raw is None:
        return []
    return json.loads(raw)

def save_history_redis(session_key: str, history: list):
    r.set(f"chat:{session_key}", json.dumps(history, ensure_ascii=False), ex=SESSION_TTL)

Redis 的 TTL 机制天然实现了"超时自动重置"——无需手动检查时间戳，Key 过期后下次读取就是空列表，相当于开启新会话。

四、上下文窗口控制策略

4.1 问题：历史越积越长

主流 LLM 的 context 窗口少则 4K token，多则 128K。微信聊天场景里用户可能一天发几十条消息，如果全量放入历史，很快就会超限——轻则 API 报错，重则 token 费用爆炸。

4.2 方案一：固定轮数滑动窗口

最简单粗暴的方案：只保留最近 N 轮对话（一轮 = 一条 user + 一条 assistant）。

pythonMAX_TURNS = 10  # 保留最近10轮

def trim_history_by_turns(history: list) -> list:
    """
    保留最近 MAX_TURNS 轮，每轮2条消息
    system prompt（如有）永远保留在最前面
    """
    system_msgs = [m for m in history if m["role"] == "system"]
    dialog_msgs = [m for m in history if m["role"] != "system"]
    # 取最后 MAX_TURNS * 2 条
    trimmed = dialog_msgs[-(MAX_TURNS * 2):]
    return system_msgs + trimmed

优点：实现简单、效果可预期。 缺点：早期关键信息（如用户自我介绍）会被截掉。

4.3 方案二：Token 估算截断

按 token 数截断，比按轮数更精准。可以用 tiktoken（OpenAI）或模型对应的 tokenizer 估算：

python# 代码为示例，具体接口/字段以官方文档为准
# 以下使用 tiktoken 估算 token 数（适用于 GPT 系列）
# 若使用其他 LLM，请替换为对应 tokenizer
import tiktoken

enc = tiktoken.get_encoding("cl100k_base")
MAX_TOKENS = 3000  # 为回复预留空间，不要占满整个 context

def count_tokens(messages: list) -> int:
    total = 0
    for m in messages:
        total += len(enc.encode(m.get("content", ""))) + 4  # 每条消息固定开销约4 token
    return total

def trim_history_by_tokens(history: list) -> list:
    system_msgs = [m for m in history if m["role"] == "system"]
    dialog_msgs = [m for m in history if m["role"] != "system"]
    result = []
    token_count = count_tokens(system_msgs)
    # 从最新消息往前保留
    for msg in reversed(dialog_msgs):
        cost = len(enc.encode(msg.get("content", ""))) + 4
        if token_count + cost > MAX_TOKENS:
            break
        result.insert(0, msg)
        token_count += cost
    return system_msgs + result

4.4 方案三：摘要压缩（适合长会话）

当历史过长时，不简单截断，而是把早期历史"摘要化"后作为一条 system 消息保留，这样关键信息不丢失：

python# 代码为示例，具体接口/字段以官方文档为准
def summarize_old_history(old_msgs: list, llm_client) -> str:
    """
    调用 LLM 把旧的对话历史压缩成一段摘要
    """
    prompt = "请把以下对话历史压缩成一段100字以内的摘要，保留关键事实和用户意图：\n"
    for m in old_msgs:
        prompt += f"{m['role']}: {m['content']}\n"
    summary = llm_client.chat(prompt)
    return f"[历史摘要] {summary}"

def smart_trim(history: list, threshold: int, llm_client) -> list:
    if count_tokens(history) <= threshold:
        return history
    # 把前半段摘要化
    mid = len(history) // 2
    summary_text = summarize_old_history(history[:mid], llm_client)
    new_history = [{"role": "system", "content": summary_text}] + history[mid:]
    return new_history

五、完整处理流程

5.1 消息处理主循环

python# 代码为示例，具体接口/字段以官方文档为准
BASE  = "https://你的接口域名"   # 注册后在官方文档获取
TOKEN = "你的Token"
APPID = "你的appId"
HEADERS = {"token": TOKEN}       # 鉴权字段名以官方文档为准

import requests

def send_text(to_wxid: str, content: str):
    url = f"{BASE}/message/postText"
    payload = {"appId": APPID, "toWxid": to_wxid, "content": content}
    resp = requests.post(url, json=payload, headers=HEADERS, timeout=10)
    return resp.json()

def call_llm(messages: list) -> str:
    """
    调用 LLM 接口，此处为占位示例
    替换为你实际使用的 LLM SDK 调用
    """
    # e.g. openai.chat.completions.create(model="gpt-4o", messages=messages)
    raise NotImplementedError("请替换为实际的LLM调用")

def process_message(msg: dict):
    session_key = make_session_key(msg)
    user_text = msg.get("content", "").strip()
    from_wxid = msg.get("fromWxid", "")

    # 支持用户主动清除上下文
    if user_text in ["重置", "清除记忆", "/reset"]:
        conversation_store[session_key] = []
        send_text(from_wxid, "好的，已清除历史记录，我们重新开始吧。")
        return

    # 获取并更新历史
    history = get_history(session_key)
    history.append({"role": "user", "content": user_text})

    # 截断控制（选用上文任意一种方案）
    trimmed = trim_history_by_turns(history)

    # 调用 LLM
    try:
        reply = call_llm(trimmed)
    except Exception as e:
        reply = "抱歉，我暂时无法回答，请稍后再试。"
        print(f"LLM error: {e}")

    # 把 assistant 回复也存入历史
    history.append({"role": "assistant", "content": reply})

    # 发送回复
    send_text(from_wxid, reply)

5.2 System Prompt 的设计建议

System prompt 是塑造机器人"人格"和"能力边界"的关键。建议写清楚：

pythonSYSTEM_PROMPT = """你是一个专业的客服助手，性格友好、简洁。
- 不确定的事情请明确说"我不确定"，不要捏造信息
- 每次回复控制在200字以内
- 如果用户问到非本业务范围的问题，礼貌拒绝并引导回主题
"""

在组织传给 LLM 的 messages 时，system prompt 始终放在第一条：

pythondef build_llm_messages(session_key: str, new_user_msg: str) -> list:
    history = get_history(session_key)
    history.append({"role": "user", "content": new_user_msg})
    trimmed = trim_history_by_turns(history)
    return [{"role": "system", "content": SYSTEM_PROMPT}] + trimmed

六、消息收发接口接入说明

微信消息的收发需要借助能操控微信客户端的 HTTP 接口层。WechatApi 提供扫码登录、消息收发、好友与群管理等 REST 接口，HTTP 调用即可，适合快速集成到上述多轮对话框架中（WechatApi）。

登录完成后，用 setCallback 把你的服务地址注册为回调，此后平台会将收到的微信消息实时推送过来；发消息则调用 postText 接口，传入 appId、目标 toWxid 和 content 即可。具体接口路径、字段名以官方文档为准，代码中的占位符需替换为实际值。

七、常见问题与调试技巧

7.1 机器人重复回复

原因：回调接口响应超时，平台重试；或者消息队列重复消费。

解法：用 msgId 做去重，消费前先检查该 ID 是否已处理：

python# 代码为示例，具体接口/字段以官方文档为准
processed_ids = set()   # 生产环境用 Redis SET + TTL

def is_duplicate(msg_id: str) -> bool:
    if msg_id in processed_ids:
        return True
    processed_ids.add(msg_id)
    return False

7.2 上下文窗口污染

现象：用户换了话题，机器人还在回答上一个话题的内容。

解法：

1. 增加超时重置（30分钟无消息自动清除）
2. 识别话题切换关键词触发重置
3. 在 system prompt 里明确"如果用户明显切换话题，请承认切换并从新话题出发"

7.3 LLM 调用延迟导致用户体验差

微信用户等待超过 3-5 秒就会感到不耐烦。几个缓解技巧：

• 收到消息后先发一条"思考中..."占位，再用异步任务调 LLM
• 对明显无需 LLM 的指令（如"重置"、"帮助"）走快速分支直接回复
• 使用流式输出（streaming）并在完整回复到达后一次性发送

7.4 群聊中的 @ 处理

群消息里机器人通常只响应被 @ 的消息，需要从 content 里解析出 @ 内容：

python# 代码为示例，具体接口/字段以官方文档为准
def extract_group_mention(content: str, bot_wxid: str) -> str | None:
    """
    如果消息包含 @机器人，返回去掉@部分后的纯文本；否则返回 None
    实际 @ 格式以平台文档为准
    """
    mention_tag = f"@{bot_wxid}"
    if mention_tag in content:
        return content.replace(mention_tag, "").strip()
    return None

7.5 防封建议

调用发消息接口时注意频率控制：同一个 toWxid 不要连续快速发送多条消息，建议加随机延迟（1-3 秒）；批量场景下使用队列限速，避免短时间内发送频率过高触发风控。

总结

多轮对话的核心是在无状态的消息推送体系之上，用业务代码维护好每个用户的对话历史。关键点包括：以 fromWxid（或群 ID）为 Key 隔离会话，选择合适的上下文截断策略（固定轮数、Token 限制或摘要压缩），以及通过超时重置和主动清除指令让用户有掌控感。把这套状态管理逻辑和可靠的微信消息收发接口结合起来，就能搭建出一个真正"记得住"用户意图的微信 AI 机器人。