OTel Token Usage Processing#

Token usage normalization during OTel span ingestion is handled by extractUsageDetails() inside OtelIngestionProcessor. It runs on every span and produces a standardized usage_details map that downstream cost calculation depends on.

Entry Points#

Ingestion pipeline path: createObservationEvent() calls extractUsageDetails() and stores the result as usageDetails on the observation body.
Event processing path (used by the non-queue path): processToEvent() calls extractUsageDetails() via UsageDetails.safeParse() , storing the result as providedUsageDetails.

The attribute key for the Langfuse-native usage blob is defined as OBSERVATION_USAGE_DETAILS = "langfuse.observation.usage_details" in LangfuseOtelSpanAttributes.

Extraction Priority#

extractUsageDetails() applies sources in this priority order :

langfuse.observation.usage_details — JSON-encoded object on the span attribute. Parsed via parseJsonObjectAttribute(); if present (even as {}), all other sources are skipped.
Genkit (genkit-tracer scope) — reads genkit:output.usage.{inputTokens, outputTokens, totalTokens, thoughtsTokens}.
Vercel AI SDK (ai scope) — reads gen_ai.usage.prompt_tokens / gen_ai.usage.input_tokens, gen_ai.usage.completion_tokens / gen_ai.usage.output_tokens, and ai.usage.tokens for total. Also enriches with provider metadata from ai.response.providerMetadata (OpenAI, Anthropic, Bedrock cache details).
Pydantic AI (pydantic-ai scope) — delegates to extractGenericGenAiUsageDetails().
Generic gen_ai.usage.* / llm.token_count.* — fallback for all other instrumentation (OpenInference, TraceLoop, etc.) via extractGenericGenAiUsageDetails().

Generic `gen_ai.usage.*` Normalization#

extractGenericGenAiUsageDetails() collects all keys matching gen_ai.usage.* (excluding gen_ai.usage.cost, which goes to cost details) or llm.token_count.*, strips the prefix, then normalizes to canonical Langfuse keys :

OTel attribute suffix(es)	Normalized key
`prompt_tokens`, `input_tokens`, `prompt`	`input` (minus cache tokens)
`completion_tokens`, `output_tokens`, `completion`	`output`
`total_tokens`, `total`	`total`
`cache_read.input_tokens`, `cache_read_tokens`, `details.cache_read_tokens`, `details.cache_read_input_tokens`	`input_cached_tokens`
`cache_creation.input_tokens`, `cache_write_tokens`, `details.cache_write_tokens`, `details.cache_creation_input_tokens`	`input_cache_creation`
anything else (e.g. `output_reasoning_tokens`)	kept as-is (with `details.` prefix stripped)

input is deducted by cache read + cache creation tokens to avoid double-counting .

Vercel AI SDK: Provider Metadata Enrichment#

When instrumentationScopeName === "ai", the processor additionally parses ai.response.providerMetadata JSON to extract provider-specific breakdowns:

OpenAI: cachedPromptTokens → input_cached_tokens; reasoningTokens → output_reasoning_tokens; acceptedPredictionTokens, rejectedPredictionTokens.
Anthropic: cache_creation_input_tokens → input_cache_creation; cache_read_input_tokens → input_cached_tokens; Anthropic TTL-specific fields ephemeral_5m_input_tokens → input_cache_creation_5m, ephemeral_1h_input_tokens → input_cache_creation_1h.
Bedrock: cacheReadInputTokens → input_cache_read; cacheWriteInputTokens → input_cache_write; cacheCreationInputTokens → input_cache_creation.

After enrichment, input and output are floor-clamped to 0 after subtracting sub-token counts .

extractCostDetails() is a sibling method. It reads langfuse.observation.cost_details (JSON blob) first; if absent, maps gen_ai.usage.cost → { total: <value> }.

Validation#

After extraction, the result is validated against the UsageDetails Zod schema. Failures are logged as warnings but do not drop the span .

Key Source Files#

File	Purpose
`OtelIngestionProcessor.ts`	All extraction logic (`extractUsageDetails`, `extractGenericGenAiUsageDetails`, `extractCostDetails`)
`attributes.ts`	`LangfuseOtelSpanAttributes` enum — defines `OBSERVATION_USAGE_DETAILS` and `OBSERVATION_COST_DETAILS` keys