ADK-Agentenausführung mit dem BigQuery Agent Analytics-Plug-in analysieren

1. Einführung

In diesem Codelab erstellen Sie mit dem Agent Development Kit (ADK) ein Multi-Agenten-System und aktivieren die Agentenbeobachtbarkeit mit dem BigQuery Agent Analytics-Plug-in.Sie stellen dem Agenten eine Reihe von Fragen und analysieren dann mit BigQuery Konversations-Traces und die Nutzung von Agenten-Tools.

Aufgaben

Mit dem ADK einen Multi-Agenten-Einzelhandelsassistenten erstellen
Das BigQuery Agent Analytics-Plug-in initialisieren, um Trace-Daten zur Ausführung dieses Agenten zu erfassen und in BigQuery zu speichern
Die Agenten-Logdaten in BigQuery analysieren

Voraussetzungen

Ein Webbrowser wie Chrome
Ein Google Cloud-Projekt mit aktivierter Abrechnung oder
ein Gmail-Konto. Im nächsten Abschnitt erfahren Sie, wie Sie einen kostenlosen Gutschein im Wert von 5 $für dieses Codelab einlösen und ein neues Projekt einrichten.

Dieses Codelab richtet sich an Entwickler aller Erfahrungsstufen, auch an Anfänger. Sie verwenden die Befehlszeilenschnittstelle in Google Cloud Shell und Python-Code für die ADK-Entwicklung. Sie müssen kein Python-Experte sein, aber ein grundlegendes Verständnis des Lesens von Code hilft Ihnen, die Konzepte zu verstehen.

2. Hinweis

Google Cloud-Projekt erstellen

Wählen Sie in der Google Cloud Console auf der Seite der Projektauswahl ein Google Cloud-Projekt aus oder erstellen Sie eines.

Die Abrechnung für das Cloud-Projekt muss aktiviert sein. Informationen zum Prüfen, ob die Abrechnung für ein Projekt aktiviert ist Informationen zum Prüfen, ob die Abrechnung für ein Projekt aktiviert ist.

Cloud Shell starten

Die Cloud Shell ist eine Befehlszeilenumgebung, die in Google Cloud ausgeführt wird und in der die erforderlichen Tools vorinstalliert sind.

Klicken Sie oben in der Google Cloud Console auf Cloud Shell aktivieren:

Führen Sie nach der Verbindung mit Cloud Shell diesen Befehl aus, um Ihre Authentifizierung in Cloud Shell zu prüfen:

gcloud auth list

Führen Sie den folgenden Befehl aus, um zu bestätigen, dass Ihr Projekt für die Verwendung mit gcloud konfiguriert ist:

gcloud config get project

Wenn Ihr Projekt nicht wie erwartet konfiguriert ist, legen Sie es mit dem folgenden Befehl fest:

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

APIs aktivieren

Führen Sie diesen Befehl aus, um alle erforderlichen APIs und Dienste zu aktivieren:

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

Bei erfolgreicher Ausführung des Befehls sollte eine Meldung ähnlich der folgenden angezeigt werden:

Operation "operations/..." finished successfully.

3. Installation und Einrichtung

Kehren Sie zu Cloud Shell zurück und prüfen Sie, ob Sie sich in Ihrem Basisverzeichnis befinden.

Führen Sie in Cloud Shell den folgenden Befehl aus, um in BigQuery ein neues Dataset mit dem Namen „adk_logs“ zu erstellen:

bq mk --dataset --location=US adk_logs

Erstellen wir nun eine virtuelle Python-Umgebung und installieren die erforderlichen Pakete.

Öffnen Sie in Cloud Shell einen neuen Terminaltab und führen Sie diesen Befehl aus, um einen Ordner mit dem Namen adk-agent-observability zu erstellen und zu öffnen:

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

Erstellen Sie eine virtuelle Python-Umgebung:

python -m venv .venv

Aktivieren Sie die virtuelle Umgebung:

source .venv/bin/activate

Installieren Sie das ADK:

pip install --upgrade google-adk

4. ADK-Anwendung erstellen

Erstellen wir nun unseren Einzelhandelsassistenten. Dieser Agent soll Folgendes tun:

Führen Sie den Befehl des Dienstprogramms „adk create“ aus, um eine neue Agentenanwendung mit den erforderlichen Ordnern und Dateien zu erstellen:

adk create retail_assistant_app

Führen Sie die angezeigten Schritte aus:

Wählen Sie gemini-2.5-flash für das Modell aus.
Wählen Sie Vertex AI für das Back-End aus.
Bestätigen Sie Ihre Standardprojekt-ID und -region für Google Cloud.

Unten sehen Sie eine Beispielinteraktion:

Klicken Sie in Cloud Shell auf die Schaltfläche „Editor öffnen“ , um den Cloud Shell-Editor zu öffnen und die neu erstellten Ordner und Dateien anzusehen:

Beachten Sie die generierten Dateien:

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

init.py::Markiert den Ordner als Python-Modul.
agent.py::Enthält die anfängliche Agentendefinition.
.env:Möglicherweise müssen Sie auf „Ansicht“ > „Versteckte Dateien einblenden“ klicken, um diese Datei zu sehen.

Die Datei „.env“ enthält Umgebungsvariablen für Ihr Projekt. Aktualisieren Sie alle Variablen, die in den Eingabeaufforderungen nicht korrekt festgelegt wurden:

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

5. Agent definieren

Definieren wir nun ein hierarchisches Multi-Agenten-System.

Real-Time Trend Agent:Verwendet die Google Suche, um aktuelle Modetrends zu finden.
Inventory Data Agent:Verwendet das BigQuery-Toolset, um das öffentliche Dataset „thelook_ecommerce“ nach verfügbaren Produkten zu durchsuchen.
Retail assistant (Root) Agent:Orchestriert den Workflow, indem er den Trend Agent um Rat fragt und den Inventory Agent nach passenden Produkten sucht.

Ersetzen Sie den gesamten Inhalt von retail_assistant_app/agent.py durch den folgenden Code.

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. Logs mit dem BigQuery Agent Analytics-Plug-in generieren

Konfigurieren wir nun das BigQuery Agent Analytics-Plug-in, um Ausführungsdaten zu erfassen.

Dazu erstellen Sie eine Instanz der Klasse App. Diese Klasse dient als Laufzeitcontainer für Ihren Agenten. Sie verwaltet die Konversationsschleife, den Nutzerstatus und orchestriert alle angehängten Plug-ins (z. B. unseren Agenten-Analytics-Logger).

Der Code unten:

Initialisiert das Logging-Plug-in:Erstellt das BigQueryAgentAnalyticsPlugin mit den erforderlichen Verbindungsdetails.
Integriert das Plug-in:Übergibt das initialisierte BigQuery-Plug-in an den App-Konstruktor, sodass Ereignisse der Agentenausführung automatisch erfasst und protokolliert werden.
Führt die Agentenausführung aus und protokolliert sie:Führt den Konversationsablauf über runner.run_async aus, wobei das Plug-in gleichzeitig die gesamte Ereignissequenz erfasst und an BigQuery sendet, bevor es seine Ressourcen schließt.

Kopieren Sie diesen Code und fügen Sie ihn unter den Agentendefinitionen in der Datei „ agent.py“ ein:

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.")

async def run_all_prompts():
    """Runs all prompts in a single event loop."""
    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 in prompts:
        await main(prompt)

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

Nachdem die Instrumentierung eingerichtet ist, können wir den Agenten in Aktion sehen. Führen Sie das Skript aus, um den Konversationsworkflow auszulösen.

python retail_assistant_app/agent.py

Sie sollten sehen, wie der Einzelhandelsassistent den Workflow orchestriert:

Er fordert den Real-Time Trend Agent (real_time_agent) auf, das Wetter in London zu ermitteln und nach passenden Modetrends zu suchen.
Anschließend delegiert er an den Inventory Data Agent (inventory_data_agent) , um das BigQuery-Dataset thelook_ecommerce nach bestimmten Produkten zu durchsuchen, die zu diesen Trends passen.
Schließlich fasst der Root Orchestrator die Ergebnisse zu einer endgültigen Empfehlung zusammen.

Währenddessen streamt das Plug-in den Ausführungs-Trace des Agenten an BigQuery.

7. Agenten-Logs analysieren

Tool-Nutzung

Wir können jetzt sehen, was unser Agent im Hintergrund getan hat. Die Daten wurden an BigQuery gestreamt und können analysiert werden:

Suchen Sie in der Google Cloud Console nach BigQuery.
Suchen Sie im Bereich Explorer nach Ihrem Projekt.
Maximieren Sie das Dataset adk_logs.
Öffnen Sie die Tabelle retail_assitant_agent_logs und klicken Sie auf Abfrage.

