Building Production AI Applications: What They Don't Tell You

The gap between demo and production

Your demo works perfectly. A user types a question, the LLM responds in 2 seconds, everyone is impressed. Then you ship to production and reality hits:

• Response time varies from 0.5s to 30s
• The model occasionally hallucinates critical facts
• Your API bill is 10x what you estimated
• Users find prompt injection attacks within hours
• The model that worked great in testing starts giving different answers
• Rate limits kick in during traffic spikes

This guide covers what you need to know to bridge the gap.

1. Prompt management

The problem

Prompts scattered across your codebase become unmaintainable:

# DON'T DO THIS
response = client.chat.completions.create(
    model="gpt-5.6",
    messages=[
        {"role": "system", "content": f"You are a helpful assistant for {company}. Always respond in {language}. Use this context: {context}. Format as JSON with keys: summary, sentiment, actions. Never mention you are an AI. Be concise but thorough. The current date is {date}..."},
        {"role": "user", "content": user_input}
    ]
)

The solution

Centralize and version your prompts:

# prompts/summarize_v3.py
SYSTEM_PROMPT = """You are a content summarization assistant.

Rules:
- Summarize the input in 3-5 sentences
- Capture the main argument and key supporting points
- Do not add information not present in the source
- Format as JSON: {"summary": str, "key_points": list[str], "sentiment": "positive"|"neutral"|"negative"}

Context: {context}
"""

def summarize(text, context=""):
    return client.chat.completions.create(
        model="gpt-5.6",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT.format(context=context)},
            {"role": "user", "content": text}
        ],
        response_format={"type": "json_object"},
        temperature=0.3,
        max_tokens=500
    )

Prompt versioning

Track prompt changes like code changes:

PROMPT_REGISTRY = {
    "summarize": {
        "v1": "Summarize this text: {text}",  # Naive
        "v2": "Summarize in 3 sentences: {text}",  # Added constraint
        "v3": SYSTEM_PROMPT,  # Current (structured output)
    }
}

def get_prompt(name, version="latest"):
    if version == "latest":
        versions = list(PROMPT_REGISTRY[name].keys())
        return PROMPT_REGISTRY[name][versions[-1]]
    return PROMPT_REGISTRY[name][version]

2. Evaluation

The problem

Without systematic evaluation, you can't know if a prompt change improved or degraded quality.

Build an eval suite

import json

# eval_tests/summarize_tests.json
EVAL_CASES = [
    {
        "input": "The stock market fell 3% today due to inflation concerns...",
        "expected_keys": ["summary", "key_points", "sentiment"],
        "expected_sentiment": "negative",
        "max_summary_length": 500,
    },
    {
        "input": "The new product launch was a success with 10K signups...",
        "expected_keys": ["summary", "key_points", "sentiment"],
        "expected_sentiment": "positive",
        "max_summary_length": 500,
    },
]

def run_evals(model="gpt-5.6"):
    results = []
    for case in EVAL_CASES:
        response = summarize(case["input"])
        output = json.loads(response.choices[0].message.content)
        
        checks = {
            "has_all_keys": all(k in output for k in case["expected_keys"]),
            "sentiment_correct": output.get("sentiment") == case["expected_sentiment"],
            "summary_length_ok": len(output.get("summary", "")) <= case["max_summary_length"],
        }
        
        results.append({
            "input": case["input"][:50] + "...",
            "passed": all(checks.values()),
            "checks": checks,
        })
    
    passed = sum(1 for r in results if r["passed"])
    print(f"Eval results: {passed}/{len(results)} passed")
    return results

Continuous evaluation

Run evals on every prompt change and periodically in production:

# Run evals daily
def daily_eval():
    results = run_evals()
    pass_rate = sum(1 for r in results if r["passed"]) / len(results)
    
    if pass_rate < 0.9:
        alert(f"Eval pass rate dropped to {pass_rate:.0%}")
    
    # Log for trending
    log_eval_result(pass_rate, results)

3. Monitoring

What to track

from dataclasses import dataclass, field
from datetime import datetime
import time

@dataclass
class LLMCallMetrics:
    model: str
    prompt_tokens: int
    completion_tokens: int
    latency_ms: int
    status: str  # "success", "error", "timeout"
    error_type: str = ""
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    
    @property
    def cost(self):
        rates = {"gpt-5.6": (1.4e-6, 4.4e-6), "glm-5.2": (0.5e-6, 1.5e-6)}
        in_rate, out_rate = rates.get(self.model, (1e-6, 3e-6))
        return (self.prompt_tokens * in_rate) + (self.completion_tokens * out_rate)

