The Gap Between Demo and Production

Your agent works on your laptop. It handles 10 test cases. Your demo impresses the team. Then you deploy it, and within a week: it hallucinated a refund policy that doesn't exist, looped for 47 steps on a simple query costing $12, and sent a customer an email with confidential pricing data.

Welcome to production AI.

72% of enterprises have deployed multi-agent systems — and nearly half have had a Severity 1 incident involving hallucinated data or unauthorized autonomous behavior. This lesson is about being in the other half.

Guardrails: The Defense-in-Depth Model

Production agents need three layers of guardrails. Not one. Not two. Three. Because each layer catches what the others miss.

Layer 1: Input Guardrails

Validate and sanitize everything that enters the agent before the LLM sees it:

from guardrails import Guard
from guardrails.hub import ToxicLanguage, PromptInjection

# Guardrails AI library for structured validation
input_guard = Guard().use_many(
    ToxicLanguage(on_fail="exception"),
    PromptInjection(on_fail="exception")
)

def validate_input(user_message: str) -> str:
    """Validate user input before passing to agent."""
    # Length limit
    if len(user_message) > 5000:
        raise ValueError("Input too long (max 5000 chars)")

    # Guardrails AI checks
    try:
        result = input_guard.validate(user_message)
        return result.validated_output
    except Exception as e:
        raise ValueError(f"Input rejected: {str(e)}")

# Custom prompt injection detection
def detect_injection(text: str) -> bool:
    """Detect common prompt injection patterns."""
    patterns = [
        "ignore previous instructions",
        "ignore all prior",
        "disregard your instructions",
        "you are now",
        "new system prompt",
        "override your",
        "forget everything"
    ]
    text_lower = text.lower()
    return any(p in text_lower for p in patterns)

Layer 2: Execution Guardrails

Control what the agent can do during execution:

import time
from dataclasses import dataclass, field

@dataclass
class ExecutionLimits:
    """Hard limits on agent execution."""
    max_steps: int = 20
    max_tool_calls: int = 15
    max_tokens_total: int = 100_000
    max_cost_usd: float = 0.50
    max_time_seconds: int = 120
    banned_tools: list[str] = field(default_factory=lambda: ["delete_user", "drop_table"])

class ExecutionGuard:
    def __init__(self, limits: ExecutionLimits):
        self.limits = limits
        self.step_count = 0
        self.tool_calls = 0
        self.tokens_used = 0
        self.cost_usd = 0.0
        self.start_time = time.time()

    def check(self, tool_name: str = None, tokens: int = 0) -> None:
        """Check all limits. Raises if any are exceeded."""
        self.step_count += 1
        self.tokens_used += tokens
        self.cost_usd = self.tokens_used * 0.000003  # ~GPT-4o input pricing

        if tool_name:
            self.tool_calls += 1
            if tool_name in self.limits.banned_tools:
                raise PermissionError(f"Tool '{tool_name}' is banned")

        elapsed = time.time() - self.start_time

        violations = []
        if self.step_count > self.limits.max_steps:
            violations.append(f"Max steps ({self.limits.max_steps}) exceeded")
        if self.tool_calls > self.limits.max_tool_calls:
            violations.append(f"Max tool calls ({self.limits.max_tool_calls}) exceeded")
        if self.cost_usd > self.limits.max_cost_usd:
            violations.append(f"Max cost (${self.limits.max_cost_usd}) exceeded")
        if elapsed > self.limits.max_time_seconds:
            violations.append(f"Max time ({self.limits.max_time_seconds}s) exceeded")

        if violations:
            raise RuntimeError(f"Execution limits exceeded: {'; '.join(violations)}")

Layer 3: Output Guardrails

Validate the agent's final output before it reaches the user:

from guardrails import Guard
from guardrails.hub import RestrictToTopic, DetectPII

output_guard = Guard().use_many(
    RestrictToTopic(
        valid_topics=["technology", "ai", "software"],
        invalid_topics=["politics", "religion", "medical advice"],
        on_fail="reask"  # Ask the LLM to regenerate
    ),
    DetectPII(on_fail="fix")  # Automatically redact PII
)

def validate_output(agent_response: str) -> str:
    """Validate and fix agent output before returning to user."""
    result = output_guard.validate(agent_response)
    return result.validated_output

