AI Architecture and Methods

A comprehensive guide to AI architecture and methods for production environments.

Table of Contents

1. Why Architecture Matters
2. Build vs. Buy Decisions
   • 2.1 The Build vs. Buy Decision Framework
   • 2.2 Evaluating Your Options
   • 2.3 Making It Work in Production
3. Operating in Production
   • 3.1 The Evolution of Processing Patterns
   • 3.2 Matching Architecture to the Problem
   • 3.3 The Reality Check
4. Predictable Failure Patterns
   • 4.1 Why AI Integration Is Different
   • 4.2 Predictable Failures at the Boundaries
   • 4.3 Operational Reality
5. Architecture Selection
   • 5.1 The Failure Cliff
   • 5.2 Performance Trade-offs That Matter
   • 5.3 Building for Reality
6. Your Decision Framework
7. Quick Reference and Anti-Patterns

1. Why Architecture Matters

The Reality Behind Demos

Most of us experience the same moment: someone above us has watched an impressive AI demo and announces “we need AI everywhere by Q3.” What demos don’t show:

• The “perfect” model can cost $10,000 per month in GPU time
• Instant responses are often caches of hours of pre-processing
• In production at 3 AM, the same system can answer confidently… and be wrong
• All while burning your infrastructure budget

┌─────────────────────────────────────────────────────────┐
│              THE DEMO vs. PRODUCTION GAP                │
├─────────────────────────────────────────────────────────┤
│ DEMO                    │ PRODUCTION                    │
│ Fixed latency           │ Latency 2s → 30s              │
│ Clean data              │ Real-world data               │
│ Unlimited budget (demo) │ Financial constraints         │
│ No load                 │ Traffic spikes                │
│ No errors displayed     │ Silent hallucinations         │
└─────────────────────────────────────────────────────────┘

The Three Decisions That Lock Everything In

The difference between teams that succeed and those explaining failure 6 months later comes down to three decisions:

flowchart TD
    D1["Decision 1\nBuild / Buy / Rent\nyour model"] --> C["Costs\nConstraints\nFuture emergencies"]
    D2["Decision 2\nArchitecture for\nreal workloads"] --> C
    D3["Decision 3\nManage AI\nfailure modes"] --> C

    style D1 fill:#4A90D9,color:#fff
    style D2 fill:#5BA85A,color:#fff
    style D3 fill:#E8A838,color:#fff
    style C fill:#D95B5B,color:#fff

Why AI Is Different from Other Services

Golden rule: When a database fails, you get an error. When AI fails, you get a confident response that is completely wrong. There are no exceptions raised for hallucinations.

2. Build vs. Buy Decisions

2.1 The Build vs. Buy Decision Framework

There are essentially four paths to access an AI model:

flowchart LR
    subgraph Full_Control["Full Control"]
        A["Build Custom\n~$500K minimum\n6–18 months"]
    end
    subgraph Middle_Ground["Middle Ground"]
        B["Fine-tuning\nA few months\nUnique data required"]
    end
    subgraph Rent["Rent"]
        C["APIs\n$50–100/day\nImmediate start"]
    end
    subgraph Buy["Buy"]
        D["Vendor Products\n~$10K/month\nWrapper on APIs"]
    end

    style A fill:#D95B5B,color:#fff
    style B fill:#E8A838,color:#fff
    style C fill:#5BA85A,color:#fff
    style D fill:#4A90D9,color:#fff

When Does a Custom Build Make Sense?

A fully custom build only makes sense if:

• You have truly unique data that creates a competitive advantage
  • A quantitative firm with proprietary signals
  • A medical imaging company with labeled scans
• If you’re not protecting something unique, you’re paying for control you probably don’t need

Real Costs Compared

xychart-beta
    title "Estimated Annual Cost by Approach (USD)"
    x-axis ["Build Custom", "Fine-tuning", "API", "Vendor"]
    y-axis "Cost USD" 0 --> 2500000
    bar [2000000, 400000, 10200, 120000]

API example: 100,000 requests/month × 2,000 tokens at $2/million tokens = $400 in base API costs. With rate limits, fallback models, monitoring: ~$850/month.

The Four Options in Detail

1. Build Custom — For whom?

✅ Truly unique proprietary data
✅ Competitive advantage on data
✅ Budget > $2M over 2 years
✅ Senior ML team already in place
❌ If you just want control
❌ If the execution deadline is Q3

2. Fine-tuning — What to expect?

⏱️  Several months minimum
👥  Engineers who understand data prep + evaluation + drift
🗃️  Real data = PDFs, spreadsheets, emails full of contradictions
💡  You'll spend more time cleaning than training

3. APIs — The “Rent” option

# Simple OpenAI API call example
from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is our return policy?"}
    ],
    max_tokens=500
)

print(response.choices[0].message.content)

4. Vendor Products

✅ Compliance, SLA, support included
✅ Ready-made integrations
❌ Often a wrapper on the same APIs you could call directly
❌ Enterprise plan = GPT + integrations + support, but it's still GPT

2.2 Evaluating Your Options

Warning Signals to Watch For

mindmap
  root((Warning Signals))
    Vendor
      "Our solution handles everything"
      Never mentions failures
      Vague pricing ("contact us")
    Planning
      Guaranteed delivery in 6 months
      No mention of edge cases
    Costs
      Undefined per-interaction pricing
      Unclear usage metrics
      No cost ceiling

