LLM batch API cost reduction: 50% off when your workload can wait

• Nightly document processing pipelines (contract analysis, PDF summarization, report generation on the previous day’s data)
• Bulk classification and tagging (categorizing a product catalog, intent-labeling a support ticket archive, tagging historical conversation logs)
• Embedding generation for a corpus (initial embedding of a document library before launching an RAG system, re-embedding after a model upgrade)
• Test suite evaluation and LLM-as-judge scoring (running evals on a test set of 5,000 prompts nightly to detect regression)
• Content moderation for uploaded assets (user-uploaded images, documents, or audio transcripts that are queued rather than blocking the upload response)
• Weekly report generation (summarizing user activity, generating personalized digests, creating analytics narratives)
• Fine-tuning data preparation (generating synthetic training examples, reformatting existing datasets)
• Retrospective analytics (running sentiment analysis or topic modeling on last month’s customer feedback)

import anthropic
import json
import time
import runguard

client = anthropic.Anthropic()
rg = runguard.init(api_key="rg_live_...")

def submit_classification_batch(
    tickets: list[dict],
    system_prompt: str,
) -> str:
    """Submit a batch and return the batch ID."""
    requests = [
        {
            "custom_id": f"ticket-{t['id']}",
            "params": {
                "model": "claude-3-5-sonnet-20241022",
                "max_tokens": 100,
                "system": system_prompt,
                "messages": [
                    {"role": "user", "content": t["text"]}
                ]
            }
        }
        for t in tickets
    ]

    batch = client.messages.batches.create(requests=requests)
    rg.metric("batch_submitted", value=len(requests), tags={"batch_id": batch.id})
    print(f"Submitted batch {batch.id} with {len(requests)} requests")
    return batch.id


def wait_for_batch(batch_id: str, poll_interval_s: int = 60) -> anthropic.MessageBatch:
    """Poll until the batch is complete. Returns the completed batch object."""
    while True:
        batch = client.messages.batches.retrieve(batch_id)
        if batch.processing_status == "ended":
            return batch
        print(f"Batch {batch_id} still processing... sleeping {poll_interval_s}s")
        time.sleep(poll_interval_s)


def process_batch_results(batch_id: str) -> dict[str, str]:
    """
    Stream results from the completed batch.
    Returns a dict mapping custom_id to classification label.
    Logs per-request errors to RunGuard for alerting.
    """
    results: dict[str, str] = {}
    errors: list[dict] = []

    for result in client.messages.batches.results(batch_id):
        if result.result.type == "succeeded":
            # Extract the text content from the message
            label = result.result.message.content[0].text.strip()
            results[result.custom_id] = label
        elif result.result.type == "errored":
            errors.append({
                "custom_id": result.custom_id,
                "error_type": result.result.error.type,
                "error_message": result.result.error.message,
            })

    if errors:
        rg.alert(
            "batch_request_errors",
            detail=f"{len(errors)} requests failed in batch {batch_id}",
            severity="warning",
            metadata={"sample_errors": errors[:5]},
        )

    rg.metric(
        "batch_completion",
        value=len(results),
        tags={
            "batch_id": batch_id,
            "success_count": len(results),
            "error_count": len(errors),
        }
    )

    return results


# Putting it together for a nightly classification job
def nightly_classification_job(tickets: list[dict]) -> None:
    SYSTEM = (
        "Classify the support ticket into exactly one category: "
        "billing, technical, feature_request, other. "
        "Reply with only the category name."
    )

    with rg.budget(daily_usd=15.00, alert_at=0.80):
        batch_id = submit_classification_batch(tickets, SYSTEM)
        batch = wait_for_batch(batch_id)
        labels = process_batch_results(batch_id)

    print(f"Classified {len(labels)} tickets. "
          f"Estimated cost: ${batch.request_counts.succeeded * 500 / 1_000_000 * 1.50:.4f}")

import OpenAI from "openai";
import * as fs from "fs";
import * as readline from "readline";
import RunGuard from "@runguard/sdk";

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });
const rg = new RunGuard({ apiKey: process.env.RUNGUARD_API_KEY! });

