通过 Langfuse SDK 接入

最近更新时间:2025-12-05 15:26:01

我的收藏
Langfuse 是一个开源的 LLM 工程化平台,专为监控、调试和优化 LLM 应用而设计。它提供全面的追踪监控、提示词版本管理、成本分析、质量评估和数据集管理等核心功能,支持 LangChain、OpenAI SDK 等主流框架的无缝集成。通过直观的仪表板和详细的分析报告,Langfuse 帮助开发团队深入了解应用性能,控制运营成本,持续改进模型输出质量。
LLM 可观测已兼容 Langfuse 探针 协议,如果您之前使用 Langfuse 作为可观测平台,可以参考本文无缝切换到腾讯云 LLM 可观测。

前提条件

该方案支持 Langfuse SDK 3.2.6 及以上版本。

操作步骤

步骤1:配置数据发送目标

在 LLM 应用中,一般通过如下配置将监控数据发送到 Langfuse 平台:
# Langfuse authentication
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxx"
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxx"
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"


from langfuse import Langfuse

langfuse_client = Langfuse()

# ----- 业务代码开始 -----

# ----- 业务代码结束 -----

# 保证退出前数据发送完
langfuse_client.flush()
在此基础上,只需添加如下代码,就能实现将 Langfuse 探针埋点产生的链路数据发送到腾讯云 LLM 可观测。
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource

# ----- Langfuse 平台相关配置(可移除) -----
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxx"
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxx"
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"

# ----- LLM 可观测相关配置 -----
resource = Resource.create({
"service.name": "<serviceName>", # 应用名,可自行配置
"token": "<token>", # 接入 token,LLM 可观测控制台获取
"host.name": "hostName", # 主机名,可自行配置
})
tracer_provider = TracerProvider(resource=resource)
tracer_provider.add_span_processor(
BatchSpanProcessor(
OTLPSpanExporter(
endpoint="<endpoint>", # 接入点,LLM 可观测获取
)
)
)
trace.set_tracer_provider(tracer_provider)

from langfuse import Langfuse

langfuse_client = Langfuse(
# 指定 Trace Provider
tracer_provider=tracer_provider,
# 阻止 Langfuse 埋点产生的链路数据发送到 Langfuse 平台。
# 若不设置此参数,链路数据将同时发送到 Langfuse 平台和和腾讯云 LLM 可观测。
blocked_instrumentation_scopes=["langfuse-sdk"]
)


# ----- 业务代码开始 -----


# ----- 业务代码结束 -----

# 保证退出前数据发送完
langfuse_client.flush()
对应的字段说明如下:
<serviceName> :应用名,多个使用相同 serviceName 接入的应用进程,会表现为相同应用下的多个实例。应用名最长63个字符,只能包含小写字母、数字及分隔符“ - ”,且必须以小写字母开头,以数字或小写字母结尾。
<token> :前置步骤中拿到业务系统 Token。
<hostName>:该实例的主机名,是应用实例的唯一标识,通常情况下可以设置为应用实例的 IP 地址。
<endpoint> :前置步骤中拿到的接入点。
至此已完成将监控数据发送到腾讯云 LLM 可观测的配置。

步骤2:在链路埋点中标记 LLM 入口(可选)

LLM 可观测引入了 LLM 入口概念,代表一条链路中开启 LLM 调用的环节。如果您的分布式调用链路在 Langfuse SDK 链路埋点之前,链路的上游使用了其他 OpenTelemetry 方案(例如腾讯云 OpenTelemetry-Python 探针),则需要在链路埋点标记 LLM 入口,否则将影响 LLM 可观测的数据展示。



标记方式

根据开启 LLM 调用环节的实际情况,您可以基于 OpenTelemetry SDK 或 Langfuse SDK 标记 LLM 入口,标记方式为:在当前 Span 添加 Span 属性,将gen_ai.is_entry设置为True

通过 OpenTelemetry SDK 进行标记

通常标记于 LLM 调用的发起环节,例如通过 OpenAI 访问大模型。
@observe(name="run_demo")
def run_demo() -> None:
question = "什么是 OpenTelemetry?"
print(f"User: {question}")

# 在此处添加入口标记
trace.get_current_span().set_attribute("gen_ai.is_entry", True) # gen_ai.is_entry: Bool(true)

answer = chat_completion_stream(question)
print("\\n\\nAssistant (final):\\n" + answer)

通过 Langfuse SDK 添加标记

以 LangChain 和 LangGraph 为例,通常标记在组件的执行入口(invoke)之前。
with langfuse.start_as_current_span(name="langgraph-agent-demo") as span:
span.update(metadata={"gen_ai.is_entry": True}) # 在此处添加,只会在入口 span 处标记
# 注意不能通过以下方式添加在 metadata 中,这样会使所有的后继 span 都继承这个属性,让入口标记失去意义
result = agent_executor.invoke(
{"input": question},
config={
"callbacks": [callback_handler],
"metadata": {
"langfuse_user_id": "user-456",
}
}
)