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 Agent Engine da Vertex AI 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 criar uma instância do Vertex AI Agent Engine. Não precisa de implementar nenhum código para começar a usar as sessões. Sem a implementação de código, a criação de uma instância do Agent Engine do Vertex AI demora apenas alguns segundos.

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

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.

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 o seguinte:

  • PROJECT_ID: o ID do seu projeto.

  • LOCATION: a sua região.

  • 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.name.split("/")[-1].

  • USER_ID: um identificador único não vazio para o utilizador, com um comprimento máximo de 128 carateres.

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

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

client.agent_engines.update(
    resource_name=agent_engine.api_resource.name,
    agent=AGENT,
    config={"requirements": REQUIREMENTS},
)

Substitua o seguinte:

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?