interface TicketRecord {
  id: string;
  text: string;
}

interface BatchResult {
  ticketId: string;
  label: string;
}

async function submitOpenAIBatch(
  tickets: TicketRecord[],
  systemPrompt: string
): Promise<string> {
  // Build the JSONL content in memory
  const lines = tickets.map((t) =>
    JSON.stringify({
      custom_id: `ticket-${t.id}`,
      method: "POST",
      url: "/v1/chat/completions",
      body: {
        model: "gpt-4o",
        max_tokens: 100,
        messages: [
          { role: "system", content: systemPrompt },
          { role: "user", content: t.text },
        ],
      },
    })
  );
  const jsonlContent = lines.join("\n");

  // Upload the JSONL file
  const file = await openai.files.create({
    file: new File([jsonlContent], "batch_input.jsonl", {
      type: "application/jsonl",
    }),
    purpose: "batch",
  });

  // Submit the batch
  const batch = await openai.batches.create({
    input_file_id: file.id,
    endpoint: "/v1/chat/completions",
    completion_window: "24h",
  });

  await rg.metric("openai_batch_submitted", {
    value: tickets.length,
    tags: { batchId: batch.id },
  });

  return batch.id;
}

async function waitForOpenAIBatch(
  batchId: string,
  pollIntervalMs = 60_000
): Promise<OpenAI.Batches.Batch> {
  while (true) {
    const batch = await openai.batches.retrieve(batchId);
    if (batch.status === "completed" || batch.status === "failed" ||
        batch.status === "expired" || batch.status === "cancelled") {
      return batch;
    }
    console.log(`Batch ${batchId} status: ${batch.status}. Waiting...`);
    await new Promise((resolve) => setTimeout(resolve, pollIntervalMs));
  }
}

async function processOpenAIBatchResults(
  batch: OpenAI.Batches.Batch
): Promise<BatchResult[]> {
  if (batch.status !== "completed" || !batch.output_file_id) {
    throw new Error(`Batch ${batch.id} did not complete successfully: ${batch.status}`);
  }

  const fileContent = await openai.files.content(batch.output_file_id);
  const text = await fileContent.text();

  const results: BatchResult[] = [];
  const errors: string[] = [];

  for (const line of text.split("\n").filter(Boolean)) {
    const record = JSON.parse(line);
    if (record.error) {
      errors.push(`${record.custom_id}: ${record.error.message}`);
      continue;
    }
    const label =
      record.response?.body?.choices?.[0]?.message?.content?.trim() ?? "unknown";
    results.push({
      ticketId: record.custom_id.replace("ticket-", ""),
      label,
    });
  }

  if (errors.length) {
    await rg.alert("openai_batch_errors", {
      detail: `${errors.length} requests failed in batch ${batch.id}`,
      severity: "warning",
    });
  }

  await rg.metric("openai_batch_completed", {
    value: results.length,
    tags: { batchId: batch.id, errorCount: errors.length },
  });

  return results;
}

// Example nightly job
async function nightlyClassificationJob(tickets: TicketRecord[]): Promise<void> {
  const SYSTEM =
    "Classify the support ticket into exactly one category: " +
    "billing, technical, feature_request, other. Reply with only the category name.";

  // RunGuard budget guard prevents the batch submission itself if daily budget
  // would be exceeded by the estimated cost of this batch
  await rg.withBudget({ dailyUsd: 15.0, alertAt: 0.8 }, async () => {
    const batchId = await submitOpenAIBatch(tickets, SYSTEM);
    const batch = await waitForOpenAIBatch(batchId);
    const results = await processOpenAIBatchResults(batch);
    console.log(`Classified ${results.length} tickets via batch API`);
  });
}

import runguard
from runguard.batch import BatchRouter, BatchPolicy
from datetime import datetime, timedelta

rg = runguard.init(api_key="rg_live_...")

