从零实现：如何将ChatGPT无缝接入微信的工程实践

3次阅读

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

微信官方 API 的限制与第三方协议的风险权衡
微信官方未开放个人号 API，只能依赖第三方库通过 Web 协议模拟登录。ItChat 这类库虽然易用，但存在账号风控风险，需要谨慎控制调用频率
通过抓包分析发现，Web 协议单账号消息接收 QPS 超过 20 条时，极容易触发登录态失效
ChatGPT 长文本回复与微信消息长度限制的矛盾
微信单条消息长度限制为 2048 字节（约 682 个中文字符），而 GPT- 4 的典型回复常超过该限制
实测显示，超过限制的消息会被微信服务器直接丢弃，不会进入重试队列
多用户并发场景下的会话隔离需求
当 5 个以上用户同时提问时，内存中的会话字典会出现键冲突
上下文关联的对话需要保持至少 10 分钟的状态记忆，这对存储设计提出要求

协议库对比
ItChat：基于 Requests 的同步实现，开发简单但吞吐量低（实测约 15QPS）
PyWeChatSpy：使用异步 IO 但协议逆向程度较深，存在更高封号风险
最终选择 ItChat+ 多进程方案，平衡开发效率与风险控制

异步方案性能数据

# 测试环境：4 核 8G 云服务器
| 方案          | 平均响应时间 | 最大 QPS |
|---------------|-------------|--------|
| 同步请求      | 1200ms      | 18     |
| aiohttp 异步   | 380ms       | 53     |

上下文存储决策
内存缓存：实现简单但进程间无法共享，重启丢失数据
Redis：支持 TTL 自动过期，但需要处理序列化开销
采用 Redis+MessagePack 的方案，相比 JSON 节省 37% 存储空间

消息监听装饰器

def retry(max_attempts=3, delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except WeChatError as e:
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(delay * (attempt + 1))
        return wrapper
    return decorator

@itchat.msg_register(TEXT)
@retry()
def handle_message(msg):
    ...

GPT 响应分块处理

def chunk_response(text, max_len=600):
    sentences = re.split(r'(?<=[。！？])', text)
    chunks, current = [], ''
    for s in sentences:
        if len(current + s) > max_len:
            chunks.append(current)
            current = s
        else:
            current += s
    if current:
        chunks.append(current)
    return chunks

Redis 会话管理

-- KEYS[1]:user_id, ARGV[1]:new_msg, ARGV[2]:expire_time
local history = redis.call('GET', KEYS[1])
if not history then
    history = '[]'
end
local decoded = cjson.decode(history)
table.insert(decoded, ARGV[1])
redis.call('SETEX', KEYS[1], ARGV[2], cjson.encode(decoded))
return #decoded

消息队列配置

# Kafka 生产者配置
producer = KafkaProducer(bootstrap_servers=['kafka:9092'],
    value_serializer=lambda v: msgpack.packb(v),
    compression_type='gzip'
)

风控规避实践
心跳间隔设置为 25-35 秒随机值，避免固定周期
消息发送间隔加入 0.1-0.3 秒的随机延迟
每日主动退出登录一次，模拟人工行为

监控埋点示例

from prometheus_client import Counter
MSG_RECEIVED = Counter('wx_msg_total', 'Received messages', ['user'])

@itchat.msg_register(TEXT)
def handle_message(msg):
    MSG_RECEIVED.labels(user=msg['FromUserName']).inc()

内容过滤策略
使用 Trie 树实现敏感词过滤，响应时间 <2ms
政治类词汇使用 SHA256 哈希比对，避免明文存储
自动拒绝包含 15 个以上 URL 的消息

自动登录方案

def keep_alive():
    while True:
        if not itchat.check_login():
            itchat.auto_login(hotReload=True)
        time.sleep(random.randint(20, 40))

AC 自动机集成

class ACFilter:
    def __init__(self, keywords):
        self.trie = {}
        for word in keywords:
            node = self.trie
            for char in word:
                node = node.setdefault(char, {})
            node['__end__'] = True