进阶功能

使用这些方法来强化你的 Litefuse instrumentation、保护敏感数据，并让 SDK 适配你的具体环境。

按 instrumentation scope 过滤

Litefuse 现在在两套 SDK 中都默认应用了 span 过滤器，无需额外配置即可让导出的 span 聚焦在 LLM 上。

默认情况下，只要满足以下任意一条，span 就会被导出：

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

如果你希望把另一个集成加入默认的 instrumentation scope 白名单，请在 langfuse/langfuse 提一个 issue，附上 scope 名和示例 span。

你可以在 Litefuse 中通过 metadata.scope.name 查看 span 的 instrumentation scope。被过滤掉的 span 不会出现在 UI 中。

要识别被过滤的 scope：

启用 debug 日志（Python：Langfuse(debug=True) 或 LANGFUSE_DEBUG="True"；JS/TS：LANGFUSE_DEBUG="true" 或 LANGFUSE_LOG_LEVEL="DEBUG"）。
运行你的应用，在日志中查找被丢弃的 span 信息和 instrumentation scope 名。
将这些 scope 通过组合 is_default_export_span / isDefaultExportSpan 加入到你的白名单逻辑中。
可选：临时使用 should_export_span=lambda span: True 或 shouldExportSpan: () => true 查看所有 span，再恢复过滤。

早期 SDK 版本默认导出所有未被屏蔽的 span。如需恢复该行为，请提供一个总是返回 true 的自定义过滤回调。

⚠️

过滤 span 可能破坏 trace 中的父子关系。例如，如果你过滤掉了父 span 但保留了它的子节点，可能会在 Litefuse UI 中看到 “孤立的” observation。请按上面的调试流程把被过滤掉的 span 重新加入白名单。

默认行为（推荐）：

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

导出全部：

from langfuse import Langfuse
 
langfuse = Langfuse(should_export_span=lambda span: True)

传入 should_export_span 会替换默认过滤器。如果想在保留默认行为的基础上扩展，请与 is_default_export_span 组合使用。

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

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 创建的 span：

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 表达拒绝规则。

兼容性示例（已弃用）：

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

更多关于在已有 OpenTelemetry 设置上使用 Litefuse 的内容可参见这里。

屏蔽敏感数据

如果你的 trace 数据（输入、输出、metadata）可能包含敏感信息（PII、密钥），可以在客户端初始化时提供一个 mask 函数。该函数会在数据发送到 Litefuse 前应用到所有相关数据上。

mask 函数应通过关键字参数接收数据，并返回 mask 后的数据。返回的数据必须可被 JSON 序列化。

from langfuse import Langfuse
import re
 
