Gerenciar sessões com o Kit de Desenvolvimento de Agente

Nesta página, descrevemos como conectar um agente do Kit de Desenvolvimento de Agente (ADK) com sessões da plataforma de agente 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 de suporte:

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_ENTERPRISE"] = "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 Vertex AI Agent Engine

Para acessar as sessões do Agent Platform, primeiro use 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:

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 seus agentes, ferramentas e callbacks, e orquestra chamadas para ler e gravar sessões. Inicialize o Runner com VertexAiSessionService, que se conecta às sessões do Agent Platform. 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 extrair 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 da Agent Platform, interaja com ele para verificar se o histórico e os estados da sessão persistem.

Interface do ADK

Para testar seu agente com a interface do usuário do ADK e se conectar às sessões do Agent Platform, execute o comando adk web no terminal.

Embora o adk web geralmente leia variáveis de ambiente de um arquivo .env, ele ignora esse arquivo quando você usa a flag --session_service_uri. Em vez disso, defina as variáveis de ambiente GOOGLE_CLOUD_PROJECT e GOOGLE_CLOUD_LOCATION na sessão do terminal antes de executar o 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)

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. Especifique um ID de sessão personalizado para organizar as sessões usando seu próprio esquema de nomenclatura ou para fazer a transição rápida de outros serviços. Se você não especificar um ID de sessão, o Agent Platform vai gerar um automaticamente.

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

Para IDs de sessão personalizados, considere as seguintes restrições para evitar colisões com IDs gerados pelo sistema:

  • Se o primeiro caractere for uma letra, o ID poderá ter até 63 caracteres de comprimento. Os caracteres válidos são letras minúsculas, números e hifens ([a-z0-9-]). O último caractere precisa ser uma letra ou um número.
  • Se o primeiro caractere for um número, o ID poderá ter até nove caracteres. Os caracteres válidos são números ([0-9]) sem zeros à esquerda.

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

Configurar o 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. Você pode 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

Campos personalizados para eventos

O ADK é compatível com um esquema de eventos flexível para que você possa incluir dados arbitrários em eventos de sessão. Isso é útil para interoperabilidade com outros frameworks de agentes ou para armazenar dados de eventos personalizados.

Quando você anexa um evento com campos personalizados, o ADK serializa os dados do evento no campo raw_event do evento de sessão armazenado.

# 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 você recupera a sessão, os eventos recuperados retêm esses campos personalizados.

Excluir uma sessão

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

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

Implantar seu agente na Agent Platform

Depois de testar o agente localmente, implante o agente na produção atualizando a instância do Agent Platform 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 seu 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