Langfuse 文档 中文 英文原文 ↗
文档 / 埋点(Instrumentation)

埋点(Instrumentation)

用 Langfuse SDK 为你的应用埋点主要有两种方式:

所有方法都是可互操作的。你可以将装饰器创建的 observation 嵌套在上下文管理器内,或将手动 spans 与我们的原生集成混合使用。

自定义埋点

使用以下方法用 Langfuse SDK 为你的应用埋点:

上下文管理器

上下文管理器让你能够创建一个新的 span,并在其持续时间内将其设置为 OTel 上下文中当前活动的 observation。在此块内创建的所有新 observations 将自动成为其子节点。

start_as_current_observation() 是在确保活动 OpenTelemetry 上下文被更新的同时创建 observations 的主要方式。在 with 块内创建的任何子 observations 都会自动继承父节点。

通过设置 as_type 参数,observations 可以有不同的类型。

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

startActiveObservation 接受一个回调,使新 span 在回调作用域内活动,并自动结束它,即使跨异步边界。

通过设置 asType 参数,observations 可以有不同的类型。

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

Observe 包装器

observe 装饰器是一种简单的方式,无需修改函数的内部逻辑即可自动捕获被包装函数的输入、输出、计时和错误。

使用 observe() 装饰一个函数,自动捕获输入、输出、计时和错误。

通过设置 as_type 参数,observations 可以有不同的类型。

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

捕获大型输入/输出可能会增加开销。按装饰器禁用 IO 捕获(capture_input=False、capture_output=False),或通过 LANGFUSE_OBSERVE_DECORATOR_IO_CAPTURE_ENABLED 环境变量禁用。

使用 observe() 包装一个函数,自动捕获输入、输出、计时和错误。

通过设置 asType 参数,observations 可以有不同的类型。

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

捕获大型输入/输出可能会增加开销。按装饰器禁用 IO 捕获(captureInput=False、captureOutput=False),或通过 LANGFUSE_OBSERVE_DECORATOR_IO_CAPTURE_ENABLED 环境变量禁用。

手动 observations

你也可以手动创建 observations。当你需要以下情况时这很有用:

当你需要在不更改活动上下文的情况下手动控制时,使用 start_observation()。

你可以传递 as_type 参数以指定要创建的 observation 类型。

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

如果你使用 start_observation(),你有责任在返回的 observation 对象上调用 .end()。不这样做将导致 Langfuse 中出现不完整或缺失的 observations。它们与 with 语句一起使用的 start_as_current_... 对应方法会自动处理这一点。

关键特征:

带更复杂嵌套的示例:

python
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')

startObservation 让你完全控制创建 observations。

你可以传递 asType 参数以指定要创建的 observation 类型。

当你调用这些函数之一时,新 observation 会自动链接为 OpenTelemetry 上下文中当前活动操作的子节点。但是,它不会使这个新 observation 成为活动的。这意味着你追踪的任何进一步操作仍将链接到_原始_父节点,而非你刚创建的那个。

要手动创建嵌套 observations,请使用返回对象上的方法(例如 parentSpan.startObservation(...))。

typescript
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();

如果你使用 startObservation(),你有责任在返回的 observation 对象上调用 .end()。不这样做将导致 Langfuse 中出现不完整或缺失的 observations。

嵌套 observations

Langfuse SDK 方法自动处理 observations 的嵌套。

Observe 装饰器

如果你使用 observe 包装器,函数调用层级会被自动捕获并反映在 trace 中。

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

上下文管理器

如果你使用上下文管理器,嵌套由 OpenTelemetry 的上下文传播自动处理。当你使用 start_as_current_observation() 创建新 observation 时,它成为创建时上下文中活动 observation 的子节点。

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

手动 Observations

如果你手动创建 observations,你可以使用父 LangfuseSpan 或 LangfuseGeneration 对象上的方法创建子节点。这些子节点_不会_成为当前上下文,除非使用它们的 _as_current_ 变体(参见上下文管理器)。

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

嵌套通过 OpenTelemetry 上下文传播自动发生。当你使用 startActiveObservation 创建新 observation 时,它成为当时活动对象的子节点。

ts
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" });
  });
});

更新 observations

你可以在代码执行时用新信息更新 observations。

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

使用 observation.update() 更新活动的 observation。

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

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

向 observations 添加属性

你可以向 observations 添加属性,以帮助你更好地理解应用并在 Langfuse 中关联 observations:

要更新 trace 的输入和输出,参见 trace 级输入/输出。

使用 propagate_attributes() 向 observations 添加属性。 在 Python SDK 中,environment 是一等 Langfuse 环境,映射到 langfuse.environment,而非 trace 元数据。当环境是请求作用域时使用它,例如当一个共享代理处理来自多个部署环境的请求时。

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

使用 @observe() 装饰器时:

python
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

使用 propagateAttributes() 向 observations 添加属性。

ts
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();
    }
  );
});

跨服务传播

对于跨多个服务的分布式追踪,使用 as_baggage 参数(更多细节参见 OpenTelemetry 文档)通过 HTTP header 传播属性。

在 Python SDK 中,environment 也可以这样传播。环境作为 langfuse_environment baggage 发送,并优先于下游服务的本地 LANGFUSE_TRACING_ENVIRONMENT 或客户端级环境,适用于在传播上下文内创建的 spans。

python
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",
        environment="staging",
        as_baggage=True,
    ):
        requests.get("https://service-b.example.com/api")
ts
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");
    }
  );
});

安全警告:当启用 baggage 传播时,属性会被添加到所有出站 HTTP header。仅将其用于分布式追踪所需的非敏感值。

