---

背景痛点：为什么“调通”≠“好用”

第一次把 ChatGPT 接入产品，我踩了三个大坑，全是血泪。

1. 认证失败：把 OPENAI_API_KEY 硬编码在 config.py，上线前忘了改，结果 GitHub 仓库被扫描器瞬间扒走，Key 在两小时内被刷掉 18 美元额度。
2. 上下文丢失：用户问“明天北京天气如何？”紧接着再问“那我该穿羽绒服吗？”——第二句被当成独立请求，模型直接答非所问，体验瞬间崩塌。
3. 响应延迟：默认的 REST 接口要等整包返回，平均 2.3 s TTFB（Time To First Byte），在移动端弱网环境下用户疯狂点“重试”，服务器 502 一片。

后来我才意识到，REST 与 Streaming 不是谁好谁坏，而是场景不同：

• REST：适合一次性问答、对延迟不敏感、需要完整日志落盘的后台任务。
• Streaming：适合多轮对话、需要边想边吐字的前台场景，能把“首字时间”降到 400 ms 以内，用户心理等待减半。

下面把趟过的坑、沉淀出的代码全部打包，给你一份“可直接搬”的生产级模板。

---

核心实现：30 分钟跑通最小可用闭环

1. 环境准备与认证头封装

先装官方库（已升级到 ≥1.0）：

pip install openai python-dotenv

在项目根目录放 .env：

OPENAI_API_KEY="sk-xxx"
OPENAI_ORG_ID="org-xxx"   # 可选

client.py：单例客户端 + 类型注解，避免在循环里反复 openai.OpenAI() —— 那是烧钱加速器。

import os
from openai import OpenAI
from typing import Optional

_client: Optional[OpenAI] = None

def get_client() -> OpenAI:
    global _client
    if _client is None:
        _client = OpenAI(
            api_key=os.getenv("OPENAI_API_KEY"),
            organization=os.getenv("OPENAI_ORG_ID") or None,
            max_retries=0,  # 我们自己写重试策略
        )
    return _client

避坑指南：

• 不要把 api_key 参数写成 os.environ["OPENAI_API_KEY"]，一旦 CI 环境漏配会直接抛 KeyError 中断部署。用 os.getenv + 默认值最稳。

---

2. 多轮对话上下文维护（含 token 计数）

官方没有“会话”概念，得自己把 messages[] 带回来。封装一个 Conversation 类，自动统计 token 消耗（token consumption），接近上限时自动滑动窗口。

import tiktoken
from typing import List, Dict

class Conversation:
    def __init__(self, model: str = "gpt-3.5-turbo", max_tokens: int = 3500):
        self.model = model
        self.max_tokens = max_tokens
        self.messages: List[Dict[str, str]] = []
        self._encoder = tiktoken.encoding_for_model(model)

    def add(self, role: str, content: str) -> None:
        self.messages.append({"role": role, "content": content})
        self._drop_if_overflow()

    def _count(self, msgs: List[Dict[str, str]]) -> int:
        # 简单实现：每条 role + content 都算
        return sum(len(self._encoder.encode(m["role"] + m["content"])) for m in msgs)

    def _drop_if_overflow(self) -> None:
        while self._count(self.messages) > self.max_tokens:
            # 保留 system，去掉最早的用户/助手对
            if len(self.messages) > 2 and self.messages[0]["role"] == "system":
                del self.messages[1:3]
            else:
                del self.messages[:2]

调用示例：

conv = Conversation()
conv.add("system", "你是小助手，回答尽量简洁。")
conv.add("user", "明天北京天气？")
# 发请求前把 conv.messages 直接塞给 openai

---

3. 流式响应处理（chunk 拼接）

Streaming 不是“一行一句”，而是 SSE（Server-Sent Events）片段，需要把 delta.content 累加。

from openai import Stream
from openai.types.chat import ChatCompletionChunk

