llm-ops

name	llm-ops
description	Operationalize LLM-powered applications in production — covering prompt versioning, evaluation pipelines, cost tracking, latency monitoring, hallucination detection, and safe rollout of model upgrades.
argument-hint	["LLM provider","application type","traffic volume","cost budget","safety requirements"]
allowed-tools	Read, Write, Bash

LLM Operations (LLMOps)

LLMs are not like regular software. They are probabilistic, expensive, slow, and can degrade silently when the model is updated under you. LLMOps is the discipline of running LLM applications reliably: versioning prompts like code, evaluating outputs systematically, tracking costs, and catching regressions before users do.

Process

1. Version every prompt — prompts are code; track changes in git with evaluation results.
2. Build an eval pipeline — automated quality checks on every prompt change.
3. Instrument cost and latency — per-request token counts, latency percentiles, cost per operation.
4. Detect hallucinations and failures — output quality checks in production.
5. Canary model upgrades — never switch models cold; shadow-test and ramp traffic.
6. Set up fallback chains — primary model → fallback model → cached response.
7. Rate limit and queue — protect upstream LLM API from traffic spikes.
8. Monitor for prompt injection — detect adversarial user inputs targeting your prompts.

Output Format

Prompt Registry & Versioning

# prompts/registry.py
import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
import yaml

@dataclass
class PromptVersion:
    name: str
    version: str          # semver: 1.0.0
    content: str          # The actual prompt template
    model: str            # e.g. "gpt-4o", "claude-3-5-sonnet-20241022"
    temperature: float
    max_tokens: int
    content_hash: str     # SHA256 of content for integrity
    created_at: str
    eval_score: Optional[float] = None   # From eval pipeline
    notes: str = ""
    
    @classmethod
    def from_file(cls, path: Path) -> "PromptVersion":
        data = yaml.safe_load(path.read_text())
        content = data["content"]
        return cls(
            name=data["name"],
            version=data["version"],
            content=content,
            model=data.get("model", "gpt-4o"),
            temperature=data.get("temperature", ),
            max_tokens=data.get(, ),
            content_hash=hashlib.sha256(content.encode()).hexdigest()[:],
            created_at=data.get(, datetime.now(timezone.utc).isoformat()),
            eval_score=data.get(),
            notes=data.get(, ),
        )
    
     () -> :
        
         .content.(**variables)


 :
    
    
     ():
        . = Path(prompts_dir)
        ._cache: [, [, PromptVersion]] = {}
        ._load_all()
    
     ():
         file  ..glob():
            prompt = PromptVersion.from_file(file)
             prompt.name   ._cache:
                ._cache[prompt.name] = {}
            ._cache[prompt.name][prompt.version] = prompt
    
     () -> PromptVersion:
        versions = ._cache.get(name, {})
          versions:
             KeyError()
        
         version == :
            
            version = (versions.keys(), key= v: [(x)  x  v.split()])[-]
        
         versions[version]
    
     () -> []:
         (._cache.get(name, {}).keys())

# prompts/summarizer/v1.2.0.yaml
name: document-summarizer
version: "1.2.0"
model: claude-3-5-sonnet-20241022
temperature: 0.1
max_tokens: 500
eval_score: 0.87
notes: "Added instruction to preserve technical terms. Improved accuracy on code docs."
created_at: "2024-02-15T10:00:00Z"

content: |
  You are a precise technical summarizer. Your job is to summarize the document below.
  
  Rules:
  - Maximum {max_words} words
  - Preserve all technical terms, product names, and numbers exactly as written
  - Use bullet points for lists of features or steps
  - Start with a one-sentence overview, then bullet points   
                 
            
  
  
  {}
  
  

Evaluation Pipeline

# evals/pipeline.py
import anthropic
import openai
from dataclasses import dataclass
from typing import Callable
import json
import re

@dataclass
class EvalCase:
    input_variables: dict
    expected_output: str         # Golden reference
    acceptance_criteria: list[str]   # Things output must contain/do
    rejection_criteria: list[str]    # Things output must NOT contain

@dataclass  
class EvalResult:
    case_id: str
    prompt_version: str
    model: str
    output: str
    score: float               # 0.0 to 1.0
    passed: bool
    failures: list[str]
    latency_ms: float
    input_tokens: int
    output_tokens: int
    cost_usd: float