Führen Sie im BigQuery-Editor die folgende Abfrage aus, um zu sehen, welche Tool-Aufrufe Ihr Agent ausgeführt hat, und um Tool-Fehler zu erfassen:

SELECT
  -- Extract the tool name directly from the JSON key "tool"
  JSON_VALUE(content, '$.tool') AS tool_name,
  -- Count every time a tool finished (successfully or with an error)
  COUNT(*) AS total_finished_runs,
  -- Count as a failure if  event_type is ERROR, result object contains a status of 'ERROR', or error_details exist
  COUNTIF(
    event_type = 'TOOL_ERROR' OR
    JSON_VALUE(content, '$.result.status') = 'ERROR' OR
    JSON_VALUE(content, '$.result.error_details') IS NOT NULL
  ) AS failure_count
FROM
  `adk_logs.retail_assistant_agent_logs`
WHERE
  event_type IN ('TOOL_COMPLETED', 'TOOL_ERROR')
GROUP BY
  1;

Klicken Sie auf „Visualisierung“, um die Ergebnisse als Diagramm anzusehen (Ihre Ergebnisse können abweichen):

Tokennutzung

Um die Kosten Ihrer Agenten zu ermitteln, können Sie die von den einzelnen Agenten verbrauchten Prompt- und Kandidaten-Tokens zusammenfassen:

SELECT
 t.agent,
 SUM(LAX_INT64(t.content.usage.prompt)) AS prompt_tokens,
 SUM(LAX_INT64(t.content.usage.completion)) AS completion_tokens
FROM
 `adk_logs.retail_assistant_agent_logs` AS t
WHERE
 t.event_type = 'LLM_RESPONSE'
 -- Filter for records that actually contain usage metadata
 AND t.content.usage IS NOT NULL
GROUP BY 1;

Klicken Sie auf „Visualisierung“, um die Ergebnisse als Diagramm anzusehen (Ihre Ergebnisse können abweichen):

8. [Bonus] Nutzerstimmung analysieren

Analysieren wir nun die Stimmung der Nutzereingabe, die dem Agenten zur Verfügung gestellt wurde.

Erstellen Sie in Cloud Shell eine Cloud-Ressourcenverbindung, damit BigQuery mit Vertex AI-Diensten interagieren kann:

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

Sie sollten eine Antwort wie diese sehen:

Connection 517325854360.us.test_connection successfully created

Erstellen Sie eine Cloud-Ressourcenverbindung:

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

Führen Sie diesen Befehl aus, um zu prüfen, ob das Dienstkonto erfolgreich erstellt wurde:

echo $SERVICE_ACCOUNT_EMAIL

Ihr Dienstkonto sollte angezeigt werden:

Erteilen Sie dem Dienstkonto der Ressourcenverbindung die Berechtigungen auf Projektebene, die für die Interaktion mit Vertex AI erforderlich sind:

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'

Es kann einige Minuten dauern, bis die Berechtigungen übernommen werden. Kehren Sie dann zu BigQuery zurück und führen Sie die folgende Abfrage mit der Funktion AI.SCORE aus, um die Nutzerstimmung zu analysieren:

SELECT
 timestamp,
 user_id,
 content,
 AI.SCORE((
   'What is the sentiment of the user in this text:', JSON_VALUE(content.text_summary),
   '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;

Die Funktion AI.SCORE weist jeder Nutzereingabe einen Stimmungswert zwischen 1 und 5 zu. Die Ergebnisse sollten in etwa so aussehen:

9. Bereinigen

Damit Ihrem Google Cloud-Konto keine laufenden Kosten entstehen, löschen Sie die Ressourcen, die während dieses Workshops erstellt wurden.

Löschen Sie das vom Skript erstellte Logging-Dataset:

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

Löschen Sie die Cloud-Ressourcenverbindung:

bq rm --connection --project_id=$PROJECT_ID --location=us test_connection

So entfernen Sie das Verzeichnis „bigquery-adk-codelab“ und seinen Inhalt:

cd .. 
rm -rf adk-agent-observability

10. Glückwunsch

Glückwunsch! Sie haben mit dem Agent Development Kit (ADK) ein Multi-Agenten-System erstellt und das BigQuery Agent Analytics-Plug-in erfolgreich integriert, um das Verhalten Ihres Agenten zu verfolgen und zu prüfen.