OpenAI API 最佳实践指南

# 固定 API 版本，避免意外行为变更
client = OpenAI(
    api_key="sk-...",
    default_headers={"OpenAI-Beta": "assistants=v2"}
)

# 推荐的初始化模式
from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY"),
    max_retries=3,
    timeout=30.0
)

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "获取指定城市的天气信息",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名称"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["city"]
        }
    }
}]

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
    tools=tools,
    tool_choice="auto"  # "auto" | "required" | {"type": "function", "name": "xxx"}
)

from pydantic import BaseModel

class CalendarEvent(BaseModel):
    name: str
    date: str
    participants: list[str]

response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[{"role": "user", "content": "帮我提取会议信息：下周三下午3点产品评审会"}],
    response_format=CalendarEvent
)

event = response.choices[0].message.parsed
# CalendarEvent(name='产品评审会', date='...', participants=[...])

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "写一首关于春天的诗"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content is not None:
        print(chunk.choices[0].delta.content, end="")

# 1. 准备 JSONL 文件
# {"custom_id": "req-1", "method": "POST", "url": "/v1/chat/completions", "body": {...}}

# 2. 上传并创建 batch
batch = client.batches.create(
    input_file_id=file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h"
)

# 3. 轮询状态
batch = client.batches.retrieve(batch.id)
# batch.status: validating → in_progress → completed

# 4. 下载结果
result = client.files.content(batch.output_file_id)

import tenacity

@tenacity.retry(
    wait=tenacity.wait_exponential(multiplier=1, min=1, max=60),
    stop=tenacity.stop_after_attempt(5),
    retry=tenacity.retry_if_exception_type((
        openai.RateLimitError,
        openai.APITimeoutError,
        openai.InternalServerError
    ))
)
def call_openai(messages):
    return client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )

def call_with_fallback(messages):
    models = ["gpt-4o", "gpt-4o-mini", "gpt-3.5-turbo"]
    for model in models:
        try:
            return client.chat.completions.create(model=model, messages=messages)
        except openai.RateLimitError:
            continue
    raise Exception("所有模型均不可用")

response = client.chat.completions.create(
    model="o3-mini",
    reasoning_effort="medium",  # low | medium | high
    messages=[
        {"role": "user", "content": "分析这个算法的时间复杂度并给出证明:\n\n[算法代码]"}
    ]
)

# 推理 token 消耗在 response.usage 中单独列出
print(f"推理 token: {response.usage.completion_tokens_details.reasoning_tokens}")

# 1. 创建 Assistant
assistant = client.beta.assistants.create(
    name="代码审查助手",
    instructions="你是一个代码审查专家，帮助用户审查代码并提供改进建议。",
    model="gpt-4o",
    tools=[
        {"type": "code_interpreter"},
        {"type": "file_search"},
        {"type": "function", "function": {...}}
    ]
)

# 2. 创建 Thread（会话）
thread = client.beta.threads.create()

# 3. 添加消息
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="请审查这段 Python 代码：[代码内容]"
)

# 4. 运行 Assistant
run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id
)

# 5. 轮询等待完成
while run.status in ["queued", "in_progress"]:
    time.sleep(1)
    run = client.beta.threads.runs.retrieve(
        thread_id=thread.id,
        run_id=run.id
    )

# 6. 获取回复
messages = client.beta.threads.messages.list(thread_id=thread.id)

# 生成嵌入向量
response = client.embeddings.create(
    model="text-embedding-3-small",
    input=["第一条文本", "第二条文本"],
    encoding_format="float"  # "float" | "base64"
)

# 获取向量
embedding = response.data[0].embedding  # list[float], 1536 维

response = client.embeddings.create(
    model="text-embedding-3-large",
    input="示例文本",
    dimensions=1024  # 从 3072 缩减到 1024 维
)

import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 对查询和文档分别生成嵌入，计算余弦相似度
query_emb = get_embedding("搜索查询")
doc_embs = [get_embedding(doc) for doc in documents]
scores = [cosine_similarity(query_emb, doc_emb) for doc_emb in doc_embs]
top_k = np.argsort(scores)[-5:][::-1]  # 取前 5 个最相关文档

from sklearn.cluster import KMeans

embeddings = [get_embedding(text) for text in texts]
kmeans = KMeans(n_clusters=5).fit(embeddings)
labels = kmeans.labels_

# 使用 tiktoken 精确计算 token 用量
import tiktoken

encoding = tiktoken.encoding_for_model("gpt-4o")
token_count = len(encoding.encode(your_text))