Как ускорить обработку тысяч AI-запросов: пакетный режим OVHcloud AI Endpoints

Вот перевод HTML-контента с сохранением всех тегов:

Представьте, что вам нужно классифицировать 20 000 запросов в службу поддержки до завтрашнего утра или перевести полный каталог продуктов, не отправляя каждый запрос вручную. Такая рабочая нагрузка может быстро стать медленной, повторяющейся и сложной в управлении.

Пакетный режим разработан именно для таких сценариев.

Что такое пакетный режим?

При работе с LLM вы часто отправляете запросы по одному через синхронные конечные точки, такие как /v1/chat/completions или /v1/responses. Это хорошо подходит для сценариев реального времени, но что делать, если нужно обработать сотни или тысячи подсказок? Отправка их по отдельности медленна, и вы ограничены лимитами скорости.

Пакетный режим решает эту проблему. Вместо отправки запросов по одному вы загружаете файл, содержащий все ваши запросы, отправляете пакетное задание и асинхронно получаете результаты в течение максимум 24 часов. И вот вишенка на торте: пакетный режим на 50% дешевле, чем синхронные запросы. Поскольку платформа может более эффективно планировать вашу рабочую нагрузку, вы получаете значительное снижение затрат.

Это идеально подходит для:

• 📊 Задач массовой классификации или обобщения
• 🌍 Крупномасштабных переводов
• 📝 Генерации описаний для каталога продуктов
• 🧪 Оценки выходных данных модели на тестовом наборе данных

ℹ️ Пакетный API совместим с форматом пакетного API OpenAI, поэтому вы можете использовать официальный SDK OpenAI для взаимодействия с ним.

Когда не следует использовать пакетный режим!

Пакетный режим предназначен для больших рабочих нагрузок, которые не требуют немедленного ответа. Таким образом, он не подходит для сценариев реального времени, таких как чат-боты, поддержка клиентов в реальном времени, интерактивные ассистенты или приложения, где пользователи ожидают ответа в течение нескольких секунд. Для таких сценариев более подходят синхронные конечные точки. Используйте пакетный режим, когда ваши запросы могут быть обработаны асинхронно и получены позже.

ℹ️ Пакетный API в настоящее время находится в бета-версии. Дополнительную информацию о бета-версии можно найти на специальной странице.

Предварительные требования для использования пакетного режима

Перед началом работы вам понадобятся:

• Ключ API AI Endpoints
• Установленный Python 3.10+
• Пакет Python openai

⚠️ Вы можете сгенерировать свой ключ API в консоли AI Endpoints.

Установите зависимость:

pip install openai

Настройте переменные окружения:

export OVH_AI_ENDPOINTS_ACCESS_TOKEN='your_api_key'
export OVH_AI_ENDPOINTS_BASE_URL='https://oai.endpoints.kepler.ai.cloud.ovh.net/v1'

Шаг 1: Подготовка входного файла

Входной файл использует формат JSON Lines (.jsonl). Каждая строка представляет собой отдельный запрос с уникальным custom_id, который позволяет сопоставлять результаты с исходными запросами.

Вот пример requests.jsonl:

