LLM Observability in 2026: Traces, Token Cost and PII Redaction with OpenTelemetry and Langfuse

1. The Problem: Logs Say It Worked, Not What It Cost

An LLM request is rarely one call. A single user message fans out into a retrieval, one or more model calls, a tool or two and maybe an agent loop. Classic request logging flattens all of that into one line, so three questions stay unanswerable: where did the time go, which step produced the bad answer, and who is responsible for the spend.

OpenTelemetry gives you the structure and a vendor-neutral wire format. The GenAI semantic conventions give you shared names for what an LLM span should carry. Langfuse gives you an LLM-aware backend that understands those attributes: usage, model, cost, user and session. That combination is what we build below.

2. Minimum Concepts Before Touching Code

Six ideas cover what you need. Get these straight and the code reads like a checklist.

3. Step 1 — Pin Versions and Point OpenTelemetry at Langfuse

I checked PyPI on 2026-10-14: langfuse 5.0.1, Python 3.11 or newer, published that same day, so pin it exactly. It depends on opentelemetry-api, opentelemetry-sdk and the OTLP HTTP exporter at 1.45.0 or newer (below 2). Python SDK v5 needs Langfuse Cloud or self-hosted 4.59.0+; self-hosted v3 stays on SDK v4. The client reads LANGFUSE_* variables and sets OpenTelemetry up for you.

bash — install and environment

# 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)

One module then owns the OpenTelemetry boundary. The Langfuse client attaches to the global TracerProvider and, by default, exports Langfuse spans, spans with gen_ai.* attributes and known LLM instrumentation scopes, so HTTP and database spans stay out. Plain OpenTelemetry API calls are enough to emit spec-shaped spans. Auto-instrumentation covers the provider call; retrieval and tool spans are yours either way.

Python — telemetry.py: client, tracer and one span helper

# 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. Step 2 — A Chat Span With Token Usage and Finish Reason

One function wraps the provider call. The attributes are the ones the inference span defines: operation and provider (required), request and response model, response id, finish reasons and the usage counters. I also set langfuse.observation.type explicitly. A recognized value always wins, so Langfuse never has to guess that this span is a generation.

Python — chat() with semconv attributes

# 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

How these attributes land in Langfuse, from its OpenTelemetry mapping table:

5. Step 3 — Retrieval, Tool-Call and Agent Spans

The chat span alone hides most of the story. The spec defines separate spans for the steps around it: retrieval (name ’retrieval <data source id>’, CLIENT), execute_tool (’execute_tool <tool name>’, INTERNAL) and invoke_agent (’invoke_agent <agent name>’). Record ids, scores and counts. Keep retrieved text, tool arguments and results behind the same opt-in flag.

Python — retrieve(), run_tool() and the agent root span

# 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

The result is a tree you can read at a glance. This is the shape the span names and kinds above produce, not a screenshot:

Expected trace shape

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. Step 4 — User, Session and Cost Per User

A cost-per-user view needs two things: every generation carries a cost (or a model Langfuse can price) and every span carries the user. Langfuse documents that to filter or aggregate by userId or sessionId the attribute must be on every span of the trace, not only the root. propagate_attributes() does that for spans created inside its block, so call it at the request boundary. Hash the identifier: user_id is US-ASCII, 200 characters at most, and has no business being a raw email.

Python — request boundary with a pseudonymous user id

# 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)

For cost the default path is to send nothing: Langfuse matches the model name against its model definitions and multiplies by usage. Add a custom definition (Project Settings > Models, or the Models API) for models it does not know. With a contract price or an unknown model, compute cost yourself from non-overlapping buckets and send gen_ai.usage.cost; ingested values take priority. If you use Langfuse’s own update(usage_details, cost_details) instead, note that its buckets are mutually exclusive: input excludes the cached tokens that OpenTelemetry’s input_tokens includes.

Python — cost from your own price table

# 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))

Reading the view: the Users page lists users segmented by token usage, trace count and feedback, and custom dashboards chart user-level cost, tokens and trace counts. For programmatic access use the Metrics API v2.

bash — Metrics API v2: cost by model

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. Step 5 — Redact Prompts and PII Before They Leave the Process

Three layers, strongest first. One: do not capture content at all (the Opt-In default above). Two: mask what you do capture inside the process. Three: a Collector redaction processor as a gateway backstop. For layer two the docs recommend mask_otel_spans for new Python setups. It runs at export time over the raw attributes of every span the client exports, third-party instrumentations included; the older mask hook only sees data you set through SDK calls.

Python — mask_otel_spans with regex scrubbing

# 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
)

Properties worth knowing: the hook is synchronous and runs on the batch worker thread, so keep it fast and deterministic. It can change attributes only: not names, events, links or resource attributes. If it raises or returns an invalid result, Langfuse drops the whole export batch. That is fail-closed, which is what you want for PII, so do not wrap it in a try/except that returns None. In SDK v5 metadata values are JSON-encoded; decode with json.loads before inspecting them.

8. Step 6 — Sampling Without Lying About Cost

Tracing every request is fine at small scale; later you sample. Langfuse’s client-side sample_rate (or LANGFUSE_SAMPLE_RATE) is a float from 0 to 1 decided per trace, and an unsampled trace sends none of its observations or scores. The consequence people miss: at 0.25, any per-user cost read from traces covers roughly a quarter of the traffic. Billing-grade totals belong in metrics, which are not sampled with traces. The current spec names the usage counters gen_ai.client.inference.usage.input_tokens and output_tokens, with gen_ai.token.modality required.

Python — unsampled token counters (sample_rate is in the constructor above)

# 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

To keep every error and every slow trace instead of a flat percentage, decide after the trace completes. The Collector’s tail_sampling processor (Beta for traces) keeps traces by status, latency or probability, but all spans of a trace must reach the same Collector instance. The same Collector can run the redaction processor and forward to Langfuse over OTLP/HTTP.

YAML — Collector: redact, tail-sample, export to 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. Common Errors (Most Are Silent)

These are the failure modes I would check first. None of them throws; all of them make a dashboard look fine while being wrong.

Conclusion

You now have the whole loop: spec-shaped chat, retrieval, tool and agent spans exported to Langfuse 5.0.1, token usage and cost on every generation, a pseudonymous user and session on every span, a masker in front of the exporter and a sampling plan that does not corrupt your cost numbers.

Start with one flow this week: add the chat span, turn content capture off, hash the user id and compare one trace’s token counts with your provider’s usage block. Expect renames while the GenAI conventions are in Development, and keep every attribute name in a single module so a rename is a one-file change.

Sources

• OpenTelemetry — GenAI semantic conventions repository (Status: Development)
• OpenTelemetry — GenAI client inference spans (attributes, usage, Opt-In content)
• OpenTelemetry — GenAI spans: retrieval and execute_tool
• OpenTelemetry — GenAI agent spans (invoke_agent)
• OpenTelemetry — GenAI inference token metrics
• Langfuse — OpenTelemetry integration (endpoints, headers, attribute mapping)
• Langfuse — Python SDK overview (v5)
• Langfuse — Masking (mask_otel_spans)
• Langfuse — Sampling
• Langfuse — Token and cost tracking
• Langfuse — Metrics API v2
• PyPI — langfuse 5.0.1
• OpenTelemetry Collector — tail sampling processor
• OpenTelemetry Collector — redaction processor