Analiza la ejecución del agente del ADK con el complemento BigQuery Agent Analytics

1. Introducción

En este codelab, compilarás un sistema multiagente con el Kit de desarrollo de agentes (ADK) y habilitarás la observabilidad del agente con el complemento BigQuery Agent Analytics.Le harás una serie de preguntas al agente y, luego, usarás BigQuery para analizar los registros de seguimiento de la conversación y el uso de las herramientas del agente.

Actividades

• Crea un asistente de comercio minorista multiagente con el ADK
• Inicializa el complemento BigQuery Agent Analytics para capturar y almacenar datos de seguimiento sobre la ejecución de este agente en BigQuery.
• Analiza los datos de registros del agente en BigQuery

Requisitos

• Un navegador web, como Chrome
• Un proyecto de Google Cloud con la facturación habilitada
• Una cuenta de Gmail En la siguiente sección, se mostrará cómo canjear un crédito gratuito de USD 5 para este codelab y configurar un proyecto nuevo.

Este codelab está dirigido a desarrolladores de todos los niveles, incluidos principiantes. Usarás la interfaz de línea de comandos en Google Cloud Shell y código de Python para el desarrollo del ADK. No es necesario que seas un experto en Python, pero tener conocimientos básicos para leer código te ayudará a comprender los conceptos.

2. Antes de comenzar

Para obtener créditos de Google Cloud: Para ayudarte a comenzar, canjea tus créditos gratuitos de Google Cloud con este vínculo. Puedes seguir las instrucciones aquí para activar el crédito y crear un proyecto nuevo.

Crea un proyecto de Google Cloud

1. En la página del selector de proyectos de la consola de Google Cloud, selecciona o crea un proyecto de Google Cloud.

2. Asegúrate de que la facturación esté habilitada para tu proyecto de Cloud. Obtén información sobre cómo verificar si la facturación está habilitada en un proyecto.

Inicia Cloud Shell

Cloud Shell es un entorno de línea de comandos que se ejecuta en Google Cloud y que viene precargado con las herramientas necesarias.

1. Haz clic en Activar Cloud Shell en la parte superior de la consola de Google Cloud:

2. Una vez que te conectes a Cloud Shell, ejecuta este comando para verificar tu autenticación en Cloud Shell:

gcloud auth list

3. Ejecuta el siguiente comando para confirmar que tu proyecto esté configurado para usar gcloud:

gcloud config get project

4. Si tu proyecto no está configurado como se espera, usa el siguiente comando para configurarlo:

export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID

Habilita las APIs

1. Ejecuta este comando para habilitar todas las APIs y los servicios requeridos:

gcloud services enable bigquery.googleapis.com \
cloudresourcemanager.googleapis.com \
aiplatform.googleapis.com

2. Si el comando se ejecuta correctamente, verás un mensaje similar al que se muestra a continuación:

Operation "operations/..." finished successfully.

3. Instalación y configuración

Regresa a Cloud Shell y asegúrate de estar en tu directorio principal.

Ejecuta el siguiente comando en Cloud Shell para crear un nuevo conjunto de datos llamado adk_logs en BigQuery:

bq mk --dataset --location=US adk_logs

Ahora, crearemos un entorno virtual de Python y, luego, instalaremos los paquetes necesarios.

1. Abre una nueva pestaña de terminal en Cloud Shell y ejecuta este comando para crear una carpeta llamada adk-agent-observability y navegar a ella:

mkdir adk-agent-observability
cd adk-agent-observability

2. Crea un entorno virtual de Python:

python -m venv .venv

3. Activa el entorno virtual:

source .venv/bin/activate

4. Instala el ADK:

pip install --upgrade google-adk

4. Crea una aplicación del ADK

Ahora, creemos nuestro agente asistente de ventas. Este agente se diseñará para…

1. Ejecuta el comando de utilidad adk create para generar una nueva aplicación del agente con las carpetas y los archivos necesarios:

adk create retail_assistant_app

Sigue las indicaciones:

1. Elige gemini-2.5-flash para el modelo.
2. Elige Vertex AI para el backend.
3. Confirma tu ID y región del proyecto de Google Cloud predeterminado.

A continuación, se muestra una interacción de ejemplo:

2. Haz clic en el botón Abrir editor en Cloud Shell para abrir el editor de Cloud Shell y ver las carpetas y los archivos recién creados:

Ten en cuenta los archivos generados:

retail_assistant_app/
├── .venv/
└── retail_assistant_app/
    ├── __init__.py
    ├── agent.py
    └── .env

• init.py: Marca la carpeta como un módulo de Python.
• agent.py: Contiene la definición inicial del agente.
• .env: Es posible que debas hacer clic en Ver > Activar o desactivar archivos ocultos para ver este archivo.

• El archivo .env contiene variables de entorno para tu proyecto. Actualiza las variables que no se hayan configurado correctamente desde las indicaciones:

GOOGLE_GENAI_USE_VERTEXAI=1
GOOGLE_CLOUD_PROJECT=<YOUR_GOOGLE_PROJECT_ID>
GOOGLE_CLOUD_LOCATION=<YOUR_GOOGLE_CLOUD_REGION>

5. Define tu agente

Ahora definamos un sistema multiagente jerárquico.

1. Agente de tendencias en tiempo real: Usa la Búsqueda de Google para encontrar tendencias de moda actuales.
2. Agente de datos de inventario: Usa el conjunto de herramientas de BigQuery para consultar el conjunto de datos públicos thelook_ecommerce en busca de productos disponibles.
3. Agente de asistente de venta minorista (raíz): Organiza el flujo de trabajo solicitando asesoramiento al Agente de tendencias y productos coincidentes al Agente de inventario.

Reemplaza todo el contenido de retail_assistant_app/agent.py por el siguiente código.

import os
import uuid
import asyncio
import google.auth
import dotenv
from google.genai import types
from google.adk.agents import Agent
from google.adk.apps import App
from google.adk.runners import InMemoryRunner
from google.adk.tools import AgentTool, google_search
from google.adk.tools.bigquery import BigQueryCredentialsConfig, BigQueryToolset
from google.adk.plugins.bigquery_agent_analytics_plugin import BigQueryAgentAnalyticsPlugin

dotenv.load_dotenv()

# --- Configuration ---

PROJECT_ID = os.getenv('GOOGLE_CLOUD_PROJECT', 'project_not_set')
DATASET_ID = "adk_logs"
TABLE_ID = "retail_assistant_agent_logs"
APP_NAME = "retail_assistant_agent"
USER_ID = "test_user"

# --- Toolsets ---

credentials, _ = google.auth.default()
credentials_config = BigQueryCredentialsConfig(credentials=credentials)
bigquery_toolset = BigQueryToolset(
 credentials_config=credentials_config
)

# --- Agents ---

# 1. Trend Spotter
real_time_agent = Agent(
   name="real_time_agent",
   model="gemini-2.5-flash",
    description="Researches external factors like weather, local events, and current fashion trends.",
   instruction="""
       You are a real-time research agent.
       Use Google Search to find real-time information relevant to the user's request,
       such as the current weather in their location or trending styles.
   """,
   tools=[google_search]
)

# 2. Inventory Manager
inventory_data_agent = Agent(
   name="inventory_data_agent",
   model="gemini-2.5-flash",
   description="Oversees product inventory in the BigQuery `thelook_ecommerce` dataset to find available items and prices.",
   instruction=f"""
   You manage the inventory. You have access to the `bigquery-public-data.thelook_ecommerce` dataset via the BigQuery toolset.
   Run all BigQuery queries the project id of: '{PROJECT_ID}'
  
   Your workflow:
   1. Look at the products table.
   2. Find items that match the requirements, factor in the results from the trend_setter agent if there are any.
   3. Return with a user friendly response, including the list of specific products and prices.
   """,
   tools=[bigquery_toolset]
)

# 3. Root Orchestrator
root_agent = Agent(
   name="retail_assistant",
   model="gemini-2.5-flash",   
   description="The primary orchestrator, responsible for handling user input, delegating to sub-agents, and synthesizing the final product recommendation.",
   instruction="""
   You are a Retail Assistant.   
   You can ask the 'real_time_agent' agent for any realtime information needed, or style advice, include any information provided by the user.
   You should ask the 'inventory_data_agent' agent to find a maximum of 3 available items matching that style.
   Combine the results into a recommendation.
   """,
   tools=[AgentTool(agent=real_time_agent)],
   sub_agents=[inventory_data_agent]
)

