Day 44: Observability Cho LLM App

{
  "timestamp": "2026-05-10T10:15:21.120Z",
  "level": "INFO",
  "event": "retrieval_completed",
  "trace_id": "tr_01hxx",
  "tenant_id": "acme",
  "route": "/query",
  "top_k": 20,
  "candidate_count": 18,
  "empty_retrieval": false,
  "latency_ms": 42,
  "index_version": "policy-index-2026-05-01"
}

rag.query
  query.rewrite
  retrieval.hybrid_search
  rerank.cross_encoder
  context.build
  llm.generate
  citation.validate
  feedback.attach

{
  "trace_id": "tr_01hxx4e6kg9k7r8y0x",
  "request_id": "req_9d2c",
  "session_id_hash": "sha256:7c0f...",
  "tenant_id": "acme",
  "user_id_hash": "sha256:aa31...",
  "route": "/query",
  "environment": "prod",
  "query": {
    "raw_redacted": "Chính sách nghỉ phép năm 2026 là gì?",
    "raw_hash": "sha256:2f67...",
    "language": "vi",
    "length_chars": 38
  },
  "rewrite": {
    "enabled": true,
    "rewritten_query_redacted": "Chính sách nghỉ phép nhân viên năm 2026",
    "latency_ms": 35
  },
  "retrieval": {
    "strategy": "hybrid_dense_bm25_rrf",
    "index_version": "policy-index-2026-05-01",
    "embedding_model": "text-embedding-3-small",
    "top_k": 30,
    "latency_ms": 64,
    "empty": false,
    "candidates": [
      {
        "rank": 1,
        "chunk_id": "policy_2026::chunk_018",
        "document_id": "policy_2026",
        "source_uri_hash": "sha256:c91e...",
        "score_dense": 0.82,
        "score_sparse": 12.7,
        "score_rrf": 0.041,
        "acl_matched": true
      }
    ]
  },
  "rerank": {
    "enabled": true,
    "reranker_model": "bge-reranker-v2-m3",
    "candidate_count": 30,
    "selected_count": 8,
    "latency_ms": 210,
    "top_score": 0.91,
    "score_gap_top2": 0.08
  },
  "context": {
    "chunk_ids": ["policy_2026::chunk_018", "policy_2026::chunk_021"],
    "context_tokens": 1840,
    "truncated": false,
    "max_context_tokens": 6000
  },
  "generation": {
    "provider": "openai",
    "model": "gpt-4.1-mini",
    "prompt_version": "rag-answer-v7",
    "temperature": 0.1,
    "input_tokens": 2300,
    "output_tokens": 420,
    "ttft_ms": 780,
    "latency_ms": 2800,
    "finish_reason": "stop",
    "estimated_cost_usd": 0.001592
  },
  "validation": {
    "schema_valid": true,
    "citation_valid": true,
    "citation_failure_reason": null,
    "guardrail_action": "allow"
  },
  "feedback": {
    "rating": null,
    "reason": null,
    "comment_redacted": null
  },
  "result": {
    "status": "success",
    "error_type": null,
    "total_latency_ms": 3140
  }
}

pip install fastapi uvicorn pydantic prometheus-client opentelemetry-api opentelemetry-sdk

from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter


def configure_tracing() -> None:
    resource = Resource.create(
        {
            "service.name": "rag-api",
            "service.version": "1.0.0",
            "deployment.environment": "dev",
        }
    )
    provider = TracerProvider(resource=resource)
    provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))
    trace.set_tracer_provider(provider)

from __future__ import annotations

import hashlib
import json
import logging
import os
import re
import time
import uuid
from contextlib import contextmanager
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Any, Literal

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
from prometheus_client import Counter, Gauge, Histogram

LOGGER = logging.getLogger("rag.observability")
LOGGER.setLevel(logging.INFO)

SALT = os.environ.get("OBSERVABILITY_HASH_SALT", "dev-only-change-me")
RAW_CONTENT_LOGGING = os.environ.get("RAW_CONTENT_LOGGING", "false").lower() == "true"

tracer = trace.get_tracer("rag-api", "1.0.0")

