Skip to content

Writing observations

This file covers the four primitives for emitting observations and the helpers for enriching them at runtime. After reading 01-tracing-setup.md, this is where you learn to instrument code.

When to use what

Situation Use
Named function that represents a unit of work @step("name")
Code block that does not map to a single function span("name")
Point-in-time marker — cache hit, fallback trigger, decision point event("name")
Top-level entry where you set user / session context propagate(...) wrapping the root call

@step

@step wraps a function as a Langfuse observation. It works on both sync and async functions. Arguments are automatically captured as the observation input; the return value is captured as the output.

from semantic_tracing import step, update_current_step

@step("route_request")
async def route(query: str) -> RoutingDecision:
    decision = deterministic_route(query)
    update_current_step(metadata={"handoff_to": decision.handoff_to})
    return decision

Resulting trace structure:

trace
└── route_request (span)
      input:    {"query": "..."}
      output:   {"handoff_to": "A", "reason": "..."}
      metadata: {"handoff_to": "A"}

Observation type

Pass as_type to give the observation a semantic type visible in the Langfuse UI:

@step("fetch_documents", as_type="retriever")
async def retrieve(query: str) -> list[Document]:
    ...

@step("check_safety", as_type="guardrail")
def validate_output(text: str) -> bool:
    ...

Controlling capture

Pass capture_input=False or capture_output=False to suppress automatic recording of arguments or return values for a specific step:

@step("process_image", capture_input=False)
async def handle(image_bytes: bytes) -> dict:
    ...

See 04-tracing-input-capture.md for argument-level filtering without disabling capture entirely.

span()

span() is a context manager for tracing a block of code that does not map to a single function. It yields a span object whose .update() method lets you attach data from inside the block.

from semantic_tracing import span

async def process_batch(items: list[Item]) -> list[Result]:
    results = []
    for item in items:
        with span(f"validate_{item.id}") as s:
            outcome = validate(item)
            if not outcome.valid:
                s.update(level="WARNING", output={"reason": outcome.reason})
                break
            results.append(outcome)
            s.update(output={"status": "passed"})
    return results

span() accepts the same as_type parameter as @step.

Generation observations with span()

When the code that calls the LLM is already inside another with block — for example, a route handler that opens a chain span first — extract it into its own function just to use @step would be artificial. Use span(as_type="generation") and call .update() with the same generation kwargs that update_current_step() accepts: model, usage, completion_start_time, and usage_details.

with span("answer", as_type="generation") as gen:
    response = await call_llm(prompt)
    gen.update(
        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},
    )

usage={"input": N, "output": M} is translated to usage_details= before reaching the Langfuse SDK (which uses usage_details exclusively). The public kwarg name is the same as in update_current_step() so call sites are consistent. For full TTFT and streaming guidance see 03-tracing-generation.md.

event()

event() records a point-in-time marker with no duration. Use it to mark decisions, fallbacks, or conditions that are worth surfacing in a trace without creating a time range.

from semantic_tracing import event

if not tool_state["called"]:
    decision = deterministic_fallback(query)
    event("routing_fallback_used", metadata={
        "reason": "agent did not call route_query",
        "fallback_target": decision.handoff_to,
    })

propagate()

propagate() stamps all child spans with user_id, session_id, and tags. Wrap it around the top-level call where those values become known:

from semantic_tracing import propagate, step

@step("workflow_root")
async def run_workflow(request: WorkflowRequest):
    with propagate(
        user_id=request.user_id,
        session_id=request.session_id,
        tags=["offline-eval", "v2"],
    ):
        result = await orchestrate(request)
    return result

All @step and span() calls inside the propagate() scope inherit the user_id, session_id, and tags — they appear on every observation in Langfuse without being set individually.

Trace-level input

Supply input to record the semantically meaningful user input on the trace's root observation. In Langfuse a trace has no input of its own: it shows the root observation's. This is the text a human reviewer should see at a glance when opening the trace:

with propagate(
    user_id=request.user_id,
    session_id=request.session_id,
    input={"query": request.query},
):
    result = await handle(request)

input is written as the input of the first span started inside the with block, which is the root observation when propagate() wraps the turn. A span that sets its own input keeps it. Pass the user-facing query text, not the full internal request object — reviewers need the signal, not the noise.

Suppressing Langfuse export for a scope

Set langfuse_export=False to exclude the entire scope from Langfuse while keeping the OTel trace intact. App Insights still receives every span; context propagates normally; only the Langfuse exporter is bypassed for that scope.

with propagate(langfuse_export=False):
    await run_smoke_test(request)

Use cases: - Background batch jobs that do not need Langfuse visibility - Synthetic smoke tests running in production - Feature-flagged paths you want to keep invisible in Langfuse during a rollout

This is independent of sample_rate in setup_tracing(). langfuse_export=False is a hard exclusion for a specific call; sample_rate is a statistical exclusion across all calls.

