Batch API: пакетная обработка запросов к LLM

import json
import time
from openai import OpenAI

client = OpenAI()


def prepare_batch_file(
    requests: list[dict],
    model: str = "gpt-4.1-mini",
) -> str:
    """Подготовка JSONL-файла для batch."""
    lines = []
    for i, req in enumerate(requests):
        lines.append(json.dumps({
            "custom_id": req.get("id", f"req-{i}"),
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": model,
                "max_tokens": req.get("max_tokens", 512),
                "messages": req["messages"],
            },
        }))
    return "\n".join(lines)


def run_openai_batch(
    requests: list[dict],
    model: str = "gpt-4.1-mini",
    description: str = "batch job",
) -> dict:
    """Полный цикл OpenAI Batch API."""
    # 1. Подготовить JSONL
    jsonl_content = prepare_batch_file(requests, model)

    # 2. Загрузить файл
    with open("/tmp/batch_input.jsonl", "w") as f:
        f.write(jsonl_content)

    input_file = client.files.create(
        file=open("/tmp/batch_input.jsonl", "rb"),
        purpose="batch",
    )
    print(f"File uploaded: {input_file.id}")

    # 3. Создать batch
    batch = client.batches.create(
        input_file_id=input_file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h",
        metadata={"description": description},
    )
    print(f"Batch created: {batch.id}")

    # 4. Поллить статус
    while batch.status not in ("completed", "failed", "expired", "cancelled"):
        time.sleep(30)
        batch = client.batches.retrieve(batch.id)
        counts = batch.request_counts
        print(
            f"Status: {batch.status}, "
            f"completed: {counts.completed}/{counts.total}, "
            f"failed: {counts.failed}"
        )

    if batch.status != "completed":
        raise RuntimeError(f"Batch failed: {batch.status}, errors: {batch.errors}")

    # 5. Скачать результаты
    output_file = client.files.content(batch.output_file_id)
    results = {}
    for line in output_file.text.strip().split("\n"):
        item = json.loads(line)
        custom_id = item["custom_id"]
        if item["response"]["status_code"] == 200:
            body = item["response"]["body"]
            results[custom_id] = {
                "content": body["choices"][0]["message"]["content"],
                "usage": body["usage"],
            }
        else:
            results[custom_id] = {"error": item["response"]["body"]}

    return results


# Пример: классификация 1000 тикетов
tickets = [
    "Не могу войти в аккаунт после смены пароля",
    "Хочу оформить возврат товара, заказ #12345",
    "Ваше приложение вылетает при открытии камеры",
    # ... ещё 997 тикетов
]

requests = [
    {
        "id": f"ticket-{i}",
        "messages": [
            {
                "role": "system",
                "content": (
                    "Classify support ticket into one of: "
                    "billing, technical, account, shipping, other. "
                    "Reply with JSON: {\"category\": \"...\", \"priority\": \"low|medium|high\"}"
                ),
            },
            {"role": "user", "content": ticket},
        ],
        "max_tokens": 64,
    }
    for i, ticket in enumerate(tickets)
]

results = run_openai_batch(requests, model="gpt-4.1-mini", description="ticket classification")
for custom_id, result in list(results.items())[:3]:
    print(f"{custom_id}: {result.get('content', result.get('error'))}")

import time
import anthropic

client = anthropic.Anthropic()


def run_anthropic_batch(
    requests: list[dict],
    model: str = "claude-haiku-3.5-20241022",
) -> dict:
    """Полный цикл Anthropic Message Batches."""
    # 1. Подготовить запросы
    batch_requests = []
    for i, req in enumerate(requests):
        batch_requests.append({
            "custom_id": req.get("id", f"req-{i}"),
            "params": {
                "model": model,
                "max_tokens": req.get("max_tokens", 512),
                "messages": req["messages"],
                "system": req.get("system", ""),
            },
        })

    # 2. Создать batch
    batch = client.messages.batches.create(requests=batch_requests)
    print(f"Batch created: {batch.id}, status: {batch.processing_status}")

    # 3. Поллить статус
    while batch.processing_status != "ended":
        time.sleep(30)
        batch = client.messages.batches.retrieve(batch.id)
        counts = batch.request_counts
        print(
            f"Status: {batch.processing_status}, "
            f"succeeded: {counts.succeeded}, "
            f"errored: {counts.errored}, "
            f"processing: {counts.processing}"
        )

    # 4. Получить результаты
    results = {}
    for entry in client.messages.batches.results(batch.id):
        custom_id = entry.custom_id
        if entry.result.type == "succeeded":
            msg = entry.result.message
            results[custom_id] = {
                "content": msg.content[0].text,
                "usage": {
                    "input_tokens": msg.usage.input_tokens,
                    "output_tokens": msg.usage.output_tokens,
                },
            }
        else:
            results[custom_id] = {"error": str(entry.result)}

    return results


# Пример: eval-прогон с LLM-as-judge
eval_samples = [
    {
        "id": f"eval-{i}",
        "system": (
            "You are an evaluator. Score the AI response on a scale 1-5 "
            "for accuracy, helpfulness, and safety. "
            "Reply with JSON: {\"accuracy\": N, \"helpfulness\": N, \"safety\": N}"
        ),
        "messages": [
            {
                "role": "user",
                "content": f"Question: {sample['question']}\n\nAI Response: {sample['response']}",
            },
        ],
        "max_tokens": 128,
    }
    for i, sample in enumerate(eval_dataset)
]

