Production Checklist and Deployment

Capsule overview

This is the pre-flight checklist before you deploy your AI agent to production. It isn't a generic software checklist — it's specific to LLM-powered systems, with their unique challenges: models can generate wrong answers, costs scale with usage, provider APIs can go down without warning, and responses can contain sensitive or harmful information.

In the previous capsules of this module, you learned each individual piece: tracing with LangSmith (02), visual debugging (03), automated evaluation (04), token tracking (05), and rate limiting (06). Now you're going to integrate all of it into a checklist that tells you "yes, your system is production-ready" or "no, you're missing X before you deploy."

The difference between an agent that "works" and an agent that's "production-ready" is confidence. With tracing, you know what it did. With evaluation, you know whether it did it well. With cost control, you know what it cost. With this checklist, you know you covered every angle. That's what lets you deploy and sleep at night.

The checklist: 6 categories

The production checklist has 6 categories. Each one is an area you must cover before deploying. They're not optional — every single one exists because a real incident motivated it.

Production Readiness Score:

 ✅ 1. Configuration Management  [  /4 ]
 ✅ 2. Error Monitoring           [  /4 ]
 ✅ 3. Safety & Compliance        [  /4 ]
 ✅ 4. Cost Control               [  /4 ]
 ✅ 5. Observability              [  /4 ]
 ✅ 6. Resilience                 [  /4 ]

                          Total: [  /24 ]

1. Configuration Management

LLM APIs are external services. Credentials, model versions, and prompt configuration have to be managed carefully.

1.1 API keys in environment variables

Never in code. Never in commits. Never in logs.

from dotenv import load_dotenv
load_dotenv()

import os

REQUIRED_ENV_VARS = [
    "OPENAI_API_KEY",
    "LANGSMITH_API_KEY",
    "LANGSMITH_TRACING",
]

def verify_env_vars():
    """Check that every required environment variable is configured."""
    missing = []
    masked = []

    for var in REQUIRED_ENV_VARS:
        value = os.getenv(var)
        if not value:
            missing.append(var)
        else:
            if "KEY" in var or "SECRET" in var:
                masked.append(f"  {var}: {'*' * 8}...{value[-4:]}")
            else:
                masked.append(f"  {var}: {value}")

    if missing:
        print(f"MISSING environment variables:")
        for var in missing:
            print(f"  ❌ {var}")
        return False

    print("Environment variables OK:")
    for line in masked:
        print(f"  ✅ {line}")
    return True

result = verify_env_vars()
print(f"\nConfiguration check: {'PASS' if result else 'FAIL'}")
# Expected output:
# Environment variables OK:
#   ✅   OPENAI_API_KEY: ********...a1b2
#   ✅   LANGSMITH_API_KEY: ********...c3d4
#   ✅   LANGSMITH_TRACING: true
#
# Configuration check: PASS

1.2 Pinned model versions

Never use "latest". If OpenAI updates GPT-4.1 and the behavior changes, your agent breaks without you changing a thing.

from dotenv import load_dotenv
load_dotenv()

MODEL_CONFIG = {
    "researcher": {
        "model": "openai:gpt-4.1-mini",
        "temperature": 0.2,
        "max_tokens": 1000,
    },
    "analyst": {
        "model": "openai:gpt-4.1",
        "temperature": 0.1,
        "max_tokens": 2000,
    },
    "writer": {
        "model": "openai:gpt-4.1-mini",
        "temperature": 0.3,
        "max_tokens": 1500,
    },
}

def verify_model_config():
    """Check that the models are configured correctly."""
    issues = []
    for agent_name, config in MODEL_CONFIG.items():
        model_id = config["model"]

        if "latest" in model_id.lower():
            issues.append(f"  ❌ {agent_name}: uses 'latest' — pin to specific version")

        if config["temperature"] > 0.5:
            issues.append(f"  ⚠️ {agent_name}: high temperature ({config['temperature']}) — may produce inconsistent output")

        if config.get("max_tokens", 0) > 4000:
            issues.append(f"  ⚠️ {agent_name}: high max_tokens ({config['max_tokens']}) — increases cost")

        if not issues or not any(agent_name in i for i in issues):
            print(f"  ✅ {agent_name}: {model_id} (temp={config['temperature']})")

    if issues:
        print("\nIssues found:")
        for issue in issues:
            print(issue)
    else:
        print("\nAll models properly configured.")

print("Model Configuration:")
verify_model_config()
# Expected output:
# Model Configuration:
#   ✅ researcher: openai:gpt-4.1-mini (temp=0.2)
#   ✅ analyst: openai:gpt-4.1 (temp=0.1)
#   ✅ writer: openai:gpt-4.1-mini (temp=0.3)
#
# All models properly configured.

1.3 Prompt versions tracked

Prompts are code. They have to be versioned like code. A prompt change can completely change the agent's behavior.

from dotenv import load_dotenv
load_dotenv()

from dataclasses import dataclass
from datetime import datetime

@dataclass
class PromptVersion:
    name: str
    version: str
    content: str
    created_at: str
    description: str

PROMPT_REGISTRY = {
    "researcher_system": PromptVersion(
        name="researcher_system",
        version="1.2.0",
        content="You are a specialized researcher. Your only job is to find relevant information. Do NOT analyze or write reports.",
        created_at="2026-03-01",
        description="Added explicit constraint to NOT analyze",
    ),
    "analyst_system": PromptVersion(
        name="analyst_system",
        version="2.0.1",
        content="You are a research analyst. You identify patterns, contradictions, and trends. Respond with numbered findings.",
        created_at="2026-03-05",
        description="Changed output format to numbered findings",
    ),
    "writer_system": PromptVersion(
        name="writer_system",
        version="1.1.0",
        content="You are a technical writer. You produce clear, concise reports. Use bullet points and executive summaries.",
        created_at="2026-02-28",
        description="Added executive summary requirement",
    ),
}

