Advanced features

Default behavior (recommended):

from langfuse import Langfuse

# Smart default filter (Langfuse + GenAI/LLM spans)
langfuse = Langfuse()

Export everything:

from langfuse import Langfuse

langfuse = Langfuse(should_export_span=lambda span: True)

Passing should_export_span replaces the default filter. To keep default behavior and extend it, compose with is_default_export_span.

Compose custom logic with built-in predicates:

from langfuse import Langfuse
from langfuse.span_filter import is_default_export_span

langfuse = Langfuse(
    should_export_span=lambda span: (
        is_default_export_span(span)
        or (
            span.instrumentation_scope is not None
            and span.instrumentation_scope.name.startswith("my_framework")
        )
    )
)

Only export spans created by the Langfuse SDK:

from langfuse import Langfuse
from langfuse.span_filter import is_langfuse_span

langfuse = Langfuse(should_export_span=is_langfuse_span)

Available Python helpers: is_default_export_span, is_langfuse_span, is_genai_span, is_known_llm_instrumentor, KNOWN_LLM_INSTRUMENTATION_SCOPE_PREFIXES.

To block specific instrumentation scopes, express the deny rule in should_export_span (blocked_instrumentation_scopes was removed in Python SDK v5):

from langfuse import Langfuse
from langfuse.span_filter import is_default_export_span

blocked_scopes = {"sqlalchemy", "psycopg"}

langfuse = Langfuse(
    should_export_span=lambda span: (
        span.instrumentation_scope is None
        or span.instrumentation_scope.name not in blocked_scopes
    )
    and is_default_export_span(span)
)

Default behavior (recommended):

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";

const sdk = new NodeSDK({
  // Smart default filter (Langfuse + GenAI/LLM spans)
  spanProcessors: [new LangfuseSpanProcessor()],
});

sdk.start();

Custom filtering:

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor, ShouldExportSpan } from "@langfuse/otel";

const shouldExportSpan: ShouldExportSpan = ({ otelSpan }) =>
  otelSpan.instrumentationScope.name !== "express";

const sdk = new NodeSDK({
  spanProcessors: [new LangfuseSpanProcessor({ shouldExportSpan })],
});

sdk.start();

Passing shouldExportSpan replaces the default filter, so include default conditions in your callback if you want to extend rather than replace the default behavior.

Compose with the default filter:

import { isDefaultExportSpan, type ShouldExportSpan } from "@langfuse/otel";

const shouldExportSpan: ShouldExportSpan = ({ otelSpan }) =>
  isDefaultExportSpan(otelSpan) ||
  otelSpan.instrumentationScope.name.startsWith("my-framework");

Available JS/TS helpers from @langfuse/otel: isDefaultExportSpan, isLangfuseSpan, isGenAISpan, isKnownLLMInstrumentor, KNOWN_LLM_INSTRUMENTATION_SCOPE_PREFIXES.

Export everything:

new LangfuseSpanProcessor({ shouldExportSpan: () => true });

Use mask_otel_spans to return sparse patches for the OpenTelemetry spans that should change before export.

import re
from typing import Optional

from langfuse import Langfuse
from langfuse.types import (
    MaskOtelSpansParams,
    MaskOtelSpansResult,
    OtelSpanPatch,
)

email_pattern = re.compile(r"\b[\w.-]+?@[\w.-]+?\.\w+?\b")


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_value = email_pattern.sub("[EMAIL_REDACTED]", value)

                if masked_value != value:
                    replacements[key] = masked_value

        if replacements:
            patches[identifier] = OtelSpanPatch(set_attributes=replacements)

    return MaskOtelSpansResult(span_patches=patches)


langfuse = Langfuse(mask_otel_spans=mask_otel_spans)

mask_otel_spans is synchronous and usually runs on the OpenTelemetry batch span processor worker thread; during flush() and shutdown it may run on the caller thread. Keep the function fast to avoid backing up the export queue. See masking for the full behavior, error handling, and legacy mask comparison.

You can provide a mask function to the LangfuseSpanProcessor. This function will be applied to the input, output, and metadata of every observation.

The function receives an object { data }, where data is the stringified JSON of the attribute's value. It should return the masked data.

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";

