Dev Note

AI Agents in Production: A Complete Deployment Guide for 2026
The era of AI agents is no longer a research curiosity — it’s a production reality. In 2026, teams across industries are deploying autonomous AI systems that write code, manage workflows, and interact with external APIs. But shipping agents to production is nothing like shipping traditional software. This guide covers what you need to know.
Photo by Possessed Photography on Unsplash
What Makes Production AI Agents Different?
Traditional software has deterministic paths. AI agents don’t. They reason, plan, and take actions that can cascade across systems. This introduces three fundamental challenges:
Non-determinism — the same input may produce different actions
Tool use risk — agents can call external APIs, write files, send emails
Long-horizon failures — errors compound over multi-step tasks
Production agents need guardrails at every layer.
Architecture Patterns for Production Agents
The Supervisor-Worker Pattern
The most reliable pattern for production is hierarchical:
User Request
    │
    ▼
Supervisor Agent (planning + routing)
    │
    ├──► Code Agent (tool: code interpreter)
    ├──► Search Agent (tool: web search)
    └──► Data Agent (tool: database queries)
The supervisor handles intent classification and task decomposition. Worker agents have limited, scoped tool access. This containment reduces blast radius when something goes wrong.
The Reflection Loop
Before executing any irreversible action (sending email, calling external API, writing to DB), agents should reflect:
class ReflectiveAgent:
    async def execute(self, action: Action) -> Result:
        if action.is_irreversible:
            reflection = await self.reflect(action)
            if reflection.confidence < 0.85:
                return await self.request_human_approval(action)
        return await self._execute(action)
Checkpointing and Recovery
Long-running agents must checkpoint their state:
import json
from pathlib import Path
class CheckpointedAgent:
    def __init__(self, checkpoint_dir: str):
        self.checkpoint_dir = Path(checkpoint_dir)
    
    async def run_step(self, step_id: str, fn, *args):
        checkpoint_path = self.checkpoint_dir / f"{step_id}.json"
        
        if checkpoint_path.exists():
            # Resume from checkpoint
            return json.loads(checkpoint_path.read_text())
        
        result = await fn(*args)
        checkpoint_path.write_text(json.dumps(result))
        return result
Observability: What to Instrument
Trace Every LLM Call
Every LLM call should emit a trace with:
model — which model was used
input_tokens / output_tokens
latency_ms
tool_calls — list of tools invoked
session_id — for linking multi-turn conversations
agent_step — position in the task graph
Using OpenTelemetry with the opentelemetry-instrumentation-openai package:
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
tracer = trace.get_tracer("ai-agent")
async def call_llm(prompt: str, tools: list) -> str:
    with tracer.start_as_current_span("llm_call") as span:
        span.set_attribute("prompt_length", len(prompt))
        span.set_attribute("tool_count", len(tools))
        
        response = await llm.invoke(prompt, tools=tools)
        
        span.set_attribute("output_tokens", response.usage.output_tokens)
        return response.content
Key Metrics to Track
Metric
      Why It Matters
    
Task success rate
      Core reliability signal
    
Steps per task
      Efficiency, cost predictor
    
Tool call failure rate
      External dependency health
    
Human escalation rate
      Agent confidence calibration
    
P95 task latency
      User experience
    
Safety and Guardrails
Input Validation
Never pass raw user input directly to an agent with broad tool access:
from pydantic import BaseModel, validator
class AgentRequest(BaseModel):
    task: str
    max_steps: int = 10
    allowed_tools: list[str] = ["search", "calculator"]
    
    @validator("task")
    def task_must_be_safe(cls, v):
        forbidden = ["rm -rf", "DROP TABLE", "system("]
        for pattern in forbidden:
            if pattern in v:
                raise ValueError(f"Forbidden pattern: {pattern}")
        return v
    
    @validator("max_steps")
    def cap_max_steps(cls, v):
        return min(v, 20)  # hard cap
Tool Permission Scoping
Each agent should have an explicit allowlist:
# agent-config.yaml
agents:
  customer_support_agent:
    allowed_tools:
      - search_knowledge_base
      - create_ticket
      - send_confirmation_email
    denied_tools:
      - delete_customer_data
      - modify_billing
    max_steps: 8
    require_approval:
      - send_confirmation_email  # human in the loop for emails