How to Interrogate Vendors About Production

Evaluation rule: Teams that have actually deployed AI will tell you about failures directly — the model that produced garbage at 2 AM, the cost spike that jumped 100x, and the “competent” responses that created customer service nightmares.

Production Cost Estimation

API Scenario — Typical Application:
─────────────────────────────────────
• 100,000 requests/month
• 2,000 tokens/request (context + response)
• Rate: $2 / million tokens
• Base API cost       : $400/month
• Rate limits + upgrades: +$150
• Fallback models     : +$100
• Monitoring          : +$200
                      ─────────
  TOTAL               : ~$850/month

2.3 Making It Work in Production

The Evolution of the Simple API Call

Everything starts with a simple call. It works until the first traffic spike:

flowchart LR
    subgraph V1["Version 1 — Naive"]
        U1[User] --> |"Request"| M1[AI Model]
        M1 --> |"Response"| U1
    end

    subgraph V2["Version 2 — Queue + Worker"]
        U2[User] --> |"Instant response"| Q2[Queue]
        Q2 --> W2[Worker]
        W2 --> M2[AI Model]
        M2 --> |"Async result"| U2
    end

    V1 -->|"Traffic spike\n429 errors\nTimeouts"| V2

Base pattern with queue and worker:

# Pattern: Queue + Worker for async AI calls
import redis
import json
import uuid
from openai import OpenAI

client = OpenAI()
r = redis.Redis(host='localhost', port=6379)

def submit_ai_request(user_message: str) -> str:
    """Record the request and return an ID immediately."""
    request_id = str(uuid.uuid4())
    job = {
        "id": request_id,
        "message": user_message,
        "status": "pending"
    }
    r.lpush("ai_queue", json.dumps(job))
    return request_id  # Return immediately to the user

def process_ai_queue():
    """Worker that processes requests in the background."""
    while True:
        _, raw = r.brpop("ai_queue")
        job = json.loads(raw)
        
        try:
            response = client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": job["message"]}]
            )
            result = response.choices[0].message.content
            
            # Store the result
            r.setex(f"result:{job['id']}", 3600, result)
            
        except Exception as e:
            r.setex(f"error:{job['id']}", 3600, str(e))

Stabilization Patterns

Once the happy path is established, add elements that maintain stability:

flowchart TD
    U[User] --> CB["Circuit Breaker\n(Fail fast if service is sick)"]
    CB --> Q[Queue + Worker]
    Q --> C["Cache\n(Frequent questions)"]
    C --> |"Cache HIT"| U
    C --> |"Cache MISS"| R["Retry with exponential\nbackoff and random jitter"]
    R --> M[AI Model]
    M --> U

    style CB fill:#E8A838,color:#fff
    style C fill:#5BA85A,color:#fff
    style R fill:#4A90D9,color:#fff

Circuit Breaker — Implementation:

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"        # Normal — requests pass through
    OPEN = "open"            # Failure — fail fast
    HALF_OPEN = "half_open"  # Testing — try one request

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.last_failure_time = None

    def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker OPEN — service unavailable")

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

    def _on_success(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED

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

# Usage
ai_breaker = CircuitBreaker(failure_threshold=5, reset_timeout=60)

def call_ai_with_breaker(prompt: str):
    def _ai_call():
        return client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}]
        )
    return ai_breaker.call(_ai_call)

Retry with exponential backoff and jitter:

import random
import time
from tenacity import retry, stop_after_attempt, wait_exponential, wait_random

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10) + wait_random(0, 2)
)
def call_ai_with_retry(prompt: str) -> str:
    """AI call with automatic retry and backoff + random jitter."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

Cache for frequent questions:

import hashlib
import json
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_cached_response(prompt: str) -> str | None:
    """Check cache before calling the model."""
    cache_key = "ai:" + hashlib.sha256(prompt.encode()).hexdigest()
    cached = r.get(cache_key)
    return cached if cached else None

def store_cached_response(prompt: str, response: str, ttl: int = 3600):
    """Cache the response with expiration."""
    cache_key = "ai:" + hashlib.sha256(prompt.encode()).hexdigest()
    r.setex(cache_key, ttl, response)

def get_ai_response(prompt: str) -> str:
    """Complete pipeline: cache → AI → storage."""
    cached = get_cached_response(prompt)
    if cached:
        return cached

    response = call_ai_with_retry(prompt)
    store_cached_response(prompt, response)
    return response

Streaming to Improve Perceived Speed

from openai import OpenAI

client = OpenAI()

def stream_ai_response(prompt: str):
    """Stream the response token by token — the user sees progress."""
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )

    for chunk in stream:
        if chunk.choices[0].delta.content is not None:
            # Send each token to the client via SSE/WebSocket
            yield chunk.choices[0].delta.content

Architectural Trade-offs in Production

3. Operating in Production

3.1 The Evolution of Processing Patterns

In production, your processing pattern determines latency, cost, and debugging difficulty.

flowchart LR
    S["Sequential\nOne request\nat a time"] -->|"Requests\noverlap"| P["Parallel\nMultiple requests\nsimultaneously"]
    P -->|"CPU contention\nthroughput ceiling"| D["Distributed\nWork spread\nacross machines"]
    
    style S fill:#5BA85A,color:#fff
    style P fill:#E8A838,color:#fff
    style D fill:#D95B5B,color:#fff

When to Move to the Next Level?

Parallel Pattern — Essential Elements

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def process_request(prompt: str, request_id: str) -> dict:
    """Async AI processing with traceability ID."""
    try:
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}]
        )
        return {
            "id": request_id,
            "result": response.choices[0].message.content,
            "status": "success"
        }
    except Exception as e:
        return {"id": request_id, "error": str(e), "status": "failed"}

async def process_batch(prompts: list[dict]) -> list[dict]:
    """Process multiple requests in parallel."""
    tasks = [
        process_request(p["message"], p["id"]) 
        for p in prompts
    ]
    return await asyncio.gather(*tasks)

Distributed Partitioning — The Key

Think of supermarket checkout lanes: dedicated lanes that flow smoothly are better than one congested lane.

graph LR
    Requests --> Router[Router by partition]
    Router -->|"client_A"| W1["Worker 1\nPartition A"]
    Router -->|"client_B"| W2["Worker 2\nPartition B"]
    Router -->|"region_EU"| W3["Worker 3\nPartition EU"]

    style Router fill:#4A90D9,color:#fff
    style W1 fill:#5BA85A,color:#fff
    style W2 fill:#5BA85A,color:#fff
    style W3 fill:#5BA85A,color:#fff

Technical term: Partitioning groups work by client, document type, or region. The key: keep related work on the same partition to avoid intra-partition network calls.

Real-time vs. Batch — The Key Question

Who is waiting?

Someone is watching the screen → Real-time (< 500ms for interactive UI)
Nobody is waiting             → Batch (schedule on a cron)

def should_use_realtime(request: dict) -> bool:
    """Decide if the request requires real-time processing."""
    return (
        request.get("user_waiting", False) or
        request.get("priority", "normal") == "high" or
        request.get("type") == "interactive"
    )

def route_request(request: dict):
    if should_use_realtime(request):
        return submit_ai_request(request["message"])
    else:
        return schedule_batch_job(request)

3.2 Matching Architecture to the Problem

There is no universal architecture — first map the problem, then choose the pattern.

For Language Tasks (NLP)

flowchart TD
    A["Deploy the baseline\nA few days of real prompts"] --> B["Log failures\n(group by category)"]
    B --> C{Failure type?}
    C -->|"Unknown terms\nPrivate facts"| D["Add Retrieval\nRAG first"]
    C -->|"Repeated gaps\nCorrective examples"| E["Consider\nFine-tuning"]
    C -->|"Poorly formulated prompt"| F["Iterate on\nthe prompt"]
    D --> G["Evaluate\nimprovements"]
    E --> G
    F --> G

    style A fill:#4A90D9,color:#fff
    style D fill:#5BA85A,color:#fff
    style E fill:#E8A838,color:#fff

For Structured Decisions (yes/no, categories)

1. Start with explainable, traceable rules
2. Implement a "why trail" — each decision records its reason
3. Log cases that rules miss or misclassify
4. Add a small model ONLY for residual cases
5. Show rule reason if a rule applies
6. Show model reason if the model is used

def make_decision(input_data: dict) -> dict:
    """Decision with complete audit trail."""
    
    # 1. Try rules first
    rule_result = apply_business_rules(input_data)
    if rule_result is not None:
        return {
            "decision": rule_result["decision"],
            "reason": rule_result["reason"],
            "method": "rule",
            "rule_id": rule_result["rule_id"]
        }
    
    # 2. Fall back to the model for residual cases
    model_result = call_ai_model(input_data)
    return {
        "decision": model_result["decision"],
        "reason": model_result["explanation"],
        "method": "model",
        "confidence": model_result["confidence"]
    }

For RAG and Search

flowchart LR
    subgraph Step1["Step 1: Keyword"]
        K["Keyword\nsearch"] --> |"Failures?"| S
    end
    subgraph Step2["Step 2: Semantic"]
        S["Semantic\nsearch"] --> |"Noisy results?"| R
    end
    subgraph Step3["Step 3: Re-ranking"]
        R["Re-ranking\nby clicks or model"] --> IDX
    end
    subgraph Maintenance["Maintenance"]
        IDX["Refresh\nindex regularly"]
    end

Complete RAG pipeline:

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
llm = ChatOpenAI(model="gpt-4o", temperature=0)

def build_rag_pipeline(documents: list[str]):
    """Build a basic RAG pipeline."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        separators=["\n\n", "\n", ". ", " "]
    )
    
    docs = [Document(page_content=d) for d in documents]
    chunks = splitter.split_documents(docs)
    
    vectorstore = Chroma.from_documents(chunks, embeddings)
    retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
    
    return retriever

def query_rag(retriever, question: str) -> dict:
    """RAG query with cited sources."""
    relevant_docs = retriever.invoke(question)
    context = "\n\n".join([d.page_content for d in relevant_docs])
    
    prompt = f"""Base your answer ONLY on the following context.
If the answer is not in the context, say so clearly.

Context:
{context}

Question: {question}
"""
    
    response = llm.invoke(prompt)
    
    return {
        "answer": response.content,
        "sources": [d.metadata.get("source", "unknown") for d in relevant_docs]
    }

The Context Window Budget

Treat the context window as a budget:

Context window budget
├── System prompt         → Short (≤ 200 tokens)
├── Recent conversation   → Keep last N turns
├── Relevant facts (RAG)  → Selective chunks, not everything
└── User task             → Always first

