共计 2746 个字符,预计需要花费 7 分钟才能阅读完成。
背景痛点:初次接入的常见雷区
第一次调用 ChatGPT API 时,开发者常会遇到这些问题:

- API 密钥泄漏:将密钥硬编码在代码中或误提交到 GitHub
- 请求格式错误:未正确设置 headers 或 body 格式
- 流式响应卡顿 :不理解
stream=True参数的使用场景 - 上下文丢失:多轮对话时忘记维护 message 历史
技术选型:Completion API vs Chat API
OpenAI 提供两种接口风格:
- Completion API:适合单次补全任务
- 典型场景:代码补全、文章续写
-
优势:响应速度更快
-
Chat API(推荐新手使用):
- 支持多轮对话上下文
- 可通过 system prompt 设置 AI 角色
- 天然支持对话场景(如客服机器人)
核心实现三要素
1. 基础请求示例(Python)
import httpx
from typing import List, Dict
async def call_chatgpt(messages: List[Dict[str, str]], api_key: str) -> str:
headers = {"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
try:
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.post(
"https://api.openai.com/v1/chat/completions",
json={
"model": "gpt-3.5-turbo",
"messages": messages,
"temperature": 0.7 # 控制创造性
},
headers=headers
)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"]
except httpx.HTTPStatusError as e:
print(f"API 请求失败: {e.response.status_code}")
return ""
2. 构造带上下文的对话
有效的 message 结构应包含三种角色:
system: 设定 AI 行为(如 ” 你是一个资深的 Python 工程师 ”)user: 用户输入assistant: AI 之前的回复
conversation = [{"role": "system", "content": "你是一个幽默的技术顾问"},
{"role": "user", "content": "如何用 Python 发送 HTTP 请求?"}
]
# 每次调用后需要将 AI 回复追加到对话历史
response = await call_chatgpt(conversation, API_KEY)
conversation.append({"role": "assistant", "content": response})
3. 处理非标准响应
当 API 返回复杂 JSON 时建议:
- 使用
.get()方法避免 KeyError - 添加类型检查
- 处理可能的多选项情况
def parse_response(raw: dict) -> str:
if not (choices := raw.get("choices")):
return ""
first_choice = choices[0]
if isinstance(first_choice, dict) and "message" in first_choice:
return first_choice["message"].get("content", "")
return ""
生产环境关键策略
请求限速方案
采用指数退避(exponential backoff)处理限流:
import time
async def safe_call(max_retries=3):
base_delay = 1
for attempt in range(max_retries):
try:
return await call_chatgpt(...)
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
delay = base_delay * (2 ** attempt)
time.sleep(delay)
else:
raise
raise Exception("超出最大重试次数")
敏感内容过滤
建议在客户端增加校验层:
def contains_sensitive(text: str) -> bool:
blacklist = ["暴力", "违禁品"] # 根据业务需求扩展
return any(word in text for word in blacklist)
# 在返回给用户前检查
if contains_sensitive(response):
response = "[内容已过滤]"
六大避坑指南
- Token 计数陷阱
- 中文 1 个 token≈2 个字符
- 使用
tiktoken库精确计算 -
示例:
len(encoding.encode("你好"))输出 3 -
上下文隔离
- 为每个用户会话维护独立 message 数组
-
避免使用全局变量存储对话状态
-
错误码 428 处理
- 表示需要先决条件
-
检查请求头是否缺少
Content-Type -
超时设置
- 同步请求建议 10 秒
-
流式响应建议 30 秒以上
-
温度参数(temperature)
- 0- 2 之间浮动
- 创意场景用 0.7-1.0
-
严谨答案用 0.1-0.3
-
费用监控
- 通过 Usage 头字段统计消耗
- 设置每日预算告警
代码规范要点
- 所有函数添加类型注解
- 使用
try-except包裹网络请求 - 敏感信息从环境变量读取
from pydantic import BaseSettings
class Config(BaseSettings):
api_key: str = Field(..., env="OPENAI_KEY")
config = Config() # 自动从.env 文件加载
延伸挑战:实现对话记忆
尝试扩展上述代码:
- 使用数据库(如 SQLite)持久化对话历史
- 为每个用户会话创建唯一 ID
- 实现自动清理过期会话
# 伪代码示例
class DialogueManager:
def __init__(self):
self.sessions = {} # {session_id: messages}
def get_history(self, session_id: str) -> list:
return self.sessions.get(session_id, [])
实践心得
经过多个项目的实战验证,ChatGPT API 在正确处理以下方面后表现稳定:
- 维护完整的调用上下文
- 实现健壮的错误恢复机制
- 对用户输入做必要清洗
建议从简单场景入手,逐步增加复杂功能。当遇到 429 限流错误时,合理设置退避时间比盲目重试更有效。对于需要高并发的场景,可以考虑使用消息队列缓冲请求。
正文完
发表至: 技术分享
近一天内
