Agent Infrastructure Patterns

Building AI agents that work in production requires thinking beyond the model itself. The infrastructure around your agent determines whether it survives real-world traffic, handles failures gracefully, and scales to meet demand.

The Production Agent Architecture

A production e-commerce AI agent isn't just a prompt and an API call. It's a system:

┌─────────────────────────────────────────────────────┐
│                   Load Balancer                      │
└─────────────────────┬───────────────────────────────┘
                      │
┌─────────────────────▼───────────────────────────────┐
│              API Gateway / Rate Limiter              │
└─────────────────────┬───────────────────────────────┘
                      │
┌─────────────────────▼───────────────────────────────┐
│              Agent Orchestrator Service              │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐   │
│  │   Router    │ │  Context    │ │   Session   │   │
│  │   Agent     │ │  Manager    │ │   Store     │   │
│  └─────────────┘ └─────────────┘ └─────────────┘   │
└─────────────────────┬───────────────────────────────┘
                      │
        ┌─────────────┼─────────────┐
        ▼             ▼             ▼
┌───────────┐  ┌───────────┐  ┌───────────┐
│  Product  │  │  Returns  │  │  Order    │
│   Agent   │  │   Agent   │  │   Agent   │
└─────┬─────┘  └─────┬─────┘  └─────┬─────┘
      │              │              │
      └──────────────┼──────────────┘
                     ▼
┌─────────────────────────────────────────────────────┐
│              LLM Gateway (with fallbacks)            │
│   OpenAI │ Anthropic │ Azure │ Self-hosted          │
└─────────────────────────────────────────────────────┘
    

Pattern 1: Stateless Agent Design

The foundation of scalable agents is statelessness. The agent itself stores no conversation state - everything is externalized.

from redis import Redis
from dataclasses import dataclass
from typing import List, Optional
import json

@dataclass
class ConversationMessage:
    role: str  # "user" or "assistant"
    content: str
    timestamp: float
    metadata: dict

class StatelessEcommerceAgent:
    def __init__(self, llm_client, redis_client: Redis):
        self.llm = llm_client
        self.redis = redis_client
        self.session_ttl = 3600  # 1 hour

    async def handle_message(
        self,
        session_id: str,
        user_message: str,
        user_context: dict
    ) -> str:
        # 1. Load conversation history from external store
        history = await self._get_history(session_id)

        # 2. Build context with relevant data
        context = await self._build_context(user_message, user_context)

        # 3. Call LLM with full context
        response = await self.llm.complete(
            system_prompt=self._get_system_prompt(user_context),
            messages=history + [{"role": "user", "content": user_message}],
            context=context
        )

        # 4. Save updated history
        await self._save_message(session_id, "user", user_message)
        await self._save_message(session_id, "assistant", response)

        return response

    async def _get_history(self, session_id: str) -> List[dict]:
        """Retrieve conversation history from Redis"""
        history_key = f"chat:history:{session_id}"
        messages = self.redis.lrange(history_key, -10, -1)  # Last 10 messages
        return [json.loads(m) for m in messages]

    async def _save_message(self, session_id: str, role: str, content: str):
        """Persist message to Redis with TTL"""
        history_key = f"chat:history:{session_id}"
        message = json.dumps({"role": role, "content": content})
        self.redis.rpush(history_key, message)
        self.redis.expire(history_key, self.session_ttl)

Pattern 2: Circuit Breaker for LLM APIs

LLM APIs fail. Rate limits, outages, and latency spikes are common. Circuit breakers prevent cascading failures:

from enum import Enum
from dataclasses import dataclass
from datetime import datetime, timedelta
import asyncio

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Blocking requests
    HALF_OPEN = "half_open"  # Testing recovery

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout: timedelta = timedelta(seconds=30)
    half_open_max_calls: int = 3

    def __post_init__(self):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.half_open_calls = 0

    async def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if datetime.now() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
            else:
                raise CircuitOpenError("Circuit breaker is open")

        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.half_open_calls += 1
            if self.half_open_calls >= self.half_open_max_calls:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
        self.failure_count = 0

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN


class ResilientLLMClient:
    def __init__(self):
        self.primary = OpenAIClient()
        self.fallback = AnthropicClient()
        self.circuit_breaker = CircuitBreaker(failure_threshold=3)

    async def complete(self, prompt: str) -> str:
        try:
            # Try primary with circuit breaker
            return await self.circuit_breaker.call(
                self.primary.complete, prompt
            )
        except (CircuitOpenError, RateLimitError, TimeoutError):
            # Fallback to secondary provider
            return await self.fallback.complete(prompt)

Pattern 3: Multi-Agent Orchestration

Complex e-commerce interactions often require multiple specialized agents working together:

from typing import Dict, Any, Optional
from enum import Enum

class AgentType(Enum):
    PRODUCT_SEARCH = "product_search"
    ORDER_STATUS = "order_status"
    RETURNS = "returns"
    RECOMMENDATIONS = "recommendations"
    GENERAL = "general"