The model pays more attention to what is at
the beginning and end of the context.
Put the task first, facts second.

def build_context_budget(
    task: str,
    recent_conversation: list,
    retrieved_facts: list,
    max_tokens: int = 4000
) -> str:
    """Build the context within budget limits."""
    
    # 1. Task first (high model attention)
    context = f"Task: {task}\n\n"
    
    # 2. Recent conversation
    for turn in recent_conversation[-3:]:  # Keep last 3 turns
        context += f"{turn['role']}: {turn['content']}\n"
    
    # 3. Retrieved facts (deduplicated, compressed)
    unique_facts = list(dict.fromkeys(retrieved_facts))
    for fact in unique_facts[:5]:  # Limit to 5 facts max
        context += f"\nRelevant fact: {fact}"
    
    return context

3.3 The Reality Check

“Every augmentation is an amputation.” — For us, every AI use adds a capability but also a burden.

A simple string scan runs in microseconds. A model interpreting that string takes seconds. That’s 10,000 times different.

Define the Latency Target BEFORE Designing

flowchart LR
    I["Interactive\nInterface"] -->|"< 500ms"| Target1["P95 < 500ms"]
    B["Background\nupdate"] -->|"A few seconds"| Target2["P95 < 3s"]
    R["Non-urgent\nreport"] -->|"Offline / Batch"| Target3["SLA in hours"]

    style Target1 fill:#D95B5B,color:#fff
    style Target2 fill:#E8A838,color:#fff
    style Target3 fill:#5BA85A,color:#fff

The Tabletop Exercise with a 10x Multiplier

Before being in an incident, do the exercise on paper:

At 10x current traffic, what breaks first?
──────────────────────────────────────────────────────
□ Database connections?
□ AI provider rate limits?
□ Queue depth?
□ Available memory?

At 10x data volume:
□ Do storage and indexes keep up?

At 10x cost:
□ Is the product still financially viable?

For each breaking point:

1. Immediate response: “Increase rate limits” or “Add cache”
2. Long-term fix: The required architectural change

4. Predictable Failure Patterns

4.1 Why AI Integration Is Different

We’re accustomed to predictable systems. With AI, that predictability disappears. The same input can return a slightly different response — or in a completely different format.

Format Drift — What to Expect

You ask for: JSON with "answer" and "sources"

You might receive:
  ✅ {"answer": "...", "sources": [...]}
  ⚠️ {"answer": "...", "sources": [...], "extra_field": "..."}  # Unexpected field
  ⚠️ {"answer": "...", "sources": "source1, source2"}           # Different type
  ❌ ```json\n{"answer": "..."}```                              # Wrapped in backticks
  ❌ {'answer': '...', 'sources': [...]}                        # Single quotes

Output Validation — The Minimum Required

import json
import re
from typing import Any

def parse_ai_json_response(raw_response: str) -> dict | None:
    """
    Parse LLM JSON response with tolerance for common formats.
    Never trust raw output without validation.
    """
    
    # 1. Try direct parsing
    try:
        return json.loads(raw_response)
    except json.JSONDecodeError:
        pass
    
    # 2. Clean Markdown code blocks
    cleaned = raw_response.strip()
    if cleaned.startswith("```"):
        match = re.search(r'```(?:json)?\s*([\s\S]*?)```', cleaned)
        if match:
            try:
                return json.loads(match.group(1).strip())
            except json.JSONDecodeError:
                pass
    
    # 3. Fix single quotes
    try:
        fixed = cleaned.replace("'", '"')
        return json.loads(fixed)
    except json.JSONDecodeError:
        pass
    
    # 4. Fix trailing commas
    try:
        fixed = re.sub(r',\s*([}\]])', r'\1', cleaned)
        return json.loads(fixed)
    except json.JSONDecodeError:
        pass
    
    # 5. Total failure → take fallback path
    return None

def safe_ai_call(prompt: str, expected_schema: dict) -> dict:
    """AI call with return schema validation."""
    
    raw = call_ai_with_retry(prompt)
    parsed = parse_ai_json_response(raw)
    
    if parsed is None:
        return {"error": "invalid_response", "fallback": True}
    
    for required_field in expected_schema.get("required", []):
        if required_field not in parsed:
            return {"error": f"missing_field:{required_field}", "fallback": True}
    
    return parsed

Use JSON Mode When Available

# Preferred method: native JSON mode (more reliable than prompts)
response = client.chat.completions.create(
    model="gpt-4o",
    response_format={"type": "json_object"},  # Native JSON mode
    messages=[
        {
            "role": "system",
            "content": "Always return valid JSON with 'answer' and 'sources' fields."
        },
        {"role": "user", "content": prompt}
    ]
)

# Or with Structured Outputs (even more reliable)
from pydantic import BaseModel

class AIResponse(BaseModel):
    answer: str
    sources: list[str]
    confidence: float

response = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[{"role": "user", "content": prompt}],
    response_format=AIResponse  # Strict schema
)
result = response.choices[0].message.parsed

Monitor for Format Drift

import datetime

class FormatDriftMonitor:
    """Monitor output format changes over time."""
    
    def __init__(self):
        self.test_prompts = [
            "Return JSON: {'status': 'ok', 'count': 5}",
            "List 3 colors as a JSON array",
        ]
    
    def daily_check(self):
        """Run each morning to detect drift."""
        issues = []
        
        for prompt in self.test_prompts:
            result = call_ai_with_retry(prompt)
            parsed = parse_ai_json_response(result)
            
            if parsed is None:
                issues.append({
                    "prompt": prompt[:50],
                    "issue": "invalid_format",
                    "timestamp": datetime.datetime.utcnow().isoformat()
                })
        
        return issues  # Alert if len(issues) > 0

