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-4o",
    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-4o",
    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()

多轮对话

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

def 发问(问题):
    消息历史.append({"role": "user", "content": 问题})
    resp = client.chat.completions.create(model="gpt-4o", messages=消息历史)
    回答 = resp.choices[0].message.content
    消息历史.append({"role": "assistant", "content": 回答})
    return 回答

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

JSON 结构化输出

import json

response = client.chat.completions.create(
    model="gpt-4o",
    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

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

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

response = client.chat.completions.create(
    model="gpt-4o",
    messages=消息,
    tools=工具列表,
    tool_choice="auto",
)

if response.choices[0].finish_reason == "tool_calls":
    工具调用 = response.choices[0].message.tool_calls[0]
    参数 = json.loads(工具调用.function.arguments)
    print(f"要调用：{工具调用.function.name}")
    print(f"参数：{参数}")

嵌入向量（Embeddings）

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

向量列表 = [item.embedding for item in response.data]
print(f"向量维度：{len(向量列表[0])}")  # 1536

计算相似度：

import numpy as np

def 余弦相似度(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

相似度 = 余弦相似度(向量列表[0], 向量列表[1])
print(f"相似度：{相似度:.4f}")

异步并发（批量处理）

import asyncio
from openai import AsyncOpenAI

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

async def 批量总结(文章列表):
    任务列表 = [
        async_client.chat.completions.create(
            model="gpt-4o-mini",  # 批量任务用便宜的模型
            messages=[{"role": "user", "content": f"用一句话总结：{文章}"}],
            max_tokens=100,
        )
        for 文章 in 文章列表
    ]
    结果 = await asyncio.gather(*任务列表)
    return [r.choices[0].message.content for r in 结果]

# 并发处理 20 篇文章
摘要列表 = asyncio.run(批量总结(文章列表))

错误处理

from openai import AuthenticationError, RateLimitError, APIStatusError
import time

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

模型选择指南

场景	推荐模型	原因
文本分类、简单摘要	gpt-4o-mini	最便宜
代码生成、通用问答	gpt-4o	性价比最高
复杂推理、架构设计	gpt-5 或 o3	能力最强
实时对话应用	gpt-4o-mini	速度快，成本低

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