从零实现：如何将ChatGPT无缝接入微信的完整技术方案

2次阅读

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

将 ChatGPT 接入微信个人账号听起来很酷，但实际操作中会遇到几个棘手的问题。首先是微信协议的限制，微信官方并不鼓励自动化操作，因此使用第三方库如 itchat 存在被封号的风险。其次是 ChatGPT API 的调用频率限制，免费用户的 API 调用次数有限，频繁调用可能导致服务被暂时禁用。此外，还需要考虑如何管理会话状态，使得 ChatGPT 能够理解上下文，而不是每次回复都像是全新的对话。

在选择微信机器人框架时，我们有几个选项：itchat、WeChatPY 等。itchat 是一个基于网页版微信的 Python 库，使用简单，适合快速开发，但缺点是网页版微信可能随时被腾讯封禁。WeChatPY 则是一个更底层的库，功能强大但学习曲线较陡。考虑到快速实现和易用性，本文选择 itchat 作为微信消息处理的框架。

itchat 提供了简单的 API 来监听和发送微信消息。我们可以使用装饰器来监听特定的消息类型，例如文本消息。以下是一个简单的示例代码：

import itchat

@itchat.msg_register(itchat.content.TEXT)
def text_reply(msg):
    return "收到消息:" + msg['Text']

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

为了避免频繁调用 ChatGPT API，我们可以引入一个消息队列来控制调用频率。例如，使用 Python 的 queue 模块来实现简单的消息队列：

from queue import Queue
import time

message_queue = Queue()

def process_queue():
    while True:
        if not message_queue.empty():
            msg = message_queue.get()
            # 调用 ChatGPT API
            response = chatgpt_api_call(msg)
            # 发送回复
            itchat.send(response, toUserName=msg['FromUserName'])
        time.sleep(1)  # 控制调用频率

为了让 ChatGPT 能够理解上下文，我们需要维护一个会话状态。可以使用字典来存储每个用户的会话历史：

sessions = {}

def get_chatgpt_response(user, message):
    if user not in sessions:
        sessions[user] = []
    sessions[user].append({"role": "user", "content": message})
    response = chatgpt_api_call(sessions[user])
    sessions[user].append({"role": "assistant", "content": response})
    return response

以下是一个完整的 Python 实现，包含微信消息监听、ChatGPT API 封装和异常处理逻辑：

import itchat
import openai
from queue import Queue
import time

# 初始化 ChatGPT API
openai.api_key = "your-api-key"

# 消息队列和会话状态
message_queue = Queue()
sessions = {}

def chatgpt_api_call(messages):
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"API 调用失败: {e}")
        return "抱歉，我暂时无法处理您的请求。"

@itchat.msg_register(itchat.content.TEXT)
def text_reply(msg):
    user = msg['FromUserName']
    message_queue.put(msg)
    return "消息已接收，正在处理..."

def process_queue():
    while True:
        if not message_queue.empty():
            msg = message_queue.get()
            user = msg['FromUserName']
            response = get_chatgpt_response(user, msg['Text'])
            itchat.send(response, toUserName=user)
        time.sleep(1)

if __name__ == "__main__":
    itchat.auto_login(hotReload=True)
    import threading
    threading.Thread(target=process_queue, daemon=True).start()
    itchat.run()

为了避免微信账号被封禁，可以采取以下策略：