Пакетная обработка (Batch Processing) — это техника выполнения большого количества запросов к AI API параллельно и с управлением скоростью. Типичные сценарии: обработка 10 000 отзывов, генерация описаний для каталога товаров, массовая классификация. Без правильного батчинга такие задачи занимают часы вместо минут.
Проблема последовательных запросов
Наивный подход — отправлять запросы по одному:
# ПЛОХО: последовательно, ~3 сек на запрос
# 1000 запросов = ~50 минут
for item in items:
response = client.chat.completions.create(...)
results.append(response)Каждый запрос ждёт ответа 1-5 секунд. Для 1000 элементов это 30-80 минут. С параллельными запросами — 2-5 минут.
Параллельные запросы на Python (asyncio)
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI(
base_url="https://api.modelswitch.ru/v1",
api_key="msk_ваш_ключ"
)
async def process_item(item: str, semaphore: asyncio.Semaphore) -> dict:
"""Обработка одного элемента с ограничением параллельности."""
async with semaphore:
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "Классифицируй отзыв: positive, negative, neutral. Ответь одним словом."},
{"role": "user", "content": item}
],
temperature=0
)
return {"text": item, "label": response.choices[0].message.content.strip()}
async def batch_process(items: list[str], max_concurrent: int = 20) -> list[dict]:
"""Пакетная обработка с ограничением параллельных запросов."""
semaphore = asyncio.Semaphore(max_concurrent)
tasks = [process_item(item, semaphore) for item in items]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Фильтруем ошибки
successful = [r for r in results if isinstance(r, dict)]
errors = [r for r in results if isinstance(r, Exception)]
print(f"Успешно: {len(successful)}, Ошибки: {len(errors)}")
return successful
# Пример: классификация 500 отзывов
reviews = ["Отличный товар!", "Не понравилось качество", ...] * 100
results = asyncio.run(batch_process(reviews, max_concurrent=20))
for r in results[:5]:
print(f"[{r['label']}] {r['text'][:50]}")Управление Rate Limits
AI API имеют лимиты: 60-300 запросов в минуту (RPM) и 100K-1M токенов в минуту (TPM). Для управления используем token bucket:
import time
class RateLimiter:
def __init__(self, rpm: int = 60):
self.rpm = rpm
self.interval = 60.0 / rpm
self.last_request = 0.0
async def wait(self):
"""Ожидание до следующего разрешённого запроса."""
now = time.time()
elapsed = now - self.last_request
if elapsed < self.interval:
await asyncio.sleep(self.interval - elapsed)
self.last_request = time.time()
rate_limiter = RateLimiter(rpm=50) # 50 запросов в минуту
async def rate_limited_request(item: str) -> dict:
await rate_limiter.wait()
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": item}],
temperature=0
)
return {"text": item, "result": response.choices[0].message.content}Пакетная обработка на TypeScript
import OpenAI from "openai";
const client = new OpenAI({
baseURL: "https://api.modelswitch.ru/v1",
apiKey: "msk_ваш_ключ",
});
async function batchProcess<T>(
items: string[],
processor: (item: string) => Promise<T>,
concurrency = 20
): Promise<T[]> {
const results: T[] = [];
const queue = [...items];
async function worker() {
while (queue.length > 0) {
const item = queue.shift()!;
try {
const result = await processor(item);
results.push(result);
} catch (e) {
console.error(`Ошибка: ${e}`);
}
}
}
const workers = Array.from({ length: concurrency }, () => worker());
await Promise.all(workers);
return results;
}
// Использование: генерация описаний для 200 товаров
const products = ["Ноутбук ASUS VivoBook", "Клавиатура Logitech MX Keys", ...];
const descriptions = await batchProcess(
products,
async (product) => {
const response = await client.chat.completions.create({
model: "gpt-4o-mini",
messages: [
{ role: "system", content: "Напиши краткое SEO-описание товара (2-3 предложения)." },
{ role: "user", content: product },
],
});
return { product, description: response.choices[0].message.content };
},
15
);
console.log(`Обработано: ${descriptions.length} товаров`);Советы по оптимизации
- Выбор модели — для массовых задач используйте
gpt-4o-mini($0.15/1M input). Она в 15 раз дешевле GPT-4o. - Параллельность — начните с 10-20 параллельных запросов. Увеличивайте, если не получаете 429 ошибок.
- Прогресс-бар — для длинных задач используйте
tqdm(Python) для отслеживания прогресса. - Checkpoint — сохраняйте промежуточные результаты каждые N запросов. При сбое не нужно начинать заново.
- Стоимость — перед запуском 10 000 запросов протестируйте на 100 и оцените стоимость.
ModelSwitch не взимает дополнительную плату за пакетные запросы. Платите только за токены. Дашборд покажет расход в реальном времени.