基于AI Agent的微信自动回复系统：技术实现与避坑指南

1次阅读

没有评论

共计 1923 个字符，预计需要花费 5 分钟才能阅读完成。

微信作为国民级 IM 工具，其自动化处理一直存在几个核心痛点：

API 限制 ：微信官方没有提供开放的消息 API，第三方库都是基于网页版协议逆向实现，稳定性存疑
频率控制 ：消息发送频率过高会导致账号被限制，需要精确控制响应间隔
上下文维护 ：连续对话需要保持会话状态，而微信协议本身是无状态的

目前主流的微信自动化方案有：

itchat：
优点：接口简洁，文档完善，支持热登录
缺点：基于网页版协议，2021 年后可能被逐步限制
wxpy：
优点：封装更高级的接口
缺点：已停止维护
企业微信 API：
优点：官方支持
缺点：需要企业资质

我们选择 itchat+LLM 的组合，因其成本最低且能满足基础需求。

使用 itchat 实现基础消息接收功能：

import itchat

@itchat.msg_register([itchat.content.TEXT])
def text_reply(msg):
    print(f"收到消息: {msg['Text']}")
    # 将消息放入处理队列
    reply_queue.put(msg)

itchat.auto_login(hotReload=True)
itchat.run()

封装 ChatGPT API 作为回复引擎：

import openai

class ChatGPT:
    def __init__(self, api_key):
        openai.api_key = api_key
        self.conversations = {}  # 维护对话上下文

    def reply(self, user_id, text):
        if user_id not in self.conversations:
            self.conversations[user_id] = []

        self.conversations[user_id].append({"role": "user", "content": text})

        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=self.conversations[user_id]
        )

        reply = response.choices[0].message.content
        self.conversations[user_id].append({"role": "assistant", "content": reply})

        # 限制上下文长度
        if len(self.conversations[user_id]) > 6:
            self.conversations[user_id] = self.conversations[user_id][-6:]

        return reply

使用装饰器实现消息间隔控制：

from datetime import datetime, timedelta
import time

last_send_time = {}

def frequency_control(interval=5):
    def decorator(func):
        def wrapper(user_id, *args, **kwargs):
            now = datetime.now()
            if user_id in last_send_time:
                elapsed = (now - last_send_time[user_id]).seconds
                if elapsed < interval:
                    time.sleep(interval - elapsed)

            result = func(user_id, *args, **kwargs)
            last_send_time[user_id] = datetime.now()
            return result
        return wrapper
    return decorator

使用小号测试，避免主号被封
定期检查登录状态
避免发送敏感词汇

try:
    reply = chatgpt.reply(user_id, text)
    itchat.send(reply, toUserName=msg['FromUserName'])
except Exception as e:
    print(f"Error: {e}")
    # 记录错误日志