基于Python的OpenAI API调用实践：从环境搭建到流式响应

云川学术

354人浏览 · 2026-08-02 03:16:46

云川学术 · 2026-08-02 03:16:46 发布

本文介绍如何通过OpenAI官方API接口，在Python环境中调用大语言模型，包含同步调用、流式响应及错误处理等完整示例。

一、背景说明

OpenAI为开发者提供了标准化的API接口，允许将大语言模型能力集成到各类应用中。无论是搭建对话系统、实现文本自动化处理，还是构建内容分析工具，API调用都是基础环节。

本文基于OpenAI官方公开的API文档，梳理在Python环境下调用API的完整技术流程，供开发者参考。

说明：本文仅涉及API调用的技术实现层面，所有代码均基于官方接口规范编写，不涉及账户注册、计费方式等商业操作。

二、环境准备

2.1 Python版本要求

建议使用 Python 3.8 及以上版本。可通过以下命令查看当前环境：

python --version

2.2 安装官方SDK

OpenAI官方提供Python SDK，使用pip安装：

pip install openai

若需要异步请求支持，可额外安装：

pip install aiohttp

三、基础调用示例

3.1 同步请求

以下代码演示了最基础的API调用方式：

import openai

# 配置API密钥（生产环境建议从环境变量读取）
openai.api_key = "your-api-key-here"

response = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=[
        {"role": "system", "content": "你是一名Python编程助手"},
        {"role": "user", "content": "请用Python实现一个快速排序算法"}
    ],
    temperature=0.7,
    max_tokens=1024
)

print(response.choices[0].message.content)

3.2 参数说明

参数	类型	说明
`model`	string	指定使用的模型标识符
`messages`	list	对话消息列表，包含角色（system/user/assistant）与内容
`temperature`	float	控制生成内容的随机性，取值范围0~2，值越低输出越确定
`max_tokens`	int	限制模型输出的最大token长度

四、流式响应模式

对于需要实时展示生成内容的场景（如聊天界面），可使用流式响应逐字返回：

import openai

openai.api_key = "your-api-key-here"

stream = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=[
        {"role": "user", "content": "请写一篇150字左右的科技短文"}
    ],
    stream=True
)

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

流式模式下，API会分块返回生成内容，适合需要逐字显示的用户交互场景。

五、错误处理与重试机制

生产环境中，网络波动或接口限流可能导致请求失败，建议实现重试逻辑：

import time
import openai

def call_with_retry(messages, max_retries=3, base_delay=1):
    """带自动重试的API调用函数"""
    for attempt in range(max_retries):
        try:
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=messages,
                timeout=30
            )
            return response
        except openai.error.RateLimitError:
            if attempt < max_retries - 1:
                wait_time = base_delay * (attempt + 1)
                time.sleep(wait_time)
                continue
            raise
        except openai.error.APIConnectionError:
            if attempt < max_retries - 1:
                time.sleep(base_delay)
                continue
            raise
        except openai.error.AuthenticationError:
            # 密钥无效，不应重试
            raise

六、异步调用示例（高并发场景）

对于需要同时处理多个请求的应用，推荐使用异步方式：

import asyncio
import openai
from openai import AsyncOpenAI

async def async_call():
    client = AsyncOpenAI(api_key="your-api-key-here")
    
    response = await client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "user", "content": "解释一下什么是异步编程"}
        ]
    )
    print(response.choices[0].message.content)

asyncio.run(async_call())