Gestire le sessioni con Agent Development Kit

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

Prima di iniziare

Queste istruzioni utilizzano la seguente struttura di base dei file di progetto per definire un agente ADK e il relativo codice di runner e deployment di supporto:

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

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

Imposta le variabili di ambiente

Per utilizzare l'ADK, imposta le variabili di ambiente:

import os

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

Sostituisci quanto segue:

  • PROJECT_ID: il tuo ID progetto.
  • LOCATION: la tua regione. Consulta le regioni supportate per Memory Bank.

Crea un'istanza di Vertex AI Agent Engine

Per accedere alle sessioni di Agent Platform, devi prima utilizzare un'istanza di Vertex AI Agent Engine. Non è necessario eseguire il deployment di alcun codice per iniziare a utilizzare le sessioni. 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 richiedere 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 Sessioni.

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. Salva questo codice in un file denominato 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]
)

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 a Agent Platform Sessions. Salva questo codice in un file denominato runner.py.

Progetto Google Cloud 

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

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

Interagire con l'agente

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

UI ADK

Per testare l'agente con l'interfaccia utente dell'ADK e connetterti a Agent Platform Sessions, esegui il comando adk web nel terminale.

Anche se adk web in genere legge le variabili di ambiente da un file .env, ignora questo file quando utilizzi il flag --session_service_uri. Devi invece impostare le variabili di ambiente GOOGLE_CLOUD_PROJECT e GOOGLE_CLOUD_LOCATION nella sessione del terminale prima di eseguire il comando.

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)

UI ADK

Python

Utilizza il codice Python dell'ADK per gestire sessioni e stati. Aggiungi il seguente codice alla fine del file runner.py per interagire con l'agente.

Per brevità, i seguenti snippet contengono chiamate await di primo livello. Per eseguire questo codice come script Python, inserisci gli snippet all'interno di una funzione async e utilizza asyncio.run() per eseguirlo, come mostrato in questo esempio:

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

Crea una sessione ed esegui query sull'agente

Utilizza il seguente codice per creare una sessione e inviare una query all'agente. Specifica un ID sessione personalizzato per organizzare le sessioni utilizzando il tuo schema di denominazione o per eseguire rapidamente la transizione da altri servizi. Se non specifichi un ID sessione, Agent Platform ne genera automaticamente uno.

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

Per gli ID sessione personalizzati, tieni presenti le seguenti limitazioni per evitare collisioni con gli ID generati dal sistema:

  • Se il primo carattere è una lettera, l'ID può contenere fino a 63 caratteri. I caratteri validi sono lettere minuscole, numeri e trattini ([a-z0-9-]). L'ultimo carattere deve essere una lettera o un numero.
  • Se il primo carattere è un numero, l'ID può contenere fino a 9 caratteri. I caratteri validi sono numeri ([0-9]) senza zeri iniziali.

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

Configurare la durata (TTL) della sessione

Tutte le sessioni devono avere un tempo di scadenza. Puoi definire questa scadenza quando crei o aggiorni una sessione. La sessione e i relativi eventi secondari vengono eliminati automaticamente dopo la scadenza del periodo di tempo. Puoi impostare direttamente l'ora di scadenza (expire_time) o impostare la durata (ttl) in secondi. Se non viene specificato nessuno dei due, il sistema applica un TTL predefinito di 365 giorni.

Durata

Se imposti il tempo di permanenza, il server calcola la data di scadenza come create_time + ttl per le sessioni appena create o update_time + ttl per le sessioni aggiornate.

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

Scadenza 

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

Elenca le sessioni esistenti

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

# List sessions
sessions = await session_service.list_sessions(app_name=app_name,user_id=user_id)
print(sessions)
# 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:

# 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

Campi personalizzati per gli eventi

L'ADK supporta uno schema di eventi flessibile, in modo da poter includere dati arbitrari negli eventi di sessione. Ciò è utile per l'interoperabilità con altri framework di agenti o per archiviare dati sugli eventi personalizzati.

Quando aggiungi un evento con campi personalizzati, l'ADK serializza i dati sugli eventi nel campo raw_event dell'evento di sessione memorizzato.

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

Quando recuperi la sessione, gli eventi recuperati conservano questi campi personalizzati.

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 su Agent Platform

Dopo aver testato l'agente localmente, puoi eseguirne il deployment in produzione aggiornando l'istanza di Agent Platform 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 facile da ricordare 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://.

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