OpenAI API Python 教程

本文提供使用 Python 调用 OpenAI API 的完整代码，覆盖从基础入门到生产环境的各种用法。示例通过 APIMaster 接入可用。

安装

pip install openai  # Python 3.8+

初始化

from openai import OpenAI

client = OpenAI(
    api_key="你的 APIMaster Key",
    base_url="https://apimaster.ai/v1",
)

推荐用环境变量管理 Key：

export OPENAI_API_KEY="你的Key"
export OPENAI_BASE_URL="https://apimaster.ai/v1"

from openai import OpenAI
client = OpenAI()  # 自动读取环境变量

基础对话

response = client.chat.completions.create(
    model="gpt-5.4",
    messages=[
        {"role": "system", "content": "你是一位 Python 专家，回答简洁，给出可运行的代码。"},
        {"role": "user", "content": "如何用 Python 合并两个字典？"},
    ],
)
print(response.choices[0].message.content)
print(f"消耗 Token：{response.usage.total_tokens}")

流式输出

stream = client.chat.completions.create(
    model="gpt-5.4",
    messages=[{"role": "user", "content": "用 Python 实现一个完整的链表类，包含增删查操作。"}],
    stream=True,
)

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

多轮对话

history = [
    {"role": "system", "content": "你是一个经验丰富的软件架构师。"}
]

def ask(question):
    history.append({"role": "user", "content": question})
    resp = client.chat.completions.create(model="gpt-5.4", messages=history)
    answer = resp.choices[0].message.content
    history.append({"role": "assistant", "content": answer})
    return answer

print(ask("微服务和单体架构各有什么优缺点？"))
print(ask("什么情况下应该拆分微服务？"))

JSON 结构化输出

import json

response = client.chat.completions.create(
    model="gpt-5.4",
    messages=[
        {
            "role": "user",
            "content": "从以下文本提取联系方式，返回 JSON：'请联系张明，邮箱 zhangming@example.com，电话 138-0000-0000，就职于北京科技有限公司'",
        }
    ],
    response_format={"type": "json_object"},
)

data = json.loads(response.choices[0].message.content)
print(data)
# {"姓名": "张明", "邮箱": "zhangming@example.com", "电话": "138-0000-0000", "公司": "北京科技有限公司"}

Function Calling（工具调用）

import json

tools = [
    {
        "type": "function",
        "function": {
            "name": "查询股价",
            "description": "查询指定股票的当前价格",
            "parameters": {
                "type": "object",
                "properties": {
                    "股票代码": {"type": "string", "description": "如 600519、000858"},
                    "市场": {"type": "string", "enum": ["沪市", "深市", "港股"]},
                },
                "required": ["股票代码"],
            },
        },
    }
]

messages = [{"role": "user", "content": "茅台现在股价是多少？"}]

response = client.chat.completions.create(
    model="gpt-5.4",
    messages=messages,
    tools=tools,
    tool_choice="auto",
)

if response.choices[0].finish_reason == "tool_calls":
    tool_call = response.choices[0].message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    print(f"Tool: {tool_call.function.name}")
    print(f"Args: {args}")

嵌入向量（Embeddings）

response = client.embeddings.create(
    model="text-embedding-3-small",
    input=["Python 是一门编程语言", "机器学习需要大量数据"],
)

vectors = [item.embedding for item in response.data]
print(f"Dimensions: {len(vectors[0])}")  # 1536

计算相似度：

import numpy as np

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

similarity = cosine_sim(vectors[0], vectors[1])
print(f"Similarity: {similarity:.4f}")

异步并发（批量处理）

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    api_key="你的Key",
    base_url="https://apimaster.ai/v1",
)

async def batch_summarize(articles):
    tasks = [
        async_client.chat.completions.create(
            model="gpt-4o-mini",  # 批量任务用低成本模型
            messages=[{"role": "user", "content": f"用一句话总结：{article}"}],
            max_tokens=100,
        )
        for article in articles
    ]
    results = await asyncio.gather(*tasks)
    return [r.choices[0].message.content for r in results]

# 并发处理 20 篇文章
summaries = asyncio.run(batch_summarize(articles))

错误处理

from openai import AuthenticationError, RateLimitError, APIStatusError
import time

def safe_call(client, **kwargs):
    for attempt in range(3):
        try:
            return client.chat.completions.create(**kwargs)
        except AuthenticationError:
            raise  # Key 错误，不重试
        except RateLimitError:
            time.sleep(2 ** attempt)  # 指数退避
        except APIStatusError as e:
            if e.status_code >= 500:
                time.sleep(1)  # 服务器错误，重试
            else:
                raise
    raise RuntimeError("重试 3 次后仍然失败")

模型选择指南

场景	推荐模型	原因
文本分类、简单摘要	gpt-4o-mini	低成本
代码生成、通用问答	gpt-5.4	平衡成本与能力
复杂推理、架构设计	gpt-5.5 或 o3	高能力
实时对话应用	gpt-4o-mini 或 gpt-5.4	按质量和预算选择

获取 OpenAI API 接入 → · 查看价格 →