Gestire le sessioni con Agent Development Kit

Questa pagina descrive come connettere un agente 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 creare un'istanza di Vertex AI Agent Engine. Per iniziare a utilizzare le sessioni, non devi eseguire il deployment di alcun codice. Senza il deployment del codice, la creazione di un'istanza di Vertex AI Agent Engine richiede solo pochi secondi.

import vertexai
from vertexai import agent_engines

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

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.

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)

Sostituisci quanto segue:

• PROJECT_ID: il tuo ID progetto.

• LOCATION: La tua regione.

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

• USER_ID: un identificatore univoco non vuoto per l'utente, con una lunghezza massima di 128 caratteri.

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.

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)

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:

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

Sostituisci quanto segue:

• AGENT: l'applicazione che implementa il metodo query / stream_query (ad esempio, AdkApp per un agente ADK). Per ulteriori informazioni, consulta Considerazioni sul 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

• Gestire le sessioni utilizzando le chiamate API.