文档 / 高级功能

高级功能

使用这些方法来加固你的 Langfuse 埋点、保护敏感数据,并使 SDK 适应你的特定环境。

按埋点作用域过滤

Langfuse 现在在两个 SDK 中都应用默认 span 过滤器,无需额外配置即可保持导出聚焦于 LLM。

默认情况下,如果满足以下任一条件,则导出 span:

它由 Langfuse SDK 创建(instrumentation_scope.name == "langfuse-sdk")
它至少有一个 gen_ai.* 属性
它来自已知的 LLM 埋点作用域(例如 openinference.*、langsmith、haystack、litellm、agent_framework、strands-agents、vllm、opentelemetry.instrumentation.anthropic)

如果你希望将另一个集成添加到默认埋点作用域允许列表,请在 langfuse/langfuse 中开一个 issue,附上作用域名称和示例 span。

你可以在 Langfuse 的 metadata.scope.name 下检查 span 的埋点作用域。被过滤掉的 spans 不会出现在 UI 中。

要识别被过滤的作用域:

启用调试日志(Python 上 Langfuse(debug=True) 或 LANGFUSE_DEBUG="True",JS/TS 上 LANGFUSE_DEBUG="true" 或 LANGFUSE_LOG_LEVEL="DEBUG")。
运行你的应用,检查日志中的 dropped-span 消息和埋点作用域名称。
通过与 is_default_export_span / isDefaultExportSpan 组合,将这些作用域添加到你的允许列表逻辑中。
可选:临时使用 should_export_span=lambda span: True 或 shouldExportSpan: () => true 检查所有 spans,然后恢复过滤。

早期 SDK 版本默认导出所有未被阻止的 spans。要恢复该行为,请提供一个始终为 true 的自定义过滤回调。

过滤 spans 可能会破坏 traces 中的父子关系。例如,如果你过滤掉父 span 但保留其子节点,你可能会在 Langfuse UI 中看到"孤立"的 observations。请使用上面的调试流程重新添加被过滤掉的 spans。

默认行为(推荐):

python

from langfuse import Langfuse

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

导出所有内容:

python

from langfuse import Langfuse

langfuse = Langfuse(should_export_span=lambda span: True)

传递 should_export_span 会替换默认过滤器。要保留默认行为并扩展它,请与 is_default_export_span 组合。

将自定义逻辑与内置谓词组合:

python

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

仅导出由 Langfuse SDK 创建的 spans:

python

from langfuse import Langfuse
from langfuse.span_filter import is_langfuse_span

langfuse = Langfuse(should_export_span=is_langfuse_span)

可用的 Python 辅助函数:is_default_export_span、is_langfuse_span、is_genai_span、is_known_llm_instrumentor、KNOWN_LLM_INSTRUMENTATION_SCOPE_PREFIXES。

blocked_instrumentation_scopes 仍可用于向后兼容,但已弃用并计划在未来版本中移除。优先在 should_export_span 中表达拒绝规则。

已弃用的兼容性示例:

python

from langfuse import Langfuse

langfuse = Langfuse(
    should_export_span=lambda span: True,
    blocked_instrumentation_scopes=["sqlalchemy", "psycopg"],
)

默认行为(推荐):

tsinstrumentation.ts

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

自定义过滤:

tsinstrumentation.ts

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

传递 shouldExportSpan 会替换默认过滤器,因此如果你想扩展而非替换默认行为,请在回调中包含默认条件。

与默认过滤器组合:

tsinstrumentation.ts

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

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

来自 @langfuse/otel 的可用 JS/TS 辅助函数:isDefaultExportSpan、isLangfuseSpan、isGenAISpan、isKnownLLMInstrumentor、KNOWN_LLM_INSTRUMENTATION_SCOPE_PREFIXES。

导出所有内容:

tsinstrumentation.ts

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

你可以在这里阅读更多关于将 Langfuse 与现有 OpenTelemetry 设置一起使用的内容。

遮蔽敏感数据

如果你的追踪数据可能包含敏感信息(如 PII 或 secrets),请在将 spans 发送到 Langfuse 之前配置遮蔽钩子。对于 Python SDK 应用,优先使用 mask_otel_spans,因为它在导出阶段对原始 OpenTelemetry span 属性运行,包括由第三方埋点创建的 spans。

使用 mask_otel_spans 在导出前为应更改的 OpenTelemetry spans 返回稀疏补丁。

python

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 是同步的,通常运行在 OpenTelemetry 批量 span processor 工作线程上;在 flush() 和关闭期间它可能在调用线程上运行。保持函数快速,以避免使导出队列积压。完整行为、错误处理和旧版 mask 对比参见遮蔽。

