Mastra 接入 Langfuse 可观测性指南导出 LLM 追踪、Prompt 关联与评分【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra本篇技术指南介绍如何在 Mastra 框架中通过mastra/langfuse将 Agent、Workflow、工具调用与模型生成的全量追踪数据导出到 Langfuse实现开源 LLM 可观测性。你将掌握LangfuseExporter的完整配置方式密钥、Endpoint、实时/批量导出、环境与版本标签、withLangfusePrompt的 Prompt 关联用法以及 Mastra 追踪属性到 Langfuse 语义字段的底层映射机制可直接照搬用于生产环境的接入与排障。一、包定位与整体架构mastra/langfuse是 Mastra 官方提供的 Langfuse 可观测性 Provider。从 包描述 可以看到它基于 Langfuse v5 官方 SDKlangfuse/otel与langfuse/client实现完整功能支持其中langfuse/otel提供LangfuseSpanProcessor负责把 OpenTelemetry Span 批量/实时写入 Langfuselangfuse/client提供非追踪类能力如评分scoring、Prompt 管理与评估。包的对外出口在 index.ts只做两件事从./tracing导出追踪相关实现从./helpers导出构建 Langfuse 兼容追踪选项的辅助函数。核心类LangfuseExporter继承自mastra/observability的BaseExporter见 tracing.ts因此它可以无缝挂载到 Mastra 的Observability注册表中与其他 Exporter 共存。从架构上看数据链路为Mastra 运行 Agent/Workflow 时产生 Span 事件Observability实例将SPAN_ENDED事件路由到已注册的 ExporterLangfuseExporter用SpanConverter来自mastra/otel-exporter格式为GenAI_v1_38_0把 Mastra Span 转成 OTel Span再经mapMastraToLangfuseAttributes把mastra.*属性映射为 Langfuse 可读的langfuse.*字段最后交给LangfuseSpanProcessor写入 Langfuse。二、安装与最小接入安装命令与官方 README 一致见 README.mdnpm install mastra/langfuse接入前必须准备 Langfuse 的凭据。按 README 的要求在创建 Exporter 之前设置两个环境变量export LANGFUSE_PUBLIC_KEYpk-lf-xxxx export LANGFUSE_SECRET_KEYsk-lf-xxxx然后在 Mastra 实例中注册LangfuseExporterimport { Mastra } from mastra/core/mastra; import { Observability } from mastra/observability; import { LangfuseExporter } from mastra/langfuse; export const mastra new Mastra({ observability: new Observability({ configs: { langfuse: { serviceName: my-service, exporters: [new LangfuseExporter()], }, }, }), });这段代码里configs.langfuse.serviceName会被传给Observability的配置校验Observability构造时会用 Zod schema 校验整个注册配置见 default.ts最终由LangfuseExporter.init()透传给SpanConverter作为写入 Langfuse 的service.name。三、LangfuseExporter 配置项全解LangfuseExporterConfig定义于 tracing.ts在BaseExporterConfig之上提供了以下选项配置项类型默认值说明publicKeystring环境变量LANGFUSE_PUBLIC_KEYLangfuse 公钥secretKeystring环境变量LANGFUSE_SECRET_KEYLangfuse 密钥baseUrlstring环境变量LANGFUSE_BASE_URL否则https://cloud.langfuse.comLangfuse 服务地址自托管/私有部署时必填additionalHeadersRecordstring, string无附加请求头用于代理鉴权等场景realtimebooleanfalse开启实时模式每个事件后立即 flushflushAtnumberSDK 默认每个 OTel 导出批次的最大 Span 数flushIntervalnumberSDK 默认待导出 Span 的最大等待秒数environmentstring环境变量LANGFUSE_TRACING_ENVIRONMENT写入 trace 的 Langfuse 环境标签如 production/stagingreleasestring环境变量LANGFUSE_RELEASE写入 trace 的 Langfuse 发布版本标签3.1 凭据解析与缺失处理构造器中的解析顺序为「配置对象优先、环境变量兜底」见 tracing.tsconst publicKey config.publicKey ?? process.env.LANGFUSE_PUBLIC_KEY; const secretKey config.secretKey ?? process.env.LANGFUSE_SECRET_KEY; const baseUrl stripTrailingSlashes(config.baseUrl ?? process.env.LANGFUSE_BASE_URL ?? LANGFUSE_DEFAULT_BASE_URL);当公钥或密钥任一缺失时Exporter 会调用setDisabled(...)进入禁用状态并输出明确日志标明密钥是来自 config、来自 env 还是缺失此时不会创建 SpanProcessor 与 Client后续事件全部丢弃。这一行为被 tracing.test.ts 中的disables when publicKey is missing等测试用例覆盖。另外注意baseUrl会经过stripTrailingSlashes处理以遍历方式逐字符去除末尾的/实现刻意避免了可能被攻击者利用的正则回溯见 tracing.ts 底部所以传入https://my-langfuse.example.com///也会被规整为规范地址。3.2 导出模式批量 vs 实时realtime控制LangfuseSpanProcessor的exportModerealtime: false默认→exportMode: batched按flushAt/flushInterval批量导出吞吐更高realtime: true→exportMode: immediate每个 Span 结束后立即导出调试时可见性最好。对应测试uses immediate export mode when realtime is true验证了参数传递。此外代码里还显式设置了shouldExportSpan: () true这是因为 Langfuse SpanProcessor 的默认过滤器只放行带gen_ai.*属性的 Span而 Mastra 的 Span 使用mastra.*命名空间必须全量放行源码注释对此有明确说明。3.3 生命周期flush 与 shutdownasync flush(): Promisevoid { await Promise.all([this.#processor?.forceFlush(), this.#client?.flush()]); } async shutdown(): Promisevoid { await Promise.all([this.#processor?.shutdown(), this.#client?.shutdown()]); }flush()同时冲刷 SpanProcessor 与 Clientshutdown()则同时优雅关闭两者保证进程退出前数据不丢失见 tracing.ts。四、mastra.* 属性到 langfuse.* 的映射原理这是本包最有价值的部分SpanConverter生成的 OTel Span 属性以mastra.*为前缀而 Langfuse 的 OTLP 端点只读取langfuse.*命名空间。mapMastraToLangfuseAttributes在导出前就地in-place完成映射见 tracing.ts。映射规则可归纳如下4.1 保留字段映射可筛选的顶级字段Mastra 属性Langfuse 字段用途mastra.metadata.userIduser.id关联用户mastra.metadata.sessionId或mastra.metadata.threadIdsession.id关联会话/线程mastra.metadata.traceNamelangfuse.trace.name自定义 trace 名mastra.metadata.versionlangfuse.trace.versiontrace 版本mastra.tagslangfuse.trace.tags标签JSON 序列化mastra.completion_start_timelangfuse.observation.completion_start_time首 Token 时间TTFTmastra.span.typelangfuse.observation.metadata.spanTypeSpan 类型gen_ai.agent.id/gen_ai.agent.namelangfuse.observation.metadata.agentId/agentNameAgent 身份gen_ai.operation.namelangfuse.observation.metadata.operationName操作名mastra.*.input/mastra.*.outputlangfuse.observation.input/output非 gen_ai Span 的输入输出对于gen_ai类 Span输入输出保持gen_ai.input.messages/gen_ai.output.messages原生读取路径只有不存在这些字段时才回退到mastra.*前缀匹配详见 tracing.ts 中Input/Output映射段。4.2 根 Span 身份与 Trace 级元数据当span.isRootSpan为真时映射逻辑会额外做四件事Trace 级输入输出把根 Span 的input/output镜像到langfuse.trace.input/langfuse.trace.output。源码注释说明没有这一步Langfuse trace 顶层的输入输出为空会破坏映射到 Trace input/output 的 LLM-as-a-judge 评估器。实体身份AGENT_RUN根 Span 写入langfuse.trace.name取entityName ?? entityId以及langfuse.trace.metadata.agentId/agentNameWORKFLOW_RUN同理写入workflowId/workflowName。这样每个 Langfuse trace 都被限定到发起它的 Agent/Workflow便于按 trace 名或元数据过滤来限定 Langfuse 评估器的作用域。用户 traceName 优先如果用户通过mastra.metadata.traceName显式设置了名字则保留用户值不被实体名覆盖测试preserves user-provided traceName over the agent default验证。剩余元数据前转其余mastra.metadata.*键如runId、resourceId、用户自定义键转发到langfuse.trace.metadata.*使它们成为可筛选的顶级 trace 元数据非字符串值用 JSON 序列化Langfuse 摄入时会还原类型。注意只处理根 Span——因为 Langfuse 会从任意 Span 应用langfuse.trace.*子 Span 可能覆盖 trace 级信息所以刻意只在根 Span 上做。有DEDICATED_METADATA_KEYSuserId、sessionId、threadId、traceName、version、langfuse用于排除已映射到专用字段的键避免重复。4.3 容错设计所有序列化与解析都是「尽力而为」的serializeTraceIo对无法 JSON 序列化的值循环引用、bigint返回undefined并跳过该属性而不是让导出失败mastra.metadata.langfuse解析失败非法 JSON会被静默忽略。对应的测试omits trace input/output that cannot be serialized instead of failing the export验证了这一容错路径。五、Prompt 关联withLangfusePromptmastra/langfuse还提供了withLangfusePrompt辅助函数见 helpers.ts用于启用 Langfuse Prompt TracingPrompt 关联。它配合mastra/observability的buildTracingOptions使用import { buildTracingOptions } from mastra/observability; import { withLangfusePrompt } from mastra/langfuse; import { Agent } from mastra/core/agent; import { openai } from ai-sdk/openai; const agent new Agent({ name: support-agent, instructions: You are a helpful assistant, model: openai(gpt-4o), defaultGenerateOptions: { tracingOptions: buildTracingOptions( withLangfusePrompt({ name: customer-support, version: 1 }), ), }, });withLangfusePrompt接收一个LangfusePromptInput把name与version合并进metadata.langfuse.promptexport function withLangfusePrompt(prompt: LangfusePromptInput): TracingOptionsUpdater { return opts ({ ...opts, metadata: { ...opts.metadata, langfuse: { ...(opts.metadata?.langfuse as Recordstring, unknown), prompt: { ...(prompt.name ! undefined { name: prompt.name }), ... }, }, }, }); }关键细节Langfuse v5 只支持按 name version 关联。接口里的id字段已被标记deprecated注释明确说明 v5 会忽略该字段因此生产环境请始终提供name和version也可以直接传入 Langfuse SDK 的 prompt 对象例如langfuse.getPrompt()的返回值它只会提取其中的name/version/id字段其余字段如prompt文本、config、labels不会混入 tracing 元数据多个 updater 可以组合buildTracingOptions(withLangfusePrompt(...), withUserId(user-123))会深合并 metadata互不覆盖见 helpers.test.ts 的should compose with other updaters用例。当生成 Span 携带metadata.langfuse.prompt时导出器会在映射阶段把它转成langfuse.observation.prompt.name与langfuse.observation.prompt.version测试maps prompt metadata to langfuse.observation.prompt.* attributes验证Langfuse 控制台即可看到该 generation 关联到了具体 Prompt 版本。六、自定义 Trace 元数据与 Prompt 链接mastra.metadata.langfuse是留给用户的特殊命名空间支持两类键见 tracing.ts 映射逻辑保留键prompt用于 Prompt 链接{ prompt: { name, version } }会被映射为langfuse.observation.prompt.*任意自定义键其余键会被转发为langfuse.trace.metadata.key作为 trace 顶级可筛选元数据Langfuse 只允许按顶级元数据过滤/分组 trace。字符串直接透传数字、布尔、对象以 JSON 序列化后由 Langfuse 摄入时还原类型。// 通过 tracingOptions 注入 buildTracingOptions( withLangfusePrompt({ name: customer-support, version: 2 }), opts ({ ...opts, metadata: { ...opts.metadata, customerId: abc, tier: enterprise, }, }), );优先级规则均有测试覆盖根 Span 的实体身份键agentId/agentName/workflowId/workflowName优先于用户自定义的metadata.langfuse.*冲突键显式的metadata.langfuse.*值优先于根 Span 普通元数据中同名键值为null/undefined的键不会转发。七、评估与评分onScoreEvent 与 addScoreToTraceLangfuseExporter支持把 Mastra 的评分结果写入 Langfuse用于评估链路闭环新路径onScoreEvent推荐Observability的评分事件流水线mastra.observability.addScore产生ScoreEvent后Exporter 会调用LangfuseClient.score.create把scoreId作为评分 ID、scorerName ?? scorerId作为评分名、score作为数值、reason作为注释、metadata原样透传并附加dataType: NUMERIC若配置了environment含LANGFUSE_TRACING_ENVIRONMENT兜底评分也会带上该环境标签。traceId缺失时直接跳过。旧路径addScoreToTrace已弃用为向后兼容保留转发到同一个submitScore底层调用评分 ID 由traceId-spanId-scorerName拼装而成。源码注释建议迁移到新的mastra.observability.addScore评分事件流水线。此外Exporter 通过clientgetter 暴露LangfuseClient实例可进一步使用 Prompt 管理、数据集datasets等高级 API。八、版本与更新包的版本历史与发布说明见 observability/langfuse/CHANGELOG.md。依赖约束方面package.json声明了peerDependenciesmastra/core 1.16.0-0 2.0.0-0、opentelemetry/api ^1.9.0、opentelemetry/sdk-trace-base ^2.0.1并指定 Node.js22.13.0。接入前请确认项目满足这些版本要求。九、接入自检清单LANGFUSE_PUBLIC_KEY与LANGFUSE_SECRET_KEY均已设置或通过new LangfuseExporter({ publicKey, secretKey })显式传入——否则 Exporter 会静默禁用自托管 Langfuse 时设置LANGFUSE_BASE_URL或baseUrl确保无尾随/需要即时可见时开启realtime: true生产环境建议保持批量模式并调整flushAt/flushInterval需要在 Langfuse 中按环境/版本区分时设置environment/release或对应环境变量Prompt 关联请使用withLangfusePrompt({ name, version })勿依赖已弃用的id进程退出前调用exporter.flush()/shutdown()或交给 Mastra 生命周期管理避免批量缓冲区数据丢失。至此你已经可以从「安装接入」到「属性映射原理」再到「评分闭环」完整掌握mastra/langfuse的用法足以在生产环境独立完成 Langfuse 观测接入与问题排查。【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考