def track_llm_call(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            response = func(*args, **kwargs)
            metrics = LLMCallMetrics(
                model=kwargs.get("model", "unknown"),
                prompt_tokens=response.usage.prompt_tokens,
                completion_tokens=response.usage.completion_tokens,
                latency_ms=int((time.time() - start) * 1000),
                status="success"
            )
            log_metrics(metrics)
            return response
        except Exception as e:
            metrics = LLMCallMetrics(
                model=kwargs.get("model", "unknown"),
                prompt_tokens=0,
                completion_tokens=0,
                latency_ms=int((time.time() - start) * 1000),
                status="error",
                error_type=type(e).__name__
            )
            log_metrics(metrics)
            raise
    return wrapper

Key metrics to monitor

| Metric | Why it matters | Alert threshold | |---|---|---| | Error rate | Provider outages, key issues | > 5% | | P95 latency | User experience | > 10s | | Cost per request | Budget tracking | > 2x average | | Token usage | Cost forecasting | > 2x daily average | | Hallucination rate | Quality degradation | > 1% | | Empty response rate | Silent failures | > 0.5% |

4. Cost control

Token budgets

class TokenBudget:
    def __init__(self, daily_limit=1_000_000):
        self.daily_limit = daily_limit
        self.used = 0
        self.date = datetime.now().date()
    
    def check_and_deduct(self, tokens):
        # Reset daily
        if datetime.now().date() != self.date:
            self.date = datetime.now().date()
            self.used = 0
        
        if self.used + tokens > self.daily_limit:
            raise Exception(f"Token budget exceeded: {self.used}/{self.daily_limit}")
        
        self.used += tokens

Model routing by complexity

Route simple tasks to cheaper models:

def route_model(prompt, max_length=None):
    # Simple, short prompts → cheap model
    if len(prompt) < 200 and "?" in prompt:
        return "glm-5.2"  # $0.50/$1.50 per 1M
    
    # Code-related → specialized model
    if "code" in prompt.lower() or "function" in prompt.lower():
        return "deepseek-v3"  # $0.27/$1.10 per 1M
    
    # Long context → model with large window
    if max_length and max_length > 64000:
        return "claude-opus-5"  # 200K context
    
    # Default → general purpose
    return "gpt-5.6"  # $1.40/$4.40 per 1M

Caching

Cache identical requests to avoid redundant API calls:

import hashlib
import json
from functools import lru_cache

def cache_key(model, messages, **kwargs):
    content = json.dumps({"model": model, "messages": messages, **kwargs})
    return hashlib.sha256(content.encode()).hexdigest()

# Use Redis for production
response_cache = {}

def cached_chat(model, messages, **kwargs):
    key = cache_key(model, messages, **kwargs)
    if key in response_cache:
        return response_cache[key]
    
    response = client.chat.completions.create(model=model, messages=messages, **kwargs)
    response_cache[key] = response
    return response

5. Security

Prompt injection defense

def sanitize_input(user_input):
    # Remove common injection patterns
    dangerous_patterns = [
        "ignore previous instructions",
        "system:",
        "you are now",
        "new instructions:",
    ]
    
    lowered = user_input.lower()
    for pattern in dangerous_patterns:
        if pattern in lowered:
            return f"[FILTERED: potential prompt injection detected]"
    
    return user_input

def build_safe_prompt(system_prompt, user_input):
    sanitized = sanitize_input(user_input)
    return [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": f"<user_input>{sanitized}</user_input>"},
    ]

Output validation

Always validate LLM output before using it:

def safe_parse_json(response_text):
    try:
        data = json.loads(response_text)
    except json.JSONDecodeError:
        return None
    
    # Validate structure
    if not isinstance(data, dict):
        return None
    if "summary" not in data:
        return None
    
    # Sanitize values
    for key, value in data.items():
        if isinstance(value, str):
            data[key] = value[:1000]  # Limit length
        elif isinstance(value, list):
            data[key] = value[:10]  # Limit array size
    
    return data

6. Graceful degradation

When the LLM fails, your app should still work:

def chat_with_degradation(user_input):
    try:
        # Try primary model
        return llm_chat("gpt-5.6", user_input)
    except Exception:
        try:
            # Fall back to cheaper model
            return llm_chat("glm-5.2", user_input)
        except Exception:
            # Fall back to cached/static response
            return get_cached_response(user_input)

7. Testing

Unit tests for prompts

def test_summarize_prompt():
    response = summarize("The meeting was productive. We discussed Q3 targets.")
    data = json.loads(response.choices[0].message.content)
    
    assert "summary" in data
    assert "sentiment" in data
    assert data["sentiment"] in ["positive", "neutral", "negative"]
    assert len(data["summary"]) > 0

Load testing

import concurrent.futures
import time

def load_test(model, num_requests=50):
    def make_request():
        start = time.time()
        try:
            response = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": "Hello"}],
                max_tokens=10
            )
            return time.time() - start
        except:
            return None
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        latencies = list(executor.map(lambda _: make_request(), range(num_requests)))
    
    successful = [l for l in latencies if l is not None]
    print(f"Success rate: {len(successful)}/{num_requests}")
    print(f"P50: {sorted(successful)[len(successful)//2]:.2f}s")
    print(f"P95: {sorted(successful)[int(len(successful)*0.95)]:.2f}s")

Getting started

Riven provides the infrastructure — 75+ models, automatic failover, rate limiting, and usage tracking — so you can focus on your application logic. Start building free.