Building Production AI Agents: Architecture Patterns and Best Practices 2026

Shipping an AI agent demo takes a weekend. Shipping one that’s still running six months later — without silently hallucinating, exhausting your token budget, or locking up on unhandled errors — is a different problem entirely.

This guide is about the second problem. We cover the architectural decisions that separate demos from production systems: memory design, tool registry patterns, error recovery, observability, and deployment topology. Framework-agnostic where possible, with code in Python.

The Production Gap

Most AI agent tutorials stop at “chain a few LLM calls together.” Production agents need:

Deterministic state management — knowing exactly what the agent knows and when
Graceful degradation — failing predictably when tools or the model returns garbage
Cost control — not burning tokens on retries for recoverable errors
Debuggability — a complete trace of every decision the agent made

Every pattern below addresses one or more of these requirements.

1. Memory Architecture

Memory is the most underspecified part of most agent designs. “Give it the conversation history” breaks down fast. Production agents use three distinct memory types.

1.1 Episodic Memory (What Happened)

Short-lived working memory for the current task. Stored in context. Think of it as RAM.

from dataclasses import dataclass, field
from typing import Any

@dataclass
class EpisodicMemory:
    messages: list[dict] = field(default_factory=list)
    tool_results: list[dict] = field(default_factory=list)
    observations: list[str] = field(default_factory=list)

    def add_message(self, role: str, content: str) -> None:
        self.messages.append({"role": role, "content": content})

    def add_tool_result(self, tool_name: str, result: Any, success: bool) -> None:
        self.tool_results.append({
            "tool": tool_name,
            "result": result,
            "success": success,
            "timestamp": __import__("time").time()
        })

    def to_context_window(self) -> list[dict]:
        """Flatten for LLM consumption."""
        context = list(self.messages)
        for tr in self.tool_results:
            context.append({
                "role": "tool",
                "content": f"[{tr['tool']}] {'OK' if tr['success'] else 'ERROR'}: {tr['result']}"
            })
        return context

1.2 Semantic Memory (What It Knows)

Long-term factual knowledge, retrieved via embedding similarity. Backed by a vector store (pgvector, Chroma, Pinecone).

from openai import OpenAI
import chromadb

class SemanticMemory:
    def __init__(self, collection_name: str = "agent_knowledge"):
        self.client = chromadb.PersistentClient(path="./chroma_data")
        self.collection = self.client.get_or_create_collection(collection_name)
        self.embed_client = OpenAI()

    def store(self, content: str, metadata: dict | None = None) -> str:
        embedding = self._embed(content)
        doc_id = __import__("uuid").uuid4().hex
        self.collection.add(
            documents=[content],
            embeddings=[embedding],
            ids=[doc_id],
            metadatas=[metadata or {}]
        )
        return doc_id

    def retrieve(self, query: str, top_k: int = 5) -> list[str]:
        embedding = self._embed(query)
        results = self.collection.query(
            query_embeddings=[embedding],
            n_results=top_k
        )
        return results["documents"][0] if results["documents"] else []

    def _embed(self, text: str) -> list[float]:
        resp = self.embed_client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return resp.data[0].embedding

1.3 Procedural Memory (What It Can Do)

The tool registry — the agent’s “muscle memory” for how to accomplish tasks. More on this in section 3.

Key insight: Use episodic memory aggressively and semantic memory sparingly. Embedding retrieval adds latency and can surface irrelevant context. Only pull from semantic memory when the task explicitly requires it (e.g., “what did we discuss last month about X?“).

2. Agent State Machine

Reactive agents (“just loop until done”) are a support nightmare. Explicit state machines give you observability and control flow for free.

from enum import Enum
from typing import Callable

class AgentState(Enum):
    IDLE = "idle"
    PLANNING = "planning"
    EXECUTING = "executing"
    REFLECTING = "reflecting"
    WAITING_FOR_HUMAN = "waiting_for_human"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class StateTransition:
    from_state: AgentState
    to_state: AgentState
    condition: Callable[[dict], bool]
    action: Callable[[dict], dict]

class AgentStateMachine:
    def __init__(self, transitions: list[StateTransition]):
        self.state = AgentState.IDLE
        self.transitions = transitions
        self.history: list[tuple[AgentState, AgentState]] = []

    def tick(self, context: dict) -> AgentState:
        for transition in self.transitions:
            if transition.from_state == self.state and transition.condition(context):
                new_context = transition.action(context)
                self.history.append((self.state, transition.to_state))
                self.state = transition.to_state
                context.update(new_context)
                break
        return self.state

Wire up transitions explicitly:

transitions = [
    StateTransition(
        from_state=AgentState.IDLE,
        to_state=AgentState.PLANNING,
        condition=lambda ctx: ctx.get("task") is not None,
        action=lambda ctx: {"plan": generate_plan(ctx["task"])}
    ),
    StateTransition(
        from_state=AgentState.PLANNING,
        to_state=AgentState.EXECUTING,
        condition=lambda ctx: ctx.get("plan") is not None,
        action=lambda ctx: {"current_step": ctx["plan"][0]}
    ),
    StateTransition(
        from_state=AgentState.EXECUTING,
        to_state=AgentState.REFLECTING,
        condition=lambda ctx: ctx.get("step_result") is not None,
        action=lambda ctx: {"reflection": reflect_on_result(ctx)}
    ),
    StateTransition(
        from_state=AgentState.REFLECTING,
        to_state=AgentState.COMPLETED,
        condition=lambda ctx: ctx.get("reflection", {}).get("done") is True,
        action=lambda ctx: {"final_result": ctx.get("step_result")}
    ),
    StateTransition(
        from_state=AgentState.REFLECTING,
        to_state=AgentState.EXECUTING,
        condition=lambda ctx: ctx.get("reflection", {}).get("done") is False,
        action=lambda ctx: {"current_step": ctx.get("reflection", {}).get("next_step")}
    ),
]

This pattern makes the agent’s decision process auditable — you can replay any run from history.

3. Tool Registry and Routing

Hardcoding tool lists in system prompts is fine for prototypes. Production agents need a registry that handles capability discovery, schema validation, and fallback routing.

from typing import Any, Callable
from pydantic import BaseModel, ValidationError
import inspect
import json

class ToolSpec(BaseModel):
    name: str
    description: str
    parameters: dict  # JSON Schema
    handler: Callable
    timeout_seconds: float = 30.0
    retry_count: int = 2
    requires_confirmation: bool = False

class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, ToolSpec] = {}

    def register(self, spec: ToolSpec) -> None:
        self._tools[spec.name] = spec

    def get_tool_schemas(self) -> list[dict]:
        """Return OpenAI-compatible tool schemas for the LLM."""
        return [
            {
                "type": "function",
                "function": {
                    "name": name,
                    "description": spec.description,
                    "parameters": spec.parameters
                }
            }
            for name, spec in self._tools.items()
        ]

    async def execute(self, tool_name: str, arguments: dict) -> dict:
        spec = self._tools.get(tool_name)
        if not spec:
            return {"error": f"Unknown tool: {tool_name}", "success": False}

        for attempt in range(spec.retry_count + 1):
            try:
                import asyncio
                result = await asyncio.wait_for(
                    self._call_handler(spec.handler, arguments),
                    timeout=spec.timeout_seconds
                )
                return {"result": result, "success": True}
            except asyncio.TimeoutError:
                return {"error": f"Tool {tool_name} timed out after {spec.timeout_seconds}s", "success": False}
            except Exception as e:
                if attempt == spec.retry_count:
                    return {"error": str(e), "success": False}
                await asyncio.sleep(2 ** attempt)  # exponential backoff

    @staticmethod
    async def _call_handler(handler: Callable, arguments: dict) -> Any:
        if inspect.iscoroutinefunction(handler):
            return await handler(**arguments)
        return handler(**arguments)

Registering tools is then declarative:

registry = ToolRegistry()

registry.register(ToolSpec(
    name="search_web",
    description="Search the web for recent information on a topic.",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "The search query"},
            "max_results": {"type": "integer", "default": 5}
        },
        "required": ["query"]
    },
    handler=web_search_function,
    timeout_seconds=15.0,
    retry_count=1
))

Circuit Breaker for Tools

Wrap the registry with a circuit breaker so a flaky external API doesn’t kill the agent:

from collections import deque
from datetime import datetime, timedelta

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, reset_timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self._failures: deque = deque(maxlen=failure_threshold)
        self._open_since: datetime | None = None

    def is_open(self) -> bool:
        if self._open_since is None:
            return False
        if datetime.now() - self._open_since > timedelta(seconds=self.reset_timeout):
            self._open_since = None
            self._failures.clear()
            return False
        return True

    def record_failure(self) -> None:
        self._failures.append(datetime.now())
        if len(self._failures) >= self.failure_threshold:
            self._open_since = datetime.now()

    def record_success(self) -> None:
        self._failures.clear()
        self._open_since = None

4. Error Recovery Patterns

The naive approach: catch exceptions and retry. The production approach: classify errors and respond differently.

from enum import Enum

class ErrorClass(Enum):
    RETRYABLE = "retryable"          # Network timeout, rate limit
    CORRECTABLE = "correctable"      # Wrong tool arguments (ask LLM to fix)
    FATAL = "fatal"                  # Auth failure, quota exceeded
    HUMAN_REQUIRED = "human_required"  # Ambiguous input, policy violation

