OpenAI 是 AI API 的事实标准，其接口设计被绝大多数厂商效仿。截至 2026 年中，OpenAI 提供从轻量推理到旗舰多模态的完整模型谱系，日均处理数万亿 token 的推理请求。本章深入解析 API 的全部核心参数、高级功能、定价策略和最佳实践。
| 属性 | 值 |
|---|---|
| 官网 | https://openai.com |
| API 文档 | https://platform.openai.com/docs |
| 基础 URL | https://api.openai.com/v1 |
| 认证方式 | Bearer Token (API Key) |
| SDK 支持 | Python / Node.js / Go / Java / .NET / Rust |
| API 风格 | RESTful + SSE 流式 |
API Key 通过 platform.openai.com/api-keys 生成。最佳实践包括：
OPENAI_API_KEY 环境变量注入，绝不硬编码在代码中| 模型 | 上下文 | 特点 | 价格（输入/输出，每百万 token） |
|---|---|---|---|
gpt-4o |
128K | 旗舰多模态，支持图像/音频输入 | $2.50 / $10.00 |
gpt-4o-mini |
128K | 轻量快速，性价比最高 | $0.15 / $0.60 |
o4-mini |
200K | 最新推理模型 | $1.10 / $4.40 |
o3 |
200K | 高端推理模型 | $10.00 / $40.00 |
gpt-4.1 |
1M | 超长上下文 | $2.00 / $8.00 |
gpt-4.1-mini |
1M | 超长上下文轻量版 | $0.40 / $1.60 |
gpt-4.1-nano |
1M | 超长上下文经济版 | $0.10 / $0.40 |
选择模型时需综合考虑任务复杂度、成本约束和延迟要求：
| 使用场景 | 推荐模型 | 理由 |
|---|---|---|
| 简单对话/客服 | gpt-4o-mini |
成本低、响应快，足够应对大部分日常问答 |
| 代码生成/调试 | gpt-4o / o4-mini |
代码推理需要更强能力 |
| 数学/逻辑推理 | o3 / o4-mini |
推理模型在 STEM 领域有显著优势 |
| 长文档分析 | gpt-4.1 系列 |
1M 上下文可处理整本书或大型代码库 |
| 多模态理解 | gpt-4o |
唯一支持图像/音频原生输入的主流模型 |
| 成本敏感型 | gpt-4o-mini / gpt-4.1-nano |
极低成本的优质输出 |
OpenAI 遵循以下模型生命周期策略：
o3）先以 preview 版本发布，API 可能随时变更gpt-4o），API 版本锁定但会持续更新权重gpt-4o-2025-08-06）可获得固定版本建议：生产环境使用不带日期后缀的模型名，依赖 OpenAI 的后向兼容承诺；对结果一致性要求严格的场景使用带日期后缀的固定版本。
{
"model": "gpt-4o",
"messages": [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "请用三个要点解释量子计算。"}
],
"temperature": 0.7,
"max_tokens": 4096,
"top_p": 0.9,
"frequency_penalty": 0.2,
"presence_penalty": 0.1,
"seed": 42,
"stream": false,
"stop": ["\n\n", "。"],
"user": "user-123",
"response_format": {"type": "text"}
}
| 参数 | 类型 | 范围 | 默认值 | 行为说明 |
|---|---|---|---|---|
temperature |
float | 0-2 | 1.0 | 控制输出随机性。0=贪心采样（总是选最高概率 token），2=极高随机性 |
top_p |
float | 0-1 | 1.0 | Nucleus Sampling：只考虑累积概率质量达到 top_p 的 token。0.1=只保留前 10% |
max_tokens |
int | 1-上限 | 动态 | 每次响应最多生成的 token 数，非字符数 |
frequency_penalty |
float | -2.0~2.0 | 0.0 | 基于 token 在已生成文本中出现频率的惩罚系数 |
presence_penalty |
float | -2.0~2.0 | 0.0 | 基于 token 是否已在文本中出现的惩罚系数 |
seed |
int | null/any | null | 设置种子可获得确定性输出（当前为 beta 功能） |
stop |
string[] | - | null | 遇到这些字符串时停止生成 |
stream |
bool | - | false | 是否使用 SSE 流式输出 |
user |
string | - | null | 终端用户标识，用于监控和滥用检测 |
temperature 和 top_p 都用于控制输出的多样性，但机制完全不同：
Temperature 的工作原理：
对于一个给定的 token 概率分布 ，应用 temperature 后的新分布为：
其中 为 temperature 值。具体行为：
| 值 | 效果 | 示例场景 |
|---|---|---|
| 0 | 完全确定性，总是选最高概率 token | 数学计算、代码生成 |
| 0.3 | 轻度随机，高概率 token 仍占绝对优势 | 事实性问答 |
| 0.7 | 适中随机性，创意和准确性平衡 | 通用对话 |
| 1.0 | 自然随机，原始概率分布 | 创意写作 |
| 1.5 | 高度随机，低概率 token 被大幅提升 | 头脑风暴 |
| 2.0 | 近乎均匀分布，所有 token 几乎等概率 | 纯随机生成 |
示例：假设模型预测下一个 token 的概率为 {"好": 0.6, "很": 0.2, "不错": 0.1, "超级": 0.05, "极": 0.05}：
Top-P 的工作原理：选择累积概率质量达到 的最小 token 集合，然后在这些 token 中重新归一化采样。例如 时：
推荐组合策略：
温度优先策略：固定 top_p=1.0，只调整 temperature ← 最常用
精准控制策略：固定 temperature=1.0，降低 top_p ← 需要更确定输出时
保守策略：temperature=0.3, top_p=0.5 ← 需要高度确定性的输出
OpenAI 官方建议： 只调整其中一个参数，不要同时大幅调整两者。同时低 temperature + 低 top_p 会导致输出过于受限。
max_tokens 控制每次生成的最大 token 数量，注意以下几点：
不同模型的 max_tokens 上限：
| 模型 | 默认 max_tokens | 最大可达 | 备注 |
|---|---|---|---|
| gpt-4o | 4096 | 16384 | 使用 max_completion_tokens 字段 |
| gpt-4o-mini | 4096 | 16384 | 同上 |
| o3 | 4096 | 100000 | 推理模型输出长度可以很大 |
| gpt-4.1 | 8192 | 1000000 | 配合超长上下文使用 |
这两个参数常被混淆，理解其核心区别至关重要：
| 维度 | frequency_penalty | presence_penalty |
|---|---|---|
| 计算依据 | token 出现的次数（频率） | token 是否出现过（有无） |
| 影响范围 | 高频重复 token 受更大惩罚 | 所有已出现 token 受同等惩罚 |
| 典型值 | 0.1-0.5 | 0.1-0.5 |
| 负值用法 | 鼓励重复（用于诗歌/歌词） | 鼓励固定话题 |
工作原理示意（假设助手已输出 "这个系统非常好用"）：
已生成 token: "这个", "系统", "非常", "好用"
frequency_penalty=0.5:
"系统" 的惩罚 = 0.5 × 出现次数(1) → 轻度抑制
"非常" 的惩罚 = 0.5 × 出现次数(1) → 轻度抑制
presence_penalty=0.5:
"系统" 的惩罚 = 0.5（因子为 1）→ 抑制
"非常" 的惩罚 = 0.5（因子为 1）→ 同程度抑制
如果 "系统" 出现了 3 次：
frequency_penalty=0.5 → 惩罚 0.5 × 3 = 1.5（强抑制）
presence_penalty=0.5 → 惩罚 0.5 × 1 = 0.5（不变，只看有无）
实际应用场景：
| 场景 | 推荐参数 | 原因 |
|---|---|---|
| 去除非结构化的重复 | frequency_penalty=0.3 |
精准打击重复模式 |
| 鼓励话题多样性 | presence_penalty=0.5 |
让模型聊更多不同的东西 |
| 写诗/歌词 | frequency_penalty=-0.2 |
鼓励韵律和重复 |
| 代码生成 | 两者都为 0 | 代码不需要随机性 |
| 客服回复 | frequency_penalty=0.1 |
轻微去重，保持自然 |
OpenAI Chat API 的核心是 messages 数组，支持多种消息角色：
| 角色 | 用途 | 说明 |
|---|---|---|
system |
系统指令 | 设置助手的行为、语气、约束 |
user |
用户输入 | 用户的提问或指令 |
assistant |
助手回复 | 模型的回答（历史对话） |
tool |
工具结果 | function calling 调用返回的结果 |
有效的 system 消息结构：
{
"role": "system",
"content": "你是一位资深软件架构师。请遵循以下规则：\n1. 回答必须基于事实，不确定时明确说明\n2. 优先推荐成熟稳定的技术方案\n3. 代码示例使用 Python 或 TypeScript\n4. 不主动提供安全漏洞利用方法"
}
关键原则：
messages = [
{"role": "system", "content": "你是一位有耐心的数学老师。"},
{"role": "user", "content": "什么是导数？"},
{"role": "assistant", "content": "导数描述函数在某一点的瞬时变化率。用极限定义：$f'(x) = \\lim_{h \\to 0} \\frac{f(x+h) - f(x)}{h}$"},
{"role": "user", "content": "能不能给个具体例子？"},
# 模型会基于完整对话历史生成响应
]
OpenAI 使用 Server-Sent Events (SSE) 协议进行流式输出：
data: {"id":"chatcmpl-9a8b7c6d","object":"chat.completion.chunk","created":1715000000,"model":"gpt-4o-2024-08-06","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}
data: {"id":"chatcmpl-9a8b7c6d","object":"chat.completion.chunk","created":1715000000,"model":"gpt-4o-2024-08-06","choices":[{"index":0,"delta":{"content":"量子"},"finish_reason":null}]}
data: {"id":"chatcmpl-9a8b7c6d","object":"chat.completion.chunk","created":1715000000,"model":"gpt-4o-2024-08-06","choices":[{"index":0,"delta":{"content":"计算"},"finish_reason":null}]}
data: [DONE]
from openai import OpenAI
client = OpenAI()
stream = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "写一个关于 AI 的短诗"}],
stream=True
)
collected = []
for chunk in stream:
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
collected.append(content)
print(content, end="", flush=True)
print(f"\n\n完整响应: {''.join(collected)}")
当 stream=true 与 response_format: {"type": "json_object"} 结合使用时：
chunk.choices[0].delta.content 拼接后统一解析stream = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "北京的天气怎么样？"}],
tools=[weather_tool],
stream=True
)
tool_calls = {}
for chunk in stream:
delta = chunk.choices[0].delta
if delta.tool_calls:
for tc in delta.tool_calls:
idx = tc.index
if idx not in tool_calls:
tool_calls[idx] = {"id": tc.id, "function": {"name": "", "arguments": ""}}
if tc.function:
if tc.function.name:
tool_calls[idx]["function"]["name"] += tc.function.name
if tc.function.arguments:
tool_calls[idx]["function"]["arguments"] += tc.function.arguments
{
"tools": [
{
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": "搜索内部知识库获取相关信息",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"filters": {
"type": "object",
"properties": {
"category": {"type": "string", "enum": ["tech", "business", "legal"]},
"date_from": {"type": "string", "format": "date"}
}
},
"max_results": {
"type": "integer",
"description": "返回结果数量上限",
"default": 5
}
},
"required": ["query"]
}
}
}
],
"tool_choice": "auto"
}
| 值 | 行为 | 适用场景 |
|---|---|---|
"auto" |
模型自主决定是否调用工具 | 通用场景，模型自行判断 |
"none" |
禁止调用工具 | 纯文本对话 |
"required" |
强制调用某个工具（任意一个） | 需要确保工具被使用 |
{"type": "function", "function": {"name": "..."}} |
强制调用指定工具 | 路由/分类场景 |
串行调用：模型先调一个工具，得到结果后再调下一个。
用户：帮我查纽约天气，然后推荐餐厅
↓
模型：[call get_weather(location="纽约")]
↓
结果返回
↓
模型：[call search_restaurants(city="纽约", weather="晴天")]
并行调用：模型一次请求调用多个工具（OpenAI 原生支持）：
{
"choices": [{
"finish_reason": "tool_calls",
"message": {
"tool_calls": [
{"id": "call_1", "function": {"name": "get_weather", "arguments": "{\"location\": \"纽约\"}"}},
{"id": "call_2", "function": {"name": "get_weather", "arguments": "{\"location\": \"东京\"}"}},
{"id": "call_3", "function": {"name": "get_weather", "arguments": "{\"location\": \"伦敦\"}"}}
]
}
}]
}
复杂的工具使用场景通常采用 ReAct (Reasoning + Acting) 模式：
def react_loop(messages, tools, max_iterations=5):
"""ReAct: 模型交替推理和行动，直到得到最终答案"""
client = OpenAI()
for i in range(max_iterations):
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
msg = response.choices[0].message
if msg.tool_calls:
# 添加模型的决定
messages.append(msg)
for tc in msg.tool_calls:
# 执行工具
result = execute_tool(tc.function.name, json.loads(tc.function.arguments))
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": json.dumps(result)
})
else:
# 模型给出最终答案
return msg.content
return "Reached max iterations"
{
"response_format": {"type": "json_object"}
}
重要约束：
Structured Outputs 使用 JSON Schema 严格约束输出格式，保证：
{
"response_format": {
"type": "json_schema",
"json_schema": {
"name": "math_reasoning",
"strict": true,
"schema": {
"type": "object",
"properties": {
"reasoning_steps": {
"type": "array",
"items": {
"type": "object",
"properties": {
"step_number": {"type": "integer"},
"description": {"type": "string"},
"intermediate_result": {"type": "string"}
},
"required": ["step_number", "description", "intermediate_result"],
"additionalProperties": false
}
},
"final_answer": {"type": "string"},
"confidence": {
"type": "number",
"minimum": 0,
"maximum": 1
}
},
"required": ["reasoning_steps", "final_answer", "confidence"],
"additionalProperties": false
}
}
}
}
| 场景 | JSON Mode | Structured Outputs | 普通模式 |
|---|---|---|---|
| 简单键值对 | ✅ 够用 | ✅ 推荐 | ❌ 不可靠 |
| 复杂嵌套对象 | ❌ 不支持 | ✅ 完美 | ❌ 不可靠 |
| 多候选集提取 | ❌ 不可靠 | ✅ 推荐 | ❌ 不可靠 |
| 任意格式文本 | ❌ 不可用 | ❌ 不可用 | ✅ 推荐 |
GPT-4o 支持通过两种方式传入图像：
方式一：URL 引用
{
"type": "image_url",
"image_url": {
"url": "https://example.com/diagram.png",
"detail": "high"
}
}
方式二：Base64 编码
import base64
with open("screenshot.png", "rb") as f:
b64 = base64.b64encode(f.read()).decode("utf-8")
content = [
{"type": "text", "text": "这个截图里有什么错误？"},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
]
| detail 值 | 行为 | Token 消耗 |
|---|---|---|
"high" |
详细分析，先 2048×2048 缩放后平铺 | ~765 tokens |
"low" |
快速分析，直接 512×512 低分辨率 | ~85 tokens |
"auto" |
模型自动选择（默认） | 动态 |
实际差异对比：对于一张 1920×1080 的截图：
low：~85 tokens，约 $0.0001（GPT-4o-mini）high：~765 tokens，约 $0.0019（GPT-4o-mini）high 的准确率通常比 low 高 5-15%{
"role": "user",
"content": [
{"type": "text", "text": "比较这两个 UI 设计的差异，列出 5 个具体的改进建议。"},
{"type": "image_url", "image_url": {"url": "https://example.com/design_v1.png"}},
{"type": "image_url", "image_url": {"url": "https://example.com/design_v2.png"}}
]
}
GPT-4o 支持同一请求中传入最多 10 张图像，模型会自动关联分析。
GPT-4o 的实时 API (Realtime API) 支持原生音频输入输出：
会话模式：WebSocket → 双向流式音频对话
非流式模式：POST /v1/audio/transcriptions → 文字转录
目前音频功能的使用场景：
| 层级 | 条件 | RPM | TPM (GPT-4o) | RPD | 并发 |
|---|---|---|---|---|---|
| Free | 注册即得 | 3 | 150K | 200 | 1 |
| Tier 1 | $5 已消费 | 500 | 1M | 10K | 3 |
| Tier 2 | $50 已消费 | 5,000 | 2M | 10K | 5 |
| Tier 3 | $250 已消费 | 5,000 | 4M | 10K | 5 |
| Tier 4 | $1,000 已消费 | 10,000 | 8M | 10K | 10 |
| Tier 5 | $10,000 已消费 | 10,000 | 16M | 10K | 10 |
关键解读：
当收到 HTTP 429 (Rate Limit Exceeded) 时，推荐实现指数退避重试（Exponential Backoff）：
import time
import random
def retry_with_backoff(func, max_retries=3, base_delay=1.0):
for attempt in range(max_retries):
try:
return func()
except Exception as e:
if hasattr(e, 'status_code') and e.status_code == 429:
delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
print(f"Rate limited, retrying in {delay:.1f}s...")
time.sleep(delay)
else:
raise
raise Exception("Max retries exceeded")
高级策略：使用令牌桶算法监控剩余配额：
class RateLimiter:
def __init__(self, rpm=5000, tpm=2000000):
self.rpm = rpm
self.tpm = tpm
self.tokens = rpm
self.last_refill = time.time()
def wait_if_needed(self, estimated_tokens=1000):
now = time.time()
elapsed = now - self.last_refill
self.tokens = min(self.rpm, self.tokens + elapsed * self.rpm / 60)
self.last_refill = now
if self.tokens < 1:
wait = (1 - self.tokens) * 60 / self.rpm
time.sleep(wait)
self.tokens -= 1
| 模型 | 输入 | 缓存命中输入 | 输出 | 批量 API 输入 | 批量 API 输出 |
|---|---|---|---|---|---|
| gpt-4o | $2.50 $1.25 | $10.00 $1.25 | $5.00 | ||
| gpt-4o-mini | $0.15 $0.075 | $0.60 $0.075 | $0.30 | ||
| o3 | $10.00 - $40.00 | $5.00 $20.00 | |||
| o4-mini | $1.10 - $4.40 | $0.55 $2.20 | |||
| gpt-4.1 | $2.00 $1.00 | $8.00 $1.00 | $4.00 |
OpenAI 从 2024 年起支持自动 Prompt 缓存，无需任何代码修改。当 System 消息或对话前缀在连续请求中重复时自动触发。
缓存命中条件：
成本节省计算示例：
假设每天处理 10,000 个请求，每个请求包含：
Batch API 提供异步批量处理，价格减半：
# 1. 准备批处理文件（JSONL 格式）
# 每一行是一个独立的 completion 请求
batch_content = """{"custom_id": "request-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o", "messages": [{"role": "user", "content": "Hello"}]}}
{"custom_id": "request-2", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-4o", "messages": [{"role": "user", "content": "World"}]}}"""
# 2. 上传文件
with open("batch_input.jsonl", "w") as f:
f.write(batch_content)
client.files.create(
file=open("batch_input.jsonl", "rb"),
purpose="batch"
)
# 3. 创建批处理作业
batch = client.batches.create(
input_file_id=file.id,
endpoint="/v1/chat/completions",
completion_window="24h"
)
# 4. 轮询直到完成
while batch.status not in ("completed", "failed"):
time.sleep(60)
batch = client.batches.retrieve(batch.id)
# 5. 下载结果
result = client.files.content(batch.output_file_id)
Batch API 使用场景：
| 场景 | 推荐方式 | 原因 |
|---|---|---|
| 用户实时请求 | 同步/流式 | 需要即时响应 |
| 离线数据处理 | Batch | 成本减半，24h内完成 |
| 大规模数据标注 | Batch | 可并行处理百万级请求 |
| 模型评估 | Batch | 无需实时性 |
def estimate_cost(model: str, input_tokens: int, output_tokens: int,
batch: bool = False, cache_hit_pct: float = 0) -> float:
pricing = {
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"o3": {"input": 10.00, "output": 40.00},
"o4-mini": {"input": 1.10, "output": 4.40},
"gpt-4.1": {"input": 2.00, "output": 8.00},
}
p = pricing.get(model)
if not p:
return 0
# 缓存折扣
cached_input = input_tokens * cache_hit_pct * 0.5 # 缓存打5折
uncached_input = input_tokens * (1 - cache_hit_pct)
effective_input_cost = (cached_input + uncached_input) * p["input"]
output_cost = output_tokens * p["output"]
# Batch API 5折
if batch:
return (effective_input_cost + output_cost) / 2 / 1_000_000
return (effective_input_cost + output_cost) / 1_000_000
| 特性 | OpenAI | Anthropic Claude | Google Gemini | DeepSeek |
|---|---|---|---|---|
| API 格式 | Chat Completions（标准） | Messages API（兼容） | 专有格式 | 完全兼容 OpenAI |
| temperature | 0-2 | 0-1 | 0-2 | 0-2 |
| system role | 原生支持 | 原生支持 | 独立参数 | 同 OpenAI |
| function calling | 并行调用 | 工具调用 | 函数声明 | 兼容 |
| 多模态 | 图像/音频 | 图像/PDF | 图像/音频/视频 | 图像 |
| 流式格式 | SSE | SSE | SSE | SSE |
| 最大上下文 | 1M (GPT-4.1) | 200K | 1M | 128K |
| JSON 模式 | Structured Outputs | Tool Use | 原生 | 兼容 |
| 缓存机制 | 自动缓存 | 显式缓存 | 显式缓存 | 无 |
从 OpenAI 迁移到兼容 API 时需注意：
"auto"，部分厂商默认 "none"content 数组，部分厂商用独立参数{
"model": "gpt-4o",
"messages": [{"role": "user", "content": "今天的天气怎么样？"}],
"logprobs": true,
"top_logprobs": 3
}
返回每个 token 位置的概率分布，用于：
对于 gpt-4o 系列，可以通过这个参数突破默认的 4096 上限：
{
"model": "gpt-4o",
"max_completion_tokens": 16384
}
配合 128K 上下文，可以产出非常长的内容（如整篇文章或长篇代码）。
OpenAI 对所有 API 请求实施多层审核：
最佳实践：在客户端实现自定义内容过滤作为补充层：
def safe_chat_completion(client, messages, **kwargs):
# 输入检查
for msg in messages:
content = msg.get("content", "")
if isinstance(content, str) and contains_unsafe_content(content):
return {"safety_blocked": True, "reason": "Input filtered"}
response = client.chat.completions.create(
messages=messages,
**kwargs
)
# 输出检查
content = response.choices[0].message.content
if not content or is_empty_content(content):
return {"safety_blocked": True, "reason": "Output was empty"}
return response
| 状态码 | 含义 | 处理策略 |
|---|---|---|
| 200 | 成功 | 正常处理 |
| 400 | 请求错误（参数无效） | 检查请求格式和必填字段 |
| 401 | 认证失败 | 检查 API Key 是否有效 |
| 403 | 权限不足 | 检查 API Key 的模型访问权限 |
| 404 | 模型不存在 | 检查模型名是否正确 |
| 429 | 速率限制 | 指数退避重试 |
| 500 | 服务器内部错误 | 重试 1-2 次后降级 |
| 503 | 服务暂时不可用 | 延迟后重试 |
import time
from openai import APIError, RateLimitError, APITimeoutError
class OpenAIErrorHandler:
MAX_RETRIES = 3
RETRYABLE_CODES = {429, 500, 502, 503}
@staticmethod
def handle_api_error(func):
def wrapper(*args, **kwargs):
last_error = None
for attempt in range(OpenAIErrorHandler.MAX_RETRIES):
try:
return func(*args, **kwargs)
except RateLimitError as e:
wait = min(2 ** attempt * 2, 60)
print(f"Rate limited, waiting {wait}s (attempt {attempt+1})")
time.sleep(wait)
last_error = e
except APITimeoutError as e:
wait = 2 ** attempt
time.sleep(wait)
last_error = e
except APIError as e:
if e.status_code in OpenAIErrorHandler.RETRYABLE_CODES:
wait = 2 ** attempt
time.sleep(wait)
last_error = e
else:
# 非可重试错误直接抛出
raise
raise last_error
return wrapper
from openai import OpenAI
client = OpenAI()
def chat_with_history(user_input, history=None):
if history is None:
history = [
{"role": "system", "content": "你是一位精通技术的助手，回答简洁准确。"}
]
history.append({"role": "user", "content": user_input})
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=history,
temperature=0.7,
max_tokens=1024,
)
assistant_msg = response.choices[0].message.content
history.append({"role": "assistant", "content": assistant_msg})
return assistant_msg, history
# 使用
msg, history = chat_with_history("什么是 Rust 的所有权系统？")
print(msg)
msg2, history = chat_with_history("它和垃圾回收有什么区别？", history)
print(msg2)
from openai import OpenAI
from concurrent.futures import ThreadPoolExecutor
client = OpenAI()
def process_batch(prompts, model="gpt-4o-mini", max_workers=5):
def process_single(prompt):
try:
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
max_tokens=500
)
return prompt, response.choices[0].message.content, None
except Exception as e:
return prompt, None, str(e)
with ThreadPoolExecutor(max_workers=max_workers) as executor:
results = list(executor.map(process_single, prompts))
return results
# 同时处理 20 个分类任务
prompts = [f"将这段文本分类：{text}" for text in texts]
results = process_batch(prompts, max_workers=5)
参见：Anthropic API 行为与参数详解 | DeepSeek API 详解 | Google Gemini API 详解 | 智谱 AI API 详解