class AgentOrchestrator:
    def __init__(self):
        self.router = RouterAgent()
        self.agents: Dict[AgentType, BaseAgent] = {
            AgentType.PRODUCT_SEARCH: ProductSearchAgent(),
            AgentType.ORDER_STATUS: OrderStatusAgent(),
            AgentType.RETURNS: ReturnsAgent(),
            AgentType.RECOMMENDATIONS: RecommendationAgent(),
            AgentType.GENERAL: GeneralAgent(),
        }

    async def process(
        self,
        session_id: str,
        user_message: str,
        context: dict
    ) -> AgentResponse:
        # 1. Route to appropriate agent
        intent = await self.router.classify(user_message, context)

        # 2. Get specialized agent
        agent = self.agents.get(intent.agent_type, self.agents[AgentType.GENERAL])

        # 3. Execute with confidence check
        response = await agent.handle(
            session_id=session_id,
            message=user_message,
            context=context,
            intent=intent
        )

        # 4. Post-process and validate
        validated = await self._validate_response(response, intent)

        # 5. Check if handoff needed
        if validated.needs_handoff:
            return await self._handoff_to_agent(
                validated.handoff_target,
                session_id,
                user_message,
                context
            )

        return validated


class RouterAgent:
    """Lightweight agent that classifies intent and routes"""

    ROUTING_PROMPT = """Classify the customer intent. Return JSON:
    {
        "agent_type": "product_search|order_status|returns|recommendations|general",
        "confidence": 0.0-1.0,
        "extracted_entities": {}
    }

    Customer message: {message}
    Recent context: {context}
    """

    async def classify(self, message: str, context: dict) -> Intent:
        # Use smaller/faster model for routing
        response = await self.llm.complete(
            self.ROUTING_PROMPT.format(message=message, context=context),
            model="gpt-4o-mini",  # Fast, cheap for classification
            temperature=0
        )
        return Intent.parse(response)

Pattern 4: Request Queue with Priority

Not all customer queries are equal. VIP customers, checkout-related questions, and time-sensitive issues should be prioritized:

import heapq
from dataclasses import dataclass, field
from typing import Any
import asyncio

@dataclass(order=True)
class PrioritizedRequest:
    priority: int
    request: Any = field(compare=False)
    timestamp: float = field(compare=False)

class PriorityAgentQueue:
    def __init__(self, max_workers: int = 10):
        self.queue: list = []
        self.workers = max_workers
        self.semaphore = asyncio.Semaphore(max_workers)

    def calculate_priority(self, request: AgentRequest) -> int:
        """Lower number = higher priority"""
        base_priority = 100

        # VIP customers get priority
        if request.user.is_vip:
            base_priority -= 30

        # Checkout/cart queries are urgent
        if request.context.get("page") == "checkout":
            base_priority -= 20

        # Order issues during delivery window
        if request.intent == "order_status" and request.has_active_delivery:
            base_priority -= 25

        # Complaints get attention
        if request.sentiment == "negative":
            base_priority -= 15

        return base_priority

    async def enqueue(self, request: AgentRequest):
        priority = self.calculate_priority(request)
        item = PrioritizedRequest(
            priority=priority,
            request=request,
            timestamp=time.time()
        )
        heapq.heappush(self.queue, item)

    async def process_queue(self):
        while True:
            if not self.queue:
                await asyncio.sleep(0.1)
                continue

            async with self.semaphore:
                item = heapq.heappop(self.queue)
                await self.process_request(item.request)

Pattern 5: Graceful Degradation Ladder

When things go wrong, degrade gracefully through multiple fallback levels:

class DegradationLadder:
    """Fallback chain for agent responses"""

    async def get_response(
        self,
        query: str,
        context: dict,
        session_id: str
    ) -> AgentResponse:

        # Level 1: Full AI agent with all capabilities
        try:
            return await self._try_full_agent(query, context)
        except (TimeoutError, RateLimitError):
            pass

        # Level 2: Simplified agent (fewer tools, shorter context)
        try:
            return await self._try_simple_agent(query, context)
        except (TimeoutError, RateLimitError):
            pass

        # Level 3: Cached/templated responses
        cached = await self._try_cached_response(query)
        if cached:
            return cached

        # Level 4: Rule-based fallback
        rule_based = self._try_rule_based(query, context)
        if rule_based:
            return rule_based

        # Level 5: Human handoff
        return await self._escalate_to_human(session_id, query, context)

    async def _try_full_agent(self, query: str, context: dict):
        """Full agent with all tools and long context"""
        async with timeout(8):  # 8 second timeout
            return await self.full_agent.process(
                query=query,
                context=context,
                tools=["search", "orders", "inventory", "returns"],
                max_tokens=1000
            )

    async def _try_simple_agent(self, query: str, context: dict):
        """Simplified agent - faster model, no tools"""
        async with timeout(3):  # 3 second timeout
            return await self.simple_agent.process(
                query=query,
                context=self._truncate_context(context),
                tools=[],  # No tool calls
                max_tokens=300,
                model="gpt-4o-mini"
            )

    def _try_rule_based(self, query: str, context: dict):
        """Pattern matching for common queries"""
        patterns = {
            r"where.*order": "You can track your order at [order tracking link]",
            r"return.*policy": "Our return policy allows returns within 30 days...",
            r"cancel.*order": "To cancel, please visit your orders page or call...",
        }
        for pattern, response in patterns.items():
            if re.search(pattern, query.lower()):
                return AgentResponse(
                    content=response,
                    confidence=0.6,
                    is_fallback=True
                )
        return None

Real-World: Klarna's AI Assistant

Klarna deployed AI assistants handling millions of customer interactions. Their architecture insights:

Key Takeaways

Next up: Scaling Agent Infrastructure - How to handle Black Friday traffic with AI agents.