LLM generation spans

as_type="generation" marks an observation as an LLM call in Langfuse. Combined with model and token usage data, it activates the Langfuse cost dashboard and TTFT panel. This file covers everything beyond the basic @step wrapper that matters specifically for generation spans.

Read 02-tracing-observations.md first — the @step, update_current_step(), and as_type mechanics covered there apply here too.

Why generation spans are special

A generation span in Langfuse has fields that other span types do not use:

Field	Set via	Langfuse effect
`model`	`update_current_step(model=...)`	Groups cost by model in the cost dashboard
`usage`	`update_current_step(usage={"input": N, "output": M})`	Shows prompt/completion token counts; drives cost calculation
`completion_start_time`	`update_current_step(completion_start_time=dt)`	Drives the native TTFT metric shown in the generation panel
`usage_details`	`update_current_step(usage_details={...})`	Extended token breakdown — cached tokens, reasoning tokens
`model_parameters`	`update_current_step(model_parameters={...})`	Sampling parameters (temperature, reasoning_effort) shown in the generation panel

All five require as_type="generation" on the enclosing @step or span(). Passing them through metadata does not activate cost or TTFT tracking.

When any of these generation-specific kwargs is present, update_current_step() dispatches to langfuse.update_current_generation() instead of update_current_span(). Both functions are available when there is an active Langfuse span in context — from @step or from span() — so the SDK-level helpers work in both patterns.

Generation levels (0–3)

Four instrumentation levels cover the range from non-streaming to fully-automated streaming. Pick the level that matches your call pattern.

Level 0Level 1Level 2Level 3

The simplest case: the client.chat.completions.create() call is synchronous or awaited, and the full response object is available before returning. TTFT is not tracked because the first-token time is not distinguishable from the final-response time in a non-streaming call.

from semantic_tracing import step, update_current_step

@step("summarize", as_type="generation")
async def summarize(text: str) -> str:
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"Summarize:\n{text}"}],
    )
    update_current_step(
        model="gpt-4o",
        usage={
            "input": response.usage.prompt_tokens,
            "output": response.usage.completion_tokens,
        },
        output=response.choices[0].message.content,
    )
    return response.choices[0].message.content

When streaming, you control the chunk loop. Record TTFT yourself by calling update_current_step(completion_start_time=datetime.now(timezone.utc)) at the moment the first content chunk arrives. After the loop, call it again with usage data.

from datetime import datetime, timezone
from semantic_tracing import step, update_current_step

@step("stream_response", as_type="generation")
async def stream(prompt: str) -> str:
    stream = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        stream_options={"include_usage": True},  # required for usage on last chunk
    )
    chunks = []
    first_token_recorded = False
    model = None

    async for chunk in stream:
        if model is None:
            model = chunk.model
        delta = chunk.choices[0].delta.content if chunk.choices else None
        if delta is not None and not first_token_recorded:
            first_token_recorded = True
            # stamp TTFT the moment the first content token arrives
            update_current_step(completion_start_time=datetime.now(timezone.utc))
        if delta:
            chunks.append(delta)
        if chunk.usage:
            # Final chunk carries usage when stream_options={"include_usage": True}
            update_current_step(
                model=model,
                usage={
                    "input": chunk.usage.prompt_tokens,
                    "output": chunk.usage.completion_tokens,
                },
                usage_details={
                    k: v for k, v in {
                        "cached_tokens": getattr(
                            chunk.usage.prompt_tokens_details, "cached_tokens", None
                        ),
                        "reasoning_tokens": getattr(
                            chunk.usage.completion_tokens_details, "reasoning_tokens", None
                        ),
                    }.items() if v
                },
            )

    return "".join(chunks)

record_first_token() is a convenience wrapper for the TTFT stamp. It calls update_current_step(completion_start_time=datetime.now(timezone.utc)) internally, removing the boilerplate from the chunk loop.

from semantic_tracing import record_first_token, step, update_current_step

@step("stream_response", as_type="generation")
async def stream(prompt: str) -> str:
    stream = await client.chat.completions.create(...)
    chunks = []
    first_token_recorded = False

    async for chunk in stream:
        delta = chunk.choices[0].delta.content if chunk.choices else None
        if delta is not None and not first_token_recorded:
            first_token_recorded = True
            record_first_token()  # cleaner than datetime boilerplate
        ...

wrap_azure_stream() handles TTFT recording and usage extraction for you, wrapping the chunk loop. It is compatible with openai.AsyncOpenAI and openai.AsyncAzureOpenAI streaming responses. The caller's loop is unchanged — wrap_azure_stream() transparently yields the same chunks the original stream yields, then calls update_current_step() once after the stream ends.

from semantic_tracing import step, wrap_azure_stream

@step("stream_response", as_type="generation")
async def stream(prompt: str) -> str:
    raw = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        stream_options={"include_usage": True},
    )
    chunks = []
    async for chunk in wrap_azure_stream(raw):
        delta = chunk.choices[0].delta.content if chunk.choices else None
        if delta:
            chunks.append(delta)
    return "".join(chunks)

Using span() instead of @step

The examples above use @step because most LLM calls live inside a dedicated function. When the call is already inside a with block — a route handler that opens a chain span first, or a loop that manages its own context — wrapping it in an extra function just to use @step is unnecessary. Use span() and update_current_step() directly:

with span("answer", as_type="generation"):
    response = await call_llm(prompt)
    update_current_step(
        output=response.text,
        model="gpt-4o",
        usage={"input": response.usage.input_tokens, "output": response.usage.output_tokens},
        usage_details={"cached_tokens": response.usage.cached_tokens},
    )

update_current_step() looks up the current Langfuse observation from the OTel context stack — inside with span("answer", as_type="generation"): that observation is the answer generation span. The same dispatch logic applies: generation-specific kwargs trigger update_current_generation() so they are stored correctly.

The levels (0–3) for TTFT and streaming apply inside with span(...): blocks exactly as they do inside @step functions — call update_current_step(completion_start_time=...) manually (Level 1), record_first_token() (Level 2), or wrap_azure_stream() (Level 3).

When to use each level

Situation	Level
Non-streaming response, no TTFT needed	Level 0 — `update_current_step(model=, usage=)` after the call
Streaming, you write your own chunk loop, TTFT matters	Level 1 — `update_current_step(completion_start_time=...)` manually
Same as Level 1 but want less boilerplate	Level 2 — `record_first_token()` inside your loop
OpenAI / Azure OpenAI streaming, no custom loop logic needed	Level 3 — `wrap_azure_stream()` around the raw stream

Important notes for Level 3

Requires stream_options={"include_usage": True} in the create() call. Without it, the final chunk has no usage field and token counts are not recorded.
Does not import from openai — duck-typed on chunk attributes. Works with any client that produces chunks with .choices[0].delta.content, .model, and .usage (on the final chunk).
wrap_azure_stream() calls update_current_step() once after the stream ends. If your loop also calls update_current_step(), both calls are merged by Langfuse.