更新 trace

默认情况下,trace 的 input/output 镜像你在根 observation(trace 中的第一个 observation)上设置的内容。如果需要,你可以自定义 trace 级信息,用于 LLM-as-a-Judge、AB 测试或 UI 清晰度。

Langfuse 中的 LLM-as-a-Judge 工作流可能依赖 trace 级输入/输出。如果你的评估负载不同,请确保刻意设置它们,而非依赖根 observation。

默认行为

python
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

覆盖默认行为

如果你需要与根 observation 不同的 trace 输入/输出,使用 observation.set_trace_io() 或 langfuse.set_current_trace_io():

set_trace_io() 和 set_current_trace_io() 已弃用,仅为与依赖 trace 输入/输出的 trace 级 LLM-as-a-judge 评估器向后兼容而存在。对于新代码,请直接在根 observation 上设置输入/输出。参见 Python v3 → v4 迁移指南。

python
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

使用 propagateAttributes 设置关联的 trace 属性,使用 setTraceIO 设置 trace 级输入/输出。

.setTraceIO() 已弃用,仅为与 trace 级 LLM-as-a-judge 评估器向后兼容而存在。参见 JS/TS v4 → v5 迁移指南。

ts
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();
  }
);

Trace 和 observation ID

Langfuse 遵循 W3C Trace Context 标准:

你不能设置任意 observation ID,但可以生成确定性的 trace ID 以与外部系统关联。

关于跨服务关联 traces 的更多信息,参见 Trace ID 与分布式追踪。

使用 create_trace_id() 生成 trace ID。如果提供了 seed,则 ID 是确定性的。使用相同的 seed 会得到相同的 ID。这对于将外部 ID 与 Langfuse traces 关联很有用。

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

使用 get_current_trace_id() 获取当前 trace ID,使用 get_current_observation_id 获取当前 observation ID。

你也可以使用 observation.trace_id 和 observation.id 直接从 LangfuseSpan 或 LangfuseGeneration 对象访问 trace 和 observation ID。

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

使用 createTraceId 从 seed 生成确定性的 trace ID。

ts
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,
    },
  }
);

使用 getActiveTraceId 获取活动的 trace ID,使用 getActiveSpanId 获取当前 observation ID。

ts
import { startObservation, getActiveTraceId } from "@langfuse/tracing";

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

链接到现有 traces

当与已有 trace ID 的上游服务集成时,提供 W3C trace 上下文,以便 Langfuse spans 加入现有树,而非创建新树。

使用 trace_context 参数设置自定义 trace 上下文信息。

python
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

使用 parentSpanContext 参数设置自定义 trace 上下文信息。

ts
import { startObservation } from "@langfuse/tracing";

const span = startObservation(
  "downstream-task",
  {},
  {
    parentSpanContext: {
      traceId: "abcdef1234567890abcdef1234567890",
      spanId: "fedcba0987654321",
      traceFlags: 1,
    },
  }
);

span.end();

客户端生命周期与刷新

由于 Langfuse SDK 是异步的,它们在后台缓冲 spans。在短生命周期进程(脚本、serverless 函数、worker)中始终 flush() 或 shutdown() 客户端,以避免丢失数据。

flush()

手动触发将所有缓冲的 observations(spans、generations、scores、媒体元数据)发送到 Langfuse API。这在短生命周期脚本中或退出应用之前很有用,以确保所有数据被持久化。

python
from langfuse import get_client

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

flush() 方法会阻塞,直到排队的数据被相应的后台线程处理。

shutdown()

优雅地关闭 Langfuse 客户端。这包括:

  1. 刷新所有缓冲的数据(类似于 flush())。
  2. 等待后台线程(用于数据接入和媒体上传)完成当前任务并终止。

在应用退出前调用 shutdown() 至关重要,以防止数据丢失并确保干净的资源释放。SDK 会自动注册一个 atexit 钩子,在正常程序终止时调用 shutdown(),但在以下场景中建议手动调用:

python
from langfuse import get_client

langfuse = get_client()
# ... application logic ...

# Before exiting:
langfuse.shutdown()

通用 Serverless 函数

从你的 OTEL SDK 设置文件导出 processor,以便稍后调用 forceFlush()。

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

// Export the processor to be able to flush it
export const langfuseSpanProcessor = new LangfuseSpanProcessor({
  exportMode: "immediate" // optional: configure immediate span export in serverless environments
});

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

sdk.start();

在你的 serverless 函数 handler 中,在函数退出前对 LangfuseSpanProcessor 调用 forceFlush()。

tshandler.ts
import { langfuseSpanProcessor } from "./instrumentation";

export async function handler(event, context) {
  // ... your application logic ...

  // Flush before exiting
  await langfuseSpanProcessor.forceFlush();
}

Vercel Cloud Functions

从你的 instrumentation.ts 文件导出 processor,以便稍后刷新它。

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

// Export the processor to be able to flush it
export const langfuseSpanProcessor = new LangfuseSpanProcessor();

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

sdk.start();

在 Vercel Cloud Functions 中,请使用 after 工具在请求完成后安排一次刷新。

tsroute.ts
import { after } from "next/server";

import { langfuseSpanProcessor } from "./instrumentation.ts";

export async function POST() {
  // ... existing request logic ...

  // Schedule flush after request has completed
  after(async () => {
    await langfuseSpanProcessor.forceFlush();
  });

  // ... send response ...
}
非官方中文翻译 · 图片/视频/代码均链接官方资源 · 版权归 Langfuse GmbH 所有 查看英文原文 ↗