Observabilidad de LLMs en 2026: Trazas, Costo por Token y Redacción de PII con OpenTelemetry y Langfuse

1. El Problema: Los Logs Dicen Que Funcionó, No Cuánto Costó

Un request a un LLM rara vez es una sola llamada. Un mensaje del usuario se abre en un retrieval, una o más llamadas al modelo, un tool o dos y quizás un loop de agente. El logging clásico de requests aplasta todo eso en una línea, así que tres preguntas quedan sin respuesta: dónde se fue el tiempo, qué paso produjo la mala respuesta y quién es responsable del gasto.

OpenTelemetry te da la estructura y un formato de transporte neutral. Las convenciones semánticas GenAI te dan nombres compartidos para lo que debe llevar un span de LLM. Langfuse te da un backend que entiende LLMs y esos atributos: uso, modelo, costo, usuario y sesión. Esa combinación es lo que armamos abajo.

2. Conceptos Mínimos Antes de Tocar Código

Seis ideas cubren lo que necesitás. Con esto claro, el código se lee como una checklist.

3. Paso 1 — Pineá Versiones y Apuntá OpenTelemetry a Langfuse

Revisé PyPI el 2026-10-14: langfuse 5.0.1, Python 3.11 o superior, publicada ese mismo día, así que pineala exacta. Depende de opentelemetry-api, opentelemetry-sdk y el exporter OTLP HTTP en 1.45.0 o superior (menor a 2). El SDK Python v5 requiere Langfuse Cloud o self-hosted 4.59.0+; el self-hosted v3 se queda en el SDK v4. El cliente lee las variables LANGFUSE_* y configura OpenTelemetry por vos.

bash — instalación y entorno

# verified 2026-10-14: langfuse 5.0.1 on PyPI (Python >=3.11); it pulls
# opentelemetry-api / -sdk / -exporter-otlp-proto-http >=1.45.0,<2
pip install "langfuse==5.0.1" openai

export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"   # US: https://us.cloud.langfuse.com
export LLM_MODEL="<your-model-id>"                      # no model ids hardcoded in telemetry code
export LLM_CAPTURE_CONTENT="false"                      # prompts/completions are Opt-In (step 5)

Un módulo se queda con la frontera de OpenTelemetry. El cliente de Langfuse se engancha al TracerProvider global y, por defecto, exporta spans de Langfuse, spans con atributos gen_ai.* y scopes de instrumentación LLM conocidos, así que los spans de HTTP y base de datos quedan afuera. Alcanza con la API de OpenTelemetry para emitir spans con la forma de la spec. La auto-instrumentación cubre la llamada al proveedor; los spans de retrieval y tools los ponés vos igual.

Python — telemetry.py: cliente, tracer y un helper de spans

# telemetry.py
import json, os
from contextlib import contextmanager

from opentelemetry import trace
from opentelemetry.trace import SpanKind, Status, StatusCode
from langfuse import Langfuse, propagate_attributes

# One client per process. It sets up OpenTelemetry and exports Langfuse spans plus
# spans with gen_ai.* attributes (default filter). Steps 5 and 6 add options here.
langfuse = Langfuse()
tracer = trace.get_tracer("support-bot.genai")

CAPTURE_CONTENT = os.getenv("LLM_CAPTURE_CONTENT") == "true"

def j(value) -> str:
    return json.dumps(value, ensure_ascii=False, default=str)

@contextmanager
def genai_span(name, kind, op, lf_type, attrs=None):
    with tracer.start_as_current_span(
        name, kind=kind,
        record_exception=False, set_status_on_exception=False,
        attributes={"gen_ai.operation.name": op, "langfuse.observation.type": lf_type, **(attrs or {})},
    ) as span:
        try:
            yield span
        except Exception as exc:
            # error.type only: exception text often echoes user input
            span.set_attribute("error.type", type(exc).__name__)
            span.set_status(Status(StatusCode.ERROR))
            raise

4. Paso 2 — Un Span de Chat con Uso de Tokens y Finish Reason

Una función envuelve la llamada al proveedor. Los atributos son los que define el span de inferencia: operación y proveedor (requeridos), modelo pedido y de respuesta, id de respuesta, finish reasons y los contadores de uso. Además seteo langfuse.observation.type explícito. Un valor reconocido siempre gana, así Langfuse no tiene que adivinar que el span es una generation.

Python — chat() con atributos semconv

# llm.py
import os
from openai import OpenAI
from telemetry import CAPTURE_CONTENT, SpanKind, genai_span, j

