1. Einführung
In diesem Codelab erfahren Sie, wie Sie mit dem Google Agent Development Kit (ADK) einen komplexen KI-Agenten erstellen. Wir werden einen natürlichen Entwicklungspfad verfolgen, der mit einem grundlegenden Conversational Agent beginnt und nach und nach spezialisierte Funktionen hinzufügt.
Der Agent, den wir erstellen, ist ein persönlicher Assistent, der Sie bei alltäglichen Aufgaben wie der Verwaltung Ihres Kalenders, dem Erinnern an Aufgaben, der Recherche und dem Zusammenstellen von Notizen unterstützt. Er wird von Grund auf mit ADK, Gemini und Vertex AI erstellt.
Am Ende dieses Labs haben Sie einen voll funktionsfähigen Agenten und das Wissen, das Sie benötigen, um ihn an Ihre eigenen Anforderungen anzupassen.
Vorbereitung
- Grundlegende Kenntnisse der Programmiersprache Python
- Grundkenntnisse der Google Cloud Console zum Verwalten von Cloud-Ressourcen
Lerninhalte
- Google Cloud-Infrastruktur für KI-Agents bereitstellen
- Implementierung eines persistenten langfristigen Speichers mit Vertex AI Memory Bank.
- Erstellen einer Hierarchie von spezialisierten Sub-Agents.
- Einbindung externer Datenbanken und des Google Workspace-Ökosystems
Voraussetzungen
Dieser Workshop kann vollständig in der Google Cloud Shell durchgeführt werden, in der alle erforderlichen Abhängigkeiten (gcloud CLI, Code-Editor, Go, Gemini CLI) vorinstalliert sind.
Alternativ benötigen Sie Folgendes, wenn Sie lieber auf Ihrem eigenen Computer arbeiten möchten:
- Python (Version 3.12 oder höher)
- Einen Codeeditor oder eine IDE (z. B. VS Code oder
vim). - Ein Terminal zum Ausführen von Python- und
gcloud-Befehlen. - Empfohlen:Ein Coding-Agent wie die Gemini CLI oder Antigravity
Wichtige Technologien
Hier finden Sie weitere Informationen zu den Technologien, die wir verwenden:
2. Umgebung einrichten
Wählen Sie eine der folgenden Optionen aus: Umgebung selbst einrichten, wenn Sie dieses Codelab auf Ihrem eigenen Computer ausführen möchten, oder Cloud Shell starten, wenn Sie dieses Codelab vollständig in der Cloud ausführen möchten.
Umgebung zum selbstbestimmten Lernen einrichten
- Melden Sie sich in der Google Cloud Console an und erstellen Sie ein neues Projekt oder verwenden Sie ein vorhandenes. Wenn Sie noch kein Gmail- oder Google Workspace-Konto haben, müssen Sie eines erstellen.
- Der Projektname ist der Anzeigename für die Teilnehmer dieses Projekts. Es handelt sich um einen String, der nicht von Google APIs verwendet wird. Sie können sie jederzeit aktualisieren.
- Die Projekt-ID ist für alle Google Cloud-Projekte eindeutig und unveränderlich (kann nach dem Festlegen nicht mehr geändert werden). In der Cloud Console wird automatisch ein eindeutiger String generiert. Normalerweise ist es nicht wichtig, wie dieser String aussieht. In den meisten Codelabs müssen Sie auf Ihre Projekt-ID verweisen (in der Regel als
PROJECT_IDangegeben). Wenn Ihnen die generierte ID nicht gefällt, können Sie eine andere zufällige ID generieren. Alternativ können Sie es mit einem eigenen Namen versuchen und sehen, ob er verfügbar ist. Sie kann nach diesem Schritt nicht mehr geändert werden und bleibt für die Dauer des Projekts bestehen. - Zur Information: Es gibt einen dritten Wert, die Projektnummer, die von einigen APIs verwendet wird. Weitere Informationen zu diesen drei Werten
- Als Nächstes müssen Sie die Abrechnung in der Cloud Console aktivieren, um Cloud-Ressourcen/-APIs zu verwenden. Die Durchführung dieses Codelabs kostet wenig oder gar nichts. Wenn Sie Ressourcen herunterfahren möchten, um Kosten zu vermeiden, die über diese Anleitung hinausgehen, können Sie die erstellten Ressourcen oder das Projekt löschen. Neue Google Cloud-Nutzer können am kostenlosen Testzeitraum mit einem Guthaben von 300$ teilnehmen.
Cloud Shell starten
Während Sie Google Cloud von Ihrem Laptop aus per Fernzugriff nutzen können, wird in diesem Codelab Google Cloud Shell verwendet, eine Befehlszeilenumgebung, die in der Cloud ausgeführt wird.
Klicken Sie in der Google Cloud Console rechts oben in der Symbolleiste auf das Cloud Shell-Symbol:
Die Bereitstellung und Verbindung mit der Umgebung sollte nur wenige Augenblicke dauern. Anschließend sehen Sie in etwa Folgendes:
Diese virtuelle Maschine verfügt über sämtliche Entwicklertools, die Sie benötigen. Sie bietet ein Basisverzeichnis mit 5 GB nichtflüchtigem Speicher und läuft in Google Cloud, was die Netzwerkleistung und Authentifizierung erheblich verbessert. Alle Aufgaben in diesem Codelab können in einem Browser ausgeführt werden. Sie müssen nichts installieren.
3. Projekt einrichten
Bevor wir Code schreiben, müssen wir die erforderliche Infrastruktur und die erforderlichen Berechtigungen in Google Cloud bereitstellen.
Umgebungsvariablen festlegen
Öffnen Sie das Terminal und legen Sie die folgenden Umgebungsvariablen fest:
export PROJECT_ID=`gcloud config get project`
export LOCATION=us-central1
Erforderliche APIs aktivieren
Ihr Agent benötigt Zugriff auf mehrere Google Cloud-Dienste. Führen Sie den folgenden Befehl aus, um sie zu aktivieren:
gcloud services enable \
aiplatform.googleapis.com \
calendar-json.googleapis.com \
sqladmin.googleapis.com
Mit Standardanmeldedaten für Anwendungen authentifizieren
Wir müssen uns mit Standardanmeldedaten für Anwendungen (Application Default Credentials, ADC) authentifizieren, um mit den Google Cloud-Diensten in Ihrer Umgebung zu kommunizieren.
Führen Sie den folgenden Befehl aus, um sicherzustellen, dass Ihre Standardanmeldedaten für Anwendungen aktiv und aktuell sind:
gcloud auth application-default login
4. Basis-Agent erstellen
Als Nächstes müssen wir das Verzeichnis initialisieren, in dem wir den Quellcode des Projekts speichern:
# setup project directory
mkdir -p adk_ea_codelab && cd adk_ea_codelab
# prepare virtual environment
uv init
# install dependencies
uv add google-adk google-api-python-client tzlocal python-dotenv
uv add cloud-sql-python-connector[pg8000] sqlalchemy
Zuerst legen wir die Identität des Agenten und seine grundlegenden Konversationsfunktionen fest. Im ADK definiert die Agent-Klasse die Persona des Agenten und seine Anweisungen.
Jetzt ist ein guter Zeitpunkt, um über einen Namen für den Agent nachzudenken. Ich möchte, dass meine Agents richtige Namen wie Aida oder Sharon haben, da ich denke, dass dies dazu beiträgt, ihnen eine gewisse „Persönlichkeit“ zu verleihen. Sie können den Agent aber auch einfach nach seiner Funktion benennen, z. B. „executive_assistant“, „travel_agent“ oder „code_executor“.
Führen Sie den Befehl adk create aus, um einen Boilerplate-Agenten zu erstellen:
# replace with your desired agent name
uv run adk create executive_assistant
Wählen Sie gemini-2.5-flash als Modell und Vertex AI als Backend aus. Prüfen Sie, ob die vorgeschlagene Projekt-ID mit der übereinstimmt, die Sie für dieses Lab erstellt haben, und drücken Sie die Eingabetaste, um die Auswahl zu bestätigen. Für die Google Cloud-Region können Sie die Standardeinstellung (us-central1) übernehmen. Ihr Terminal sieht dann so aus:
daniela_petruzalek@cloudshell:~/adk_ea_codelab (your-project-id)$ uv run adk create executive_assistant Choose a model for the root agent: 1. gemini-2.5-flash 2. Other models (fill later) Choose model (1, 2): 1 1. Google AI 2. Vertex AI Choose a backend (1, 2): 2 You need an existing Google Cloud account and project, check out this link for details: https://google.github.io/adk-docs/get-started/quickstart/#gemini---google-cloud-vertex-ai Enter Google Cloud project ID [your-project-id]: Enter Google Cloud region [us-central1]: Agent created in /home/daniela_petruzalek/adk_ea_codelab/executive_assistant: - .env - __init__.py - agent.py daniela_petruzalek@cloudshell:~/adk_ea_codelab (your-project-id)$
Nach Abschluss des vorherigen Befehls wird ein Ordner mit dem Namen des Agents (z.B. executive_assistant) mit einigen Dateien erstellt, darunter eine agent.py-Datei mit der grundlegenden Agentdefinition:
from google.adk.agents.llm_agent import Agent
root_agent = Agent(
model='gemini-2.5-flash',
name='root_agent',
description='A helpful assistant for user questions.',
instruction='Answer user questions to the best of your knowledge',
)
Wenn Sie mit diesem Agenten interagieren möchten, können Sie uv run adk web in der Befehlszeile ausführen und die Entwicklungs-UI in Ihrem Browser öffnen. Die Ansicht sieht ungefähr so aus:
$ uv run adk web ... INFO: Started server process [1244] INFO: Waiting for application startup. +-----------------------------------------------------------------------------+ | ADK Web Server started | | | | For local testing, access at http://127.0.0.1:8000. | +-----------------------------------------------------------------------------+ INFO: Application startup complete. INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
Auch wenn dieser KI-Agent recht einfach ist, sollten Sie diese Schritte mindestens einmal durchführen, um sicherzugehen, dass die Einrichtung richtig funktioniert, bevor Sie mit der Bearbeitung des KI-Agenten beginnen. Der Screenshot unten zeigt eine einfache Interaktion über die Entwickler-UI:
Jetzt ändern wir die Agentendefinition mit der Persona unseres Executive Assistant. Kopieren Sie den folgenden Code und ersetzen Sie damit den Inhalt von agent.py. Passen Sie den Namen und die Persona des Agenten an Ihre Vorlieben an.
from google.adk.agents.llm_agent import Agent
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='A professional AI Executive Assistant',
instruction='''
You are an elite, warm, and highly efficient AI partner.
Your primary goal is to help the user manage their tasks, schedule, and research.
Always be direct, concise, and high-signal.
''',
)
Die Eigenschaft „name“ definiert den internen Namen des Agents. In den Anweisungen können Sie ihm jedoch auch einen freundlicheren Namen als Teil seiner Persona für die Interaktionen mit dem Endnutzer geben. Der interne Name wird hauptsächlich für die Beobachtbarkeit und Übergaben in Multi-Agenten-Systemen mit dem Tool transfer_to_agent verwendet. Sie müssen sich nicht selbst darum kümmern, da das ADK das Tool automatisch registriert, wenn Sie einen oder mehrere untergeordnete Vertreter deklarieren.
Verwenden Sie „adk web“, um den gerade erstellten Agenten auszuführen:
uv run adk web
Öffne die ADK-Benutzeroberfläche im Browser und begrüße deinen neuen Assistenten.
5. Persistenten Speicher mit Vertex AI Memory Bank hinzufügen
Ein echter Assistent muss sich an Vorlieben und frühere Interaktionen erinnern, um ein nahtloses, personalisiertes Erlebnis zu bieten. In diesem Schritt integrieren wir Vertex AI Agent Engine Memory Bank, eine Vertex AI-Funktion, die dynamisch Langzeitgedächtnisse auf Grundlage von Nutzerkonversationen generiert.
Mit der Memory Bank kann Ihr KI-Agent personalisierte Informationen erstellen, auf die über mehrere Sitzungen hinweg zugegriffen werden kann. So wird die Kontinuität über Sitzungen hinweg gewährleistet. Im Hintergrund wird die chronologische Reihenfolge der Nachrichten in einer Sitzung verwaltet. Außerdem kann die Ähnlichkeitssuche verwendet werden, um dem Agenten die relevantesten Erinnerungen für den aktuellen Kontext zur Verfügung zu stellen.
Memory Service initialisieren
Das ADK verwendet Vertex AI zum Speichern und Abrufen von Langzeitgedächtnis. Sie müssen eine „Memory Engine“ in Ihrem Projekt initialisieren. Dies ist im Wesentlichen eine Reasoning Engine-Instanz, die als Memory Bank konfiguriert ist.
Erstellen Sie das folgende Skript als setup_memory.py:
setup_memory.py
import vertexai
import os
PROJECT_ID=os.getenv("PROJECT_ID")
LOCATION=os.getenv("LOCATION")
client = vertexai.Client(project=PROJECT_ID, location=LOCATION)
# Create Reasoning Engine for Memory Bank
agent_engine = client.agent_engines.create()
# You will need this resource name to give it to ADK
print(agent_engine.api_resource.name)
Führen Sie nun setup_memory.py aus, um die Reasoning Engine für die Memory Bank bereitzustellen:
uv run python setup_memory.py
Die Ausgabe sollte ungefähr so aussehen:
$ uv run python setup.py projects/1234567890/locations/us-central1/reasoningEngines/1234567890
Speichern Sie den Ressourcennamen der Engine in einer Umgebungsvariablen:
export ENGINE_ID="<insert the resource name above>"
Jetzt müssen wir den Code aktualisieren, damit er den persistenten Speicher verwendet. Ersetzen Sie den Inhalt von agent.py durch Folgendes:
agent.py
from google.adk.agents.llm_agent import Agent
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from google.adk.tools.load_memory_tool import load_memory_tool
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
# Update root_agent with memory tools and callback
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='Executive Assistant with Persistent Memory',
instruction='''
You are an elite AI partner with long-term memory.
Use load_memory to find context about the user when needed.
Always be direct, concise, and high-signal.
''',
tools=[PreloadMemoryTool(), load_memory_tool],
after_agent_callback=auto_save_session_to_memory_callback,
)
Bei PreloadMemoryTool wird automatisch relevanter Kontext aus früheren Unterhaltungen in jede Anfrage eingefügt (mithilfe der Ähnlichkeitssuche), während load_memory_tool es dem Modell ermöglicht, bei Bedarf explizit Fakten aus dem Memory Bank abzurufen. Diese Kombination bietet Ihrem Agenten einen umfassenden, persistenten Kontext.
Wenn Sie Ihren Agenten mit Unterstützung für das Gedächtnis starten möchten, müssen Sie ihm beim Ausführen von „adk web“ memory_service_uri übergeben:
uv run adk web --memory_service_uri="agentengine://$ENGINE_ID"
Geben Sie den Agenten einige Fakten über sich selbst und fragen Sie dann in einer anderen Sitzung danach. Sagen Sie zum Beispiel Ihren Namen:
Sie können die vom KI-Agenten gespeicherten gemerkten Informationen in der Cloud Console ansehen. Rufen Sie die Produktseite für „Agent Engine“ auf (verwenden Sie die Suchleiste).
Klicken Sie dann auf den Namen Ihrer Agent Engine (achten Sie darauf, dass Sie die richtige Region auswählen):
Rufen Sie dann den Tab „Erinnerungen“ auf:
Es sollten einige Erinnerungen hinzugefügt worden sein.
6. Webrecherchefunktionen hinzufügen
Um hochwertige Informationen zu liefern, muss unser KI-Assistent gründliche Recherchen durchführen, die über eine einzelne Suchanfrage hinausgehen. Durch die Übertragung der Recherche an einen spezialisierten untergeordneten Agenten bleibt die Reaktionsfähigkeit der Hauptpersona erhalten, während der Researcher im Hintergrund komplexe Datenerhebungen durchführt.
In diesem Schritt implementieren wir einen LoopAgent, um „Recherchetiefe“ zu erreichen. So kann der Agent iterativ suchen, Ergebnisse auswerten und seine Anfragen verfeinern, bis er ein vollständiges Bild hat. Wir legen auch Wert auf technische Genauigkeit, indem wir Inline-Zitationen für alle Ergebnisse verlangen. So wird sichergestellt, dass jede Behauptung durch einen Quelllink belegt ist.
Research Specialist erstellen (research.py)
Hier definieren wir einen Basis-Agenten, der mit dem Google-Suchtool ausgestattet ist, und umschließen ihn mit einem LoopAgent. Der Parameter „max_iterations“ fungiert als Governor und sorgt dafür, dass der Agent die Suche bis zu dreimal wiederholt, wenn Lücken in seinem Verständnis bestehen.
research.py
from google.adk.agents.llm_agent import Agent
from google.adk.agents.loop_agent import LoopAgent
from google.adk.tools.google_search_tool import GoogleSearchTool
from google.adk.tools.tool_context import ToolContext
def exit_loop(tool_context: ToolContext):
"""Call this function ONLY when no further research is needed, signaling the iterative process should end."""
print(f" [Tool Call] exit_loop triggered by {tool_context.agent_name}")
tool_context.actions.escalate = True
# Return empty dict as tools should typically return JSON-serializable output
return {}
# --- RESEARCH LOGIC ---
_research_worker = Agent(
model='gemini-2.5-flash',
name='research_worker',
description='Worker agent that performs a single research step.',
instruction='''
Use google_search to find facts and synthesize them for the user.
Critically evaluate your findings. If the data is incomplete or you need more context, prepare to search again in the next iteration.
You must include the links you found as references in your response, formatting them like citations in a research paper (e.g., [1], [2]).
Use the exit_loop tool to terminate the research early if no further research is needed.
If you need to ask the user for clarifications, call the exit_loop function early to interrupt the research cycle.
''',
tools=[GoogleSearchTool(bypass_multi_tools_limit=True), exit_loop],
)
# The LoopAgent iterates the worker up to 3 times for deeper research
research_agent = LoopAgent(
name='research_specialist',
description='Deep web research specialist.',
sub_agents=[_research_worker],
max_iterations=3,
)
Root-Agent (agent.py) aktualisieren
Importieren Sie den research_agent und fügen Sie ihn Sharon als Tool hinzu:
agent.py
from google.adk.agents.llm_agent import Agent
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from google.adk.tools.load_memory_tool import load_memory_tool
# Import our new sub agent
from .research import research_agent
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
# Update root_agent with memory tools and callback
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='Executive Assistant with persistent memory and research capabilities',
instruction='''
You are an elite AI partner with long-term memory.
1. Use load_memory to recall facts.
2. Delegate research tasks to the research_specialist.
Always be direct, concise, and high-signal.
''',
tools=[PreloadMemoryTool(), load_memory_tool],
sub_agents=[research_agent],
after_agent_callback=auto_save_session_to_memory_callback,
)
Starten Sie ADK Web noch einmal, um den Forschungs-KI-Agenten zu testen.
uv run adk web --memory_service_uri="agentengine://$ENGINE_ID"
Geben Sie ihm eine einfache Rechercheaufgabe, z. B. „Wie schreibt man einen guten Tech-Blog?“
Vielleicht haben Sie bemerkt, dass sich der KI-Agent meinen Namen gemerkt hat, obwohl es sich um eine neue Sitzung handelt. Beachten Sie auch den Tool-Aufruf „transfer_to_agent“: Mit diesem Tool wird die Aufgabe an unseren neuen Research-Agent übergeben.
Kommen wir nun zur Aufgabenverwaltung.
7. Aufgabenverwaltung mit Cloud SQL hinzufügen
Der Agent hat zwar ein Langzeitgedächtnis, es ist aber nicht für detaillierte, strukturierte Daten wie eine Aufgabenliste geeignet. Für Aufgaben verwenden wir eine herkömmliche relationale Datenbank. Wir verwenden SQLAlchemy und eine Google Cloud SQL-Datenbank (PostgreSQL). Bevor wir den Code schreiben können, müssen wir die Infrastruktur bereitstellen.
Infrastruktur bereitstellen
Führen Sie diese Befehle aus, um die Datenbank zu erstellen. Hinweis:Das Erstellen der Instanz dauert etwa 5–10 Minuten. Sie können mit dem nächsten Schritt fortfahren, während dieser Vorgang im Hintergrund ausgeführt wird.
# 1. Define instance variables
export INSTANCE_NAME="assistant-db"
export USER_EMAIL=$(gcloud config get-value account)
# 2. Create the Cloud SQL instance
gcloud sql instances create $INSTANCE_NAME \
--database-version=POSTGRES_18 \
--tier=db-f1-micro \
--region=us-central1 \
--edition=ENTERPRISE
# 3. Create the database for our tasks
gcloud sql databases create tasks --instance=$INSTANCE_NAME
Die Bereitstellung der Datenbankinstanz dauert einige Minuten. Vielleicht ist jetzt ein guter Zeitpunkt, um eine Tasse Kaffee oder Tee zu trinken oder den Code zu aktualisieren, während Sie warten. Vergessen Sie aber nicht, zurückzukehren und die Zugriffssteuerung abzuschließen.
Zugriffssteuerung konfigurieren
Jetzt müssen wir Ihr Nutzerkonto so konfigurieren, dass es Zugriff auf die Datenbank hat. Führen Sie die folgenden Befehle im Terminal aus:
# change this to your favorite password
export DB_PASS="correct-horse-battery-staple"
# Create a regular database user
gcloud sql users create assistant_user \
--instance=$INSTANCE_NAME \
--password=$DB_PASS
Umgebungskonfiguration aktualisieren
Das ADK lädt die Konfiguration zur Laufzeit aus einer .env-Datei. Aktualisieren Sie die Umgebung Ihres KI-Agenten mit den Details zur Datenbankverbindung.
# Retrieve the unique connection name
export DB_CONN=$(gcloud sql instances describe $INSTANCE_NAME --format='value(connectionName)')
# Append configuration to your .env file
cat <<EOF >> executive_assistant/.env
DB_CONNECTION_NAME=$DB_CONN
DB_USER=assistant_user
DB_PASSWORD=$DB_PASS
DB_NAME=tasks
EOF
Fahren wir nun mit den Codeänderungen fort.
To-do-Spezialisten erstellen (todo.py)
Ähnlich wie beim Research-Agenten erstellen wir den To-do-Spezialisten in einer eigenen Datei. todo.py erstellen:
todo.py
import os
import uuid
import sqlalchemy
from datetime import datetime
from typing import Optional, List
from sqlalchemy import (
Column,
String,
DateTime,
Enum,
select,
delete,
update,
)
from sqlalchemy.orm import declarative_base, Session
from google.cloud.sql.connector import Connector
from google.adk.agents.llm_agent import Agent
# --- DATABASE LOGIC ---
Base = declarative_base()
connector = Connector()
def getconn():
db_connection_name = os.environ.get("DB_CONNECTION_NAME")
db_user = os.environ.get("DB_USER")
db_password = os.environ.get("DB_PASSWORD")
db_name = os.environ.get("DB_NAME", "tasks")
return connector.connect(
db_connection_name,
"pg8000",
user=db_user,
password=db_password,
db=db_name,
)
engine = sqlalchemy.create_engine(
"postgresql+pg8000://",
creator=getconn,
)
class Todo(Base):
__tablename__ = "todos"
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
title = Column(String(255), nullable=False)
priority = Column(
Enum("high", "medium", "low", name="priority_levels"), nullable=False, default="medium"
)
due_date = Column(DateTime, nullable=True)
status = Column(Enum("pending", "done", name="status_levels"), default="pending")
created_at = Column(DateTime, default=datetime.utcnow)
def init_db():
"""Builds the table if it's missing."""
Base.metadata.create_all(bind=engine)
def add_todo(
title: str, priority: str = "medium", due_date: Optional[str] = None
) -> dict:
"""
Adds a new task to the list.
Args:
title (str): The description of the task.
priority (str): The urgency level. Must be one of: 'high', 'medium', 'low'.
due_date (str, optional): The due date in ISO format (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS).
Returns:
dict: A dictionary containing the new task's ID and a status message.
"""
init_db()
with Session(engine) as session:
due = datetime.fromisoformat(due_date) if due_date else None
item = Todo(
title=title,
priority=priority.lower(),
due_date=due,
)
session.add(item)
session.commit()
return {"id": item.id, "status": f"Task added ✅"}
def list_todos(status: str = "pending") -> list:
"""
Lists tasks from the database, optionally filtering by status.
Args:
status (str, optional): The status to filter by. 'pending', 'done', or 'all'.
"""
init_db()
with Session(engine) as session:
query = select(Todo)
s_lower = status.lower()
if s_lower != "all":
query = query.where(Todo.status == s_lower)
query = query.order_by(Todo.priority, Todo.created_at)
results = session.execute(query).scalars().all()
return [
{
"id": t.id,
"task": t.title,
"priority": t.priority,
"status": t.status,
}
for t in results
]
def complete_todo(task_id: str) -> str:
"""Marks a specific task as 'done'."""
init_db()
with Session(engine) as session:
session.execute(update(Todo).where(Todo.id == task_id).values(status="done"))
session.commit()
return f"Task {task_id} marked as done."
def delete_todo(task_id: str) -> str:
"""Permanently removes a task from the database."""
init_db()
with Session(engine) as session:
session.execute(delete(Todo).where(Todo.id == task_id))
session.commit()
return f"Task {task_id} deleted."
# --- TODO SPECIALIST AGENT ---
todo_agent = Agent(
model='gemini-2.5-flash',
name='todo_specialist',
description='A specialist agent that manages a structured SQL task list.',
instruction='''
You manage the user's task list using a PostgreSQL database.
- Use add_todo when the user wants to remember something. If no priority is mentioned, mark it as 'medium'.
- Use list_todos to show tasks.
- Use complete_todo to mark a task as finished.
- Use delete_todo to remove a task entirely.
When marking a task as complete or deleting it, if the user doesn't provide the ID,
use list_todos first to find the correct ID for the task they described.
''',
tools=[add_todo, list_todos, complete_todo, delete_todo],
)
Der obige Code ist für zwei Hauptaufgaben zuständig: die Verbindung zur Cloud SQL-Datenbank und die Bereitstellung einer Liste von Tools für alle gängigen Aufgabenlistenvorgänge, einschließlich Hinzufügen, Entfernen und Markieren als erledigt.
Da diese Logik sehr spezifisch für den Aufgaben-Agent ist und wir diese detaillierte Verwaltung aus Sicht des Executive Assistant (Root-Agent) nicht unbedingt benötigen, wird dieser Agent als „AgentTool“ anstelle eines untergeordneten Agents verpackt.
Bei der Entscheidung, ob Sie einen AgentTool oder einen Sub-Agent verwenden möchten, sollten Sie berücksichtigen, ob Kontext geteilt werden muss:
- Verwenden Sie ein AgentTool, wenn Ihr Agent keinen Kontext mit dem Root-Agent teilen muss.
- Sie verwenden einen Sub-Agent, wenn Ihr Agent Kontext mit dem Root-Agent teilen soll.
Bei einem Recherche-Agenten kann es nützlich sein, Kontext zu teilen, bei einem einfachen To-do-Agenten ist das jedoch nicht der Fall.
Implementieren wir AgentTool in agent.py.
Root-Agent (agent.py) aktualisieren
Importieren Sie nun den todo_agent in Ihre Hauptdatei und hängen Sie ihn als Tool an:
agent.py
import os
from datetime import datetime
from google.adk.agents.llm_agent import Agent
from google.adk.tools.agent_tool import AgentTool
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from google.adk.tools.load_memory_tool import load_memory_tool
# Import our specialized sub-agents
from .research import research_agent
from .todo import todo_agent
# Callback for persistent memory storage
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
# --- ROOT AGENT DEFINITION ---
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='A professional AI Executive Assistant with memory and specialized tools.',
instruction='''
You are an elite, high-signal AI Executive Assistant.
Your goal is to help the user manage their knowledge, tasks, and research.
## Your Capabilities:
1. Memory: Use load_memory to recall personal facts or past context about the user.
2. Research: Delegate complex web-based investigations to the research_specialist.
3. Tasks: Delegate all to-do list management (adding, listing, or completing tasks) to the todo_specialist.
Always be direct and professional. If a task is successful, provide a brief confirmation.
''',
tools=[
PreloadMemoryTool(),
load_memory_tool,
AgentTool(todo_agent) # Exposes the Todo Specialist as a tool
],
sub_agents=[research_agent], # Exposes the Research Specialist for direct handover
after_agent_callback=auto_save_session_to_memory_callback,
)
Führen Sie adk web noch einmal aus, um die neue Funktion zu testen:
uv run adk web --memory_service_uri="agentengine://$ENGINE_ID"
So erstellen Sie eine Aufgabe:
8. Kalenderverwaltung hinzufügen
Schließlich werden wir Google Kalender einbinden, damit der Agent Termine verwalten kann. In diesem Codelab erstellen wir einen unabhängigen Kalender, den der Agent verwalten kann, anstatt dem Agenten Zugriff auf Ihren persönlichen Kalender zu gewähren. Das könnte nämlich gefährlich sein, wenn es nicht richtig gemacht wird.
Zuerst erstellen wir ein dediziertes Dienstkonto, das als Identität des Agents fungiert. Anschließend erstellen wir den Kalender des Kundenservicemitarbeiters programmatisch über das Dienstkonto.
Dienstkonto bereitstellen
Öffnen Sie das Terminal und führen Sie die folgenden Befehle aus, um die Identität zu erstellen und Ihrem persönlichen Konto die Berechtigung zu erteilen, sie zu übernehmen:
export SA_NAME="ea-agent"
export SA_EMAIL="${SA_NAME}@${PROJECT_ID}.iam.gserviceaccount.com"
# Create the service account
gcloud iam service-accounts create $SA_NAME \
--display-name="Executive Assistant Agent"
# Allow your local user to impersonate it
gcloud iam service-accounts add-iam-policy-binding $SA_EMAIL \
--member="user:$(gcloud config get-value account)" \
--role="roles/iam.serviceAccountTokenCreator"
# Save it to the agent's environment
echo "SERVICE_ACCOUNT_EMAIL=$SA_EMAIL" >> executive_assistant/.env
Kalender programmatisch erstellen
Wir schreiben ein Skript, mit dem das Dienstkonto den Kalender erstellen soll. Erstellen Sie im Stammverzeichnis Ihres Projekts (neben setup_memory.py) eine neue Datei mit dem Namen setup_calendar.py:
setup_calendar.py
import os
import google.auth
from googleapiclient.discovery import build
from google.auth.transport.requests import Request
from google.auth import impersonated_credentials
from dotenv import load_dotenv
load_dotenv('executive_assistant/.env')
SA_EMAIL = os.environ.get("SERVICE_ACCOUNT_EMAIL")
def setup_sa_calendar():
print(f"Authenticating to impersonate {SA_EMAIL}...")
# 1. Base credentials
creds, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
creds.refresh(Request())
# 2. Impersonate the Service Account
impersonated = impersonated_credentials.Credentials(
source_credentials=creds,
target_principal=SA_EMAIL,
target_scopes=["https://www.googleapis.com/auth/calendar"],
)
service = build("calendar", "v3", credentials=impersonated)
# 3. Create the calendar
print("Creating independent Service Account calendar...")
calendar = service.calendars().insert(body={
"summary": "AI Assistant (SA Owned)",
"description": "An independent calendar managed purely by the AI."
}).execute()
calendar_id = calendar['id']
# 4. Save the ID
with open("executive_assistant/.env", "a") as f:
f.write(f"\nCALENDAR_ID={calendar_id}\n")
print(f"Setup complete! CALENDAR_ID {calendar_id} added to .env")
if __name__ == "__main__":
setup_sa_calendar()
Führen Sie das Skript über Ihr Terminal aus:
uv run python setup_calendar.py
Calendar Specialist erstellen (calendar.py)
Konzentrieren wir uns nun auf den Kalenderspezialisten. Wir werden diesen Agent mit einer vollständigen Reihe von Kalendertools ausstatten: Auflisten, Erstellen, Aktualisieren, Löschen und sogar eine „Schnell hinzufügen“-Funktion, die natürliche Sprache versteht.
Kopieren Sie den folgenden Code in die Datei calendar.py.
calendar.py
import os
from datetime import datetime, timedelta, timezone
import google.auth
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google.adk.agents.llm_agent import Agent
def _get_calendar_service():
"""Build the Google Calendar API service using Service Account Impersonation."""
from google.auth.transport.requests import Request
from google.auth import impersonated_credentials
target_principal = os.environ.get("SERVICE_ACCOUNT_EMAIL")
if not target_principal:
raise ValueError("SERVICE_ACCOUNT_EMAIL environment variable is missing.")
base_scopes = ["https://www.googleapis.com/auth/cloud-platform"]
creds, _ = google.auth.default(scopes=base_scopes)
creds.refresh(Request())
target_scopes = ["https://www.googleapis.com/auth/calendar"]
impersonated = impersonated_credentials.Credentials(
source_credentials=creds,
target_principal=target_principal,
target_scopes=target_scopes,
)
return build("calendar", "v3", credentials=impersonated)
def _format_event(event: dict) -> dict:
"""Format a raw Calendar API event into a clean dict for the LLM."""
start = event.get("start", {})
end = event.get("end", {})
return {
"id": event.get("id"),
"title": event.get("summary", "(No title)"),
"start": start.get("dateTime", start.get("date")),
"end": end.get("dateTime", end.get("date")),
"location": event.get("location", ""),
"description": event.get("description", ""),
"attendees": [
{"email": a["email"], "status": a.get("responseStatus", "unknown")}
for a in event.get("attendees", [])
],
"link": event.get("htmlLink", ""),
"conference_link": (
event.get("conferenceData", {}).get("entryPoints", [{}])[0].get("uri", "")
if event.get("conferenceData")
else ""
),
"status": event.get("status", ""),
}
def list_events(days_ahead: int = 7) -> dict:
"""List upcoming calendar events."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
now = datetime.now(timezone.utc).isoformat()
end = (datetime.now(timezone.utc) + timedelta(days=days_ahead)).isoformat()
events_result = service.events().list(
calendarId=calendar_id, timeMin=now, timeMax=end,
maxResults=50, singleEvents=True, orderBy="startTime"
).execute()
events = events_result.get("items", [])
if not events:
return {"status": "success", "count": 0, "events": []}
return {"status": "success", "count": len(events), "events": [_format_event(e) for e in events]}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
def create_event(title: str, start_time: str, end_time: str, description: str = "", location: str = "", attendees: str = "", add_google_meet: bool = False) -> dict:
"""Create a new calendar event."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
event_body = {
"summary": title,
"start": {"dateTime": start_time},
"end": {"dateTime": end_time},
}
if description: event_body["description"] = description
if location: event_body["location"] = location
if attendees:
email_list = [e.strip() for e in attendees.split(",") if e.strip()]
event_body["attendees"] = [{"email": e} for e in email_list]
conference_version = 0
if add_google_meet:
event_body["conferenceData"] = {
"createRequest": {"requestId": f"event-{datetime.now().strftime('%Y%m%d%H%M%S')}", "conferenceSolutionKey": {"type": "hangoutsMeet"}}
}
conference_version = 1
event = service.events().insert(calendarId=calendar_id, body=event_body, conferenceDataVersion=conference_version).execute()
return {"status": "success", "message": f"Event created ✅", "event": _format_event(event)}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
def update_event(event_id: str, title: str = "", start_time: str = "", end_time: str = "", description: str = "") -> dict:
"""Update an existing calendar event."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
patch_body = {}
if title: patch_body["summary"] = title
if start_time: patch_body["start"] = {"dateTime": start_time}
if end_time: patch_body["end"] = {"dateTime": end_time}
if description: patch_body["description"] = description
if not patch_body: return {"status": "error", "message": "No fields to update."}
event = service.events().patch(calendarId=calendar_id, eventId=event_id, body=patch_body).execute()
return {"status": "success", "message": "Event updated ✅", "event": _format_event(event)}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
def delete_event(event_id: str) -> dict:
"""Delete a calendar event by its ID."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
service.events().delete(calendarId=calendar_id, eventId=event_id).execute()
return {"status": "success", "message": f"Event '{event_id}' deleted ✅"}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
def quick_add_event(text: str) -> dict:
"""Create an event using natural language (e.g. 'Lunch with Sarah next Monday noon')."""
calendar_id = os.environ.get("CALENDAR_ID")
try:
service = _get_calendar_service()
event = service.events().quickAdd(calendarId=calendar_id, text=text).execute()
return {"status": "success", "message": "Event created from text ✅", "event": _format_event(event)}
except HttpError as e:
return {"status": "error", "message": f"Calendar API error: {e}"}
calendar_agent = Agent(
model='gemini-2.5-flash',
name='calendar_specialist',
description='Manages the user schedule and calendar events.',
instruction='''
You manage the user's Google Calendar.
- Use list_events to check the schedule.
- Use quick_add_event for simple, conversational scheduling requests (e.g., "Lunch tomorrow at noon").
- Use create_event for complex meetings that require attendees, specific durations, or Google Meet links.
- Use update_event to change details of an existing event.
- Use delete_event to cancel or remove an event.
CRITICAL: For update_event and delete_event, you must provide the exact `event_id`.
If the user does not provide the ID, you MUST call list_events first to find the correct `event_id` before attempting the update or deletion.
Always use the current date/time context provided by the root agent to resolve relative dates like "tomorrow".
''',
tools=[list_events, create_event, update_event, delete_event, quick_add_event],
)
Root-Agent fertigstellen (agent.py)
Aktualisieren Sie Ihre agent.py-Datei mit dem folgenden Code:
agent.py
import os
from datetime import datetime
from zoneinfo import ZoneInfo
from google.adk.agents.llm_agent import Agent
from google.adk.tools.agent_tool import AgentTool
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from google.adk.tools.load_memory_tool import load_memory_tool
# Import all our specialized sub-agents
from .research import research_agent
from .todo import todo_agent
from .calendar import calendar_agent
import tzlocal
# Automatically detect the local system timezone
TIMEZONE = tzlocal.get_localzone_name()
# Callback for persistent memory storage
async def auto_save_session_to_memory_callback(callback_context):
await callback_context._invocation_context.memory_service.add_session_to_memory(
callback_context._invocation_context.session)
# Callback to inject the current time into the prompt
async def setup_agent_context(callback_context, **kwargs):
now = datetime.now(ZoneInfo(TIMEZONE))
callback_context.state["current_time"] = now.strftime("%A, %Y-%m-%d %I:%M %p")
callback_context.state["timezone"] = TIMEZONE
# --- ROOT AGENT DEFINITION ---
root_agent = Agent(
model='gemini-2.5-flash',
name='executive_assistant',
description='A professional AI Executive Assistant with memory and specialized tools.',
instruction='''
You are an elite, high-signal AI Executive Assistant.
Your goal is to help the user manage their knowledge, tasks, research, and schedule.
## Your Capabilities:
1. Memory: Use load_memory to recall personal facts.
2. Research: Delegate complex web investigations to the research_specialist.
3. Tasks: Delegate all to-do list management to the todo_specialist.
4. Scheduling: Delegate all calendar queries to the calendar_specialist.
## 🕒 Current State
- Time: {current_time?}
- Timezone: {timezone?}
Always be direct and professional.
''',
tools=[
PreloadMemoryTool(),
load_memory_tool,
AgentTool(todo_agent),
AgentTool(calendar_agent)
],
sub_agents=[research_agent],
before_agent_callback=[setup_agent_context],
after_agent_callback=[auto_save_session_to_memory_callback],
)
Neben dem Kalendertool haben wir auch eine neue Funktion Callback vor dem Kundenservicemitarbeiter hinzugefügt: setup_agent_context. Durch diese Funktion kennt der Agent das aktuelle Datum, die aktuelle Uhrzeit und die aktuelle Zeitzone, sodass er den Kalender effizienter nutzen kann. Dazu werden Sitzungsstatusvariablen festgelegt, eine andere Art von Agent-Speicher, die für die kurzfristige Persistenz konzipiert ist.
Führen Sie „adk web“ ein letztes Mal aus, um den vollständigen Agenten zu testen.
uv run adk web --memory_service_uri="agentengine://$ENGINE_ID"
Sie können den Sitzungsstatus auf dem Tab „Status“ in der Entwickler-UI prüfen:
Sie haben jetzt einen Agent, der Termine, To-do-Listen und Recherchen im Blick behalten kann und ein Langzeitgedächtnis hat.
Nach dem Lab bereinigen
9. Fazit
Glückwunsch! Sie haben einen multifunktionalen KI-Assistenten für Führungskräfte in fünf Entwicklungsphasen erfolgreich konzipiert.
Worüber haben wir gesprochen?
- Infrastruktur für KI-Agents bereitstellen
- Persistenten Speicher und spezialisierte Sub-Agents mit ADK-Build-ins implementieren.
- Einbindung externer Datenbanken und Produktivitäts-APIs
Nächste Schritte
Sie können Ihre Lernreise fortsetzen, indem Sie sich andere Codelabs auf dieser Plattform ansehen oder den Executive Assistant selbst verbessern.
Wenn Sie Ideen für Verbesserungen benötigen, können Sie Folgendes ausprobieren:
- Ereigniskompaktierung implementieren, um die Leistung bei langen Unterhaltungen zu optimieren
- Fügen Sie einen Artefaktdienst hinzu, damit der Agent Notizen für Sie erstellen und als Dateien speichern kann.
- Stellen Sie Ihren Agent als Backend-Dienst mit Google Cloud Run bereit.
Denken Sie daran, die Umgebung nach dem Testen zu bereinigen, um unerwartete Gebühren für Ihr Abrechnungskonto zu vermeiden.
Viel Spaß beim Programmieren!