const spanProcessor = new LangfuseSpanProcessor({
  mask: ({ data }) =>
    data.replace(/\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b/g, "***MASKED_CREDIT_CARD***"),
});

const sdk = new NodeSDK({ spanProcessors: [spanProcessor] });

sdk.start();

Via environment variable:

You can set the log level using the LANGFUSE_DEBUG environment variable to enable the debug mode.

export LANGFUSE_DEBUG="True"

In code:

The Langfuse SDK uses Python's standard logging module. The main logger is named "langfuse". To enable detailed debug logging, you can either:

1. Set the debug=True parameter when initializing the Langfuse client.
2. Configure the "langfuse" logger manually:

import logging

langfuse_logger = logging.getLogger("langfuse")
langfuse_logger.setLevel(logging.DEBUG)

The default log level for the langfuse logger is logging.WARNING.

You can configure the global SDK logger to control the verbosity of log output. This is useful for debugging.

Via environment variable:

You can set the log level using the LANGFUSE_LOG_LEVEL environment variable to enable the debug mode.

export LANGFUSE_LOG_LEVEL="DEBUG"

In code:

import { configureGlobalLogger, LogLevel } from "@langfuse/core";

// Set the log level to DEBUG to see all log messages
configureGlobalLogger({ level: LogLevel.DEBUG });

Available log levels are DEBUG, INFO, WARN, and ERROR.

In code:

You can configure the SDK to sample traces by setting the sample_rate parameter during client initialization. This value should be a float between 0.0 (sample 0% of traces) and 1.0 (sample 100% of traces).

If a trace is not sampled, none of its observations or associated scores will be sent to Langfuse.

from langfuse import Langfuse

# Sample approximately 20% of traces
langfuse_sampled = Langfuse(sample_rate=0.2)

Via environment variable:

You can also set the sample rate using the LANGFUSE_SAMPLE_RATE environment variable.

export LANGFUSE_SAMPLE_RATE="0.2"

In code:

Langfuse respects OpenTelemetry's sampling decisions. Configure a sampler on your OTEL NodeSDK to control which traces reach Langfuse and reduce noise/costs in high-volume workloads.

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";
import { TraceIdRatioBasedSampler } from "@opentelemetry/sdk-trace-base";

const sdk = new NodeSDK({
  sampler: new TraceIdRatioBasedSampler(0.2),
  spanProcessors: [new LangfuseSpanProcessor()],
});

sdk.start();

Via environment variable:

You can also set the sample rate using the LANGFUSE_SAMPLE_RATE environment variable.

export LANGFUSE_SAMPLE_RATE="0.2"

Isolate Langfuse spans with a custom provider and avoid sending them to other exporters.

import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
import { setLangfuseTracerProvider } from "@langfuse/tracing";

// Create a new TracerProvider and register the LangfuseSpanProcessor
// do not set this TracerProvider as the global TracerProvider
const langfuseTracerProvider = new NodeTracerProvider({
  spanProcessors: [new LangfuseSpanProcessor()],
})

// Register the isolated TracerProvider
setLangfuseTracerProvider(langfuseTracerProvider)

The Langfuse Python SDK supports routing traces to different projects within the same application by using multiple public keys. This works because the Langfuse SDK adds a specific span attribute containing the public key to all spans it generates.

How it works:

1. Span Attributes: The Langfuse SDK adds a specific span attribute containing the public key to spans it creates
2. Multiple Processors: Multiple span processors are registered onto the global tracer provider, each with their respective exporters bound to a specific public key
3. Filtering: Within each span processor, spans are filtered based on the presence and value of the public key attribute

Important Limitation with Third-Party Libraries:

Third-party libraries that emit OpenTelemetry spans automatically (e.g., HTTP clients, databases, other instrumentation libraries) do not have the Langfuse public key span attribute. As a result:

• Third-party spans that pass the export filter and have no public key cannot be routed to a specific project
• These spans are processed by all span processors and can be sent to all projects
• With the default filter, this mainly affects GenAI/LLM spans from third-party instrumentors (infrastructure spans are typically filtered out)

Why is this experimental? This approach requires that the public_key parameter be passed to all Langfuse SDK executions across all integrations to ensure proper routing, and third-party spans that pass filtering may appear in all projects.

