ChatGPT API实战指南：从接入到生产环境优化的全流程解析

1次阅读

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

典型问题分析
Token 计算误差：实际消耗 token 常超出预估，导致账单不可控
流式响应中断：网络波动时半截回复影响用户体验
速率限制：突发流量触发 429 错误（每分钟 3,500 请求限制）
上下文丢失：超过 4,096 tokens 时历史对话被截断
业务影响
直接调用 API 的失败率可达 15%（基于压测数据）
未优化的对话系统单次调用延迟可能超过 5 秒

API 选型对比
| 特性 | Completion API | Chat API |
|————|———————|———————|
| 适用场景 | 单轮文本生成 | 多轮对话 |
| 上下文管理 | 需手动拼接 | 自动维护对话状态 |
| 成本控制 | 难预测 token 消耗 | 可估算对话回合数 |
流式传输决策树
启用 stream 的场景：
- 响应内容超过 512 tokens
- 需要实时显示生成过程
- 移动端网络环境较差时
禁用 stream 的场景：
- 需要精确计算 token 消耗
- 必须保证原子性响应

import httpx
from typing import AsyncGenerator

class ChatGPTClient:
    def __init__(self, api_key: str):
        self.base_url = "https://api.openai.com/v1"
        self.headers = {"Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    async def stream_chat(
        self, 
        messages: list[dict],
        model: str = "gpt-3.5-turbo"
    ) -> AsyncGenerator[str, None]:
        payload = {
            "model": model,
            "messages": messages,
            "stream": True,
            "temperature": 0.7
        }

        async with httpx.AsyncClient(timeout=30.0) as client:
            try:
                response = await client.post(f"{self.base_url}/chat/completions",
                    headers=self.headers,
                    json=payload
                )
                response.raise_for_status()

                async for chunk in response.aiter_lines():
                    if chunk.startswith("data:"):
                        yield chunk[6:].strip()
            except httpx.ReadTimeout:
                yield "[ERROR] 响应超时，请重试"
            except httpx.HTTPStatusError as e:
                yield f"[ERROR] API 请求失败: {e.response.status_code}"

网络中断补偿方案：
记录最后收到的 chunk ID
重连时携带 last_id 参数继续请求
性能优化技巧：
使用 HTTP/ 2 多路复用降低连接开销
设置合理的 TCP keepalive（建议 60 秒）

成本控制矩阵
| 参数 | 质量倾向设置 | 成本倾向设置 |
|————–|——————–|——————–|
| max_tokens | 1024 | 256 |
| temperature | 0.9 | 0.3 |
| top_p | 0.95 | 0.5 |

Redis 幂等设计

import redis
from hashlib import md5

r = redis.Redis()

def request_hash(prompt: str, user_id: str) -> str:
    return md5(f"{prompt}_{user_id}".encode()).hexdigest()

def check_duplicate(hash_key: str, ttl: int = 300) -> bool:
    if r.exists(hash_key):
        return True
    r.setex(hash_key, ttl, "1")
    return False