Sitzungen mit dem Agent Development Kit verwalten

Auf dieser Seite wird beschrieben, wie Sie einen Agent Development Kit-Agenten (ADK) mit Agent Platform-Sitzungen verbinden und verwaltete Sitzungen in der lokalen und Produktionsumgebung verwenden.

angezeigt.

Hinweis

In dieser Anleitung wird die folgende grundlegende Projektdateistruktur verwendet, um einen ADK-Agenten und den zugehörigen Runner- und Bereitstellungscode zu definieren:

my_agent/
    agent.py      # main agent code
    runner.py     # code for interacting with the agent
    deploy.py     # code for deploying the agent to Google Cloud

Stellen Sie sicher, dass Ihre Umgebung eingerichtet ist, indem Sie die Schritte unter Umgebung einrichten befolgen: Erforderliche Rollen abrufen und Authentifizierung.

Umgebungsvariablen festlegen

Zwecks Verwendung des ADK legen Sie Ihre Umgebungsvariablen fest:

import os

os.environ["GOOGLE_GENAI_USE_ENTERPRISE"] = "TRUE"
os.environ["GOOGLE_CLOUD_PROJECT"] = "PROJECT_ID"
os.environ["GOOGLE_CLOUD_LOCATION"] = "LOCATION"

Ersetzen Sie Folgendes:

  • PROJECT_ID: Ihre Projekt-ID.
  • LOCATION: Ihre Region. Informationen zu den unterstützten Regionen für Memory Bank

Vertex AI Agent Engine-Instanz erstellen

Um auf Agent Platform-Sitzungen zuzugreifen, müssen Sie zuerst eine Vertex AI Agent Engine-Instanz verwenden. Sie müssen keinen Code bereitstellen, um Sitzungen zu verwenden. Wenn Sie Agent Engine bereits verwendet haben, dauert das Erstellen einer Vertex AI Agent Engine-Instanz nur wenige Sekunden, ohne dass Code bereitgestellt werden muss. Wenn Sie Agent Engine zum ersten Mal verwenden, kann es länger dauern.

Google Cloud-Projekt 

import vertexai

client = vertexai.Client(
  project="PROJECT_ID",
  location="LOCATION"
)

# If you don't have an Agent Engine instance already, create an instance.
agent_engine = client.agent_engines.create()

# Print the agent engine ID, you will need it in the later steps to initialize
# the ADK `VertexAiSessionService`.
print(agent_engine.api_resource.name.split("/")[-1])

Ersetzen Sie Folgendes:

ADK-Agenten entwickeln

Folgen Sie der Anleitung unter Agent Development Kit, um Ihren ADK-Agenten zu erstellen, oder verwenden Sie den folgenden Code, um einen Agenten zu erstellen, der einen Nutzer mit festen Begrüßungen begrüßt. Speichern Sie diesen Code in einer Datei mit dem Namen agent.py.

# file: my_agent/agent.py
from google import adk

def greetings(query: str):
  """Tool to greet user."""
  if 'hello' in query.lower():
    return {"greeting": "Hello, world"}
  else:
    return {"greeting": "Goodbye, world"}

# Define an ADK agent
root_agent = adk.Agent(
    model="gemini-2.0-flash",
    name='my_agent',
    instruction="You are an Agent that greet users, always use greetings tool to respond.",
    tools=[greetings]
)

ADK-Runner einrichten

Die ADK-Laufzeit orchestriert die Ausführung Ihrer Agenten, Tools und Callbacks sowie Aufrufe zum Lesen und Schreiben von Sitzungen. Initialisieren Sie den Runner mit VertexAiSessionService, der eine Verbindung zu Agent Platform-Sitzungen herstellt. Speichern Sie diesen Code in einer Datei mit dem Namen runner.py.

Google Cloud-Projekt 

# file: my_agent/runner.py
import agent # Import from your agent.py
from google.adk import Runner
from google.adk.sessions import VertexAiSessionService
from google.genai import types

app_name="APP_NAME"
user_id="USER_ID"

# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
      project="PROJECT_ID",
      location="LOCATION",
      agent_engine_id="AGENT_ENGINE_ID"
)
runner = Runner(
    agent=agent.root_agent,
    app_name=app_name,
    session_service=session_service)

# Helper method to send query to the runner
async def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  async for event in runner.run_async(
      user_id=user_id, session_id=session_id, new_message=content):
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent Response: ", final_response)

Ersetzen Sie Folgendes:

  • APP_NAME: Der Name Ihrer Agentenanwendung.

  • USER_ID: Wählen Sie eine eigene Nutzer-ID mit einer Zeichenbegrenzung von 128. Beispiel: user-123.

  • AGENT_ENGINE_ID: Die Ressourcen-ID einer Vertex AI Agent Engine-Instanz.

  • Bei bereitgestellten Agenten wird die Ressourcen-ID als Umgebungsvariable GOOGLE_CLOUD_AGENT_ENGINE_ID aufgeführt.

  • Bei lokalen Agenten können Sie die Ressourcen-ID mit agent_engine.api_resource.name.split("/")[-1] abrufen.