Monitoring: Seeing What Your Agent Actually Does

You can't fix what you can't see. Production agents need observability across four dimensions:

1. Operational Metrics

import structlog
from prometheus_client import Counter, Histogram, Gauge

logger = structlog.get_logger()

# Prometheus metrics
agent_runs_total = Counter("agent_runs_total", "Total agent runs", ["status"])
agent_duration = Histogram("agent_duration_seconds", "Agent run duration")
agent_cost = Histogram("agent_cost_usd", "Agent run cost in USD")
agent_steps = Histogram("agent_steps_total", "Steps per agent run")
active_agents = Gauge("active_agents", "Currently running agents")

class MonitoredAgent:
    def __init__(self, agent_app, limits: ExecutionLimits):
        self.app = agent_app
        self.limits = limits

    async def run(self, input_message: str, user_id: str) -> str:
        guard = ExecutionGuard(self.limits)
        active_agents.inc()
        start = time.time()

        try:
            result = await self.app.ainvoke(
                {"messages": [HumanMessage(content=input_message)]},
                config={"recursion_limit": self.limits.max_steps}
            )
            agent_runs_total.labels(status="success").inc()
            output = result["messages"][-1].content
            return validate_output(output)

        except RuntimeError as e:
            agent_runs_total.labels(status="limit_exceeded").inc()
            logger.error("agent_limit_exceeded", user_id=user_id, error=str(e))
            return "I wasn't able to complete this task within the allowed limits. Please try a simpler request."

        except Exception as e:
            agent_runs_total.labels(status="error").inc()
            logger.error("agent_error", user_id=user_id, error=str(e))
            return "Something went wrong. Our team has been notified."

        finally:
            duration = time.time() - start
            agent_duration.observe(duration)
            agent_cost.observe(guard.cost_usd)
            agent_steps.observe(guard.step_count)
            active_agents.dec()

            logger.info("agent_run_complete",
                user_id=user_id,
                steps=guard.step_count,
                tool_calls=guard.tool_calls,
                cost_usd=round(guard.cost_usd, 4),
                duration_seconds=round(duration, 2)
            )

2. Quality Metrics

Track output quality over time using automated evaluation:

import random

async def sample_and_evaluate(agent_output: str, user_query: str):
    """Evaluate a random sample of agent outputs."""
    # Evaluate 10% of responses
    if random.random() > 0.10:
        return

    eval_prompt = f"""Rate this AI agent response on a 1-5 scale for:
    - Accuracy (factually correct)
    - Relevance (answers the question)
    - Completeness (thorough answer)

    Query: {user_query}
    Response: {agent_output[:2000]}

    Return JSON: {{"accuracy": N, "relevance": N, "completeness": N}}"""

    scores = await evaluator.ainvoke([HumanMessage(content=eval_prompt)])
    # Log to your metrics system
    logger.info("quality_eval", scores=scores.content, query=user_query[:100])

Cost Control: The Part Nobody Tells You About

Agent costs scale non-linearly. A simple query costs $0.01. A complex one costs $0.50. A confused agent in a loop costs $10+. At scale, this adds up fast.

Strategy 1: Tiered Model Selection

def select_model(task_complexity: str) -> str:
    """Route tasks to appropriately-priced models."""
    models = {
        "simple": "gpt-4o-mini",    # $0.15/1M input tokens
        "medium": "gpt-4o",          # $2.50/1M input tokens
        "complex": "claude-sonnet-4", # For hardest tasks
    }
    return models.get(task_complexity, "gpt-4o-mini")

# Classify complexity before running the agent
classifier_prompt = """Classify this task as simple, medium, or complex:
- Simple: single fact lookup, yes/no question, basic formatting
- Medium: multi-step research, comparison, analysis
- Complex: multi-source synthesis, code generation, planning

Task: {task}
Classification (one word):"""

# Use the cheapest model for classification itself
complexity = await cheap_llm.ainvoke(classifier_prompt.format(task=user_input))
model = select_model(complexity.content.strip().lower())

Strategy 2: Caching

import hashlib
import json
from functools import lru_cache

# Cache tool results to avoid redundant API calls
tool_cache: dict[str, str] = {}