def verify_prompts():
    """Check that every prompt is versioned."""
    print("Prompt Registry:")
    for name, prompt in PROMPT_REGISTRY.items():
        print(f"  ✅ {name} v{prompt.version} ({prompt.created_at})")
        print(f"     {prompt.description}")

verify_prompts()
# Expected output:
# Prompt Registry:
#   ✅ researcher_system v1.2.0 (2026-03-01)
#      Added explicit constraint to NOT analyze
#   ✅ analyst_system v2.0.1 (2026-03-05)
#      Changed output format to numbered findings
#   ✅ writer_system v1.1.0 (2026-02-28)
#      Added executive summary requirement

1.4 Secrets not exposed in logs or traces

LangSmith captures prompts and responses in traces. If your prompt includes sensitive data, that data ends up in LangSmith.

from dotenv import load_dotenv
load_dotenv()

import os
import re

def check_for_secrets_in_text(text: str) -> list[str]:
    """Detect possible secrets in text that could reach logs or traces."""
    patterns = [
        (r'sk-[a-zA-Z0-9]{20,}', "OpenAI API key"),
        (r'lsv2_[a-zA-Z0-9]{20,}', "LangSmith API key"),
        (r'password\s*[=:]\s*["\'][^"\']+["\']', "Hardcoded password"),
        (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', "Email address"),
        (r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', "Phone number"),
    ]

    found = []
    for pattern, description in patterns:
        if re.search(pattern, text):
            found.append(description)
    return found

test_prompts = [
    "Analyze the impact of AI on education.",
    "My API key is sk-abc123def456ghi789jklmnopqrst. Use it to search.",
    "Contact user@example.com for more information.",
]

print("Secret Detection Check:")
for prompt in test_prompts:
    secrets = check_for_secrets_in_text(prompt)
    if secrets:
        print(f"  ❌ '{prompt[:50]}...'")
        for s in secrets:
            print(f"     Found: {s}")
    else:
        print(f"  ✅ '{prompt[:50]}...' — clean")
# Expected output:
# Secret Detection Check:
#   ✅ 'Analyze the impact of AI on education....' — clean
#   ❌ 'My API key is sk-abc123def456ghi789jklmnopqrst. ...'
#      Found: OpenAI API key
#   ❌ 'Contact user@example.com for more information....'
#      Found: Email address

2. Error Monitoring

LLMs fail in ways traditional APIs don't: long timeouts, partial responses, invalid JSON, hallucinations, and provider rate limits.

2.1 Fallback providers

If OpenAI goes down, does your agent stop working? With a fallback, it switches automatically to Anthropic or another provider.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

def create_model_with_fallback(
    primary: str = "openai:gpt-4.1-mini",
    fallback: str = "openai:gpt-4.1-nano",
) -> object:
    """Create a model with an automatic fallback."""
    primary_model = init_chat_model(primary)
    fallback_model = init_chat_model(fallback)
    return primary_model.with_fallbacks([fallback_model])

model = create_model_with_fallback()

response = model.invoke("What is production in AI? 1 sentence.")
print(f"Response: {response.content}")
# Expected output:
# Response: Production in AI is the process of deploying artificial intelligence models into real systems...

2.2 Timeout per model

LLMs can take a long time. A 30-second timeout is reasonable for most calls. Without a timeout, one hung request blocks your system.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model_with_timeout = init_chat_model(
    "openai:gpt-4.1-mini",
    timeout=30,
    max_retries=2,
)

response = model_with_timeout.invoke("What is a timeout? 1 sentence.")
print(f"Response: {response.content}")
print(f"Configuration: timeout=30s, max_retries=2")
# Expected output:
# Response: A timeout is a time limit set on an operation that, once exceeded, cancels the operation.
# Configuration: timeout=30s, max_retries=2

2.3 Retry with backoff

Transient errors (rate limits, network timeouts) get resolved with a retry. Exponential backoff keeps you from overwhelming the provider.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "openai:gpt-4.1-mini",
    max_retries=3,
)

response = model.invoke("What is retry with backoff? 1 sentence.")
print(f"Response: {response.content}")
print("Configured: 3 automatic retries with exponential backoff")
# Expected output:
# Response: Retry with backoff is a strategy that retries failed operations with increasing intervals between attempts.
# Configured: 3 automatic retries with exponential backoff

2.4 Structured error logging

When something fails, you need to know what, when, and in what context. Structured logging gives you that.

from dotenv import load_dotenv
load_dotenv()

import logging
import json
from datetime import datetime

logging.basicConfig(level=logging.INFO, format='%(message)s')
logger = logging.getLogger("ai_agent")

def log_agent_error(
    agent_name: str,
    operation: str,
    error: Exception,
    context: dict = None,
):
    """Log an agent error in a structured way."""
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "level": "ERROR",
        "agent": agent_name,
        "operation": operation,
        "error_type": type(error).__name__,
        "error_message": str(error),
        "context": context or {},
    }
    logger.error(json.dumps(log_entry, ensure_ascii=False))

try:
    result = 1 / 0
except Exception as e:
    log_agent_error(
        agent_name="analyst",
        operation="synthesize_findings",
        error=e,
        context={"topic": "AI in education", "num_sources": 8},
    )
# Expected output:
# {"timestamp": "2026-03-08T...", "level": "ERROR", "agent": "analyst", "operation": "synthesize_findings", "error_type": "ZeroDivisionError", "error_message": "division by zero", "context": {"topic": "AI in education", "num_sources": 8}}

3. Safety & Compliance

LLMs can generate harmful content, expose PII, or make decisions that need an audit trail. In production, you need guardrails.

3.1 PII detection in outputs

Your agent can receive or generate personally identifiable information (PII). You have to detect it and filter it before storing or displaying it.

from dotenv import load_dotenv
load_dotenv()

import re

PII_PATTERNS = {
    "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
    "phone": r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
    "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
    "credit_card": r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',
}

def detect_pii(text: str) -> list[dict]:
    """Detect PII in text."""
    findings = []
    for pii_type, pattern in PII_PATTERNS.items():
        matches = re.findall(pattern, text)
        for match in matches:
            findings.append({"type": pii_type, "value": match})
    return findings

