☰
Google Antigravity SDK可观测性实战:OpenTelemetry追踪与Token成本审计配置指南
2026/9/27 18:33:08 网站建设 项目流程

Google Antigravity SDK可观测性实战:OpenTelemetry追踪与Token成本审计配置指南

【免费下载链接】antigravity-sdk-pythonA Python library for building AI agents that leverage the full power of Google Antigravity.项目地址: https://gitcode.com/gh_mirrors/an/antigravity-sdk-python

Google Antigravity SDK是用于构建 AI Agent 的 Python 库,而它的可观测性能力(OpenTelemetry 分布式追踪 + Token 成本审计)往往被新手忽略。本文带你用不到 30 行配置,给 Agent 加上完整的执行链路追踪与 Token 用量账单,轻松定位"Agent 为什么这么慢、花了这么多钱"。

💡 阅读时长约 8 分钟 | 无需额外经验,会基本 Python 即可上手

为什么 AI Agent 需要可观测性?

传统程序出了 bug 看日志就行,但 AI Agent 的"一次回答"背后藏着复杂的链路:模型调用、工具执行、子代理(Subagent)派生……任何一环变慢或报错,肉眼都难以察觉。

SDK 提供三层观测手段:

手段解决什么问题核心入口
OpenTelemetry 追踪每一步耗时、哪次工具调用出错、子代理调用树utils/otel.py
Token 用量审计会话总消耗、单个回合增量、每个子代理的花费conversation.py
预算上限(Budget)超支前主动刹车,控制模型/工具/Token 次数budget_limits.py

一键开启 OpenTelemetry 追踪

1. 安装 OTel 依赖

SDK 的追踪模块通过可选依赖提供(见 pyproject.toml):

pip install "google-antigravity[otel]"

2. 注册追踪钩子

所有追踪逻辑都基于 SDK 的**生命周期钩子(Hooks)**实现,一行注册即可:

from google import antigravity from google.antigravity.utils import otel as otel_hooks config = antigravity.LocalAgentConfig( hooks=otel_hooks.get_otel_hooks(), # 9 个钩子一次注册 )

get_otel_hooks()会返回 9 个开箱即用的钩子(定义于 otel.py),覆盖从会话开始到工具结束的完整生命周期。

3. 配置导出器

以打印到控制台为例(官方示例 observability_otel.py):

from opentelemetry import trace from opentelemetry.sdk import trace as sdk_trace from opentelemetry.sdk.trace import export provider = sdk_trace.TracerProvider() provider.add_span_processor( export.SimpleSpanProcessor(export.ConsoleSpanExporter()) ) trace.set_tracer_provider(provider)

生产环境只需把ConsoleSpanExporter换成 OTLP 导出器,即可接入任意 OpenTelemetry 后端。

追踪结构解析:4 层 Span 是怎么嵌套的

运行一次 Agent 后,控制台会输出树状的 Span。SDK 自动构建了四层结构(实现逻辑见 otel.py):

antigravity.session ← 整个会话 └── invoke_agent Antigravity ← 每一轮对话(Turn) ├── antigravity.step.0 ← 第 0 步 │ └── execute_tool get_weather ← 工具调用 └── invoke_agent Poet ← 子代理独立成子树 └── antigravity.step.0

每个 Span 都带上了通用的gen_ai标准属性(gen_ai.operation.name、gen_ai.agent.name、gen_ai.tool.name),方便在 Jaeger、Grafana 等平台上按 Agent 维度聚合。

⚠️两个值得知道的细节:

  • 工具 Span 的开始时机在安全策略检查之前,因此人工确认(Human-in-the-loop)的等待时间会被计入工具耗时;
  • 工具执行抛异常时,钩子会自动将异常记录到 Span 并打上ERROR状态(OTelOnToolErrorHook),无需自己写 try/except。

Token 成本审计:3 个属性看懂整张账单

追踪解决"怎么跑的",用量审计解决"花了多少"。Agent 的会话对象暴露了 3 个现成属性(定义见 conversation.py):

async with antigravity.Agent(config) as agent: await agent.chat("What is the weather in Seattle?") usage = agent.conversation.total_usage # ① 会话累计 turn = agent.conversation.last_turn_usage # ② 最近一轮增量 by_agent = agent.conversation.trajectory_usages # ③ 按子代理拆分

① 会话累计用量:返回UsageMetadata对象(types.py),包含 5 个关键字段:

字段含义
prompt_token_count输入 Token 数
cached_content_token_count命中缓存的 Token 数(输入的子集,通常更便宜)
candidates_token_count模型生成的输出 Token 数(不含思考)
thoughts_token_count思考/推理消耗的 Token 数
total_token_count以上三者之和

⚠️避坑提示:如果你用的是支持"扩展思考"的模型,thoughts_token_count可能远超输出 Token,成为成本大头——审计时请优先关注它(参考文档 observability.md)。另外若 Agent 执行失败(如 API Key 无效),用量可能报告为 0。

② 单轮增量用量:last_turn_usage通过"当前累计 − 本轮开始前快照"自动算出差值,适合做每轮成本告警。

③ 子代理拆分:trajectory_usages返回{轨迹ID: 用量}字典,主代理的轨迹 ID 与会话 ID 相同,每个子代理拥有独立轨迹——可以精确回答"是主代理花了钱,还是某个子代理在烧钱"。

预算刹车:用 BudgetConfig 控制成本上限

审计是"事后看账单",预算配置是"事前装刹车"。通过LocalAgentConfig传入 BudgetConfig,可打开 5 个限流阀门:

config = antigravity.LocalAgentConfig( budget_config=antigravity.types.BudgetConfig( max_model_calls=10, # 最多调用模型 10 次 max_tool_calls=20, # 最多执行 20 次工具 max_input_tokens=50_000, # 输入 Token 上限(推理前拦截) max_output_tokens=10_000, # 累计输出上限 max_total_tokens=80_000, # 累计总 Token 上限 ) )

任一阀门耗尽时,回合会携带对应的停止原因(如MAX_MODEL_CALLS_EXCEEDED、MAX_TOTAL_TOKENS_EXCEEDED),你可以在代码里读取response.stop_reason并优雅降级。完整的五阀门触发演示见 budget_limits.py。

🛑 小技巧:max_input_tokens是推理前主动拦截——在还没花钱之前就刹车,是唯一能"零消耗"阻止回合的阀门。

进阶:标准日志与自定义审计钩子

不想接 OpenTelemetry 平台?SDK 内置标准 Python 日志,两行即可开启运行详情(示例 observability.py):

import logging logging.getLogger("google.antigravity").setLevel(logging.INFO)

而基于装饰器的钩子可以零框架写出审计日志——每次工具执行后自动落一条审计记录:

from google.antigravity.hooks import hooks @hooks.post_tool_call async def audit_log_tool_call(data): print(f"[AUDIT] Tool execution completed. Result: {data}")

把它和total_usage结合,就是一份极简的"工具 × Token"审计方案。

关键文件速查

文件说明
google/antigravity/utils/otel.py9 个 OTel 追踪钩子的完整实现
examples/deep_dives/observability_otel.py含子代理的追踪端到端示例
examples/getting_started/observability.py日志 + 审计钩子 + 用量打印入门示例
examples/getting_started/budget_limits.py5 个预算阀门逐一触发演示
google/antigravity/conversation/conversation.pytotal_usage/last_turn_usage/trajectory_usages定义
google/antigravity/types.pyUsageMetadata数据模型
skills/google-antigravity-sdk/references/observability.md官方可观测性参考文档

小结

  1. 追踪:pip install "google-antigravity[otel]"+get_otel_hooks()一行注册,自动获得"会话 → 回合 → 步骤 → 工具"四层 Span 树;
  2. 审计:total_usage/last_turn_usage/trajectory_usages三个属性,从全局到子代理逐层对账,重点关注thoughts_token_count;
  3. 控费:BudgetConfig五个阀门在超支前主动刹车,配合stop_reason优雅降级。

把这三件套配上,你的 Agent 从此每一分钱、每一毫秒都有据可查 📊

【免费下载链接】antigravity-sdk-pythonA Python library for building AI agents that leverage the full power of Google Antigravity.项目地址: https://gitcode.com/gh_mirrors/an/antigravity-sdk-python

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询