client = OpenAI()
MODEL = os.environ["LLM_MODEL"]

def to_parts(m: dict) -> dict:  # Chat Completions message -> semconv {role, parts}
    parts = []
    if m.get("content"):
        parts.append({"type": "text", "content": m["content"]})
    for tc in m.get("tool_calls") or []:
        parts.append({"type": "tool_call", "id": tc["id"],
                      "name": tc["function"]["name"], "arguments": tc["function"]["arguments"]})
    if m["role"] == "tool":
        parts = [{"type": "tool_call_response",
                  "id": m["tool_call_id"], "response": m["content"]}]
    return {"role": m["role"], "parts": parts}

def chat(messages: list, tools: list | None = None):
    with genai_span(f"chat {MODEL}", SpanKind.CLIENT, "chat", "generation", {
        "gen_ai.provider.name": "openai",
        "gen_ai.request.model": MODEL,
    }) as span:
        kwargs = {"tools": tools} if tools else {}
        resp = client.chat.completions.create(model=MODEL, messages=messages, **kwargs)
        u = resp.usage
        cached = getattr(u.prompt_tokens_details, "cached_tokens", 0) or 0       # details can be None
        reasoning = getattr(u.completion_tokens_details, "reasoning_tokens", 0) or 0
        span.set_attributes({
            "gen_ai.response.model": resp.model,
            "gen_ai.response.id": resp.id,
            "gen_ai.response.finish_reasons": [c.finish_reason for c in resp.choices],
            "gen_ai.usage.input_tokens": u.prompt_tokens,         # includes cached tokens
            "gen_ai.usage.output_tokens": u.completion_tokens,    # includes reasoning tokens
            "gen_ai.usage.cache_read.input_tokens": cached,       # subset of input_tokens
            "gen_ai.usage.reasoning.output_tokens": reasoning,    # subset of output_tokens
        })
        if CAPTURE_CONTENT:  # Opt-In: "likely to contain sensitive information including PII"
            out = resp.choices[0].message.model_dump(exclude_none=True)
            span.set_attributes({
                "gen_ai.input.messages": j([to_parts(m) for m in messages]),
                "gen_ai.output.messages": j([to_parts(out)]),
                "langfuse.observation.input": j(messages),   # Langfuse's table does not list
                "langfuse.observation.output": j(out),       # gen_ai.input.messages: mirror it
            })
        return resp

Cómo aterrizan estos atributos en Langfuse, según su tabla de mapeo de OpenTelemetry:

5. Paso 3 — Spans de Retrieval, Tool Calls y Agente

El span de chat solo esconde casi toda la historia. La spec define spans separados para los pasos alrededor: retrieval (nombre ’retrieval <data source id>’, CLIENT), execute_tool (’execute_tool <tool name>’, INTERNAL) e invoke_agent (’invoke_agent <agent name>’). Registrá ids, scores y conteos. El texto recuperado y los argumentos y resultados de tools quedan detrás del mismo flag de opt-in.

Python — retrieve(), run_tool() y el span raíz del agente

# agent.py
import json
from telemetry import CAPTURE_CONTENT, SpanKind, genai_span, j
from llm import chat

def retrieve(query: str, top_k: int = 4):
    with genai_span("retrieval kb-articles", SpanKind.CLIENT, "retrieval", "retriever", {
        "gen_ai.data_source.id": "kb-articles",
        "gen_ai.retrieval.top_k": top_k,
    }) as span:
        hits = vector_store.search(query, k=top_k)            # your retriever
        span.set_attribute("app.retrieval.doc_ids", [h.id for h in hits])
        span.set_attribute("app.retrieval.scores", [round(h.score, 3) for h in hits])
        if CAPTURE_CONTENT:                                   # Opt-In per the spec
            span.set_attribute("gen_ai.retrieval.query.text", query)
        return hits

def run_tool(call: dict):
    name = call["function"]["name"]
    with genai_span(f"execute_tool {name}", SpanKind.INTERNAL, "execute_tool", "tool", {
        "gen_ai.tool.name": name,
        "gen_ai.tool.call.id": call["id"],
        "gen_ai.tool.type": "function",
    }) as span:
        args = json.loads(call["function"]["arguments"])
        result = TOOLS[name](**args)                          # your tool registry
        if CAPTURE_CONTENT:
            span.set_attribute("gen_ai.tool.call.arguments", j(args))
            span.set_attribute("gen_ai.tool.call.result", j(result))
        return result

