LLM Batch Inference Cut Costs 50% Production Guide 2026

# batch_inference_service.py - Production vLLM Continuous Batching
import asyncio
from typing import List, Dict, Any, Optional
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import vllm
from vllm import AsyncLLMEngine, AsyncEngineArgs, SamplingParams
import logging
import time
from collections import deque
import prometheus_client

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Prometheus metrics
BATCH_SIZE = prometheus_client.Histogram(
    'batch_size',
    'Batch size distribution',
    buckets=[1, 2, 4, 8, 16, 32, 64, 128]
)
THROUGHPUT = prometheus_client.Gauge(
    'throughput_tokens_per_sec',
    'Tokens per second throughput'
)
LATENCY = prometheus_client.Histogram(
    'request_latency_seconds',
    'Request latency distribution'
)

# Initialize vLLM with continuous batching
engine_args = AsyncEngineArgs(
    model="meta-llama/Llama-2-7b-chat-hf",
    tensor_parallel_size=1,  # Use 1 GPU
    dtype="float16",  # FP16 for 2x speedup
    max_num_batched_tokens=8192,  # Batch capacity
    max_num_seqs=128,  # Max concurrent sequences
    gpu_memory_utilization=0.90,  # Use 90% of GPU memory
    enable_prefix_caching=True,  # Cache system prompts
    enable_chunked_prefill=True,  # Faster TTFT
    swap_space=4,  # 4GB CPU swap for overflow
)

# Create AsyncLLMEngine for continuous batching
llm_engine = AsyncLLMEngine.from_engine_args(engine_args)

# FastAPI application
app = FastAPI(title="Continuous Batching LLM Service")

# Request queue for monitoring
request_queue = deque(maxlen=1000)

class BatchRequest(BaseModel):
    prompts: List[str]
    temperature: float = 0.7
    max_tokens: int = 512
    top_p: float = 0.9

class BatchResponse(BaseModel):
    results: List[str]
    metadata: Dict[str, Any]

@app.post("/batch/generate", response_model=BatchResponse)
async def batch_generate(request: BatchRequest):
    """
    Batch generation with continuous batching
    Requests are processed immediately, no waiting for batch to fill
    """
    start_time = time.time()

    try:
        # Record batch size
        batch_size = len(request.prompts)
        BATCH_SIZE.observe(batch_size)
        logger.info(f"Processing batch of {batch_size} requests")

        # Configure sampling parameters
        sampling_params = SamplingParams(
            temperature=request.temperature,
            top_p=request.top_p,
            max_tokens=request.max_tokens,
            frequency_penalty=0.1,
            presence_penalty=0.1,
        )

        # Submit all prompts to continuous batching engine
        # vLLM handles dynamic batch composition internally
        request_ids = []
        for i, prompt in enumerate(request.prompts):
            request_id = f"batch_{start_time}_{i}"
            request_queue.append({
                "id": request_id,
                "prompt": prompt,
                "timestamp": start_time
            })
            request_ids.append(request_id)

        # Generate responses with continuous batching
        # Engine dynamically adds/removes sequences from batch
        results = []
        total_tokens = 0

        for i, prompt in enumerate(request.prompts):
            # Each request processed independently
            # Continuous batching happens automatically
            output = await llm_engine.generate(
                prompt,
                sampling_params,
                request_id=request_ids[i]
            )

            # Extract generated text
            generated_text = output.outputs[0].text
            results.append(generated_text)

            # Track token count for throughput
            total_tokens += len(output.outputs[0].token_ids)

        # Calculate metrics
        end_time = time.time()
        latency = end_time - start_time
        throughput = total_tokens / latency if latency > 0 else 0

        # Update Prometheus metrics
        LATENCY.observe(latency)
        THROUGHPUT.set(throughput)

        logger.info(
            f"Batch completed: {batch_size} requests, "
            f"{latency:.2f}s, {throughput:.0f} tokens/sec"
        )

        return BatchResponse(
            results=results,
            metadata={
                "batch_size": batch_size,
                "latency_seconds": latency,
                "throughput_tokens_per_sec": throughput,
                "total_tokens": total_tokens,
                "avg_tokens_per_request": total_tokens / batch_size
            }
        )

    except Exception as e:
        logger.error(f"Batch generation failed: {str(e)}")
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/metrics")
async def get_metrics():
    """Prometheus metrics endpoint"""
    return prometheus_client.generate_latest()

@app.get("/health")
async def health_check():
    """Health check for load balancers"""
    queue_size = len(request_queue)
    return {
        "status": "healthy",
        "model": "Llama-2-7b-chat",
        "queue_size": queue_size,
        "continuous_batching": True
    }