Rate Limiting and Cost Controls
from functools import lru_cache
import time
class CostGuard:
    def __init__(self, max_tokens_per_hour: int = 500_000):
        self.max_tokens = max_tokens_per_hour
        self.window_start = time.time()
        self.tokens_used = 0
    
    def check(self, estimated_tokens: int) -> bool:
        now = time.time()
        if now - self.window_start > 3600:
            self.tokens_used = 0
            self.window_start = now
        
        if self.tokens_used + estimated_tokens > self.max_tokens:
            raise Exception("Token budget exceeded")
        
        self.tokens_used += estimated_tokens
        return True
Deployment Infrastructure
Containerization
Agents should run in isolated containers with strict resource limits:
FROM python:3.12-slim
# Non-root user
RUN useradd -m -u 1000 agent
USER agent
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY --chown=agent:agent . .
# No network by default — use explicit service mesh rules
ENV PYTHONPATH=/app
CMD ["python", "-m", "agent.main"]
# kubernetes deployment
resources:
  limits:
    memory: "2Gi"
    cpu: "2"
  requests:
    memory: "512Mi"
    cpu: "500m"
Queue-Based Architecture
For production workloads, use a message queue rather than synchronous HTTP:
Client → SQS Queue → Agent Worker (ECS/K8s) → Result Store → Client Polling
This decouples task submission from execution, enables retry logic, and allows horizontal scaling based on queue depth.
Testing AI Agents
Traditional unit tests aren’t enough. You need:
Behavioral Test Suites
import pytest
from agent import CustomerSupportAgent
@pytest.mark.asyncio
async def test_agent_does_not_escalate_simple_queries():
    agent = CustomerSupportAgent()
    result = await agent.run("What are your business hours?")
    
    assert result.resolved is True
    assert result.escalated_to_human is False
    assert result.steps_taken <= 3
@pytest.mark.asyncio
async def test_agent_escalates_billing_disputes():
    agent = CustomerSupportAgent()
    result = await agent.run("You charged me twice and I want a refund NOW")
    
    assert result.escalated_to_human is True
    assert "billing" in result.escalation_reason.lower()
LLM-as-Judge Evaluation
Use a separate LLM to grade agent outputs:
async def evaluate_response(task: str, agent_response: str) -> dict:
    judge_prompt = f"""
    Task: {task}
    Agent Response: {agent_response}
    
    Rate this response on:
    1. Helpfulness (1-5)
    2. Accuracy (1-5)  
    3. Safety (1-5)
    
    Respond in JSON.
    """
    scores = await judge_llm.invoke(judge_prompt)
    return scores
Common Production Pitfalls
1. Infinite Loops
Agents can get stuck in reasoning loops. Always set a hard step limit and implement cycle detection:
async def run(self, task: str) -> Result:
    seen_states = set()
    for step in range(self.max_steps):
        state_hash = hash(str(self.current_state))
        if state_hash in seen_states:
            return Result(error="Loop detected", partial=self.current_result)
        seen_states.add(state_hash)
        await self.step()
2. Context Window Overflow
As conversations grow, agents hit token limits. Implement sliding window memory:
def trim_context(messages: list, max_tokens: int = 8000) -> list:
    total = 0
    trimmed = []
    for msg in reversed(messages):
        tokens = estimate_tokens(msg)
        if total + tokens > max_tokens:
            break
        trimmed.insert(0, msg)
        total += tokens
    return trimmed
3. Tool Call Hallucinations
LLMs sometimes call tools with invalid arguments. Validate every tool call:
def validate_tool_call(tool_name: str, args: dict) -> bool:
    schema = TOOL_SCHEMAS[tool_name]
    try:
        schema(**args)  # pydantic validation
        return True
    except ValidationError as e:
        logger.warning(f"Invalid tool call {tool_name}: {e}")
        return False
Conclusion
Production AI agents require a fundamentally different engineering mindset. The key principles:
Contain, then expand — start with narrow tool access, broaden only with evidence
Observe everything — LLM calls are black boxes; traces are your lifeline
Test behavior, not implementation — agents change; desired outcomes don’t
Budget hard limits — tokens cost money; runaway agents cost more
The teams winning with AI agents in production are the ones who treat them like distributed systems with probabilistic behavior — not magic boxes. Design for failure, observe obsessively, and ship incrementally.
Tags: AI, LLM, Agents, Production, Kubernetes, Observability, LangGraph

AI Agents in Production: A Complete Deployment Guide for 2026

댓글