Skip to content

用于 ADK 的 Future AGI 可观测性

Supported in ADKPython

Future AGI 是一个面向 AI 智能体的可观测性和评估平台。 traceai-google-adk 包自动检测 ADK 智能体，并将每次智能体运行、模型调用、工具执行和事件循环周期作为 OpenTelemetry span 导出到 Future AGI，在那里你可以检查运行树、评估行为并运行实验。

Future AGI ADK traces

概述

traceai-google-adk 包为 ADK 添加了 OpenTelemetry 检测能力， 使你能够：

  • 追踪智能体运行：捕获每次智能体调用、工具调用、模型请求和响应，包括提示词、补全内容、参数和令牌使用情况。
  • 评估行为：针对捕获的追踪运行预构建或自定义评估器。
  • 调试智能体：深入分层运行树，查找失败的工具调用、延迟热点和意外分支。

先决条件

  1. 在 app.futureagi.com 注册。
  2. 从仪表盘复制你的 FI_API_KEY 和 FI_SECRET_KEY。
  3. 设置环境变量：
export FI_API_KEY=<你的-fi-api-密钥>
export FI_SECRET_KEY=<你的-fi-secret-密钥>
export GOOGLE_API_KEY=<你的-google-api-密钥>

安装

pip install traceai-google-adk

traceai-google-adk 包将 google-adk 和 google-genai 声明为运行时依赖，因此它们会被自动传递安装。

将追踪数据发送到 Future AGI

在启动时注册 Future AGI 追踪器，并在运行任何智能体之前附加 GoogleADKInstrumentor。后续的所有 ADK 智能体调用都会被自动捕获。

import asyncio

from fi_instrumentation import register
from fi_instrumentation.fi_types import ProjectType
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types
from traceai_google_adk import GoogleADKInstrumentor

tracer_provider = register(
    project_type=ProjectType.OBSERVE,
    project_name="adk-weather-agent",
)
GoogleADKInstrumentor().instrument(tracer_provider=tracer_provider)


def get_weather(city: str) -> dict:
    """获取指定城市的当前天气报告。"""
    if city.lower() == "new york":
        return {
            "status": "success",
            "report": "The weather in New York is sunny with a temperature of 25°C.",
        }
    return {
        "status": "error",
        "error_message": f"Weather information for '{city}' is not available.",
    }


agent = Agent(
    name="weather_agent",
    model="gemini-flash-latest",
    description="回答天气问题的智能体。",
    instruction="你必须使用可用的工具来寻找答案。",
    tools=[get_weather],
)


async def main():
    runner = InMemoryRunner(agent=agent, app_name="weather_app")
    await runner.session_service.create_session(
        app_name="weather_app", user_id="user", session_id="session"
    )
    async for event in runner.run_async(
        user_id="user",
        session_id="session",
        new_message=types.Content(
            role="user",
            parts=[types.Part(text="纽约的天气怎么样？")],
        ),
    ):
        if event.is_final_response():
            print(event.content.parts[0].text.strip())


if __name__ == "__main__":
    asyncio.run(main())

在仪表盘中查看追踪

运行智能体，然后在 Future AGI 仪表盘 中打开你的项目。每次 ADK 智能体运行都会生成一个分层追踪，包含提示词、补全内容、模型参数、令牌使用情况、工具输入和输出以及事件循环周期，供你检查。

资源