Instrumentation

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": "..."})

import { startActiveObservation, startObservation } from "@langfuse/tracing";
 
await startActiveObservation("user-request", async (span) => {
  span.update({ input: { query: "Capital of France?" } });
 
  const generation = startObservation(
    "llm-call",
    { model: "gpt-4", input: [{ role: "user", content: "Capital of France?" }] },
    { asType: "generation" }
  );
  generation.update({ output: { content: "Paris." } }).end();
 
  span.update({ output: "Answered." });
});

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"

import { observe, updateActiveObservation } from "@langfuse/tracing";
 
async function fetchData(source: string) {
  updateActiveObservation({ metadata: { source: "API" } });
  return { data: `some data from ${source}` };
}
 
const tracedFetchData = observe(fetchData, {
  name: "fetch-data",
  asType: "span",
});
 
const result = await tracedFetchData("API");

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

from langfuse import get_client
 
langfuse = get_client()
 
# This outer span establishes an active context.
with langfuse.start_as_current_observation(as_type="span", name="main-operation") as main_operation_span:
    # 'main_operation_span' is the current active context.
 
    # 1. Create a "manual" span using langfuse.start_observation().
    #    - It becomes a child of 'main_operation_span'.
    #    - Crucially, 'main_operation_span' REMAINS the active context.
    #    - 'manual_side_task' does NOT become the active context.
    manual_side_task = langfuse.start_observation(name="manual-side-task")
    manual_side_task.update(input="Data for side task")
 
    # 2. Start another operation that DOES become the active context.
    #    This will be a child of 'main_operation_span', NOT 'manual_side_task',
    #    because 'manual_side_task' did not alter the active context.
    with langfuse.start_as_current_observation(as_type="span", name="core-step-within-main") as core_step_span:
        # 'core_step_span' is now the active context.
        # 'manual_side_task' is still open but not active in the global context.
        core_step_span.update(input="Data for core step")
        # ... perform core step logic ...
        core_step_span.update(output="Core step finished")
    # 'core_step_span' ends. 'main_operation_span' is the active context again.
 
    # 3. Complete and end the manual side task.
    # This could happen at any point after its creation, even after 'core_step_span'.
    manual_side_task.update(output="Side task completed")
    manual_side_task.end() # Manual end is crucial for 'manual_side_task'
 
    main_operation_span.update(output="Main operation finished")
# 'main_operation_span' ends automatically here.
 
# Expected trace structure in Langfuse:
# - main-operation
#   |- manual-side-task
#   |- core-step-within-main
#     (Note: 'core-step-within-main' is a sibling to 'manual-side-task', both children of 'main-operation')

import { startObservation } from "@langfuse/tracing";
 
// Start a root span for a user request
const span = startObservation(
  // name
  "user-request",
  // params
  {
    input: { query: "What is the capital of France?" },
  }
);
 
// Create a nested span for, e.g., a tool call
const toolCall = span.startObservation(
  // name
  "fetch-weather",
  // params
  {
    input: { city: "Paris" },
  },
  // Specify observation type in asType
  // This will type the attributes argument accordingly
  // Default is 'span'
  { asType: "tool" }
);
 
// Simulate work and end the tool call span
await new Promise((resolve) => setTimeout(resolve, 100));
toolCall.update({ output: { temperature: "15°C" } }).end();
 
// Create a nested generation for the LLM call
const generation = span.startObservation(
  "llm-call",
  {
    model: "gpt-4",
    input: [{ role: "user", content: "What is the capital of France?" }],
  },
  { asType: "generation" }
);
 
generation.update({
  usageDetails: { input: 10, output: 5 },
  output: { content: "The capital of France is Paris." },
});
 
generation.end();
 
// End the root span
span.update({ output: "Successfully answered user request." }).end();

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

from langfuse import get_client
 
langfuse = get_client()
 
with langfuse.start_as_current_observation(as_type="span", name="outer-process") as outer_span:
    # outer_span is active
 
    with langfuse.start_as_current_observation(as_type="generation", name="llm-step-1") as gen1:
        # gen1 is active, child of outer_span
        gen1.update(output="LLM 1 output")
 
    with outer_span.start_as_current_observation(name="intermediate-step") as mid_span:
        # mid_span is active, also a child of outer_span
        # This demonstrates using the yielded span object to create children
 
        with mid_span.start_as_current_observation(as_type="generation", name="llm-step-2") as gen2:
            # gen2 is active, child of mid_span
            gen2.update(output="LLM 2 output")
 
        mid_span.update(output="Intermediate processing done")
 
    outer_span.update(output="Outer process finished")

from langfuse import get_client
 
langfuse = get_client()
 
parent = langfuse.start_observation(name="manual-parent")
 
child_span = parent.start_observation(name="manual-child-span")
# ... work ...
child_span.end()
 
child_gen = parent.start_observation(name="manual-child-generation", as_type="generation")
# ... work ...
child_gen.end()
 
parent.end()

import { startActiveObservation } from "@langfuse/tracing";
 
await startActiveObservation("outer-process", async () => {
  await startActiveObservation("llm-step-1", async (span) => {
    span.update({ output: "LLM 1 output" });
  });
 
  await startActiveObservation("intermediate-step", async (span) => {
    await startActiveObservation("llm-step-2", async (child) => {
      child.update({ output: "LLM 2 output" });
    });
 
    span.update({ output: "Intermediate processing done" });
  });
});

from langfuse import get_client
 
langfuse = get_client()
 