REQUESTS = Counter(
    "rag_request_total",
    "Total RAG requests",
    ["route", "status"],
)
STAGE_LATENCY = Histogram(
    "rag_stage_latency_seconds",
    "Latency by RAG stage",
    ["stage"],
    buckets=(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30),
)
TTFT = Histogram(
    "llm_ttft_seconds",
    "Time to first token",
    ["model"],
    buckets=(0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10),
)
TOKENS = Counter(
    "llm_token_total",
    "LLM token usage",
    ["model", "type"],
)
COST = Counter(
    "llm_cost_usd_total",
    "Estimated LLM cost in USD",
    ["model"],
)
IN_FLIGHT = Gauge(
    "rag_requests_in_flight",
    "RAG requests currently being processed",
    ["route"],
)
EMPTY_RETRIEVAL = Counter(
    "rag_empty_retrieval_total",
    "RAG requests with zero retrieved chunks",
    ["index_version"],
)
CITATION_INVALID = Counter(
    "rag_citation_invalid_total",
    "Invalid citation count",
    ["reason"],
)

MODEL_PRICE_USD_PER_1M = {
    "gpt-4.1-mini": {"input": Decimal("0.40"), "output": Decimal("1.60")},
    "gpt-4.1": {"input": Decimal("2.00"), "output": Decimal("8.00")},
}

PII_PATTERNS = [
    (re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"), "[EMAIL]"),
    (re.compile(r"\b(?:\+?84|0)(?:\d[\s.-]?){8,10}\b"), "[PHONE]"),
    (re.compile(r"\b\d{9,12}\b"), "[ID_NUMBER]"),
]


def hash_value(value: str | None) -> str | None:
    if not value:
        return None
    digest = hashlib.sha256(f"{SALT}:{value}".encode("utf-8")).hexdigest()
    return f"sha256:{digest}"


def redact_text(text: str | None) -> str | None:
    if text is None:
        return None
    redacted = text
    for pattern, replacement in PII_PATTERNS:
        redacted = pattern.sub(replacement, redacted)
    return redacted


def safe_content(text: str | None) -> str | None:
    if text is None:
        return None
    return redact_text(text) if RAW_CONTENT_LOGGING else None


def estimate_cost_usd(model: str, input_tokens: int, output_tokens: int) -> Decimal:
    price = MODEL_PRICE_USD_PER_1M.get(model)
    if not price:
        return Decimal("0")
    input_cost = Decimal(input_tokens) * price["input"] / Decimal(1_000_000)
    output_cost = Decimal(output_tokens) * price["output"] / Decimal(1_000_000)
    return (input_cost + output_cost).quantize(Decimal("0.000001"))


def log_event(event: str, **fields: Any) -> None:
    payload = {
        "event": event,
        "timestamp_ms": int(time.time() * 1000),
        **fields,
    }
    LOGGER.info(json.dumps(payload, ensure_ascii=False, default=str))


@dataclass
class RagTrace:
    trace_id: str
    route: str
    tenant_id: str
    user_id_hash: str | None
    prompt_version: str
    model: str
    index_version: str
    status: Literal["success", "error"] = "success"
    error_type: str | None = None
    stage_latency_ms: dict[str, int] = field(default_factory=dict)
    token_usage: dict[str, int] = field(default_factory=lambda: {"input": 0, "output": 0})
    estimated_cost_usd: Decimal = Decimal("0")


@contextmanager
def measured_stage(trace_record: RagTrace, stage: str, **span_attrs: Any):
    start = time.perf_counter()
    with tracer.start_as_current_span(f"rag.{stage}") as span:
        span.set_attributes(
            {
                "rag.trace_id": trace_record.trace_id,
                "rag.tenant_id": trace_record.tenant_id,
                "rag.stage": stage,
                **{k: v for k, v in span_attrs.items() if v is not None},
            }
        )
        try:
            yield span
            span.set_status(Status(StatusCode.OK))
        except Exception as exc:
            span.record_exception(exc)
            span.set_status(Status(StatusCode.ERROR, str(exc)))
            trace_record.status = "error"
            trace_record.error_type = type(exc).__name__
            raise
        finally:
            elapsed_seconds = time.perf_counter() - start
            elapsed_ms = round(elapsed_seconds * 1000)
            trace_record.stage_latency_ms[stage] = elapsed_ms
            STAGE_LATENCY.labels(stage=stage).observe(elapsed_seconds)

from typing import Annotated

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field
from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
from starlette.responses import Response

app = FastAPI(title="RAG API with Observability")


class QueryRequest(BaseModel):
    query: str = Field(min_length=1, max_length=4000)
    session_id: str | None = None
    top_k: int = Field(default=20, ge=1, le=100)


class QueryResponse(BaseModel):
    trace_id: str
    answer: str
    citations: list[dict[str, str]]
    usage: dict[str, int]
    estimated_cost_usd: str


@app.get("/metrics")
def metrics() -> Response:
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)