Model Version Pinning

# Pin the model version in production
MODEL_VERSION = "gpt-4o-2024-08-06"  # Specific version

def validate_model_at_startup():
    test_response = client.chat.completions.create(
        model=MODEL_VERSION,
        messages=[{"role": "user", "content": "Reply with just: OK"}]
    )
    assert "OK" in test_response.choices[0].message.content, \
        f"Model {MODEL_VERSION} not responding as expected"

4.2 Predictable Failures at the Boundaries

These failures appear early and often. Plan for them to avoid incidents.

1. Timeout Mismatches

sequenceDiagram
    participant U as User
    participant W as Web Tier (timeout: 1s)
    participant Q as Queue
    participant WK as Worker (timeout: 60s)
    participant AI as AI Model (20s)

    U->>W: Request
    W->>Q: Enqueue + return immediately
    W-->>U: "Processing..."
    Q->>WK: Forward
    WK->>AI: AI call (may take 20s)
    AI-->>WK: Response
    WK-->>U: Result via push/polling

    Note over W: Short timeout (1s) for the page
    Note over WK: Long timeout (60s) for the worker

# Timeout configuration by tier
WEB_TIER_TIMEOUT = 1.0    # Seconds — user is waiting
WORKER_TIMEOUT = 60.0     # Seconds — background worker
AI_MODEL_TIMEOUT = 30.0   # Seconds — AI API call

# NEVER do this on the web path
def bad_web_handler(request):
    # ❌ User waits 20-30s — bad UX + likely web timeout
    result = call_ai_model(request.message)
    return result

# Do this instead
def good_web_handler(request):
    # ✅ Return immediately with an ID
    job_id = submit_ai_request(request.message)
    return {"job_id": job_id, "status": "processing"}

2. Shared Rate Limits

import threading

class RateLimitManager:
    """Manage shared API quotas across multiple services."""
    
    def __init__(self, requests_per_minute: int = 100):
        self.rpm_limit = requests_per_minute
        self.request_counts = {"interactive": 0, "batch": 0}
        self._lock = threading.Lock()
    
    def allocate(self, request_type: str = "interactive") -> bool:
        """Allocate a request slot based on priority."""
        with self._lock:
            total = sum(self.request_counts.values())
            if total >= self.rpm_limit:
                return False
            
            # Interactive requests have priority
            if request_type == "batch" and \
               self.request_counts["interactive"] > self.rpm_limit * 0.7:
                return False  # Reserve 30% for interactive
            
            self.request_counts[request_type] += 1
            return True

3. Data Path — The Size Multiplication

1 document
  → N chunks (×10 typically)
    → N tokens (×500 tokens/chunk)
      → N embeddings (×vector dimension)
        = Load × 5,000 times the initial size

# Strict caps to prevent overload
MAX_DOCUMENT_SIZE_MB = 10
MAX_CHUNKS_PER_DOC = 500
MAX_TOKENS_PER_REQUEST = 4000

def validate_document(content: str) -> None:
    """Validate before processing — fail early."""
    if len(content.encode('utf-8')) > MAX_DOCUMENT_SIZE_MB * 1024 * 1024:
        raise ValueError(f"Document too large: max {MAX_DOCUMENT_SIZE_MB}MB")
    
    estimated_chunks = len(content) // 500
    if estimated_chunks > MAX_CHUNKS_PER_DOC:
        raise ValueError(f"Too many estimated chunks: {estimated_chunks}")

4. Duplicate Work

import hashlib

def get_stable_job_id(content: str, job_type: str) -> str:
    """Generate a stable ID based on content to avoid duplicates."""
    content_hash = hashlib.sha256(content.encode()).hexdigest()[:16]
    return f"{job_type}:{content_hash}"

def submit_idempotent_job(content: str, job_type: str) -> str:
    """Submit a job guaranteeing idempotence."""
    job_id = get_stable_job_id(content, job_type)
    
    existing = r.get(f"job:{job_id}")
    if existing:
        return job_id  # Return existing ID, don't reprocess
    
    r.setex(f"job:{job_id}", 86400, "processing")
    submit_ai_request(content)
    return job_id

5. Don’t Block the Whole Page

# Pattern: Progressive loading with Server-Sent Events
from flask import Flask, Response, stream_with_context

app = Flask(__name__)

@app.route('/ai-response')
def stream_ai():
    def generate():
        # 1. Send the base page immediately
        yield "data: {\"status\": \"loading\"}\n\n"
        
        # 2. Compute in background
        try:
            result = call_ai_with_retry(request.args.get('q'))
            yield f"data: {json.dumps({'result': result})}\n\n"
        except Exception:
            # 3. Fallback if AI unavailable
            yield "data: {\"result\": \"Service temporarily unavailable\", \"cached\": true}\n\n"
    
    return Response(
        stream_with_context(generate()),
        mimetype='text/event-stream'
    )

4.3 Operational Reality

The 5 Essential Metrics

Most teams monitor too many things and learn too little. If your dashboard has 50 graphs and you still don’t know why something broke, focus on these 5:

graph TD
    M1["P95 Latency\n(at user level)"] --> D["What users\nactually experience"]
    M2["User-visible errors\n(not internal retries)"] --> D
    M3["Cost per request\n(detect prompt bloat)"] --> D
    M4["Quality signal\n(confidence scores)"] --> D
    M5["Queue depth / backlog\n(is pipeline behind?)"] --> D

    style M1 fill:#4A90D9,color:#fff
    style M2 fill:#D95B5B,color:#fff
    style M3 fill:#E8A838,color:#fff
    style M4 fill:#9B59B6,color:#fff
    style M5 fill:#5BA85A,color:#fff

import time
import logging
from dataclasses import dataclass

logger = logging.getLogger(__name__)

@dataclass
class RequestMetrics:
    request_id: str
    latency_ms: float
    tokens_used: int
    cost_usd: float
    success: bool
    quality_score: float | None = None

def track_ai_request(func):
    """Decorator to automatically track metrics."""
    def wrapper(prompt: str, *args, **kwargs):
        start_time = time.time()
        request_id = str(uuid.uuid4())
        
        try:
            result = func(prompt, *args, **kwargs)
            latency_ms = (time.time() - start_time) * 1000
            
            metrics = RequestMetrics(
                request_id=request_id,
                latency_ms=latency_ms,
                tokens_used=result.get("usage", {}).get("total_tokens", 0),
                cost_usd=calculate_cost(result),
                success=True
            )
            
            log_metrics(metrics)
            
            if latency_ms > 5000:
                logger.warning(f"High latency detected: {latency_ms:.0f}ms [req:{request_id}]")
            
            return result
            
        except Exception as e:
            latency_ms = (time.time() - start_time) * 1000
            logger.error(f"AI request failed [req:{request_id}]: {e}")
            raise
    
    return wrapper

Graceful Degradation — Service Levels

flowchart TD
    A["Full Service\nPrimary model, real-time"] -->|"Primary model unavailable"| B
    B["Reduced Service\nSimpler model or cache"] -->|"Processing overloaded"| C
    C["Batch Service\nDeferred results + wait"] -->|"AI feature down"| D
    D["Minimal Service\nCore features without AI"]

    style A fill:#5BA85A,color:#fff
    style B fill:#E8A838,color:#fff
    style C fill:#E8763A,color:#fff
    style D fill:#D95B5B,color:#fff

class GracefulDegradation:
    """Manage service levels based on availability."""
    
    def __init__(self):
        self.primary_model = "gpt-4o"
        self.fallback_model = "gpt-4o-mini"
    
    def get_response(self, prompt: str) -> dict:
        # Level 1: Primary model
        try:
            result = call_with_timeout(prompt, self.primary_model, timeout=5)
            return {"result": result, "level": "full"}
        except Exception:
            pass
        
        # Level 2: Cache or simpler model
        cached = get_cached_response(prompt)
        if cached:
            return {"result": cached, "level": "cached", "stale": True}
        
        try:
            result = call_with_timeout(prompt, self.fallback_model, timeout=3)
            return {"result": result, "level": "degraded"}
        except Exception:
            pass
        
        # Level 3: Minimal response
        return {
            "result": "Service temporarily limited. Please try again.",
            "level": "minimal"
        }

The “When AI Goes Wrong” Runbook Document

Create and maintain this document. Practice it every quarter.

╔══════════════════════════════════════════════════════════╗
║              WHEN AI GOES DOWN                           ║
╠══════════════════════════════════════════════════════════╣
║ SYMPTOMS           CHECKS                ACTIONS         ║
╠══════════════════════════════════════════════════════════╣
║ High latency    → Provider status     → Circuit breaker  ║
║                 → Rate limits         → Reduce concurr.  ║
║                 → Prompt length       → Trim context     ║
╠══════════════════════════════════════════════════════════╣
║ Bad responses   → Prompt changed?    → Rollback prompt   ║
║                 → Model version?     → Pin version       ║
║                 → Retrieval data     → Check index       ║
╠══════════════════════════════════════════════════════════╣
║ Cost spikes     → Infinite loop?     → Feature flag OFF  ║
║                 → Context bloat?     → Cap tokens        ║
║                 → Traffic spike?     → Autoscaling       ║
╚══════════════════════════════════════════════════════════╝

5. Architecture Selection

5.1 The Failure Cliff

Systems never fail in a straight line. One moment everything is fine, and the next, nothing works.

xychart-beta
    title "The Capacity Cliff — Non-linear Behavior"
    x-axis ["60%", "70%", "80%", "90%", "95%", "100%"]
    y-axis "Response time (ms)" 0 --> 5000
    line [200, 250, 400, 1200, 3000, 5000]

Why the Cliff Is Steeper with AI

A 7-billion parameter model requires 15–20 GB of RAM before processing a single request.

AI memory footprint:
├── Model (weights)         : 15–20 GB
├── KV Cache (attention)    : 2–5 GB per session
├── Inference buffers       : 1–3 GB
└── Operating system        : 2–4 GB
                            ──────────
   Total at 80% utilization : 20–32 GB

At 100%: GC pause → backlog → timeout → cascade

The Honest Reaction: Intentional Load Shedding

import asyncio