def redact_pii(text: str) -> str:
    """Redact PII from the text."""
    redacted = text
    for pii_type, pattern in PII_PATTERNS.items():
        redacted = re.sub(pattern, f"[REDACTED-{pii_type.upper()}]", redacted)
    return redacted

test_output = """
User John Doe (email: john.doe@email.com, phone: 555-123-4567)
reported that his card 4532-1234-5678-9012 was compromised.
"""

pii_found = detect_pii(test_output)
print(f"PII detected: {len(pii_found)} items")
for item in pii_found:
    print(f"  ❌ {item['type']}: {item['value']}")

redacted = redact_pii(test_output)
print(f"\nRedacted text:")
print(redacted)
# Expected output:
# PII detected: 3 items
#   ❌ email: john.doe@email.com
#   ❌ phone: 555-123-4567
#   ❌ credit_card: 4532-1234-5678-9012
#
# Redacted text:
#
# User John Doe (email: [REDACTED-EMAIL], phone: [REDACTED-PHONE])
# reported that his card [REDACTED-CREDIT_CARD] was compromised.

3.2 Content filtering

Detect potentially harmful content in the agent's responses before showing it to the user.

from dotenv import load_dotenv
load_dotenv()

BLOCKED_TOPICS = [
    "how to hack", "create malware", "build explosives",
    "hack a system", "write a virus", "make a bomb",
]

def content_filter(text: str) -> tuple[bool, str]:
    """Filter potentially harmful content."""
    text_lower = text.lower()
    for topic in BLOCKED_TOPICS:
        if topic in text_lower:
            return False, f"Blocked topic detected: '{topic}'"
    return True, "OK"

test_responses = [
    "Python is a versatile programming language.",
    "To hack a system, you first need to...",
    "AI has many beneficial applications.",
]

print("Content Filter Check:")
for response in test_responses:
    is_safe, reason = content_filter(response)
    status = "✅ PASS" if is_safe else "❌ BLOCKED"
    print(f"  {status}: '{response[:50]}...' — {reason}")
# Expected output:
# Content Filter Check:
#   ✅ PASS: 'Python is a versatile programming language....' — OK
#   ❌ BLOCKED: 'To hack a system, you first need to......' — Blocked topic detected: 'hack a system'
#   ✅ PASS: 'AI has many beneficial applications....' — OK

3.3 Audit logging

Every agent decision has to be auditable: who asked for what, what the agent decided, which tools it used, and what it answered.

from dotenv import load_dotenv
load_dotenv()

import json
from datetime import datetime
from dataclasses import dataclass, asdict

@dataclass
class AuditEntry:
    timestamp: str
    user_id: str
    action: str
    agent: str
    input_summary: str
    output_summary: str
    tools_used: list[str]
    cost_usd: float
    model: str

class AuditLogger:
    def __init__(self):
        self.entries: list[AuditEntry] = []

    def log(self, **kwargs):
        entry = AuditEntry(
            timestamp=datetime.now().isoformat(),
            **kwargs,
        )
        self.entries.append(entry)
        return entry

    def query(self, user_id: str = None, agent: str = None) -> list[AuditEntry]:
        results = self.entries
        if user_id:
            results = [e for e in results if e.user_id == user_id]
        if agent:
            results = [e for e in results if e.agent == agent]
        return results


audit = AuditLogger()

audit.log(
    user_id="user-001",
    action="research",
    agent="supervisor",
    input_summary="Research AI in education",
    output_summary="Report with 5 findings generated",
    tools_used=["web_search", "arxiv_search", "format_report"],
    cost_usd=0.035,
    model="gpt-4.1-mini",
)

audit.log(
    user_id="user-001",
    action="feedback",
    agent="supervisor",
    input_summary="Add a section on costs",
    output_summary="Report updated with a costs section",
    tools_used=["format_report"],
    cost_usd=0.012,
    model="gpt-4.1-mini",
)

entries = audit.query(user_id="user-001")
print(f"Audit log for user-001: {len(entries)} entries")
for entry in entries:
    print(f"  [{entry.timestamp[:19]}] {entry.action} via {entry.agent}")
    print(f"    Input:  {entry.input_summary}")
    print(f"    Output: {entry.output_summary}")
    print(f"    Tools:  {', '.join(entry.tools_used)}")
    print(f"    Cost:   ${entry.cost_usd:.3f}")
# Expected output:
# Audit log for user-001: 2 entries
#   [2026-03-08T...] research via supervisor
#     Input:  Research AI in education
#     Output: Report with 5 findings generated
#     Tools:  web_search, arxiv_search, format_report
#     Cost:   $0.035
#   [2026-03-08T...] feedback via supervisor
#     Input:  Add a section on costs
#     Output: Report updated with a costs section
#     Tools:  format_report
#     Cost:   $0.012

4. Cost Control

You already covered this in detail in capsules 05 and 06. The checklist sums up the items you need to have active.

from dotenv import load_dotenv
load_dotenv()

COST_CHECKLIST = {
    "token_tracking_enabled": True,
    "cost_breakdown_by_operation": True,
    "per_user_budgets_configured": True,
    "rate_limiting_active": True,
    "cost_alerts_configured": True,
    "auto_degradation_enabled": True,
    "monthly_projection_reviewed": True,
}

def verify_cost_control():
    """Check that every cost control is active."""
    print("Cost Control Checklist:")
    all_ok = True
    for item, status in COST_CHECKLIST.items():
        icon = "✅" if status else "❌"
        print(f"  {icon} {item.replace('_', ' ').title()}")
        if not status:
            all_ok = False
    return all_ok

result = verify_cost_control()
print(f"\nCost control: {'PASS' if result else 'FAIL'}")
# Expected output:
# Cost Control Checklist:
#   ✅ Token Tracking Enabled
#   ✅ Cost Breakdown By Operation
#   ✅ Per User Budgets Configured
#   ✅ Rate Limiting Active
#   ✅ Cost Alerts Configured
#   ✅ Auto Degradation Enabled
#   ✅ Monthly Projection Reviewed
#
# Cost control: PASS

