Mastra Observability：用 @mastra/observability 构建 Agent 全链路可观测体系

Mastra Observability用 mastra/observability 构建 Agent 全链路可观测体系【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra本指南以 Mastra 仓库中 observability/mastra 包为主线完整讲解mastra/observability的安装、配置与工作原理如何在 Mastra 应用中接入分层 Trace、自动提取指标、将结构化日志与活动 Trace 关联并通过采样、敏感数据过滤、基数控制等手段在生产环境中掌控观测成本与安全边界。读完本文你将能够独立配置一个多实例、多导出目标的 Agent 观测栈并把 trace、日志、指标、评分与反馈信号打通到 Studio、Mastra Platform 或任意第三方后端。一、mastra/observability 是什么mastra/observability是 Mastra 的核心可观测性包见 package.json它负责对 Mastra 应用中的Agent 运行、模型生成model generation、工具与 MCP 调用、处理器processor执行、工作流运行workflow run与工作流步骤workflow step进行全链路插桩。该包以分层 Tracehierarchical traces为核心并自动提取指标、把结构化日志与当前活动的 Trace 相关联。从源码结构看见 src 目录该包由几个相互协作的模块组成default.ts/registry.ts/instances/对外入口Observability类、实例注册表以及BaseObservabilityInstance/DefaultObservabilityInstance等实例实现bus/统一的可观测性事件总线ObservabilityBus负责把 tracing、log、metric、score、feedback 事件路由给注册的 exporter 与 bridgeexporters/MastraStorageExporter、MastraPlatformExporter、ConsoleExporter、TestExporter等导出器span_processors/输出端处理器其中SensitiveDataFilter默认启用负责脱敏metrics/指标自动提取duration、token、cost与基数过滤CardinalityFilterclient/客户端可观测性代理W3C trace context 注入与 OTLP/JSON 载荷接收spans//context//model-tracing.tsSpan 模型、日志与指标上下文、模型调用追踪。该包以mastra/core的observability类型体系为基础实现peer 依赖为mastra/core 1.16.0-0 2.0.0-0与zod ^3.25.0 || ^4.0.0Node 版本要求22.13.0。二、安装与最小接入在项目中使用pnpm或npm安装npm install mastra/observability安装后即可在 Mastra 实例上挂载观测配置这是文档给出的最小用法import { Mastra } from mastra/core; import { Observability, MastraStorageExporter, MastraPlatformExporter } from mastra/observability; export const mastra new Mastra({ observability: new Observability({ configs: { default: { serviceName: my-app, exporters: [new MastraStorageExporter(), new MastraPlatformExporter()], }, }, }), });这段配置的含义是注册一个名为default的观测实例服务名my-app同时把观测事件导出到 Mastra 存储供 Studio 查询与 Mastra Platform。MastraStorageExporter在 mastra-storage.ts 中实现负责把事件批量写入配置的ObservabilityStorageMastraPlatformExporter在 mastra-platform.ts 中实现负责把事件发布到 Mastra Platform 的 collector。三、核心配置项逐项拆解在 config.ts 中每个观测实例ObservabilityInstanceConfig支持以下字段配置项类型默认值说明serviceNamestring必填服务名用于区分不同服务的 tracesamplingSamplingStrategy{ type: always }采样策略控制是否采集 traceexportersObservabilityExporter[][]自定义导出器至少配置一个 exporter 或 bridgebridgeObservabilityBridge无可观测性桥如 OpenTelemetry 桥用于上下文提取spanOutputProcessorsSpanOutputProcessor[][]自定义 Span 输出处理器includeInternalSpansbooleanfalse是否导出 Mastra 内部 Span如 MODEL_CHUNKexcludeSpanTypesSpanType[]无从导出中剔除的 Span 类型可降低按 Span 计费的成本spanFilter(span) boolean无细粒度过滤函数返回false则丢弃该 SpanrequestContextKeysstring[][]从 RequestContext 自动提取到 Span 元数据的键支持点号嵌套serializationOptionsSerializationOptions无控制 input/output/attributes 的截断与序列化cardinalityCardinalityConfig无指标标签基数保护配置logging{ enabled, level }双写开启级别warn控制观测日志loggerVNext的过滤级别与是否双写示例排除噪声 Span、只保留成功工具调用import { SpanType } from mastra/core/observability; const config { configs: { default: { serviceName: my-app, // 不计费的链路内 Span全部剔除减少平台成本 excludeSpanTypes: [SpanType.MODEL_CHUNK, SpanType.MODEL_STEP], spanFilter: span { // 丢弃失败的模型生成块 if (span.type SpanType.MODEL_CHUNK span.attributes?.success false) return false; return true; }, exporters: [new MastraStorageExporter()], }, }, };从源码看getSpanForExportinstances/base.ts的执行顺序是先检查isValid与内部 Span 过滤再检查excludeSpanTypes随后依次运行spanOutputProcessors最后应用spanFilter——因此用户过滤器拿到的是最终将要导出的 Span 数据可按类型、属性、实体、元数据任意组合过滤。3.1 注册表级配置default、configs 与 configSelectorObservabilityRegistryConfigconfig.ts在实例之上再做一层组织default: { enabled: true }旧式快捷开关已标记deprecated默认会注册一个serviceName: mastra、使用MastraStorageExporter MastraPlatformExporter的实例configs命名实例的映射表值可以是普通配置对象也可以是预实例化的ObservabilityInstanceconfigSelector当存在多个config 时必须提供用于在运行时根据ConfigSelectorOptions选择实例sensitiveDataFilter控制是否自动给每个实例套上SensitiveDataFiltertrue默认开启 /false关闭 / 传对象自定义选项。Zod 校验规则config.ts明确约束了这三种模式不能混用default开启时不能再给configs多个 configs 必须有configSelector有configSelector就必须有 configs 或 default。3.2 多实例与配置选择器多实例场景适合“不同服务、不同导出目标”的部署例如同一进程内同时运行网关服务与 Agent 服务import { Mastra } from mastra/core; import { Observability, MastraStorageExporter } from mastra/observability; import { LangfuseExporter } from mastra/langfuse; const mastra new Mastra({ observability: new Observability({ configs: { gateway: { serviceName: api-gateway, exporters: [new MastraStorageExporter()], }, agents: { serviceName: my-agents, exporters: [new MastraStorageExporter(), new LangfuseExporter({ publicKey: ..., secretKey: ... })], }, }, configSelector: ({ operation, entityType }) entityType agent || operation?.includes(agent) ? agents : gateway, }), });每个配置的观测实例拥有独立的 serviceName、exporter、采样策略与 Span 处理器这一点在包文档中有明确说明也与BaseObservabilityInstance的构造逻辑instances/base.ts一致每个实例单独初始化CardinalityFilter、ObservabilityBus并注册自己的 exporter 与 bridge。四、导出器从存储到第三方平台的完整清单4.1 MastraStorageExporterStudio 数据源MastraStorageExporter把 trace、log、metric、score、feedback 事件缓冲后批量写入ObservabilityStorage即 Mastra 配置的存储后端Studio 便可以从存储中查询这些 trace。其关键行为参数mastra-storage.ts参数默认值说明maxBatchSize1000触发批量刷新的 Span 数量阈值maxBufferSize10000缓冲区上限超出后紧急刷新maxBatchWaitMs5000时间触发阈值缓冲非空时最迟等待毫秒数maxRetries4失败重试次数retryDelayMs500指数退避的基础延迟retryDelayMs * 2^(n-1)strategyauto存储写入策略realtime/ 缓冲等auto时按存储适配器偏好选择写入时按信号类型分别调用batchCreateSpans、batchCreateLogs、batchCreateMetrics、batchCreateScores、batchCreateFeedback见 mastra-storage.ts。Span 的 update/end 事件会检查对应 Span 是否已创建未创建则延迟到下一批flushSpanUpdatesmastra-storage.ts从而保证事件顺序正确。4.2 MastraPlatformExporterMastra 云平台MastraPlatformExporter把信号发布到 Mastra Platform 的 collector 路由/spans/publish、/logs/publish、/metrics/publish、/scores/publish、/feedback/publish见 mastra-platform.ts。它默认过滤掉MODEL_CHUNKSpanDEFAULT_PLATFORM_SPAN_FILTER并实现了配额暂停协议当 collector 返回402或x-mastra-observability: disabled时停止发送并按照x-mastra-observability-retry-after头默认 300 秒周期性探测恢复。4.3 其他可用导出器仓库的 observability 目录下还有多个配套包例如 Arize、Braintrust、Langfuse、LangSmith、Sentry 等参见 observability 下的arize、braintrust、langfuse、langsmith、sentry等子目录。包文档特别指出额外的包提供了面向这些服务以及 OpenTelemetry 兼容后端的导出器。4.4 导出器如何收到事件ObservabilityBus所有导出器都通过中央可观测性总线接收事件。ObservabilityBusbus/observability-bus.ts将事件按类型路由到实现了对应方法onTracingEvent、onLogEvent、onMetricEvent、onScoreEvent、onFeedbackEvent的处理器未实现该方法的处理器会被静默跳过。事件在扇出前会经过deepClean对 log/metric/score/feedback 的整包做深度清洗tracing 事件在 Span 构造时已清洗保证JSON.stringify安全。flush()采用两阶段流程先排空在途 handler Promise再调用各 exporter 与 bridge 的flush()排空其 SDK 内部缓冲如 OTEL BatchSpanProcessor这对 Inngest 等持久化执行引擎尤其重要——步骤完成后进程可能被中断调用flush()能确保数据送达外部系统。五、敏感数据过滤默认开启的安全底线SensitiveDataFilterspan_processors/sensitive-data-filter.ts是一个默认启用的 Span 输出处理器在 Span 到达 exporter之前统一脱敏。它覆盖 attributes、metadata、input、output、errorInfo、requestContext 六个字段并且对 JSON 字符串做解析后再脱敏。匹配规则字段名大小写不敏感并归一化分隔符——api-key、api_key、Api Key都会归一化为apikey再精确匹配因此promptTokens不会被误伤成token。默认敏感字段包括password、token、secret、key、apikey、auth、authorization、bearer、bearertoken、jwt、credential、clientsecret、privatekey、refresh、ssn脱敏有三种风格redactionStyle风格行为示例full默认一律替换为固定令牌sk-abc…→[REDACTED]partial保留首尾各 3 个字符中间省略长度 ≤6 时整体脱敏sk-abc123xyz→sk-…xyzindexed同一 trace 内同一值映射到稳定令牌[LABEL_N]可在不暴露原文的前提下做关联分析sk-abc→[APIKEY_1]indexed模式的状态按 trace 作用域管理使用 SHA-256 摘要做键不保留原始值最多跟踪最近 1000 个 trace、每个 trace 最多 1000 个唯一值超出后退化为完整脱敏令牌。在 default.ts 中可以确认自动应用逻辑sensitiveDataFilter默认为true若用户自己的spanOutputProcessors里已包含SensitiveDataFilter则跳过自动追加避免双重脱敏自动追加的过滤器排在其他处理器之后运行确保上游处理器引入的敏感数据仍会被脱敏预实例化的ObservabilityInstance不会被修改只会打印警告。自定义脱敏示例import { Observability } from mastra/observability; const observability new Observability({ configs: { default: { serviceName: my-app, exporters: [new MastraStorageExporter()], }, }, // 关闭默认敏感数据过滤不推荐仅示例 sensitiveDataFilter: false, });需要完全关闭时设sensitiveDataFilter: false需要定制时传入选项对象const observability new Observability({ configs: { default: { serviceName: my-app, exporters: [new MastraStorageExporter()], }, }, sensitiveDataFilter: { sensitiveFields: [password, token, secret, key, apikey, jwt, api_key_custom], redactionToken: ***, redactionStyle: indexed, }, });六、采样策略控制采集成本与数据量采样策略在 config.ts 中定义支持四种类型// 1. 全量采集默认 { type: always } // 2. 完全不采集 { type: never } // 3. 按比例随机采样probability 必须在 [0,1] { type: ratio, probability: 0.1 } // 4. 应用自定义采样逻辑 { type: custom, sampler: ({ requestContext, metadata }) { // 例如只采样登录/支付相关请求 return metadata?.businessCritical true; }, }shouldSample的实现instances/base.ts对ratio使用Math.random() probability概率越界会告警并退化为不采样custom直接调用用户提供的sampler函数可接收requestContext与metadata。需要特别强调的是trace 级采样startSpaninstances/base.ts只在根 Span 上执行采样判定子 Span 直接继承父 Span 的采样结果父是NoOpSpan则子也是NoOpSpan。这保证了“要么整条 trace 都被采集要么都不被采集”避免出现只有子 Span、没有根 Span 的断裂链路。七、自动提取的指标与结构化日志7.1 指标从 Span 生命周期自动派生包文档说明指标从 Span 生命周期事件自动派生涵盖duration耗时、status状态、模型 token 数、缓存 token 数。对应实现是 metrics/auto-extract.tsemitDurationMetrics根据startTime与endTime计算耗时并带上status: ok | error标签emitTokenMetrics仅对MODEL_GENERATION类型 Span 读取usage字段派发 token 指标成本估算由 metrics/estimator.ts 结合 metrics/pricing-data.jsonl 的定价数据完成模型 ID 解析逻辑见 model-id.ts。值得注意的细节指标发射独立于导出过滤。即使用户通过excludeSpanTypes剔除了AGENT_RUN、TOOL_CALL等 Span 以降低按 Span 计费的成本duration/token/cost 指标依然会生成见 instances/base.ts 中emitSpanEnded的处理。当内部MODEL_GENERATIONSpan 被过滤时其 usage 会“回滚”累加到最近的可见祖先 Span 的internalUsage属性上captureModelUsageRollup/applyUsageRollup让成本归属到用户可见的 Span。7.2 指标基数保护metrics 标签会经过基数过滤cardinality filtering防止 user ID、trace ID 等无界值压垮指标后端。CardinalityFiltermetrics/cardinality.ts默认阻止一组标签键DEFAULT_BLOCKED_LABELS如trace_id等并默认拦截符合 UUID v4 形态的值正则见 cardinality.ts。可通过cardinality配置覆盖configs: { default: { serviceName: my-app, exporters: [new MastraStorageExporter()], cardinality: { blockedLabels: [trace_id, user_id, conversation_id], blockUUIDs: true, }, }, }该过滤器同样作用于用户自定义指标MetricsContext中的标签会先经过CardinalityFilter.filterLabels再发射。7.3 结构化日志与 trace 关联包文档指出结构化日志会继承 trace ID、span ID、tags 与实体元数据。LoggerContextImplcontext/logger.ts由getLoggerContextinstances/base.ts创建当logging.enabled false时返回 no-op logger否则把当前 Span 的traceId、可导出的spanIdresolveExportedSpanId会跳过内部/被排除的 Span、correlation context 与 metadata 一并注入。日志级别按logging.level过滤默认warn并支持logging.enabled关闭向观测存储的双写。这样在 Studio 中点击某条 trace即可看到与之关联的所有日志无需手动拼接 traceId 去检索。八、Span 模型与生命周期Span 的生命周期由startSpan统一管理BaseObservabilityInstance会在创建后自动接线wireSpanLifecycleinstances/base.tsspan.end(options)包装原始end捕获 usage 回滚数据调用原始结束逻辑随后发出SPAN_ENDED事件并触发自动指标提取span.update(options)包装原始update随后发出SPAN_UPDATED事件事件型 Spanspan.isEvent不经过生命周期接线直接在创建时发出结束事件。根 Span 的 metadata 会自动注入environment来自 Mastra 实例环境前提是用户未显式设置RequestContext 中配置的requestContextKeys支持点号嵌套如headers.x-request-id也会被提取进根 Span metadata子 Span 通过继承父 metadata 自动获得这些上下文。这对跨服务的链路贯穿很有用例如在网关层把x-request-id写入 RequestContext所有下游 Agent trace 都会自动带上该请求 ID。九、分数Score与反馈Feedback除 trace/log/metric 外Observability还提供addScore与addFeedbackAPIdefault.ts。可以通过correlationContext自动继承当前 Span 的 trace/span ID快速打分也可以通过traceId/spanId对历史 trace 打分。后者会先从观测存储中重建 tracegetRecordedTrace并以重试调度累计约 5.85 秒规避 exporter 异步刷新的竞态——Span 刚结束时马上打分可能遇到数据尚未落库而丢失该重试机制覆盖了默认刷新间隔并留有余量见 default.ts。分数与反馈事件同样经由ObservabilityBus路由到所有注册的 exporter可被存储或平台消费。十、生命周期管理flush 与 shutdownflush()强制排空缓冲事件并刷新 exporter/bridge 的 SDK 内部缓冲适合 Serverless 环境在运行时实例终止前调用或持久化执行引擎如 Inngest在 durable step 之外调用shutdown()先关闭总线排空剩余事件、清理订阅者再并行关闭所有 exporter、Span 输出处理器与 bridgeinstances/base.ts。Observability注册表层面同样暴露flush()与shutdown()分别对所有已注册实例生效。十一、相关测试与延伸阅读该包带有丰富的测试覆盖是理解行为细节的绝佳入口config.test.ts验证采样策略、序列化、日志配置等 Zod 校验逻辑trace-level-sampling.test.ts验证 trace 级采样继承行为span-filtering.test.ts验证excludeSpanTypes与spanFilter的过滤语义senstive-data-filter.test.ts验证三种脱敏风格与字段归一化mastra-storage.test.ts 与 mastra-platform.test.ts验证两个核心导出器的缓冲、重试与配额协议auto-extract.test.ts 与 cardinality.test.ts验证指标自动提取与基数过滤observability-bus.test.ts验证事件路由与两阶段 flushprocessor-tracing.test.ts 与 model-tracing.test.ts验证处理器与模型调用的插桩链路。__snapshots__/目录下的 trace 快照如agent-tool-call-trace.json、workflow-branching-trace.json展示了真实生成的 trace 结构是理解分层 Span 数据形态的第一手资料。结语mastra/observability把 Agent 应用的观测从“打日志”升级为“一套体系”分层 trace 刻画每次 Agent/工作流/工具/模型调用的完整脉络指标与日志自动挂靠到活动 trace采样、脱敏、基数过滤与 Span 级过滤四道闸门共同守住成本与安全而事件总线让存储、平台与第三方后端可以同时、无侵入地消费同一份观测数据。在生产环境中建议至少配置MastraStorageExporter供 Studio 本地查询并按需叠加平台或第三方 exporter同时保留默认的SensitiveDataFilter再根据业务重要性选择 ratio 或 custom 采样即可获得一套既完整又受控的 Agent 可观测栈。【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考