What Is an AI Agent in Production?
An AI agent in production is a software system that uses one or more machine learning models, reasoning loops, tool integrations, and memory stores to autonomously perform tasks in a live environment — serving real users, processing real data, and operating under real constraints like latency budgets, error budgets, and compliance requirements. Unlike a model endpoint that merely returns a single inference result, an AI agent typically executes a sense → reason → act → observe loop, potentially over many iterations, invoking external APIs, querying databases, and maintaining conversational or task-specific state.
In practice, deploying an AI agent to production means moving beyond a Jupyter notebook or a local prototype into a hardened, observable, scalable, and secure runtime that can handle concurrent users, recover from failures gracefully, and be updated without downtime.
Why Production Deployment Matters
Many AI agents work beautifully in a sandbox but collapse under real-world load. Production deployment forces you to address:
- Latency and throughput — users expect sub-second tool calls or multi-minute background jobs, depending on the use case.
- Cost control — LLM API calls, vector store queries, and tool executions consume budget; production requires throttling, caching, and usage tracking.
- State management — sessions, memory, and intermediate results must persist across restarts and scale horizontally.
- Observability — without tracing, logging, and metrics, debugging a looping agent is nearly impossible.
- Security and sandboxing — agents that execute code, run SQL, or call external tools need strict boundaries.
- Continuous improvement — prompts, models, and tool definitions evolve; you need a deployment pipeline that supports rapid iteration.
Ignoring production concerns leads to runaway costs, broken user experiences, and in the worst case, agents that take irreversible destructive actions. A disciplined deployment approach transforms your agent from a fragile experiment into a reliable software component.
Architecture Overview
A production-grade AI agent stack typically consists of these layers:
- API Gateway / Reverse Proxy — handles authentication, rate limiting, and routing (e.g., Nginx, Cloudflare, AWS API Gateway).
- Agent Runtime Service — the core process that runs the agent loop, manages sessions, and orchestrates tool calls.
- Model Inference Layer — either self-hosted (vLLM, TGI) or managed (OpenAI, Anthropic, Azure, AWS Bedrock).
- Tool Execution Sandbox — a secure environment for code execution, API calls, or database queries.
- State & Memory Store — PostgreSQL, Redis, Pinecone, or Weaviate for session state, episodic memory, and vector search.
- Observability Stack — Prometheus, Grafana, OpenTelemetry, and log aggregation for metrics, traces, and alerts.
- CI/CD Pipeline — automated testing, canary deployments, and rollback mechanisms.
Step-by-Step Deployment Guide
Step 1: Containerize Your Agent
Start by packaging your agent as a Docker container. This ensures consistency between development, staging, and production. Below is a sample Dockerfile for a Python-based agent built with LangChain and FastAPI:
# Dockerfile
FROM python:3.12-slim AS builder
RUN pip install --upgrade pip && \
pip install poetry
WORKDIR /app
COPY pyproject.toml poetry.lock ./
RUN poetry export -f requirements.txt --output requirements.txt
FROM python:3.12-slim AS runtime
WORKDIR /app
COPY --from=builder /app/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY src/ ./src/
COPY config/ ./config/
EXPOSE 8080
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8080"]
The key principle here is the multi-stage build: you export dependencies with Poetry in the builder stage, then copy only the requirements file into the slim runtime image. This keeps the final image small and avoids shipping development tools.
Step 2: Expose Your Agent as a REST API
Wrap your agent loop in an async-capable web framework. FastAPI is an excellent choice for Python agents because it supports streaming responses (SSE) and background tasks natively. Below is a minimal but production-ready API structure:
# src/main.py
import uuid
import asyncio
from typing import AsyncGenerator
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from src.agent import AgentOrchestrator
from src.tools.registry import ToolRegistry
from src.memory.session_store import SessionStore
from src.observability import setup_tracing, log_event
app = FastAPI(title="AI Agent Service", version="1.0.0")
setup_tracing()
agent = AgentOrchestrator()
tool_registry = ToolRegistry()
session_store = SessionStore()
class RunRequest(BaseModel):
session_id: str | None = None
user_input: str
metadata: dict = {}
class RunResponse(BaseModel):
session_id: str
output: str
tool_calls_made: list[str]
tokens_used: int
@app.post("/run", response_model=RunResponse)
async def run_agent(req: RunRequest):
session_id = req.session_id or str(uuid.uuid4())
try:
session = await session_store.get_or_create(session_id)
result = await agent.execute(
session=session,
user_input=req.user_input,
tool_registry=tool_registry,
)
await session_store.update(session_id, result.new_state)
log_event("agent_run_completed", {
"session_id": session_id,
"tokens_used": result.tokens_used,
"tool_calls": result.tool_calls,
})
return RunResponse(
session_id=session_id,
output=result.final_output,
tool_calls_made=result.tool_calls,
tokens_used=result.tokens_used,
)
except Exception as exc:
log_event("agent_run_failed", {"session_id": session_id, "error": str(exc)})
raise HTTPException(status_code=500, detail=str(exc))
@app.post("/run/stream")
async def run_agent_stream(req: RunRequest):
"""Stream agent output as Server-Sent Events."""
session_id = req.session_id or str(uuid.uuid4())
async def event_generator() -> AsyncGenerator[str, None]:
session = await session_store.get_or_create(session_id)
async for chunk in agent.execute_streaming(
session=session,
user_input=req.user_input,
tool_registry=tool_registry,
):
yield f"data: {chunk.model_dump_json()}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"X-Session-Id": session_id},
)
Notice how session management is explicit — each request carries a session_id that links to persistent state. The streaming endpoint uses Server-Sent Events, which is simpler than WebSockets for most agent streaming needs and works well with CDNs and reverse proxies.
Step 3: Implement the Agent Orchestrator
The orchestrator is the brain of your deployment. It manages the reasoning loop, decides when to call tools, and handles retries. Here's a robust implementation pattern:
# src/agent.py
import time
from dataclasses import dataclass, field
from typing import AsyncIterator
from src.llm.client import LLMClient
from src.tools.registry import ToolRegistry
from src.memory.models import Session
from src.observability import trace_span
@dataclass
class AgentResult:
final_output: str
tool_calls: list[str] = field(default_factory=list)
tokens_used: int = 0
new_state: dict | None = None
class AgentOrchestrator:
MAX_ITERATIONS = 10
RECURSION_LIMIT = 3
def __init__(self, llm_client: LLMClient | None = None):
self.llm = llm_client or LLMClient()
async def execute(
self, session: Session, user_input: str, tool_registry: ToolRegistry
) -> AgentResult:
total_tokens = 0
tool_calls_log = []
conversation = session.messages + [{"role": "user", "content": user_input}]
for iteration in range(self.MAX_ITERATIONS):
with trace_span(f"agent_iteration_{iteration}"):
response = await self.llm.chat_completion(
messages=conversation,
tools=tool_registry.schema(),
tool_choice="auto",
)
total_tokens += response.usage.total_tokens
assistant_msg = response.choices[0].message
conversation.append(assistant_msg)
if not assistant_msg.tool_calls:
# Agent decided to respond directly
return AgentResult(
final_output=assistant_msg.content,
tool_calls=tool_calls_log,
tokens_used=total_tokens,
new_state={"messages": conversation},
)
# Execute tool calls in order
for tool_call in assistant_msg.tool_calls:
tool_calls_log.append(tool_call.function.name)
try:
tool_result = await tool_registry.execute(
name=tool_call.function.name,
arguments=tool_call.function.arguments,
session_id=session.id,
)
conversation.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": tool_result,
})
except Exception as e:
conversation.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": f"Error: {str(e)}. Please handle this gracefully.",
})
# Fallback: force final response after max iterations
fallback = await self.llm.chat_completion(
messages=conversation + [{
"role": "system",
"content": "You have reached the maximum number of steps. Summarize what you've done and provide the best answer possible."
}],
tools=[],
)
return AgentResult(
final_output=fallback.choices[0].message.content,
tool_calls=tool_calls_log,
tokens_used=total_tokens + fallback.usage.total_tokens,
new_state={"messages": conversation},
)
async def execute_streaming(
self, session: Session, user_input: str, tool_registry: ToolRegistry
) -> AsyncIterator[dict]:
"""Yields intermediate chunks for SSE streaming."""
conversation = session.messages + [{"role": "user", "content": user_input}]
for iteration in range(self.MAX_ITERATIONS):
yield {"type": "iteration_start", "iteration": iteration}
response = await self.llm.chat_completion(
messages=conversation,
tools=tool_registry.schema(),
tool_choice="auto",
)
assistant_msg = response.choices[0].message
conversation.append(assistant_msg)
if not assistant_msg.tool_calls:
yield {"type": "final_output", "content": assistant_msg.content}
return
for tool_call in assistant_msg.tool_calls:
yield {
"type": "tool_call",
"tool_name": tool_call.function.name,
"arguments": tool_call.function.arguments,
}
tool_result = await tool_registry.execute(
name=tool_call.function.name,
arguments=tool_call.function.arguments,
session_id=session.id,
)
yield {"type": "tool_result", "content": tool_result[:500]}
conversation.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": tool_result,
})
Critical safety features in this orchestrator: a hard MAX_ITERATIONS cap prevents infinite loops, tool execution errors are fed back to the model so it can self-correct, and a fallback summary prompt ensures the agent always returns something useful even if it exceeds its step budget.
Step 4: Build a Secure Tool Execution Layer
Tools are the most dangerous part of an AI agent. A tool that writes files, executes SQL, or makes HTTP requests can cause catastrophic damage if not sandboxed. Here's a production-safe tool registry with built-in validation:
# src/tools/registry.py
import json
import hashlib
import asyncio
from typing import Any
from pydantic import BaseModel, ValidationError
from src.tools.sandbox import SandboxExecutor
from src.tools.approval import ApprovalGate
from src.observability import log_tool_execution
class ToolDefinition(BaseModel):
name: str
description: str
parameters: dict # JSON Schema
handler: callable
requires_approval: bool = False
max_execution_time_seconds: int = 30
rate_limit_per_session: int | None = None
class ToolRegistry:
def __init__(self):
self._tools: dict[str, ToolDefinition] = {}
self._approval_gate = ApprovalGate()
self._sandbox = SandboxExecutor()
def register(self, tool: ToolDefinition) -> None:
self._tools[tool.name] = tool
def schema(self) -> list[dict]:
"""Return OpenAI-compatible function definitions."""
return [{
"type": "function",
"function": {
"name": t.name,
"description": t.description,
"parameters": t.parameters,
}
} for t in self._tools.values()]
async def execute(self, name: str, arguments: str | dict, session_id: str) -> str:
if name not in self._tools:
return f"Error: Tool '{name}' not found. Available: {list(self._tools.keys())}"
tool = self._tools[name]
# Parse and validate arguments
if isinstance(arguments, str):
try:
parsed_args = json.loads(arguments)
except json.JSONDecodeError:
return "Error: Invalid JSON arguments."
else:
parsed_args = arguments
# Approval check for dangerous tools
if tool.requires_approval:
approved = await self._approval_gate.check(
tool_name=name,
arguments=parsed_args,
session_id=session_id,
)
if not approved:
return "Error: This action requires explicit approval and was denied."
# Execute in sandbox with timeout
try:
result = await asyncio.wait_for(
self._sandbox.run(tool.handler, parsed_args, session_id),
timeout=tool.max_execution_time_seconds,
)
log_tool_execution(name, parsed_args, result, session_id)
return str(result)
except asyncio.TimeoutError:
return f"Error: Tool '{name}' exceeded {tool.max_execution_time_seconds}s timeout."
except Exception as e:
return f"Error executing tool '{name}': {type(e).__name__}: {str(e)}"
And here's a sandbox implementation that uses ephemeral Docker containers for code execution tools:
# src/tools/sandbox.py
import tempfile
import subprocess
import os
class SandboxExecutor:
"""Execute tool handlers inside ephemeral Docker containers."""
SANDBOX_IMAGE = "python:3.12-slim"
MEMORY_LIMIT = "256m"
CPU_LIMIT = "0.5"
READ_ONLY_FS = True
async def run(self, handler: callable, arguments: dict, session_id: str) -> Any:
# For native Python tools that don't need sandboxing, run directly
if getattr(handler, '__sandbox_free__', False):
return await handler(arguments, session_id=session_id)
# For code execution tools, use Docker sandbox
script_content = self._generate_script(handler, arguments)
with tempfile.TemporaryDirectory() as tmpdir:
script_path = os.path.join(tmpdir, "exec.py")
with open(script_path, "w") as f:
f.write(script_content)
cmd = [
"docker", "run", "--rm",
"--memory", self.MEMORY_LIMIT,
"--cpus", self.CPU_LIMIT,
"--network", "none" if os.environ.get("DENY_NETWORK", "true") == "true" else "host",
"--read-only" if self.READ_ONLY_FS else "",
"-v", f"{script_path}:/code/exec.py:ro",
"-w", "/code",
self.SANDBOX_IMAGE,
"python", "exec.py",
]
cmd = [c for c in cmd if c] # Remove empty strings
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=30,
)
if result.returncode != 0:
raise RuntimeError(f"Sandbox error: {result.stderr}")
return result.stdout.strip()
def _generate_script(self, handler: callable, arguments: dict) -> str:
# Serialize handler and arguments into a standalone script
# In production, use a pre-built tool library
return f"""
import json
import sys
args = json.loads({json.dumps(arguments)!r})
# Tool handler would be imported from a pre-installed package
# For brevity, we inline the logic here
result = "Tool execution simulated"
print(json.dumps({{"status": "ok", "output": result}}))
"""
Step 5: Add Persistent Memory and Session Storage
Agents need to remember context across invocations. Use a dedicated session store backed by PostgreSQL or Redis:
# src/memory/session_store.py
import json
from datetime import datetime, timezone
from typing import Optional
import asyncpg
from src.memory.models import Session
class SessionStore:
def __init__(self, dsn: str | None = None):
self.dsn = dsn or "postgresql://user:pass@localhost:5432/agent_db"
self._pool: asyncpg.Pool | None = None
async def _get_pool(self) -> asyncpg.Pool:
if self._pool is None:
self._pool = await asyncpg.create_pool(self.dsn, min_size=2, max_size=10)
return self._pool
async def get_or_create(self, session_id: str) -> Session:
pool = await self._get_pool()
async with pool.acquire() as conn:
row = await conn.fetchrow(
"""SELECT messages, metadata, created_at, updated_at
FROM sessions WHERE id = $1""",
session_id,
)
if row:
return Session(
id=session_id,
messages=json.loads(row['messages']),
metadata=json.loads(row['metadata'] or '{}'),
created_at=row['created_at'],
)
else:
now = datetime.now(timezone.utc)
await conn.execute(
"""INSERT INTO sessions (id, messages, metadata, created_at, updated_at)
VALUES ($1, $2, $3, $4, $4)""",
session_id, json.dumps([]), '{}', now,
)
return Session(
id=session_id,
messages=[],
metadata={},
created_at=now,
)
async def update(self, session_id: str, new_state: dict) -> None:
pool = await self._get_pool()
async with pool.acquire() as conn:
await conn.execute(
"""UPDATE sessions
SET messages = $1, metadata = $2, updated_at = $3
WHERE id = $4""",
json.dumps(new_state.get('messages', [])),
json.dumps(new_state.get('metadata', {})),
datetime.now(timezone.utc),
session_id,
)
Create the sessions table with this migration:
-- migrations/001_create_sessions.sql
CREATE TABLE IF NOT EXISTS sessions (
id UUID PRIMARY KEY,
messages JSONB NOT NULL DEFAULT '[]',
metadata JSONB DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_sessions_updated_at ON sessions(updated_at DESC);
-- Auto-cleanup sessions older than 30 days
CREATE OR REPLACE FUNCTION cleanup_old_sessions() RETURNS void AS $$
DELETE FROM sessions WHERE updated_at < NOW() - INTERVAL '30 days';
$$ LANGUAGE SQL;
Step 6: Implement Comprehensive Observability
You cannot debug what you cannot see. Instrument every layer with OpenTelemetry:
# src/observability.py
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider, BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
import structlog
import prometheus_client as prom
from contextlib import contextmanager
# Metrics
agent_runs_total = prom.Counter(
"agent_runs_total", "Total agent executions", ["status"]
)
agent_tokens_used = prom.Histogram(
"agent_tokens_used", "Tokens per agent run",
buckets=[100, 500, 1000, 2000, 5000, 10000, 20000],
)
tool_executions_total = prom.Counter(
"tool_executions_total", "Tool executions", ["tool_name", "status"]
)
agent_iterations = prom.Histogram(
"agent_iterations", "Iterations per run",
buckets=[1, 2, 3, 5, 7, 10],
)
def setup_tracing():
resource = Resource.create({"service.name": "ai-agent-service"})
provider = TracerProvider(resource=resource)
exporter = OTLPSpanExporter(endpoint="http://jaeger:4317", insecure=True)
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
# Auto-instrument frameworks
FastAPIInstrumentor.instrument_app(app)
HTTPXClientInstrumentor().instrument()
logger = structlog.get_logger()
def log_event(event: str, data: dict) -> None:
logger.info(event, **data)
@contextmanager
def trace_span(name: str, attributes: dict | None = None):
span = trace.get_current_span()
if span.is_recording():
span.update_name(name)
if attributes:
span.set_attributes(attributes)
yield
def log_tool_execution(name: str, args: dict, result: str, session_id: str) -> None:
tool_executions_total.labels(tool_name=name, status="success").inc()
logger.info("tool_executed", tool=name, args=args, result=result[:200], session=session_id)
Pair this with a docker-compose snippet that runs Jaeger and Prometheus alongside your agent:
# docker-compose.observability.yml
version: "3.9"
services:
jaeger:
image: jaegertracing/all-in-one:latest
ports:
- "16686:16686" # UI
- "4317:4317" # OTLP gRPC
environment:
- COLLECTOR_OTLP_ENABLED=true
prometheus:
image: prom/prometheus:latest
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
command: ["--config.file=/etc/prometheus/prometheus.yml"]
grafana:
image: grafana/grafana:latest
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
volumes:
- grafana-storage:/var/lib/grafana
volumes:
grafana-storage:
Step 7: Rate Limiting and Cost Control
Uncontrolled agent runs can exhaust your LLM budget. Add middleware that enforces per-user and global limits:
# src/middleware/rate_limit.py
import time
from collections import defaultdict
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
class AgentRateLimiter(BaseHTTPMiddleware):
def __init__(self, app, max_requests_per_minute: int = 30, max_tokens_per_request: int = 4000):
super().__init__(app)
self.max_rpm = max_requests_per_minute
self.max_tokens = max_tokens_per_request
self._buckets: dict[str, list[float]] = defaultdict(list)
async def dispatch(self, request: Request, call_next):
user_id = request.headers.get("X-User-Id", request.client.host)
now = time.time()
# Sliding window rate limit
window_start = now - 60
self._buckets[user_id] = [t for t in self._buckets.get(user_id, []) if t > window_start]
if len(self._buckets[user_id]) >= self.max_rpm:
raise HTTPException(status_code=429, detail="Rate limit exceeded. Retry after 60 seconds.")
self._buckets[user_id].append(now)
response = await call_next(request)
response.headers["X-RateLimit-Remaining"] = str(self.max_rpm - len(self._buckets[user_id]))
return response
For LLM cost tracking, implement a token budget per session:
# src/cost/token_budget.py
class TokenBudgetExceeded(Exception):
pass
class TokenBudgetManager:
def __init__(self, default_budget: int = 20000):
self.default_budget = default_budget
self._session_usage: dict[str, int] = {}
def check(self, session_id: str, tokens_to_add: int) -> None:
current = self._session_usage.get(session_id, 0)
if current + tokens_to_add > self.default_budget:
raise TokenBudgetExceeded(
f"Token budget exceeded for session {session_id}: "
f"used {current}/{self.default_budget}, attempting to add {tokens_to_add}"
)
def record(self, session_id: str, tokens: int) -> None:
self._session_usage[session_id] = self._session_usage.get(session_id, 0) + tokens
def reset(self, session_id: str) -> None:
self._session_usage.pop(session_id, None)
Step 8: Docker Compose for Local Production-like Environment
Before pushing to cloud, run a full local stack that mirrors production:
# docker-compose.yml
version: "3.9"
services:
agent-service:
build: .
ports:
- "8080:8080"
environment:
- DATABASE_URL=postgresql://agent:agentpass@postgres:5432/agent_db
- REDIS_URL=redis://redis:6379
- LLM_API_KEY=${LLM_API_KEY}
- OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger:4317
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_started
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
interval: 30s
timeout: 5s
retries: 3
postgres:
image: pgvector/pgvector:pg16
ports:
- "5432:5432"
environment:
POSTGRES_USER: agent
POSTGRES_PASSWORD: agentpass
POSTGRES_DB: agent_db
volumes:
- pgdata:/var/lib/postgresql/data
- ./migrations:/docker-entrypoint-initdb.d
healthcheck:
test: ["CMD-SHELL", "pg_isready -U agent -d agent_db"]
interval: 5s
timeout: 3s
retries: 5
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
- redisdata:/data
volumes:
pgdata:
redisdata:
Step 9: Kubernetes Deployment (Production Scale)
For multi-replica deployments, Kubernetes provides rolling updates, auto-scaling, and self-healing. Here's a production-grade deployment manifest:
# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: agent-service
namespace: ai-agents
labels:
app: agent-service
spec:
replicas: 3
strategy:
type: RollingUpdate
rollingUpdate:
maxUnavailable: 1
maxSurge: 1
selector:
matchLabels:
app: agent-service
template:
metadata:
labels:
app: agent-service
annotations:
prometheus.io/scrape: "true"
prometheus.io/port: "8080"
spec:
containers:
- name: agent
image: ghcr.io/yourorg/agent-service:1.2.3
ports:
- containerPort: 8080
name: http
env:
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: agent-db-credentials
key: url
- name: LLM_API_KEY
valueFrom:
secretKeyRef:
name: llm-credentials
key: api_key
- name: OTEL_EXPORTER_OTLP_ENDPOINT
value: "http://jaeger-collector.observability:4317"
resources:
requests:
cpu: "500m"
memory: "512Mi"
limits:
cpu: "2000m"
memory: "2Gi"
livenessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 15
periodSeconds: 20
readinessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 5
periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
name: agent-service
namespace: ai-agents
spec:
selector:
app: agent-service
ports:
- port: 80
targetPort: 8080
protocol: TCP
type: ClusterIP
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: agent-service
namespace: ai-agents
annotations:
cert-manager.io/cluster-issuer: letsencrypt-prod
nginx.ingress.kubernetes.io/proxy-read-timeout: "300"
nginx.ingress.kubernetes.io/proxy-send-timeout: "300"
spec:
ingressClassName: nginx
tls:
- hosts:
- agent-api.yourdomain.com
secretName: agent-api-tls
rules:
- host: agent-api.yourdomain.com
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: agent-service
port:
number: 80
The Ingress annotations for proxy timeouts are critical — agent runs can take 30-120 seconds, and the default Nginx 60-second timeout will cut them off. Setting proxy-read-timeout to 300 seconds gives headroom for long-running agent tasks.
Step 10: CI/CD Pipeline with Automated Testing
Agents require specialized testing beyond unit tests. You need evaluation-driven testing that verifies agent behavior against a golden dataset. Here's a GitHub Actions workflow that runs evals before deployment:
# .github/workflows/deploy-agent.yml
name: Deploy AI Agent
on:
push:
branches: [main]
paths:
- 'src/**'
- 'config/**'
- 'evals/**'
jobs:
unit-tests:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.12'
- run: pip install -r requirements-dev.txt
- run: pytest tests/unit/ -v --junitxml=results.xml
eval-tests:
runs-on: ubuntu-latest
needs: unit-tests
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.12'
- run: pip install -r requirements-dev.txt
- run: python evals/run_evals.py --dataset evals/golden_tasks.json --threshold 0.85
env:
LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
build-and-push:
runs-on: ubuntu-latest
needs: eval-tests
steps:
- uses: actions/checkout@v4
- uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- uses: docker/build-push-action@v5
with:
context: .
push: true
tags: ghcr.io/${{ github.repository }}:${{ github.sha }},ghcr.io/${{ github.repository }}:latest
deploy:
runs-on: ubuntu-latest
needs: build-and-push
steps:
- uses: actions/checkout@v4
- name: Deploy to Kubernetes
run: |
kubectl set image deployment/agent-service \
agent=