Instrumentation

Langfuse SDKs lets you manually create observations and traces. You can also use the manual creation patters together with one of the native integrations.

Native integrations

Langfuse supports native integrations for popular LLM and agent libraries such as OpenAI, LangChain or the Vercel AI SDK. They automatically create observations and traces and capture prompts, responses, usage, and errors.

Custom observations

For some use cases you might want to have more control over the observations and traces. For this, you can create custom observations using the Langfuse SDK. The SDKs provide 3 ways to create custom observations:

Context manager
Observe wrapper
Manual observations

All custom patterns are interoperable. You can nest a decorator-created observation inside a context manager or mix manual spans with native integrations.

Context manager

The context manager allows you to create a new span and set it as the currently active observation in the OTel context for its duration. Any new observations created within this block will be its children.

langfuse.start_as_current_observation() is the primary way to create observations while ensuring the active OpenTelemetry context is updated. Any child observations created inside the with block inherit the parent automatically.

from langfuse import get_client, propagate_attributes
 
langfuse = get_client()
 
with langfuse.start_as_current_observation(
    as_type="span",
    name="user-request-pipeline",
    input={"user_query": "Tell me a joke"},
) as root_span:
    with propagate_attributes(user_id="user_123", session_id="session_abc"):
        with langfuse.start_as_current_observation(
            as_type="generation",
            name="joke-generation",
            model="gpt-4o",
        ) as generation:
            generation.update(output="Why did the span cross the road?")
 
    root_span.update(output={"final_joke": "..."})

Observe wrapper

Use the observe wrapper to automatically capture inputs, outputs, timings, and errors of a wrapped function without modifying the function’s internal logic.

from langfuse import observe
 
@observe()
def my_data_processing_function(data, parameter):
    return {"processed_data": data, "status": "ok"}
 
@observe(name="llm-call", as_type="generation")
async def my_async_llm_call(prompt_text):
    return "LLM response"

Parameters: name, as_type, capture_input, capture_output, transform_to_string. Special kwargs such as langfuse_trace_id or langfuse_parent_observation_id let you stitch into existing traces.

The decorator automatically propagates the OTEL trace context. Pass langfuse_trace_id when you need to force a specific trace ID (e.g., to align with an external system) and langfuse_parent_observation_id to attach to an existing parent span.

Capturing large inputs/outputs may add overhead. Disable IO capture per decorator (capture_input=False, capture_output=False) or via the LANGFUSE_OBSERVE_DECORATOR_IO_CAPTURE_ENABLED env var.

Option	Description	Default
`name`	Observation name	Function name
`asType`	Observation type (`span`, `generation`, etc.)	`"span"`
`captureInput`	Capture arguments as `input`	`true`
`captureOutput`	Capture return values/errors as `output`	`true`

Manual observations

Manual APIs are useful when you need to create spans without altering the currently active context (e.g., background work or parallel tasks).

Use start_span() / start_generation() when you need manual control without changing the active context.

from langfuse import get_client
 
langfuse = get_client()
 
span = langfuse.start_span(name="manual-span")
span.update(input="Data for side task")
child = span.start_span(name="child-span")
child.end()
span.end()

⚠️

Spans created via start_span() / start_generation() must be ended explicitly via .end().

Nesting observations

The function call hierarchy is automatically captured by the @observe decorator and reflected in the trace.

from langfuse import observe
 
@observe
def my_data_processing_function(data, parameter):
    return {"processed_data": data, "status": "ok"}
 
 
@observe
def main_function(data, parameter):
    return my_data_processing_function(data, parameter)

Update observations

Update observation objects directly with .update() or use context-aware helpers with .update_current_span().

from langfuse import get_client
 
langfuse = get_client()
 
with langfuse.start_as_current_observation(as_type="generation", name="llm-call") as gen:
    gen.update(
        input={"prompt": "Why is the sky blue?"},
        output="Rayleigh scattering",
        usage_details={"input_tokens": 5, "output_tokens": 50},
    )
 
with langfuse.start_as_current_observation(as_type="span", name="data-processing"):
    langfuse.update_current_span(metadata={"step1_complete": True})

Add attributes to observations

Propagate attributes such as userId, sessionId, metadata, version, and tags to keep downstream analytics consistent. These helpers mirror the Python propagate_attributes context manager and the TypeScript propagateAttributes callback wrapper from the standalone SDK docs. Use propagation for attributes that should appear on every observation and updateTrace()/update_current_trace() for single-trace fields like name, input, output, or public.

Propagatable attributes

userId / user_id
sessionId / session_id
metadata
version
tags

Trace-only attributes (use updateTrace / update_current_trace)

