使用 Cloud Trace 进行智能体可观测性

使用 ADK，你已经能够利用强大的 Web 开发 UI 在本地检查和观察智能体交互，这在这里已有讨论。但是，如果我们的目标是云端部署，我们将需要一个集中式仪表板来观察真实流量。

Cloud Trace 是 Google Cloud Observability 的一个组件。它是一个强大的工具，专门专注于跟踪功能，用于监控、调试和改进应用程序的性能。对于智能体开发套件 (ADK) 应用程序，Cloud Trace 实现了全面的跟踪，帮助你了解请求如何流经智能体的交互，并识别 AI 智能体中的性能瓶颈或错误。

概述

Cloud Trace 基于 OpenTelemetry 构建，这是一个支持多种语言和摄取方法的开源标准，用于生成跟踪数据。这与 ADK 应用程序的可观测性实践保持一致，后者也利用 OpenTelemetry 兼容的工具，允许你：

• 跟踪智能体交互：Cloud Trace 持续收集和分析项目中的跟踪数据，使你能够快速诊断 ADK 应用程序中的延迟问题和错误。这种自动数据收集简化了识别复杂智能体工作流中问题的过程。
• 调试问题：通过分析详细跟踪快速诊断延迟问题和错误。对于理解表现为跨不同服务通信延迟增加或特定智能体操作（如工具调用）期间问题的理解至关重要。
• 深度分析和可视化：Trace Explorer 是分析跟踪的主要工具，提供可视化辅助工具，如跨持续时间的热图和请求/错误率的折线图。它还提供了一个按服务和操作分组的跨度表，提供代表性跟踪的一键访问和瀑布视图，以便轻松识别智能体执行路径中的瓶颈和错误来源

以下示例将假设以下智能体目录结构

working_dir/
├── weather_agent/
│   ├── agent.py
│   └── __init__.py
└── deploy_agent_engine.py
└── deploy_fast_api_app.py
└── agent_runner.py

# weather_agent/agent.py

import os
from google.adk.agents import Agent

os.environ.setdefault("GOOGLE_CLOUD_PROJECT", "{your-project-id}")
os.environ.setdefault("GOOGLE_CLOUD_LOCATION", "global")
os.environ.setdefault("GOOGLE_GENAI_USE_VERTEXAI", "True")


# 定义一个工具函数
def get_weather(city: str) -> dict:
    """检索指定城市的当前天气报告。

    Args:
        city (str): 要检索天气报告的城市名称。

    Returns:
        dict: 状态和结果或错误消息。
    """
    if city.lower() == "new york":
        return {
            "status": "success",
            "report": (
                "纽约的天气是晴天，温度为 25 度"
                " 摄氏度（77 华氏度）。"
            ),
        }
    else:
        return {
            "status": "error",
            "error_message": f"'{city}' 的天气信息不可用。",
        }


# 创建一个带有工具的智能体
root_agent = Agent(
    name="weather_agent",
    model="gemini-2.5-flash",
    description="使用天气工具回答问题的智能体。",
    instruction="你必须使用可用的工具来找到答案。",
    tools=[get_weather],
)

Cloud Trace 设置

智能体引擎部署设置

智能体引擎部署 - 从 ADK CLI

你可以通过在部署智能体时添加 --trace_to_cloud 标志来启用云跟踪，使用 adk deploy agent_engine 命令进行智能体引擎部署。

adk deploy agent_engine \
    --project=$GOOGLE_CLOUD_PROJECT \
    --region=$GOOGLE_CLOUD_LOCATION \
    --staging_bucket=$STAGING_BUCKET \
    --trace_to_cloud \
    $AGENT_PATH

智能体引擎部署 - 从 Python SDK

如果你更喜欢使用 Python SDK，你可以在初始化 AdkApp 对象时通过添加 enable_tracing=True 来启用云跟踪

# deploy_agent_engine.py

from vertexai.preview import reasoning_engines
from vertexai import agent_engines
from weather_agent.agent import root_agent

import vertexai

PROJECT_ID = "{your-project-id}"
LOCATION = "{your-preferred-location}"
STAGING_BUCKET = "{your-staging-bucket}"