class LoadShedder:
    """Intentionally shed load rather than accept everything."""
    
    def __init__(self, max_concurrent: int = 10, max_queue: int = 50):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.queue_size = 0
        self.max_queue = max_queue
    
    async def execute(self, task_func, *args):
        if self.queue_size >= self.max_queue:
            # ✅ Return "try again soon" rather than blocking everything
            return {
                "error": "service_busy",
                "message": "Service overloaded, please retry in a moment",
                "retry_after": 30
            }
        
        self.queue_size += 1
        try:
            async with self.semaphore:
                return await task_func(*args)
        finally:
            self.queue_size -= 1

Monitoring Capacity Limits

import psutil

class CapacityMonitor:
    """Monitor capacity indicators before hitting the cliff."""
    
    THRESHOLDS = {
        "memory_pct": 80,        # % RAM used
        "queue_depth": 100,      # Requests waiting
        "p95_latency_ms": 3000,  # P95 latency
        "disk_free_gb": 5        # GB free on disk
    }
    
    def check_health(self) -> dict:
        memory = psutil.virtual_memory()
        disk = psutil.disk_usage('/')
        
        alerts = []
        
        if memory.percent > self.THRESHOLDS["memory_pct"]:
            alerts.append(f"Memory: {memory.percent:.1f}% used")
        
        free_gb = disk.free / (1024**3)
        if free_gb < self.THRESHOLDS["disk_free_gb"]:
            alerts.append(f"Disk: only {free_gb:.1f}GB free")
        
        return {
            "healthy": len(alerts) == 0,
            "alerts": alerts,
            "memory_pct": memory.percent,
            "disk_free_gb": round(free_gb, 1)
        }

5.2 Performance Trade-offs That Matter

Every architectural decision is a trade-off. With AI, the ranges are wide: milliseconds to minutes, cents to dollars, perfect to deeply problematic.

quadrantChart
    title Speed vs Quality Trade-off
    x-axis Slow --> Fast
    y-axis Low quality --> High quality
    quadrant-1 Ideal (hard)
    quadrant-2 Fast response, medium quality
    quadrant-3 Avoid
    quadrant-4 Deep analysis, slow
    Password reset bot: [0.95, 0.55]
    Product recommendations: [0.7, 0.72]
    Financial analysis: [0.25, 0.90]
    Interactive assistant: [0.75, 0.75]

Fast Model vs. Smart Model

Precomputation and Semantic Cache

from langchain_openai import OpenAIEmbeddings
import numpy as np

class SemanticCache:
    """Cache based on semantic similarity — not just exact match."""
    
    def __init__(self, similarity_threshold: float = 0.95):
        self.embeddings_model = OpenAIEmbeddings(model="text-embedding-3-small")
        self.threshold = similarity_threshold
        self.cache_entries = []  # [(embedding, response)]
    
    def _cosine_similarity(self, a: list, b: list) -> float:
        a, b = np.array(a), np.array(b)
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
    
    def get(self, query: str) -> str | None:
        """Search for a semantically similar response in the cache."""
        query_embedding = self.embeddings_model.embed_query(query)
        
        for cached_embedding, cached_response in self.cache_entries:
            similarity = self._cosine_similarity(query_embedding, cached_embedding)
            if similarity >= self.threshold:
                return cached_response
        
        return None
    
    def set(self, query: str, response: str):
        """Add to semantic cache."""
        embedding = self.embeddings_model.embed_query(query)
        self.cache_entries.append((embedding, response))

The Three Limits to Define Per Feature

For each AI feature, define before building:

┌─────────────────────────────────────────────────────────┐
│ Feature: [name]                                         │
├────────────────────┬────────────────────────────────────┤
│ Latency limit      │ P95 must not exceed ___ms          │
│ (max speed)        │ If exceeded → change the design    │
├────────────────────┼────────────────────────────────────┤
│ Model budget       │ Max ___$ / 1,000 requests           │
│ (cost limit)       │ If exceeded → cache or rules       │
├────────────────────┼────────────────────────────────────┤
│ Quality bar        │ Min score ___ (out of 10)          │
│ (quality threshold)│ If below → escalate to human       │
└────────────────────┴────────────────────────────────────┘

When you hit a limit → DO NOT move the limit.
Change the design.

Non-linear Capacity Growth

100 req/min  → Synchronous API OK
200 req/min  → Cache hit rate drops, queues deepen
500 req/min  → Provider rate limits hit, latency spikes for everyone
1,000 req/min → Distributed architecture required

5.3 Building for Reality

Limits that don’t go away: models run slower than traditional code, compute costs money, outputs always vary.

These constraints are not problems to solve — they are constraints to design within.

Design for Graceful Degradation — The Levels

flowchart LR
    subgraph Normal["Normal service"]
        A["Advanced model\n+ real-time\n+ all features"]
    end
    subgraph Reduced["Reduced service"]
        B["Simpler model\n+ cached ok\n+ essential features"]
    end
    subgraph Minimal["Minimal service"]
        C["Basic features\nwithout AI\n+ product still works"]
    end

    Normal -->|"Pressure"| Reduced
    Reduced -->|"Critical pressure"| Minimal

    style Normal fill:#5BA85A,color:#fff
    style Reduced fill:#E8A838,color:#fff
    style Minimal fill:#D95B5B,color:#fff

Feature Flags for Degradation Without Redeployment

import os

