Faça a gestão de sessões com o Agent Development Kit

Esta página descreve como pode associar um agente do Agent Development Kit (ADK) a sessões do Vertex AI Agent Engine e usar sessões geridas no ambiente local e de produção.

Antes de começar

Certifique-se de que o seu ambiente está configurado seguindo os passos Obtenha as funções necessárias e Autenticação em Configure o seu ambiente.

Crie uma instância do Vertex AI Agent Engine

Para aceder às sessões do Vertex AI Agent Engine, primeiro tem de usar uma instância do Vertex AI Agent Engine. Não precisa de implementar nenhum código para começar a usar as sessões. Se já usou o Agent Engine, a criação de uma instância do Agent Engine da Vertex AI demora apenas alguns segundos sem implementação de código. Pode demorar mais tempo se for a primeira vez que 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 o seguinte:

  • PROJECT_ID: o ID do projeto.

  • LOCATION: a sua região. Consulte as regiões suportadas para as Sessões.

Modo expresso do Vertex AI

Pode usar o SDK Vertex AI para Python com a chave da API do modo expresso:

import vertexai

client = vertexai.Client(  # For service interactions via client.agent_engines
    api_key="API_KEY",
)

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

Desenvolva o seu agente ADK

Para criar o seu agente ADK, siga as instruções no Agent Development Kit ou use o seguinte código para criar um agente que cumprimenta um utilizador com cumprimentos fixos:

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

Configure o executor do ADK

O ADK Runtime orquestra a execução dos seus agentes, ferramentas e callbacks, bem como as chamadas para ler e escrever sessões. Inicialize o Runner com VertexAiSessionService, que se liga às sessões do Vertex AI Agent Engine.

Projeto do Google Cloud 

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_ID",
      "LOCATION",
      "AGENT_ENGINE_ID"
)
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 o seguinte:

  • APP_NAME: o nome da sua aplicação de agente.

  • USER_ID: escolha o seu próprio ID do utilizador com um limite de 128 carateres. Por exemplo, user-123.

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

  • Para agentes implementados, o ID do recurso é apresentado como a variável de ambiente GOOGLE_CLOUD_AGENT_ENGINE_ID

  • Para agentes locais, pode obter o ID do recurso através de agent_engine.api_resource.name.split("/")[-1].

Modo expresso do Vertex AI 

import os
from google.adk.sessions import VertexAiSessionService
from google.genai import types

# Set environment variables
os.environ["GOOGLE_API_KEY"] = "API_KEY"
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"

# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
      agent_engine_id="AGENT_ENGINE_ID"
)
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 o seguinte:

  • API_KEY: a chave da API do modo expresso da Vertex AI.

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

  • Para agentes implementados, o ID do recurso é apresentado como a variável de ambiente GOOGLE_CLOUD_AGENT_ENGINE_ID

  • Para agentes locais, pode obter o ID do recurso através de agent_engine.api_resource.name.split("/")[-1].

Interaja com o seu agente

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

IU do ADK

Teste o seu agente com a interface do utilizador do ADK e estabeleça ligação à sessão do Vertex AI Agent Engine através da opção de linha de comandos session_service_uri:

agent_engine_id="AGENT_ENGINE_ID"

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)

IU do ADK

Python

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

Crie uma sessão e consulte o agente

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

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

Depois de a sessão ser criada e transmitida ao executor, o ADK usa a sessão para armazenar eventos da interação atual. Também pode retomar uma sessão anterior indicando o ID dessa sessão.

Apresente as sessões existentes

Liste todas as sessões existentes associadas a um determinado ID do utilizador.

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

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

Faça a gestão dos estados das sessões

Os estados contêm informações de que o agente precisa para uma conversa. Pode fornecer um estado inicial como um dicionário quando cria 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 executor, acrescente um novo evento à sessão através de 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
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

Elimine uma sessão

Elimine uma sessão específica associada a um User-ID:

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

Implemente o seu agente no Vertex AI Agent Engine

Depois de testar o agente localmente, pode implementá-lo na produção atualizando a instância do Vertex AI Agent Engine com os seguintes 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 o seguinte:

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

  • DISPLAY_NAME: um nome intuitivo para o seu agente.

  • REQUIREMENTS: uma lista de pacotes pip necessários para o seu agente. Por exemplo, ["google-cloud-storage", "google-cloud-aiplatform[agent_engines,adk]"].

  • STAGING_BUCKET: um contentor do Cloud Storage com o prefixo gs://.

Modo expresso do Vertex AI

Implemente o seu agente através de ficheiros de origem seguindo estes passos de implementação.

Limpar

Para limpar todos os recursos usados neste projeto, pode eliminar a instância do Vertex AI Agent Engine juntamente com os respetivos recursos secundários:

agent_engine.delete(force=True)

O que se segue?