results = run_anthropic_batch(eval_samples, model="claude-sonnet-4-20250514")

from typing import Iterator


def chunk_requests(
    requests: list[dict],
    chunk_size: int = 10_000,
) -> Iterator[list[dict]]:
    """Разбивка на чанки для надёжности."""
    for i in range(0, len(requests), chunk_size):
        yield requests[i : i + chunk_size]


def run_chunked_batch(
    requests: list[dict],
    model: str = "gpt-4.1-mini",
    chunk_size: int = 10_000,
) -> dict:
    """Batch-обработка с чанкингом и retry."""
    all_results = {}
    failed_chunks = []

    for chunk_idx, chunk in enumerate(chunk_requests(requests, chunk_size)):
        print(f"Processing chunk {chunk_idx} ({len(chunk)} requests)...")
        try:
            results = run_openai_batch(
                chunk,
                model=model,
                description=f"chunk-{chunk_idx}",
            )
            all_results.update(results)
        except RuntimeError as e:
            print(f"Chunk {chunk_idx} failed: {e}")
            failed_chunks.append((chunk_idx, chunk))

    # Retry неудачных чанков
    for chunk_idx, chunk in failed_chunks:
        print(f"Retrying chunk {chunk_idx}...")
        try:
            results = run_openai_batch(chunk, model=model)
            all_results.update(results)
        except RuntimeError:
            print(f"Chunk {chunk_idx} failed again, skipping")

    return all_results


# 100 000 запросов, чанками по 10 000
all_results = run_chunked_batch(large_dataset, chunk_size=10_000)

# .github/workflows/eval-batch.yml
name: LLM Eval (Batch)

on:
  pull_request:
    paths: ["prompts/**", "src/llm/**"]

jobs:
  eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Run batch eval
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python scripts/submit_eval_batch.py \
            --input eval/test_cases.jsonl \
            --model gpt-4.1-mini \
            --output /tmp/batch_id.txt

      - name: Wait for results
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python scripts/poll_batch.py \
            --batch-id $(cat /tmp/batch_id.txt) \
            --timeout 3600 \
            --output eval/results.json

      - name: Check quality threshold
        run: |
          python scripts/check_eval_threshold.py \
            --results eval/results.json \
            --min-score 0.85

# scripts/nightly_batch.py
import schedule
import time
from datetime import datetime


def nightly_classification():
    """Ночная классификация накопленных тикетов."""
    # 1. Забрать необработанные записи из БД
    unprocessed = db.fetch_unprocessed_tickets(limit=50_000)
    if not unprocessed:
        print("No tickets to process")
        return

    print(f"Processing {len(unprocessed)} tickets at {datetime.now()}")

    # 2. Подготовить batch-запросы
    requests = [
        {
            "id": f"ticket-{t['id']}",
            "messages": [
                {"role": "system", "content": CLASSIFICATION_PROMPT},
                {"role": "user", "content": t["text"]},
            ],
            "max_tokens": 64,
        }
        for t in unprocessed
    ]

    # 3. Отправить batch (50% скидка)
    results = run_openai_batch(requests, model="gpt-4.1-mini")

    # 4. Записать результаты в БД
    for ticket_id, result in results.items():
        db_id = int(ticket_id.split("-")[1])
        if "content" in result:
            db.update_ticket_classification(db_id, result["content"])
        else:
            db.mark_ticket_error(db_id, str(result["error"]))

    print(f"Done. Processed: {len(results)}")


# Запуск каждый день в 02:00
schedule.every().day.at("02:00").do(nightly_classification)

while True:
    schedule.run_pending()
    time.sleep(60)

from dataclasses import dataclass, field
from datetime import datetime
import logging

logger = logging.getLogger(__name__)


@dataclass
class BatchMonitor:
    """Мониторинг batch-задач с алертами."""
    batches: dict = field(default_factory=dict)
    stale_threshold_hours: float = 12.0

    def track(self, batch_id: str, total_requests: int):
        self.batches[batch_id] = {
            "submitted_at": datetime.now(),
            "total": total_requests,
            "status": "submitted",
        }

    def update(self, batch_id: str, status: str, completed: int, failed: int):
        if batch_id not in self.batches:
            return
        entry = self.batches[batch_id]
        entry["status"] = status
        entry["completed"] = completed
        entry["failed"] = failed
        entry["last_update"] = datetime.now()

        # Алерт на зависший batch
        elapsed = (datetime.now() - entry["submitted_at"]).total_seconds() / 3600
        if elapsed > self.stale_threshold_hours and status not in ("completed", "failed"):
            logger.warning(
                f"Batch {batch_id} stale: {elapsed:.1f}h elapsed, "
                f"status={status}, completed={completed}/{entry['total']}"
            )

    def report(self) -> dict:
        """Сводка по всем batch-задачам."""
        total_cost_saved = 0
        for batch_id, entry in self.batches.items():
            if entry["status"] == "completed":
                # Примерная экономия: 50% от стоимости realtime
                total_cost_saved += entry["total"] * 0.005  # ~$0.005 на запрос
        return {
            "active_batches": sum(
                1 for b in self.batches.values()
                if b["status"] not in ("completed", "failed")
            ),
            "total_batches": len(self.batches),
            "estimated_savings": round(total_cost_saved, 2),
        }