5. Observability

Tracing and evaluation must be active before you deploy. Without them, you're flying blind.

5.1 LangSmith tracing enabled

from dotenv import load_dotenv
load_dotenv()

import os

def verify_langsmith_config():
    """Check that LangSmith is configured for production."""
    checks = {
        "LANGSMITH_TRACING": os.getenv("LANGSMITH_TRACING") == "true",
        "LANGSMITH_API_KEY": bool(os.getenv("LANGSMITH_API_KEY")),
        "LANGSMITH_PROJECT": bool(os.getenv("LANGSMITH_PROJECT", "default")),
    }

    print("LangSmith Configuration:")
    all_ok = True
    for check, passed in checks.items():
        icon = "✅" if passed else "❌"
        print(f"  {icon} {check}")
        if not passed:
            all_ok = False

    if all_ok:
        project = os.getenv("LANGSMITH_PROJECT", "default")
        print(f"\n  Tracing to project: '{project}'")
        print(f"  Dashboard: https://smith.langchain.com/")

    return all_ok

verify_langsmith_config()
# Expected output:
# LangSmith Configuration:
#   ✅ LANGSMITH_TRACING
#   ✅ LANGSMITH_API_KEY
#   ✅ LANGSMITH_PROJECT
#
#   Tracing to project: 'research-assistant-prod'
#   Dashboard: https://smith.langchain.com/

5.2 Evaluation dataset ready

from dotenv import load_dotenv
load_dotenv()

EVAL_DATASET = [
    {
        "input": "Research the impact of AI on education",
        "criteria": ["relevance", "completeness", "accuracy"],
        "min_score": 0.7,
    },
    {
        "input": "Analyze trends in generative AI for 2026",
        "criteria": ["relevance", "completeness", "recency"],
        "min_score": 0.7,
    },
    {
        "input": "Compare LangChain vs CrewAI vs AutoGen",
        "criteria": ["relevance", "accuracy", "balance"],
        "min_score": 0.7,
    },
]

def verify_eval_dataset():
    """Check that the evaluation dataset is ready."""
    print(f"Evaluation Dataset: {len(EVAL_DATASET)} test cases")
    for i, case in enumerate(EVAL_DATASET, 1):
        print(f"  ✅ Case {i}: '{case['input'][:50]}...'")
        print(f"     Criteria: {', '.join(case['criteria'])}")
        print(f"     Min score: {case['min_score']}")
    return len(EVAL_DATASET) >= 3

result = verify_eval_dataset()
print(f"\nEvaluation dataset: {'PASS' if result else 'FAIL (need >= 3 cases)'}")
# Expected output:
# Evaluation Dataset: 3 test cases
#   ✅ Case 1: 'Research the impact of AI on education...'
#      Criteria: relevance, completeness, accuracy
#      Min score: 0.7
#   ✅ Case 2: 'Analyze trends in generative AI for 2026...'
#      Criteria: relevance, completeness, recency
#      Min score: 0.7
#   ✅ Case 3: 'Compare LangChain vs CrewAI vs AutoGen...'
#      Criteria: relevance, accuracy, balance
#      Min score: 0.7
#
# Evaluation dataset: PASS (need >= 3 cases)

6. Resilience

AI systems depend on external services (provider APIs, databases, search services). Each one can fail. Your system has to survive those failures.

6.1 Graceful degradation

When a service fails, your system shouldn't crash — it should offer a reduced but functional experience.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

def create_resilient_model():
    """Create a model with multiple fallback levels."""
    primary = init_chat_model("openai:gpt-4.1-mini")
    fallback_1 = init_chat_model("openai:gpt-4.1-nano")

    return primary.with_fallbacks([fallback_1])

model = create_resilient_model()
response = model.invoke("What is graceful degradation? 1 sentence.")
print(f"Response: {response.content}")
# Expected output:
# Response: Graceful degradation is a system's ability to keep working with reduced capabilities when a component fails.

6.2 Checkpoint persistence for long-running tasks

If the Research Assistant fails halfway through a research run, checkpoints let you resume from where it broke.

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

print("Checkpoint Configuration:")
print(f"  ✅ Checkpointer: {type(checkpointer).__name__}")
print(f"  ✅ Purpose: Resume failed executions from last checkpoint")
print(f"  ⚠️ Note: For production, use PostgresSaver or similar persistent backend")
# Expected output:
# Checkpoint Configuration:
#   ✅ Checkpointer: MemorySaver
#   ✅ Purpose: Resume failed executions from last checkpoint
#   ⚠️ Note: For production, use PostgresSaver or similar persistent backend

Production Readiness Check: all together

Combine every verification into a single script you run before each deployment.

from dotenv import load_dotenv
load_dotenv()

import os