def pii_masker(data: any, **kwargs) -> any:
    if isinstance(data, str):
        return re.sub(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+", "[EMAIL_REDACTED]", data)
    elif isinstance(data, dict):
        return {k: pii_masker(data=v) for k, v in data.items()}
    elif isinstance(data, list):
        return [pii_masker(data=item) for item in data]
    return data
 
langfuse = Langfuse(mask=pii_masker)

日志与调试

Langfuse SDK 可以输出详细的日志和调试信息，帮助你排查应用问题。

通过环境变量：

可以通过 LANGFUSE_DEBUG 环境变量设置日志级别，启用 debug 模式。

export LANGFUSE_DEBUG="True"

在代码中：

Langfuse SDK 使用 Python 标准的 logging 模块。主 logger 名为 "langfuse"。要启用详细的 debug 日志，可以：

在初始化 Langfuse 客户端时设置 debug=True。
手动配置 "langfuse" logger：

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

langfuse logger 默认日志级别为 logging.WARNING。

采样

采样允许你只把一部分 trace 发送到 Litefuse。这对于降低高流量应用的成本和噪音非常有用。

在代码中：

可以在客户端初始化时通过 sample_rate 参数配置 SDK 进行采样。值应是 0.0（采样 0% 的 trace）到 1.0（采样 100% 的 trace）之间的浮点数。

如果某条 trace 没有被采样，则它的所有 observation（span、generation）和关联的分数都不会发送到 Litefuse。

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

通过环境变量：

也可以通过 LANGFUSE_SAMPLE_RATE 环境变量设置采样率。

export LANGFUSE_SAMPLE_RATE="0.2"

隔离的 TracerProvider

你可以为 Litefuse 配置一个独立的 OpenTelemetry TracerProvider。这能在 Litefuse 的 tracing 与其他可观测性系统之间形成隔离。

隔离带来的好处：

Litefuse 的 span 不会被发送到你的其他可观测性后端（例如 Datadog、Jaeger、Zipkin）
第三方库的 span 不会被发送到 Litefuse
配置和采样率相互独立

⚠️

虽然 TracerProvider 之间相互隔离，但它们共用同一个 OpenTelemetry context 来跟踪活跃 span。这可能导致 span 关系问题：

来自一个 TracerProvider 的父 span 可能拥有来自另一个 TracerProvider 的子节点
一些 span 如果其父 span 属于另一个 TracerProvider，就可能显得 “孤立”
trace 层级可能不完整或令人困惑

请在做 instrumentation 时仔细规划，避免出现令人困惑的 trace 结构。

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

更多关于在已有 OpenTelemetry 设置上使用 Litefuse 的内容可参见这里。

多项目设置

⚠️

多项目设置在 Python SDK 中是实验性功能，与第三方 OpenTelemetry 集成相关有重要限制。

Langfuse Python SDK 支持在同一个应用中使用多个 public key 将 trace 路由到不同项目。其原理是 Langfuse SDK 会向它产生的所有 span 添加一个包含 public key 的特定 span 属性。

工作原理：

Span 属性：Langfuse SDK 会向其创建的 span 添加一个包含 public key 的特定属性
多个 processor：在全局 tracer provider 上注册多个 span processor，每个都绑定到特定 public key 的 exporter
过滤：每个 span processor 内部会根据 public key 属性的存在与值过滤 span

与第三方库相关的重要限制：

自动发送 OpenTelemetry span 的第三方库（例如 HTTP 客户端、数据库、其他 instrumentation 库）没有 Litefuse 的 public key span 属性。结果是：

通过导出过滤器但没有 public key 的第三方 span 无法路由到特定项目
这些 span 会被所有 span processor 处理，可能被发送到所有项目
在默认过滤器下，这主要影响来自第三方 instrumentation 的 GenAI/LLM span（基础设施 span 通常会被过滤掉）

为何是实验性的？ 该方案要求所有集成中的 Langfuse SDK 调用都传入 public_key 参数，才能保证正确路由；并且通过过滤的第三方 span 可能出现在所有项目中。

初始化

要设置多项目，请为每个项目分别初始化一个 Litefuse 客户端：

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://litefuse.cloud"
)
 
project_b_client = Langfuse(
    public_key="pk-lf-project-b-...",
    secret_key="sk-lf-project-b-...",
    base_url="https://litefuse.cloud"
)

集成的使用方式

在多项目设置中，所有集成都必须指定 public_key 参数，以确保 trace 路由到正确的项目。

Observe 装饰器：

将 langfuse_public_key 作为关键字参数传给 最外层 被 observe 的函数（不是装饰器）。从 Python SDK >= 3.2.2 起，嵌套的被装饰函数会自动从所处的执行上下文中获取 public key。同样，get_client 调用也会感知当前装饰函数执行上下文中的 langfuse_public_key，因此无需在这里再传一次。

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 调用上：

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 构造函数：

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 调用都必须包含相应的 public key 参数
缺失 public key 参数可能导致 trace 被路由到默认项目或丢失
通过过滤的第三方 OpenTelemetry span 可能出现在所有项目中，因为它们没有 Litefuse public key 属性

Time to first token (TTFT)

你可以手动设置 LLM 调用的 time to first token（TTFT）。这对衡量 LLM 调用延迟、识别慢速 LLM 调用非常有用。

可以使用 completion_start_time 属性手动设置 LLM 调用的 TTFT。这对衡量 LLM 调用延迟、识别慢速 LLM 调用非常有用。

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

自签名 SSL 证书（自托管 Litefuse）

如果你自托管 Litefuse 并希望使用自签名 SSL 证书，需要配置 SDK 信任该自签名证书：

⚠️

修改 SSL 设置可能对你的环境带来重大安全影响。请在操作前充分理解这些影响。

1. 设置 OpenTelemetry span exporter 信任自签名证书

.env

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

2. 设置 HTTPX 信任该证书，用于发往 Litefuse 实例的所有其他 API 请求

main.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 和 Litefuse，由于两者都用 OpenTelemetry 做 tracing，你需要配置自定义的 OpenTelemetry 设置。本指南介绍了如何同时把错误监控数据送到 Sentry，以及把 LLM 可观测性 trace 捕获到 Litefuse。

线程池与多进程

使用 OpenTelemetry threading instrumentor 让 context 在 worker 线程间传播。

from opentelemetry.instrumentation.threading import ThreadingInstrumentor
 
ThreadingInstrumentor().instrument()

对于多进程，请遵循 OpenTelemetry 的指引。如果你使用 Pydantic Logfire，请启用 distributed_tracing=True。如需跨独立服务或进程进行 tracing，请参见 Trace ID 与分布式 tracing。

插桩故障排查与 FAQ

这个页面对你有帮助吗？

支持