vertexai.init(
    project=PROJECT_ID,
    location=LOCATION,
    staging_bucket=STAGING_BUCKET,
)

adk_app = reasoning_engines.AdkApp(
    agent=root_agent,
    enable_tracing=True,
)


remote_app = agent_engines.create(
    agent_engine=adk_app,
    extra_packages=[
        "./weather_agent",
    ],
    requirements=[
        "google-cloud-aiplatform[adk,agent_engines]",
    ],
)

Cloud Run 部署设置

Cloud Run 部署 - 从 ADK CLI

你可以通过在部署智能体时添加 --trace_to_cloud 标志来启用云跟踪，使用 adk deploy cloud_run 命令进行 cloud run 部署。

adk deploy cloud_run \
    --project=$GOOGLE_CLOUD_PROJECT \
    --region=$GOOGLE_CLOUD_LOCATION \
    --trace_to_cloud \
    $AGENT_PATH

如果你想启用云跟踪并在 Cloud Run 上使用自定义智能体服务部署，可以参考下面的自定义部署设置部分

自定义部署设置

从内置 get_fast_api_app 模块

如果你想自定义自己的智能体服务，你可以通过使用内置的 get_fast_api_app 模块初始化 FastAPI 应用并设置 trace_to_cloud=True 来启用云跟踪

# deploy_fast_api_app.py

import os
from google.adk.cli.fast_api import get_fast_api_app
from fastapi import FastAPI

# 为云跟踪设置 GOOGLE_CLOUD_PROJECT 环境变量
os.environ.setdefault("GOOGLE_CLOUD_PROJECT", "alvin-exploratory-2")

# 发现当前工作目录中的 `weather_agent` 目录
AGENT_DIR = os.path.dirname(os.path.abspath(__file__))

# 创建启用了云跟踪的 FastAPI 应用
app: FastAPI = get_fast_api_app(
    agents_dir=AGENT_DIR,
    web=True,
    trace_to_cloud=True,
)

app.title = "weather-agent"
app.description = "用于与智能体 weather-agent 交互的 API"


# 主执行
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8080)

从自定义智能体运行器

如果你想完全自定义 ADK 智能体运行时，你可以通过使用 Opentelemetry 的 CloudTraceSpanExporter 模块来启用云跟踪。

# agent_runner.py

from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from weather_agent.agent import root_agent as weather_agent
from google.genai.types import Content, Part
from opentelemetry import trace
from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
from opentelemetry.sdk.trace import export
from opentelemetry.sdk.trace import TracerProvider

APP_NAME = "weather_agent"
USER_ID = "u_123"
SESSION_ID = "s_123"

provider = TracerProvider()
processor = export.BatchSpanProcessor(
    CloudTraceSpanExporter(project_id="{your-project-id}")
)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

session_service = InMemorySessionService()
runner = Runner(agent=weather_agent, app_name=APP_NAME, session_service=session_service)


async def main():
    session = await session_service.get_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
    )
    if session is None:
        session = await session_service.create_session(
            app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
        )

    user_content = Content(
        role="user", parts=[Part(text="巴黎的天气怎么样？")]
    )

    final_response_content = "没有响应"
    async for event in runner.run_async(
        user_id=USER_ID, session_id=SESSION_ID, new_message=user_content
    ):
        if event.is_final_response() and event.content and event.content.parts:
            final_response_content = event.content.parts[0].text

    print(final_response_content)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

检查 Cloud Traces

设置完成后，每当你与智能体交互时，它都会自动将跟踪数据发送到 Cloud Trace。你可以通过访问 console.cloud.google.com 并在配置的 Google Cloud Project 上访问 Trace Explorer 来检查跟踪

然后你将看到由 ADK 智能体产生的所有可用跟踪，这些跟踪配置在多个跨度名称中，如 invocation、agent_run、call_llm 和 execute_tool

如果你点击其中一个跟踪，你将看到详细过程的瀑布视图，类似于我们在 Web 开发 UI 中使用 adk web 命令看到的视图。

资源

• Google Cloud Trace 文档