Lower per-token costs with LiteLLM request batching

Batching requests is one of the most effective ways to lower per-token costs when running inference at scale. By grouping multiple prompts into a single API call, you amortize fixed overhead — connection setup, TLS handshake, provider routing — across hundreds or thousands of tokens. This guide walks through implementing litellm request batching reduce token costs in a production setting, from configuration to verification.

Step 1: Understand the cost model before you batch

Most providers charge per token, but the effective cost per token drops when you batch because you eliminate per-request fixed costs. A single request carrying 4,000 tokens costs the same in provider fees as four requests of 1,000 tokens each, but you save on:

Network round trips (latency and bandwidth)
Load balancer and gateway overhead
Rate-limit slots consumed
Logging and observability ingestion

The tradeoff is latency: the batch completes when the slowest sub-request finishes. For throughput-oriented workloads (batch evaluation, offline enrichment, nightly jobs), this is acceptable. For user-facing chat, batch only non-critical background work.

Step 2: Install and configure LiteLLM with batch support

LiteLLM’s batch_completion function handles the orchestration. Start with a clean environment:

pip install litellm==1.48.0 tenacity prometheus-client

Create a configuration file that centralizes model routing, fallback, and batch defaults. This keeps your application code free of provider-specific logic.

# config/litellm_config.py
import os
from litellm import Router

router = Router(
    model_list=[
        {
            "model_name": "gpt-4o-mini-batch",
            "litellm_params": {
                "model": "openai/gpt-4o-mini",
                "api_key": os.getenv("OPENAI_API_KEY"),
                "max_tokens": 4096,
                "temperature": 0.1,
            },
            "tpm": 200_000,      # tokens per minute
            "rpm": 3_000,        # requests per minute
        },
        {
            "model_name": "gpt-4o-mini-batch-fallback",
            "litellm_params": {
                "model": "azure/gpt-4o-mini",
                "api_key": os.getenv("AZURE_API_KEY"),
                "api_base": os.getenv("AZURE_API_BASE"),
                "api_version": "2024-06-01",
                "max_tokens": 4096,
                "temperature": 0.1,
            },
            "tpm": 150_000,
            "rpm": 2_000,
        },
    ],
    routing_strategy="usage-based-routing",
    fallbacks=[{"gpt-4o-mini-batch": ["gpt-4o-mini-batch-fallback"]}],
    set_verbose=False,
)

The usage-based-routing strategy sends traffic to the deployment with the most available capacity, which matters when you’re pushing large batches.

Step 3: Define batch parameters for your workload

Batch size controls the throughput-latency curve. Start with these heuristics:

Workload type	Batch size	Max tokens per request	Timeout
Classification / tagging	100–500	500	60s
Summarization	20–100	2,000	120s
Code generation	10–50	4,000	180s
Embedding (if supported)	500–2,000	8,000	60s

Encode these as a dataclass so they’re version-controlled and testable:

# batching/params.py
from dataclasses import dataclass
from typing import Literal

@dataclass(frozen=True)
class BatchConfig:
    model: str
    batch_size: int
    max_tokens_per_request: int
    timeout_seconds: int
    max_retries: int = 3
    retry_backoff_base: float = 1.5

PRESETS = {
    "classification": BatchConfig(
        model="gpt-4o-mini-batch",
        batch_size=200,
        max_tokens_per_request=500,
        timeout_seconds=60,
    ),
    "summarization": BatchConfig(
        model="gpt-4o-mini-batch",
        batch_size=50,
        max_tokens_per_request=2000,
        timeout_seconds=120,
    ),
    "code_gen": BatchConfig(
        model="gpt-4o-mini-batch",
        batch_size=25,
        max_tokens_per_request=4000,
        timeout_seconds=180,
    ),
}

Step 4: Build a resilient batch client

Wrap batch_completion with retries, metrics, and structured error handling. The client should:

Chunk input lists into configured batch sizes
Emit per-batch latency and token counts
Surface partial failures without losing successful results
Respect provider rate limits via the router

# batching/client.py
import asyncio
import time
from dataclasses import dataclass
from typing import Any
from tenacity import retry, stop_after_attempt, wait_exponential_jitter

from litellm import batch_completion
from config.litellm_config import router
from batching.params import BatchConfig, PRESETS

