从零开始：给微信群聊机器人接入ChatGPT的完整指南

2次阅读

没有评论

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

微信机器人开发本质上是通过模拟用户行为或调用官方 / 非官方 API 实现自动消息处理。常见技术路线包括：

网页版协议 ：基于逆向工程模拟微信网页端操作，但稳定性差且易被封号
桌面协议 ：hook 微信客户端实现，风险更高
企业微信接口 ：官方支持但需企业资质
第三方框架 ：如 Wechaty、itchat 等封装了底层协议

ChatGPT API 则提供了标准化的对话接口，核心优势在于：

上下文感知能力
支持多轮对话
丰富的知识覆盖

优点：
多协议支持（PadLocal/Windows 等）
TypeScript 生态完善
社区活跃
缺点：
需要付费协议
学习曲线较陡

优点：
Python 原生支持
简单易上手
缺点：
仅支持网页协议
已停止维护

优点：
完全可控
缺点：
开发成本高
风控风险大

使用 Wechaty 接收消息示例：

from wechaty import Wechaty, Message

class MyBot(Wechaty):
    async def on_message(self, msg: Message):
        text = msg.text()
        room = msg.room()
        if room:
            print(f'群 [{await room.topic()}] {msg.talker().name}: {text}')
        else:
            print(f'私聊 {msg.talker().name}: {text}')

关键处理逻辑：

过滤系统消息
识别 @消息
处理图片 / 语音等多媒体

推荐使用官方 openai 库：

import openai
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def chatgpt_query(prompt, context=None):
    messages = [{"role": "system", "content": "你是有帮助的 AI 助手"}]
    if context:
        messages.extend(context)
    messages.append({"role": "user", "content": prompt})

    try:
        response = await openai.ChatCompletion.acreate(
            model="gpt-3.5-turbo",
            messages=messages,
            temperature=0.7
        )
        return response['choices'][0]['message']['content']
    except Exception as e:
        print(f"API 调用失败: {str(e)}")
        return "服务暂时不可用，请稍后再试"

import asyncio
from concurrent.futures import ThreadPoolExecutor

# 创建线程池处理阻塞操作
executor = ThreadPoolExecutor(max_workers=4)

async def process_message(msg):
    # 步骤 1：预处理（IO 密集型）loop = asyncio.get_event_loop()
    clean_text = await loop.run_in_executor(executor, preprocess, msg.text())

    # 步骤 2：调用 ChatGPT（网络 IO）response = await chatgpt_query(clean_text)

    # 步骤 3：发送回复
    await msg.say(response)

推荐使用百度内容审核 API：

import aiohttp

async def content_check(text):
    url = "https://aip.baidubce.com/rest/2.0/solution/v1/text_censor/v2/user_defined"
    params = {"text": text}
    headers = {"Content-Type": "application/x-www-form-urlencoded"}

    async with aiohttp.ClientSession() as session:
        async with session.post(url, data=params, headers=headers) as resp:
            result = await resp.json()
            return result["conclusionType"] == 1

控制消息频率（<30 条 / 分钟）
避免相同内容重复发送
使用真人操作模式（随机延迟）

from ratelimit import limits, sleep_and_retry

# 限制每分钟 3 次调用
@sleep_and_retry
@limits(calls=3, period=60)
async def safe_api_call():
    # API 调用代码

使用 Redis 存储对话上下文：

import redis
import pickle

r = redis.Redis()

def save_context(user_id, messages):
    r.set(f"ctx:{user_id}", pickle.dumps(messages), ex=3600)

def load_context(user_id):
    data = r.get(f"ctx:{user_id}")
    return pickle.loads(data) if data else []

# bot.py
import asyncio
from wechaty import Wechaty, Message
from chatgpt import chatgpt_query
from safety import content_check

class ChatGPTBot(Wechaty):
    async def on_message(self, msg: Message):
        # 跳过自己发送的消息
        if msg.is_self():
            return

        # 只处理文本消息
        if not msg.text():
            return

        try:
            # 内容安全检查
            if not await content_check(msg.text()):
                await msg.say("内容包含敏感信息")
                return

            # 获取回复
            response = await chatgpt_query(msg.text())
            await msg.say(response[:500])  # 限制回复长度

        except Exception as e:
            print(f"处理消息失败: {e}")
            await msg.say("机器人暂时无法响应")

async def main():
    bot = ChatGPTBot()
    await bot.start()

asyncio.run(main())