Guia Google ADK + VertexAI: Construção do agente, deploy e consumo

Author

João Quentino

Construindo e Publicando um Agente com Google ADK + Vertex AI Agent Engine

Introdução

O Google Agent Development Kit (ADK) permite criar agentes inteligentes personalizados utilizando os modelos Gemini da Vertex AI.

Esses agentes podem integrar-se a diversas fontes de dados, utilizar ferramentas externas e serem implantados no Vertex AI Agent Engine , tornando-se acessíveis por APIs.

Neste documento, você aprenderá a:

  1. Estruturar um projeto de agente com o ADK.
  2. Configurar variáveis e permissões no Google Cloud.
  3. Fazer o deploy no Agent Engine .
  4. Enviar requisições para o agente via API REST.

Estrutura do Projeto

Um projeto básico de agente com ADK DEVE seguir esta estrutura:

├── chatbot_xique
│   ├── agent.py                # Código principal do agente
│   ├── __init__.py
│   └── test.py                 # Testes locais
├── deployment
│   ├── deploy.py               # Script de implantação (opcional)
│   └── __init__.py
├── requirements.txt            # Dependências
├── pyproject.toml              # Metadados do projeto (para ADK)
├── service-account.json        # Credenciais do Google Cloud
├── uv.lock                     # Gerenciador de pacotes (opcional)
└── README.md

Cada agente construído deve ser mantido em uma pasta separada, com agent.py e um arquivo __init__.py instanciando o agente declarado no arquivo agent.py: from .agent import root_agent

1. Configuração Inicial

1.1 Variáveis de ambiente

Crie um arquivo .env na raiz do projeto:

PROJECT_ID="id_do_projeto" # id do projeto
LOCATION="us-central1" # localização do projeto
DATASTORE_ID="id_do_datastore" # caso tenha um datastore
MODEL="gemini-2.0-flash"
OPENAI_API_KEY="SUA_CHAVE_OPENAI"
GOOGLE_API_KEY="SUA_CHAVE_GOOGLE"

DISCLAIMER: Caso queira utilizar a função embutida do adk, google_search, só irá funcionar com modelos do gemini. Caso o usuário queira utilizar um modelo da OpenAI ou outros, terá que construir sua própria ferramenta de busca. Mas um modelo da OpenAi faz busca em um datastore da Google facilmente.

1.2 Dependências

Crie e ative um ambiente virtual, e baixe essa lista de requirements.txt:

google-cloud-aiplatform
google-adk
google-cloud-discoveryengine
python-dotenv
requests
fastapi
uvicorn
pydantic

2. Construindo o Agente

2.1 Estrutura básica (agent.py)

O ADK fornece o objeto Agent para construir um assistente com ferramentas e instruções específicas:

import re
import vertexai
from vertexai import agent_engines
from google.cloud import discoveryengine
from google.adk.agents import Agent
from google.adk.tools import google_search
from dotenv import load_dotenv
import os

load_dotenv()

PROJECT_ID = os.getenv("PROJECT_ID")
LOCATION = "global"
DATASTORE_ID = os.getenv("DATASTORE_ID")
MODEL = "gemini-2.0-flash"

client = vertexai.Client(project=PROJECT_ID, location="us-central1")

2.2 Função de busca no Datastore

def search_in_datastore(query: str) -> str:
    client = discoveryengine.SearchServiceClient()
    serving_config = client.serving_config_path(
        project=PROJECT_ID,
        location=LOCATION,
        data_store=DATASTORE_ID,
        serving_config="default_config",
    )

    request = discoveryengine.SearchRequest(
        serving_config=serving_config,
        query=query,
        page_size=3,
        content_search_spec=discoveryengine.SearchRequest.ContentSearchSpec(
            snippet_spec=discoveryengine.SearchRequest.ContentSearchSpec.SnippetSpec(return_snippet=True)
        ),
    )
    response = client.search(request)

    results = []
    for result in response.results:
        title = result.document.derived_struct_data.get("title", "Fonte")
        snippet = result.document.derived_struct_data.get("snippet", "")
        results.append(f"[{title}] {snippet}")

    return "\n".join(results)

2.3 Instanciando o Agente

root_agent = Agent(
    name="xique_agent",
    model=MODEL,
    description="Chatbot para a cidade de Xique-xique, Bahia",
    instruction=(
        "Você é um assistente especializado na cidade de Xique-Xique, Bahia. "
        "Use as fontes do datastore primeiro; se necessário, utilize o google_search. "
        "Responda apenas perguntas sobre Xique-xique, BA."
    ),
    tools=[google_search, search_in_datastore],
)

Para testar o agente, basta realizar o comando: adk web

3. Deploy no Vertex AI Agent Engine

3.1 Comando de Deploy

Execute no terminal:

adk deploy agent_engine \
  --project=id_do_projeto \
  --region=loc_do_projeto \
  --staging_bucket=gs://nome-do-bucket \
  --display_name="xique-chatbot" \
  ./chatbot_xique # nome da pasta que está o agente

O staging_bucket deve ser um bucket pré-criado no Google Cloud Storage.

4. Permissões Necessárias

O runtime do agente precisa de permissões para acessar o Discovery Engine e Document AI.

4.1 Obter o número do projeto

PROJECT_NUMBER=$(gcloud projects describe id-do-projeto --format="value(projectNumber)")

4.2 Conceder permissões

gcloud projects add-iam-policy-binding id-do-projeto \
  --member="serviceAccount:${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" \
  --role="roles/documentai.apiUser"

gcloud projects add-iam-policy-binding id-do-projeto \
  --member="serviceAccount:${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" \
  --role="roles/aiplatform.user"

gcloud projects add-iam-policy-binding id-do-projeto \
  --member="serviceAccount:${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" \
  --role="roles/discoveryengine.viewer"

5. Fazendo Requisições ao Agente

Após o deploy, você pode interagir com o agente via API REST do Reasoning Engine .

5.1 Criando sessão

import requests, uuid, os
from dotenv import load_dotenv
load_dotenv()

PROJECT_ID = os.getenv("PROJECT_ID")
LOCATION = os.getenv("LOCATION")
REASONING_ENGINE_ID = os.getenv("REASONING_ENGINE_ID")
TOKEN = os.getenv("ACCESS_TOKEN")  # obtido via get_auth_token()

url = f"https://{LOCATION}-aiplatform.googleapis.com/v1/projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{REASONING_ENGINE_ID}:query"

headers = {
    "Authorization": f"Bearer {TOKEN}",
    "Content-Type": "application/json"
}

payload = {
    "class_method": "create_session",
    "input": {"user_id": f"u_{uuid.uuid4().hex[:8]}"}
}

response = requests.post(url, headers=headers, json=payload)
print(response.json())

5.2 Enviando mensagem ao agente

session_id = "<ID_DA_SESSÃO>"
user_id = "<ID_DO_USUÁRIO>"

chat_url = f"https://{LOCATION}-aiplatform.googleapis.com/v1/projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{REASONING_ENGINE_ID}:streamQuery?alt=sse"

payload = {
    "class_method": "async_stream_query",
    "input": {
        "user_id": user_id,
        "session_id": session_id,
        "message": "Qual é a principal festa tradicional de Xique-xique BA?"
    }
}

response = requests.post(chat_url, headers=headers, json=payload, stream=True)
for line in response.iter_lines():
    if line:
        print(line.decode())

6. Gerando Token de Autenticação

Crie uma função utilitária:

import google.auth
import google.auth.transport.requests
import google.oauth2.service_account
import os, json

def get_auth_token():
    creds_json = os.getenv('GOOGLE_CREDENTIALS')
    if creds_json:
        creds_info = json.loads(creds_json)
        creds = google.oauth2.service_account.Credentials.from_service_account_info(
            creds_info,
            scopes=['https://www.googleapis.com/auth/cloud-platform']
        )
    else:
        creds, _ = google.auth.default(scopes=['https://www.googleapis.com/auth/cloud-platform'])

    creds.refresh(google.auth.transport.requests.Request())
    return creds.token

7. Criando API FastAPI para o Agente

Você pode disponibilizar uma API intermediária em FastAPI para facilitar o consumo:

from fastapi import FastAPI
from src.utils.auth import get_auth_token
import requests, uuid, json, os

app = FastAPI(title="Xique-xique AI")

@app.post("/session/create")
def create_session():
    token = get_auth_token()
    url = f"https://us-central1-aiplatform.googleapis.com/v1/projects/{os.getenv('PROJECT_ID')}/locations/us-central1/reasoningEngines/{os.getenv('REASONING_ENGINE_ID')}:query"
    user_id = f"u_{uuid.uuid4().hex[:8]}"
    payload = {"class_method": "create_session", "input": {"user_id": user_id}}
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    return requests.post(url, headers=headers, json=payload).json()

@app.post("/chat")
def chat(user_id: str, session_id: str, message: str):
    token = get_auth_token()
    url = f"https://us-central1-aiplatform.googleapis.com/v1/projects/{os.getenv('PROJECT_ID')}/locations/us-central1/reasoningEngines/{os.getenv('REASONING_ENGINE_ID')}:streamQuery?alt=sse"
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    payload = {"class_method": "async_stream_query", "input": {"user_id": user_id, "session_id": session_id, "message": message}}
    r = requests.post(url, headers=headers, json=payload, stream=True)
    for line in r.iter_lines():
        if line:
            print(line.decode())