ChatGPT多人使用实战指南：从零搭建高并发对话系统

1次阅读

没有评论

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

当多个开发者共享同一个 ChatGPT API 时，通常会遇到以下几个典型问题：

会话交叉污染：不同用户的对话历史混淆，导致上下文混乱。
突发流量导致 429 错误：短时间内大量请求触发 OpenAI 的速率限制。
配额超限：共享 API 密钥容易在短时间内耗尽配额。

直接调用 API 虽然简单，但无法解决多人共享时的并发控制和会话隔离问题。搭建代理层则可以提供以下优势：

限流控制：通过令牌桶算法平滑请求流量。
会话隔离：确保每个用户的对话上下文独立。
缓存优化：减少重复请求，节省 API 调用次数。

Redis 作为高性能内存数据库，适合存储临时会话数据。以下是实现方案：

每个用户分配唯一的user_id，作为 Redis 键的一部分。
存储对话上下文时，以 user_id 为前缀，确保隔离。
设置合理的 TTL，避免内存泄漏。

import time
from redis import Redis

class RateLimiter:
    def __init__(self, redis_client: Redis, max_tokens: int, refill_rate: float):
        self.redis = redis_client
        self.max_tokens = max_tokens
        self.refill_rate = refill_rate  # tokens per second

    def acquire(self, user_id: str, tokens: int = 1) -> bool:
        key = f"rate_limit:{user_id}"
        last_time = self.redis.hget(key, "last_time")
        current_tokens = self.redis.hget(key, "tokens")
        now = time.time()

        if last_time is None or current_tokens is None:
            # First request or key expired
            self.redis.hset(key, mapping={
                "last_time": now,
                "tokens": self.max_tokens - tokens
            })
            return True

        last_time = float(last_time)
        current_tokens = float(current_tokens)

        # Calculate refilled tokens
        time_passed = now - last_time
        refilled_tokens = time_passed * self.refill_rate
        new_tokens = min(current_tokens + refilled_tokens, self.max_tokens)

        if new_tokens >= tokens:
            self.redis.hset(key, mapping={
                "last_time": now,
                "tokens": new_tokens - tokens
            })
            return True
        return False

def get_user_context(user_id: str) -> list:
    context_key = f"chat_context:{user_id}"
    context = redis.lrange(context_key, 0, -1)
    return [msg.decode('utf-8') for msg in context]

def save_user_context(user_id: str, messages: list):
    context_key = f"chat_context:{user_id}"
    redis.delete(context_key)
    redis.rpush(context_key, *messages)
    redis.expire(context_key, 3600)  # Expire after 1 hour

在不同并发量下测试 API 响应时间，建议使用 locust 进行压力测试。典型结果如下：