def run_agent(question: str, conversation_id: str) -> str | None:
    with genai_span("invoke_agent support-bot", SpanKind.INTERNAL, "invoke_agent", "agent", {
        "gen_ai.agent.name": "support-bot",
        "gen_ai.conversation.id": conversation_id,
    }) as span:
        hits = retrieve(question)
        messages = [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": question + "\n\n" + format_context(hits)},
        ]
        content = None
        for _ in range(4):                                    # hard cap on tool loops
            m = chat(messages, tools=TOOL_SCHEMAS).choices[0].message
            content = m.content
            turn = {"role": "assistant", "content": m.content}
            if m.tool_calls:
                turn["tool_calls"] = [tc.model_dump() for tc in m.tool_calls]
            messages.append(turn)
            if not m.tool_calls:
                break
            for call in turn["tool_calls"]:
                messages.append({"role": "tool", "tool_call_id": call["id"],
                                 "content": j(run_tool(call))})
        if CAPTURE_CONTENT:
            span.set_attribute("langfuse.observation.input", question)
            span.set_attribute("langfuse.observation.output", content or "")
        return content

El resultado es un árbol que se lee de un vistazo. Esta es la forma que producen los nombres y kinds de arriba, no una captura:

Forma esperada de la traza

invoke_agent support-bot          INTERNAL   langfuse.observation.type=agent
|-- retrieval kb-articles         CLIENT     retriever: top_k, doc ids, scores
|-- chat <model>                  CLIENT     generation: usage.input_tokens / output_tokens
|-- execute_tool <tool>           INTERNAL   tool: gen_ai.tool.name, call id
|-- chat <model>                  CLIENT     generation: finish_reasons=["stop"]

(every span: user.id / session.id inherited from propagate_attributes)

6. Paso 4 — Usuario, Sesión y Costo por Usuario

Una vista de costo por usuario necesita dos cosas: que cada generation lleve un costo (o un modelo que Langfuse pueda tarifar) y que cada span lleve el usuario. Langfuse documenta que para filtrar o agregar por userId o sessionId el atributo tiene que estar en todos los spans de la traza, no solo en la raíz. propagate_attributes() lo hace para los spans creados dentro de su bloque, así que llamalo en la frontera del request. Hasheá el identificador: user_id es US-ASCII, 200 caracteres como máximo, y no tiene por qué ser un email crudo.

Python — frontera del request con user id seudónimo

# handler.py
import hashlib, hmac, os
from langfuse import propagate_attributes
from agent import run_agent

def pseudonym(email: str) -> str:
    key = os.environ["USER_ID_HMAC_KEY"].encode()
    return hmac.new(key, email.strip().lower().encode(), hashlib.sha256).hexdigest()[:32]

def handle_request(question: str, email: str, conversation_id: str, plan: str) -> str | None:
    # Call this FIRST: only spans created inside the block get the attributes.
    with propagate_attributes(
        user_id=pseudonym(email),         # US-ASCII, max 200 chars, no raw email
        session_id=conversation_id,
        metadata={"plan": plan},          # few keys, short low-cardinality values
        tags=["support-bot"],
    ):
        return run_agent(question, conversation_id)

Para el costo, el camino por defecto es no mandar nada: Langfuse compara el nombre del modelo contra sus definiciones de modelo y multiplica por el uso. Agregá una definición custom (Project Settings > Models, o la Models API) para los modelos que no conoce. Con un precio de contrato o un modelo desconocido, calculá el costo vos con buckets que no se solapen y mandá gen_ai.usage.cost; los valores ingestados tienen prioridad. Si usás el update(usage_details, cost_details) propio de Langfuse, ojo: sus buckets son mutuamente excluyentes, así que input excluye los tokens cacheados que el input_tokens de OpenTelemetry sí incluye.

Python — costo desde tu propia tabla de precios

# pricing.py: prices come from YOUR contract or config, never from a blog post
import os

PRICE_PER_MTOK = {
    "input": float(os.environ["PRICE_INPUT_USD"]),
    "cached": float(os.environ["PRICE_CACHED_INPUT_USD"]),
    "output": float(os.environ["PRICE_OUTPUT_USD"]),
}

def cost_usd(input_tokens: int, cached: int, output_tokens: int) -> float:
    fresh = input_tokens - cached            # cached tokens are a SUBSET of input_tokens
    total = (
        fresh * PRICE_PER_MTOK["input"]
        + cached * PRICE_PER_MTOK["cached"]
        + output_tokens * PRICE_PER_MTOK["output"]   # reasoning tokens are already inside
    )
    return total / 1_000_000

