Sitzungen mit dem Agent Development Kit verwalten

Google Cloud-Projekt

import agentplatform

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

# If you don't have an Agent Runtime instance already, create an instance.
remote_agent = client.runtimes.create(
    config={
      "display_name": "My Session Store",
    }
)

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

Ersetzen Sie Folgendes:

• PROJECT_ID: Ihre Projekt-ID.

• LOCATION: Ihre Region. Unterstützte Regionen für Sessions

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 Agent Runtime-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 remote_agent.api_resource.name.split("/")[-1] abrufen.

ADK-Benutzeroberfläche

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

adk web liest Umgebungsvariablen normalerweise 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)

Python

ADK-Python-Code zum Verwalten von Sitzungen und Status verwenden 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 await-Aufrufe auf oberster Ebene. 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 Agenten zu senden. Sie können eine benutzerdefinierte Sitzungs-ID angeben, um Sitzungen mit Ihrem eigenen Namensschema zu organisieren oder schnell von anderen Diensten zu wechseln. Wenn Sie keine Sitzungs-ID angeben, generiert Agent Platform automatisch eine.

# 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, darf die ID bis zu 63 Zeichen lang sein. Gültige Zeichen sind Kleinbuchstaben, Ziffern und Bindestriche ([a-z0-9-]). Das letzte Zeichen muss ein Buchstabe oder eine Ziffer sein.
• Wenn das erste Zeichen eine Zahl ist, darf 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, verwendet das ADK die Sitzung, 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 (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 Gültigkeitsdauer automatisch gelöscht. Sie können entweder die Ablaufzeit (expire_time) direkt oder die Gültigkeitsdauer (ttl) in Sekunden festlegen. Wenn keines der beiden angegeben ist, wendet das System eine Standard-TTL von 365 Tagen an.

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

  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

Zustände 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

Benutzerdefinierte Felder für Ereignisse

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

Wenn Sie einem Ereignis benutzerdefinierte Felder hinzufügen, serialisiert das ADK die Ereignisdaten im 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, bleiben diese benutzerdefinierten Felder in den abgerufenen Ereignissen erhalten.

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)

Google Cloud-Projekt

client.runtimes.update(
    resource_name=remote_agent.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 nutzerfreundlicher Name für Ihren Agenten.

• REQUIREMENTS: Eine Liste der von Ihrem Agent benötigten pip-Pakete. Beispiel: ["google-cloud-storage", "google-cloud-agentplatform[runtimes,adk]"].

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