name
input
output
public

from langfuse import get_client, propagate_attributes
 
langfuse = get_client()
 
with langfuse.start_as_current_observation(as_type="span", name="user-workflow"):
    with propagate_attributes(
        user_id="user_123",
        session_id="session_abc",
        metadata={"experiment": "variant_a"},
        version="1.0",
    ):
        with langfuse.start_as_current_observation(as_type="generation", name="llm-call"):
            pass

Note on Attribute Propagation

We use Attribute Propagation to propagate specific attributes (userId, sessionId, version, tags, metadata) across all observations in an execution context. We will use all observations with these attributes to calculate attribute-level metrics. Please consider the following when using Attribute Propagation:

Values must be strings ≤200 characters
Metadata keys: Alphanumeric characters only (no whitespace or special characters)
Call early in your trace to ensure all observations are covered. This way you make sure that all Metrics in Langfuse are accurate.
Invalid values are dropped with a warning

Cross-service propagation

Use baggage propagation only when you need to forward attributes across HTTP boundaries. It pushes the values into every outbound request header, so prefer non-sensitive identifiers (session IDs, experiment versions, etc.).

from langfuse import get_client, propagate_attributes
import requests
 
langfuse = get_client()
 
with langfuse.start_as_current_observation(as_type="span", name="api-request"):
    with propagate_attributes(
        user_id="user_123",
        session_id="session_abc",
        as_baggage=True,
    ):
        requests.get("https://service-b.example.com/api")

⚠️

When baggage propagation is enabled, attributes are added to all outbound HTTP headers. Only use it for non-sensitive values needed for distributed tracing.

Trace-level metadata & inputs/outputs

By default, trace input/output mirror whatever you set on the root observation. Override them explicitly whenever evaluations, AB-tests, or judge models need a different payload than the root span captured.

The snippets below illustrate both the default behavior and how to call update_current_trace / updateActiveTrace() to set trace-level payloads later in the workflow.

LLM-as-a-judge and evaluation workflows typically rely on trace-level inputs/outputs. Make sure to set them deliberately rather than relying on the root span if your evaluation payload differs.

Trace input/output default to the root observation. Override them explicitly when needed (e.g., for evaluations).

from langfuse import get_client
 
langfuse = get_client()
 
with langfuse.start_as_current_observation(as_type="span", name="complex-pipeline") as root_span:
    root_span.update(input="Step 1 data", output="Step 1 result")
    root_span.update_trace(
        input={"original_query": "User question"},
        output={"final_answer": "Complete response", "confidence": 0.95},
    )

from langfuse import observe, get_client
 
langfuse = get_client()
 
@observe()
def process_user_query(user_question: str):
    answer = call_llm(user_question)
    langfuse.update_current_trace(
        input={"question": user_question},
        output={"answer": answer},
    )
    return answer

Trace and observation IDs

Langfuse follows the W3C Trace Context standard: trace IDs are 32-character lowercase hex strings (16 bytes) and observation IDs are 16-character lowercase hex strings (8 bytes). You cannot set arbitrary observation IDs, but you can generate deterministic trace IDs to correlate with external systems.

Langfuse uses W3C Trace Context IDs. Access current IDs or create deterministic ones.

from langfuse import get_client, Langfuse
 
langfuse = get_client()
 
with langfuse.start_as_current_observation(as_type="span", name="my-op") as current_op:
    trace_id = langfuse.get_current_trace_id()
    observation_id = langfuse.get_current_observation_id()
    print(trace_id, observation_id)
 
external_request_id = "req_12345"
deterministic_trace_id = Langfuse.create_trace_id(seed=external_request_id)

Link to existing traces

When integrating with upstream services that already have trace IDs, supply the W3C trace context so Langfuse spans join the existing tree rather than creating a new one.

from langfuse import get_client
 
langfuse = get_client()
 
existing_trace_id = "abcdef1234567890abcdef1234567890"
existing_parent_span_id = "fedcba0987654321"
 
with langfuse.start_as_current_observation(
    as_type="span",
    name="process-downstream-task",
    trace_context={
        "trace_id": existing_trace_id,
        "parent_span_id": existing_parent_span_id,
    },
):
    pass

Client lifecycle & flushing

Both SDKs buffer spans in the background. Always flush or shut down the exporter in short-lived processes (scripts, serverless functions, workers) to avoid losing data.

Flush or shut down the client to ensure all buffered data is delivered—especially in short-lived jobs.

from langfuse import get_client
 
langfuse = get_client()
# ... create traces ...
langfuse.flush()
langfuse.shutdown()

Overview Advanced Features

Was this page helpful?

Support