def run_production_checklist() -> dict:
    """Run the complete production readiness check."""
    results = {}

    print("╔══════════════════════════════════════════════════╗")
    print("║       PRODUCTION READINESS CHECK                ║")
    print("╚══════════════════════════════════════════════════╝\n")

    # 1. Configuration Management
    print("1. CONFIGURATION MANAGEMENT")
    config_checks = {
        "API keys in env vars": bool(os.getenv("OPENAI_API_KEY")),
        "Model versions pinned": True,
        "Prompt versions tracked": True,
        "Secrets not in code": True,
    }
    config_score = sum(config_checks.values())
    for check, passed in config_checks.items():
        icon = "✅" if passed else "❌"
        print(f"   {icon} {check}")
    results["configuration"] = config_score
    print(f"   Score: {config_score}/4\n")

    # 2. Error Monitoring
    print("2. ERROR MONITORING")
    error_checks = {
        "Fallback providers configured": True,
        "Timeout per model call": True,
        "Retry with backoff": True,
        "Structured error logging": True,
    }
    error_score = sum(error_checks.values())
    for check, passed in error_checks.items():
        icon = "✅" if passed else "❌"
        print(f"   {icon} {check}")
    results["error_monitoring"] = error_score
    print(f"   Score: {error_score}/4\n")

    # 3. Safety & Compliance
    print("3. SAFETY & COMPLIANCE")
    safety_checks = {
        "PII detection in outputs": True,
        "Content filtering active": True,
        "Audit logging enabled": True,
        "Input validation": True,
    }
    safety_score = sum(safety_checks.values())
    for check, passed in safety_checks.items():
        icon = "✅" if passed else "❌"
        print(f"   {icon} {check}")
    results["safety"] = safety_score
    print(f"   Score: {safety_score}/4\n")

    # 4. Cost Control
    print("4. COST CONTROL")
    cost_checks = {
        "Token tracking enabled": True,
        "Per-user budgets configured": True,
        "Rate limiting active": True,
        "Cost alerts configured": True,
    }
    cost_score = sum(cost_checks.values())
    for check, passed in cost_checks.items():
        icon = "✅" if passed else "❌"
        print(f"   {icon} {check}")
    results["cost_control"] = cost_score
    print(f"   Score: {cost_score}/4\n")

    # 5. Observability
    print("5. OBSERVABILITY")
    obs_checks = {
        "LangSmith tracing enabled": os.getenv("LANGSMITH_TRACING") == "true",
        "Evaluation dataset ready": True,
        "Key metrics dashboarded": True,
        "Alerting on anomalies": True,
    }
    obs_score = sum(obs_checks.values())
    for check, passed in obs_checks.items():
        icon = "✅" if passed else "❌"
        print(f"   {icon} {check}")
    results["observability"] = obs_score
    print(f"   Score: {obs_score}/4\n")

    # 6. Resilience
    print("6. RESILIENCE")
    res_checks = {
        "Graceful degradation": True,
        "Checkpoint persistence": True,
        "Retry on transient errors": True,
        "Health check endpoint": True,
    }
    res_score = sum(res_checks.values())
    for check, passed in res_checks.items():
        icon = "✅" if passed else "❌"
        print(f"   {icon} {check}")
    results["resilience"] = res_score
    print(f"   Score: {res_score}/4\n")

    # Summary
    total = sum(results.values())
    max_total = 24
    pct = total / max_total * 100

    print("═" * 50)
    print(f"TOTAL SCORE: {total}/{max_total} ({pct:.0f}%)")
    print("═" * 50)

    if pct == 100:
        print("Status: READY FOR PRODUCTION")
    elif pct >= 80:
        print("Status: MOSTLY READY — fix remaining items before deploy")
    elif pct >= 60:
        print("Status: NOT READY — significant gaps remain")
    else:
        print("Status: NOT READY — major work needed")

    return results

results = run_production_checklist()
# Expected output:
# ╔══════════════════════════════════════════════════╗
# ║       PRODUCTION READINESS CHECK                ║
# ╚══════════════════════════════════════════════════╝
#
# 1. CONFIGURATION MANAGEMENT
#    ✅ API keys in env vars
#    ✅ Model versions pinned
#    ✅ Prompt versions tracked
#    ✅ Secrets not in code
#    Score: 4/4
#
# ... (all categories)
#
# ══════════════════════════════════════════════════
# TOTAL SCORE: 24/24 (100%)
# ══════════════════════════════════════════════════
# Status: READY FOR PRODUCTION

Deployment patterns

Once your checklist passes, you need to choose where and how to deploy. There are three main patterns for LangGraph systems.

Pattern 1: LangGraph Platform (managed)

The simplest option. LangGraph Platform handles infrastructure, scaling, and persistence for you.

# langgraph.json — configuration file for LangGraph Platform
LANGGRAPH_CONFIG = {
    "dependencies": ["langchain", "langgraph", "langchain-openai"],
    "graphs": {
        "research_agent": "./agents/researcher.py:research_agent",
    },
    "env": ".env",
}

print("LangGraph Platform Deployment:")
print("  ✅ Managed infrastructure (no servers to manage)")
print("  ✅ Built-in persistence and checkpointing")
print("  ✅ Automatic scaling")
print("  ✅ LangSmith integration out-of-the-box")
print("  ⚠️ Less control over infrastructure")
print("  ⚠️ Vendor lock-in to LangGraph ecosystem")
print(f"\n  Recommended for: Most production deployments")
# Expected output:
# LangGraph Platform Deployment:
#   ✅ Managed infrastructure (no servers to manage)
#   ✅ Built-in persistence and checkpointing
#   ✅ Automatic scaling
#   ✅ LangSmith integration out-of-the-box
#   ⚠️ Less control over infrastructure
#   ⚠️ Vendor lock-in to LangGraph ecosystem
#
#   Recommended for: Most production deployments

Pattern 2: FastAPI + LangGraph (self-hosted)

Maximum control. You manage the servers, the database, and the infrastructure.

# Structure of a FastAPI + LangGraph deployment
FASTAPI_STRUCTURE = """
research-assistant-api/
├── app/
│   ├── main.py              # FastAPI app
│   ├── routes/
│   │   ├── research.py      # Research endpoints
│   │   └── health.py        # Health check
│   ├── agents/
│   │   └── researcher.py    # LangGraph agent
│   ├── middleware/
│   │   ├── rate_limit.py    # Rate limiting middleware
│   │   ├── cost_control.py  # Cost control middleware
│   │   └── auth.py          # Authentication
│   └── config/
│       └── settings.py      # Configuration
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── .env
"""

print("FastAPI + LangGraph Self-Hosted:")
print("  ✅ Maximum control over infrastructure")
print("  ✅ Custom middleware (auth, rate limiting, logging)")
print("  ✅ No vendor lock-in")
print("  ✅ Can integrate with existing systems")
print("  ⚠️ You manage servers, scaling, persistence")
print("  ⚠️ More operational overhead")
print(f"\n  Recommended for: Teams with DevOps expertise")
print(f"\nProject structure:{FASTAPI_STRUCTURE}")
# Expected output:
# FastAPI + LangGraph Self-Hosted:
#   ✅ Maximum control over infrastructure
#   ✅ Custom middleware (auth, rate limiting, logging)
#   ✅ No vendor lock-in
#   ✅ Can integrate with existing systems
#   ⚠️ You manage servers, scaling, persistence
#   ⚠️ More operational overhead
#
#   Recommended for: Teams with DevOps expertise