Initialization

To set up multiple projects, initialize separate Langfuse clients for each project:

from langfuse import Langfuse

# Initialize clients for different projects
project_a_client = Langfuse(
    public_key="pk-lf-project-a-...",
    secret_key="sk-lf-project-a-...",
    base_url="https://cloud.langfuse.com"
)

project_b_client = Langfuse(
    public_key="pk-lf-project-b-...",
    secret_key="sk-lf-project-b-...",
    base_url="https://cloud.langfuse.com"
)

Integration Usage

For all integrations in multi-project setups, you must specify the public_key parameter to ensure traces are routed to the correct project.

Observe Decorator:

Pass langfuse_public_key as a keyword argument to the top-most observed function (not the decorator). From Python SDK >= 3.2.2, nested decorated functions will automatically pick up the public key from the execution context they are currently into. Also, calls to get_client will be also aware of the current langfuse_public_key in the decorated function execution context, so passing the langfuse_public_key here again is not necessary.

from langfuse import observe

@observe
def nested():
    # get_client call is context aware
    # if it runs inside another decorated function that has
    # langfuse_public_key passed, it does not need passing here again


@observe
def process_data_for_project_a(data):
    # passing `langfuse_public_key` here again is not necessarily
    # as it is stored in execution context
    nested()

    return {"processed": data}

@observe
def process_data_for_project_b(data):
    # passing `langfuse_public_key` here again is not necessarily
    # as it is stored in execution context
    nested()

    return {"enhanced": data}

# Route to Project A
# Top-most decorated function needs `langfuse_public_key` kwarg
result_a = process_data_for_project_a(
    data="input data",
    langfuse_public_key="pk-lf-project-a-..."
)

# Route to Project B
# Top-most decorated function needs `langfuse_public_key` kwarg
result_b = process_data_for_project_b(
    data="input data",
    langfuse_public_key="pk-lf-project-b-..."
)

OpenAI Integration:

Add langfuse_public_key as a keyword argument to the OpenAI execution:

from langfuse.openai import openai

client = openai.OpenAI()

# Route to Project A
response_a = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello from Project A"}],
    langfuse_public_key="pk-lf-project-a-..."
)

# Route to Project B
response_b = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello from Project B"}],
    langfuse_public_key="pk-lf-project-b-..."
)

Langchain Integration:

Add public_key to the CallbackHandler constructor:

from langfuse.langchain import CallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Create handlers for different projects
handler_a = CallbackHandler(public_key="pk-lf-project-a-...")
handler_b = CallbackHandler(public_key="pk-lf-project-b-...")

llm = ChatOpenAI(model_name="gpt-4o")
prompt = ChatPromptTemplate.from_template("Tell me about {topic}")
chain = prompt | llm

# Route to Project A
response_a = chain.invoke(
    {"topic": "machine learning"},
    config={"callbacks": [handler_a]}
)

# Route to Project B
response_b = chain.invoke(
    {"topic": "data science"},
    config={"callbacks": [handler_b]}
)

Important Considerations:

• Every Langfuse SDK execution across all integrations must include the appropriate public key parameter
• Missing public key parameters may result in traces being routed to the default project or lost
• Third-party OpenTelemetry spans that pass filtering may appear in all projects since they lack the Langfuse public key attribute

The JS/TS SDK supports two multi-project patterns:

• Fixed set of projects: register one LangfuseSpanProcessor per project. This works best when the projects are known at startup, for example one project per environment.
• Project per request: choose the project for each request at runtime, for example in a multi-tenant application where every customer brings their own Langfuse credentials.

Fixed set of projects

Register multiple LangfuseSpanProcessor instances, each with its own credentials.

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";

const sdk = new NodeSDK({
  spanProcessors: [
    new LangfuseSpanProcessor({
      publicKey: "pk-lf-public-key-project-1",
      secretKey: "sk-lf-secret-key-project-1",
    }),
    new LangfuseSpanProcessor({
      publicKey: "pk-lf-public-key-project-2",
      secretKey: "sk-lf-secret-key-project-2",
    }),
  ],
});

sdk.start();

