Gerenciar sessões com o Agent Development Kit

Esta página descreve como conectar um agente do Kit de desenvolvimento de agentes (ADK) a sessões do mecanismo de agente da Vertex AI e usar sessões gerenciadas no ambiente local e de produção.

Antes de começar

Confira se o ambiente está configurado seguindo as etapas Receber as funções necessárias e Autenticação em Configurar o ambiente.

Criar uma instância do Vertex AI Agent Engine

Para acessar as sessões do Vertex AI Agent Engine, primeiro é necessário criar uma instância do Vertex AI Agent Engine. Não é necessário implantar nenhum código para começar a usar as Sessões. Sem a implantação de código, a criação de uma instância do Vertex AI Agent Engine leva apenas alguns segundos.

import vertexai
from vertexai import agent_engines

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

Desenvolver seu agente do ADK

Para criar seu agente do ADK, siga as instruções no Agent Development Kit (link em inglês) ou use o código a seguir para criar um agente que cumprimenta um usuário com saudações fixas:

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

Configurar o executor do ADK

O ADK Runtime orquestra a execução de agentes, ferramentas e callbacks, além de chamadas para ler e gravar sessões. Inicialize o Runner com VertexAiSessionService, que se conecta às sessões do 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)

Substitua:

  • PROJECT_ID: o ID do projeto.

  • LOCATION: sua região. Somente us-central1 é aceito para sessões do mecanismo do agente da Vertex AI.

  • AGENT_ENGINE_ID: o ID do recurso de uma instância do Vertex AI Agent Engine.

    • Para agentes implantados, o ID do recurso é listado como a variável de ambiente GOOGLE_CLOUD_AGENT_ENGINE_ID.

    • Para agentes locais, é possível extrair o ID do recurso usando agent_engine.name.split("/")[-1].

  • USER_ID: um identificador exclusivo não vazio do usuário, com um comprimento máximo de 128 caracteres.

Interaja com seu agente

Depois de definir o agente e configurar as sessões do Vertex AI Agent Engine, você pode interagir com ele para verificar se o histórico e os estados da sessão persistem.

Interface do ADK

Teste seu agente com a interface do usuário do ADK e se conecte à sessão do motor do agente do Vertex AI usando a opção de linha de comando session_db_url:

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)

Interface do ADK

Python

Use o código Python do ADK para gerenciar sessões e estados.

Criar uma sessão e consultar o agente

Use o código abaixo para criar uma sessão e enviar uma consulta ao agente:

# Create a session
session = 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"

Depois que a sessão é criada e transmitida ao executor, o ADK usa a sessão para armazenar eventos da interação atual. Também é possível retomar uma sessão anterior fornecendo o ID dela.

Listar sessões

Lista todas as sessões associadas a um determinado ID do usuário.

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

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

Listar eventos de uma sessão

Lista todos os eventos associados a uma sessão.

# List all the events of the session
for event in session_service.list_events(app_name=app_name, user_id=user_id, session_id=session.id):
  if event.is_final_response():
    print(event.content.parts[0].text)

# Expected output
# Hello!
# Hello, world
# Thanks!
# Goodbye, world

Gerenciar estados da sessão

Os estados contêm informações necessárias para uma conversa. É possível fornecer um estado inicial como um dicionário ao criar uma sessão:

# Create a session with state
session = session_service.create_session(
      app_name=app_name,
      user_id=user_id,
      state={'key': 'value'})

print(session.state['key'])
# value

Para atualizar o estado da sessão fora do executor, anexe um novo evento à sessão usando 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
session_service.append_event(session, system_event)

# Check updated state
updated_session = 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

Excluir uma sessão

Exclua uma sessão específica associada a um ID de usuário:

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

Implantar seu agente no Vertex AI Agent Engine

Depois de testar o agente localmente, você pode implantar o agente na produção atualizando a instância do Vertex AI Agent Engine com parâmetros:

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

Substitua:

  • AGENT: o aplicativo que implementa o método query / stream_query (por exemplo, AdkApp para um agente do ADK). Para mais informações, consulte Considerações sobre a implantação.

Limpar

Para limpar todos os recursos usados neste projeto, exclua a instância do Vertex AI Agent Engine e os recursos filhos:

agent_engine.delete(force=True)

A seguir