Deploying Agents to Production

What is Agent Deployment?

Agent deployment transforms development prototypes into production-ready services. It encompasses containerization, API serving, monitoring, scaling, security, and operational reliability. The goal is five-nines availability with automatic recovery.

The deployment stack typically includes: FastAPI for HTTP serving, Docker for containerization, Redis for caching and sessions, Prometheus/Grafana for monitoring, and Kubernetes or ECS for orchestration.

Why This Matters

A working prototype is not a production system. Production requires handling failures gracefully, scaling under load, monitoring for issues, and recovering from crashes automatically. Without proper deployment, agents fail at the worst possible moments — when users need them most.

Real-World Analogy

Deploying an agent to production is like opening a restaurant. Cooking a great meal (building the agent) is only part of the challenge. You also need a kitchen (infrastructure), waitstaff (API layer), a reservation system (load balancing), health inspections (monitoring), emergency protocols (circuit breakers), and the ability to handle a rush of customers (auto-scaling).

Project Overview

We will deploy an agent with:

FastAPI REST API with async support and health checks
Docker containerization with multi-stage builds
Redis caching and session management
Prometheus metrics and Grafana dashboards
Kubernetes auto-scaling configuration
Circuit breakers and graceful degradation
Structured logging and distributed tracing

Expected outcome: A production deployment template for any agent.

Difficulty: Advanced (requires understanding of DevOps, containerization, and production operations)

Tools & Setup

Tool	Version	Purpose
Python	3.11+	Core language
FastAPI	0.109+	HTTP framework
uvicorn	0.27+	ASGI server
redis	5.0+	Caching
docker	24.0+	Containerization
prometheus-client	0.19+	Metrics

Step 1: FastAPI Application

# app/main.py
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
import redis.asyncio as redis
import logging
import time

logger = logging.getLogger(__name__)

redis_client = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    global redis_client
    redis_client = redis.from_url(
        "redis://localhost:6379",
        decode_responses=True,
        max_connections=20,
    )
    logger.info("Connected to Redis")
    yield
    await redis_client.close()
    logger.info("Disconnected from Redis")


app = FastAPI(
    title="Agent API",
    version="1.0.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)


@app.get("/health")
async def health():
    return {"status": "healthy", "version": "1.0.0", "timestamp": time.time()}


@app.get("/ready")
async def ready():
    try:
        await redis_client.ping()
        return {"status": "ready", "redis": "connected"}
    except Exception:
        raise HTTPException(status_code=503, detail="Service not ready")

Step 2: Routes and Metrics

# app/routes.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from typing import Dict, Optional
import logging
import time

logger = logging.getLogger(__name__)


class QueryRequest(BaseModel):
    query: str = Field(..., min_length=1, max_length=5000)
    user_id: str = Field(default="anonymous", max_length=100)
    session_id: Optional[str] = None


class QueryResponse(BaseModel):
    answer: str
    request_id: str
    latency_ms: float
    tokens_used: int
    model: str


router = APIRouter()


@router.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    from app.agent import ProductionAgent
    agent = ProductionAgent()
    start = time.time()
    try:
        result = await agent.process(request.query, request.user_id)
        latency_ms = (time.time() - start) * 1000
        return QueryResponse(
            answer=result["answer"],
            request_id=result["request_id"],
            latency_ms=round(latency_ms, 2),
            tokens_used=result["tokens_used"],
            model=result["model"],
        )
    except Exception as e:
        logger.error("Query failed: %s", e)
        raise HTTPException(status_code=500, detail=str(e))


# app/metrics.py
from prometheus_client import Counter, Histogram, Gauge
import time
from functools import wraps
from typing import Callable

REQUEST_COUNT = Counter(
    "agent_requests_total",
    "Total requests",
    ["method", "endpoint", "status"],
)
REQUEST_LATENCY = Histogram(
    "agent_request_latency_seconds",
    "Request latency",
    ["endpoint"],
    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0],
)
TOKEN_USAGE = Counter(
    "agent_tokens_used_total",
    "Total tokens used",
    ["model", "type"],
)
ACTIVE_REQUESTS = Gauge(
    "agent_active_requests",
    "Currently active requests",
)