@dataclass
class BatchResult:
    responses: list[dict[str, Any]]
    errors: list[dict[str, Any]]
    total_prompt_tokens: int
    total_completion_tokens: int
    latency_seconds: float

class BatchClient:
    def __init__(self, config: BatchConfig):
        self.config = config
        self._semaphore = asyncio.Semaphore(4)  # limit concurrent batches

    @retry(
        wait=wait_exponential_jitter(initial=1, max=30),
        stop=stop_after_attempt(3),
        reraise=True,
    )
    async def _execute_batch(self, messages_batch: list[list[dict]]) -> dict:
        """Single batch call with router-managed fallback."""
        start = time.perf_counter()
        try:
            response = await batch_completion(
                model=self.config.model,
                messages=messages_batch,
                max_tokens=self.config.max_tokens_per_request,
                temperature=0.1,
                timeout=self.config.timeout_seconds,
                router=router,  # enables fallback + usage-based routing
            )
            return {
                "response": response,
                "latency": time.perf_counter() - start,
                "error": None,
            }
        except Exception as e:
            return {
                "response": None,
                "latency": time.perf_counter() - start,
                "error": {"type": type(e).__name__, "message": str(e)},
            }

    async def process(self, all_messages: list[list[dict]]) -> BatchResult:
        """Process arbitrary-length message list in configured batches."""
        all_responses = []
        all_errors = []
        total_prompt = 0
        total_completion = 0
        total_latency = 0.0

        for i in range(0, len(all_messages), self.config.batch_size):
            chunk = all_messages[i : i + self.config.batch_size]
            async with self._semaphore:
                result = await self._execute_batch(chunk)

            total_latency += result["latency"]

            if result["error"]:
                for idx, _ in enumerate(chunk):
                    all_errors.append({
                        "batch_index": i + idx,
                        "error": result["error"],
                    })
                continue

            resp = result["response"]
            all_responses.extend(resp.choices)

            # Accumulate token usage from response metadata
            if hasattr(resp, "usage"):
                total_prompt += resp.usage.prompt_tokens
                total_completion += resp.usage.completion_tokens

        return BatchResult(
            responses=all_responses,
            errors=all_errors,
            total_prompt_tokens=total_prompt,
            total_completion_tokens=total_completion,
            latency_seconds=total_latency,
        )

Step 5: Implement the application workflow

Wire the client into your data pipeline. This example processes a JSONL file of classification tasks — a common use case where litellm request batching reduce token costs measurably.

# batching/run_classification.py
import asyncio
import json
import sys
from pathlib import Path

from batching.client import BatchClient
from batching.params import PRESETS

SYSTEM_PROMPT = """Classify the customer support ticket into exactly one category:
- billing
- technical
- account
- general
Return only the category name."""

def build_messages(ticket: dict) -> list[dict]:
    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": f"Subject: {ticket['subject']}\nBody: {ticket['body']}"},
    ]

async def main(input_path: Path, output_path: Path):
    config = PRESETS["classification"]
    client = BatchClient(config)

    # Load tickets
    tickets = []
    with input_path.open() as f:
        for line in f:
            tickets.append(json.loads(line))

    print(f"Loaded {len(tickets)} tickets. Batching with size={config.batch_size}...")

    all_messages = [build_messages(t) for t in tickets]
    result = await client.process(all_messages)

    # Write results preserving input order
    with output_path.open("w") as f:
        for idx, ticket in enumerate(tickets):
            # Find response for this index
            resp = next((r for r in result.responses if r.get("index") == idx), None)
            if resp:
                ticket["predicted_category"] = resp["message"]["content"].strip()
            else:
                ticket["predicted_category"] = "ERROR"
                ticket["error"] = next(
                    (e["error"] for e in result.errors if e["batch_index"] == idx),
                    "unknown",
                )
            f.write(json.dumps(ticket) + "\n")

    # Summary
    total_tokens = result.total_prompt_tokens + result.total_completion_tokens
    print(f"Completed in {result.latency_seconds:.1f}s")
    print(f"Tokens: {result.total_prompt_tokens:,} prompt + {result.total_completion_tokens:,} completion = {total_tokens:,} total")
    print(f"Errors: {len(result.errors)} / {len(tickets)}")
    print(f"Effective throughput: {total_tokens / result.latency_seconds:,.0f} tokens/sec")

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: python run_classification.py input.jsonl output.jsonl")
        sys.exit(1)
    asyncio.run(main(Path(sys.argv[1]), Path(sys.argv[2])))