# inside chat(), after the usage attributes (ingested cost beats inferred cost):
span.set_attribute("gen_ai.usage.cost", cost_usd(u.prompt_tokens, cached, u.completion_tokens))

Para leer la vista: la página Users lista usuarios segmentados por uso de tokens, cantidad de trazas y feedback, y los dashboards custom grafican costo, tokens y trazas por usuario. Para acceso programático usá la Metrics API v2.

bash — Metrics API v2: costo por modelo

curl -G \
  -H "Authorization: Basic $AUTH_STRING" \
  --data-urlencode 'query={
    "view": "observations",
    "metrics": [{"measure": "totalCost", "aggregation": "sum"}],
    "dimensions": [{"field": "providedModelName"}],
    "filters": [],
    "fromTimestamp": "2026-10-01T00:00:00Z",
    "toTimestamp": "2026-10-15T00:00:00Z",
    "orderBy": [{"field": "sum_totalCost", "direction": "desc"}],
    "config": {"row_limit": 1000}
  }' \
  https://cloud.langfuse.com/api/public/v2/metrics

7. Paso 5 — Redactá Prompts y PII Antes de Que Salgan del Proceso

Tres capas, de la más fuerte a la más débil. Una: no capturar contenido (el default Opt-In de arriba). Dos: enmascarar adentro del proceso lo que sí capturás. Tres: un processor de redacción en el Collector como respaldo en el gateway. Para la capa dos los docs recomiendan mask_otel_spans en setups Python nuevos. Corre al exportar sobre los atributos crudos de cada span que exporta el cliente, incluidas instrumentaciones de terceros; el hook viejo mask solo ve lo que seteás vía llamadas del SDK.

Python — mask_otel_spans con scrubbing por regex

# telemetry.py (final constructor)
import re
from typing import Optional
from langfuse.types import MaskOtelSpansParams, MaskOtelSpansResult, OtelSpanPatch

PATTERNS = [
    (re.compile(r"\b[\w.-]+?@[\w.-]+?\.\w+?\b"), "[EMAIL]"),
    (re.compile(r"\b(?:\d[ -]*?){13,19}\b"), "[CARD]"),
    (re.compile(r"\b\d{3}[-. ]?\d{3}[-. ]?\d{4}\b"), "[PHONE]"),
]
DROP = ("gen_ai.system_instructions", "gen_ai.tool.definitions")

def scrub(text: str) -> str:
    for pattern, label in PATTERNS:
        text = pattern.sub(label, text)
    return text

def mask_otel_spans(*, params: MaskOtelSpansParams) -> Optional[MaskOtelSpansResult]:
    patches = {}
    for identifier, span in params.spans.items():
        replacements = {}
        for key, value in span.attributes.items():
            if isinstance(value, str):
                masked = scrub(value)
                if masked != value:
                    replacements[key] = masked
        drop = tuple(k for k in DROP if k in span.attributes)
        if replacements or drop:
            patches[identifier] = OtelSpanPatch(delete_attributes=drop, set_attributes=replacements)
    return MaskOtelSpansResult(span_patches=patches)

langfuse = Langfuse(
    mask_otel_spans=mask_otel_spans,
    sample_rate=float(os.getenv("LANGFUSE_SAMPLE_RATE", "0.25")),  # trace-level, client-side
)

Propiedades que conviene conocer: el hook es síncrono y corre en el thread worker del batch, así que mantenelo rápido y determinista. Solo puede cambiar atributos: no nombres, eventos, links ni atributos de resource. Si lanza una excepción o devuelve un resultado inválido, Langfuse descarta todo el batch de exportación. Eso es fail-closed, que es lo que querés para PII, así que no lo envuelvas en un try/except que devuelva None. En el SDK v5 los valores de metadata van codificados en JSON; decodificá con json.loads antes de inspeccionarlos.

8. Paso 6 — Sampling Sin Mentirte con el Costo

Trazar cada request está bien a escala chica; después hacés sampling. El sample_rate del cliente de Langfuse (o LANGFUSE_SAMPLE_RATE) es un float de 0 a 1 que se decide por traza, y una traza no muestreada no manda ninguna de sus observaciones ni scores. La consecuencia que se pasa por alto: con 0.25, cualquier costo por usuario leído de las trazas cubre más o menos un cuarto del tráfico. Los totales de facturación van en métricas, que no se muestrean con las trazas. La spec actual nombra los contadores de uso gen_ai.client.inference.usage.input_tokens y output_tokens, con gen_ai.token.modality requerido.