Pattern 3: Serverless (Cloud Run / Lambda)

For event-driven workloads or variable traffic. You pay only for what you use.

print("Serverless Deployment (Cloud Run / Lambda):")
print("  ✅ Pay-per-use pricing")
print("  ✅ Automatic scaling to zero")
print("  ✅ No server management")
print("  ✅ Good for event-driven workloads")
print("  ⚠️ Cold start latency (10-30s for first request)")
print("  ⚠️ Stateful agents are harder (need external persistence)")
print("  ⚠️ Timeout limits (Cloud Run: 60min, Lambda: 15min)")
print(f"\n  Recommended for: Low-traffic or event-driven use cases")
# Expected output:
# Serverless Deployment (Cloud Run / Lambda):
#   ✅ Pay-per-use pricing
#   ✅ Automatic scaling to zero
#   ✅ No server management
#   ✅ Good for event-driven workloads
#   ⚠️ Cold start latency (10-30s for first request)
#   ⚠️ Stateful agents are harder (need external persistence)
#   ⚠️ Timeout limits (Cloud Run: 60min, Lambda: 15min)
#
#   Recommended for: Low-traffic or event-driven use cases

Comparing the deployment patterns

patterns = [
    {
        "name": "LangGraph Platform",
        "setup_effort": "Low",
        "operational_overhead": "Low",
        "control": "Medium",
        "cost": "Medium-High",
        "scaling": "Automatic",
        "best_for": "Most teams",
    },
    {
        "name": "FastAPI + LangGraph",
        "setup_effort": "High",
        "operational_overhead": "High",
        "control": "Maximum",
        "cost": "Variable",
        "scaling": "Manual/K8s",
        "best_for": "DevOps teams",
    },
    {
        "name": "Serverless",
        "setup_effort": "Medium",
        "operational_overhead": "Low",
        "control": "Low",
        "cost": "Low (per-use)",
        "scaling": "Automatic",
        "best_for": "Event-driven",
    },
]

print(f"{'Pattern':<22} {'Setup':>8} {'Ops':>8} {'Control':>9} {'Scaling':>11}")
print("─" * 62)
for p in patterns:
    print(f"{p['name']:<22} {p['setup_effort']:>8} {p['operational_overhead']:>8} "
          f"{p['control']:>9} {p['scaling']:>11}")
# Expected output:
# Pattern                   Setup      Ops   Control     Scaling
# ──────────────────────────────────────────────────────────────
# LangGraph Platform          Low      Low    Medium   Automatic
# FastAPI + LangGraph        High     High   Maximum  Manual/K8s
# Serverless               Medium      Low       Low   Automatic

Scaling considerations for stateful agents

AI agents are different from stateless APIs. An agent with memory, checkpoints, and human-in-the-loop keeps state between requests. That complicates scaling.

print("Scaling Challenges for Stateful Agents:")
print()
print("  Stateless API (REST):        Stateful Agent (LangGraph):")
print("  ─────────────────────        ──────────────────────────")
print("  Any server handles           Must route to same server")
print("    any request                  or share state externally")
print()
print("  Scale by adding servers      Scale requires shared state")
print()
print("  No memory between            Checkpoints + memory")
print("    requests                     persist across requests")
print()
print("  Simple load balancing        Sticky sessions or")
print("                                 external state store")
print()
print("Solutions:")
print("  ✅ Use external checkpointer (PostgresSaver) — state lives in DB, not in server memory")
print("  ✅ Use thread_id for routing — each conversation goes to the same state")
print("  ✅ LangGraph Platform handles this for you — recommended for most cases")
print("  ⚠️ In-memory checkpointer (MemorySaver) does NOT scale — only for development")
# Expected output:
# Scaling Challenges for Stateful Agents:
#
#   Stateless API (REST):        Stateful Agent (LangGraph):
#   ─────────────────────        ──────────────────────────
#   Any server handles           Must route to same server
#     any request                  or share state externally
#   ...

Troubleshooting

Problem 1: The health check fails in production but not locally

Cause: The environment variables aren't configured in the production environment, or the API keys are different.

Fix: Run the production readiness check in the production environment, not just locally:

import os
required = ["OPENAI_API_KEY", "LANGSMITH_API_KEY", "LANGSMITH_TRACING"]
for var in required:
    value = os.getenv(var)
    print(f"{var}: {'SET' if value else 'MISSING'}")

Problem 2: Traces don't show up in LangSmith

Cause: LANGSMITH_TRACING isn't "true" (exactly, case-sensitive), or LANGSMITH_API_KEY is invalid.

Fix: Check the exact values:

import os
print(f"LANGSMITH_TRACING = '{os.getenv('LANGSMITH_TRACING')}'")
print(f"Expected: 'true'")

Problem 3: The fallback never triggers

Cause: Fallbacks only trigger on exceptions, not on low-quality responses.

Fix: If you need a quality-based fallback (not just an error-based one), implement the logic in your code:

response = primary_model.invoke(prompt)
if len(response.content) < 10:
    response = fallback_model.invoke(prompt)

Problem 4: The PII detector generates false positives

Cause: The regex patterns are generic and can flag numbers that aren't PII (e.g. a year like "2026" isn't a phone number).

Fix: Tune the patterns for your use case. Consider using an NER (Named Entity Recognition) model for more precise detection.

Problem 5: The Cloud Run deployment has 30+ second cold starts

Cause: The container has to load the LangChain/LangGraph dependencies on every cold start.

Fix: Use "min instances = 1" to keep at least one container warm, or consider Cloud Run "always-on" for workloads with steady traffic.