@app.post("/query", response_model=QueryResponse)
async def query(
    request: QueryRequest,
    x_tenant_id: Annotated[str, Header(alias="X-Tenant-Id")] = "demo",
    x_user_id: Annotated[str | None, Header(alias="X-User-Id")] = None,
) -> QueryResponse:
    trace_id = f"tr_{uuid.uuid4().hex}"
    route = "/query"
    model = "gpt-4.1-mini"
    prompt_version = "rag-answer-v7"
    index_version = "policy-index-2026-05-01"
    trace_record = RagTrace(
        trace_id=trace_id,
        route=route,
        tenant_id=x_tenant_id,
        user_id_hash=hash_value(x_user_id),
        prompt_version=prompt_version,
        model=model,
        index_version=index_version,
    )

    IN_FLIGHT.labels(route=route).inc()
    start = time.perf_counter()

    with tracer.start_as_current_span("rag.query") as root_span:
        root_span.set_attributes(
            {
                "rag.trace_id": trace_id,
                "rag.tenant_id": x_tenant_id,
                "rag.prompt_version": prompt_version,
                "rag.model": model,
                "rag.index_version": index_version,
            }
        )
        log_event(
            "query_received",
            trace_id=trace_id,
            tenant_id=x_tenant_id,
            user_id_hash=trace_record.user_id_hash,
            query_hash=hash_value(request.query),
            query_redacted=safe_content(request.query),
            query_length_chars=len(request.query),
            top_k=request.top_k,
        )

        try:
            with measured_stage(trace_record, "retrieval", index_version=index_version):
                retrieved = await retrieve_chunks(request.query, top_k=request.top_k)
                if not retrieved:
                    EMPTY_RETRIEVAL.labels(index_version=index_version).inc()

            log_event(
                "retrieval_completed",
                trace_id=trace_id,
                tenant_id=x_tenant_id,
                index_version=index_version,
                top_k=request.top_k,
                candidate_count=len(retrieved),
                empty_retrieval=len(retrieved) == 0,
                latency_ms=trace_record.stage_latency_ms["retrieval"],
                chunk_ids=[chunk["chunk_id"] for chunk in retrieved[:10]],
            )

            with measured_stage(trace_record, "rerank", candidate_count=len(retrieved)):
                reranked = await rerank_chunks(request.query, retrieved)

            with measured_stage(trace_record, "context_build", selected_count=len(reranked[:8])):
                context = build_context(reranked[:8])

            log_event(
                "context_built",
                trace_id=trace_id,
                chunk_count=len(context["chunk_ids"]),
                context_tokens=context["token_count"],
                truncated=context["truncated"],
            )

            with measured_stage(trace_record, "generation", model=model, prompt_version=prompt_version):
                generation = await generate_answer_streaming(
                    query=request.query,
                    context=context["text"],
                    model=model,
                    prompt_version=prompt_version,
                    trace_id=trace_id,
                )

            input_tokens = int(generation["usage"]["input_tokens"])
            output_tokens = int(generation["usage"]["output_tokens"])
            cost = estimate_cost_usd(model, input_tokens, output_tokens)
            trace_record.token_usage = {"input": input_tokens, "output": output_tokens}
            trace_record.estimated_cost_usd = cost

            TOKENS.labels(model=model, type="input").inc(input_tokens)
            TOKENS.labels(model=model, type="output").inc(output_tokens)
            COST.labels(model=model).inc(float(cost))
            TTFT.labels(model=model).observe(generation["ttft_ms"] / 1000)

            with measured_stage(trace_record, "citation_validation"):
                validation = validate_citations(generation["citations"], context["chunk_ids"])
                if not validation["valid"]:
                    CITATION_INVALID.labels(reason=validation["reason"]).inc()

            log_event(
                "generation_completed",
                trace_id=trace_id,
                tenant_id=x_tenant_id,
                model=model,
                prompt_version=prompt_version,
                input_tokens=input_tokens,
                output_tokens=output_tokens,
                ttft_ms=generation["ttft_ms"],
                latency_ms=trace_record.stage_latency_ms["generation"],
                estimated_cost_usd=str(cost),
                finish_reason=generation["finish_reason"],
            )
            log_event(
                "citation_validated",
                trace_id=trace_id,
                valid=validation["valid"],
                failure_reason=validation["reason"],
            )

            REQUESTS.labels(route=route, status="success").inc()
            total_latency_ms = round((time.perf_counter() - start) * 1000)
            log_event(
                "query_completed",
                trace_id=trace_id,
                status="success",
                total_latency_ms=total_latency_ms,
                stage_latency_ms=trace_record.stage_latency_ms,
                token_usage=trace_record.token_usage,
                estimated_cost_usd=str(trace_record.estimated_cost_usd),
            )

            return QueryResponse(
                trace_id=trace_id,
                answer=generation["answer"],
                citations=generation["citations"],
                usage=trace_record.token_usage,
                estimated_cost_usd=str(trace_record.estimated_cost_usd),
            )

        except TimeoutError as exc:
            REQUESTS.labels(route=route, status="timeout").inc()
            log_event(
                "request_failed",
                trace_id=trace_id,
                status="timeout",
                error_type=type(exc).__name__,
                stage_latency_ms=trace_record.stage_latency_ms,
            )
            raise HTTPException(status_code=504, detail={"trace_id": trace_id, "error": "timeout"}) from exc
        except Exception as exc:
            REQUESTS.labels(route=route, status="error").inc()
            log_event(
                "request_failed",
                trace_id=trace_id,
                status="error",
                error_type=type(exc).__name__,
                stage_latency_ms=trace_record.stage_latency_ms,
            )
            raise HTTPException(status_code=500, detail={"trace_id": trace_id, "error": "internal_error"}) from exc
        finally:
            IN_FLIGHT.labels(route=route).dec()