class LLMEvalPipeline:
    def __init__(self, judge_model: str = "claude-3-5-sonnet-20241022"):
        self.judge = anthropic.Anthropic()
        self.judge_model = judge_model
    
     () -> :
        results = []
        
         i,   (test_cases):
             run  (n_runs):
                result = ._evaluate_single(prompt, , )
                results.append(result)
        
        
        passed = (  r  results  r.passed)
        scores = [r.score  r  results]
        
         {
            : prompt.name,
            : prompt.version,
            : prompt.model,
            : (test_cases),
            : n_runs,
            : passed / (results),
            : (scores) / (scores),
            : (scores),
            : (scores)[((scores) * )],
            : (r.cost_usd  r  results),
            : (r.latency_ms  r  results) / (results),
            : (r.latency_ms  r  results)[((results) * )],
            : [(r)  r  results],
        }
    
     () -> EvalResult:
         time
        
        rendered = prompt.render(**.input_variables)
        
        start = time.perf_counter()
        response = ._call_model(prompt.model, rendered, prompt.temperature, prompt.max_tokens)
        latency_ms = (time.perf_counter() - start) * 
        
        output = response[]
        failures = []
        
        
         criterion  .acceptance_criteria:
              ._check_criterion(output, criterion):
                failures.append()
        
        
         criterion  .rejection_criteria:
             ._check_criterion(output, criterion):
                failures.append()
        
        
        judge_score = ._llm_judge(
            input_text=rendered,
            expected=.expected_output,
            actual=output,
        )
        
        score = judge_score * ( - (failures) * )
        score = (, (, score))
        
         EvalResult(
            case_id=case_id,
            prompt_version=prompt.version,
            model=prompt.model,
            output=output,
            score=score,
            passed=(failures) ==   score >= ,
            failures=failures,
            latency_ms=latency_ms,
            input_tokens=response[],
            output_tokens=response[],
            cost_usd=._compute_cost(prompt.model, response[], response[]),
        )
    
     () -> :
        
        judge_prompt = 

        response = .judge.messages.create(
            model=.judge_model,
            max_tokens=,
            messages=[{: , : judge_prompt}]
        )
        
        :
            scores = json.loads(response.content[].text)
             (scores[] + scores[] + scores[]) / 
         Exception:
               
    
     () -> :
        
        pricing = {
            : {: , : },
            : {: , : },
            : {: , : },
            : {: , : },
        }
        p = pricing.get(model, {: , : })
         (input_tokens * p[] + output_tokens * p[]) / 

Cost & Latency Instrumentation

# instrumentation/llm_client.py
import time
import anthropic
from prometheus_client import Counter, Histogram, Gauge

# Metrics
llm_requests = Counter("llm_requests_total", "LLM API calls", ["model", "prompt_name", "status"])
llm_latency = Histogram("llm_latency_seconds", "LLM request latency", ["model", "prompt_name"],
    buckets=[0.5, 1, 2, 5, 10, 30, 60])
llm_tokens = Counter("llm_tokens_total", "Tokens consumed", ["model", "prompt_name", "token_type"])
llm_cost = Counter("llm_cost_usd_total", "Estimated cost in USD", ["model", "prompt_name"])

class InstrumentedLLMClient:
    """Wraps LLM API with observability."""
    
    def __init__(self):
        self.client = anthropic.Anthropic()
        self.prompt_registry = PromptRegistry()
    
    async def complete(
        self,
        prompt_name: str,
        variables: dict,
        prompt_version: str = "latest",
        user_id:  = ,
    ) -> :
        prompt = .prompt_registry.get(prompt_name, prompt_version)
        rendered = prompt.render(**variables)
        
        start = time.perf_counter()
        status = 
        
        :
            response = .client.messages.create(
                model=prompt.model,
                max_tokens=prompt.max_tokens,
                temperature=prompt.temperature,
                messages=[{: , : rendered}],
                metadata={: user_id  },
            )
            
            output = response.content[].text
            input_tokens = response.usage.input_tokens
            output_tokens = response.usage.output_tokens
            
         anthropic.RateLimitError:
            status = 
            
         anthropic.APIError  e:
            status = 
            
        :
            latency = time.perf_counter() - start
            llm_requests.labels(model=prompt.model, prompt_name=prompt_name, status=status).inc()
            llm_latency.labels(model=prompt.model, prompt_name=prompt_name).observe(latency)
        
        
        llm_tokens.labels(model=prompt.model, prompt_name=prompt_name, token_type=).inc(input_tokens)
        llm_tokens.labels(model=prompt.model, prompt_name=prompt_name, token_type=).inc(output_tokens)
        
        cost = ._compute_cost(prompt.model, input_tokens, output_tokens)
        llm_cost.labels(model=prompt.model, prompt_name=prompt_name).inc(cost)
        
         output

