很多开发者用 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_PROXYHTTPS_PROXY 环境变量。

os.environ["HTTP_PROXY"] = ""
os.environ["HTTPS_PROXY"] = ""
client = OpenAI()

基础调用:发请求与解析回复

构造消息时包含 rolecontent,并检查 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 消耗。

实战踩坑汇总

  1. .env 中密钥不加引号,避免读取时带引号导致认证失败。

  2. 代理需同时设 HTTP_PROXYHTTPS_PROXY,并确认协议支持。

  3. max_tokens 不要硬编码,按模型上限动态调整。

  4. 流式输出最后一个 chunk 不含 usage,统计消耗需用非流式方式或手动拼接。

  5. 使用 AsyncOpenAI 后记得调用 await async_client.close() 释放连接。

总结

面对 GPT 调用任务,按以下决策树快速选择:

  • 需实时显示?→ 流式输出 + SSE

  • 需并发处理?→ AsyncOpenAI + gather

  • 成本敏感?→ tiktoken 预估算 + 参数优化 + 缓存

  • 需结构化输出?→ 考虑函数调用或 response_format

  • 模型更新?→ 先小流量验证

本文提供了从安装到上线的完整代码片段,理解原理后再因地制宜调整参数,才能真正解决生产问题。