Python OpenAI
openai — мощная библиотека Python для взаимодействия с набором моделей и сервисов OpenAI.
openai обёртывает все вызовы RESTful API, позволяя разработчикам легко интегрировать мощные возможности ИИ — такие как обработка естественного языка, генерация изображений и распознавание речи — в свои приложения на Python.
Ключевые особенности:
- Генерация текста: используйте такие модели, как GPT-4 или GPT-5, для создания статей, кода, резюме, диалогов и многого другого.
- Генерация изображений: создавайте изображения по текстовым описаниям с помощью моделей DALL-E.
- Эмбеддинги: преобразование текста в векторные представления, обычно используемые для таких задач, как семантический поиск, классификация текста и кластеризация.
- Речь в текст: используйте модель Whisper для транскрипции аудиофайлов в текст.
- Тонкая настройка: обучите более целенаправленную модель, предоставив собственный набор данных.
- Assistants API: создавайте сложные приложения, которые понимают контекст, вызывают инструменты и выполняют долгосрочные взаимодействия.
Репозиторий openai с открытым исходным кодом:https://github.com/openai/openai-python

Как использовать?
Требования:
- Версия Python: 3.9 или выше.
- Зависимости: httpx (по умолчанию), aiohttp (опционально), websockets (требуется для Realtime API).
Сначала необходимо установить библиотеку openai с помощью pip:
pip install openai или pip3 install openai
Затем перейдите на официальный сайт OpenAIhttps://platform.openai.com/ , чтобы зарегистрировать аккаунт и сгенерировать API-ключ на странице API keys.
Проверьте установленную версию:
import openai print(openai.__version__) # 输出当前 SDK 版本
Пример
from openai import OpenAI
client = OpenAI(
# Это значение по умолчанию, и его можно опустить
api_key="API-ключ, который вы запросили",
)
response = client.responses.create(
model="gpt-4o",
instructions="Ты — ассистент по программированию, который говорит как пират.",
input="Как проверить, является ли объект Python экземпляром класса?",
)
print(response.output_text)
Описание параметра:
| Параметр | Обязательный | Type | Описание |
|---|---|---|---|
api_key |
Да | str |
Ключ OpenAI, который вы запросили |
model |
Да | str |
Указывает используемую модель OpenAI, определяя её возможности, уровень рассуждений и стоимость |
instructions |
нет | str |
Системный промпт, определяющий идентичность модели, нормы поведения и стиль выражения, имеющий приоритет надinput |
input |
Да | str / list |
Содержимое пользовательского ввода, описывающее конкретный вопрос или задачу |
Сторонние модели
Доступ к OpenAI из Китая всё ещё несколько затруднён. Многие местные провайдеры также поддерживают совместимость с OpenAI, например DeepSeek, Qwen и GLM.
DeepSeek
API DeepSeek полностью совместим с форматом API OpenAI. Вам нужно лишь изменить несколько настроек конфигурации, чтобы напрямую использовать OpenAI SDK или совместимые инструменты для доступа к API DeepSeek.
| Параметр | Значение / Описание |
|---|---|
| base_url | Обязательно, фиксированное значение:https://api.deepseek.com(Вы также можете использоватьhttps://api.deepseek.com/v1, только для совместимости с OpenAI; v1 не связан с версиями модели) |
| api_key | Обязательно. Сначала вам нужно подать заявку на получение выделенного API-ключа на официальном сайте DeepSeek (URL для подачи заявки:https://platform.deepseek.com/) |
| model | Обязательно. Модель, которую нужно указать:
|
Пример
from openai import OpenAI
client = OpenAI(
api_key=os.environ.get('DEEPSEEK_API_KEY'), # Рекомендуется настраивать через переменные окружения; их также можно жёстко прописать в коде (не рекомендуется)
base_url="https://api.deepseek.com")
response = client.chat.completions.create(
model="deepseek-v4-flash",
messages=[
{"role": "system", "content": "Вы — полезный помощник"},
{"role": "user", "content": "Привет"},
],
stream=False # stream=False без потоковой передачи (возвращается сразу), stream=True потоковая передача (возвращается в реальном времени)
)
print(response.choices[0].message.content)
Использование
Далее мы используем OpenAI SDK для доступа к моделям Qwen в сервисе Bailian.
Пример вызова без потоковой передачи
Пример
import os
def get_response():
client = OpenAI(
api_key="sk-xxx", # Используйте ваш ключ API Alibaba Cloud Bailian
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", # Укажите base_url для DashScope SDK
)
completion = client.chat.completions.create(
model="qwen-plus", # Здесь мы используем qwen-plus в качестве примера; при необходимости можно изменить название модели. Список моделей: https://help.aliyun.com/zh/model-studio/getting-started/models
messages=[{'role': 'system', 'content': 'Вы — полезный помощник.'},
{'role': 'user', 'content': 'Кто вы?'}]
)
# данные JSON
#print(completion.model_dump_json())
print(completion.choices[0].message.content)
if __name__ == '__main__':
get_response()
Выполнение кода даёт следующий результат:
I am Qwen, an ultra-large-scale language model independently developed by the Tongyi Laboratory under Alibaba Group. I can help you answer questions and create text, such as writing stories, official documents, emails, scripts, logical reasoning, programming, and more. I can also express opinions and play games. If you have any questions or need help, feel free to tell me at any time!
Пример потокового вызова
Пример
def get_response():
client = OpenAI(
api_key="sk-xxx",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
completion = client.chat.completions.create(
model="qwen-plus",
messages=[
{'role': 'system', 'content': 'Вы — полезный помощник.'},
{'role': 'user', 'content': 'Кто вы?'}
],
stream=True,
stream_options={"include_usage": True}
)
for chunk in completion:
# chunk может не содержать choices или delta
if hasattr(chunk, "choices") and len(chunk.choices) > 0:
choice = chunk.choices[0]
if hasattr(choice, "delta") and hasattr(choice.delta, "content"):
print(choice.delta.content, end='', flush=True)
if __name__ == '__main__':
get_response()
Выполнение кода даёт следующий результат:
I am Qwen, an ultra-large-scale language model independently developed by the Tongyi Laboratory under Alibaba Group. I can help you answer questions and create text, such as writing stories, official documents, emails, scripts, logical reasoning, programming, and more. I can also express opinions and play games. If you have any questions or need help, feel free to tell me at any time!
Ключевые возможности
Возможность работы с изображениями
Поддерживает ввод изображений (URL или в кодировке Base64) для мультимодального взаимодействия.
Ввод URL изображения:
Пример
img_url = "https://upload.wikimedia.org/wikipedia/commons/thumb/d/d5/2023_06_08_Raccoon1.jpg/1599px-2023_06_08_Raccoon1.jpg"
response = client.responses.create(
model="gpt-5.2",
input=[
{
"role": "user",
"content": [
{"type": "input_text", "text": prompt},
{"type": "input_image", "image_url": img_url},
],
}
],
)
print(response.output_text)
Ввод изображения в кодировке Base64:
Пример
from openai import OpenAI
client = OpenAI()
prompt = "Что на этом изображении?"
# Прочитать локальное изображение и закодировать его в Base64
with open("path/to/image.png", "rb") as image_file:
b64_image = base64.b64encode(image_file.read()).decode("utf-8")
response = client.responses.create(
model="gpt-5.2",
input=[
{
"role": "user",
"content": [
{"type": "input_text", "text": prompt},
{"type": "input_image", "image_url": f"data:image/png;base64,{b64_image}"},
],
}
],
)
Асинхронное использование (Async usage)
Замените OpenAI на AsyncOpenAI и вызовите API с помощью await:
Базовый пример async:
Пример
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI(
api_key=os.environ.get("OPENAI_API_KEY"),
)
async def main() -> None:
response = await client.responses.create(
model="gpt-5.2",
input="Объясни дизестеблишментарианизм умному пятилетнему ребёнку."
)
print(response.output_text)
asyncio.run(main())
Для оптимизации асинхронного сценария (бэкенд aiohttp) установите расширенную версию:
pip install openai[aiohttp]
Оптимизация бэкенда aiohttp:
Пример
import asyncio
from openai import DefaultAioHttpClient, AsyncOpenAI
async def main() -> None:
# Контекстный менеджер гарантирует освобождение ресурсов
async with AsyncOpenAI(
api_key=os.environ.get("OPENAI_API_KEY"),
http_client=DefaultAioHttpClient(), # Включить бэкенд aiohttp
) as client:
chat_completion = await client.chat.completions.create(
messages=[{"role": "user", "content": "Say this is a test"}],
model="gpt-5.2",
)
asyncio.run(main())
Потоковые ответы
На основе Server Side Events (SSE) для получения ответов в реальном времени; синхронные и асинхронные интерфейсы согласованы.
Пример синхронной потоковой передачи:
Пример
client = OpenAI()
stream = client.responses.create(
model="gpt-5.2",
input="Write a one-sentence bedtime story about a unicorn.",
stream=True, # Включить потоковый вывод
)
for event in stream:
print(event) # Выводить ответ по сегментам
Пример асинхронной потоковой передачи:
Пример
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def main():
stream = await client.responses.create(
model="gpt-5.2",
input="Write a one-sentence bedtime story about a unicorn.",
stream=True,
)
async for event in stream:
print(event)
asyncio.run(main())
API в реальном времени
Мультимодальный диалог с низкой задержкой (текст / аудио), реализованный через WebSocket, требует библиотеки websockets.
Базовый пример текста:
Пример
from openai import AsyncOpenAI
async def main():
client = AsyncOpenAI()
# Установить соединение в реальном времени
async with client.realtime.connect(model="gpt-realtime") as connection:
# Обновить конфигурацию сессии (вывод только текста)
await connection.session.update(
session={"type": "realtime", "output_modalities": ["text"]}
)
# Отправить сообщение пользователя
await connection.conversation.item.create(
item={
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "Поздоровайся!"}],
}
)
# Запустить ответ модели
await connection.response.create()
# Прослушивать события в реальном времени
async for event in connection:
if event.type == "response.output_text.delta":
print(event.delta, flush=True, end="") # Выводить текстовые фрагменты в реальном времени
elif event.type == "response.output_text.done":
print() # Добавить перевод строки при завершении ответа
elif event.type == "response.done":
break # Прекратить прослушивание
asyncio.run(main())
Обработка ошибок Realtime API:
Пример
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def main():
async with client.realtime.connect(model="gpt-realtime") as connection:
async for event in connection:
if event.type == 'error':
# Перехватывать и обрабатывать события ошибок (соединение не будет разорвано)
print(f"Тип ошибки: {event.error.type}")
print(f"Код ошибки: {event.error.code}")
print(f"Сообщение об ошибке: {event.error.message}")
asyncio.run(main())
Справочное руководство
Следующая таблица содержит:
- Инициализация ядра: для синхронного использования,
OpenAIдля асинхронного использования,AsyncOpenAIдля использования с Azure,AzureOpenAIрекомендуется настраивать API Key через переменные среды. - Генерация текста: предпочитайте новую версию,
responses.create()для классических сценариев используйтеchat.completions.create()потоковый вывод требует настройкиstream=True。 - Расширенные возможности: поддержка мультимодальности (ввод изображений), Realtime API, загрузка файлов/тонкая настройка; при обработке ошибок необходимо перехватывать
APIErrorподклассы, а тайм-аут/повторы можно гибко настраивать.
Установка и импорт
Установите официальный SDK:
pip install --upgrade openai
Импортируйте основные классы:
from openai import OpenAI, AsyncOpenAI, AzureOpenAI
Создание клиента
Создайте стандартный клиент OpenAI:
from openai import OpenAI client = OpenAI(api_key="YOUR_API_KEY")
Создайте клиент через переменные окружения:
import os from openai import OpenAI os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY" client = OpenAI()
Установите тайм-аут запроса и количество повторных попыток:
client = OpenAI(
timeout=30,
max_retries=2
)
Генерация текста (Responses API)
Самый базовый запрос на генерацию текста:
response = client.responses.create(
model="gpt-5.2",
input="用一句话解释什么是 Python"
)
print(response.output_text)
Ограничьте поведение модели с помощью инструкций:
response = client.responses.create(
model="gpt-5.2",
instructions="你是一个严谨的 Python 教程作者",
input="解释什么是列表推导式"
)
print(response.output_text)
Многоходовой диалог
Используйте массив input для построения контекстного диалога:
response = client.responses.create(
model="gpt-5.2",
input=[
{"role": "user", "content": "Python 是什么?"},
{"role": "assistant", "content": "Python 是一门高级编程语言。"},
{"role": "user", "content": "它适合做什么?"}
]
)
print(response.output_text)
Структурированный вывод (JSON Schema)
Попросите модель вернуть данные JSON, соответствующие Schema:
response = client.responses.create(
model="gpt-5.2",
response_format={
"type": "json_schema",
"json_schema": {
"name": "tech_summary",
"schema": {
"type": "object",
"properties": {
"title": {"type": "string"},
"summary": {"type": "string"}
},
"required": ["title", "summary"]
}
}
},
input="介绍 asyncio"
)
print(response.output_parsed)
Вызов инструментов (Function Calling)
Определите инструменты и позвольте модели автоматически решать, вызывать ли их:
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取城市天气",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string"}
},
"required": ["city"]
}
}
}
]
response = client.responses.create(
model="gpt-5.2",
tools=tools,
input="北京今天天气怎么样?"
)
print(response.output)
Потоковый вывод (Streaming)
Синхронный потоковый вывод:
with client.responses.stream(
model="gpt-5.2",
input="写一段 Python 示例代码"
) as stream:
for event in stream:
if event.type == "response.output_text.delta":
print(event.delta, end="")
Асинхронный потоковый вывод:
stream = await client.responses.create(
model="gpt-5.2",
input="解释什么是生成式 AI",
stream=True
)
async for event in stream:
print(event)
Асинхронные вызовы (AsyncOpenAI)
Использование асинхронного клиента в asyncio:
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def main():
response = await client.responses.create(
model="gpt-5.2",
input="解释什么是事件循环"
)
print(response.output_text)
asyncio.run(main())
Realtime API (Realtime)
Использование WebSocket для низколатентных разговоров в реальном времени:
async with client.realtime.connect(model="gpt-realtime") as conn:
await conn.session.update(
session={"type": "realtime", "output_modalities": ["text"]}
)
await conn.conversation.item.create(
item={
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "你好"}]
}
)
async for event in conn:
print(event)
Генерация изображений
Генерация изображения из текста:
image = client.images.generate(
model="gpt-image-1",
prompt="一只在写代码的猫",
size="1024x1024"
)
print(image.data[0].url)
Генерация эмбеддингов
Генерация векторных эмбеддингов текста:
embedding = client.embeddings.create(
model="text-embedding-3-small",
input="Hello world"
)
print(embedding.data[0].embedding)
Речь в текст
Преобразование аудиофайла в текст:
audio_file = open("speech.mp3", "rb")
transcript = client.audio.transcriptions.create(
model="whisper-1",
file=audio_file
)
print(transcript.text)
Текст в речь
Преобразовать текст в аудиофайл:
speech = client.audio.speech.create(
model="gpt-4o-mini-tts",
voice="alloy",
input="你好,欢迎使用 OpenAI"
)
with open("output.mp3", "wb") as f:
f.write(speech)
Управление файлами (Files API)
Загрузить файл:
from pathlib import Path
client.files.create(
file=Path("data.jsonl"),
purpose="fine-tune"
)
Список файлов:
files = client.files.list()
for f in files:
print(f.id, f.filename)
Удалить файл:
client.files.delete(file_id="file-xxx")
Тонкая настройка модели (Fine-tuning Jobs)
Создать задание тонкой настройки:
job = client.fine_tuning.jobs.create(
training_file="file-xxx",
model="gpt-3.5-turbo"
)
print(job.id, job.status)
Проверка Webhook
Проверить и разобрать запрос Webhook:
event = client.webhooks.unwrap(request_body, request_headers)
Проверить только подпись:
client.webhooks.verify_signature(request_body, request_headers)
Обработка ошибок
Перехватывать исключения, выбрасываемые SDK:
import openai
try:
client.responses.create(
model="gpt-5.2",
input="test"
)
except openai.RateLimitError as e:
print("触发速率限制:", e)
except openai.APIConnectionError as e:
print("连接失败:", e)
Получить Request ID неудачного запроса:
try:
client.responses.create(model="gpt-5.2", input="test")
except openai.APIStatusError as exc:
print(exc.request_id)
Совместимость с Azure OpenAI
Инициализируйте клиент Azure OpenAI:
from openai import AzureOpenAI
client = AzureOpenAI(
api_version="2023-07-01-preview",
azure_endpoint="https://xxx.openai.azure.com"
)
Версии и журнал изменений
Проверьте версию SDK:
import openai print(openai.__version__)
Включите отладочное логирование SDK:
export OPENAI_LOG=debug
Другие расширенияДополнительные справочные материалы по API:https://github.com/openai/openai-python/blob/main/api.md