Gestire le sessioni con Agent Development Kit

Questa pagina descrive come connettere un agente dell'Agent Development Kit (ADK) con le sessioni di Vertex AI Agent Engine e utilizzare le sessioni gestite nell'ambiente locale e di produzione.

Prima di iniziare

Assicurati che l'ambiente sia configurato seguendo i passaggi Ottieni i ruoli richiesti e Autenticazione in Configura l'ambiente.

Crea un'istanza di Vertex AI Agent Engine

Per accedere alle sessioni di Vertex AI Agent Engine, devi prima utilizzare un'istanza di Vertex AI Agent Engine. Per iniziare a utilizzare le sessioni, non devi eseguire il deployment di alcun codice. Se hai già utilizzato Agent Engine, la creazione di un'istanza di Vertex AI Agent Engine richiede solo pochi secondi senza il deployment del codice. Potrebbe essere necessario più tempo se è la prima volta che utilizzi Agent Engine.

Progetto Google Cloud 

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

Sostituisci quanto segue:

  • PROJECT_ID: il tuo ID progetto.

  • LOCATION: La tua regione. Consulta le regioni supportate per le sessioni.

Modalità Express di Vertex AI

Puoi utilizzare l'SDK Vertex AI per Python con la chiave API in modalità express:

import vertexai

client = vertexai.Client(  # For service interactions via client.agent_engines
    api_key="API_KEY",
)

# 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])

Sviluppare l'agente ADK

Per creare l'agente ADK, segui le istruzioni riportate in Agent Development Kit o utilizza il seguente codice per creare un agente che saluti un utente con saluti fissi:

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

Configurare lo strumento di esecuzione ADK

ADK Runtime orchestra l'esecuzione di agenti, strumenti e callback e orchestra le chiamate per leggere e scrivere sessioni. Inizializza Runner con VertexAiSessionService, che si connette alle sessioni di Vertex AI Agent Engine.

Progetto Google Cloud 

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_ID",
      "LOCATION",
      "AGENT_ENGINE_ID"
)
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)

Sostituisci quanto segue:

  • APP_NAME: il nome dell'applicazione agente.

  • USER_ID: scegli il tuo ID utente con un limite di 128 caratteri. Ad esempio, user-123.

  • AGENT_ENGINE_ID: l'ID risorsa di un'istanza di Vertex AI Agent Engine.

  • Per gli agenti di cui è stato eseguito il deployment, l'ID risorsa è elencato come variabile di ambiente GOOGLE_CLOUD_AGENT_ENGINE_ID

  • Per gli agenti locali, puoi recuperare l'ID risorsa utilizzando agent_engine.api_resource.name.split("/")[-1].

Modalità Express di Vertex AI 

import os
from google.adk.sessions import VertexAiSessionService
from google.genai import types

# Set environment variables
os.environ["GOOGLE_API_KEY"] = "API_KEY"
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"

# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
      agent_engine_id="AGENT_ENGINE_ID"
)
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)

Sostituisci quanto segue:

  • API_KEY: la chiave API di Vertex AI Express Mode.

  • AGENT_ENGINE_ID: l'ID risorsa di un'istanza di Vertex AI Agent Engine.

  • Per gli agenti di cui è stato eseguito il deployment, l'ID risorsa è elencato come variabile di ambiente GOOGLE_CLOUD_AGENT_ENGINE_ID

  • Per gli agenti locali, puoi recuperare l'ID risorsa utilizzando agent_engine.api_resource.name.split("/")[-1].

Interagire con l'agente

Dopo aver definito l'agente e configurato le sessioni di Vertex AI Agent Engine, puoi interagire con l'agente per verificare che la cronologia e gli stati delle sessioni vengano mantenuti.

UI ADK

Testa l'agente con l'interfaccia utente dell'ADK e connettiti a Vertex AI Agent Engine Session utilizzando l'opzione della riga di comando session_service_uri:

agent_engine_id="AGENT_ENGINE_ID"

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)

UI ADK

Python

Utilizza il codice Python dell'ADK per gestire sessioni e stati.

Crea una sessione ed esegui query sull'agente

Utilizza il seguente codice per creare una sessione e inviare una query all'agente:

# 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"

Dopo che la sessione è stata creata e trasmessa al runner, ADK la utilizza per archiviare gli eventi dell'interazione corrente. Puoi anche riprendere una sessione precedente fornendo il relativo ID.

Elenca le sessioni esistenti

Elenca tutte le sessioni esistenti associate a un determinato ID utente.

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

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

Gestire gli stati della sessione

Gli stati contengono le informazioni di cui l'agente ha bisogno per una conversazione. Puoi fornire uno stato iniziale come dizionario quando crei una sessione:

# 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

Per aggiornare lo stato della sessione al di fuori del runner, aggiungi un nuovo evento alla sessione utilizzando state_delta:

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

Eliminare una sessione

Elimina una sessione specifica associata a un User-ID:

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

Esegui il deployment dell'agente in Vertex AI Agent Engine

Dopo aver testato l'agente localmente, puoi eseguirne il deployment in produzione aggiornando l'istanza di Vertex AI Agent Engine con i parametri:

Progetto Google Cloud

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.
    },
)

Sostituisci quanto segue:

  • AGENT: l'applicazione che implementa il metodo query / stream_query (ad esempio, AdkApp per un agente ADK). Per ulteriori informazioni, vedi Considerazioni sul deployment.

  • DISPLAY_NAME: un nome intuitivo per l'agente.

  • REQUIREMENTS: un elenco di pacchetti pip richiesti dall'agente. Ad esempio, ["google-cloud-storage", "google-cloud-aiplatform[agent_engines,adk]"].

  • STAGING_BUCKET: un bucket Cloud Storage con il prefisso gs://.

Modalità Express di Vertex AI

Esegui il deployment dell'agente tramite i file di origine seguendo questi passaggi di deployment.

Esegui la pulizia

Per eliminare tutte le risorse utilizzate in questo progetto, puoi eliminare l'istanza di Vertex AI Agent Engine insieme alle relative risorse secondarie:

agent_engine.delete(force=True)

Passaggi successivi