with langfuse.start_as_current_observation(as_type="generation", name="llm-call", model="gpt-5-mini") as gen:
    gen.update(input={"prompt": "Why is the sky blue?"})
    
    # ... make LLM call ...
    response_text = "Rayleigh scattering..."
    
    gen.update(
        output=response_text,
        usage_details={"input_tokens": 5, "output_tokens": 50},
        metadata={"confidence": 0.9}
    )
 
# Alternatively, update the current observation in context:
with langfuse.start_as_current_observation(as_type="span", name="data-processing"):
    # ... some processing ...
    langfuse.update_current_span(metadata={"step1_complete": True})
    # ... more processing ...
    langfuse.update_current_span(output={"result": "final_data"})

import { startActiveObservation } from "@langfuse/tracing";
 
await startActiveObservation("user-request", async (span) => {
  span.update({
    input: { path: "/api/process" },
    output: { status: "success" },
  });
});

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",
        trace_name="user-workflow",
    ):
        with langfuse.start_as_current_observation(as_type="generation", name="llm-call"):
            pass

from langfuse import observe, propagate_attributes
 
@observe()
def my_llm_pipeline(user_id: str, session_id: str):
    # Propagate early in the trace
    with propagate_attributes(
        user_id=user_id,
        session_id=session_id,
        metadata={"pipeline": "main"}
    ):
        # All nested @observe functions inherit these attributes
        result = call_llm()
        return result
 
@observe()
def call_llm():
    # This automatically has user_id, session_id, metadata from parent
    pass

import { startActiveObservation, propagateAttributes, startObservation } from "@langfuse/tracing";
 
await startActiveObservation("user-workflow", async () => {
  await propagateAttributes(
    {
      userId: "user_123",
      sessionId: "session_abc",
      metadata: { experiment: "variant_a", env: "prod" },
      version: "1.0",
      traceName: "user-workflow",
    },
    async () => {
      const generation = startObservation("llm-call", { model: "gpt-4" }, { asType: "generation" });
      generation.end();
    }
  );
});

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")

import { propagateAttributes, startActiveObservation } from "@langfuse/tracing";
 
await startActiveObservation("api-request", async () => {
  await propagateAttributes(
    {
      userId: "user_123",
      sessionId: "session_abc",
      asBaggage: true,
    },
    async () => {
      await fetch("https://service-b.example.com/api");
    }
  );
});

from langfuse import get_client
 
langfuse = get_client()
 
# Using the context manager
with langfuse.start_as_current_observation(
    as_type="span",
    name="user-request",
    input={"query": "What is the capital of France?"}  # This becomes the trace input
) as root_span:
 
    with langfuse.start_as_current_observation(
        as_type="generation",
        name="llm-call",
        model="gpt-4o",
        input={"messages": [{"role": "user", "content": "What is the capital of France?"}]}
    ) as gen:
        response = "Paris is the capital of France."
        gen.update(output=response)
        # LLM generation input/output are separate from trace input/output
 
    root_span.update(output={"answer": "Paris"})  # This becomes the trace output

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 has its own input/output
    root_span.update(input="Step 1 data", output="Step 1 result")
 
    # But trace should have different input/output (e.g., for LLM-as-a-judge)
    root_span.set_trace_io(
        input={"original_query": "User's actual question"},
        output={"final_answer": "Complete response", "confidence": 0.95}
    )
 
    # Now trace input/output are independent of root span input/output
 
# Using the observe decorator
@observe()
def process_user_query(user_question: str):
    # LLM processing...
    answer = call_llm(user_question)
 
    # Explicitly set trace input/output for evaluation features
    langfuse.set_current_trace_io(
        input={"question": user_question},
        output={"answer": answer}
    )
 
    return answer

import { propagateAttributes, startObservation } from "@langfuse/tracing";
 
const userId = "user-123";
const sessionId = "session-abc";
 
propagateAttributes(
  {
    userId: userId,
    sessionId: sessionId,
    tags: ["authenticated-user"],
    metadata: { plan: "premium" },
  },
  () => {
    const rootSpan = startObservation("data-processing");
 
    const generation = rootSpan.startObservation(
      "llm-call",
      {},
      { asType: "generation" }
    );
 
    generation.end();
 
    rootSpan.end();
  }
);

from langfuse import get_client, Langfuse
langfuse = get_client()
 
external_request_id = "req_12345"
deterministic_trace_id = langfuse.create_trace_id(seed=external_request_id)

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)

import { createTraceId, startObservation } from "@langfuse/tracing";
 
const externalId = "support-ticket-54321";
 
const langfuseTraceId = await createTraceId(externalId);
 
const rootSpan = startObservation(
  "process-ticket",
  {},
  {
    parentSpanContext: {
      traceId: langfuseTraceId,
      spanId: "0123456789abcdef",
      traceFlags: 1,
    },
  }
);

import { startObservation, getActiveTraceId } from "@langfuse/tracing";
 
await startObservation("run", async (span) => {
  const traceId = getActiveTraceId();
  console.log(`Current trace ID: ${traceId}`);
});

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

import { startObservation } from "@langfuse/tracing";
 
const span = startObservation(
  "downstream-task",
  {},
  {
    parentSpanContext: {
      traceId: "abcdef1234567890abcdef1234567890",
      spanId: "fedcba0987654321",
      traceFlags: 1,
    },
  }
);
 
span.end();

from langfuse import get_client
 
langfuse = get_client()
# ... create traces and observations ...
langfuse.flush() # Ensures all pending data is sent

from langfuse import get_client
 
langfuse = get_client()
# ... application logic ...
 
# Before exiting:
langfuse.shutdown()