Langfuse 是一个开源的 LLM 工程化平台,专为监控、调试和优化 LLM 应用而设计。它提供全面的追踪监控、提示词版本管理、成本分析、质量评估和数据集管理等核心功能,支持 LangChain、OpenAI SDK 等主流框架的无缝集成。通过直观的仪表板和详细的分析报告,Langfuse 帮助开发团队深入了解应用性能,控制运营成本,持续改进模型输出质量。
前提条件
该方案支持 Langfuse SDK 3.2.6 及以上版本。
操作步骤
步骤1:配置数据发送目标
在 LLM 应用中,一般通过如下配置将监控数据发送到 Langfuse 平台:
# Langfuse authenticationos.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 Langfuselangfuse_client = Langfuse()# ----- 业务代码开始 -----# ----- 业务代码结束 -----# 保证退出前数据发送完langfuse_client.flush()
在此基础上,只需添加如下代码,就能实现将 Langfuse 探针埋点产生的链路数据发送到腾讯云 LLM 可观测。
from opentelemetry import tracefrom opentelemetry.sdk.trace import TracerProviderfrom opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporterfrom opentelemetry.sdk.trace.export import BatchSpanProcessorfrom 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 Langfuselangfuse_client = Langfuse(# 指定 Trace Providertracer_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",}})