ChatGPT API 高效集成指南：从零构建企业级对话系统

1次阅读

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

直接调用 ChatGPT API 时，开发者常遇到三类典型问题：

Token 计算误差：GPT 模型按 token 计费，但中文 / 英文 token 折算比例不同（中文通常 1 字≈1.5token），手动统计容易产生预算偏差
上下文丢失：默认 API 无状态，多轮对话需自行维护历史消息，不当的截断策略会导致 ” 遗忘 ” 关键信息
突发流量限流：免费层每分钟仅 3 次请求，即使付费版也有 TPM（Tokens Per Minute）限制，突发流量易触发 429 错误

采用分层设计，核心组件包括：

持久化层：使用 SQLite 存储对话元数据（session_id, user_id, created_at）
缓存层：Redis 存储最近 20 轮对话的原始消息（设置 TTL 为 24 小时）
服务层：异步处理请求，实现消息分块和 token 计数

# 数据库模型示例（SQLAlchemy）class Conversation(Base):
    __tablename__ = 'conversations'
    id = Column(String(36), primary_key=True)
    user_id = Column(String(64), index=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    title = Column(String(128))  # 自动生成对话摘要

通过 Redis 的有序集合 (zset) 实现消息时序管理：

使用 session_id 作为 key，消息时间戳作为 score
每次新增消息执行 ZADD 操作
获取上下文时按 score 范围查询

# Redis 操作示例
async def add_message(session_id: str, message: dict):
    timestamp = time.time()
    await redis.zadd(f"chat:{session_id}",
        {json.dumps(message): timestamp}
    )
    # 保持最近 20 条
    await redis.zremrangebyrank(f"chat:{session_id}", 0, -21
    )

采用指数退避算法处理限流错误：

初始延迟 1 秒
每次重试延迟时间倍增
最大重试 3 次

from openai import RateLimitError
import asyncio

async def request_with_retry(prompt, max_retries=3):
    delay = 1
    for attempt in range(max_retries):
        try:
            return await openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=prompt
            )
        except RateLimitError:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(delay)
            delay *= 2

通过迭代器逐步返回结果，提升用户体验：

async def stream_response(messages):
    response = await openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=messages,
        stream=True,
        temperature=0.7  # 控制创造性
    )

    full_content = ""
    async for chunk in response:
        content = chunk["choices"][0].get("delta", {}).get("content", "")
        if content:
            full_content += content
            yield content

    # 最终保存完整响应
    await save_completion(full_content)

精确计算每次请求的 token 消耗（使用官方 tiktoken 库）：

import tiktoken

encoder = tiktoken.encoding_for_model("gpt-3.5-turbo")

def count_tokens(messages):
    """计算消息列表的总 token 数"""
    tokens_per_message = 3  # 每条消息的系统开销
    tokens_per_name = 1     # 角色名的额外开销

    total = 0
    for message in messages:
        total += tokens_per_message
        for key, value in message.items():
            total += len(encoder.encode(value))
            if key == "name":
                total += tokens_per_name
    return total

模型版本	输入单价（/1K tokens）	输出单价（/1K tokens）
gpt-3.5-turbo	$0.0015	$0.002
gpt-4	$0.03	$0.06

建议策略：

对非关键业务使用 3.5 版本
设置每次对话的 max_tokens 限制
监控每日 token 消耗（可通过 CloudWatch 等工具）

使用 Prometheus + Grafana 监控 P99 延迟：

from prometheus_client import Summary

REQUEST_LATENCY = Summary(
    'openai_api_latency_seconds',
    'API response latency distribution'
)

@REQUEST_LATENCY.time()
async def call_openai_api():
    # API 调用代码

推荐两步过滤法：

前置过滤：使用正则匹配明显违规词
后置过滤：调用 OpenAI 的 moderation 端点

from openai import Moderation

async def safety_check(text):
    response = await Moderation.create(input=text)
    return response["results"][0]["flagged"]

当对话历史超过模型限制（如 4096 tokens）时：

优先保留最近消息
使用摘要压缩早期对话
关键信息提取为元数据

def truncate_conversation(messages, max_tokens=3000):
    """智能截断对话历史"""
    while count_tokens(messages) > max_tokens:
        # 移除最早的非系统消息
        for i, msg in enumerate(messages):
            if msg["role"] != "system":
                messages.pop(i)
                break
    return messages

国内访问需设置代理，注意两点：

使用 HTTP_PROXY 环境变量
配置 requests/httpx 的代理参数

import os
from openai import OpenAI

os.environ["HTTP_PROXY"] = "http://127.0.0.1:10809"

client = OpenAI(
    timeout=10,
    http_client=httpx.Client(proxies="http://127.0.0.1:10809")
)