Gerenciar sessões com o Agent Development Kit

Nesta página, descrevemos como conectar um agente do Kit de desenvolvimento de agentes (ADK) com 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

Verifique se o ambiente está configurado seguindo as etapas Receber os papéis necessários 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 crie 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 implantação de código, criar 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 em Kit de desenvolvimento de agentes 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 organiza a execução dos seus agentes, ferramentas e callbacks, além de organizar 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.

• 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 para o 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, interaja com ele para verificar se o histórico e os estados da sessão persistem.

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)

Implantar o agente no Vertex AI Agent Engine

Depois de testar o agente localmente, é possível implantá-lo na produção atualizando a instância do Vertex AI Agent Engine com os seguintes 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 mecanismo de agente da Vertex AI e os recursos filhos dela:

agent_engine.delete(force=True)

A seguir

• Gerenciar sessões usando chamadas de API.