⚙️ 2.04 – Production AI

📑 On this page

Streaming Responses
Caching & Rate Limiting
Cost Optimization
Monitoring & Logging
Prompt Management
Exercises

📡 Streaming Responses

📌 Why Stream?

LLMs can take 2-10 seconds to generate full responses. Streaming shows tokens as they're generated, improving perceived latency and user experience.

OpenAI Streaming (Python):

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True
)

for chunk in response:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

FastAPI Streaming Endpoint:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI

app = FastAPI()

def generate_stream(prompt):
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    for chunk in response:
        if chunk.choices[0].delta.content:
            yield f"data: {chunk.choices[0].delta.content}\n\n"
    yield "data: [DONE]\n\n"

@app.get("/stream")
async def stream(prompt: str):
    return StreamingResponse(generate_stream(prompt), media_type="text/event-stream")

💾 Caching & Rate Limiting

📌 Caching Strategies

Strategy	When to use	Example
Exact-match cache	Repeated identical queries	FAQ chatbot
Semantic cache	Similar but not identical queries	Customer support (same question, different wording)
Time-based cache	Data that changes slowly	Weekly report summaries

Simple Redis Cache:

import redis
import hashlib

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

def cached_completion(prompt, model="gpt-4"):
    # Create cache key
    cache_key = hashlib.md5(f"{prompt}{model}".encode()).hexdigest()
    
    # Check cache
    cached = redis_client.get(cache_key)
    if cached:
        return cached
    
    # Call API
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    )
    answer = response.choices[0].message.content
    
    # Store in cache (60 minutes TTL)
    redis_client.setex(cache_key, 3600, answer)
    return answer

📌 Semantic Cache (using embeddings):

from sentence_transformers import SentenceTransformer, util

model = SentenceTransformer('all-MiniLM-L6-v2')
cache_store = []  # In production, use vector DB

def semantic_cached_completion(prompt, threshold=0.95):
    prompt_embedding = model.encode(prompt)
    
    # Check for similar prompts in cache
    for cached_prompt, cached_response in cache_store:
        similarity = util.cos_sim(prompt_embedding, cached_prompt)[0][0]
        if similarity > threshold:
            return cached_response
    
    response = call_llm(prompt)
    cache_store.append((prompt_embedding, response))
    return response

💰 Cost Optimization

📌 Cost-Saving Techniques

Use cheaper models: GPT-3.5 ($0.50/1M tokens) vs GPT-4 ($30/1M tokens) – 60x cheaper!
Prompt compression: Reduce token count without losing meaning
Smaller context windows: Trim unnecessary history
Caching: Don't re-answer the same questions
Batch processing: Combine multiple requests

Model	Input cost/1M tokens	Output cost/1M tokens	When to use
GPT-4o	$5.00	$15.00	Complex reasoning, high accuracy
GPT-4 Turbo	$10.00	$30.00.\]
GPT-3.5 Turbo	$0.50	$1.50	Simple tasks, classification, prototyping
Claude 3 Haiku	$0.25	$1.25	Fast, cheap, good for basic tasks
Llama 3 (local)	~$0.00	~$0.00	Privacy, scale, fine-tuning

💡 Practical Strategy:

Use GPT-3.5 or Claude Haiku for 80% of requests
Only route to GPT-4 when confidence is low or complexity is high
This "router" pattern reduces costs by 80-90%

Smart Model Router:

def smart_router(question):
    # First, try GPT-3.5
    cheap_response = call_gpt35(question)
    confidence = evaluate_confidence(cheap_response)
    
    if confidence > 0.8:
        return cheap_response, "gpt-3.5"
    else:
        # Upgrade to GPT-4
        expensive_response = call_gpt4(question)
        return expensive_response, "gpt-4"

📊 Monitoring & Logging

📌 Key Metrics to Track

Latency: Time to first token (TTFT) and total response time
Token usage: Input tokens, output tokens, total cost
Error rates: Rate limits, timeouts, API errors
Usage patterns: Most common questions, peak hours
Quality: User feedback, relevance scores