Python — contadores de tokens sin muestrear (sample_rate va en el constructor de arriba)

# metrics.py: unsampled counters for billing-grade totals
from opentelemetry import metrics

meter = metrics.get_meter("support-bot.genai")
INPUT_TOKENS = meter.create_counter("gen_ai.client.inference.usage.input_tokens", unit="{token}")
OUTPUT_TOKENS = meter.create_counter("gen_ai.client.inference.usage.output_tokens", unit="{token}")

def record_usage(resp, request_model: str) -> None:
    attrs = {
        "gen_ai.operation.name": "chat",
        "gen_ai.provider.name": "openai",
        "gen_ai.token.modality": "text",
        "gen_ai.request.model": request_model,
        "gen_ai.response.model": resp.model,
        # no user id here: one time series per user explodes cardinality
    }
    INPUT_TOKENS.add(resp.usage.prompt_tokens, attrs)
    OUTPUT_TOKENS.add(resp.usage.completion_tokens, attrs)
# needs a MeterProvider + OTLP metric exporter pointing at your metrics backend

Para conservar todos los errores y todas las trazas lentas en vez de un porcentaje plano, decidí cuando la traza ya terminó. El processor tail_sampling del Collector (Beta para trazas) retiene trazas por estado, latencia o probabilidad, pero todos los spans de una traza deben llegar a la misma instancia del Collector. El mismo Collector puede correr el processor de redacción y reenviar a Langfuse por OTLP/HTTP.

YAML — Collector: redactar, tail sampling y exportar a Langfuse

receivers:
  otlp:
    protocols:
      http: {}
processors:
  redaction:
    allow_all_keys: true
    blocked_values:
      - "4[0-9]{12}(?:[0-9]{3})?"      # Visa-like card numbers
  tail_sampling:
    decision_wait: 30s
    policies:
      - name: keep-errors
        type: status_code
        status_code: { status_codes: [ERROR] }
      - name: keep-slow
        type: latency
        latency: { threshold_ms: 8000 }
      - name: baseline
        type: probabilistic
        probabilistic: { sampling_percentage: 10 }
exporters:
  otlphttp/langfuse:
    endpoint: "https://cloud.langfuse.com/api/public/otel"
    headers:
      Authorization: "Basic ${AUTH_STRING}"     # base64(public_key:secret_key)
      x-langfuse-ingestion-version: "4"
service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [redaction, tail_sampling]
      exporters: [otlphttp/langfuse]

9. Errores Comunes (La Mayoría Son Silenciosos)

Estos son los modos de falla que revisaría primero. Ninguno lanza una excepción; todos hacen que un dashboard se vea bien estando mal.

Conclusión

Ya tenés el loop completo: spans de chat, retrieval, tools y agente con la forma de la spec exportados a Langfuse 5.0.1, uso de tokens y costo en cada generation, un usuario seudónimo y una sesión en cada span, un masker delante del exporter y un plan de sampling que no corrompe tus números de costo.

Empezá con un flujo esta semana: agregá el span de chat, apagá la captura de contenido, hasheá el user id y compará los tokens de una traza con el bloque usage de tu proveedor. Esperá renombres mientras las convenciones GenAI estén en Development, y dejá todos los nombres de atributos en un solo módulo para que un renombre sea un cambio de un archivo.

Fuentes

• OpenTelemetry — Repositorio de convenciones semánticas GenAI (Status: Development)
• OpenTelemetry — Spans de inferencia GenAI (atributos, uso, contenido Opt-In)
• OpenTelemetry — Spans GenAI: retrieval y execute_tool
• OpenTelemetry — Spans de agente GenAI (invoke_agent)
• OpenTelemetry — Métricas de tokens de inferencia GenAI
• Langfuse — Integración con OpenTelemetry (endpoints, headers, mapeo de atributos)
• Langfuse — Resumen del SDK Python (v5)
• Langfuse — Masking (mask_otel_spans)
• Langfuse — Sampling
• Langfuse — Seguimiento de tokens y costo
• Langfuse — Metrics API v2
• PyPI — langfuse 5.0.1
• OpenTelemetry Collector — processor de tail sampling
• OpenTelemetry Collector — processor de redacción