This configuration sends every span accepted by each processor's filter to both projects. You can configure a custom shouldExportSpan filter for each processor to control which traces go to which project.

Each processor holds its own exporter and buffers for the lifetime of the process, so don't add a processor per tenant when you have many tenants or tenants are created at runtime. Use a project per request instead.

Project per request

To pick the Langfuse project at runtime, create a short-lived TracerProvider for each request and carry it in the OpenTelemetry context. A single routing provider, registered once with setLangfuseTracerProvider, forwards every Langfuse span to the provider in the active context:

• Each request gets its own LangfuseSpanProcessor with that project's credentials. When the request finishes, the provider is flushed and shut down, then garbage collected. Nothing accumulates per tenant.
• The per-request provider is never registered globally. Concurrent requests stay isolated because the context is scoped to the async call chain of each request.
• Spans started without a request context go to a provider without processors and are dropped instead of being sent to a default project.

import {
  context,
  createContextKey,
  type Context,
  type Tracer,
  type TracerProvider,
} from "@opentelemetry/api";
import {
  AlwaysOnSampler,
  BasicTracerProvider,
} from "@opentelemetry/sdk-trace-base";
import { setLangfuseTracerProvider } from "@langfuse/tracing";
import { LangfuseSpanProcessor } from "@langfuse/otel";

const PROJECT_PROVIDER = createContextKey("langfuse.project-tracer-provider");

// Spans started outside of `withLangfuseProject` are dropped
const NOOP_PROVIDER = new BasicTracerProvider();

class ProjectRoutingTracerProvider implements TracerProvider {
  getTracer(name: string, version?: string): Tracer {
    const resolve = (ctx: Context) =>
      (
        (ctx.getValue(PROJECT_PROVIDER) as TracerProvider | undefined) ??
        NOOP_PROVIDER
      ).getTracer(name, version);

    return {
      startSpan: (spanName, options, ctx = context.active()) =>
        resolve(ctx).startSpan(spanName, options, ctx),
      startActiveSpan: ((spanName: string, ...args: unknown[]) => {
        const ctx = args.length >= 3 ? (args[1] as Context) : context.active();
        const tracer = resolve(ctx);
        return (tracer.startActiveSpan as (...a: unknown[]) => unknown).call(
          tracer,
          spanName,
          ...args,
        );
      }) as Tracer["startActiveSpan"],
    };
  }
}

// Register once at startup. Do not register this provider globally.
export const projectRoutingTracerProvider = new ProjectRoutingTracerProvider();
setLangfuseTracerProvider(projectRoutingTracerProvider);

export async function withLangfuseProject<T>(
  credentials: { publicKey: string; secretKey: string; baseUrl: string },
  fn: () => Promise<T>,
): Promise<T> {
  const provider = new BasicTracerProvider({
    // Record spans even if an upstream sampler dropped the parent span
    sampler: new AlwaysOnSampler(),
    spanProcessors: [new LangfuseSpanProcessor(credentials)],
  });

  try {
    return await context.with(
      context.active().setValue(PROJECT_PROVIDER, provider),
      fn,
    );
  } finally {
    // Flushes all spans of this request, then releases the provider.
    // A failed export must not fail the request itself.
    await provider
      .shutdown()
      .catch((error) => console.error("Langfuse export failed", error));
  }
}

Wrap each request in withLangfuseProject. Tracing calls inside the callback, including startObservation, startActiveObservation, observe, and the OpenAI integration, are sent to that project. To get a deterministic trace ID per request, for example to link to the trace from your own system, pass a parent span context built with createTraceId:

import { createTraceId, startActiveObservation } from "@langfuse/tracing";
import { withLangfuseProject } from "./langfuse-projects";

export async function handleRequest(
  requestId: string,
  tenant: { publicKey: string; secretKey: string; baseUrl: string },
) {
  await withLangfuseProject(tenant, async () => {
    await startActiveObservation(
      "handle-request",
      async (span) => {
        span.update({ input: { message: "Hello" } });
        // ... your application logic
      },
      {
        // Optional: derive the trace ID from your own request ID
        parentSpanContext: {
          traceId: await createTraceId(requestId),
          spanId: "0123456789abcdef",
          traceFlags: 1,
        },
      },
    );
  });
}

