很多开发者用 Python 调用 GPT 时,常遇到版本不兼容、密钥泄露、代理不通、费用超支等问题。本文从零开始,覆盖 SDK 安装、认证、基础调用、流式输出、异步并发、成本控制、错误处理及模型迁移,每个环节都有可直接运行的代码,帮你避开常见坑。
环境准备与 SDK 安装
选对库的版本
OpenAI Python 库 v1.x 相比 v0.x 接口变化很大。旧版 openai.ChatCompletion.create() 已废弃,会导致 AttributeError。需安装最新版并使用客户端模式。
pip install openai
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "你好"}]
)
安全管理 API 密钥
不要将密钥硬编码,推荐使用环境变量配合 .env 文件。
from dotenv import load_dotenv
import os
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")
client = OpenAI(api_key=api_key)
若需代理,设置 HTTP_PROXY 和 HTTPS_PROXY 环境变量。
os.environ["HTTP_PROXY"] = ""
os.environ["HTTPS_PROXY"] = ""
client = OpenAI()
基础调用:发请求与解析回复
构造消息时包含 role 和 content,并检查 response.choices 是否为空。
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是一位编程助手"},
{"role": "user", "content": "写一个斐波那契数列函数"}
],
temperature=0.7,
max_tokens=500
)
reply = response.choices[0].message.content
生产环境务必设置超时和重试。
client = OpenAI(timeout=30.0, max_retries=3)
流式输出:实现打字机效果
使用 stream=True 逐块获取内容,注意 delta.content 可能为 None 需判空。
stream = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "讲个笑话"}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="")
在 FastAPI 中集成 SSE 需用 StreamingResponse,高并发时应改用 AsyncOpenAI。
异步并发:同时处理多个请求
使用 AsyncOpenAI 配合 asyncio.gather 提高吞吐量。
import asyncio
from openai import AsyncOpenAI
async_client = AsyncOpenAI()
async def ask(prompt):
response = await async_client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
max_tokens=200
)
return response.choices[0].message.content
async def main():
prompts = ["解释量子计算", "推荐三本小说", "写一段 Python 代码"]
results = await asyncio.gather(*[ask(p) for p in prompts])
for r in results:
print(r)
asyncio.run(main())
Jupyter Notebook 中需用 nest_asyncio 解决循环冲突。
成本控制:Token 统计与参数优化
用 tiktoken 预估 token 数,结合参数设置可节省 30%~60%。核心策略:
max_tokens:按任务设定,分类任务 100 即可。
stop:设置停止序列,避免无效输出。
temperature:事实类任务设为 0 减少随机性。
缓存重复请求:用内存或 Redis 减少调用。
import tiktoken
def count_tokens(text, model="gpt-4o"):
encoding = tiktoken.encoding_for_model(model)
return len(encoding.encode(text))
策略适用场景节省比例设 max_tokens=150短回答任务50%~80%使用 stop 序列结构化输出20%~40%缓存相同输入FAQ 重复查询90%~100%
错误处理:高频问题定位
401 AuthenticationError:密钥无效或未加载,检查环境变量及前缀
sk-。Timeout:网络延迟或代理不通,增大 timeout 或改用流式。
BadRequestError(上下文过长):超出模型最大 token,用 tiktoken 预先截断历史消息。
响应解析异常:始终先判断
response.choices不为空,再访问delta.content。
模型选择与迁移
模型上下文长度特点适用场景gpt-4o128K多模态,速度快,性价比高复杂推理、长文档分析gpt-3.5-turbo16K成本低,响应快简单问答、分类、清洗
迁移时注意响应格式、定价差异和系统提示敏感度变化。切换前建议在小流量测试集上对比质量和 token 消耗。
实战踩坑汇总
.env中密钥不加引号,避免读取时带引号导致认证失败。代理需同时设
HTTP_PROXY和HTTPS_PROXY,并确认协议支持。max_tokens不要硬编码,按模型上限动态调整。流式输出最后一个 chunk 不含
usage,统计消耗需用非流式方式或手动拼接。使用
AsyncOpenAI后记得调用await async_client.close()释放连接。
总结
面对 GPT 调用任务,按以下决策树快速选择:
需实时显示?→ 流式输出 + SSE
需并发处理?→ AsyncOpenAI + gather
成本敏感?→ tiktoken 预估算 + 参数优化 + 缓存
需结构化输出?→ 考虑函数调用或 response_format
模型更新?→ 先小流量验证
本文提供了从安装到上线的完整代码片段,理解原理后再因地制宜调整参数,才能真正解决生产问题。