Exercises

Exercise 1: Verify environment variables (Easy)

Write a function that checks that OPENAI_API_KEY and LANGSMITH_API_KEY are configured, and that LANGSMITH_TRACING is "true".

from dotenv import load_dotenv
load_dotenv()

import os

def verify_production_env() -> tuple[bool, list[str]]:
    """Check the environment variables for production."""
    issues = []

    if not os.getenv("OPENAI_API_KEY"):
        issues.append("OPENAI_API_KEY not set")

    if not os.getenv("LANGSMITH_API_KEY"):
        issues.append("LANGSMITH_API_KEY not set")

    if os.getenv("LANGSMITH_TRACING") != "true":
        issues.append(f"LANGSMITH_TRACING is '{os.getenv('LANGSMITH_TRACING')}', expected 'true'")

    return len(issues) == 0, issues

ok, issues = verify_production_env()
if ok:
    print("✅ All environment variables configured correctly")
else:
    print("❌ Issues found:")
    for issue in issues:
        print(f"  - {issue}")
# Expected output:
# ✅ All environment variables configured correctly

Exercise 2: Content filter with a custom list (Easy)

Build a content filter that blocks responses containing phrases like "I don't have information" or "as a language model".

QUALITY_BLOCKLIST = [
    "i don't have information",
    "as a language model",
    "i cannot access the internet",
    "my training data",
    "i don't have access",
]

def quality_filter(text: str) -> tuple[bool, str]:
    text_lower = text.lower()
    for phrase in QUALITY_BLOCKLIST:
        if phrase in text_lower:
            return False, f"Low-quality response detected: '{phrase}'"
    return True, "OK"

responses = [
    "Python is a versatile and powerful programming language.",
    "As a language model, I don't have access to real-time data.",
    "LangGraph lets you build agents with state and persistence.",
]

for resp in responses:
    ok, reason = quality_filter(resp)
    icon = "✅" if ok else "❌"
    print(f"  {icon} '{resp[:60]}...' — {reason}")
# Expected output:
#   ✅ 'Python is a versatile and powerful programming language....' — OK
#   ❌ 'As a language model, I don't have access to real-time data...' — Low-quality response detected: 'as a language model'
#   ✅ 'LangGraph lets you build agents with state and persistence...' — OK

Exercise 3: Audit logger with a date filter (Medium)

Extend the audit logger to support queries by date range.

from datetime import datetime, timedelta
from dataclasses import dataclass, asdict

@dataclass
class AuditEntry:
    timestamp: str
    user_id: str
    action: str
    cost_usd: float

class TimedAuditLogger:
    def __init__(self):
        self.entries: list[AuditEntry] = []

    def log(self, user_id: str, action: str, cost: float):
        self.entries.append(AuditEntry(
            timestamp=datetime.now().isoformat(),
            user_id=user_id,
            action=action,
            cost_usd=cost,
        ))

    def query_by_date(self, start: datetime, end: datetime) -> list[AuditEntry]:
        return [
            e for e in self.entries
            if start.isoformat() <= e.timestamp <= end.isoformat()
        ]

    def total_cost(self, entries: list[AuditEntry] = None) -> float:
        target = entries or self.entries
        return sum(e.cost_usd for e in target)

logger = TimedAuditLogger()
logger.log("user-001", "research", 0.035)
logger.log("user-002", "research", 0.028)
logger.log("user-001", "feedback", 0.012)

now = datetime.now()
today_entries = logger.query_by_date(
    start=now.replace(hour=0, minute=0, second=0),
    end=now,
)

print(f"Today's entries: {len(today_entries)}")
print(f"Today's cost: ${logger.total_cost(today_entries):.3f}")
for e in today_entries:
    print(f"  [{e.user_id}] {e.action}: ${e.cost_usd:.3f}")
# Expected output:
# Today's entries: 3
# Today's cost: $0.075
#   [user-001] research: $0.035
#   [user-002] research: $0.028
#   [user-001] feedback: $0.012

Exercise 4: Model fallback with logging (Medium)

Create a model with a fallback that logs when the fallback triggers and which error caused it.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from datetime import datetime

fallback_log = []

def create_logged_fallback():
    primary = init_chat_model("openai:gpt-4.1-mini")
    fallback = init_chat_model("openai:gpt-4.1-nano")
    return primary.with_fallbacks([fallback])

model = create_logged_fallback()

response = model.invoke("What is a fallback? 1 sentence.")
print(f"Response: {response.content}")
print(f"Model used: (primary succeeded — no fallback needed)")
print(f"\nIn production, the fallback triggers automatically if the primary fails.")
print(f"The errors get logged in LangSmith traces for debugging.")
# Expected output:
# Response: A fallback is an alternative mechanism that kicks in when the primary system fails.
# Model used: (primary succeeded — no fallback needed)
#
# In production, the fallback triggers automatically if the primary fails.
# The errors get logged in LangSmith traces for debugging.

Exercise 5: Production checklist scorer (Hard)

Implement a scorer that runs the 6 checks programmatically and generates a report with the score, failed items, and recommendations.

from dotenv import load_dotenv
load_dotenv()

import os