Structured Logging (Python):

import logging
import time

logging.basicConfig(level=logging.INFO)

def log_completion(prompt, response, model, tokens_used, latency_ms):
    logging.info({
        "event": "llm_completion",
        "model": model,
        "prompt_length": len(prompt),
        "response_length": len(response),
        "tokens": tokens_used,
        "latency_ms": latency_ms,
        "timestamp": time.time()
    })

📌 Observability Stack

OpenTelemetry: Distributed tracing
LangSmith: LLM-specific debugging and evaluation
Prometheus + Grafana: Metrics dashboard
Lunary or Helicone: LLM observability (open-source)

📝 Prompt Management

📌 Version Control for Prompts

Treat prompts like code – version them, test changes, roll back when needed.

Prompt Registry Pattern:

PROMPTS = {
    "classify_v1": "Classify the sentiment as positive/negative/neutral: {text}",
    "classify_v2": "You are a sentiment expert. Analyze this text and respond with exactly one word: POSITIVE, NEGATIVE, or NEUTRAL. Text: {text}",
    "summarize_v1": "Summarize: {text}",
}

def get_prompt(name, version="latest"):
    if version == "latest":
        # Get the highest version number
        versions = [v for v in PROMPTS if v.startswith(name)]
        version = max(versions)
    return PROMPTS[version]

📌 Prompt Testing & A/B Testing

Online A/B testing: Split traffic between prompt versions
Offline evaluation: Run both versions on test set, compare scores
Canary deployments: Roll out new prompts to 1% of users first

📌 Production Readiness Checklist

✅ Streaming for better user experience
✅ Caching for repeated queries (semantic or exact-match)
✅ Rate limiting to stay within API quotas
✅ Model routing to optimize cost/quality
✅ Monitoring and logging (latency, tokens, errors)
✅ Prompt version control and testing
✅ Error handling with graceful fallbacks

✍️ Exercises

Exercise 2.04.1 – Implement caching

Add Redis caching to an existing LLM function. Measure the latency difference on the second identical query.

First query: 2-5 seconds (API call). Cached query: < 10ms (Redis). That's 200-500x faster!

Exercise 2.04.2 – Design a model router

Design a system that uses GPT-3.5 for simple questions and GPT-4 for complex ones. How would you classify complexity?

Approach 1 (Rule-based): Route to GPT-4 if question length > 200 words, contains math, or asks for code.
Approach 2 (LLM classification): Use small model (e.g., GPT-3.5) to score complexity 1-10, route to GPT-4 if score > 7.
Approach 3 (Confidence threshold): Try GPT-3.5 first, check confidence, upgrade if needed.

Exercise 2.04.3 – Cost estimation

Estimate monthly costs: 100,000 queries, each averages 500 input tokens and 200 output tokens. Calculate cost for GPT-3.5 vs GPT-4.

GPT-3.5: (100k × $0.50/1M) × 500 + (100k × $1.50/1M) × 200 = $25 + $30 = $55/month
GPT-4: (100k × $10/1M) × 500 + (100k × $30/1M) × 200 = $500 + $600 = $1,100/month
GPT-4 is 20x more expensive!

📌 Key Takeaways

Streaming dramatically improves user-perceived latency
Caching (exact or semantic) reduces cost and latency
Model routing (cheap model first, upgrade if needed) saves 80-90% on costs
Monitor tokens, latency, and errors to optimize
Version and test prompts systematically

🎉 Level 2 Complete!

You've finished Level 2 of the AI course. You can now:

✅ Use advanced prompting techniques (CoT, ToT, ReAct)
✅ Build production RAG systems with optimal chunking and retrieval
✅ Use LangChain and LlamaIndex frameworks
✅ Deploy AI applications with caching, streaming, and cost optimization

Next: Level 3 – Professional (Architecture, Security, Fine-Tuning)