Cost, latency & observability

Cost, latency, and observability

A RAG pipeline in production is a distributed system with multiple stages, each contributing latency and cost. Without per-stage measurement, optimization is guesswork — you cannot know whether to optimize retrieval or generation without knowing which dominates your latency budget. The typical production target: p90 time-to-first-token under 2 seconds, cost per query under $0.02 for standard workloads.

Per-stage latency breakdown

A standard RAG query passes through four stages, each with characteristic latency:

• Embedding the query: ~50ms (API round-trip to embedding model; network overhead dominates over compute)
• Vector retrieval: ~20ms (pgvector HNSW index scan over 1M vectors; scales sub-linearly with index size)
• Reranking: ~100ms (cross-encoder inference over 20 candidates; scales linearly with candidate count)
• Generation: ~800ms to first token (Claude Sonnet with 2K context; dominated by model inference, scales with prompt length)

Total for a well-optimized pipeline: ~970ms. Generation alone accounts for 82% of the total. This means that optimizing retrieval from 20ms to 10ms (a 50% improvement) saves 10ms total — irrelevant. Optimizing generation (via prompt compression, model tiering, or caching) has 10–100x more impact on user-perceived latency.

Measure p50, p90, and p99 independently. A p50 of 900ms with a p99 of 8 seconds indicates tail latency problems — likely caused by cold starts (first request after idle period), cache misses (unusual queries), retry storms (upstream timeouts triggering retries), or slow cross-encoder runs on unusually long documents.

Caching strategies

Semantic caching: cache the full generated response keyed on query embedding similarity. When a new query arrives, compute its embedding and check if any cached query embedding is within cosine distance threshold (typically 0.95 similarity). If yes, return the cached response without executing the pipeline at all.

Hit rates of 15–30% are typical for enterprise knowledge bases where users ask similar questions repeatedly. Customer support, internal tooling, and documentation QA workloads benefit most. Creative or open-ended queries cache poorly.

Embedding cache: store computed embeddings for frequently-queried strings in a fast key-value store (Redis, in-memory dict). Eliminates the 50ms embedding API call on cache hit. Useful when users submit the same exact query multiple times (search-style interfaces, autocomplete).

Prompt caching (Anthropic): Anthropic's server-side prompt caching stores the static prefix of your prompt (system instructions, few-shot examples, retrieval template) after the first request. Subsequent requests sharing the same prefix skip processing those tokens, reducing both latency and cost.

Two applications in RAG:

• At index time for contextual retrieval: the full document is the cached prefix. Processing 100 chunks from the same document caches the document after chunk 1, so chunks 2–100 read from cache. Reduces per-chunk cost by ~90%.
• At query time: cache the system prompt + retrieval instructions. The cached portion is free on subsequent queries. Saves 100–300ms of processing time on the static prefix.

import hashlib
import numpy as np
import time
from typing import Optional

