Gerenciar sessões com o Agent Development Kit

Nesta página, descrevemos como conectar um agente do Kit de Desenvolvimento de Agente (ADK) com sessões do Vertex AI Agent Engine e usar sessões gerenciadas no ambiente local e de produção.

Antes de começar

Estas instruções usam a seguinte estrutura básica de arquivos de projeto para definir um agente do ADK e o runner e o código de implantação compatíveis:

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

Verifique se o ambiente está configurado seguindo as etapas Receber os papéis necessários e Autenticação em Configurar o ambiente.

Defina as variáveis de ambiente

Para usar o ADK, defina as variáveis de ambiente:

import os

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

Substitua:

PROJECT_ID: o ID do projeto.
LOCATION: sua região. Consulte as regiões compatíveis com o Memory Bank.

Criar uma instância do Agent Engine da Vertex AI

Para acessar as sessões do Vertex AI Agent Engine, primeiro você precisa usar uma instância do Vertex AI Agent Engine. Não é necessário implantar nenhum código para começar a usar as sessões. Se você já usou o Agent Engine, criar uma instância do Vertex AI Agent Engine leva apenas alguns segundos sem implantação de código. Isso pode levar mais tempo se for a primeira vez que você usa o Agent Engine.

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

Substitua:

PROJECT_ID: o ID do projeto.
LOCATION: sua região. Consulte as regiões compatíveis para sessões.

Desenvolver seu agente do ADK

Para criar seu agente do ADK, siga as instruções em Kit de desenvolvimento de agente ou use o código a seguir para criar um agente que cumprimenta um usuário com saudações fixas. Salve esse código em um arquivo chamado 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]
)

Configurar o executor do ADK

O ADK Runtime orquestra a execução de agentes, ferramentas e callbacks, além de orquestrar chamadas para ler e gravar sessões. Inicialize o Runner com VertexAiSessionService, que se conecta às sessões do Vertex AI Agent Engine. Salve esse código em um arquivo chamado runner.py.

Projeto do 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)

Substitua:

APP_NAME: o nome do aplicativo do agente.
USER_ID: escolha seu próprio ID de usuário com um limite de 128 caracteres. Por exemplo, user-123.
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 recuperar o ID do recurso usando agent_engine.api_resource.name.split("/")[-1].

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.

Interface do ADK

Teste seu agente com a interface do usuário do ADK e conecte-se à sessão do Vertex AI Agent Engine usando a opção de linha de comando session_service_uri. Ao usar session_service_uri, também é necessário definir as variáveis de ambiente GOOGLE_CLOUD_PROJECT e GOOGLE_CLOUD_LOCATION ou usar um arquivo .env no diretório principal que contém a pasta do agente. Por exemplo, se o agente estiver em agents/my_agent/, o arquivo .env vai estar na pasta agents, e você vai executar adk web nessa pasta.agents

project_id=PROJECT_ID
location=LOCATION
agent_engine_id="AGENT_ENGINE_ID"

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)

Interface do ADK

Python

Use o código Python do ADK para gerenciar sessões e estados. Adicione o seguinte código ao final do arquivo runner.py para interagir com o agente.

Os snippets a seguir contêm chamadas await de nível superior para serem mais breves. Para executar esse código como um script Python, coloque os snippets em uma função async e use asyncio.run() para executá-lo, conforme mostrado neste exemplo:

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

Criar uma sessão e consultar o agente

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

# file: my_agent/runner.py
# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_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"

Depois que a sessão é criada e transmitida ao executor, o ADK a usa para armazenar eventos da interação atual. Você também pode retomar uma sessão anterior fornecendo o ID dela.

Configurar time to live (TTL) da sessão

Todas as sessões precisam ter um tempo de expiração. É possível definir esse prazo de validade ao criar ou atualizar uma sessão. A sessão e os eventos filhos são excluídos automaticamente após o período de expiração. É possível definir o prazo de validade (expire_time) diretamente ou definir o tempo de vida (ttl) em segundos. Se nenhum for especificado, o sistema vai aplicar um TTL padrão de 365 dias.

Time to live (TTL)

Se você definir o tempo de vida, o servidor vai calcular o prazo de validade como create_time + ttl para sessões recém-criadas ou update_time + ttl para sessões atualizadas.

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

Tempo de expiração

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

Listar sessões atuais

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

# List sessions
sessions = await session_service.list_sessions(app_name=app_name,user_id=user_id)
print(sessions)
# ListSessionsResponse(session_ids=['1122334455', '9988776655'])

Gerenciar estados de sessão

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

# 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

Para atualizar o estado da sessão fora do runner, adicione um novo evento à sessão usando 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

Excluir uma sessão

Excluir uma sessão específica associada a um ID do usuário:

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 em produção atualizando a instância do Vertex AI Agent Engine com parâmetros:

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

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.
DISPLAY_NAME: um nome fácil de usar para o agente.
REQUIREMENTS: uma lista de pacotes pip exigidos pelo seu agente. Por exemplo, ["google-cloud-storage", "google-cloud-aiplatform[agent_engines,adk]"].
STAGING_BUCKET: um bucket do Cloud Storage com o prefixo gs://.

Limpar

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

agent_engine.delete(force=True)

A seguir

Gerenciar sessões usando chamadas de API.

Gerenciar sessões com o Agent Development Kit Mantenha tudo organizado com as coleções Salve e categorize o conteúdo com base nas suas preferências.

Antes de começar

Defina as variáveis de ambiente

Criar uma instância do Agent Engine da Vertex AI

Projeto do Google Cloud

Desenvolver seu agente do ADK

Configurar o executor do ADK

Projeto do Google Cloud

Interaja com seu agente

Interface do ADK

Python

Criar uma sessão e consultar o agente

Configurar time to live (TTL) da sessão

Time to live (TTL)

Tempo de expiração

Listar sessões atuais

Gerenciar estados de sessão

Excluir uma sessão

Implantar o agente no Vertex AI Agent Engine

Projeto do Google Cloud

Limpar

A seguir

Gerenciar sessões com o Agent Development Kit