Integrating LLMs Into Microservices — Async Patterns, Queues, and Service Design

Introduction

LLM calls can take 5–30 seconds to complete — far too long to block a synchronous HTTP request. Integrating large language models into microservice architectures requires async job queues, webhook callbacks, circuit breakers, and service isolation so that AI workloads never cascade into user-facing timeouts.

Synchronous vs. Asynchronous LLM Integration

The fundamental choice is whether to block the caller until the LLM responds or return immediately and deliver results later. For latency-tolerant workflows like report generation or batch enrichment, synchronous calls are acceptable. For user-facing endpoints, asynchronous patterns are mandatory.

A synchronous wrapper is the simplest starting point:

import anthropic
import time
 
client = anthropic.Anthropic()
 
def sync_llm_call(prompt: str, model: str = "claude-3-5-sonnet-20241022") -> dict:
    start = time.time()
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return {
        "text": response.content[0].text,
        "input_tokens": response.usage.input_tokens,
        "output_tokens": response.usage.output_tokens,
        "latency_ms": int((time.time() - start) * 1000)
    }

This works fine in background workers, but never call it directly inside a web request handler.

Async Job Queue Pattern

The production pattern is to accept the request, enqueue a job, return a job ID, and let the caller poll or receive a webhook callback.

import uuid
import time
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
 
class JobStatus(Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"
 
@dataclass
class LLMJob:
    job_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    prompt: str = ""
    status: JobStatus = JobStatus.PENDING
    result: Optional[str] = None
    error: Optional[str] = None
    created_at: float = field(default_factory=time.time)
    completed_at: Optional[float] = None
    callback_url: Optional[str] = None
 
# In production, back this with Redis or a real message broker
job_store: dict[str, LLMJob] = {}
 
def enqueue_llm_job(prompt: str, callback_url: Optional[str] = None) -> str:
    job = LLMJob(prompt=prompt, callback_url=callback_url)
    job_store[job.job_id] = job
    # push job_id to Redis queue or SQS here
    return job.job_id
 
def get_job_status(job_id: str) -> Optional[LLMJob]:
    return job_store.get(job_id)

A worker process pulls from the queue and executes:

import anthropic
 
def process_job(job: LLMJob) -> None:
    client = anthropic.Anthropic()
    job.status = JobStatus.PROCESSING
    try:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{"role": "user", "content": job.prompt}]
        )
        job.result = response.content[0].text
        job.status = JobStatus.COMPLETED
        job.completed_at = time.time()
        if job.callback_url:
            notify_callback(job)
    except Exception as e:
        job.error = str(e)
        job.status = JobStatus.FAILED

Webhook Callback Delivery

When a job completes, notify the originating service via HTTP POST with retry logic:

import httpx
import asyncio
 
async def notify_callback(job: LLMJob, max_retries: int = 5) -> None:
    payload = {
        "job_id": job.job_id,
        "status": job.status.value,
        "result": job.result,
        "error": job.error,
    }
    async with httpx.AsyncClient() as client:
        for attempt in range(max_retries):
            try:
                resp = await client.post(
                    job.callback_url,
                    json=payload,
                    timeout=10.0
                )
                if resp.status_code < 300:
                    return
            except Exception:
                pass
            # exponential backoff: 1s, 2s, 4s, 8s, 16s
            await asyncio.sleep(2 ** attempt)

Use at-least-once delivery and make callback endpoints idempotent by checking job_id before processing.

LLM Service as an Isolated Microservice

Separate the LLM gateway into its own service with a clean API surface. This enables independent scaling, centralized cost tracking, and easy model swapping:

from dataclasses import dataclass
import time
import anthropic
 
@dataclass
class LLMRequest:
    prompt: str
    system_prompt: str = ""
    model: str = "claude-3-5-sonnet-20241022"
    max_tokens: int = 1024
    temperature: float = 0.7
 
@dataclass
class LLMResponse:
    text: str
    input_tokens: int
    output_tokens: int
    latency_ms: int
    model: str
    cost_usd: float
 
# Pricing per 1K tokens (update as models change)
MODEL_PRICING = {
    "claude-3-5-sonnet-20241022": {"input": 0.003, "output": 0.015},
    "claude-3-haiku-20240307":    {"input": 0.00025, "output": 0.00125},
}
 
