跳到主要内容

导出、JWT 鉴权与验证

本章讲“最后一公里”：span 生产出来以后怎么带着鉴权信息安全发到后端、出错怎么优雅降级，以及唯一带“eval 味”的 validate_trace_spans 怎么回查后端做断言。

3.1 导出管道一眼​

span 结束（on_end）后进入两个 processor（§01 装的），其中负责上传的是 BatchSpanProcessor：攒一批、按 schedule_delay_millis 定期刷，调用底层 exporter。root span 结束时额外 force_flush 一下（§01 的 finalize_span），不让会话数据等批量。

真正发 HTTP 的是 AuthenticatedOTLPExporter（agentops/sdk/exporters.py:16），它继承标准的 OTLPSpanExporter，只加了“动态 JWT”和“鉴权失败降级”两件事。

3.2 为什么 JWT 要“动态”​

要解决的小问题： API key 换 JWT 要走网络，不能阻塞 init()。所以 token 是后台异步拿的——但 exporter 可能在 token 还没拿到时就创建了。怎么让 exporter “等 token 到了自动用上”而不用重建？

思路： 不存死 token，而是存一个 jwt_provider 函数。每次 export 时现调一下拿最新值。

这根线从 client 穿到 exporter：

Client 后台线程：API key → _fetch_auth_async → 拿到 token → _set_auth_data 存起
                                                       │
   Client.get_current_jwt() 读这个存起的 token ←────┘
            ↑
   jwt_provider = lambda: self.get_current_jwt()   （client.py 里定义）
            ↑
   setup_telemetry(... jwt_provider=jwt_provider)  把它传给 exporter
            ↑
   exporter.export() 每次调 _get_current_jwt() 现拿最新 Bearer
  • 后台拿 token：Client._fetch_auth_async（agentops/client/client.py:96）——拿到后 _set_auth_data 存起，同时把真实 project_id 回填进 tracer 配置。
  • exporter 取 token：_get_current_jwt（agentops/sdk/exporters.py:71）优先调 provider。
  • export 时拼头：_prepare_headers（agentops/sdk/exporters.py:104）把 Authorization: Bearer <token> 覆上。

3.3 安全坑：受保护的请求头​

用户可以传自定义 headers，但不能让他们覆掉鉴权头。_filter_user_headers（agentops/sdk/exporters.py:80）维了一张 PROTECTED_HEADERS（authorization、x-api-key、content-type…），用户试图传这些会被静默丢掉。这防了“用户误传一个 Authorization 把真正的 JWT 覆掉”。

3.4 鉴权失败降级：60 秒冷却​

要解决的小问题： 如果 token 过期或错误，exporter 会不停重试、不停报 401。怎么别辣眼又别打垮后端？

export（agentops/sdk/exporters.py:121）的做法：遇到 401/403 或 JWT 过期时，记下 _last_auth_failure 时间戳，60 秒内直接返回 FAILURE 不再尝试（_auth_failure_threshold = 60）。成功一次就重置。所有异常（网络错、服务端错）都被接住转成 SpanExportResult.FAILURE，绝不向上抛——这是“observability 不能拖垮业务”的原则。

这也解释了为什么 §01 说“不拿 API key 也能跑”：span 照样创建，只是 export 静静失败。

3.5 重放 URL：怎么拼出 dashboard 链接​

每次 start/end trace 都会打一条 session 重放链接。get_trace_url（agentops/helpers/dashboard.py:11）把 trace_id 转成 32 位十六进制（不加 UUID 连字符，因为 dashboard 也不加），拼成 {app_url}/sessions?trace_id=...。是否打受 log_session_replay_url 控制。

3.6 验证：唯一带“eval 味”的部分​

这是 AgentOps 里最接近“评测”的东西，但要看清楚它的边界。 它不打分“答得对不对”，只断言“这次跑确实产生了 span / 确实有 LLM 活动”——主要给测试与 CI 用。

validate_trace_spans（agentops/validation.py:209）的做法是回查后端公开 API，而不是看本地内存：

  1. 从 trace_id / trace_context / 当前 span 三者之一拿到 trace_id。
  2. get_jwt_token_sync 拿 token（拿不到则跳过验证而非报错，validation.py:269）。
  3. 重试轮询 GET /public/v1/traces/{id}，直到 span 数 >= min_spans。
  4. 调 /metrics：若 total_tokens > 0 就肯定有 LLM 活动；否则退而用 check_llm_spans（validation.py:157）看 span 属性里有没有 agentops.span.kind == "llm" 或 gen_ai 属性。
  5. 要求有 LLM 活动却没有 → 抛 ValidationError。

典型用法（测试里）：

# 示意，非源码：跨 init/run 后断言“这次跑真的有 LLM 活动”
ctx = agentops.start_trace("my_test")
run_my_agent()
agentops.end_trace(ctx)
result = agentops.validate_trace_spans(trace_context=ctx)   # 报错则测试失败
agentops.print_validation_summary(result)

边界提醒： 这个验证依赖后端已经收到并索引了 span，所以它带重试循环（max_retries=10，默认间隔 1 秒）；且拿不到 token 时是“跳过”而非失败，别误把它当离线断言用。

代码地图​

主题文件符号名
鉴权导出器agentops/sdk/exporters.pyAuthenticatedOTLPExporter / export
动态 tokenagentops/sdk/exporters.py_get_current_jwt / _prepare_headers
受保护头agentops/sdk/exporters.py_filter_user_headers
后台拿 tokenagentops/client/client.py_fetch_auth_async / _set_auth_data / get_current_jwt
jwt provider 接线agentops/sdk/core.pysetup_telemetry（jwt_provider 参数）
重放 URLagentops/helpers/dashboard.pyget_trace_url / log_trace_url
验证主函数agentops/validation.pyvalidate_trace_spans
LLM span 判定agentops/validation.pycheck_llm_spans
结果打印agentops/validation.pyprint_validation_summary