ChatGPT链接集成实战：如何解决API调用中的并发与稳定性问题

1次阅读

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

最近在项目中集成 ChatGPT API 时，我遇到了几个棘手的问题。首先是并发限制，ChatGPT API 对每分钟的调用次数有严格限制，超出限制会直接返回 429 错误。其次是响应延迟，特别是在高峰时段，API 的响应时间会明显变长，有时甚至超过 30 秒。最后是错误处理，网络波动或 API 临时不可用会导致请求失败，需要有一套健壮的重试机制。

为了解决这些问题，我调研了几种常见的集成方式：

轮询：简单但低效，容易触发 API 限流
Webhook：实时性好，但需要维护回调接口和状态管理
异步队列：解耦调用方和处理方，支持积压处理和自动重试

综合考虑后，我选择了基于消息队列的异步处理架构，因为它能很好地平衡性能和可靠性。

下面是用 Python 实现的基于 RabbitMQ 的异步处理方案：

import pika
import openai
import json
import time
from tenacity import retry, stop_after_attempt, wait_exponential

# 初始化 RabbitMQ 连接
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='chatgpt_requests', durable=True)

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def call_chatgpt_api(prompt):
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"API 调用失败: {str(e)}")
        raise

def process_request(ch, method, properties, body):
    try:
        data = json.loads(body)
        response = call_chatgpt_api(data['prompt'])
        # 处理成功，发送响应
        ch.basic_publish(exchange='',
                         routing_key=properties.reply_to,
                         body=json.dumps({'response': response}))
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        print(f"处理失败: {str(e)}")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='chatgpt_requests', on_message_callback=process_request)
channel.start_consuming()

这个实现有几个关键点：

使用 RabbitMQ 的持久化队列确保消息不丢失
采用 tenacity 库实现指数退避重试
限制预取数量 (prefetch_count) 避免过载
明确确认 (ack) 或拒绝 (nack) 消息

为了进一步提升系统性能，我做了以下优化：

批处理：将多个请求合并为一个 API 调用
缓存：对常见问题的回答进行缓存
指数退避：重试间隔随时间指数增长

批处理实现示例：

def batch_process_requests(requests):
    prompts = [r['prompt'] for r in requests]
    responses = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt} for prompt in prompts]
    )
    return [r.message.content for r in responses.choices]

在实际部署时，我总结了以下几点经验：