withLangfuseProject waits for the request's spans to be exported before it returns, which adds the export round trip to every request. If your runtime keeps running after the response is sent, you can remove the await before provider.shutdown() so the export runs in the background.

Limitations

This pattern relies on the OpenTelemetry async context. NodeSDK and NodeTracerProvider.register() set up the required AsyncLocalStorage context manager. If your application uses neither, register one at startup, otherwise all spans are dropped:

import { context } from "@opentelemetry/api";
import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks";

context.setGlobalContextManager(new AsyncLocalStorageContextManager().enable());

Keep these limitations in mind:

• Spans started outside the request context are dropped. This includes callbacks from long-lived event emitters, sockets, or connection pools that were created outside the request, and spans created from a stored observation handle (span.startObservation(...)) outside the request callback. To keep a callback in the request's project, bind it while the request context is active: emitter.on("data", context.bind(context.active(), handler)).
• Spans that end after the request finishes are lost. provider.shutdown() only exports spans that have already ended. End all observations, including streaming generations and background work, before the withLangfuseProject callback returns.
• Third-party OpenTelemetry instrumentation is not routed. Libraries that use the global OpenTelemetry tracer provider don't pass through the routing provider, so their spans are not sent to the request's project. Integrations that accept a tracer option, such as the Vercel AI SDK, can be routed by passing a tracer from the routing provider, for example experimental_telemetry: { isEnabled: true, tracer: projectRoutingTracerProvider.getTracer("ai") }.
• Shared work queues can send spans to the wrong project. Code that runs on a queue shared across requests may execute in the context of whichever request is active when the queue drains. Spans started there are sent to that request's project. Make sure Langfuse spans are started from the request's own async call chain.
• LangChain requires LANGCHAIN_CALLBACKS_BACKGROUND=false. By default, LangChain JS runs callbacks on a shared background queue, so the LangChain CallbackHandler can record a run in another concurrent request's project. Set the LANGCHAIN_CALLBACKS_BACKGROUND=false environment variable so callbacks run inside the request, and await your chains inside the withLangfuseProject callback.
• Always pass all credentials. LangfuseSpanProcessor falls back to the LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, and LANGFUSE_BASE_URL environment variables for missing options, and baseUrl defaults to https://cloud.langfuse.com. Pass publicKey, secretKey, and baseUrl explicitly for every project to avoid sending data to the wrong project or region.
• Only tracing is routed. LangfuseClient from @langfuse/client, used for scores, prompts, and datasets, authenticates with its own constructor options or the LANGFUSE_* environment variables. To score a tenant's traces or fetch the tenant's prompts, create a client with the same credentials inside withLangfuseProject: new LangfuseClient({ publicKey, secretKey, baseUrl }).
• Each request opens its own export pipeline. Every request creates its own exporter and sends at least one export request when it finishes. Rate limits apply per Langfuse organization, so very high request volumes can hit the API limits.

You can use the completion_start_time attribute to manually set the time to first token (TTFT) of your LLM calls. This is useful for measuring the latency of your LLM calls and for identifying slow LLM calls.

from langfuse import get_client
import datetime, time

langfuse = get_client()

with langfuse.start_as_current_observation(as_type="generation", name="TTFT-Generation") as generation:
    time.sleep(3)
    generation.update(
        completion_start_time=datetime.datetime.now(),
        output="some response",
    )

langfuse.flush()

You can use the completionStartTime attribute to manually set the time to first token (TTFT) of your LLM calls. This is useful for measuring the latency of your LLM calls and for identifying slow LLM calls.

import { startActiveObservation } from "@langfuse/tracing";

startActiveObservation("llm-call", async (span) => {
  span.update({
    completionStartTime: new Date().toISOString(),
  });
});

Use the OpenTelemetry threading instrumentor so context flows across worker threads.

from opentelemetry.instrumentation.threading import ThreadingInstrumentor

ThreadingInstrumentor().instrument()

For multiprocessing, follow the OpenTelemetry guidance. If you use Pydantic Logfire, enable distributed_tracing=True. For tracing across separate services or processes, see Trace IDs & Distributed Tracing.