Sitzungen mit dem Agent-Entwicklungs-Kit verwalten

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

Hinweise

In dieser Anleitung wird die folgende grundlegende Projektdateistruktur zum Definieren eines ADK-Agents und des zugehörigen Runner- und Bereitstellungscodes verwendet:

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

Achten Sie darauf, dass Ihre Umgebung eingerichtet ist. Folgen Sie dazu der Anleitung unter Umgebung einrichten in den Schritten Erforderliche Rollen abrufen und Authentifizierung.

Umgebungsvariablen festlegen

Zwecks Verwendung des ADK legen Sie Ihre Umgebungsvariablen fest:

import os

os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "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. Unterstützte Regionen für Memory Bank

Vertex AI Agent Engine-Instanz erstellen

Um auf Vertex AI Agent Engine-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 und es ist keine Codebereitstellung erforderlich. 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:

PROJECT_ID: Ihre Projekt-ID.
LOCATION: Ihre Region. Unterstützte Regionen für Sessions

ADK-Agent entwickeln

Folgen Sie der Anleitung im Agent Development Kit, um Ihren ADK-Agenten zu erstellen. Alternativ können Sie den folgenden Code verwenden, 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 Agents, Tools und Callbacks sowie Aufrufe zum Lesen und Schreiben von Sitzungen. Initialisieren Sie den Runner mit VertexAiSessionService, um eine Verbindung zu Vertex AI Agent Engine-Sitzungen herzustellen. 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 Agent-Anwendung.
USER_ID: Wählen Sie eine eigene Nutzer-ID mit einem Zeichenlimit von 128 Zeichen aus. Beispiel: user-123.
AGENT_ENGINE_ID: Die Ressourcen-ID einer Vertex AI Agent Engine-Instanz.
Bei bereitgestellten Agents wird die Ressourcen-ID als Umgebungsvariable GOOGLE_CLOUD_AGENT_ENGINE_ID aufgeführt.
Für lokale Agents 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 Vertex AI Agent Engine-Sitzungen eingerichtet haben, können Sie mit Ihrem Agenten interagieren, um zu prüfen, ob der Sitzungsverlauf und die Status beibehalten werden.

ADK-Benutzeroberfläche

Testen Sie Ihren Agenten mit der ADK-Benutzeroberfläche und stellen Sie mit der Befehlszeilenoption session_service_uri eine Verbindung zur Vertex AI Agent Engine-Sitzung her. Wenn Sie session_service_uri verwenden, müssen Sie auch die Umgebungsvariablen GOOGLE_CLOUD_PROJECT und GOOGLE_CLOUD_LOCATION festlegen oder eine .env-Datei im übergeordneten Verzeichnis verwenden, das Ihren Agentenordner enthält. Wenn sich Ihr Agent beispielsweise in agents/my_agent/ befindet, sollte sich die Datei .env im Ordner agents befinden und Sie sollten adk web im Ordner agents ausführen.

project_id=PROJECT_ID
location=LOCATION
agent_engine_id="AGENT_ENGINE_ID"

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 ein, um mit dem Agenten zu interagieren.

Die folgenden Snippets enthalten aus Gründen der Übersichtlichkeit Aufrufe der obersten Ebene await. Wenn Sie diesen Code als Python-Skript ausführen möchten, fügen Sie die Snippets in eine async-Funktion ein und verwenden Sie asyncio.run(), um sie auszuführen, 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 Anfrage an Ihren Agent zu senden:

# file: my_agent/runner.py
# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_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"

Nachdem die Sitzung erstellt und an den Runner übergeben wurde, verwendet das ADK sie, um Ereignisse aus der aktuellen Interaktion zu speichern. Sie können auch eine vorherige Sitzung fortsetzen, indem Sie die ID für diese Sitzung angeben.

Gültigkeitsdauer von 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 Gültigkeitsdauer automatisch gelöscht. Sie können entweder die Ablaufzeit (expire_time) direkt oder die Gültigkeitsdauer (ttl) in Sekunden festlegen. Wenn keines angegeben ist, wendet das System eine Standard-TTL von 365 Tagen an.

Gültigkeitsdauer

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

  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

Listet 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 Anfangszustand 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, hängen Sie der Sitzung mit state_delta ein neues Ereignis an:

# 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

Sitzung löschen

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

KI-Agent in Vertex AI Agent Engine bereitstellen

Nachdem Sie Ihren Agenten lokal getestet haben, können Sie ihn in der Produktion bereitstellen, indem Sie die Vertex AI Agent Engine-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). Weitere Informationen finden Sie unter Überlegungen zur Bereitstellung.
DISPLAY_NAME: Ein nutzerfreundlicher Name für Ihren KI-Agenten.
REQUIREMENTS: Eine Liste der von Ihrem Agent benötigten pip-Pakete. 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 den untergeordneten Ressourcen löschen:

agent_engine.delete(force=True)

Nächste Schritte

Sitzungen mit API-Aufrufen verwalten

Sitzungen mit dem Agent-Entwicklungs-Kit verwalten Mit Sammlungen den Überblick behalten Sie können Inhalte basierend auf Ihren Einstellungen speichern und kategorisieren.

Hinweise

Umgebungsvariablen festlegen

Vertex AI Agent Engine-Instanz erstellen

Google Cloud-Projekt

ADK-Agent entwickeln

ADK-Runner einrichten

Google Cloud-Projekt

Mit dem Agent interagieren

ADK-Benutzeroberfläche

Python

Sitzung erstellen und Agent abfragen

Gültigkeitsdauer von Sitzungen konfigurieren

Gültigkeitsdauer

Ablaufzeit

Vorhandene Sitzungen auflisten

Sitzungsstatus verwalten

Sitzung löschen

KI-Agent in Vertex AI Agent Engine bereitstellen

Google Cloud-Projekt

Bereinigen

Nächste Schritte

Sitzungen mit dem Agent-Entwicklungs-Kit verwalten