async def generate_answer_streaming(
    query: str,
    context: str,
    model: str,
    prompt_version: str,
    trace_id: str,
) -> dict[str, Any]:
    started = time.perf_counter()
    first_token_ms: int | None = None
    chunks: list[str] = []
    last_event: Any | None = None

    prompt = render_prompt(prompt_version=prompt_version, query=query, context=context)
    input_tokens_estimate = estimate_tokens(prompt)

    log_event(
        "generation_started",
        trace_id=trace_id,
        model=model,
        prompt_version=prompt_version,
        prompt_tokens_estimate=input_tokens_estimate,
    )

    async for token_event in llm_client.stream(model=model, prompt=prompt, temperature=0.1):
        last_event = token_event
        if first_token_ms is None:
            first_token_ms = round((time.perf_counter() - started) * 1000)
            log_event("first_token_received", trace_id=trace_id, model=model, ttft_ms=first_token_ms)
        chunks.append(token_event.text)

    answer = "".join(chunks)
    provider_usage = getattr(last_event, "usage", None)
    finish_reason = getattr(last_event, "finish_reason", None)
    usage = provider_usage if provider_usage else {
        "input_tokens": input_tokens_estimate,
        "output_tokens": estimate_tokens(answer),
    }

    return {
        "answer": answer,
        "citations": extract_citations(answer),
        "usage": usage,
        "ttft_ms": first_token_ms or round((time.perf_counter() - started) * 1000),
        "finish_reason": finish_reason or "unknown",
    }

from datetime import datetime, timezone
from typing import Literal

from pydantic import BaseModel, Field


class FeedbackRequest(BaseModel):
    trace_id: str = Field(pattern=r"^tr_[a-f0-9]+$")
    rating: Literal["up", "down"]
    reason: Literal[
        "helpful",
        "wrong_answer",
        "wrong_source",
        "missing_context",
        "too_slow",
        "unsafe",
        "other",
    ]
    comment: str | None = Field(default=None, max_length=2000)