# Run with: uvicorn batch_inference_service:app --host 0.0.0.0 --port 8000

# openai_batch_processor.py - Production OpenAI Batch Processing
import asyncio
import openai
from typing import List, Dict, Any
import json
import time
from pathlib import Path
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class OpenAIBatchProcessor:
    """
    Production batch processor for OpenAI Batch API
    50% cost reduction vs real-time API
    """

    def __init__(self, api_key: str):
        openai.api_key = api_key
        self.client = openai.OpenAI()

    async def submit_batch_job(
        self,
        prompts: List[str],
        model: str = "gpt-4o-mini",
        temperature: float = 0.7,
        max_tokens: int = 500
    ) -> str:
        """
        Submit batch job to OpenAI
        Returns batch_id for tracking
        """
        # Create JSONL file with batch requests
        batch_file = self._create_batch_file(
            prompts, model, temperature, max_tokens
        )

        # Upload batch file
        logger.info(f"Uploading batch file: {len(prompts)} requests")
        uploaded_file = self.client.files.create(
            file=open(batch_file, "rb"),
            purpose="batch"
        )

        # Create batch job
        batch = self.client.batches.create(
            input_file_id=uploaded_file.id,
            endpoint="/v1/chat/completions",
            completion_window="24h",  # Process within 24 hours
            metadata={"description": f"Batch job {len(prompts)} requests"}
        )

        logger.info(f"Batch job submitted: {batch.id}")
        return batch.id

    def _create_batch_file(
        self,
        prompts: List[str],
        model: str,
        temperature: float,
        max_tokens: int
    ) -> Path:
        """Create JSONL file for batch API"""
        batch_file = Path(f"batch_{int(time.time())}.jsonl")

        with open(batch_file, "w") as f:
            for i, prompt in enumerate(prompts):
                request = {
                    "custom_id": f"request_{i}",
                    "method": "POST",
                    "url": "/v1/chat/completions",
                    "body": {
                        "model": model,
                        "messages": [{"role": "user", "content": prompt}],
                        "temperature": temperature,
                        "max_tokens": max_tokens
                    }
                }
                f.write(json.dumps(request) + "\n")

        return batch_file

    async def get_batch_status(self, batch_id: str) -> Dict[str, Any]:
        """Check batch job status"""
        batch = self.client.batches.retrieve(batch_id)

        return {
            "status": batch.status,
            "completed": batch.request_counts.completed,
            "failed": batch.request_counts.failed,
            "total": batch.request_counts.total,
            "progress": (
                batch.request_counts.completed / batch.request_counts.total * 100
                if batch.request_counts.total > 0 else 0
            )
        }

    async def retrieve_results(self, batch_id: str) -> List[str]:
        """
        Retrieve completed batch results
        Polls every 30s until complete
        """
        logger.info(f"Waiting for batch {batch_id} to complete...")

        while True:
            status = await self.get_batch_status(batch_id)
            logger.info(f"Batch progress: {status['progress']:.1f}%")

            if status["status"] == "completed":
                break
            elif status["status"] == "failed":
                raise Exception(f"Batch job failed: {batch_id}")

            # Poll every 30 seconds
            await asyncio.sleep(30)

        # Download results
        batch = self.client.batches.retrieve(batch_id)
        output_file = self.client.files.content(batch.output_file_id)

        # Parse results
        results = []
        for line in output_file.text.strip().split("\n"):
            data = json.loads(line)
            if data["response"]["status_code"] == 200:
                message = data["response"]["body"]["choices"][0]["message"]["content"]
                results.append(message)
            else:
                results.append(f"Error: {data['error']}")

        logger.info(f"Retrieved {len(results)} results")
        return results

    async def process_batch(
        self,
        prompts: List[str],
        **kwargs
    ) -> List[str]:
        """
        End-to-end batch processing
        Submit job, wait for completion, return results
        """
        batch_id = await self.submit_batch_job(prompts, **kwargs)
        results = await self.retrieve_results(batch_id)
        return results

# Usage example
async def main():
    processor = OpenAIBatchProcessor(api_key="your-api-key")

    prompts = [
        "Summarize the Q4 financial report",
        "Generate product description for item X",
        # ... 1000s more prompts
    ]

    # Submit batch (50% cheaper than real-time)
    results = await processor.process_batch(prompts, model="gpt-4o-mini")

    # Cost comparison:
    # Real-time: $1.00/1M tokens = $100 for 100M tokens
    # Batch API: $0.50/1M tokens = $50 for 100M tokens
    # Savings: $50 (50% reduction)

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