class LLMGatewayService:
    def __init__(self):
        self._client = anthropic.Anthropic()
 
    def generate(self, req: LLMRequest) -> LLMResponse:
        start = time.time()
        msg = self._client.messages.create(
            model=req.model,
            max_tokens=req.max_tokens,
            system=req.system_prompt or anthropic.NOT_GIVEN,
            messages=[{"role": "user", "content": req.prompt}]
        )
        latency_ms = int((time.time() - start) * 1000)
        pricing = MODEL_PRICING.get(req.model, {"input": 0, "output": 0})
        cost = (
            msg.usage.input_tokens  * pricing["input"]  / 1000 +
            msg.usage.output_tokens * pricing["output"] / 1000
        )
        return LLMResponse(
            text=msg.content[0].text,
            input_tokens=msg.usage.input_tokens,
            output_tokens=msg.usage.output_tokens,
            latency_ms=latency_ms,
            model=req.model,
            cost_usd=cost,
        )

Circuit Breaker for Resilience

Prevent cascading failures when the LLM API is degraded. The circuit breaker opens after a threshold of failures and auto-resets after a cooldown:

import time
from enum import Enum
 
class CircuitState(Enum):
    CLOSED = "closed"       # Normal: requests pass through
    OPEN = "open"           # Failing: requests rejected immediately
    HALF_OPEN = "half_open" # Recovering: one trial request allowed
 
class LLMCircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout_sec=60):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout_sec
        self.last_failure_time = 0.0
 
    def call(self, fn, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise RuntimeError("Circuit breaker is OPEN — LLM service unavailable")
        try:
            result = fn(*args, **kwargs)
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
            raise

Shared Client Library for Consistent Integration

Every service that calls the LLM gateway should use the same client library with built-in retry, timeout, and header propagation:

import httpx
import time
 
class LLMGatewayClient:
    def __init__(self, base_url: str, api_key: str, timeout: float = 30.0):
        self.base_url = base_url.rstrip("/")
        self.headers = {"Authorization": f"Bearer {api_key}"}
        self.timeout = timeout
 
    def enqueue(self, prompt: str, callback_url: str = "") -> str:
        resp = httpx.post(
            f"{self.base_url}/jobs",
            json={"prompt": prompt, "callback_url": callback_url},
            headers=self.headers,
            timeout=5.0
        )
        resp.raise_for_status()
        return resp.json()["job_id"]
 
    def poll(self, job_id: str, poll_interval: float = 1.0, max_wait: float = 120.0) -> dict:
        deadline = time.time() + max_wait
        while time.time() < deadline:
            resp = httpx.get(
                f"{self.base_url}/jobs/{job_id}",
                headers=self.headers,
                timeout=5.0
            )
            resp.raise_for_status()
            data = resp.json()
            if data["status"] in ("completed", "failed"):
                return data
            time.sleep(poll_interval)
        raise TimeoutError(f"Job {job_id} did not complete within {max_wait}s")

Service Contract Testing

Validate that the LLM service behaves as expected after upgrades:

def test_llm_gateway_contract():
    client = LLMGatewayClient(
        base_url="http://llm-gateway:8080",
        api_key="test-key"
    )
    job_id = client.enqueue("Respond with exactly: OK")
    result = client.poll(job_id, max_wait=30.0)
    assert result["status"] == "completed"
    assert "OK" in result["result"], f"Contract violation: {result['result']}"
    print("Contract test passed")

Run contract tests in CI before every model upgrade or gateway deployment.

Key Takeaways

Never call the LLM API synchronously inside user-facing HTTP handlers; use job queues and return a job ID immediately.
Back job queues with Redis, SQS, or RabbitMQ for durability — in-memory queues lose jobs on restart.
Implement exponential-backoff webhook retries (up to 5 attempts) and make callback handlers idempotent.
Isolate the LLM gateway as a dedicated microservice to enable independent scaling and centralized cost tracking.
Circuit breakers should open after 5 consecutive failures and attempt reset after 60 seconds.
Shared client libraries enforce consistent retry, timeout, and auth behavior across all consuming services.
Contract tests run in CI catch model API changes before they reach production.
Monitor queue depth and worker lag — a growing backlog is an early warning of capacity issues.

Introduction#

Synchronous vs. Asynchronous LLM Integration#

Async Job Queue Pattern#

Webhook Callback Delivery#

LLM Service as an Isolated Microservice#

Circuit Breaker for Resilience#

Shared Client Library for Consistent Integration#

Service Contract Testing#

Key Takeaways#