{"custom_id": "request-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-oss-20b", "messages": [{"role": "user", "content": "Кратко опишите сюжет Гамлета в двух предложениях."}]}}
{"custom_id": "request-2", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-oss-20b", "messages": [{"role": "user", "content": "Переведите 'Доброе утро' на французский, испанский и немецкий языки."}]}}

Ключевые моменты:

• Каждый custom_id должен быть уникальным в пределах пакета
• Поле model должно ссылаться на модель, доступную в каталоге AI Endpoints
• Поле url указывает, какую конечную точку вызывать

Шаг 2: Загрузка файла и создание пакета

Вот полный код Python, который обрабатывает весь рабочий процесс: загрузка, создание, опрос и скачивание:

import os
import time

from openai import OpenAI

# Загрузка переменных окружения
_OVH_AI_ENDPOINTS_ACCESS_TOKEN = os.environ["OVH_AI_ENDPOINTS_ACCESS_TOKEN"]
_OVH_AI_ENDPOINTS_BASE_URL = os.environ["OVH_AI_ENDPOINTS_BASE_URL"]

# Инициализация клиента, совместимого с OpenAI, для работы с OVHcloud AI Endpoints
client = OpenAI(
    base_url=_OVH_AI_ENDPOINTS_BASE_URL,
    api_key=_OVH_AI_ENDPOINTS_ACCESS_TOKEN,
)

# 1. Загрузка входного JSONL-файла с параметром purpose="batch"
print("📤 Загрузка входного файла...")
batch_input_file = client.files.create(
    file=open("requests.jsonl", "rb"),
    purpose="batch",
)
print(f"✅ Загруженный файл id: {batch_input_file.id}")

# 2. Создание пакета со ссылкой на загруженный файл
print("🚀 Создание пакета...")
batch = client.batches.create(
    input_file_id=batch_input_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
    metadata={"description": "Пример пакетного режима - OVHcloud AI Endpoints"},
)
print(f"✅ Пакет создан: {batch.id} (статус: {batch.status})")

# 3. Опрос статуса пакета до достижения конечного состояния
print("⏳ Опрос статуса пакета...")
while True:
    current = client.batches.retrieve(batch.id)
    print(f"   статус={current.status} количество={current.request_counts}")
    if current.status in ("completed", "failed", "expired", "cancelled"):
        break
    time.sleep(30)

# 4. Скачивание результатов (и ошибок, если есть)
final = client.batches.retrieve(batch.id)

if final.output_file_id:
    print("📥 Скачивание results.jsonl...")
    output = client.files.content(final.output_file_id)
    with open("results.jsonl", "wb") as f:
        f.write(output.read())
    print("✅ Результаты записаны в results.jsonl")

if final.error_file_id:
    print("🐛 Скачивание errors.jsonl...")
    errors = client.files.content(final.error_file_id)
    with open("errors.jsonl", "wb") as f:
        f.write(errors.read())
    print("🐛 Ошибки записаны в errors.jsonl")

print(f"🏁 Итоговый статус пакета: {final.status}")

Давайте разберем ключевые шаги:

Загрузка входного файла

batch_input_file = client.files.create(
    file=open("requests.jsonl", "rb"),
    purpose="batch",
)

Параметр purpose=”batch” указывает API, что этот файл будет использоваться как входные данные для пакета.

Создание пакета

batch = client.batches.create(
    input_file_id=batch_input_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
)

Параметр completion_window=”24h” означает, что пакет будет остановлен через 24 часа, если не завершится раньше.

Опрос статуса пакета

while True:
    current = client.batches.retrieve(batch.id)
    print(f"   статус={current.status} количество={current.request_counts}")
    if current.status in ("completed", "failed", "expired", "cancelled"):
        break
    time.sleep(30)

Вызов client.batches.retrieve(batch.id) возвращает текущее состояние пакета. Поле request_counts показывает разбивку: сколько запросов выполнено, сколько завершилось ошибкой или еще в обработке. Это полезно для мониторинга больших пакетов.

Возможные конечные состояния:

• completed: все запросы успешно обработаны
• failed: пакет столкнулся с фатальной ошибкой
• expired: пакет превысил длительность completion_window
• cancelled: пакет был вручную отменен через API

Мы опрашиваем каждые 30 секунд, но вы можете настроить этот интервал в зависимости от вашего сценария использования. Для очень больших пакетов более длительный интервал (например, 60–120 секунд) более подходит.

Загрузка результатов

final = client.batches.retrieve(batch.id)

if final.output_file_id:
    output = client.files.content(final.output_file_id)
    with open("results.jsonl", "wb") as f:
        f.write(output.read())

Когда пакет завершен, поле output_file_id содержит ID файла результатов. Вы загружаете его с помощью client.files.content(), который возвращает необработанное содержимое файла.

Загрузка ошибок (если есть)

if final.error_file_id:
    errors = client.files.content(final.error_file_id)
    with open("errors.jsonl", "wb") as f:
        f.write(errors.read())

Если некоторые запросы в вашем пакете не удались (например, неверное имя модели, некорректный ввод, превышение лимита токенов), их детали будут доступны в отдельном файле ошибок. Поле error_file_id будет равно None, если все запросы выполнены успешно. Каждая строка в errors.jsonl содержит custom_id неудачного запроса и детали ошибки, что позволяет легко определить и повторить только неудачные запросы.

Шаг 3: Чтение результатов

Выходной файл (results.jsonl) содержит один JSON-объект на строку. Каждый объект включает:

• custom_id, соответствующий вашему исходному запросу
• Полное тело ответа (тот же формат, что и у синхронных ответов /v1/chat/completions)

Вот как выглядит результат:

{
  "id": "964e007472a557240221910ba143bb03",
  "custom_id": "request-1",
  "response": {
    "status_code": 200,
    "body": {
      "id": "chatcmpl-9879ebff777795a3",
      "choices": [
        {
          "index": 0,
          "message": {
            "role": "assistant",
            "content": "Hamlet, the Prince of Denmark, is driven to madness and vengeance after learning that his father was murdered by his uncle Claudius..."
          },
          "finish_reason": "stop"
        }
      ],
      "model": "gpt-oss-20b",
      "usage": {
        "prompt_tokens": 78,
        "completion_tokens": 297,
        "total_tokens": 375
      }
    }
  },
  "error": null
}

Если некоторые запросы не удались, файл errors.jsonl будет содержать детали о том, что пошло не так для каждого неудачного запроса.

Другие доступные примеры

В руководстве по пакетному режиму AI Endpoints также содержатся примеры на:

• JavaScript: с использованием OpenAI Node.js SDK
• Чистых HTTP-запросах: с использованием curl без какого-либо фреймворка, если вы предпочитаете языконезависимый подход

Эти примеры показывают, что вы можете использовать Batch API из любого языка или инструмента, который может делать HTTP-запросы, поскольку он следует стандартному формату API, совместимому с OpenAI.

Заключение

Пакетный режим — это мощная функция, когда вам нужно обрабатывать большие объемы повторяющихся, нечувствительных ко времени запросов на вывод, не беспокоясь об ограничениях скорости или проблемах с тайм-аутом. Загрузите файл, отправьте пакет и вернитесь позже за результатами — это так просто.

Совместимый с OpenAI API упрощает интеграцию в существующие рабочие процессы, а с примерами на Python, JavaScript и чистых HTTP-запросах вы можете использовать любой подход, который лучше всего подходит для вашего стека.

У вас есть специальный канал Discord (#ai-endpoints) на нашем сервере Discord, увидимся там!

Для получения дополнительной информации о AI Endpoints найдите наши предыдущие сообщения в блоге.