class SemanticCache:
    """Cache RAG responses keyed on query embedding similarity."""

    def __init__(self, similarity_threshold: float = 0.95,
                 ttl_seconds: int = 3600, max_entries: int = 10000):
        self.threshold = similarity_threshold
        self.ttl = ttl_seconds
        self.max_entries = max_entries
        self.entries: list[dict] = []

    def get(self, query_embedding: list[float]) -> Optional[str]:
        """Return cached response if a sufficiently similar query exists."""
        query_vec = np.array(query_embedding)
        now = time.time()

        for entry in self.entries:
            if now - entry["timestamp"] > self.ttl:
                continue
            similarity = np.dot(query_vec, entry["embedding"]) / (
                np.linalg.norm(query_vec) * np.linalg.norm(entry["embedding"])
            )
            if similarity >= self.threshold:
                entry["hit_count"] += 1
                return entry["response"]
        return None

    def put(self, query_embedding: list[float], response: str):
        """Cache a response with its query embedding."""
        if len(self.entries) >= self.max_entries:
            self.entries.sort(key=lambda e: e["timestamp"])
            self.entries = self.entries[self.max_entries // 2:]

        self.entries.append({
            "embedding": np.array(query_embedding),
            "response": response,
            "timestamp": time.time(),
            "hit_count": 0
        })

Cost optimization with model tiering

Not every query requires the most capable model. Route by estimated complexity:

• Simple factoid queries (single-hop, answer clearly stated in context): Claude Haiku — $0.25/M input, $1.25/M output tokens
• Standard reasoning queries (synthesis across 2–3 sources): Claude Sonnet — $3/M input, $15/M output tokens
• Complex multi-step synthesis (long-form, multi-document reasoning): Claude Opus — $15/M input, $75/M output tokens

A routing classifier assigns each query to a tier. For workloads where 60% of queries are simple factoid lookups, routing those to Haiku reduces average generation cost by 40–50% with negligible quality loss on the simple tier.

Implement circuit breakers: if the accumulated cost for a single query exceeds a threshold ($0.05 for standard, $0.20 for complex), terminate the pipeline and return the best available answer with a confidence disclaimer. This prevents runaway costs from agentic loops that grade, rewrite, and re-retrieve multiple times.

Per-query cost and latency instrumentation

import time
import hashlib
from dataclasses import dataclass, field

@dataclass
class StageMetrics:
    name: str
    latency_ms: float
    input_tokens: int = 0
    output_tokens: int = 0

@dataclass
class QueryTrace:
    query_id: str
    stages: list[StageMetrics] = field(default_factory=list)

    def add_stage(self, name: str, latency_ms: float,
                  input_tokens: int = 0, output_tokens: int = 0):
        self.stages.append(StageMetrics(
            name=name, latency_ms=latency_ms,
            input_tokens=input_tokens, output_tokens=output_tokens
        ))

    @property
    def total_latency_ms(self) -> float:
        return sum(s.latency_ms for s in self.stages)

    @property
    def total_input_tokens(self) -> int:
        return sum(s.input_tokens for s in self.stages)

    @property
    def total_output_tokens(self) -> int:
        return sum(s.output_tokens for s in self.stages)

    def cost_usd(self, input_rate: float = 3.0,
                 output_rate: float = 15.0) -> float:
        """Total cost at given per-million-token rates."""
        return (
            self.total_input_tokens * input_rate / 1_000_000 +
            self.total_output_tokens * output_rate / 1_000_000
        )

    def summary(self) -> dict:
        return {
            "query_id": self.query_id,
            "total_ms": round(self.total_latency_ms, 1),
            "cost_usd": round(self.cost_usd(), 6),
            "stages": {
                s.name: {"ms": round(s.latency_ms, 1), "tokens": s.input_tokens + s.output_tokens}
                for s in self.stages
            }
        }


def instrumented_rag_query(query: str, conn, cache: SemanticCache) -> tuple[str, QueryTrace]:
    """Full RAG pipeline with per-stage timing and cost tracking."""
    trace = QueryTrace(query_id=hashlib.sha256(query.encode()).hexdigest()[:12])

    # Stage 1: Embed
    t0 = time.perf_counter()
    query_embedding = embed_text(query)
    trace.add_stage("embed", (time.perf_counter() - t0) * 1000)

    # Check cache
    cached = cache.get(query_embedding)
    if cached:
        trace.add_stage("cache_hit", 0.1)
        return cached, trace

    # Stage 2: Hybrid retrieve
    t0 = time.perf_counter()
    dense_results = dense_retrieve(query_embedding, conn, k=20)
    sparse_results = sparse_retrieve(query, conn, k=20)
    fused = reciprocal_rank_fusion([dense_results, sparse_results])
    trace.add_stage("retrieve", (time.perf_counter() - t0) * 1000)

    # Stage 3: Rerank
    t0 = time.perf_counter()
    reranked = rerank(query, fused[:20], top_k=5)
    trace.add_stage("rerank", (time.perf_counter() - t0) * 1000)

    # Stage 4: Generate
    t0 = time.perf_counter()
    answer, usage = generate_answer(query, [r.chunk for r in reranked])
    trace.add_stage("generate", (time.perf_counter() - t0) * 1000,
                    input_tokens=usage["input_tokens"],
                    output_tokens=usage["output_tokens"])

    # Cache the result
    cache.put(query_embedding, answer)

    return answer, trace

Observability platforms

Production RAG systems need per-query tracing with stage breakdowns, cost attribution, and quality monitoring. Three platforms cover the LLM observability space:

• LangSmith (LangChain): native integration with LangChain/LangGraph, automatic trace capture for every chain node, built-in eval framework with human annotation queues. Best when already using LangChain. Managed SaaS.
• LangFuse: open-source, self-hostable, model-agnostic. Manual span instrumentation via SDK decorators. Supports cost tracking, user feedback collection, and prompt versioning. Good for teams that want to own their observability data or have compliance constraints against external SaaS.
• Arize Phoenix: open-source, strong on embedding drift detection and retrieval quality visualization. Plots embedding distributions over time to detect when the corpus is going stale relative to user queries. Good for teams focused on retrieval quality monitoring.

Key signals to monitor in production:

• Retrieval drift: the distribution of query embeddings shifts over time as user needs evolve, but the corpus stays static. Monitor cosine distance between query embeddings and their nearest corpus matches — a rising average distance means the corpus is going stale.
• Relevance degradation: track the reranker's mean confidence score over time. A downward trend means retrieval quality is decaying (corpus staleness, query distribution shift, or embedding model degradation).
• Cost anomalies: alert when per-query cost exceeds 3x the 7-day rolling average. Catches runaway agentic loops, prompt length explosions, or model tier misrouting.
• User feedback loops: collect thumbs up/down or 1–5 ratings on generated answers. Feed negative examples into your eval test set. A baseline 5% negative rate is typical; above 10% sustained over a week requires investigation.

Key takeaways

• A well-optimized RAG pipeline targets p90 TTFT under 2 seconds: ~50ms embed, ~20ms retrieve, ~100ms rerank, ~800ms generate
• Generation dominates latency (82%); optimizing retrieval yields marginal gains compared to generation-side improvements
• Semantic caching achieves 15–30% hit rates on repetitive workloads, eliminating the full pipeline for cached queries
• Model tiering (simple → Haiku, standard → Sonnet, complex → Opus) reduces average cost 40–50% with minimal quality loss
• Anthropic prompt caching reduces contextual retrieval indexing cost by ~90% and saves query-time latency on static prompt prefixes
• Monitor retrieval drift, relevance degradation, cost anomalies, and user feedback — these four signals catch the majority of production quality regressions