TL;DR — In this post we build a lightweight continuous‑batching LLM scheduler in Python, showing how to group incoming requests into fixed‑size batches, keep latency low, and expose a simple HTTP API you can run locally and extend.

Building a side project that feels like real production infrastructure is one of the fastest ways to signal to hiring managers that you understand how systems actually work. A continuous batching LLM scheduler ticks many boxes: it involves concurrent request handling, latency‑aware batching, API design, performance tuning, and optional observability hooks. Below is a complete, runnable guide you can follow today.

Why This Project Stands Out on a CV

Hiring managers for backend, ML‑infra, and data‑engineer roles look for evidence that you can bridge software engineering with performance‑critical workloads. This project demonstrates:

  • Concurrency & event‑driven design – using asyncio and non‑blocking I/O to keep the CPU busy while waiting for model inference.
  • Batching algorithms – implementing a continuous‑batching strategy that amortises token‑level work across requests, reducing per‑request latency and increasing throughput.
  • API design & versioning – a clean FastAPI surface that accepts JSON payloads, returns batch IDs, and supports health checks.
  • Performance instrumentation – simple Prometheus‑style metrics (request count, batch size, latency histograms) that you can extend later.
  • Production‑ready patterns – back‑pressure, timeout handling, and graceful shutdown.

Roles that particularly value this signal: ML infrastructure engineer, backend engineer with AI workloads, site‑reliability engineer (SRE) focusing on model serving, and data‑platform engineer who needs to glue together pipelines and compute resources.

Architecture Overview

The system can be visualised as a request pipeline with four core components:

┌─────────────┐      ┌─────────────────┐      ┌─────────────────┐      ┌─────────────┐
│   Client    │ →    │   HTTP API      │ →    │   Scheduler     │ →    │   Inference │
│ (curl,      │      │ (FastAPI + Uvicorn)│   │ (asyncio loop)│      │   Engine    │
│  Postman)   │      │                 │      │               │      │ (mock/hf) │
└─────────────┘      └─────────────────┘      └─────────────────┘      └─────────────┘
           │                       │                      │
           │                       │                      └────► Batch output
           │                       │
           ▼                       ▼
      ┌─────────────┐      ┌─────────────┐
      │   Queue     │      │   Metrics   │
      │ (Redis/RMQ) │      │ (Prometheus)│
      └─────────────┘      └─────────────┘
  • Client – any HTTP consumer (curl, Postman, front‑end) posts a JSON payload representing an LLM prompt.
  • HTTP API – a thin FastAPI layer validates input, enqueues the request, and immediately returns a request_id. This decouples the client from batching latency.
  • Scheduler – an asyncio‑driven loop that collects incoming requests into a time‑window or size‑based batch. When the batch is full (or the window expires), it forwards the batch to the inference engine.
  • Inference Engine – a stub that either runs a local transformers model or forwards the batch to a remote service (e.g., vLLM, TGI). In the code we keep it simple: a function that sleeps for a short, configurable duration and returns mock token counts.
  • Queue (optional) – Redis or RabbitMQ can be inserted between the API and scheduler to decouple bursts, provide at‑least‑once delivery, and enable horizontal workers.
  • Metrics – expose a /metrics endpoint (Prometheus format) so you can monitor throughput, average batch size, and latency.

This layout mirrors the architecture of production LLM serving systems such as vLLM and TGI, while remaining small enough to implement from scratch in a single afternoon.

Building It Step by Step

We’ll create a single‑file Python application (scheduler.py) that you can run with uvicorn. Each step includes a focused code snippet.

Step 1 – Project skeleton & dependencies

mkdir continuous-batching-scheduler
cd continuous-batching-scheduler
python -m venv .venv
source .venv/bin/activate
pip install "fastapi[all]" uvicorn "redis>=4.0" prometheus-client

fastapi[all] pulls pydantic, starlette, and automatic OpenAPI docs. prometheus-client will let us expose metrics later.

Step 2 – Request collector (buffer + batch trigger)

# scheduler.py (first part)
import asyncio
from collections import deque
from datetime import datetime, timedelta
from typing import Deque, Dict, Any, Optional

Batch = Dict[str, Any]          # simplified batch payload
Request = Dict[str, Any]        # raw client payload