你可以向 LangfuseSpanProcessor 提供一个 mask 函数。该函数将应用于每个 observation 的输入、输出和元数据。

该函数接收一个对象 ``,其中 data 是属性值的字符串化 JSON。它应返回遮蔽后的数据。

tsinstrumentation.ts

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

日志记录与调试

Langfuse SDK 可以暴露详细的日志记录和调试信息,帮助你排查应用问题。

通过环境变量:

你可以使用 LANGFUSE_DEBUG 环境变量设置日志级别以启用调试模式。

bash

export LANGFUSE_DEBUG="True"

在代码中:

Langfuse SDK 使用 Python 的标准 logging 模块。主日志记录器名为 "langfuse"。要启用详细的调试日志记录,你可以:

在初始化 Langfuse 客户端时设置 debug=True 参数。
手动配置 "langfuse" 日志记录器:

python

import logging

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

langfuse 日志记录器的默认日志级别是 logging.WARNING。

你可以配置全局 SDK 日志记录器以控制日志输出的详细程度。这对于调试很有用。

通过环境变量:

你可以使用 LANGFUSE_LOG_LEVEL 环境变量设置日志级别以启用调试模式。

bash

export LANGFUSE_LOG_LEVEL="DEBUG"

在代码中:

typescript

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

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

可用的日志级别是 DEBUG、INFO、WARN 和 ERROR。

采样

采样让你只将 traces 的一个子集发送到 Langfuse。这对于减少高流量应用中的成本和噪声很有用。

在代码中:

你可以通过在客户端初始化期间设置 sample_rate 参数来配置 SDK 对 traces 采样。该值应为 0.0(采样 0% 的 traces)和 1.0(采样 100% 的 traces)之间的浮点数。

如果一个 trace 未被采样,它的任何 observations(spans、generations)或关联的 scores 都不会发送到 Langfuse。

python

from langfuse import Langfuse

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

通过环境变量:

你也可以使用 LANGFUSE_SAMPLE_RATE 环境变量设置采样率。

bash

export LANGFUSE_SAMPLE_RATE="0.2"

在代码中:

Langfuse 遵循 OpenTelemetry 的采样决策。在你的 OTEL NodeSDK 上配置采样器,以控制哪些 traces 到达 Langfuse,并减少高流量工作负载中的噪声/成本。

tsinstrumentation.ts

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

通过环境变量:

你也可以使用 LANGFUSE_SAMPLE_RATE 环境变量设置采样率。

bash

export LANGFUSE_SAMPLE_RATE="0.2"

隔离的 TracerProvider

你可以为 Langfuse 配置一个单独的 OpenTelemetry TracerProvider。这在 Langfuse 追踪和你的其他可观测性系统之间创建隔离。

隔离的好处:

Langfuse spans 不会被发送到你的其他可观测性后端(例如 Datadog、Jaeger、Zipkin)
第三方库 spans 不会被发送到 Langfuse
独立的配置和采样率

虽然 TracerProviders 是隔离的,但它们共享相同的 OpenTelemetry 上下文以追踪活动 spans。这可能导致 span 关系问题,其中:

来自一个 TracerProvider 的父 span 可能拥有来自另一个 TracerProvider 的子节点

如果某些 spans 的父 span 属于不同的 TracerProvider,它们可能显得"孤立"

Trace 层级可能不完整或令人困惑

仔细规划你的埋点,以避免令人困惑的 trace 结构。

python

from opentelemetry.sdk.trace import TracerProvider
from langfuse import Langfuse

langfuse_tracer_provider = TracerProvider() # do not set to global tracer provider to keep isolation
langfuse = Langfuse(tracer_provider=langfuse_tracer_provider)
langfuse.start_observation(name="myspan").end() # Span will be isolated from remaining OTEL instrumentation

用自定义 provider 隔离 Langfuse spans,避免将它们发送到其他导出器。

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)

你可以在这里阅读更多关于将 Langfuse 与现有 OpenTelemetry 设置一起使用的内容。

多项目设置

多项目设置在 Python SDK 中是实验性的,并且对第三方 OpenTelemetry 集成有重要限制。

Langfuse Python SDK 支持通过使用多个公钥在同一应用内将 traces 路由到不同项目。这之所以有效,是因为 Langfuse SDK 向其生成的所有 spans 添加一个包含公钥的特定 span 属性。

工作原理:

Span 属性:Langfuse SDK 向其创建的 spans 添加一个包含公钥的特定 span 属性
多个 Processor:多个 span processor 注册到全局 tracer provider,每个都有其绑定到特定公钥的相应导出器
过滤:在每个 span processor 内,基于公钥属性的存在和值过滤 spans