6. Genera registros con el complemento BigQuery Agent Analytics

Ahora, configuremos el complemento BigQuery Agent Analytics para capturar datos de ejecución.

Para ello, crearás una instancia de la clase App. Esta clase funciona como el contenedor de tiempo de ejecución de tu agente, ya que administra el bucle de conversación, controla el estado del usuario y coordina los complementos adjuntos (como nuestro registrador de estadísticas del agente).

El código que se muestra a continuación:

• Inicializa el complemento de Logging: Crea el objeto BigQueryAgentAnalyticsPlugin con los detalles de conexión requeridos.
• Integra el complemento: Pasa el complemento de BigQuery inicializado al constructor de App, lo que garantiza que los eventos de ejecución del agente se capturen y registren automáticamente.
• Ejecuta y registra la ejecución del agente: Ejecuta el flujo de conversación a través de runner.run_async, y el complemento recopila y envía simultáneamente toda la secuencia de eventos a BigQuery antes de cerrar sus recursos.

Copia y pega este código debajo de las definiciones del agente en el archivo agent.py:

async def main(prompt: str):
    """Runs a conversation with the BigQuery agent using the ADK Runner."""
    bq_logger_plugin = BigQueryAgentAnalyticsPlugin(
        project_id=PROJECT_ID, dataset_id=DATASET_ID, table_id=TABLE_ID
    )
    app = App(name=APP_NAME, root_agent=root_agent, plugins=[bq_logger_plugin])
    runner = InMemoryRunner(app=app)

    try:
        session_id = f"{USER_ID}_{uuid.uuid4().hex[:8]}"

        my_session = await runner.session_service.create_session(
            app_name=APP_NAME, user_id=USER_ID, session_id=session_id
        )

        async for event in runner.run_async(
            user_id=USER_ID,
            new_message=types.Content(
                role="user", parts=[types.Part.from_text(text=prompt)]
            ),
            session_id=my_session.id,
        ):
            if event.content.parts and event.content.parts[0].text:
                print(f"** {event.author}: {event.content.parts[0].text}")

    except Exception as e:
        print(f"Error in main: {e}")
    finally:
        print("Closing BQ Plugin...")
        await bq_logger_plugin.close()
        print("BQ Plugin closed.")

if __name__ == "__main__":
   prompts = [
       "what outfits do you have available that are suitable for the weather in london this week?",      
       "You are such a cool agent! I need a gift idea for my friend who likes yoga.",       
       "I'd like to complain - the products sold here are not very good quality!"
   ]
   for prompt, prompt in enumerate(prompts):
       asyncio.run(main(prompt))

Con la instrumentación en su lugar, es hora de ver el agente en acción. Ejecuta la secuencia de comandos para activar el flujo de trabajo de conversación.

python retail_assistant_app/agent.py

Deberías ver al asistente de venta minorista coordinando el flujo de trabajo:

1. Le pide al Agente de tendencias en tiempo real (real_time_agent) que identifique el clima en Londres y busque tendencias de moda adecuadas.
2. Luego, delega en el agente de datos de inventario (inventory_data_agent) para consultar el conjunto de datos de thelook_ecommerce BigQuery en busca de productos específicos que coincidan con esas tendencias.
3. Por último, el Root Orchestrator sintetiza los resultados en una recomendación final.

Mientras tanto, el complemento transmite el registro de ejecución del agente a BigQuery.

7. Analiza los registros de agentes

Uso de herramientas

Ahora podemos ver lo que hizo nuestro agente en segundo plano. Los datos se transmitieron a BigQuery y están listos para el análisis:

1. En la consola de Google Cloud, busca BigQuery.
2. En el panel Explorador, ubica tu proyecto.
3. Expande el conjunto de datos adk_logs.
4. Abre la tabla retail_assitant_agent_logs y haz clic en Consulta.

Para ver qué llamadas a herramientas realizó tu agente y capturar cualquier error de herramientas, ejecuta la siguiente consulta en el editor de BigQuery:

SELECT
   -- Extract text between "Tool Name: " and the next comma (or end of line)
   REGEXP_EXTRACT(content, r'Tool Name: ([^,]+)') AS tool_name,
   -- Count every time a tool finished (successfully or with an error)
   COUNT(*) AS total_finished_runs,
   -- Count it as a failure if it's an explicit system error OR contains "error" in the text
   COUNTIF(event_type = 'TOOL_ERROR' OR REGEXP_CONTAINS(content, r'(?i)\berror\b')) AS failure_count
FROM
   `.adk_logs.retail_assistant_agent_logs`
WHERE
   event_type IN ('TOOL_COMPLETED', 'TOOL_ERROR')
GROUP BY
   1

Haz clic en Visualización para ver esto como un gráfico:

Uso de tokens

Para inferir el costo de tus agentes, puedes agregar los tokens de instrucciones y los tokens de candidatos que consume cada agente distinto:

SELECT
     t.agent,
     SUM(CAST(REGEXP_EXTRACT(t.content, r'prompt:\s*(\d+)') AS INT64)) AS prompt_tokens,
     SUM(CAST(REGEXP_EXTRACT(t.content, r'candidates:\s*(\d+)') AS INT64)) AS candidate_tokens
   FROM
     `adk_logs.retail_assistant_agent_logs` AS t
   WHERE
     t.event_type = 'LLM_RESPONSE'
     AND t.content LIKE '%Token Usage: %'
   GROUP BY 1

Haz clic en Visualización para ver esto como un gráfico:

8. [Bonus] Analiza las opiniones de los usuarios

Ahora, analicemos el sentimiento de la entrada del usuario proporcionada al agente.

1. Crea una conexión de recursos de Cloud para permitir que BigQuery interactúe con los servicios de Vertex AI:

bq mk --connection --location=us \
    --connection_type=CLOUD_RESOURCE test_connection

Deberías ver una respuesta como la siguiente:

Se creó correctamente la conexión 517325854360.us.test_connection

2. Crea una conexión de recurso en la nube:

export SERVICE_ACCOUNT_EMAIL=$(bq show --format=prettyjson --connection us.test_connection | grep "serviceAccountId" | cut -d '"' -f 4)

3. Ejecuta este comando para validar que la cuenta de servicio se haya creado correctamente:

echo $SERVICE_ACCOUNT_EMAIL

Deberías ver tu cuenta de servicio:

4. Otorga a la cuenta de servicio de la conexión al recurso los permisos a nivel del proyecto necesarios para interactuar con Vertex AI:

gcloud projects add-iam-policy-binding $(gcloud config get-value project) \
    --member="serviceAccount:$SERVICE_ACCOUNT_EMAIL" \
    --role='roles/bigquery.connectionUser' \
gcloud projects add-iam-policy-binding $(gcloud config get-value project) \
    --member="serviceAccount:$SERVICE_ACCOUNT_EMAIL" \
    --role='roles/aiplatform.user'

Espera unos minutos y, luego, ejecuta la función AI.SCORE de BigQuery para analizar el sentimiento del usuario:

SELECT
 timestamp,
 user_id,
 content,
 AI.SCORE((
   'What is the sentiment of the user in this text:', content,
   'Use a scale from 1 to 5.'),
 connection_id => 'us.test_connection') AS user_sentiment
FROM
 `adk_logs.retail_assistant_agent_logs`
WHERE
  event_type = 'USER_MESSAGE_RECEIVED'
ORDER BY
 user_sentiment DESC;

La función AI.SCORE asignará un valor de opinión entre 1 y 5 para cada entrada del usuario. Deberías ver resultados como los siguientes:

9. Limpieza

Para evitar que se apliquen cargos a tu cuenta de Google Cloud, borra los recursos que creaste durante este taller.

Borra el conjunto de datos de registro que creó la secuencia de comandos:

bq rm -r -f -d $PROJECT_ID:adk_logs

Para quitar el directorio bigquery-adk-codelab y su contenido, haz lo siguiente:

cd .. 
rm -rf adk-agent-observability

10. Felicitaciones

¡Felicitaciones! Creaste un sistema multiagente con el Kit de desarrollo de agentes (ADK) y, luego, integraste correctamente el complemento BigQuery Agent Analytics para hacer un seguimiento del comportamiento de tu agente y auditarlo.

Documentos de referencia

• Documentación del complemento de registro de BigQuery del ADK
• Formulario de interés del complemento de registro de BigQuery