@app.post("/feedback")
async def feedback(
    request: FeedbackRequest,
    x_tenant_id: Annotated[str, Header(alias="X-Tenant-Id")] = "demo",
    x_user_id: Annotated[str | None, Header(alias="X-User-Id")] = None,
) -> dict[str, str]:
    feedback_record = {
        "trace_id": request.trace_id,
        "tenant_id": x_tenant_id,
        "user_id_hash": hash_value(x_user_id),
        "rating": request.rating,
        "reason": request.reason,
        "comment_redacted": safe_content(request.comment),
        "comment_hash": hash_value(request.comment),
        "created_at": datetime.now(timezone.utc).isoformat(),
        "triage_status": "new",
    }
    await feedback_store.insert(feedback_record)
    log_event("feedback_received", **feedback_record)
    return {"status": "accepted", "trace_id": request.trace_id}

Client
  -> API Gateway
  -> RAG API
       -> Trace Context
       -> Query Rewrite
       -> Hybrid Retrieval
       -> Reranker
       -> Context Builder
       -> LLM Gateway
       -> Citation Validator
       -> Feedback API
  -> Observability Outputs
       -> Metrics: Prometheus -> Grafana
       -> Logs: JSON -> ELK/OpenSearch
       -> Traces: OpenTelemetry -> Tempo/Jaeger/Datadog/etc.
       -> LLM Trace Store: Langfuse/LangSmith/custom table

groups:
  - name: rag-observability
    rules:
      - alert: RAGHighErrorRate
        expr: |
          sum(rate(rag_request_total{status!="success"}[5m]))
          /
          sum(rate(rag_request_total[5m])) > 0.03
        for: 10m
        labels:
          severity: page
        annotations:
          summary: "RAG error rate > 3%"

      - alert: RAGHighP95Latency
        expr: |
          histogram_quantile(
            0.95,
            sum(rate(rag_request_latency_seconds_bucket[5m])) by (le)
          ) > 8
        for: 10m
        labels:
          severity: ticket
        annotations:
          summary: "RAG p95 latency > 8s"

      - alert: LLMHighTTFT
        expr: |
          histogram_quantile(
            0.95,
            sum(rate(llm_ttft_seconds_bucket[5m])) by (le, model)
          ) > 3
        for: 10m
        labels:
          severity: ticket
        annotations:
          summary: "LLM p95 TTFT > 3s"

      - alert: RAGCostSpike
        expr: |
          sum(increase(llm_cost_usd_total[1h]))
          >
          2 * (sum(increase(llm_cost_usd_total[24h])) / 24)
        for: 15m
        labels:
          severity: ticket
        annotations:
          summary: "LLM hourly cost is above normal baseline"

      - alert: RAGCitationFailureSpike
        expr: |
          sum(rate(rag_citation_invalid_total[15m]))
          /
          sum(rate(rag_request_total[15m])) > 0.05
        for: 15m
        labels:
          severity: ticket
        annotations:
          summary: "Citation failure rate > 5%"

