GenAI MLOps: The Missing Manual for LLMs in Production - AI Engineering Hub

← Back to Home GenAI MLOps: The Missing Manual for LLMs in Production Operationalizing LLMs: prompt versioning, model routing, cost optimization, guardrails, observability, CI/CD for prompts, and governance. Table of Contents Prompt Versioning & Management Model Routing & Fallback Cost Optimization Guardrails & Safety Observability for LLMs CI/CD for Prompts Governance & Compliance Deployment Patterns Prompt Versioning & Management Prompts are code. Version them, test them, and deploy them like any other artifact. Prompt Registry from dataclasses import dataclass, field from datetime import datetime from typing import Optional import hashlib import json @dataclass class PromptVersion: id: str name: str version: str # semver: major.minor.patch template: str variables: list[str] model: str parameters: dict # temperature, max_tokens, etc. description: str author: str created_at: datetime parent_version: Optional[str] = None tags: list[str] = field(default_factory=list) metrics: dict = field(default_factory=dict) # eval results hash: str = "" def __post_init__(self): content = f"{self.template}:{json.dumps(self.parameters, sort_keys=True)}" self.hash = hashlib.sha256(content.encode()).hexdigest()[:12] class PromptRegistry: def __init__(self, storage_backend): self.storage = storage_backend # Git, DB, S3, etc. def register(self, prompt: PromptVersion) -> str: key = f"prompts/{prompt.name}/{prompt.version}.json" self.storage.put(key, prompt) # Also update "latest" pointer self.storage.put(f"prompts/{prompt.name}/latest.json", prompt.version) return prompt.id def get(self, name: str, version: str = "latest") -> PromptVersion: if version == "latest": version = self.storage.get(f"prompts/{name}/latest.json") return self.storage.get(f"prompts/{name}/{version}.json") def list_versions(self, name: str) -> list[str]: return self.storage.list(f"prompts/{name}/") # Usage registry = PromptRegistry(GitStorage("github.com/org/prompts")) prompt = PromptVersion( id="rag-qa-v1.2.0", name="rag-qa", version="1.2.0", template="""Answer using ONLY the context below. Context: {context} Question: {question} Answer:""", variables=["context", "question"], model="gpt-4o-mini", parameters={"temperature": 0.1, "max_tokens": 500}, description="RAG QA with strict grounding", author="engineer@company.com", created_at=datetime.now(), parent_version="1.1.0", tags=["rag", "qa", "production"], ) registry.register(prompt) Prompt Templating with Jinja2 from jinja2 import Environment, BaseLoader, StrictUndefined class PromptTemplate: def __init__(self, template_str: str): self.env = Environment(loader=BaseLoader(), undefined=StrictUndefined) self.template = self.env.from_string(template_str) def render(self, **kwargs) -> str: return self.template.render(**kwargs) # Advanced: conditional sections, loops, includes template_str = """ {% if system_prompt %}{{ system_prompt }}{% endif %} {% for example in few_shot_examples %} Example {{ loop.index }}: Input: {{ example.input }} Output: {{ example.output }} {% endfor %} Task: {{ task }} {% if constraints %}Constraints: {{ constraints }}{% endif %} """ template = PromptTemplate(template_str) rendered = template.render( system_prompt="You are a helpful assistant.", few_shot_examples=[{"input": "2+2", "output": "4"}], task="What is 3+3?", constraints="Answer in one word.", ) Prompt Testing in CI # tests/test_prompts.py import pytest from prompt_registry import PromptRegistry registry = PromptRegistry(LocalStorage("./prompts")) @pytest.mark.parametrize("prompt_name", ["rag-qa", "summarize", "code-gen"]) def test_prompt_renders(prompt_name): prompt = registry.get(prompt_name) # Test with sample variables sample_vars = get_sample_vars(prompt_name) rendered = PromptTemplate(prompt.template).render(**sample_vars) assert len(rendered) > 0 assert all(var in rendered for var in sample_vars.values()) @pytest.mark.parametrize("prompt_name", ["rag-qa"]) async def test_prompt_quality(prompt_name, judge_llm): prompt = registry.get(prompt_name) test_cases = load_test_cases(prompt_name) for tc in test_cases: rendered = PromptTemplate(prompt.template).render(**tc["input"]) response = await llm.generate(rendered, **prompt.parameters) score = await judge_llm.score(tc["input"], response) assert score["accuracy"] >= 4, f"Accuracy too low: {score}" assert score["instruction_following"] >= 4 Model Routing & Fallback Route requests to the optimal model based on task, cost, latency, and quality requirements. Intelligent Router from dataclasses import dataclass from enum import Enum from typing import Optional class ModelTier(Enum): PREMIUM = "premium" # GPT-4o, Claude 3.5 Sonnet STANDARD = "standard" # GPT-4o-mini, Claude 3 Haiku ECONOMY = "economy" # Local models, small open models @dataclass class ModelConfig: name: str tier: ModelTier cost_per_1m_input: float cost_per_1m_output: float max_tokens: int avg_latency_ms: int capabilities: list[str] # ["reasoning", "coding", "long_context", "vision"] MODEL_CATALOG = { "gpt-4o": ModelConfig("gpt-4o", ModelTier.PREMIUM, 5.00, 15.00, 128000, 2000, ["reasoning", "coding", "vision"]), "gpt-4o-mini": ModelConfig("gpt-4o-mini", ModelTier.STANDARD, 0.15, 0.60, 128000, 800, ["reasoning", "coding"]), "claude-3.5-sonnet": ModelConfig("claude-3.5-sonnet", ModelTier.PREMIUM, 3.00, 15.00, 200000, 2500, ["reasoning", "coding", "long_context"]), "claude-3-haiku": ModelConfig("claude-3-haiku", ModelTier.STANDARD, 0.25, 1.25, 200000, 600, ["coding", "speed"]), "llama-3.1-70b": ModelConfig("llama-3.1-70b", ModelTier.STANDARD, 0.80, 0.80, 128000, 1500, ["reasoning", "coding", "open"]), "llama-3.1-8b": ModelConfig("llama-3.1-8b", ModelTier.ECONOMY, 0.10, 0.10, 128000, 400, ["speed", "open"]), } class ModelRouter: def __init__(self, catalog=MODEL_CATALOG): self.catalog = catalog self.performance_history = {} def select_model(self, task: TaskRequest) -> str: """Select best model for task.""" candidates = self._filter_candidates(task) if not candidates: return "gpt-4o-mini" # default # Score by: quality (historical), cost, latency scored = [] for model_name in candidates: config = self.catalog[model_name] quality = self._get_quality_score(model_name, task.type) cost_score = 1 / (config.cost_per_1m_input + config.cost_per_1m_output) latency_score = 1 / config.avg_latency_ms # Weighted score score = 0.5 * quality + 0.3 * cost_score + 0.2 * latency_score scored.append((model_name, score)) scored.sort(key=lambda x: -x[1]) return scored[0][0] def _filter_candidates(self, task: TaskRequest) -> list[str]: candidates = [] for name, config in self.catalog.items(): # Check capability requirements if task.required_capabilities: if not all(cap in config.capabilities for cap in task.required_capabilities): continue # Check context window if task.estimated_tokens > config.max_tokens * 0.8: continue # Check budget if task.max_cost and config.cost_per_1m_input > task.max_cost: continue candidates.append(name) return candidates @dataclass class TaskRequest: type: str # "rag", "coding", "reasoning", "chat", "summarization" required_capabilities: list[str] = field(default_factory=list) estimated_tokens: int = 1000 max_cost: Optional[float] = None priority: str = "balanced" # "quality", "speed", "cost", "balanced" Fallback Chain class FallbackChain: def __init__(self, primary: str, fallbacks: list[str]): self.chain = [primary] + fallbacks self.current_index = 0 async def execute(self, request, executor) -> any: last_error = None for i, model in enumerate(self.chain): try: self.current_index = i return await executor(model, request) except Exception as e: last_error = e logger.warning(f"Model {model} failed: {e}, trying fallback") continue raise AllModelsFailed(f"All models failed. Last error: {last_error}") # Usage chain = FallbackChain( primary="gpt-4o", fallbacks=["claude-3.5-sonnet", "gpt-4o-mini", "llama-3.1-70b", "llama-3.1-8b"] ) result = await chain.execute(request, lambda model, req: llm_client.generate(model, req)) Cost Optimization Token Budgeting class TokenBudget: def __init__(self, daily_budget_usd: float, model_pricing: dict): self.daily_budget = daily_budget_usd self.pricing = model_pricing self.spent_today = 0.0 self.reset_time = datetime.now().replace(hour=0, minute=0, second=0) + timedelta(days=1) def can_afford(self, model: str, estimated_input: int, estimated_output: int) -> bool: config = MODEL_CATALOG[model] cost = (estimated_input * config.cost_per_1m_input + estimated_output * config.cost_per_1m_output) / 1_000_000 return self.spent_today + cost <= self.daily_budget def record_usage(self, model: str, input_tokens: int, output_tokens: int): config = MODEL_CATALOG[model] cost = (input_tokens * config.cost_per_1m_input + output_tokens * config.cost_per_1m_output) / 1_000_000 self.spent_today += cost # Check daily reset if datetime.now() >= self.reset_time: self.spent_today = 0 self.reset_time += timedelta(days=1) def get_remaining_budget(self) -> float: return max(0, self.daily_budget - self.spent_today) # Prompt compression def compress_prompt(prompt: str, target_tokens: int, compressor_model: str = "gpt-4o-mini") -> str: """Use a smaller model to compress prompt while preserving meaning.""" compression_prompt = f"""Compress this prompt to ~{target_tokens} tokens while preserving all key instructions and context: {prompt} Compressed version:""" return llm.generate(compression_prompt, model=compressor_model, max_tokens=target_tokens) Caching Strategies import redis import hashlib import json class LLMCache: def __init__(self, redis_client, ttl=3600): self.redis = redis_client self.ttl = ttl def _cache_key(self, model: str, prompt: str, params: dict) -> str: content = f"{model}:{prompt}:{json.dumps(params, sort_keys=True)}" return f"llm_cache:{hashlib.sha256(content.encode()).hexdigest()}" def get(self, model: str, prompt: str, params: dict) -> Optional[str]: key = self._cache_key(model, prompt, params) cached = self.redis.get(key) if cached: return json.loads(cached) return None def set(self, model: str, prompt: str, params: dict, response: str): key = self._cache_key(model, prompt, params) self.redis.setex(key, self.ttl, json.dumps(response)) # Semantic caching: cache based on embedding similarity def semantic_get(self, model: str, prompt: str, params: dict, threshold=0.95) -> Optional[str]: prompt_embedding = embed(prompt) # Search for similar cached prompts similar = vector_db.search(prompt_embedding, filter={"model": model}, k=5) for item in similar: if item.score >= threshold: return item.payload["response"] return None cache = LLMCache(redis_client) async def cached_generate(model, prompt, params, cache=cache): # Check exact cache cached = cache.get(model, prompt, params) if cached: return cached # Check semantic cache cached = cache.semantic_get(model, prompt, params) if cached: return cached # Generate response = await llm_client.generate(model, prompt, **params) # Store in both caches cache.set(model, prompt, params, response) vector_db.upsert(embed(prompt), {"model": model, "prompt": prompt, "response": response, "params": params}) return response Guardrails & Safety Input/Output Guards from guardrails import Guard from guardrails.hub import ( DetectPII, CompetitorCheck, ToxicLanguage, ProfanityFree, RegexMatch, ValidJson, RestrictToTopic, DetectJailbreak ) # Input guards input_guard = Guard().use_many( DetectPII(["EMAIL", "PHONE", "SSN", "CREDIT_CARD"], on_fail="filter"), DetectJailbreak(on_fail="exception"), RestrictToTopic(valid_topics=["technical", "general", "coding"], on_fail="filter"), ToxicLanguage(threshold=0.5, on_fail="exception"), ) # Output guards output_guard = Guard().use_many( DetectPII(["EMAIL", "PHONE", "SSN"], on_fail="filter"), ProfanityFree(on_fail="filter"), ValidJson(on_fail="reask"), # For structured output CompetitorCheck(competitors=["competitor1", "competitor2"], on_fail="filter"), ) async def guarded_generation(prompt, model, params, input_guard=input_guard, output_guard=output_guard): # Validate input validated_prompt = input_guard(prompt) # Generate response = await llm_client.generate(model, validated_prompt, **params) # Validate output validated_response = output_guard(response) return validated_response Custom Guardrails from guardrails import Validator, register_validator @register_validator(name="no-fabricated-citations", data_type="string") class NoFabricatedCitations(Validator): """Ensure all citations in response exist in provided context.""" def __init__(self, context_docs: list[str], on_fail="filter"): self.context_docs = context_docs self.on_fail = on_fail def validate(self, value: str) -> ValidationResult: # Extract citations like [doc_1], [doc_2] import re citations = re.findall(r'\[([^\]]+)\]', value) valid = True for cite in citations: found = any(cite in doc for doc in self.context_docs) if n