Building Production AI Agents: Architecture, Deployment, and Scaling

AI agents have moved from research demos to production systems handling real user requests. Building reliable, scalable AI agents requires careful architecture decisions, robust error handling, and cost-effective deployment strategies.

This guide covers everything we’ve learned deploying AI agents at scale — from simple automation scripts to complex multi-agent systems processing thousands of requests daily.

Architecture Patterns

The Agent Loop

Every AI agent follows a fundamental loop:

while not task_complete:
    observation = get_observation()
    action = llm.decide(observation, available_tools)
    result = execute_tool(action)
    update_state(result)

The complexity comes from making this loop reliable, observable, and cost-effective.

Tool Design

Tools are how agents interact with the world. Design them carefully:

from pydantic import BaseModel

class SearchInput(BaseModel):
    query: str
    max_results: int = 10

class SearchTool:
    name = "web_search"
    description = "Search the web for information"
    
    def run(self, input: SearchInput) -> str:
        # Implementation
        return results

Best practices:

• Type-safe inputs: Use Pydantic or similar for validation
• Clear descriptions: The LLM uses these to decide when to use the tool
• Error handling: Return structured errors the LLM can understand
• Idempotency: Design for safe retries

LLM Integration

Provider Abstraction

Abstract your LLM provider to enable switching and fallback:

from abc import ABC, abstractmethod

class LLMProvider(ABC):
    @abstractmethod
    async def complete(self, messages: list, tools: list) -> dict:
        pass

class OpenAIProvider(LLMProvider):
    async def complete(self, messages, tools):
        # OpenAI-specific implementation
        pass

class AnthropicProvider(LLMProvider):
    async def complete(self, messages, tools):
        # Anthropic-specific implementation
        pass

Context Management

Managing context window is crucial for cost and performance:

def manage_context(messages: list, max_tokens: int = 100000) -> list:
    """Keep conversation within token limits."""
    total_tokens = count_tokens(messages)
    
    while total_tokens > max_tokens:
        # Remove oldest non-essential messages
        messages = prune_messages(messages)
        total_tokens = count_tokens(messages)
    
    return messages

Memory and State

Short-term Memory

Store conversation history in a database, not just in-memory:

class ConversationMemory:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def add_message(self, session_id: str, message: dict):
        key = f"conversation:{session_id}"
        self.redis.lpush(key, json.dumps(message))
        self.redis.ltrim(key, 0, 99)  # Keep last 100 messages
    
    def get_history(self, session_id: str) -> list:
        key = f"conversation:{session_id}"
        messages = self.redis.lrange(key, 0, -1)
        return [json.loads(m) for m in reversed(messages)]

Long-term Memory

For persistent knowledge, use vector databases:

from qdrant_client import QdrantClient

class LongTermMemory:
    def __init__(self):
        self.client = QdrantClient("localhost", port=6333)
        self.collection = "agent_memory"
    
    def store(self, text: str, metadata: dict):
        embedding = get_embedding(text)
        self.client.upsert(
            collection_name=self.collection,
            points=[{
                "vector": embedding,
                "payload": {"text": text, **metadata}
            }]
        )
    
    def recall(self, query: str, limit: int = 5) -> list:
        embedding = get_embedding(query)
        results = self.client.search(
            collection_name=self.collection,
            query_vector=embedding,
            limit=limit
        )
        return [r.payload for r in results]

Production Deployment

Containerization

Package your agent as a container:

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Run with proper signal handling
CMD ["python", "-m", "agent.main"]

Scaling Strategies

For high-throughput agents, use async processing:

import asyncio
from aiohttp import web

async def handle_request(request):
    task_id = request.match_info['id']
    
    # Process in background
    asyncio.create_task(process_agent_task(task_id))
    
    return web.json_response({"status": "processing", "task_id": task_id})

app = web.Application()
app.router.add_post('/task/{id}', handle_request)

Rate Limiting and Queues

Protect your LLM budget with rate limiting:

from aiolimiter import AsyncLimiter

# 10 requests per second per user
limiter = AsyncLimiter(10, 1)

async def rate_limited_completion(user_id: str, messages: list):
    async with limiter:
        return await llm.complete(messages)

Cost Optimization

LLM API costs can spiral quickly. Implement these controls:

Token Counting

import tiktoken