def classify_error(error: Exception, context: dict) -> ErrorClass:
    error_str = str(error).lower()

    if "rate limit" in error_str or "timeout" in error_str or "503" in error_str:
        return ErrorClass.RETRYABLE

    if "invalid argument" in error_str or "validation" in error_str:
        return ErrorClass.CORRECTABLE

    if "unauthorized" in error_str or "quota" in error_str:
        return ErrorClass.FATAL

    if context.get("requires_confirmation"):
        return ErrorClass.HUMAN_REQUIRED

    return ErrorClass.RETRYABLE  # default: try again

class ErrorRecoveryHandler:
    async def handle(self, error: Exception, context: dict, retry_fn) -> dict:
        error_class = classify_error(error, context)

        match error_class:
            case ErrorClass.RETRYABLE:
                await asyncio.sleep(context.get("backoff", 1))
                return await retry_fn()

            case ErrorClass.CORRECTABLE:
                # Ask the LLM to fix its own arguments
                corrected_args = await ask_llm_to_fix(error, context)
                return await retry_fn(corrected_args)

            case ErrorClass.FATAL:
                return {"error": str(error), "fatal": True, "success": False}

            case ErrorClass.HUMAN_REQUIRED:
                return {
                    "paused": True,
                    "reason": str(error),
                    "awaiting_input": True,
                    "success": False
                }

5. Observability

You cannot debug what you cannot see. Every agent interaction should emit structured traces.

import structlog
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

log = structlog.get_logger()
tracer = trace.get_tracer("agent")

class ObservableAgent:
    def __init__(self, registry: ToolRegistry):
        self.registry = registry

    async def run_step(self, state: AgentState, context: dict) -> dict:
        with tracer.start_as_current_span(f"agent.step.{state.value}") as span:
            span.set_attribute("agent.state", state.value)
            span.set_attribute("agent.task", context.get("task", "")[:200])

            log.info(
                "agent_step_start",
                state=state.value,
                task=context.get("task", "")[:100]
            )

            try:
                result = await self._execute_step(state, context)
                span.set_attribute("agent.step.success", True)
                log.info("agent_step_complete", state=state.value, result_keys=list(result.keys()))
                return result
            except Exception as e:
                span.record_exception(e)
                span.set_attribute("agent.step.success", False)
                log.error("agent_step_failed", state=state.value, error=str(e))
                raise

    async def call_tool(self, tool_name: str, arguments: dict) -> dict:
        with tracer.start_as_current_span(f"tool.{tool_name}") as span:
            span.set_attribute("tool.name", tool_name)
            span.set_attribute("tool.arguments", json.dumps(arguments)[:500])

            start = __import__("time").monotonic()
            result = await self.registry.execute(tool_name, arguments)
            duration = __import__("time").monotonic() - start

            span.set_attribute("tool.duration_ms", int(duration * 1000))
            span.set_attribute("tool.success", result.get("success", False))

            log.info(
                "tool_call",
                tool=tool_name,
                success=result.get("success"),
                duration_ms=int(duration * 1000)
            )
            return result

Ship traces to Jaeger, Tempo, or Honeycomb. The key metrics to alert on:

Metric	Alert Threshold
`agent.step.duration_p99`	> 30s
`tool.success_rate`	< 95% over 5min
`agent.token_cost_per_task`	> 2x baseline
`error_class.fatal_rate`	> 0

6. Multi-Agent Topology

When a single agent isn’t enough, choose your topology deliberately.

Supervisor Pattern

One orchestrator delegates to specialist agents. Best for tasks with clearly separable subtasks.

class SupervisorAgent:
    def __init__(self, specialists: dict[str, "SpecialistAgent"]):
        self.specialists = specialists
        self.llm = get_llm_client()

    async def route_task(self, task: str) -> str:
        # Ask LLM to classify task into specialist domains
        routing_prompt = f"""Given this task, which specialist should handle it?
Specialists: {list(self.specialists.keys())}
Task: {task}
Respond with just the specialist name."""

        response = await self.llm.complete(routing_prompt)
        specialist_name = response.strip()

        if specialist_name not in self.specialists:
            specialist_name = "general"  # fallback

        return await self.specialists[specialist_name].execute(task)

Peer-to-Peer (Debate) Pattern

Two agents with opposing perspectives reach consensus. Good for decisions requiring risk assessment.

class DebateOrchestrator:
    async def debate(self, proposal: str, rounds: int = 2) -> str:
        proponent = ProponentAgent()
        critic = CriticAgent()

        pro_args = await proponent.argue(proposal)
        con_args = await critic.argue(proposal, pro_args)

        for _ in range(rounds - 1):
            pro_args = await proponent.rebut(con_args)
            con_args = await critic.rebut(pro_args)

        return await self.synthesize(pro_args, con_args)

