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

1. Einführung

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

Aufgaben

• Mit dem ADK einen Multi-Agenten-Einzelhandelsassistenten erstellen
• Initialisieren Sie das BigQuery Agent Analytics-Plug-in, um Ablaufverfolgungsdaten zur Ausführung dieses Agents in BigQuery zu erfassen und zu speichern.
• Agent-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 ein kostenloses Guthaben in Höhe 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 Befehlszeile in Google Cloud Shell und Python-Code für die ADK-Entwicklung. Sie müssen kein Python-Experte sein, aber ein grundlegendes Verständnis dafür, wie man Code liest, wird Ihnen helfen, die Konzepte zu verstehen.

2. Hinweis

Google Cloud-Projekt erstellen

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

2. Die Abrechnung für das Cloud-Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist.

Cloud Shell starten

Cloud Shell ist eine Befehlszeilenumgebung, die in Google Cloud ausgeführt wird und mit den erforderlichen Tools vorinstalliert ist.

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

2. Sobald Sie mit Cloud Shell verbunden sind, führen Sie diesen Befehl aus, um Ihre Authentifizierung in Cloud Shell zu überprüfen:

gcloud auth list

3. 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

4. Wenn Ihr Projekt nicht wie erwartet konfiguriert ist, verwenden Sie den folgenden Befehl, um es festzulegen:

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

APIs aktivieren

1. 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

2. Bei erfolgreicher Ausführung des Befehls sollte eine Meldung wie die folgende angezeigt werden:

Operation "operations/..." finished successfully.

3. Installation und Einrichtung

Kehren Sie zur Cloud Shell zurück und achten Sie darauf, dass Sie sich in Ihrem Basisverzeichnis befinden.

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

bq mk --dataset --location=US adk_logs

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

1. Öffnen Sie einen neuen Terminaltab in Cloud Shell 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

2. Erstellen Sie eine virtuelle Python-Umgebung:

python -m venv .venv

3. Aktivieren Sie die virtuelle Umgebung:

source .venv/bin/activate

4. Installieren Sie das ADK:

pip install --upgrade google-adk

4. ADK-Anwendung erstellen

Erstellen wir nun unseren Einzelhandelsassistenten-Agenten. Dieser KI‑Agent wird für Folgendes entwickelt:

1. Führen Sie den Befehl zum Erstellen des ADK-Dienstprogramms aus, um eine neue Agent-Anwendung mit den erforderlichen Ordnern und Dateien zu erstellen:

adk create retail_assistant_app

Folgen Sie der Anleitung:

1. Wählen Sie gemini-3.5-flash als Modell aus.
2. Wählen Sie Vertex AI als Backend aus.
3. Bestätigen Sie Ihre Standard-Google Cloud-Projekt-ID und -Region.

Unten sehen Sie ein Beispiel für eine Interaktion:

2. 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:

Sehen Sie sich die generierten Dateien an:

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

• init.py::Kennzeichnet den Ordner als Python-Modul.
• agent.py::Enthält die ursprüngliche Definition des KI-Agenten.
• .env:Möglicherweise müssen Sie auf „Ansicht“ > „Versteckte Dateien einblenden“ klicken, um diese Datei zu sehen.

• Die .env-Datei enthält Umgebungsvariablen für Ihr Projekt. Aktualisieren Sie alle Variablen, die nicht korrekt über die Prompts festgelegt wurden. Achten Sie darauf, dass Ihr Standort auf einen globalen oder multiregionalen Standort (z.B. us oder eu) festgelegt ist, da dies die derzeit von gemini-3.5-flash unterstützten Regionstypen sind.

GOOGLE_GENAI_USE_VERTEXAI=1
GOOGLE_CLOUD_PROJECT=<YOUR_GOOGLE_PROJECT_ID>
GOOGLE_CLOUD_LOCATION=us

5. KI-Agenten definieren

Definieren wir nun ein hierarchisches Multi-Agenten-System.

1. Echtzeit-Trend-Agent:Nutzt die Google Suche, um aktuelle Modetrends zu finden.
2. Inventory Data Agent:Verwendet das BigQuery-Toolset, um das öffentliche Dataset „thelook_ecommerce“ nach verfügbaren Produkten abzufragen.
3. Retail Assistant (Root) Agent:Orchestriert den Workflow, indem er den Trend Agent um Rat und den Inventory Agent um passende Produkte bittet.

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
import warnings
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.integrations.bigquery import BigQueryCredentialsConfig, BigQueryToolset
from google.adk.plugins.bigquery_agent_analytics_plugin import BigQueryAgentAnalyticsPlugin
from google.adk.agents.context_cache_config import ContextCacheConfig

dotenv.load_dotenv()

