企业官网建设流程全解析

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 官方 SDK（@langfuse/otel与@langfuse/client）实现完整功能支持，其中：

• @langfuse/otel提供LangfuseSpanProcessor，负责把 OpenTelemetry Span 批量/实时写入 Langfuse；
• @langfuse/client提供非追踪类能力，如评分（scoring）、Prompt 管理与评估。

包的对外出口在 index.ts，只做两件事：从./tracing导出追踪相关实现，从./helpers导出构建 Langfuse 兼容追踪选项的辅助函数。核心类LangfuseExporter继承自@mastra/observability的BaseExporter（见 tracing.ts），因此它可以无缝挂载到 Mastra 的Observability注册表中，与其他 Exporter 共存。

从架构上看，数据链路为：

1. Mastra 运行 Agent/Workflow 时产生 Span 事件；
2. Observability实例将SPAN_ENDED事件路由到已注册的 Exporter；
3. LangfuseExporter用SpanConverter（来自@mastra/otel-exporter，格式为GenAI_v1_38_0）把 Mastra Span 转成 OTel Span；
4. 再经mapMastraToLangfuseAttributes把mastra.*属性映射为 Langfuse 可读的langfuse.*字段，最后交给LangfuseSpanProcessor写入 Langfuse。

二、安装与最小接入

安装命令与官方 README 一致（见 README.md）：

npm install @mastra/langfuse

接入前必须准备 Langfuse 的凭据。按 README 的要求，在创建 Exporter 之前设置两个环境变量：

export LANGFUSE_PUBLIC_KEY=pk-lf-xxxx export LANGFUSE_SECRET_KEY=sk-lf-xxxx

然后在 Mastra 实例中注册LangfuseExporter：

import { 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之上提供了以下选项：

3.1 凭据解析与缺失处理

构造器中的解析顺序为「配置对象优先、环境变量兜底」（见 tracing.ts）：

const 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的exportMode：

• realtime: 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 与 shutdown

async flush(): Promise<void> { await Promise.all([this.#processor?.forceFlush(), this.#client?.flush()]); } async shutdown(): Promise<void> { await Promise.all([this.#processor?.shutdown(), this.#client?.shutdown()]); }

flush()同时冲刷 SpanProcessor 与 Client，shutdown()则同时优雅关闭两者，保证进程退出前数据不丢失（见 tracing.ts）。

四、mastra.* 属性到 langfuse.* 的映射原理

这是本包最有价值的部分：SpanConverter生成的 OTel Span 属性以mastra.*为前缀，而 Langfuse 的 OTLP 端点只读取langfuse.*命名空间。mapMastraToLangfuseAttributes在导出前就地（in-place）完成映射（见 tracing.ts）。映射规则可归纳如下：

4.1 保留字段映射（可筛选的顶级字段）

对于gen_ai类 Span，输入输出保持gen_ai.input.messages/gen_ai.output.messages原生读取路径；只有不存在这些字段时才回退到mastra.*前缀匹配（详见 tracing.ts 中Input/Output映射段）。

4.2 根 Span 身份与 Trace 级元数据

当span.isRootSpan为真时，映射逻辑会额外做四件事：

1. Trace 级输入输出：把根 Span 的input/output镜像到langfuse.trace.input/langfuse.trace.output。源码注释说明：没有这一步，Langfuse trace 顶层的输入输出为空，会破坏映射到 Trace input/output 的 LLM-as-a-judge 评估器。
2. 实体身份：AGENT_RUN根 Span 写入langfuse.trace.name（取entityName ?? entityId）以及langfuse.trace.metadata.agentId/agentName；WORKFLOW_RUN同理写入workflowId/workflowName。这样每个 Langfuse trace 都被限定到发起它的 Agent/Workflow，便于按 trace 名或元数据过滤来限定 Langfuse 评估器的作用域。
3. 用户 traceName 优先：如果用户通过mastra.metadata.traceName显式设置了名字，则保留用户值，不被实体名覆盖（测试preserves user-provided traceName over the agent default验证）。
4. 剩余元数据前转：其余mastra.metadata.*键（如runId、resourceId、用户自定义键）转发到langfuse.trace.metadata.*，使它们成为可筛选的顶级 trace 元数据；非字符串值用 JSON 序列化（Langfuse 摄入时会还原类型）。注意只处理根 Span——因为 Langfuse 会从任意 Span 应用langfuse.trace.*，子 Span 可能覆盖 trace 级信息，所以刻意只在根 Span 上做。有DEDICATED_METADATA_KEYS（userId、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 关联：withLangfusePrompt

@mastra/langfuse还提供了withLangfusePrompt辅助函数（见 helpers.ts），用于启用 Langfuse Prompt Tracing（Prompt 关联）。它配合@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.prompt：

export function withLangfusePrompt(prompt: LangfusePromptInput): TracingOptionsUpdater { return opts => ({ ...opts, metadata: { ...opts.metadata, langfuse: { ...(opts.metadata?.langfuse as Record<string, 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 与 addScoreToTrace

LangfuseExporter支持把 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声明了peerDependencies：@mastra/core >=1.16.0-0 <2.0.0-0、@opentelemetry/api ^1.9.0、@opentelemetry/sdk-trace-base ^2.0.1，并指定 Node.js>=22.13.0。接入前请确认项目满足这些版本要求。

九、接入自检清单

1. LANGFUSE_PUBLIC_KEY与LANGFUSE_SECRET_KEY均已设置，或通过new LangfuseExporter({ publicKey, secretKey })显式传入——否则 Exporter 会静默禁用；
2. 自托管 Langfuse 时设置LANGFUSE_BASE_URL或baseUrl，确保无尾随/；
3. 需要即时可见时开启realtime: true，生产环境建议保持批量模式并调整flushAt/flushInterval；
4. 需要在 Langfuse 中按环境/版本区分时，设置environment/release（或对应环境变量）；
5. Prompt 关联请使用withLangfusePrompt({ name, version })，勿依赖已弃用的id；
6. 进程退出前调用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