Mit dem Agent interagieren

Nachdem Sie Ihren Agenten definiert und Agent Platform-Sitzungen eingerichtet haben, können Sie mit Ihrem Agenten interagieren, um zu prüfen, ob der Sitzungsverlauf und die Sitzungsstatus beibehalten werden.

ADK-UI

Wenn Sie Ihren Agenten mit der ADK-Benutzeroberfläche testen und eine Verbindung zu Agent Platform-Sitzungen herstellen möchten, führen Sie den Befehl adk web im Terminal aus.

Normalerweise liest adk web Umgebungsvariablen aus einer .env-Datei. Diese Datei wird jedoch ignoriert, wenn Sie das Flag --session_service_uri verwenden. Stattdessen müssen Sie die Umgebungsvariablen GOOGLE_CLOUD_PROJECT und GOOGLE_CLOUD_LOCATION in Ihrer Terminalsitzung festlegen, bevor Sie den Befehl ausführen.

export GOOGLE_CLOUD_PROJECT="PROJECT_ID"
export GOOGLE_CLOUD_LOCATION="LOCATION"
adk web --session_service_uri=agentengine://AGENT_ENGINE_ID
# Sample output
+-----------------------------------------------------------------------------+
| ADK Web Server started                                                      |
|                                                                             |
| For local testing, access at http://localhost:8000.                         |
+-----------------------------------------------------------------------------+

INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

ADK-Benutzeroberfläche

Python

Verwenden Sie ADK-Python-Code, um Sitzungen und Status zu verwalten. Fügen Sie am Ende der Datei runner.py den folgenden Code hinzu, um mit dem Agenten zu interagieren.

Die folgenden Snippets enthalten aus Gründen der Kürze await-Aufrufe auf oberster Ebene. Wenn Sie diesen Code als Python-Skript ausführen möchten, platzieren Sie die Snippets in einer async-Funktion und führen Sie sie mit asyncio.run() aus, wie in diesem Beispiel gezeigt:

import asyncio

async def main():
  # Place one or more snippets here.
  # For example:
  session = await session_service.create_session(
         app_name=app_name,
         user_id=user_id)

  await call_agent("Hello!", session.id, user_id)

asyncio.run(main())

Sitzung erstellen und Agent abfragen

Verwenden Sie den folgenden Code, um eine Sitzung zu erstellen und eine Abfrage an Ihren Agenten zu senden. Geben Sie eine benutzerdefinierte Sitzungs-ID an, um Sitzungen mit einem eigenen Benennungsschema zu organisieren oder schnell von anderen Diensten zu wechseln. Wenn Sie keine Sitzungs-ID angeben, wird automatisch eine von Agent Platform generiert.

# file: my_agent/runner.py
# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_id,
       session_id=session_id)

await call_agent("Hello!", session.id, user_id)
# Agent response: "Hello, world"

await call_agent("Thanks!", session.id, user_id)
# Agent response: "Goodbye, world"

Beachten Sie bei benutzerdefinierten Sitzungs-IDs die folgenden Einschränkungen, um Konflikte mit vom System generierten IDs zu vermeiden:

  • Wenn das erste Zeichen ein Buchstabe ist, kann die ID bis zu 63 Zeichen lang sein. Gültige Zeichen sind Kleinbuchstaben, Zahlen und Bindestriche ([a-z0-9-]). Das letzte Zeichen muss ein Buchstabe oder eine Zahl sein.
  • Wenn das erste Zeichen eine Zahl ist, kann die ID bis zu 9 Zeichen lang sein. Gültige Zeichen sind Zahlen ([0-9]) ohne führende Nullen.

Nachdem die Sitzung erstellt und an den Runner übergeben wurde, werden in der Sitzung Ereignisse aus der aktuellen Interaktion gespeichert. Sie können auch eine vorherige Sitzung fortsetzen, indem Sie die ID für diese Sitzung angeben.

Gültigkeitsdauer (Time-to-Live, TTL) für Sitzungen konfigurieren

Alle Sitzungen müssen eine Ablaufzeit haben. Sie können diese Ablaufzeit beim Erstellen oder Aktualisieren einer Sitzung definieren. Die Sitzung und ihre untergeordneten Ereignisse werden nach Ablauf der Ablaufzeit automatisch gelöscht. Sie können entweder die Ablaufzeit (expire_time) direkt festlegen oder die Gültigkeitsdauer (ttl) in Sekunden angeben. Wenn nichts angegeben ist, wird standardmäßig eine Gültigkeitsdauer von 365 Tagen angewendet.

Gültigkeitsdauer