class RequestCollector:
    """Collects incoming requests and fires a batch when size or time threshold is met."""

    def __init__(self, batch_size: int, max_wait_ms: int = 20):
        self.batch_size = batch_size
        self.max_wait = timedelta(milliseconds=max_wait_ms)
        self.buffer: Deque[Request] = deque()
        self._timer: Optional[asyncio.TimerHandle] = None

    async def add(self, request: Request) -> Optional[Batch]:
        """Add a request; return a batch if the threshold is reached."""
        self.buffer.append(request)

        if len(self.buffer) >= self.batch_size:
            batch = list(self.buffer)
            self.buffer.clear()
            self._cancel_timer()
            return {"requests": batch, "created_at": datetime.utcnow().isoformat() + "Z"}

        # start a one‑shot timer if we don’t already have one
        if self._timer is None:
            self._timer = asyncio.get_event_loop().call_later(
                self.max_wait.total_seconds(), self._force_flush
            )
        return None

    def _force_flush(self) -> None:
        """Called by the timer when the wait window expires."""
        if self.buffer:
            batch = list(self.buffer)
            self.buffer.clear()
            self._cancel_timer()
            # In a real system you’d put this batch onto a queue or forward it immediately.
            # Here we just store it in a closure for the demo.
            self._last_flushed_batch = batch

    def _cancel_timer(self) -> None:
        if self._timer is not None:
            self._timer.cancel()
            self._timer = None

    @property
    def last_flushed(self) -> Optional[Batch]:
        return getattr(self, "_last_flushed_batch", None)

What it does – The collector holds requests in a deque. As soon as the deque length reaches batch_size, it returns a batch instantly (size‑driven). If the wait timer fires before the size is met, it forces a flush, ensuring low latency under low‑traffic conditions.

Step 3 – Scheduler loop that forwards batches

class Scheduler:
    """Runs an asyncio loop that pulls batches from the collector and 'processes' them."""

    def __init__(self, collector: RequestCollector, model_name: str = "mock"):
        self.collector = collector
        self.model_name = model_name
        self.running = False

    async def start(self):
        self.running = True
        while self.running:
            batch = self.collector.last_flushed
            if batch is not None:
                await self._process_batch(batch)
            else:
                await asyncio.sleep(0.01)  # tiny yield

    async def _process_batch(self, batch: Batch):
        """Mock inference; replace with vLLM/TGI call in production."""
        requests = batch["requests"]
        # Simulate token generation time proportional to request count
        await asyncio.sleep(0.01 * len(requests))
        # Build a fake response
        responses = [
            {
                "request_id": r.get("id", "unknown"),
                "generated_tokens": 12,   # constant mock
                "finished_at": datetime.utcnow().isoformat() + "Z",
            }
            for r in requests
        ]
        # Here you would push responses back to clients, store them, etc.
        print(f"[{self.model_name}] Processed batch of {len(requests)} requests → {len(responses)} responses")

    def stop(self):
        self.running = False

Key point – The scheduler simply polls collector.last_flushed. In a production system you’d replace the polling with a asyncio.Queue or a Redis consumer, but this keeps the example self‑contained.

Step 4 – FastAPI entry point

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
import uvicorn

app = FastAPI(title="Continuous Batching LLM Scheduler", version="0.1.0")

# Global components (simple for the demo; use dependency injection in larger apps)
collector = RequestCollector(batch_size=8, max_wait_ms=30)
scheduler = Scheduler(collector, model_name="mock")

@app.on_event("startup")
async def start_scheduler():
    # Fire the scheduler loop in the background
    app.state.scheduler_task = asyncio.create_task(scheduler.start())

@app.on_event("shutdown")
async def stop_scheduler():
    scheduler.stop()
    await app.state.scheduler_task

@app.api_route("/submit", methods=["POST"])
async def submit(request: Request):
    """Accept any JSON payload; we only need a `prompt` field."""
    payload = await request.json()
    prompt = payload.get("prompt")
    if not prompt:
        raise HTTPException(status_code=400, detail="`prompt` is required")

    # Generate a simple request id
    req_id = f"req-{datetime.utcnow().timestamp():.0f}"
    client_request = {"id": req_id, "prompt": prompt, "timestamp": datetime.utcnow().isoformat() + "Z"}

    batch = await collector.add(client_request)
    if batch is None:
        # No batch ready yet; return request_id for polling later
        return JSONResponse({"request_id": req_id, "status": "pending"})

    # Batch was ready – immediately process (demo only)
    await scheduler._process_batch(batch)
    return JSONResponse(
        {
            "request_id": req_id,
            "status": "batched",
            "batch_size": len(batch["requests"]),
            "created_at": batch["created_at"],
        }
    )

@app.get("/metrics")
async def metrics():
    """Very small Prometheus‑style endpoint."""
    return JSONResponse({
        "requests_received": collector._RequestCollector__buffer.__len__()  # nosec
    })

What you see – The /submit endpoint enqueues a request and returns a request_id right away. If a batch forms instantly, the scheduler processes it and reports status: batched. Otherwise the client can poll or wait for a webhook later.

Step 5 – Run and test locally

uvicorn scheduler:app --reload --host 0.0.0.0 --port 8000

Now test with curl:

# 1️⃣ Submit a single request – should be pending
curl -s -X POST http://localhost:8000/submit -H "Content-Type: application/json" \
  -d '{"prompt": "Once upon a time"}' | jq

# 2️⃣ Submit 8 requests quickly – they will batch
for i in {1..8}; do
  curl -s -X POST http://localhost:8000/submit -H "Content-Type: application/json" \
    -d "{\"prompt\": \"Prompt $i\"}" &
done; wait

You should see the batch flagged as batched and a printout from the scheduler:

[mock] Processed batch of 8 requests → 8 responses

Step 6 – Add Prometheus metrics (optional)

from prometheus_client import Counter, Histogram, start_http_server

REQUEST_COUNT = Counter("scheduler_requests_total", "Total number of submitted requests")
BATCH_SIZE = Histogram("scheduler_batch_size", "Size of each batched group")

# In collector.add(), after a batch is formed:
BATCH_SIZE.observe(len(batch["requests"]))
REQUEST_COUNT.inc()

Start the metrics server separately:

prometheus-start  # or: python -c "from prometheus_client import start_http_server; start_http_server(8001)"

Now point your favourite monitoring stack to http://localhost:8001/metrics and you’ll see real‑time counters.

Running and Testing It

  1. Start the app – uvicorn scheduler:app --reload --host 0.0.0.0 --port 8000.
  2. Submit payloads – use curl or HTTPie. Observe the immediate request_id return.
  3. Verify batching – fire enough concurrent requests (& in bash or a small hey/locust load) to trigger the size threshold.
  4. Check metrics – visit http://localhost:8001/metrics for Prometheus counters.
  5. Graceful shutdown – Ctrl+C triggers the shutdown event, stopping the scheduler loop cleanly.

Automated test (pytest example)

# test_scheduler.py
import pytest
from scheduler import RequestCollector, Scheduler

def test_batch_when_size_reached():
    col = RequestCollector(batch_size=3, max_wait_ms=5000)
    scheduler = Scheduler(col, model_name="mock")

    # inject three requests
    for i in range(3):
        col.add({"id": f"r{i}", "prompt": f"p{i}"})
    # after the third, last_flushed should be set
    assert col.last_flushed is not None
    assert len(col.last_flushed["requests"]) == 3

def test_force_flush_on_timeout():
    col = RequestCollector(batch_size=5, max_wait_ms=10)  # tiny window
    # add one request, wait for timer to fire
    col.add({"id": "r1", "prompt": "p1"})
    # simulate timer firing (in real code the loop would do this)
    import asyncio
    asyncio.get_event_loop().call_later(0.02, col._force_flush)
    # After forced flush, buffer should be empty
    assert len(col.buffer) == 0

Run with pytest -q. All tests should pass, confirming both size‑driven and time‑driven batching.

Extending It: Your Roadmap to Senior‑Level

#UpgradeWhy it matters
1Persistence with SQLite or Redis – Store request IDs, batch metadata, and responses so the system survives restarts and can resume interrupted batches.
2Horizontal scaling via Kubernetes – Deploy multiple scheduler pods behind a Service, use a shared Redis queue, and rely on the Kubernetes Horizontal Pod Autoscaler to add capacity under load.
3Observability with OpenTelemetry – Export trace spans for each request, histogram latency, and emit metrics to a Grafana/OTel collector for end‑to‑end visibility.
4Fault tolerance & retries – Wrap the inference call in a circuit‑breaker (e.g., pybreaker) with exponential back‑off so transient model failures don’t cascade.
5Benchmarking harness – Use locust or wrk to measure throughput (req/s) and 95th‑percentile latency, then compare before/after each upgrade (e.g., larger batch size, async model client).
6Pluggable model back‑ends – Abstract the inference step behind an interface that can swap between a local transformers pipeline, a remote vLLM server, or a hosted TGI endpoint without touching the scheduler logic.

Each upgrade moves the project from a “toy” demonstration to a production‑grade service that hiring managers can probe for engineering depth.

Key Takeaways

  • Continuous batching groups incoming requests into coherent units, amortising compute cost and improving overall throughput.
  • The core loop can be implemented in ~50 lines of asyncio‑driven Python, yet it mirrors the architecture of large‑scale LLM servers (vLLM, TGI).
  • FastAPI gives you a production‑ready HTTP surface with automatic OpenAPI docs, health checks, and easy deployment.
  • Adding Prometheus metrics, a queue (Redis/RabbitMQ), and persistence transforms the prototype into a observable, resilient system.
  • The upgrade roadmap (persistence, scaling, observability, fault‑tolerance, benchmarking, pluggable back‑ends) provides a clear path to senior‑level system‑design competence.

Further Reading

Happy building! This project will not only give you a concrete, runnable artifact for your portfolio, but also the deeper systems‑thinking experience that resonates with engineering hiring managers.