def cached_tool_call(tool_name: str, args: dict) -> str | None:
    """Check if we've already called this tool with these args."""
    cache_key = hashlib.sha256(
        f"{tool_name}:{json.dumps(args, sort_keys=True)}".encode()
    ).hexdigest()
    return tool_cache.get(cache_key)

def cache_tool_result(tool_name: str, args: dict, result: str):
    """Cache a tool result for future use."""
    cache_key = hashlib.sha256(
        f"{tool_name}:{json.dumps(args, sort_keys=True)}".encode()
    ).hexdigest()
    tool_cache[cache_key] = result

# In production, use Redis with TTL:
# redis_client.setex(cache_key, 3600, result)  # 1-hour cache

Strategy 3: Budget Alerts

from dataclasses import dataclass

@dataclass
class CostTracker:
    daily_budget_usd: float = 50.0
    daily_spent_usd: float = 0.0
    per_user_limit_usd: float = 2.0
    user_spent: dict = field(default_factory=dict)

    def can_run(self, user_id: str, estimated_cost: float) -> bool:
        """Check if this run is within budget."""
        if self.daily_spent_usd + estimated_cost > self.daily_budget_usd:
            logger.warning("daily_budget_approaching", spent=self.daily_spent_usd)
            return False
        user_total = self.user_spent.get(user_id, 0) + estimated_cost
        if user_total > self.per_user_limit_usd:
            logger.warning("user_budget_exceeded", user_id=user_id, spent=user_total)
            return False
        return True

    def record(self, user_id: str, cost: float):
        self.daily_spent_usd += cost
        self.user_spent[user_id] = self.user_spent.get(user_id, 0) + cost

Deployment Architecture

A production agent deployment typically looks like this:

┌──────────┐     ┌─────────────┐     ┌──────────────┐
│  Client   │────▶│  API Gateway │────▶│ Input Guards │
│  (Web/    │     │  (FastAPI)   │     │  - Length    │
│   Mobile) │     │              │     │  - Injection │
└──────────┘     └─────────────┘     │  - Toxicity  │
                                      └──────┬───────┘
                                             │
                                      ┌──────▼───────┐
                                      │  Agent Runner │
                                      │  - Model sel  │
                                      │  - Exec guard │
                                      │  - Cost track │
                                      └──────┬───────┘
                                             │
                              ┌──────────────┼──────────────┐
                              │              │              │
                       ┌──────▼──┐    ┌──────▼──┐    ┌─────▼───┐
                       │  Tools  │    │  Memory │    │  LLM    │
                       │  (MCP)  │    │  (Redis │    │  (GPT/  │
                       │         │    │  +Vector│    │  Claude)│
                       └─────────┘    └─────────┘    └─────────┘
                                             │
                                      ┌──────▼───────┐
                                      │ Output Guards│
                                      │  - Topic     │
                                      │  - PII       │
                                      │  - Quality   │
                                      └──────┬───────┘
                                             │
                                      ┌──────▼───────┐
                                      │  Monitoring  │
                                      │  - Metrics   │
                                      │  - Traces    │
                                      │  - Alerts    │
                                      └──────────────┘

The Production Checklist

Before deploying any agent to production, verify:

Input guardrails: Injection detection, length limits, toxicity filtering
Execution limits: Max steps, max cost, max time, banned tools
Output guardrails: Topic restriction, PII detection, quality validation
Monitoring: Latency, cost, error rate, quality scores — all dashboarded
Cost controls: Per-user limits, daily budgets, model tiering, caching
Error handling: Graceful degradation, user-friendly error messages, automatic alerts
Logging: Full execution traces (LangSmith or equivalent) for debugging
Testing: 20+ test cases, run 3x each, 90%+ pass rate

Miss any of these, and your first production incident is a matter of when, not if.

This is the final lesson. You now have everything you need to build, test, and deploy production AI agents. The frameworks will keep evolving — LangGraph, CrewAI, and whatever comes next — but the patterns in this course are durable. Tools, memory, planning, guardrails, monitoring, cost control. Master those, and you can build with any framework.

For the latest AI tools, frameworks, and agent infrastructure reviews, visit tools.skila.ai and repos.skila.ai.

Production Deployment — Guardrails, Monitoring, and Cost Control

Related Resources