如果你之前使用 Langfuse 作为可观测平台，可以参考这篇无缝切换到腾讯云 AI 可观测。
一、前提条件
该方案支持 Langfuse SDK 3.2.6 及以上版本。
二、配置数据发送目标
在 AI Agent 应用中，一般通过如下配置将监控数据发送到 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 探针埋点产生的链路数据发送到腾讯云 AI 可观测。
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"# —– AI 可观测相关配置 —–resource = Resource.create({ "service.name": "<serviceName>", # 应用名，可自行配置 "token": "<token>", # 接入 token，AI 可观测控制台获取 "host.name": "hostName", # 主机名，可自行配置})tracer_provider = TracerProvider(resource=resource)tracer_provider.add_span_processor( BatchSpanProcessor( OTLPSpanExporter( endpoint="<endpoint>", # 接入点，AI 可观测控制台获取 ) ))trace.set_tracer_provider(tracer_provider)from langfuse import Langfuselangfuse_client = Langfuse( # 指定 Trace Provider tracer_provider=tracer_provider, # 阻止 Langfuse 埋点产生的链路数据发送到 Langfuse 平台。 # 若不设置此参数，链路数据将同时发送到 Langfuse 平台和腾讯云 AI 可观测。 blocked_instrumentation_scopes=["langfuse-sdk"])# —– 业务代码开始 —–# —– 业务代码结束 —–# 保证退出前数据发送完 langfuse_client.flush()
对应的字段说明如下：
·<serviceName> ：应用名，多个使用相同应用名接入的应用进程，会表现为相同应用下的多个实例。应用名最长63个字符，只能包含小写字母、数字及分隔符“ – ”，且必须以小写字母开头，以数字或小写字母结尾。
·<token> ：参考 获取接入点与 Token，获取业务系统 Token。
·<hostName>：该实例的主机名，是应用实例的唯一标识，通常情况下可以设置为应用实例的 IP 地址。
·<endpoint> ：参考 获取接入点与 Token，获取接入点。
三、AI 入口标记
AI 入口代表一条链路中开启 AI 相关逻辑的环节。如果你的分布式调用链路在 Langfuse SDK 链路埋点之前，链路的上游使用了其他 OpenTelemetry 方案（例如腾讯云 OpenTelemetry-Go 探针），则需要在链路埋点标记 AI 入口，方便基于 AI 入口实现链路数据查询。
四、标记方式
你可以基于 OpenTelemetry SDK 或 Langfuse SDK 标记 AI 入口，标记方式为：在当前 Span 添加 Span 属性，将 gen_ai.is_entry 设置为 True。
五、通过 OpenTelemetry SDK 进行标记
通常标记于 AI 相关逻辑的发起环节，例如通过 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", } } )