Ordering with span()

When combining propagate() with an explicit span() in the same scope — for example, in a FastAPI route handler where the OTel root is an App Insights HTTP span rather than a Langfuse observation — always open span() first and call propagate() inside it.

# Langfuse observation is active when propagate_attributes() fires
with span("request.handle", as_type="chain", input=payload) as s:
    with propagate(user_id=..., session_id=..., trace_name="request.handle"):
        result = await handle(payload)

# propagate_attributes() runs before any Langfuse span exists
with propagate(user_id=..., session_id=..., trace_name="request.handle"):
    with span("request.handle", as_type="chain", input=payload) as s:
        result = await handle(payload)

The wrong order produces a log warning on every request: WARNING langfuse — Context error: No active span in current context.

This does not apply when propagate() is called inside a @step-decorated function — @step opens the Langfuse observation first, so the ordering is already correct.

propagate() parameter reference

Parameter Type Default Description
user_id str \| None None Langfuse user identifier
session_id str \| None None Groups turns of the same conversation in Langfuse Sessions; also used as the sampling key when sample_rate < 1.0
tags list[str] \| None None Tags applied to the trace
metadata dict \| Pydantic model \| None None Metadata propagated to all child observations. A release key (release_id, instructions_version, knowledge_version) overrides the release for this request; see Releases
version str \| None None Langfuse's version field for the trace. It is not the release; set that with Release (Releases)
trace_name str \| None None Override trace name; when omitted, auto-derived from the active OTel span name
langfuse_export bool True When False, suppresses Langfuse export for all spans in this scope
input Any None Input recorded on the first span started inside the scope (the root observation when propagate() wraps the turn)
log_trace_url bool False When True, emits the Langfuse trace URL as an INFO log line (logger semantic_tracing) when the propagate() scope exits — useful during local development so engineers can jump directly to the trace

update_current_step()

Call update_current_step() from inside a @step-decorated function to attach data that is only available after execution starts:

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(...)
    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

model and usage together with as_type="generation" activate the Langfuse token cost widget. Passing them through metadata does not trigger cost tracking — they must be top-level kwargs.

Parameter Type Description
metadata dict \| Pydantic model Merged into observation metadata
output Any Overrides the auto-captured return value
level str "DEBUG", "DEFAULT", "WARNING", or "ERROR"
input Any Overrides or sets the observation input after the function started
model str Model name or deployment ID — activates cost tracking with as_type="generation"
usage dict Token counts — {"input": N, "output": M} — activates cost tracking with model
completion_start_time datetime Timestamp of the first token — drives Langfuse native TTFT metric; only meaningful on as_type="generation" spans
usage_details dict Extended token breakdown — {"cached_tokens": N, "reasoning_tokens": M}; only meaningful on as_type="generation" spans
model_parameters dict Sampling parameters (e.g. {"temperature": 0.0}, {"reasoning_effort": "medium"}); stored in the Langfuse generation panel; only meaningful on as_type="generation" spans

When any generation-specific kwarg (model, usage, usage_details, completion_start_time, model_parameters) is present, update_current_step() dispatches to langfuse.update_current_generation() so those fields are actually stored by the Langfuse SDK. Base kwargs (metadata, output, level, input) are accepted by both dispatch paths.

For generation-specific usage of completion_start_time and usage_details, and for the record_first_token() and wrap_azure_stream() helpers, see 03-tracing-generation.md.

Observation types (as_type)

The as_type parameter is accepted by both @step and span(). The value is passed directly to the Langfuse SDK and controls the icon and category visible in the trace view.

as_type Meaning
"span" Generic unit of work with a duration (default when as_type is omitted)
"generation" LLM call — supports token usage and cost tracking
"embedding" Embedding model call — supports token usage
"agent" Agent deciding application flow
"tool" Tool call (external API, function call, etc.)
"chain" Link between steps passing context
"retriever" Data retrieval from a vector store or database
"evaluator" Assessment function
"guardrail" Content safety or validation check

If Langfuse adds new types in future SDK versions they work immediately, because the value passes through without validation at the SDK level.

Observation nesting

Each @step or span() that executes inside another creates a child observation automatically through OTel context propagation. event() creates an instant marker with no duration at the current nesting level.

flowchart TD
  TR["Langfuse trace"]
  S1["span: workflow root"]
  S2["agent: route request"]
  S3["generation: call LLM"]
  S4["tool: search db"]
  E1["event: cache hit"]

  TR --> S1
  S1 --> S2
  S1 --> S3
  S1 --> E1
  S3 --> S4

  style TR fill:#efe9ed,stroke:#62294b,stroke-width:2px
  style S3 fill:#e5f4fd,stroke:#0098eb,stroke-width:2px

The depth mirrors the call stack. There is no manual parent-ID wiring — nesting is automatic as long as child calls happen within the lexical scope of the parent @step, span(), or propagate().