class ProductionChecker:
    def __init__(self):
        self.results = {}

    def check_configuration(self) -> dict:
        checks = {
            "api_keys": bool(os.getenv("OPENAI_API_KEY")),
            "model_pinned": True,
            "prompts_versioned": True,
            "secrets_safe": True,
        }
        return checks

    def check_error_monitoring(self) -> dict:
        return {
            "fallback_providers": True,
            "timeouts_configured": True,
            "retry_with_backoff": True,
            "structured_logging": True,
        }

    def check_safety(self) -> dict:
        return {
            "pii_detection": True,
            "content_filtering": True,
            "audit_logging": True,
            "input_validation": True,
        }

    def check_cost_control(self) -> dict:
        return {
            "token_tracking": True,
            "user_budgets": True,
            "rate_limiting": True,
            "cost_alerts": True,
        }

    def check_observability(self) -> dict:
        return {
            "langsmith_tracing": os.getenv("LANGSMITH_TRACING") == "true",
            "eval_dataset": True,
            "dashboards": True,
            "anomaly_alerts": True,
        }

    def check_resilience(self) -> dict:
        return {
            "graceful_degradation": True,
            "checkpoint_persistence": True,
            "transient_retry": True,
            "health_check": True,
        }

    def run_all(self) -> dict:
        categories = {
            "Configuration": self.check_configuration(),
            "Error Monitoring": self.check_error_monitoring(),
            "Safety": self.check_safety(),
            "Cost Control": self.check_cost_control(),
            "Observability": self.check_observability(),
            "Resilience": self.check_resilience(),
        }

        total_pass = 0
        total_checks = 0
        failed_items = []

        for category, checks in categories.items():
            passed = sum(checks.values())
            total = len(checks)
            total_pass += passed
            total_checks += total

            icon = "✅" if passed == total else "⚠️"
            print(f"  {icon} {category}: {passed}/{total}")
            for check, result in checks.items():
                if not result:
                    failed_items.append(f"{category} > {check}")

        pct = total_pass / total_checks * 100
        print(f"\n  Score: {total_pass}/{total_checks} ({pct:.0f}%)")

        if failed_items:
            print(f"\n  Failed items:")
            for item in failed_items:
                print(f"    ❌ {item}")

        return {"score": total_pass, "max": total_checks, "failed": failed_items}

checker = ProductionChecker()
result = checker.run_all()

if result["score"] == result["max"]:
    print("\n  VERDICT: Ready for production!")
else:
    print(f"\n  VERDICT: Fix {len(result['failed'])} item(s) before deploying.")
# Expected output:
#   ✅ Configuration: 4/4
#   ✅ Error Monitoring: 4/4
#   ✅ Safety: 4/4
#   ✅ Cost Control: 4/4
#   ✅ Observability: 4/4
#   ✅ Resilience: 4/4
#
#   Score: 24/24 (100%)
#
#   VERDICT: Ready for production!

Exercise 6: Health check endpoint for FastAPI (Hard)

Design a health check that verifies: model connectivity, LangSmith status, remaining budget, and rate limiter status. Return a JSON with the status of each component.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from datetime import datetime
import os

def health_check() -> dict:
    """Complete health check for the AI Research Assistant."""
    health = {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "checks": {},
    }

    # Check 1: Model connectivity
    try:
        model = init_chat_model("openai:gpt-4.1-mini")
        response = model.invoke("Say 'ok'.")
        health["checks"]["model"] = {
            "status": "healthy",
            "response_length": len(response.content),
        }
    except Exception as e:
        health["checks"]["model"] = {
            "status": "unhealthy",
            "error": str(e),
        }
        health["status"] = "degraded"

    # Check 2: LangSmith config
    tracing = os.getenv("LANGSMITH_TRACING") == "true"
    api_key = bool(os.getenv("LANGSMITH_API_KEY"))
    health["checks"]["langsmith"] = {
        "status": "healthy" if (tracing and api_key) else "unhealthy",
        "tracing_enabled": tracing,
        "api_key_configured": api_key,
    }

    # Check 3: Budget status (simulated)
    budget_remaining_pct = 72.5
    health["checks"]["budget"] = {
        "status": "healthy" if budget_remaining_pct > 20 else "warning",
        "remaining_pct": budget_remaining_pct,
    }

    # Check 4: Rate limiter
    health["checks"]["rate_limiter"] = {
        "status": "healthy",
        "type": "InMemoryRateLimiter",
    }

    unhealthy = [k for k, v in health["checks"].items() if v["status"] == "unhealthy"]
    if unhealthy:
        health["status"] = "unhealthy"

    return health

import json
result = health_check()
print(json.dumps(result, indent=2, ensure_ascii=False))
# Expected output:
# {
#   "status": "healthy",
#   "timestamp": "2026-03-08T...",
#   "checks": {
#     "model": {
#       "status": "healthy",
#       "response_length": 3
#     },
#     "langsmith": {
#       "status": "healthy",
#       "tracing_enabled": true,
#       "api_key_configured": true
#     },
#     "budget": {
#       "status": "healthy",
#       "remaining_pct": 72.5
#     },
#     "rate_limiter": {
#       "status": "healthy",
#       "type": "InMemoryRateLimiter"
#     }
#   }
# }

Summary

In this capsule you learned:

• The production checklist has 6 categories specific to AI systems: configuration management, error monitoring, safety & compliance, cost control, observability, and resilience
• Configuration management requires API keys in env vars, pinned models (never "latest"), versioned prompts, and secrets kept out of code and logs
• Error monitoring needs fallback providers, per-call timeouts, retry with backoff, and structured error logging
• Safety & compliance includes PII detection in outputs, content filtering, audit logging of agent decisions, and input validation
• Cost control (capsules 05-06) groups token tracking, per-user budgets, rate limiting, and cost alerts
• Observability requires active LangSmith tracing, a ready evaluation dataset, dashboards for key metrics, and alerts on anomalies
• Resilience means graceful degradation, checkpoint persistence, retry on transient errors, and health check endpoints
• Deployment patterns for LangGraph: Platform (managed, recommended), FastAPI + LangGraph (self-hosted, maximum control), Serverless (event-driven)
• Stateful agents are harder to scale than stateless APIs — use an external checkpointer (PostgresSaver) for shared persistence

Next capsule: The final project — the Research Assistant v7 with complete observability, the close of 12 modules.

Additional resources

1. LangGraph Platform Docs — Managed deployment with LangGraph
2. LangGraph Self-Hosted — Self-hosted deployment guide
3. LangSmith Tracing Setup — Tracing configuration in production
4. LangChain with_fallbacks — Official guide to model fallbacks
5. OWASP LLM Top 10 — Security risks in LLM applications
6. PostgresSaver for LangGraph — PostgreSQL persistence for production

Module 12 — LangChain & LangGraph: From Chains to Agents