def count_tokens(messages: list, model: str = "gpt-4") -> int:
    encoding = tiktoken.encoding_for_model(model)
    total = 0
    for message in messages:
        total += len(encoding.encode(message["content"]))
    return total

Caching

Cache common queries:

import hashlib
import redis

cache = redis.Redis()

def cached_completion(messages: list, ttl: int = 3600):
    key = hashlib.md5(str(messages).encode()).hexdigest()
    
    cached = cache.get(key)
    if cached:
        return json.loads(cached)
    
    result = llm.complete(messages)
    cache.setex(key, ttl, json.dumps(result))
    return result

Model Selection

Use cheaper models for simple tasks:

def select_model(task_complexity: str) -> str:
    if task_complexity == "simple":
        return "gpt-3.5-turbo"
    elif task_complexity == "moderate":
        return "gpt-4-turbo-preview"
    else:
        return "gpt-4"

Monitoring and Observability

Track these metrics:

• Latency: Time from request to completion
• Token usage: Input and output tokens per request
• Tool usage: Which tools are used, success rates
• Error rates: LLM errors, tool errors, timeouts
• Cost per request: Track spending in real-time

Use structured logging:

import structlog

logger = structlog.get_logger()

async def process_task(task_id: str):
    logger.info("task_started", task_id=task_id)
    
    try:
        result = await agent.run(task_id)
        logger.info("task_completed", task_id=task_id, result=result)
    except Exception as e:
        logger.error("task_failed", task_id=task_id, error=str(e))
        raise

Error Handling and Retries

LLM APIs fail in ways traditional services don’t — rate limits, context-length errors, content filter rejections, and the occasional nonsense response that parses fine but means nothing. Your agent loop needs to handle all of these distinctly:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    reraise=True,
)
async def safe_completion(messages: list):
    try:
        return await llm.complete(messages)
    except RateLimitError:
        raise  # Retry — transient
    except ContextLengthError:
        # Don't retry — shrink the context and fail fast
        raise NonRetryableError("context too long")
    except ContentFilterError:
        # Don't retry — log and surface to the user
        raise NonRetryableError("content filtered")

The key distinction: transient errors get retried with backoff, permanent errors fail fast. Retrying a context-length error just burns tokens and time. Log every failure with the request ID so you can trace a bad agent run back through the LLM call chain later.

Testing Agent Behavior

Agents are notoriously hard to test because the LLM is non-deterministic. The practical approach is to test at two levels.

First, mock the LLM for unit tests — your tool logic, routing, and state management should be fully deterministic:

class MockLLM:
    def __init__(self, responses):
        self.responses = iter(responses)
    async def complete(self, messages, tools):
        return next(self.responses)

async def test_agent_uses_search_tool():
    mock = MockLLM([
        {"tool_call": {"name": "web_search", "input": {"query": "test"}}},
        {"content": "The answer is 42"},
    ])
    agent = Agent(llm=mock, tools=[SearchTool()])
    result = await agent.run("what is the answer?")
    assert "42" in result

Second, run evaluation suites against the real model on a fixed set of prompts with expected outcomes. These aren’t pass/fail tests — they’re regression detectors. When you change a prompt or upgrade the model, run the eval suite and compare scores. A drop on “correctly refuses to exfiltrate data” is a red flag you want to catch before production.

Guardrails and Safety

A production agent with tool access can do real damage — send emails, modify databases, spend money. Guardrails belong in code, not just in the prompt:

ALLOWED_ACTIONS = {"web_search", "read_file", "calculate"}
DESTRUCTIVE_ACTIONS = {"send_email", "delete_record", "charge_card"}

def authorize_action(action: str, context: dict) -> bool:
    if action not in ALLOWED_ACTIONS | DESTRUCTIVE_ACTIONS:
        return False
    if action in DESTRUCTIVE_ACTIONS:
        # Require explicit human approval for destructive ops
        return context.get("human_approved", False)
    return True

Prompt-level instructions (“never delete anything”) are suggestions the model might ignore under adversarial input. Code-level authorization is a hard boundary it cannot cross. Treat every tool that mutates state as requiring explicit gating, and log every invocation with inputs for audit.

Conclusion

Building production AI agents requires balancing capability, reliability, and cost. The patterns outlined here — proper tool design, robust memory management, and comprehensive monitoring — form the foundation of scalable agent systems.

Start simple, measure everything, and iterate based on real usage patterns. AI agents are powerful, but only when built with the same engineering rigor as traditional software systems.