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.py | 9 个 OTel 追踪钩子的完整实现 |
| examples/deep_dives/observability_otel.py | 含子代理的追踪端到端示例 |
| examples/getting_started/observability.py | 日志 + 审计钩子 + 用量打印入门示例 |
| examples/getting_started/budget_limits.py | 5 个预算阀门逐一触发演示 |
| google/antigravity/conversation/conversation.py | total_usage/last_turn_usage/trajectory_usages定义 |
| google/antigravity/types.py | UsageMetadata数据模型 |
| skills/google-antigravity-sdk/references/observability.md | 官方可观测性参考文档 |
小结
- 追踪:
pip install "google-antigravity[otel]"+get_otel_hooks()一行注册,自动获得"会话 → 回合 → 步骤 → 工具"四层 Span 树; - 审计:
total_usage/last_turn_usage/trajectory_usages三个属性,从全局到子代理逐层对账,重点关注thoughts_token_count; - 控费:
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),仅供参考