CREATE TABLE rag_traces (
    trace_id TEXT PRIMARY KEY,
    tenant_id TEXT NOT NULL,
    user_id_hash TEXT,
    session_id_hash TEXT,
    route TEXT NOT NULL,
    environment TEXT NOT NULL,
    query_hash TEXT NOT NULL,
    query_redacted TEXT,
    prompt_version TEXT NOT NULL,
    model TEXT NOT NULL,
    embedding_model TEXT,
    reranker_model TEXT,
    index_version TEXT NOT NULL,
    input_tokens INTEGER NOT NULL DEFAULT 0,
    output_tokens INTEGER NOT NULL DEFAULT 0,
    estimated_cost_usd NUMERIC(12, 6) NOT NULL DEFAULT 0,
    ttft_ms INTEGER,
    total_latency_ms INTEGER NOT NULL,
    status TEXT NOT NULL,
    error_type TEXT,
    citation_valid BOOLEAN,
    raw_trace JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX idx_rag_traces_created_at ON rag_traces (created_at DESC);
CREATE INDEX idx_rag_traces_tenant_created ON rag_traces (tenant_id, created_at DESC);
CREATE INDEX idx_rag_traces_versions ON rag_traces (prompt_version, model, index_version);
CREATE INDEX idx_rag_traces_status ON rag_traces (status, error_type);

CREATE TABLE rag_feedback (
    id BIGSERIAL PRIMARY KEY,
    trace_id TEXT NOT NULL REFERENCES rag_traces(trace_id),
    tenant_id TEXT NOT NULL,
    user_id_hash TEXT,
    rating TEXT NOT NULL CHECK (rating IN ('up', 'down')),
    reason TEXT NOT NULL,
    comment_hash TEXT,
    comment_redacted TEXT,
    triage_status TEXT NOT NULL DEFAULT 'new',
    triage_owner TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX idx_rag_feedback_trace ON rag_feedback (trace_id);
CREATE INDEX idx_rag_feedback_rating_created ON rag_feedback (rating, created_at DESC);

import uuid


def new_trace_id() -> str:
    return f"tr_{uuid.uuid4().hex}"

{
  "trace_id": "tr_abc123",
  "answer": "...",
  "citations": [],
  "usage": {
    "input": 1200,
    "output": 180
  },
  "estimated_cost_usd": "0.000768"
}

import json
import logging
import time
from typing import Any

logger = logging.getLogger("rag")


def log_event(event: str, **fields: Any) -> None:
    logger.info(
        json.dumps(
            {
                "event": event,
                "timestamp_ms": int(time.time() * 1000),
                **fields,
            },
            ensure_ascii=False,
            default=str,
        )
    )

log_event("query_received", trace_id=trace_id, query_hash=query_hash, top_k=top_k)
log_event("retrieval_completed", trace_id=trace_id, candidate_count=len(chunks), latency_ms=...)
log_event("rerank_completed", trace_id=trace_id, selected_count=len(selected), latency_ms=...)
log_event("context_built", trace_id=trace_id, context_tokens=context_tokens, truncated=truncated)
log_event("generation_completed", trace_id=trace_id, input_tokens=..., output_tokens=..., cost=...)
log_event("citation_validated", trace_id=trace_id, valid=True, failure_reason=None)

from contextlib import contextmanager
import time


@contextmanager
def timed(stage: str, latency_ms: dict[str, int]):
    start = time.perf_counter()
    try:
        yield
    finally:
        latency_ms[stage] = round((time.perf_counter() - start) * 1000)

latency_ms = {}
request_started = time.perf_counter()

with timed("retrieval", latency_ms):
    chunks = retrieve(query)

with timed("rerank", latency_ms):
    selected = rerank(query, chunks)

with timed("context_build", latency_ms):
    context = build_context(selected)

with timed("generation", latency_ms):
    answer = generate(context, query)

log_event(
    "query_completed",
    trace_id=trace_id,
    total_latency_ms=round((time.perf_counter() - request_started) * 1000),
    stage_latency_ms=latency_ms,
)

pip install prometheus-client

from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
from starlette.responses import Response

REQUESTS = Counter("rag_request_total", "Total RAG requests", ["route", "status"])
STAGE_LATENCY = Histogram("rag_stage_latency_seconds", "Latency by stage", ["stage"])
TOKENS = Counter("llm_token_total", "LLM token usage", ["model", "type"])
COST = Counter("llm_cost_usd_total", "LLM cost in USD", ["model"])
IN_FLIGHT = Gauge("rag_requests_in_flight", "Requests in flight", ["route"])


@app.get("/metrics")
def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

REQUESTS.labels(route="/query", status="success").inc()
STAGE_LATENCY.labels(stage="retrieval").observe(latency_ms["retrieval"] / 1000)
TOKENS.labels(model=model, type="input").inc(input_tokens)
TOKENS.labels(model=model, type="output").inc(output_tokens)
COST.labels(model=model).inc(float(cost_usd))

curl http://localhost:8000/metrics | grep rag_

from decimal import Decimal

MODEL_PRICE_USD_PER_1M = {
    "gpt-4.1-mini": {"input": Decimal("0.40"), "output": Decimal("1.60")},
}


def estimate_cost_usd(model: str, input_tokens: int, output_tokens: int) -> Decimal:
    price = MODEL_PRICE_USD_PER_1M[model]
    return (
        Decimal(input_tokens) * price["input"] / Decimal(1_000_000)
        + Decimal(output_tokens) * price["output"] / Decimal(1_000_000)
    ).quantize(Decimal("0.000001"))

import time


async def stream_answer(prompt: str, model: str, trace_id: str):
    started = time.perf_counter()
    first_token_seen = False

    async for token in llm_client.stream(prompt=prompt, model=model):
        if not first_token_seen:
            ttft_ms = round((time.perf_counter() - started) * 1000)
            log_event("first_token_received", trace_id=trace_id, model=model, ttft_ms=ttft_ms)
            first_token_seen = True
        yield token.text

trace_record = {
    "trace_id": trace_id,
    "tenant_id": tenant_id,
    "user_id_hash": user_id_hash,
    "query": {
        "raw_hash": query_hash,
        "raw_redacted": query_redacted,
        "length_chars": len(query),
    },
    "retrieval": {
        "strategy": "hybrid",
        "index_version": index_version,
        "top_k": top_k,
        "latency_ms": latency_ms["retrieval"],
        "candidates": candidate_summaries,
    },
    "rerank": {
        "enabled": True,
        "reranker_model": reranker_model,
        "latency_ms": latency_ms["rerank"],
        "selected_count": len(selected),
    },
    "context": {
        "chunk_ids": [chunk["chunk_id"] for chunk in selected],
        "context_tokens": context_tokens,
        "truncated": truncated,
    },
    "generation": {
        "model": model,
        "prompt_version": prompt_version,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "ttft_ms": ttft_ms,
        "latency_ms": latency_ms["generation"],
        "estimated_cost_usd": str(cost_usd),
    },
    "validation": {
        "citation_valid": citation_valid,
        "citation_failure_reason": citation_failure_reason,
    },
    "result": {
        "status": "success",
        "total_latency_ms": total_latency_ms,
    },
}

POST /feedback
{
  "trace_id": "tr_abc123",
  "rating": "down",
  "reason": "wrong_source",
  "comment": "Answer cited policy 2024, but the question asked 2026"
}

from typing import Literal
from pydantic import BaseModel, Field


class FeedbackRequest(BaseModel):
    trace_id: str
    rating: Literal["up", "down"]
    reason: Literal[
        "helpful",
        "wrong_answer",
        "wrong_source",
        "missing_context",
        "too_slow",
        "unsafe",
        "other",
    ]
    comment: str | None = Field(default=None, max_length=2000)

import hashlib
import os
import re

SALT = os.environ["OBSERVABILITY_HASH_SALT"]


def hash_value(value: str) -> str:
    return "sha256:" + hashlib.sha256(f"{SALT}:{value}".encode()).hexdigest()


def redact_text(text: str) -> str:
    text = re.sub(r"[\w.+-]+@[\w-]+\.[\w.-]+", "[EMAIL]", text)
    text = re.sub(r"\b(?:\+?84|0)(?:\d[\s.-]?){8,10}\b", "[PHONE]", text)
    text = re.sub(r"\b\d{9,12}\b", "[ID_NUMBER]", text)
    return text

{"id":"q001","query":"Chính sách nghỉ phép năm 2026 là gì?","expected_source":"policy_2026"}
{"id":"q002","query":"Nhân viên thử việc có được nghỉ phép không?","expected_source":"policy_hr"}

import json
import requests

with open("golden_questions.jsonl", "r", encoding="utf-8") as f:
    for line in f:
        item = json.loads(line)
        response = requests.post(
            "http://localhost:8000/query",
            json={"query": item["query"], "top_k": 20},
            timeout=30,
        )
        print(json.dumps({"id": item["id"], **response.json()}, ensure_ascii=False))

python run_golden_set.py > traces/golden_run_day44.jsonl

Dùng được trong production không?

Có, nhưng chỉ ở mức internal beta nếu thỏa:
- Observability: mọi request có trace_id, stage latency, token usage, cost/request và error type.
- Privacy: raw query/context/output được redact trước khi log; raw trace chỉ lưu theo sampling policy.
- Cost control: dashboard có cost/request, cost/day, token/request và alert cost spike.
- Alert/runbook: p95 latency, error rate, timeout rate, citation failure và empty retrieval đều có owner.
- Eval/feedback loop: feedback gắn trace_id và golden set chạy trước khi đổi prompt/model/index.
- Performance overhead: instrumentation overhead dưới 5% p95 latency so với baseline.

Chưa được public production nếu còn thiếu:
- Chưa có access control cho trace store, chưa có retention policy, chưa có redaction test tự động,
  hoặc chưa có load test chứng minh overhead của observability.

Có thể dùng cho internal beta. Hệ thống đã có trace_id, JSON logs, Prometheus metrics,
token/cost accounting, feedback endpoint và dashboard p95 latency/cost/citation failure.
Để lên public production cần thêm redaction test tự động, retention policy, access control
cho trace store, alert có owner, load test chứng minh overhead dưới 5%, và golden set chạy
trước mỗi lần đổi prompt/model/index.