服务器端集成ChatGPT API的架构设计与性能优化实战

2次阅读

没有评论

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

在服务器端集成 ChatGPT API 时，开发者常面临三个核心挑战：

API 调用延迟：OpenAI 的 API 服务器可能位于海外，单次请求的网络延迟通常在 200-500ms
并发限制：免费层每分钟仅允许 3 次请求，付费版也有 TPM（Tokens Per Minute）限制
token 成本控制：GPT- 4 的输入 token 成本是 $0.03/1k tokens，长文本交互可能产生意外费用

这些因素直接影响服务响应速度和运营成本，特别是对需要实时交互的应用场景。

REST API：
优点：实现简单，兼容性强
缺点：每次请求需要建立新 TCP 连接，HTTP 头开销大
gRPC：
优点：二进制协议体积小，支持多路复用
缺点：目前 OpenAI 官方未提供 gRPC 接口

通过连接池复用 TCP 连接，可以减少 TCP 三次握手和 TLS 协商的时间开销。测试显示，复用连接可使延迟降低 30% 以上。

使用 Python 的 aiohttp 库创建连接池：

import aiohttp
from contextlib import asynccontextmanager

class ChatGPTClient:
    def __init__(self, api_key, pool_size=10):
        self.api_key = api_key
        connector = aiohttp.TCPConnector(
            limit=pool_size,
            force_close=False,
            enable_cleanup_closed=True
        )
        self.session = aiohttp.ClientSession(connector=connector)

    @asynccontextmanager
    async def get_session(self):
        try:
            yield self.session
        finally:
            pass  # 保持连接活跃

收集 200ms 窗口期内的所有请求
合并相同 temperature 参数的请求
将多个 prompt 用特殊分隔符拼接
在 API 响应后拆分结果

采用 LRU 缓存 +TTL 过期策略：

from functools import lru_cache
import time

class ChatCache:
    @lru_cache(maxsize=1024)
    def get_response(self, prompt: str) -> str:
        # 实际调用 API 的逻辑
        pass

    def __call__(self, prompt):
        current_time = time.time()
        if (cached := self._cache.get(prompt)) and current_time - cached["time"] < self.ttl:
            return cached["response"]
        # ... 调用真实 API

import asyncio
import random

async def call_api_with_retry(session, payload, max_retries=3):
    backoff = 1
    for attempt in range(max_retries):
        try:
            async with session.post(
                "https://api.openai.com/v1/chat/completions",
                json=payload,
                headers={"Authorization": f"Bearer {API_KEY}"}
            ) as resp:
                return await resp.json()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(backoff + random.uniform(0, 1))
            backoff *= 2

from collections import deque
import time

class RateLimiter:
    def __init__(self, rpm):
        self.interval = 60 / rpm
        self.timestamps = deque(maxlen=rpm)

    async def acquire(self):
        now = time.time()
        if len(self.timestamps) >= self.timestamps.maxlen:
            elapsed = now - self.timestamps[0]
            if elapsed < 60:
                await asyncio.sleep(60 - elapsed)
        self.timestamps.append(time.time())

并发数	原生 API 延迟	优化后延迟
5	1200ms	600ms
20	超频错误	900ms
50	完全阻塞	1500ms

对历史对话进行语义摘要而非完整存储
设置 max_tokens 硬限制
使用 gpt-3.5-turbo 处理简单请求

import boto3
from botocore.exceptions import ClientError

def get_secret():
    client = boto3.client("kms")
    try:
        return client.decrypt(CiphertextBlob=base64.b64decode(encrypted_key)
        )["Plaintext"].decode()
    except ClientError as e:
        # 处理错误

import re

def sanitize_input(text):
    patterns = [r"\d{4}-\d{4}-\d{4}-\d{4}",  # 信用卡号
        r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"  # 邮箱
    ]
    for pattern in patterns:
        text = re.sub(pattern, "[REDACTED]", text)
    return text