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 Umgebung und in der Produktionsumgebung verwenden.

Hinweise

Prüfen Sie, ob Ihre Umgebung eingerichtet ist. Folgen Sie dazu der Anleitung unter Umgebung einrichten in den Schritten Erforderliche Rollen abrufen und Authentifizierung.

Vertex AI Agent Engine-Instanz erstellen

Um auf Vertex AI Agent Engine-Sitzungen zuzugreifen, müssen Sie zuerst eine Vertex AI Agent Engine-Instanz erstellen. Sie müssen keinen Code bereitstellen, um Sitzungen zu verwenden. Ohne Codebereitstellung dauert das Erstellen einer Vertex AI Agent Engine-Instanz nur wenige Sekunden.

import vertexai
from vertexai import agent_engines

# Create an agent engine instance
agent_engine = agent_engines.create()

ADK-Agent entwickeln

Folgen Sie der Anleitung im Agent Development Kit, um Ihren ADK-Agent zu erstellen. Alternativ können Sie den folgenden Code verwenden, um einen Agent zu erstellen, der einen Nutzer mit festen Begrüßungen begrüßt:

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.

from google.adk.sessions import VertexAiSessionService

app_name="AGENT_ENGINE_ID"
user_id="USER_ID"

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

# Helper method to send query to the runner
def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  events = runner.run(
      user_id=user_id, session_id=session_id, new_message=content)

  for event in events:
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent Response: ", final_response)

Ersetzen Sie Folgendes:

• PROJECT_ID: Ihre Projekt-ID.

• LOCATION: Ihre Region.

• 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.name.split("/")[-1] abrufen.

• USER_ID: Eine nicht leere eindeutige Kennung für den Nutzer mit einer maximalen Länge von 128 Zeichen.

Mit dem Agent interagieren

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

agent_engine_id="AGENT_ENGINE_ID"

adk web --session_db_url=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)

# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_id)

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

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

# List sessions
await session_service.list_sessions(app_name=app_name,user_id=user_id)

# ListSessionsResponse(session_ids=['1122334455', '9988776655'])

# 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

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

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

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:

agent_engines.update(resource_name=agent_engine.name, agent_engine=AGENT, requirements=REQUIREMENTS)

Ersetzen Sie Folgendes:

• AGENT: Die Anwendung, die die Methode query / stream_query implementiert (z. B. AdkApp für einen ADK-Agent). Weitere Informationen finden Sie unter Überlegungen zur Bereitstellung.

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