def stream_reply(messages: List[Dict[str, str]], temperature: float = 0.7):
    client = get_client()
    stream: Stream[ChatCompletionChunk] = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages,
        temperature=temperature,
        stream=True,
    )
    buffer = ""
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            buffer += delta
            yield delta  # 实时吐给前端
    # buffer 里即是完整回复，可入库

前端只需 EventSource 接 yield 的片段，打字机效果立现。

---

生产级考量：让服务敢在晚上安心睡觉

1. 错误处理：429 的指数退避重试

OpenAI 按“组织 + 模型”维度限流，超了返回 429。官方库默认重试 2 次，间隔 1 s 固定，不适合高并发。自写退避：

import random
import time
from openai import RateLimitError

def call_with_backoff(func, *args, **kwargs):
    delay = 1
    for attempt in range(5):
        try:
            return func(*args, **kwargs)
        except RateLimitError as e:
            sleep = delay + random.uniform(0, 0.5)
            time.sleep(sleep)
            delay *= 2  # 指数退避
    raise RuntimeError("仍被限流，人工介入")

把 func 换成 client.chat.completions.create 即可。

---

2. 性能优化：max_tokens & temperature 调节

• max_tokens：设置过小会截断，过大浪费额度。可先用 tiktoken 估算用户输入，再按“剩余 50 %”给回答。
• temperature：0.2 适合客服/知识库，0.8 适合创意写作。根据业务场景动态调整，后面留一个“课后挑战”给你练手。

---

3. 安全防护：密钥轮换 + 请求限流

• 密钥轮换：在火山引擎或 AWS Secrets Manager 存 Key，每 30 天自动滚动；服务启动时加载到内存，避免落盘。
• 请求限流：用 Redis + Token Bucket，单用户 60 req/min，超了直接返回 429，别浪费 OpenAI 额度。

---

避坑指南（持续更新）

• 避免在循环中实例化 OpenAI 客户端：全局单例可减少 30 % 延迟。
• 不要把 assistant 的完整回复再塞回 messages，只存用户输入和系统提示，否则 token 翻倍。
• 生产环境勿用 print(chunk) 调试，容易把敏感内容打到日志；用 logging.debug("delta: %s", delta) 并设日志脱敏规则。
• 别忽略 finish_reason == length：说明被 max_tokens 截断，需要前端提示“回答被截断，请缩小问题范围”。

---

互动引导：把知识变成肌肉记忆

1. Postman 集合下载
   我打包了本篇文章所有示例请求（含环境变量、测试脚本），导入即可一键发 Streaming 请求。
   下载地址：ChatGPT-API-Collection.postman_collection.json

2. 课后挑战
   实现“根据用户输入动态调整 temperature”：

   • 检测到“写一首关于秋天的诗” → temperature = 0.9
   • 检测到“Node.js 如何读取文件” → temperature = 0.2
     提交 PR 或留言贴 Gist，我会挑 3 份代码送《Python 高级编程》纸质书。

---

写在最后：把玩具变成产品，还差几步？

走完上面的代码，你已经有：

• 稳定可重试的客户端
• 带 token 计数的多轮对话
• 流式打字机效果
• 生产级限流与密钥管理

但真正的“产品”还要考虑角色人设、音色选择、打断降噪、端到端延迟监控…… 这些我在 从0打造个人豆包实时通话AI 动手实验里全部跑通：

• 用火山引擎豆包语音大模型，把 ASR→LLM→TTS 串成 400 ms 低延迟闭环
• 提供现成的 Web 脚手架，本地 npm run dev 即可麦克风对话
• 手把手教你调音色、改 Prompt、加打断唤醒词，全程注释清晰，小白也能跟下来

我亲自跑完一遍，大概 45 分钟搞定，比自己从 0 写节省至少 3 天。如果你正好想把“静态问答”升级成“实时通话”，不妨去戳链接体验，回来记得交作业：把 temperature 动态调节的代码也搬到豆包上，看能不能让诗人与程序员同屏聊天。

---