def track_request(func: Callable) -> Callable:
    @wraps(func)
    async def wrapper(*args, **kwargs):
        ACTIVE_REQUESTS.inc()
        start = time.time()
        try:
            result = await func(*args, **kwargs)
            REQUEST_COUNT.labels(method="POST", endpoint="/query", status="success").inc()
            return result
        except Exception as e:
            REQUEST_COUNT.labels(method="POST", endpoint="/query", status="error").inc()
            raise
        finally:
            latency = time.time() - start
            REQUEST_LATENCY.labels(endpoint="/query").observe(latency)
            ACTIVE_REQUESTS.dec()
    return wrapper

Step 3: Production Agent with Caching

# app/agent.py
from openai import AsyncOpenAI
import hashlib
import json
from typing import Dict, Any
import logging
import time

logger = logging.getLogger(__name__)


class ProductionAgent:
    """Production agent with Redis caching and async operations."""

    def __init__(self):
        self.client = AsyncOpenAI()
        self.model = "gpt-4o"

    async def process(self, query: str, user_id: str) -> Dict[str, Any]:
        cache_key = f"query:{hashlib.md5(query.encode()).hexdigest()}"

        cached = await self._get_cache(cache_key)
        if cached:
            logger.info("Cache hit for query: %s", query[:50])
            return cached

        start = time.time()
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": query},
            ],
            temperature=0.7,
            max_tokens=1000,
        )
        latency_ms = (time.time() - start) * 1000

        result = {
            "answer": response.choices[0].message.content,
            "request_id": f"req_{hashlib.md5(query.encode()).hexdigest()[:8]}",
            "latency_ms": round(latency_ms, 2),
            "tokens_used": response.usage.total_tokens,
            "model": self.model,
        }

        await self._set_cache(cache_key, result, ttl=300)
        return result

    async def _get_cache(self, key: str) -> Dict[str, Any] | None:
        try:
            from app.main import redis_client
            data = await redis_client.get(key)
            return json.loads(data) if data else None
        except Exception as e:
            logger.warning("Cache read failed: %s", e)
            return None

    async def _set_cache(self, key: str, value: Dict[str, Any], ttl: int = 300) -> None:
        try:
            from app.main import redis_client
            await redis_client.setex(key, ttl, json.dumps(value))
        except Exception as e:
            logger.warning("Cache write failed: %s", e)

Step 4: Docker and Kubernetes

# Dockerfile
FROM python:3.11-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY . .
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=5s --retries=3 CMD curl -f http://localhost:8000/health || exit 1
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# docker-compose.yml
version: '3.8'
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - REDIS_URL=redis://redis:6379
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    depends_on:
      - redis
    deploy:
      replicas: 3
      resources:
        limits:
          memory: 512M
          cpus: "0.5"
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
volumes:
  redis_data:

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: agent-api
  template:
    metadata:
      labels:
        app: agent-api
    spec:
      containers:
      - name: agent-api
        image: agent-api:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 3
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: agent-api-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: agent-api
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70

Why This Matters

The gap between a working prototype and a production system is enormous. Production systems need to handle failures, scale under load, maintain security, and provide observability. Without proper deployment, agents crash under real-world conditions.

Real-World Analogy

Production deployment is like building a highway system. A dirt road (prototype) works for one car, but a highway needs multiple lanes (scaling), traffic lights (load balancing), guard rails (circuit breakers), emergency services (health checks), and monitoring cameras (observability).

Mathematical Foundation

Capacity Planning:

Intuition: Number of pods needed to handle expected load. For 100 RPS at 200ms latency at 70% utilization: pods.

Auto-scaling Threshold:

Intuition: Scale up when CPU exceeds 120% of target to handle increasing load with headroom.

Performance Considerations

Metric	Value	Notes
Request Latency	200ms-2s	Depends on LLM
Throughput	100+ RPS	Per pod
Cache Hit Rate	30-60%	Depends on query patterns
Memory Usage	256-512MB	Per pod
Startup Time	5-10s	Cold start
Health Check Interval	30s	Configurable
Auto-scaling Delay	60-90s	Time to add new pods

Security Considerations

