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 版本

Пример

import os
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

Обязательно. Модель, которую нужно указать:

  • deepseek-v4-flash: Соответствуетрежиму без размышленийDeepSeek, с высокой скоростью ответа, подходит для обычных вопросов и ответов.
  • deepseek-v4-pro: Соответствуетрежиму размышленийDeepSeek, с более мощными возможностями рассуждения, подходит для решения сложных задач.

Пример

import os
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.

Пример вызова без потоковой передачи

Пример

from openai import OpenAI
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!

Пример потокового вызова

Пример

from openai import OpenAI

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 изображения:

Пример

prompt = "Что на этом изображении?"
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:

Пример

import 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 os
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 os
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) для получения ответов в реальном времени; синхронные и асинхронные интерфейсы согласованы.

Пример синхронной потоковой передачи:

Пример

from openai import OpenAI

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)  # Выводить ответ по сегментам

Пример асинхронной потоковой передачи:

Пример

import asyncio
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.

Базовый пример текста:

Пример

import asyncio
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:

Пример

import asyncio
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

Другие расширения