对第三方库的重要限制:

自动发出 OpenTelemetry spans 的第三方库(例如 HTTP 客户端、数据库、其他埋点库)没有 Langfuse 公钥 span 属性。因此:

通过导出过滤且没有公钥的第三方 spans 无法路由到特定项目
这些 spans 由所有 span processor 处理,并可能发送到所有项目
使用默认过滤器,这主要影响来自第三方 instrumentors 的 GenAI/LLM spans(基础设施 spans 通常被过滤掉)

为什么这是实验性的? 此方法要求将 public_key 参数传递给所有集成中的所有 Langfuse SDK 执行,以确保正确路由,并且通过过滤的第三方 spans 可能出现在所有项目中。

初始化

要设置多个项目,为每个项目初始化单独的 Langfuse 客户端:

python

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

集成使用

对于多项目设置中的所有集成,你必须指定 public_key 参数,以确保 traces 路由到正确的项目。

Observe 装饰器:

将 langfuse_public_key 作为关键字参数传递给_最顶层_的被观察函数(而非装饰器)。从 Python SDK >= 3.2.2 起,嵌套的装饰函数会自动从它们当前所在的执行上下文中获取公钥。此外,对 get_client 的调用也会感知装饰函数执行上下文中当前的 langfuse_public_key,因此在此处再次传递 langfuse_public_key 不是必需的。

python

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 集成:

将 langfuse_public_key 作为关键字参数添加到 OpenAI 执行:

python

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 集成:

将 public_key 添加到 CallbackHandler 构造函数:

python

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]}
)

重要考虑:

所有集成中的每次 Langfuse SDK 执行都必须包含适当的公钥参数
缺少公钥参数可能导致 traces 被路由到默认项目或丢失
通过过滤的第三方 OpenTelemetry spans 可能出现在所有项目中,因为它们缺少 Langfuse 公钥属性

你可以配置 SDK 将 traces 发送到多个 Langfuse 项目。这对于多租户应用或将 traces 发送到不同环境很有用。只需注册多个 LangfuseSpanProcessor 实例,每个都有自己的凭证。

tsinstrumentation.ts

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

此配置将每个 processor 过滤器接受的每个 span 发送到两个项目。你可以为每个 processor 配置自定义 shouldExportSpan 过滤器,以控制哪些 traces 去往哪个项目。

首 token 时间(TTFT)

你可以手动设置 LLM 调用的首 token 时间(TTFT)。这对于衡量 LLM 调用的延迟和识别缓慢的 LLM 调用很有用。

你可以使用 completion_start_time 属性手动设置 LLM 调用的首 token 时间(TTFT)。这对于衡量 LLM 调用的延迟和识别缓慢的 LLM 调用很有用。

python

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

你可以使用 completionStartTime 属性手动设置 LLM 调用的首 token 时间(TTFT)。这对于衡量 LLM 调用的延迟和识别缓慢的 LLM 调用很有用。

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

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

自签名 SSL 证书(自托管 Langfuse)

如果你自托管 Langfuse 并想使用自签名 SSL 证书,你需要配置 SDK 以信任自签名证书:

更改 SSL 设置会根据你的环境产生重大安全影响。在继续之前,请确保你理解这些影响。

1. 设置 OpenTelemetry span 导出器以信任自签名证书

bash.env

OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE="/path/to/my-selfsigned-cert.crt"

2. 设置 HTTPX 以信任证书,用于所有其他到 Langfuse 实例的 API 请求

pythonmain.py

import os

import httpx

from langfuse import Langfuse

httpx_client = httpx.Client(verify=os.environ["OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE"])

langfuse = Langfuse(httpx_client=httpx_client)

与 Sentry 一起设置

如果你的应用中同时使用 Sentry 和 Langfuse,你需要配置自定义的 OpenTelemetry 设置,因为两个工具都使用 OpenTelemetry 进行追踪。本指南展示了如何在将错误监控数据发送到 Sentry 的同时在 Langfuse 中捕获 LLM 可观测性 traces。

线程池和多进程

使用 OpenTelemetry 线程埋点器,以便上下文跨工作线程流动。

python

from opentelemetry.instrumentation.threading import ThreadingInstrumentor

ThreadingInstrumentor().instrument()

对于多进程,请遵循 OpenTelemetry 指南。如果你使用 Pydantic Logfire,请启用 distributed_tracing=True。关于跨单独服务或进程的追踪,参见 Trace ID 与分布式追踪。