API Key Authentication: Require API keys for all non-health endpoints
Rate Limiting: Implement per-user and per-IP rate limits
TLS Encryption: Always use HTTPS in production
Secrets Management: Use Kubernetes secrets or Vault, never commit keys
CORS Restrictions: Allow only specific origins
Network Policies: Restrict pod-to-pod communication
Container Security: Use non-root users, scan images for vulnerabilities

Testing & Evaluation

import pytest
from httpx import AsyncClient
from app.main import app


@pytest.mark.asyncio
async def test_health():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/health")
        assert response.status_code == 200
        assert response.json()["status"] == "healthy"


@pytest.mark.asyncio
async def test_query():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post("/query", json={"query": "test"})
        assert response.status_code == 200

Interview Q&A

Q1: What is the purpose of health checks in production deployments? A: Liveness probes detect deadlocks and restart unhealthy containers. Readiness probes ensure traffic only reaches ready pods. Together they enable automatic recovery without manual intervention, achieving self-healing infrastructure.

Q2: How does Redis caching reduce LLM API costs? A: Redis caches identical or semantically similar query responses, avoiding redundant LLM API calls. For repeat queries, cache hits reduce latency from 2s to 10ms and eliminate token costs entirely.

Q3: What is the difference between horizontal and vertical scaling? A: Horizontal scaling adds more pods (scale out), improving fault tolerance and throughput. Vertical scaling increases resources per pod (scale up), improving single-instance performance. Horizontal is preferred for LLM agents due to stateless nature.

Q4: How do you handle cold starts in serverless deployments? A: Use provisioned concurrency to keep warm instances, implement connection pooling for databases, preload model weights, use container pre-warming, and design for graceful degradation during cold starts.

Q5: What is a circuit breaker and when should it be used? A: A circuit breaker stops calling failing services after a threshold, returning cached/fallback responses. It prevents cascade failures in microservice architectures. Use when external services (LLM APIs, databases) may fail.

Q6: How would you implement zero-downtime deployments? A: Use rolling updates with readiness probes, implement blue-green deployments, use canary releases for gradual traffic shifting, ensure graceful shutdown handling, and maintain backward compatibility.

Q7: What metrics should be monitored in production? A: Request latency (P50, P95, P99), error rate, throughput (RPS), token usage and cost, cache hit rate, queue depth, CPU/memory utilization, and external API latency.

Q8: How do you secure agent deployments? A: Use API key authentication, implement rate limiting per user/IP, encrypt data in transit (TLS) and at rest, use secrets management (Vault), enable CORS restrictions, and audit all requests.

Common Pitfalls & Solutions

Pitfall	Impact	Solution
Cold start latency	Slow initial responses	Pre-warm containers, use provisioned concurrency
Memory leaks	Container crashes	Monitor memory, implement restart policies, use leak detectors
Rate limiting	Service disruption	Implement per-user quotas and API key management
Cost overruns	Budget exceeded	Set spending limits, monitor token usage, implement caching
Single point of failure	Complete outage	Multi-AZ deployment, redundant components, circuit breakers
Secret exposure	Security breach	Use secrets management (Vault, K8s secrets), never commit keys
Configuration drift	Inconsistent environments	Use infrastructure-as-code (Terraform, Helm)
No observability	Blind to issues	Implement structured logging, metrics, and distributed tracing

Summary with Key Takeaways

FastAPI provides high-performance async HTTP serving for production agents
Docker ensures consistent environments across development and production
Redis caching reduces LLM costs and improves latency for repeat queries
Kubernetes auto-scaling handles traffic spikes with horizontal pod autoscalers
Comprehensive monitoring (Prometheus + Grafana) enables proactive issue detection
Health checks, circuit breakers, and graceful degradation ensure reliability
Zero-downtime deployments require rolling updates and readiness probes

Deploying Agents to Production

Deploying Agents to Production

What is Agent Deployment?

Why This Matters

Real-World Analogy

Project Overview

Tools & Setup

Step 1: FastAPI Application

Step 2: Routes and Metrics

Step 3: Production Agent with Caching

Step 4: Docker and Kubernetes

Why This Matters

Real-World Analogy

Mathematical Foundation

Performance Considerations

Security Considerations

Testing & Evaluation

Interview Q&A

Common Pitfalls & Solutions

Summary with Key Takeaways

Need Expert AI Agents Help?