Run it:

python -m batching.run_classification data/tickets.jsonl results/tickets_classified.jsonl

Step 6: Measure and verify cost reduction

You need three numbers to prove the batching worked: baseline cost, batched cost, and the delta. Capture them in a reproducible script.

# batching/verify_cost.py
import asyncio
import json
import statistics
from pathlib import Path

from batching.client import BatchClient
from batching.params import PRESETS

async def run_baseline(client: BatchClient, messages: list[list[dict]], runs: int = 3) -> list[float]:
    """Run unbatched (batch_size=1) to establish baseline."""
    original_batch_size = client.config.batch_size
    client.config = client.config.__class__(
        **{**client.config.__dict__, "batch_size": 1}
    )
    latencies = []
    for _ in range(runs):
        result = await client.process(messages)
        latencies.append(result.latency_seconds)
    client.config = client.config.__class__(
        **{**client.config.__dict__, "batch_size": original_batch_size}
    )
    return latencies

async def run_batched(client: BatchClient, messages: list[list[dict]], runs: int = 3) -> list[float]:
    latencies = []
    for _ in range(runs):
        result = await client.process(messages)
        latencies.append(result.latency_seconds)
    return latencies

async def main():
    # Load test data
    tickets = []
    with Path("data/tickets.jsonl").open() as f:
        for line in f:
            tickets.append(json.loads(line))

    messages = [
        [
            {"role": "system", "content": "Classify: billing, technical, account, general"},
            {"role": "user", "content": f"Subject: {t['subject']}\nBody: {t['body']}"},
        ]
        for t in tickets[:200]  # 200 tickets for test
    ]

    config = PRESETS["classification"]
    client = BatchClient(config)

    print("Running baseline (batch_size=1)...")
    baseline_latencies = await run_baseline(client, messages)
    print(f"  Latencies: {[f'{l:.1f}s' for l in baseline_latencies]}")
    print(f"  Median: {statistics.median(baseline_latencies):.1f}s")

    print(f"\nRunning batched (batch_size={config.batch_size})...")
    batched_latencies = await run_batched(client, messages)
    print(f"  Latencies: {[f'{l:.1f}s' for l in batched_latencies]}")
    print(f"  Median: {statistics.median(batched_latencies):.1f}s")

    speedup = statistics.median(baseline_latencies) / statistics.median(batched_latencies)
    print(f"\nSpeedup factor: {speedup:.1f}x")
    print(f"Cost reduction estimate: {(1 - 1/speedup) * 100:.0f}% (from eliminated per-request overhead)")

if __name__ == "__main__":
    asyncio.run(main())

Expected output on a typical workload:

Running baseline (batch_size=1)...
  Latencies: ['42.3s', '41.8s', '43.1s']
  Median: 42.3s

Running batched (batch_size=200)...
  Latencies: ['8.7s', '8.4s', '8.9s']
  Median: 8.7s

Speedup factor: 4.9x
Cost reduction estimate: 80% (from eliminated per-request overhead)

The 80% figure represents overhead elimination — provider token costs stay the same, but you consume fewer rate-limit slots, less gateway CPU, and less logging volume. At scale, that translates directly to infrastructure savings.

Step 7: Handle partial failures and retries gracefully

Production batches will hit transient errors. The client above retries whole batches, but you often want per-item retry for permanent failures (e.g., one prompt triggers a content filter). Extend the result processor:

# batching/retry_failed.py
async def retry_failed_items(
    client: BatchClient,
    original_messages: list[list[dict]],
    result: BatchResult,
    max_item_retries: int = 2,
) -> BatchResult:
    """Retry only the failed indices individually."""
    if not result.errors:
        return result

    failed_indices = [e["batch_index"] for e in result.errors]
    retry_messages = [original_messages[i] for i in failed_indices]

    print(f"Retrying {len(retry_messages)} failed items individually...")
    retry_result = await client.process(retry_messages)

    # Merge successful retries back
    merged_responses = list(result.responses)
    merged_errors = [e for e in result.errors if e["batch_index"] not in failed_indices]

    for idx, resp in enumerate(retry_result.responses):
        original_idx = failed_indices[idx]
        resp.index = original_idx  # preserve ordering
        merged_responses.append(resp)

    for err in retry_result.errors:
        merged_errors.append({
            "batch_index": failed_indices[err["batch_index"]],
            "error": err["error"],
        })

    return BatchResult(
        responses=merged_responses,
        errors=merged_errors,
        total_prompt_tokens=result.total_prompt_tokens + retry_result.total_prompt_tokens,
        total_completion_tokens=result.total_completion_tokens + retry_result.total_completion_tokens,
        latency_seconds=result.latency_seconds + retry_result.latency_seconds,
    )

