一、操作场景
这篇说说如何将 Langfuse SDK 生成的 Trace 数据接入 Agent 可观测。这篇提供以下接入方式：
·通过 Skill 快速接入与分析：如果你使用支持 Skill 的 AI 工具，可由 AI 自动创建或复用日志主题、识别项目语言并生成接入代码，同时完成上报验证与数据分析。
·手动配置接入：按步骤完成环境变量配置、代码接入和上报验证。
| 接入方式 | 数据路径 | 推荐场景 |
| OTLP/HTTP 标准直传 | 应用 > 标准 OTLP HTTP Exporter > Agent 可观测 | 手动接入推荐首选。适用于新项目、快速试用、希望减少自定义代码、希望按 OpenTelemetry 标准语义上报 Trace 的场景。 |
注意：说明：
注意：本方案不需要部署 Langfuse Server，也不需要将数据发送到 Langfuse Cloud。业务代码仍使用 Langfuse SDK 生成 Trace，数据出口改为 Agent 可观测。
二、前提条件
开始接入前，请确保已完成以下准备工作：
·已开通 日志服务 CLS。
·已准备具备 CLS 写入权限的访问凭证，例如 CAM 子账号、CAM Role 或临时密钥。云 API 密钥信息请前往 API 密钥管理 获取，密钥需对目标主题有写入权限，权限模板参见 采集 Agent 可观测数据，临时密钥请参见 使用临时密钥访问 CLS。
·应用使用 TypeScript/Node.js 或 Python，并已接入或计划接入 Langfuse SDK。
三、为临时凭证授予写入权限
为 STS 临时凭证设置策略时，至少需要允许向目标 Trace 日志主题写入日志。建议遵循最小权限原则，仅授权目标地域和目标日志主题的 cls:pushLog 操作。
{ "version": "2.0", "statement": [ { "effect": "allow", "action": [ "cls:pushLog" ], "resource": [ "qcs::cls:${region}:uin/${uin}:topic/${topicId}" ] } ]}
参数说明：
| 参数 | 说明 |
| ${region} | CLS Trace Topic 所在地域，例如 ap-guangzhou。 |
| ${uin} | 腾讯云账号 UIN。 |
| ${topicId} | 用于上报 Trace 的 CLS 日志主题 ID，即 CLS_TOPIC_ID。 |
注意：这篇上报所需权限为 cls:pushLog。仅授予其他日志上传动作时，可能无法向目标 Trace Topic 写入数据。
四、通过 Skill 快速接入与分析
如果你使用支持 Skill 的 AI 工具，可使用 腾讯云 Agent 可观测助手 自动完成接入与分析。该 Skill 已合并接入与分析能力，可自动创建或复用日志主题、生成标准 OTLP/HTTP Exporter 直传方案，并处理 Endpoint 拼接、Basic 鉴权、topic_id header、字段映射、时间单位、退出 flush 和验证说明，还支持基于上报数据的错误诊断、性能、Token 成本、用户会话等多维度分析。
使用前请准备以下信息：
| 参数 | 说明 |
| CLS_DEFAULT_REGION | Agent 应用所在地域，例如 ap-guangzhou。可参见 地域和访问域名。 |
| TENCENTCLOUD_SECRET_ID | 腾讯云访问凭证 SecretId。 |
| TENCENTCLOUD_SECRET_KEY | 腾讯云访问凭证 SecretKey。 |
在 AI 工具中输入以下提示词：
请使用腾讯云 Agent 可观测接入 Skill：https://skillhub.cn/skills/tencentcloud-cls-agent-obs帮我把当前项目接入腾讯云 Agent 可观测。接入方式：Langfuse地域：ap-guangzhou（替换为你的实际地域）
五、步骤1：配置环境变量
手动接入前，请先登录 日志服务控制台 > Agent 可观测，通过应用接入创建应用，并在日志主题列表中找到名称为 {应用名称}-trace-topic 的日志主题，复制其日志主题 ID（用于下方 CLS_TOPIC_ID）。
请在应用项目中创建或更新 .env 文件。请确保 .env 已加入 .gitignore，避免密钥泄露。
·云 API 长期密钥配置：
CLS_DEFAULT_REGION=ap-guangzhouCLS_TOPIC_ID=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxxTENCENTCLOUD_SECRET_ID=<SecretId>TENCENTCLOUD_SECRET_KEY=<SecretKey>SERVICE_NAME=my-llm-app# Python 额外需要。Langfuse Python SDK 启动时会校验密钥是否存在，此处可填写任意非空占位值。LANGFUSE_PUBLIC_KEY=pk-placeholderLANGFUSE_SECRET_KEY=sk-placeholder
·STS 临时密钥配置：
CLS_DEFAULT_REGION=ap-guangzhouCLS_TOPIC_ID=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxxTENCENTCLOUD_SECRET_ID=<TmpSecretId>TENCENTCLOUD_SECRET_KEY=<TmpSecretKey>TENCENTCLOUD_TOKEN=<Token>SERVICE_NAME=my-llm-app# Python 额外需要。Langfuse Python SDK 启动时会校验密钥是否存在，此处可填写任意非空占位值。LANGFUSE_PUBLIC_KEY=pk-placeholderLANGFUSE_SECRET_KEY=sk-placeholder
环境变量说明如下：
| 环境变量 | 云 API 长期密钥 | STS 临时密钥 | 说明 |
| TENCENTCLOUD_SECRET_ID | SecretId | TmpSecretId | 腾讯云访问标识。 |
| TENCENTCLOUD_SECRET_KEY | SecretKey | TmpSecretKey | 腾讯云访问密钥。 |
| TENCENTCLOUD_TOKEN | 不配置 | Token | STS 临时凭证 Token。 |
| CLS_DEFAULT_REGION | 必填 | 必填 | Trace Topic 所在地域。 |
| CLS_TOPIC_ID | 必填 | 必填 | Trace 日志主题 ID。 |
注意：说明：
注意：OTLP/HTTP 标准直传不需要单独配置 CLS_ENDPOINT，Endpoint 由 CLS_DEFAULT_REGION 自动拼接而成。
常见地域 Endpoint 如下：
| 地域 | 公网 Endpoint | 内网 Endpoint |
| 广州 | ap-guangzhou.cls.tencentcs.com | ap-guangzhou.cls.tencentyun.com |
| 上海 | ap-shanghai.cls.tencentcs.com | ap-shanghai.cls.tencentyun.com |
| 北京 | ap-beijing.cls.tencentcs.com | ap-beijing.cls.tencentyun.com |
| 新加坡 | ap-singapore.cls.tencentcs.com | ap-singapore.cls.tencentyun.com |
注意：说明：
注意：公网 Endpoint 适用于公网访问场景。内网 Endpoint 适用于同地域 VPC/CVM 环境，可降低网络延迟并避免公网流量。
六、步骤2：接入应用
手动接入使用标准 OTLP/HTTP Exporter 将 Trace 上报到 Agent 可观测，无需编写自定义导出代码。请根据项目语言选择 TypeScript/Node.js 或 Python 示例完成接入。
注意：说明：
注意：不同语言的 OpenTelemetry SDK 对 Endpoint 参数的要求不同。这篇 TypeScript/Node.js 和 Python 示例使用完整 OTLP Trace URL，即 https://{CLS_DEFAULT_REGION}.cls.tencentcs.com/v1/traces。
TypeScript/Node.jsPython1. 安装依赖。npm install @langfuse/tracing @langfuse/otel @langfuse/openai \\ @opentelemetry/sdk-node \\ @opentelemetry/exporter-trace-otlp-proto \\ dotenv2. 创建 instrumentation.ts。import "dotenv/config";import { NodeSDK } from "@opentelemetry/sdk-node";import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto";import { LangfuseSpanProcessor } from "@langfuse/otel";const secretId = process.env.TENCENTCLOUD_SECRET_ID!;const secretKey = process.env.TENCENTCLOUD_SECRET_KEY!;const token = process.env.TENCENTCLOUD_TOKEN;const region = process.env.CLS_DEFAULT_REGION!;const topicId = process.env.CLS_TOPIC_ID!;if (!secretId || !secretKey || !region || !topicId) { throw new Error("Missing required CLS environment variables");}// 长期密钥：SecretId:SecretKey// 临时密钥：TmpSecretId#Token:TmpSecretKeyconst authText = token ? `${secretId}#${token}:${secretKey}` : `${secretId}:${secretKey}`;const auth = Buffer.from(authText, "utf8").toString("base64");export const sdk = new NodeSDK({ spanProcessors: [ new LangfuseSpanProcessor({ exporter: new OTLPTraceExporter({ url: `https://${region}.cls.tencentcs.com/v1/traces`, headers: { Authorization: `Basic ${auth}`, topic_id: topicId, }, }), flushAt: 512, flushInterval: 5, mediaUploadEnabled: false, }), ],});sdk.start();process.on("SIGTERM", async () => { await sdk.shutdown();});3. 在业务入口文件顶部引入 instrumentation.ts。请确保该引入早于 OpenAI、LangChain 等会被 instrumentation patch 的模块。import { sdk } from "./instrumentation";import OpenAI from "openai";import { observeOpenAI } from "@langfuse/openai";import { startActiveObservation, propagateAttributes } from "@langfuse/tracing";const openai = observeOpenAI(new OpenAI());async function chat(userMessage: string) { return startActiveObservation("chat", async (span) => { span.update({ input: [{ role: "user", content: userMessage }] }); return propagateAttributes( { userId: "u-123", sessionId: "s-456", tags: ["model:gpt-4o"] }, async () => { const res = await openai.chat.completions.create({ model: "gpt-4o", messages: [{ role: "user", content: userMessage }], }); span.update({ output: res.choices[0].message }); return res; }, ); });}await chat("你好");await sdk.shutdown();1. 安装依赖。pip install langfuse \\ opentelemetry-sdk \\ opentelemetry-exporter-otlp-proto-http \\ python-dotenv \\ openai2. 创建 instrumentation.py。该文件必须早于任何 langfuse 模块导入。import base64import osfrom dotenv import load_dotenvfrom langfuse import Langfusefrom opentelemetry import tracefrom opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporterfrom opentelemetry.sdk.resources import Resourcefrom opentelemetry.sdk.trace import TracerProviderload_dotenv()secret_id = os.environ["TENCENTCLOUD_SECRET_ID"]secret_key = os.environ["TENCENTCLOUD_SECRET_KEY"]token = os.environ.get("TENCENTCLOUD_TOKEN")region = os.environ["CLS_DEFAULT_REGION"]topic_id = os.environ["CLS_TOPIC_ID"]# 长期密钥：SecretId:SecretKey# 临时密钥：TmpSecretId#Token:TmpSecretKeyauth_text = ( f"{secret_id}#{token}:{secret_key}" if token else f"{secret_id}:{secret_key}")auth = base64.b64encode(auth_text.encode("utf-8")).decode("ascii")provider = TracerProvider( resource=Resource.create({ "service.name": os.environ.get("SERVICE_NAME", "my-llm-app") }))trace.set_tracer_provider(provider)exporter = OTLPSpanExporter( endpoint=f"https://{region}.cls.tencentcs.com/v1/traces", headers={ "Authorization": f"Basic {auth}", "topic_id": topic_id, },)# 使用自定义 CLS exporter，避免 SDK 默认向 Langfuse 服务端创建额外 Trace 出口。langfuse = Langfuse( public_key=os.environ.get("LANGFUSE_PUBLIC_KEY", "pk-cls-placeholder"), secret_key=os.environ.get("LANGFUSE_SECRET_KEY", "sk-cls-placeholder"), tracer_provider=provider, span_exporter=exporter, flush_at=512, flush_interval=5,)3. 在业务入口中引入 instrumentation.py。import instrumentationfrom instrumentation import langfusefrom langfuse import observe, propagate_attributesfrom langfuse.openai import OpenAIopenai = OpenAI()@observe(name="chat", as_type="generation")def chat(user_message: str): with propagate_attributes( user_id="u-123", session_id="s-456", tags=["model:gpt-4o"], ): return openai.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": user_message}], )if __name__ == "__main__": chat("你好") # 短生命周期脚本必须在退出前 flush 并关闭 exporter。 langfuse.shutdown()
七、进阶配置
·采样配置：你可以在 TracerProvider 上配置采样策略，控制上报的 Trace 数据量。
| 采样器 | 说明 | 适用场景 |
| ALWAYS_ON | 全量采样 | 测试环境或低流量环境。 |
| ALWAYS_OFF | 全部不采样 | 临时关闭 Trace。 |
| TraceIdRatioBased(0.1) | 按比例采样 10% | 生产环境控制写入量。 |
| ParentBased(root) | 根据父 Span 决定是否采样 | 分布式链路，保证整条链路采样一致。 |
TypeScript/Node.jsPython1. 安装额外依赖。npm install @opentelemetry/sdk-trace-base2. instrumentation.ts 中配置采样器。import { ParentBasedSampler, TraceIdRatioBasedSampler,} from "@opentelemetry/sdk-trace-base";export const sdk = new NodeSDK({ sampler: new ParentBasedSampler({ root: new TraceIdRatioBasedSampler(0.1), }), // 其他配置保持不变});在 instrumentation.py 中配置采样器。from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBasedprovider = TracerProvider( sampler=ParentBased(TraceIdRatioBased(0.1)), resource=Resource.create({ "service.name": os.environ.get("SERVICE_NAME", "my-llm-app") }))
·跨地域上报：如果上报链路需要跨地域，可在 OTLP 请求 Header 中添加源地域标识。
TypeScript/Node.jsPythonnew OTLPTraceExporter({ url: `https://${region}.cls.tencentcs.com/v1/traces`, headers: { Authorization: `Basic ${auth}`, topic_id: topicId, "x-cross-region": "ap-guangzhou", // 源地域标识 },})OTLPSpanExporter( endpoint=f"https://{region}.cls.tencentcs.com/v1/traces", headers={ "Authorization": f"Basic {auth}", "topic_id": topic_id, "x-cross-region": "ap-guangzhou", # 源地域标识 },)
·关闭 HTTPS（仅测试环境）。
TypeScript/Node.jsPython可传入 http:// 协议的 URL 来使用明文 HTTP。url: `http://${region}.cls.tencentcs.com/v1/traces`,可使用 insecure=True 参数。 OTLPSpanExporter( endpoint=f"http://{region}.cls.tencentcs.com/v1/traces", insecure=True, # 仅测试环境 # …)
八、步骤3：验证上报结果
完成接入后，在应用中触发一次 LLM 调用，然后在 CLS 控制台中查询 Trace 数据。
1. 登录 日志服务控制台，在左侧导航栏中，选择检索分析。
2. 选择用于存储 Trace 数据的日志主题。
3. 在检索框中输入以下语句，查询最新上报的 Trace 数据。
* | SELECT traceID, spanID, name, duration, statusCode ORDER BY __TIMESTAMP__ DESC LIMIT 10
也可以查询指定调用链：
traceID:"4bf92f3577b34da6a3ce929d0e0e4736"
如需按模型统计平均耗时，请确保模型字段已写入 attribute 并已配置索引或使用 JSON 提取函数：
* | SELECT json_extract_scalar(attribute,'$."gen_ai.request.model"') AS model, AVG(duration) AS avg_duration_ns GROUP BY model
九、注意事项
·start、end、duration 字段需要保持纳秒单位。请勿将这些字段除以 1000 转成微秒，否则耗时会显示为实际值的千分之一。
·短生命周期脚本退出前需要执行 flush() 或 shutdown()，避免最后一批 Span 未完成上报。
·访问凭证请通过环境变量或密钥管理工具注入，不要硬编码到代码中。
十、相关文档
·选择接入方式
·Agent 可观测应用详情
·Agent Trace 字段定义说明