# Suppress experimental and AFC warnings
warnings.filterwarnings("ignore", message=".*\\[EXPERIMENTAL\\].*")
warnings.filterwarnings("ignore", message=".*Direct use of automatic function calling.*")

# --- 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-3.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-3.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-3.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 alle angehängten Plug-ins (z. B. unseren Agenten-Analysen-Logger).

Der folgende Code:

• Initialisiert das Logging-Plug-in:Erstellt das BigQueryAgentAnalyticsPlugin mit den erforderlichen Verbindungsdetails.
• Plug-in einbinden:Das initialisierte BigQuery-Plug-in wird an den App-Konstruktor übergeben. So wird dafür gesorgt, dass Ereignisse zur Agent-Ausführung automatisch erfasst und protokolliert werden.
• Agent-Ausführung ausführen und protokollieren:Der Konversationsablauf wird über runner.run_async ausgeführt. Gleichzeitig werden mit dem Plug-in die gesamte Ereignissequenz erfasst und an BigQuery gesendet, bevor die Ressourcen geschlossen werden.

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

async def main(prompt: str):
    print(f"\n{'='*60}")
    print(f"👤 USER: {prompt}")
    print(f"{'='*60}")

    print("📊 [Analytics]: Initializing BigQuery Agent Analytics Plugin...")
    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],
        context_cache_config=ContextCacheConfig()
    )
    runner = InMemoryRunner(app=app)

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

        print(f"⚙️  [System]: Initializing session {session_id}...")
        print("📊 [Analytics]: Streaming execution trace to BigQuery...")

        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:
                for part in event.content.parts:
                    if part.text:
                        print(f"\n🤖 [{event.author}]: {part.text.strip()}")
                    elif part.function_call:
                        print(f"\n🛠️  [{event.author}] used tool: {part.function_call.name}")

    except Exception as e:
        print(f"\n❌ Error in main: {e}")
    finally:
        print("\n📊 [Analytics]: Closing BigQuery Plugin and flushing logs...")
        await bq_logger_plugin.close()
        print("📊 [Analytics]: BQ Plugin closed successfully.")
        print(f"✅ [System]: Finished processing prompt.")

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())

Bevor Sie den Agent ausführen, müssen Sie dafür sorgen, dass die erforderlichen Abhängigkeiten in Ihrer virtuellen Umgebung installiert sind. Führen Sie dazu Folgendes aus:

pip install "google-adk[bigquery-analytics]"

Nachdem die Instrumentierung eingerichtet ist, ist es an der Zeit, den Agenten in Aktion zu sehen. Führen Sie das Skript aus, um den Unterhaltungs-Workflow auszulösen.

python retail_assistant_app/agent.py

Der Einzelhandelsassistent sollte den Workflow orchestrieren:

1. Er bittet den Real Time Trend Agent (real_time_agent), das Wetter in London zu ermitteln und nach passenden Modetrends zu suchen.
2. Anschließend wird der Inventory Data Agent (inventory_data_agent) aufgerufen, um das thelook_ecommerce-BigQuery-Dataset nach bestimmten Produkten abzufragen, die diesen Trends entsprechen.
3. Schließlich fasst der Root Orchestrator die Ergebnisse in einer endgültigen Empfehlung zusammen.

Gleichzeitig streamt das Plug-in den Ausführungs-Trace des Agenten an BigQuery.

7. Agent-Logs analysieren

Tool-Nutzung

Wir können jetzt sehen, was unser Agent hinter den Kulissen gemacht hat! Die Daten wurden in BigQuery gestreamt und können jetzt analysiert werden:

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

Wenn Sie sehen möchten, welche Tool-Aufrufe Ihr Agent ausgeführt hat, und Tool-Fehler erfassen möchten, führen Sie die folgende Abfrage im BigQuery-Editor aus:

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 KI-Agenten zu ermitteln, können Sie die von jedem einzelnen Agenten verbrauchten Prompt-Tokens 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 Agent bereitgestellt wurde.

1. 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 die folgende sehen:

Connection 517325854360.us.test_connection successfully created

2. Cloud-Ressourcenverbindung erstellen:

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

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

echo $SERVICE_ACCOUNT_EMAIL

Ihr Dienstkonto sollte nun angezeigt werden:

4. 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;

Mit der Funktion AI.SCORE wird jeder Nutzereingabe ein Sentimentwert zwischen 1 und 5 zugewiesen. Die Ergebnisse sollten in etwa so aussehen:

9. Bereinigen

Löschen Sie die in diesem Workshop erstellten Ressourcen, um laufende Gebühren für Ihr Google Cloud-Konto zu vermeiden.

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 dessen 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.

Referenzdokumente

• Dokumentation zum ADK BigQuery Logging Plugin
• Interessenbekundung für das BigQuery Logging-Plug-in
• BigQuery Agent Analytics SDK