Hallucination & Quality Guardrails

# guardrails/output_checker.py
import re
from dataclasses import dataclass

@dataclass
class GuardrailResult:
    passed: bool
    violations: list[str]
    filtered_output: str

class OutputGuardrails:
    """Post-process LLM outputs to catch common failure modes."""
    
    def check(self, output: str, context: dict) -> GuardrailResult:
        violations = []
        filtered = output
        
        # 1. Detect confident hallucination signals
        hallucination_phrases = [
            r"as of \d{4}",          # "As of 2019" when we're in 2024
            r"I don't have access",   # Model refusing when it should answer
            r"I cannot browse",
            r"my knowledge cutoff",   # Fine for chatbots, bad for RAG with current data
        ]
        for pattern in hallucination_phrases:
            if re.search(pattern, output, re.IGNORECASE):
                violations.append(f"Hallucination signal: matched '{pattern}'")
        
        # 2. Check factual grounding (if RAG context provided)
        if context.get("source_documents"):
            citation_score = self._check_grounding(output, context["source_documents"])
            if citation_score < 0.3:
                violations.append(f"Low grounding score:  — output may not be supported by sources")
        
        
        pii_patterns = {
            : ,
            : ,
            : ,
        }
         pii_type, pattern  pii_patterns.items():
             re.search(pattern, output):
                violations.append()
                filtered = re.sub(pattern, , filtered)
        
        
        injection_signals = [
            ,
            ,
            ,
            ,
        ]
         signal  injection_signals:
             signal.lower()  output.lower():
                violations.append()
        
        
         (output.split()) < :
            violations.append()
        
         GuardrailResult(
            passed=(violations) == ,
            violations=violations,
            filtered_output=filtered,
        )
    
     () -> :
        
        output_words = (output.lower().split())
        source_words = (.join(source_docs).lower().split())
          output_words:
             
        overlap = output_words & source_words
         (overlap) / (output_words)

Model Upgrade Canary

# deployment/model_canary.py
import random
from dataclasses import dataclass

@dataclass
class ModelCanaryConfig:
    current_model: str        # "claude-3-5-sonnet-20241022"
    candidate_model: str      # "claude-3-7-sonnet-20250219"
    candidate_pct: float = 0.05   # 5% traffic to candidate
    compare_outputs: bool = True  # Log both outputs for offline comparison

class ModelCanaryRouter:
    def __init__(self, config: ModelCanaryConfig):
        self.config = config
        self._shadow_log = []
    
    def select_model(self, request_id: str) -> str:
        """Deterministic routing by request ID."""
        hash_val = int(hashlib.md5(request_id.encode()).hexdigest(), 16)
        pct = (hash_val % 1000) / 1000
        
        if pct < self.config.candidate_pct:
            return self.config.candidate_model
        return self.config.current_model
    
    async def complete_with_shadow(self, prompt: str, request_id: ) -> :
        
        primary_model = .config.current_model
        primary_output =  ._call_model(primary_model, prompt)
        
         .config.compare_outputs  random.random() < :
            
            candidate_output =  ._call_model(.config.candidate_model, prompt)
            ._shadow_log.append({
                : request_id,
                : primary_output,
                : candidate_output,
            })
        
         primary_output

Rules

• Prompts are code — version them in git, review changes, require eval results before merging.
• Evals before every prompt change — no gut-feel deployments; measure before and after.
• Never hard-code model names in application code — use the registry; model upgrades should be a config change.
• Track cost per operation, not just total — "summarize" costs $0.002, "analyze document" costs $0.08; know both.
• Set absolute cost budgets — cap monthly LLM spend at the infrastructure level, not just in monitoring.
• Latency SLOs must account for LLM variability — P95, not mean; LLM latency tail is long.
• Fallback chains are mandatory — primary model → smaller/faster model → cached or degraded response.
• Log every prompt + response — you need this to debug failures, detect drift, and build eval datasets.
• Test model upgrades with shadow traffic — never switch production models cold.
• Prompt injection is a real attack vector — sanitize user inputs before interpolating into prompts.