Call this after the initial client.process() if your error rate exceeds a threshold.

Step 8: Monitor in production

Add Prometheus metrics to track batch efficiency over time:

# batching/metrics.py
from prometheus_client import Counter, Histogram, Gauge

BATCH_REQUESTS = Counter(
    "litellm_batch_requests_total",
    "Total batch requests",
    ["model", "status"],
)
BATCH_LATENCY = Histogram(
    "litellm_batch_latency_seconds",
    "Batch request latency",
    ["model"],
    buckets=[1, 5, 10, 30, 60, 120, 300],
)
BATCH_SIZE = Histogram(
    "litellm_batch_size",
    "Number of requests per batch",
    ["model"],
    buckets=[1, 5, 10, 25, 50, 100, 200, 500, 1000],
)
TOKENS_PROCESSED = Counter(
    "litellm_tokens_processed_total",
    "Total tokens processed",
    ["model", "type"],  # prompt / completion
)
QUEUE_DEPTH = Gauge(
    "litellm_batch_queue_depth",
    "Pending items waiting for batch",
    ["model"],
)

Instrument the client:

# In BatchClient.process, after chunking:
QUEUE_DEPTH.labels(model=self.config.model).set(len(all_messages) - i)

# After each batch:
BATCH_REQUESTS.labels(model=self.config.model, status="success" if not err else "error").inc()
BATCH_LATENCY.labels(model=self.config.model).observe(latency)
BATCH_SIZE.labels(model=self.config.model).observe(len(chunk))
TOKENS_PROMPT.labels(model=self.config.model, type="prompt").inc(prompt_tokens)
TOKENS_PROMPT.labels(model=self.config.model, type="completion").inc(completion_tokens)

Alert on:

rate(litellm_batch_requests_total{status="error"}[5m]) > 0.05 — error rate above 5%
histogram_quantile(0.95, litellm_batch_latency_seconds) > 120 — p95 latency breach
litellm_batch_queue_depth > 10000 — backlog growing

Common pitfalls to avoid

Batching user-facing requests. Don’t batch chat completions where users wait for each response. The tail latency of the slowest item in the batch becomes everyone’s latency.

Ignoring token limits. A batch of 200 requests at 4,000 tokens each = 800,000 tokens. Many providers cap per-request tokens at 128k–1M. Validate batch_size * max_tokens_per_request against the model’s context window and provider limits.

Losing ordering. batch_completion returns responses in the same order as input messages, but only if you don’t filter or reorder. Keep the original index on each message if you need to join back to source data.

Over-retrying. Retrying a full batch of 200 because one item hit a content filter wastes 199 successful completions. Use the per-item retry pattern in Step 7.

Skipping the router. Direct provider calls bypass fallback and usage-based routing. Always pass router=router to batch_completion so the router can shift traffic when a deployment hits limits.

Scaling beyond a single machine

When one process can’t consume the queue fast enough, run multiple workers with a shared message queue (Redis, RabbitMQ, Kafka). Each worker pulls a chunk, processes via the batch client, and acknowledges. The batch client stays stateless — all coordination lives in the queue.

# batching/worker.py
import asyncio
import json
import redis.asyncio as redis

async def worker(worker_id: int, queue_name: str, client: BatchClient):
    r = redis.from_url("redis://localhost:6379")
    while True:
        # Blocking pop with 5s timeout
        item = await r.blpop(queue_name, timeout=5)
        if not item:
            continue
        _, payload = item
        batch = json.loads(payload)  # list of message lists
        result = await client.process(batch)
        # Write results to output store, acknowledge, etc.

Deploy N workers behind the same queue. The router distributes load across provider deployments automatically.

You now have a complete, production-ready batching pipeline: configuration, client, verification, monitoring, and scaling pattern. The cost reduction comes from eliminating per-request overhead — not from cheaper tokens — so the savings compound with volume. Start with the classification preset, verify the 4–5x speedup on your data, then tune batch sizes for your specific workloads.