Map-Reduce Pattern

Fan out work across N parallel agents, aggregate results. Best for large-scale data processing.

import asyncio

async def map_reduce_agent(items: list, process_fn, aggregate_fn):
    # Map phase: process items in parallel
    tasks = [process_fn(item) for item in items]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Filter out failures
    valid_results = [r for r in results if not isinstance(r, Exception)]
    failed_count = len(results) - len(valid_results)

    if failed_count > 0:
        log.warning("map_reduce_partial_failure", failed=failed_count, total=len(items))

    return await aggregate_fn(valid_results)

7. Token Budget Management

Token costs compound fast in multi-agent systems. Track and enforce budgets explicitly.

from dataclasses import dataclass

@dataclass
class TokenBudget:
    total_limit: int = 100_000
    used: int = 0
    per_step_limit: int = 10_000

    def can_proceed(self, estimated_tokens: int) -> bool:
        return (self.used + estimated_tokens) <= self.total_limit

    def record_usage(self, tokens: int) -> None:
        self.used += tokens
        if self.used > self.total_limit * 0.8:
            log.warning(
                "token_budget_warning",
                used=self.used,
                limit=self.total_limit,
                pct=self.used / self.total_limit
            )

    def remaining(self) -> int:
        return max(0, self.total_limit - self.used)

    @property
    def exhausted(self) -> bool:
        return self.used >= self.total_limit

Integrate the budget into every LLM call:

async def llm_call_with_budget(prompt: str, budget: TokenBudget, model: str = "claude-sonnet-4-6") -> str:
    estimated = estimate_tokens(prompt)

    if not budget.can_proceed(estimated):
        raise RuntimeError(f"Token budget exhausted: {budget.used}/{budget.total_limit}")

    response = await call_llm(prompt, model=model)
    actual_tokens = response.usage.total_tokens
    budget.record_usage(actual_tokens)

    return response.content

8. Deployment Patterns

Stateless Workers + External State Store

Never store agent state in memory. Use Redis or a database so any worker can resume any agent run.

import redis.asyncio as redis
import json

class AgentStateStore:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)

    async def save(self, run_id: str, state: dict, ttl_seconds: int = 3600) -> None:
        await self.redis.setex(
            f"agent:run:{run_id}",
            ttl_seconds,
            json.dumps(state)
        )

    async def load(self, run_id: str) -> dict | None:
        data = await self.redis.get(f"agent:run:{run_id}")
        return json.loads(data) if data else None

    async def delete(self, run_id: str) -> None:
        await self.redis.delete(f"agent:run:{run_id}")

Task Queue for Long-Running Agents

Never run agents synchronously in an HTTP request. Queue them.

# Using Celery + Redis
from celery import Celery

app = Celery("agents", broker="redis://localhost:6379/0")

@app.task(bind=True, max_retries=3, acks_late=True)
def run_agent_task(self, run_id: str, task_spec: dict):
    try:
        agent = AgentFactory.create(task_spec["agent_type"])
        result = agent.run_sync(run_id, task_spec)
        return {"run_id": run_id, "status": "completed", "result": result}
    except Exception as exc:
        self.retry(exc=exc, countdown=2 ** self.request.retries)

Health Checks

Expose a health endpoint that checks your LLM provider, vector store, and tool dependencies:

from fastapi import FastAPI

app = FastAPI()

@app.get("/health")
async def health():
    checks = {
        "llm": await check_llm_latency(),
        "vector_store": await check_chroma_connection(),
        "redis": await check_redis_connection(),
    }
    all_healthy = all(v["ok"] for v in checks.values())
    return {"status": "ok" if all_healthy else "degraded", "checks": checks}

Production Readiness Checklist

Before you call an agent “production-ready”:

Memory bounded — episodic memory has a max message count; old messages are summarized, not dropped
All tool calls have timeouts — no hanging goroutines
Circuit breakers on every external dependency
Structured logging with run IDs on every log line
Distributed traces exportable to your observability stack
Token budget enforced — agent stops and reports before hitting limits
State externalized — any worker can resume any run
Human escalation path — HUMAN_REQUIRED errors have a defined destination
Replay possible — you can rerun any failed step with the same inputs

Quick Reference: Pattern Decision Matrix

Problem	Pattern	Key Tool
Long task history	Semantic memory + summarization	pgvector, Chroma
Flaky external APIs	Circuit breaker + retry	Custom or `tenacity`
Parallel subtasks	Map-reduce topology	`asyncio.gather`
Complex decisions	Debate (peer-to-peer)	Two specialized agents
Cost control	Token budget enforcer	Usage tracking middleware
Debuggability	State machine + OTEL tracing	Jaeger/Honeycomb
Horizontal scaling	Stateless workers + Redis	Celery + Redis