Wenn Sie die Gültigkeitsdauer festlegen, berechnet der Server die Ablaufzeit als create_time + ttl für neu erstellte Sitzungen oder update_time + ttl für aktualisierte Sitzungen.

  session = await session_service.create_session(
        app_name=app_name,
        user_id=user_id,
        # Session will be deleted 10 days after creation time.
        ttl=f"{24 * 60 * 60 * 10}s"
  )
  ```

Ablaufzeit 

  import datetime

  expire_time = datetime.datetime.now(
        tz=datetime.timezone.utc) + datetime.timedelta(seconds=24 * 60 * 60 * 10)

  session = await session_service.create_session(
        app_name=app_name,
        user_id=user_id,
        # Session will be deleted at the provided time (10 days after current time).
        expire_time=expire_time.isoformat()
  )

Vorhandene Sitzungen auflisten

Listen Sie alle vorhandenen Sitzungen auf, die mit einer bestimmten Nutzer-ID verknüpft sind.

# List sessions
sessions = await session_service.list_sessions(app_name=app_name,user_id=user_id)
print(sessions)
# ListSessionsResponse(session_ids=['1122334455', '9988776655'])

Sitzungsstatus verwalten

Status enthalten Informationen, die der Agent für eine Unterhaltung benötigt. Sie können beim Erstellen einer Sitzung einen Anfangsstatus als Dictionary angeben:

# Create a session with state
session = await session_service.create_session(
    app_name=app_name,
    user_id=user_id,
    state={'key': 'value'})

print(session.state['key'])
# value

Wenn Sie den Sitzungsstatus außerhalb des Runners aktualisieren möchten, fügen Sie der Sitzung mit state_delta ein neues Ereignis hinzu:

# file: my_agent/runner.py
from google.adk.events import Event, EventActions
import time

# Define state changes
state_changes = {'key': 'new_value'}

# Create event with actions
actions_with_update = EventActions(state_delta=state_changes)
system_event = Event(
invocation_id="invocation_id",
author="system", # Or 'agent', 'tool' etc.
actions=actions_with_update,
timestamp=time.time()
)

# Append the event
await session_service.append_event(session, system_event)

# Check updated state
updated_session = await session_service.get_session(
    app_name=app_name,
    user_id=user_id,
    session_id=session.id)
# State is updated to new value
print(updated_session.state['key'])
# new_value

Benutzerdefinierte Felder für Ereignisse

Das ADK unterstützt ein flexibles Ereignisschema, sodass Sie beliebige Daten in Sitzungsereignisse einfügen können. Das ist nützlich für die Interoperabilität mit anderen Agenten-Frameworks oder zum Speichern benutzerdefinierter Ereignisdaten.

Wenn Sie ein Ereignis mit benutzerdefinierten Feldern anhängen, serialisiert das ADK die Ereignisdaten in das Feld raw_event des gespeicherten Sitzungsereignisses.

# Create an event with custom fields
custom_event = Event(
  invocation_id="invocation_id",
  author="user",
  timestamp=time.time(),
  custom_field="custom_value",
  another_field={"nested_key": "nested_value"}
)

# Append the event
await session_service.append_event(session, custom_event)

Wenn Sie die Sitzung abrufen, behalten die abgerufenen Ereignisse diese benutzerdefinierten Felder bei.

Eine Sitzung löschen

Löschen Sie eine bestimmte Sitzung, die mit einer Nutzer-ID verknüpft ist:

await session_service.delete_session(app_name=app_name, user_id=user_id, session_id=session.id)

Agent in der Agent Platform bereitstellen

Nachdem Sie Ihren Agenten lokal getestet haben, können Sie ihn in der Produktion bereitstellen, indem Sie die Agent Platform-Instanz mit Parametern aktualisieren:

Google Cloud-Projekt

client.agent_engines.update(
    resource_name=agent_engine.api_resource.name,
    agent=AGENT,
    config={
      "display_name": DISPLAY_NAME,      # Optional.
      "requirements": REQUIREMENTS,      # Optional.
      "staging_bucket": STAGING_BUCKET,  # Required.
    },
)

Ersetzen Sie Folgendes:

  • AGENT: Die Anwendung, die die Methode query / stream_query implementiert (z. B. AdkApp für einen ADK-Agenten).

  • DISPLAY_NAME: Ein benutzerfreundlicher Name für Ihren Agenten.

  • REQUIREMENTS: Eine Liste der Pip-Pakete, die von Ihrem Agenten benötigt werden. Beispiel: ["google-cloud-storage", "google-cloud-aiplatform[agent_engines,adk]"].

  • STAGING_BUCKET: Ein Cloud Storage-Bucket mit dem Präfix gs://.

Bereinigen

Wenn Sie alle in diesem Projekt verwendeten Ressourcen bereinigen möchten, können Sie die Vertex AI Agent Engine-Instanz zusammen mit ihren untergeordneten Ressourcen löschen:

agent_engine.delete(force=True)

Nächste Schritte