# Configure routing policy
policy = BatchPolicy(
    # Route to batch if the request is tagged batch_eligible
    tag_rules={"batch_eligible": True},
    # Route to batch if deadline is >30 minutes away
    deadline_threshold_minutes=30,
    # Route to batch if current UTC hour is in the off-peak window
    off_peak_hours=range(22, 6),  # 22:00–05:59 UTC
    # Flush accumulated batch requests every 15 minutes
    flush_interval_seconds=900,
    # Use Anthropic batch endpoint (or "openai" for OpenAI)
    provider="anthropic",
)

router = BatchRouter(rg, policy)

# Wrap your LLM client with the router
@router.wrap
async def call_llm(
    messages: list[dict],
    model: str = "claude-3-5-sonnet-20241022",
    batch_eligible: bool = False,
    deadline: datetime | None = None,
    **kwargs,
) -> str:
    """
    This function is intercepted by BatchRouter.
    If the request is batch-eligible, it is queued and a placeholder
    Future is returned. The Future resolves when the batch completes.
    If the request is real-time, it is forwarded immediately.
    """
    import anthropic
    client = anthropic.AsyncAnthropic()
    response = await client.messages.create(
        model=model, messages=messages, max_tokens=kwargs.get("max_tokens", 1024)
    )
    return response.content[0].text

# Usage: real-time (user-facing)
async def handle_user_message(user_input: str) -> str:
    return await call_llm(
        messages=[{"role": "user", "content": user_input}],
        batch_eligible=False,  # explicit: user is waiting
    )

# Usage: background enrichment (batch-eligible)
async def enrich_crm_record(record_id: str, notes: str) -> str:
    return await call_llm(
        messages=[
            {"role": "user", "content": f"Summarize these CRM notes: {notes}"}
        ],
        batch_eligible=True,
        deadline=datetime.utcnow() + timedelta(hours=4),
    )
    # BatchRouter queues this request; result available when batch completes

import runguard
from runguard.batch import BatchCostReconciler

rg = runguard.init(api_key="rg_live_...")

async def run_nightly_jobs() -> None:
    async with rg.budget(
        daily_usd=50.00,
        channel="batch",
        alert_at=0.80,  # alert at $40 spend
        alert_channels=["slack", "pagerduty"],
    ) as batch_budget:

        # Job 1: ticket classification
        tickets_cost = await run_ticket_classification_batch()
        batch_budget.record_cost(tickets_cost, job="ticket_classification")

        # Job 2: document summarization
        docs_cost = await run_document_summarization_batch()
        batch_budget.record_cost(docs_cost, job="document_summarization")

        # Job 3: embedding refresh
        embeddings_cost = await run_embedding_refresh_batch()
        batch_budget.record_cost(embeddings_cost, job="embedding_refresh")

    # RunGuard writes a daily batch cost report to the dashboard
    # with per-job breakdown and comparison against the budget


class BatchCostReconciler:
    """
    Utility class that parses Anthropic/OpenAI batch result files
    and computes actual cost from token usage fields.
    """
    def compute_anthropic_cost(
        self,
        results_file_path: str,
        input_price_per_mtoken: float,
        output_price_per_mtoken: float,
    ) -> float:
        total_cost = 0.0
        with open(results_file_path) as f:
            for line in f:
                record = json.loads(line)
                if record.get("result", {}).get("type") == "succeeded":
                    usage = record["result"]["message"].get("usage", {})
                    input_tokens = usage.get("input_tokens", 0)
                    output_tokens = usage.get("output_tokens", 0)
                    total_cost += (
                        input_tokens / 1_000_000 * input_price_per_mtoken
                        + output_tokens / 1_000_000 * output_price_per_mtoken
                    )
        return total_cost

rg.configure_batch_alert(
    error_rate_threshold=0.02,  # alert if >2% of requests fail
    alert_message_template=(
        "Batch {batch_id} has {error_rate:.1%} error rate "
        "({error_count}/{total_count} requests failed). "
        "Check batch results at {results_url}."
    ),
    channels=["slack"],
)