class FeatureFlags:
    """Control AI features without redeployment."""
    
    @staticmethod
    def is_enabled(feature: str) -> bool:
        return os.getenv(f"FEATURE_{feature.upper()}", "true").lower() == "true"
    
    @staticmethod
    def ai_response_enabled() -> bool:
        return FeatureFlags.is_enabled("AI_RESPONSES")
    
    @staticmethod
    def advanced_model_enabled() -> bool:
        return FeatureFlags.is_enabled("ADVANCED_MODEL")

def get_response(prompt: str) -> str:
    """Response with degradation via feature flags."""
    if not FeatureFlags.ai_response_enabled():
        return get_static_fallback_response(prompt)
    
    model = "gpt-4o" if FeatureFlags.advanced_model_enabled() else "gpt-4o-mini"
    return call_ai_model(prompt, model=model)

Planning for Exit — Exit Criteria

Document these criteria now, while you’re calm:

EXIT CRITERIA FOR THIS ARCHITECTURE
────────────────────────────────────────────────────────
Trigger a review if:
□ Traffic > ___ req/min sustained
□ Monthly cost > $___ for 3 consecutive months
□ P95 latency > ___ms over 7 days
□ Fundamental requirements change (compliance, data)

Migration plan:
□ Data to preserve    : [list]
□ Data to rebuild     : [list]
□ Estimated parallel duration: [X weeks]
□ Switchover criterion: [metrics]

Basic Practices That Catch 80% of Problems

# The boring patterns that let you sleep at night

# 1. Daily test requests
import schedule

def daily_health_check():
    """Automated tests each morning."""
    test_cases = [
        ("Simple question", "expected_format_check"),
        ("Question with context", "contains_sources"),
        ("Empty edge case", "handles_gracefully"),
    ]
    
    failures = []
    for prompt, expectation in test_cases:
        try:
            result = get_ai_response(prompt)
            if not validate_expectation(result, expectation):
                failures.append({"prompt": prompt, "issue": expectation})
        except Exception as e:
            failures.append({"prompt": prompt, "error": str(e)})
    
    if failures:
        send_alert(f"{len(failures)} AI tests failed this morning", failures)

schedule.every().day.at("07:00").do(daily_health_check)

# 2. Cost alerts
def check_cost_anomaly(current_cost: float, baseline_cost: float):
    if current_cost > baseline_cost * 2:
        send_alert(f"AI cost 2x above baseline: ${current_cost:.2f}")

# 3. Request ID for complete traceability
import contextvars

request_id_var = contextvars.ContextVar('request_id', default=None)

def traced_ai_call(prompt: str) -> str:
    """AI call with request ID traced from browser to model."""
    req_id = request_id_var.get() or str(uuid.uuid4())
    
    logger.info(f"[{req_id}] AI call: {prompt[:50]}...")
    result = get_ai_response(prompt)
    logger.info(f"[{req_id}] Result: {result[:50]}...")
    
    return result

6. Your Decision Framework

Back to the Starting Point

You now understand why perfect demos hide $10,000/month GPU bills, how confident responses can be completely wrong, and how a simple API call becomes a distributed systems problem.

Constraints First — Not Wishes

flowchart TD
    A["Write the real constraints"] --> B["Real budget\n(not the deck number)"]
    A --> C["Real timeline\n(with consequences if missed)"]
    A --> D["Non-negotiable rules\n(compliance, retention, privacy)"]
    
    B --> E["The choices that remain\n= The viable path"]
    C --> E
    D --> E

    style A fill:#4A90D9,color:#fff
    style E fill:#5BA85A,color:#fff

How Constraints Eliminate Options

Constraint: No cloud services
→ Eliminates most APIs
→ Remaining: Custom build or locally hosted open-source models

Constraint: Budget < $500K
→ Eliminates custom build
→ Remaining: APIs or vendor products

Constraint: Results this quarter
→ Eliminates fine-tuning (too long)
→ Remaining: APIs with some quick wins

What remains after constraints = the path that can actually work.

The Decision Cascade

flowchart TD
    C1["Budget"] --> Decision["Build / Buy / Rent"]
    C2["Timeline"] --> Decision
    C3["Data policies"] --> Decision
    
    Decision --> A1["Sequential processing\n→ parallel\n→ distributed"]
    Decision --> A2["Slow model\n→ queue + cache\n→ don't make users wait"]
    Decision --> A3["Quality drifts\n→ daily checks\n→ no customer complaints"]

    style Decision fill:#4A90D9,color:#fff

Final Checklist Before Building

BEFORE STARTING
──────────────────────────────────────────────────────────
□ Real budget written (not the optimistic number)
□ Timeline with documented consequences
□ Compliance constraints listed
□ P95 target defined per feature
□ Budget per request defined
□ Fallback plan documented

AFTER DEPLOYMENT
──────────────────────────────────────────────────────────
□ AI calls moved off the web path
□ Circuit breaker in place
□ Cache for frequent questions
□ Monitoring for P95, visible errors, cost/req, quality
□ Automated daily tests
□ "When AI Goes Wrong" runbook
□ Exit criteria documented

What Will Happen — Be Ready

These things don’t might happen — they will happen:

The difference: Teams that succeed expected these events from the start.

7. Quick Reference and Anti-Patterns

The 10 Anti-Patterns to Avoid

Key Terms Glossary

Search Terms

ai · architecture · methods · genai · foundations · artificial · intelligence · generative · production · reality · decision · decisions · degradation · limits · options · patterns · anti-patterns · buy · capacity · cliff · constraints · define · drift · essential