Curso
Desenvolvimento de aplicativos de LLM com LangChain
3 h
34.6K
Neste blog, mostrarei a você como criar um assistente de viagem com vários agentes e com tecnologia de IA usando o novo código aberto do Google, o Kit de desenvolvimento de agentes (ADK) de código aberto do Google e o A2A (agente para agente) (agente para agente).
Usaremos vários agentes para lidar com recomendações de voos, hotéis e atividades, comunicando-nos por meio de APIs REST e servidores FastAPI. No final, vamos encerrar tudo com um front-end limpo do Streamlit para proporcionar uma experiência de usuário intuitiva.
Mantemos nossos leitores atualizados sobre as últimas novidades em IA enviando o The Median, nosso boletim informativo gratuito de sexta-feira que detalha as principais histórias da semana. Inscreva-se e fique atento em apenas alguns minutos por semana:
O que é o ADK (Agent Development Kit) do Google?
O ADK do Google é uma estrutura modular, pronta para produção, para a criação de agentes acionados por LLM. É o mesmo kit de ferramentas que capacita os agentes dentro dos produtos do Google, como o Agentspace e o Customer Engagement Suite. Agora de código abertoele ajuda os desenvolvedores a criar aplicativos multiagentes avançados, flexíveis e interoperáveis.
Por que usar o Agent Development Kit (ADK)?
O ADK oferece a flexibilidade do Python com estruturas integradas para gerenciamento de estado, retornos de chamada, streaming e entrada/saída estruturada. Vamos dar uma olhada em seus principais recursos:
- Multiagente por design: O ADK pode compor agentes em fluxos de trabalho paralelos, sequenciais ou hierárquicos.
- Model-agnostic: Ele funciona com Gemini, GPT-4o, Claude, Mistral e outros via
LiteLlm. - Modular e dimensionável: O usuário pode definir agentes especializados e delegar de forma inteligente usando a orquestração integrada.
- Pronto para streaming: Ele oferece suporte à interação em tempo real, incluindo áudio/vídeo bidirecional.
- Ferramentas integradas: Ele oferece suporte à CLI local e à interface do usuário da Web para depuração e avaliação.
- Oferece suporte à implementação: O ADK facilmente coloca em contêineres e implementa agentes em vários ambientes.
O que é o protocolo Agent2Agent (A2A) do Google?
O protocolo Agent2Agent (A2A) é um padrão aberto e independente de fornecedor desenvolvido pelo Google para facilitar a comunicação e a colaboração entre agentes de IA em diversas plataformas e estruturas.
Os agentes do ADK expõem um endpoint HTTP padrão /run e metadados por meio de .well-known/agent.json. Isso permite a descoberta de agentes e a fácil comunicação entre eles (ou até mesmo com orquestradores externos, como LangGraph ou CrewAI).
Embora seja opcional, a adição do arquivo de metadados A2A torna seus agentes interoperáveis com o ecossistema mais amplo de ferramentas de agentes e orquestradores.
Visão geral do projeto: Planejador de viagens com tecnologia de IA com ADK e A2A
Este projeto cria um planejador de viagens que:
- Recebe como entrada o destino, as datas e o orçamento.
- Chama três agentes separados:
flight_agent: Recomenda opções de voo.stay_agent: Encontra hotéis dentro do orçamento.activities_agent: Sugere atividades envolventes.- Em seguida, usa um site central
host_agentpara orquestrar todas as solicitações. - Por fim, usa uma interface de usuário Streamlit para interação com o usuário.
Todos os agentes são hospedados como servidores FastAPI separados e expõem um ponto de extremidade /run. A comunicação é feita por meio do cliente compartilhado compatível com A2A.
Observação: Este projeto é executado inteiramente em sua máquina local para simplificar, mas você pode facilmente implantar cada agente e a interface do usuário em plataformas de nuvem como Render, Railway ou Google Cloud Run para obter acesso dimensionável.
Etapa 1: Pré-requisitos
Vamos começar instalando as seguintes bibliotecas:
pip install google-adk litellm fastapi uvicorn httpx pydantic openai streamlitEm seguida, configure sua chave de API da OpenAI - você pode usar outro provedor de modelos. Para saber como configurar sua chave de API da OpenAI, recomendo que você leia este tutorial introdutório sobre a API GPT-4o.
export OPENAI_API_KEY="your_key_here"Etapa 2: Esquema e utilitários compartilhados
Antes de podermos criar agentes inteligentes, precisamos definir uma linguagem comum para que eles conversem entre si. Em nossa configuração, isso é feito usando:
- Um esquema compartilhado para entrada (definido via Pydantic)
- Um utilitário de cliente REST para chamar agentes
- Um wrapper de servidor REST para padronizar o ponto de extremidade
/runem todos os agentes
Eles são colocados nas pastas shared/ e common/ para manter o código modular. Vamos dar uma olhada em cada um deles.
Criando um shared/schemas.py arquivo
Definimos um esquema TravelRequest usando Pydantic. Isso garante que todos os agentes concordem com a estrutura das solicitações recebidas, que inclui o destino, as datas da viagem e o orçamento do usuário.
from pydantic import BaseModel
class TravelRequest(BaseModel):
destination: str
start_date: str
end_date: str
budget: floatEsta aula ajuda você a:
- Manter a entrada consistente para todos os agentes.
- Adição de validação automática com FastAPI.
- Simplificando a reutilização do código.
Criando um common/a2a_client.py arquivo
Esse utilitário assíncrono leve permite que qualquer agente (especialmente o host) invoque outro agente usando o protocolo A2A chamando o ponto de extremidade /run.
import httpx
async def call_agent(url, payload):
async with httpx.AsyncClient() as client:
response = await client.post(url, json=payload, timeout=60.0)
response.raise_for_status()
return response.json()Esse utilitário envia de forma assíncrona uma solicitação POST para o ponto de extremidade /run de outro agente usando httpx. Ele retorna a resposta JSON analisada e gera um erro se a solicitação falhar.
Usaremos esse utilitário em nosso agente host para chamar flight_agent, stay_agent e activities_agent.
Criando um common/a2a_server.py arquivo
Em vez de escrever uma rota FastAPI personalizada para cada agente, nós a generalizamos usando a função create_app(agent), que manipula:
- Servindo o agente em
/run - Recebimento de uma solicitação de viagem
- Retornando uma resposta estruturada
from fastapi import FastAPI
import uvicorn
def create_app(agent):
app = FastAPI()
@app.post("/run")
async def run(payload: dict):
return await agent.execute(payload)
return appEste utilitário cria um aplicativo FastAPI com uma rota /run padrão que delega a execução ao agente fornecido. Ele garante uma interface agente a agente (A2A) consistente para todos os serviços que usam entrada JSON estruturada.
Juntos, esses componentes compartilhados tornam nosso sistema multiagente mais sustentável, reutilizável e alinhado com a filosofia A2A do Google de mensagens simples e estruturadas entre agentes.
Etapa 3: Criação do sistema multiagente com ADK e A2A
Agora que já temos contratos e utilitários compartilhados, vamos começar a criar os agentes individuais. Para transformar isso em um sistema verdadeiramente modular e multiagente, usaremos o protocolo A2A do Google, uma interface simples baseada em HTTP que permite que os agentes se comuniquem de forma consistente e interoperável.
O A2A (Agent-to-Agent) permite a coordenação plug-and-play entre agentes, sejam eles funções Python locais ou hospedados em redes. Cada agente expõe um ponto de extremidade /run com um esquema comum e atua como um serviço.
Em nossa demonstração, temos quatro agentes:
host_agent: Orquestra todos os outros agentesflight_agent: Encontrar voos adequadosstay_agent: Sugere acomodaçõesactivities_agent: Recomenda que você participe de atividades locais
Todos os agentes são estruturados de forma semelhante, com 3 arquivos e uma subpasta:
agents/
├── host_agent/
│ │ ├── agent.py # Optional if host logic is minimal
│ │ ├── task_manager.py # Calls other agents and aggregates responses
│ │ ├── __main__.py # Starts FastAPI app via common/a2a_server.py
│ │ └── .well-known/
│ │ └── agent.json # A2A Agent Card metadata
├── flight_agent/
├── stay_agent/
└── activities_agent/Cada agente usa google.adk.agents.Agent, um wrapper de modelo LiteLlm e Runner para execução. Comece criando os seguintes arquivos na pasta activities_agent e repita o mesmo para flight_agent e stay_agent.
Criando um agent.py arquivo
Vamos agora definir a lógica do nosso site activities_agent, que será responsável por gerar experiências locais envolventes com base no itinerário de viagem do usuário.
Etapa 1: Importações
Começamos importando os módulos essenciais para configurar e executar nosso agente.
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import jsonEsse agente usa os componentes do Google ADK, como Agent, ' Runner, ' e LiteLlm, e lida com o gerenciamento de estado usando InMemorySessionService.. A biblioteca Types é usada para criar prompts estruturados.
Etapa 2: Agente de atividades
Agora, instanciamos o próprio agente usando a classe Agent do ADK.
activities_agent = Agent(
name="activities_agent",
model=LiteLlm("openai/gpt-4o"),
description="Suggests interesting activities for the user at a destination.",
instruction=(
"Given a destination, dates, and budget, suggest 2-3 engaging tourist or cultural activities. "
"For each activity, provide a name, a short description, price estimate, and duration in hours. "
"Respond in plain English. Keep it concise and well-formatted."
)
)O parâmetro instruction define o prompt do sistema que orienta o comportamento do LLM. Embora este exemplo use inglês simples, você pode ajustar a instrução para retornar JSON estruturado para facilitar a análise.
Etapa 3: Gerenciamento de sessões
Em seguida, para manter o controle das interações do usuário, configuramos um programa Runner junto com as informações da sessão.
session_service = InMemorySessionService()
runner = Runner(
agent=activities_agent,
app_name="activities_app",
session_service=session_service
)
USER_ID = "user_activities"
SESSION_ID = "session_activities"O Runner gerencia a execução do agente para uma determinada sessão do aplicativo. Já a classe InMemorySessionService armazena o contexto na memória. Em seguida, definimos IDs de usuário e de sessão. No entanto, na produção, eles podem ser dinâmicos ou específicos do usuário. Isso garante a existência de uma nova sessão do ADK antes de enviar qualquer prompt para o agente LLM.
Etapa 4: Execução da lógica do agente
A função execute() lida com as solicitações recebidas, cria um prompt, invoca o modelo e analisa a saída.
async def execute(request):
session_service.create_session(
app_name="activities_app",
user_id=USER_ID,
session_id=SESSION_ID
)
prompt = (
f"User is flying to {request['destination']} from {request['start_date']} to {request['end_date']}, "
f"with a budget of {request['budget']}. Suggest 2-3 activities, each with name, description, price estimate, and duration. "
f"Respond in JSON format using the key 'activities' with a list of activity objects."
)
message = types.Content(role="user", parts=[types.Part(text=prompt)])
async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=message):
if event.is_final_response():
response_text = event.content.parts[0].text
try:
parsed = json.loads(response_text)
if "activities" in parsed and isinstance(parsed["activities"], list):
return {"activities": parsed["activities"]}
else:
print("'activities' key missing or not a list in response JSON")
return {"activities": response_text} # fallback to raw text
except json.JSONDecodeError as e:
print("JSON parsing failed:", e)
print("Response content:", response_text)
return {"activities": response_text} # fallback to raw textA função execute() cria dinamicamente um prompt usando os parâmetros da solicitação de entrada, como destino, datas e orçamento. Aqui está o que acontece sob o capô:
- O prompt instrui o modelo a retornar um objeto JSON estruturado contendo uma lista de atividades.
- Um objeto
Contenté construído e passado para o ADK Runner, que aguarda de forma assíncrona a resposta final do modelo usando um gerador de streaming. - Quando a resposta final é recebida, o agente extrai a saída de texto bruto e tenta analisá-la como JSON.
- Se a análise for bem-sucedida e a chave de atividades esperada existir, os dados estruturados serão retornados.
- Se a chave estiver ausente ou malformada, a alternativa é retornar a resposta de texto bruto para que a interface do usuário ainda tenha uma saída utilizável.
- Essa abordagem de manuseio duplo garante uma degradação graciosa quando o LLM retorna texto simples em vez de JSON estruturado.
Essa estratégia melhora a robustez e a experiência do usuário, especialmente quando os resultados do modelo variam ligeiramente devido à temperatura ou à interpretação imediata.
Criando um task_manager.py arquivo
Depois de definir a lógica do execute() dentro do agent.py, agora vamos conectá-lo à configuração do servidor compatível com o ADK usando o task_manager.py.
from .agent import execute
async def run(payload):
return await execute(payload)Esse arquivo funciona como um invólucro fino em torno da função execute() definida anteriormente. Ele torna o método run() disponível para módulos externos, especialmente o script do servidor em __main__.py.
Criando um __main__.py arquivo
O arquivo __main__.py inicia um servidor FastAPI na porta 8003, servindo o agente no ponto de extremidade /run.
from common.a2a_server import create_app
from .task_manager import run
app = create_app(agent=type("Agent", (), {"execute": run}))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, port=8003)Aqui está o que está acontecendo:
- O
create_app()(de common/a2a_server.py) envolve nosso agente em uma interface FastAPI compatível com A2A padrão. - Construímos dinamicamente um objeto com um método
execute()para que o ADK possa invocarrun()corretamente. - Essa separação nos permite manter interfaces de API sem estado e reutilizar a lógica do agente principal.
Criando um .well-known/agent.json arquivo
Usamos esse arquivo JSON para descrever a identidade e a finalidade do agente de acordo com o protocolo A2A (Agent-to-Agent).
{
"name": "activity_agent",
"description": "Agent providing activity details."
}Observação: Embora o arquivo .well-known/agent.json não seja usado diretamente pelos nossos agentes neste projeto, ele segue a especificação A2A e é importante para descoberta, introspecção e compatibilidade futura com orquestradores como LangGraph, CrewAI ou o registro de agentes do Google.
Uma lógica semelhante é usada para flight_agent e stay_agent também.
Etapa 4: Coordenação com host_agent
O host_agent atua como um planejador central para a demonstração. O site host_agent exemplifica o padrão de controle em sistemas multiagentes. Ele separa a tomada de decisões e a execução, permitindo que cada agente downstream se concentre em seu nicho enquanto centraliza a lógica de coordenação. Isso não apenas simplifica os testes e o dimensionamento, mas também reflete a arquitetura de microsserviço do mundo real em sistemas distribuídos.
Ele envia a mesma carga útil para todos os três agentes usando suas APIs /run expostas e mescla os resultados. Vamos adicionar os seguintes arquivos à pasta host_agent.
Criando um agent.py arquivo
Vamos começar com as importações básicas.
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import typesEsse bloco de importação traz todos os principais componentes necessários para você definir e executar um agente baseado em LLM usando o Google ADK: classe Agent, wrapper LLM leve, Runner para lidar com a execução e gerenciamento de sessões na memória.
host_agent = Agent(
name="host_agent",
model=LiteLlm("openai/gpt-4o"),
description="Coordinates travel planning by calling flight, stay, and activity agents.",
instruction="You are the host agent responsible for orchestrating trip planning tasks. "
"You call external agents to gather flights, stays, and activities, then return a final result."
)
session_service = InMemorySessionService()
runner = Runner(
agent=host_agent,
app_name="host_app",
session_service=session_service
)
USER_ID = "user_host"
SESSION_ID = "session_host"O código acima define um agente ADK de nível superior responsável por coordenar o plano de viagem completo. Embora não invoquemos subagentes do LLM nesta implementação, o prompt do sistema define a função para uma extensão futura em que o LLM poderia lidar com o uso de ferramentas e o meta-raciocínio.
async def execute(request):
# Ensure session exists
session_service.create_session(
app_name="host_app",
user_id=USER_ID,
session_id=SESSION_ID
)
prompt = (
f"Plan a trip to {request['destination']} from {request['start_date']} to {request['end_date']} "
f"within a total budget of {request['budget']}. Call the flights, stays, and activities agents for results."
)
message = types.Content(role="user", parts=[types.Part(text=prompt)])
async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=message):
if event.is_final_response():
return {"summary": event.content.parts[0].text}Essa função execute() serve como o principal ponto de entrada para o LLM do agente host. It:
- Inicializa uma sessão (para suporte de memória, se necessário)
- Constrói dinamicamente um prompt de usuário
- Envia-o para o modelo usando runner.run_async() do ADK
method - Por fim, você aguarda e extrai a resposta final
Criando um task_manager.py arquivo
O gerenciador de tarefas executa a lógica de orquestração chamando agentes remotos e lidando com o fluxo de trabalho completo de planejamento de viagens. Para sua implementação prática, definimos as URLs de serviço para cada agente filho. Esses pontos de extremidade estão em conformidade com o esquema JSON A2A /run protocol and expect a shared TravelRequest`.
from common.a2a_client import call_agent
FLIGHT_URL = "http://localhost:8001/run"
STAY_URL = "http://localhost:8002/run"
ACTIVITIES_URL = "http://localhost:8003/run"Agora, definimos a carga útil.
async def run(payload):
#Print what the host agent is sending
print("Incoming payload:", payload)
flights = await call_agent(FLIGHT_URL, payload)
stay = await call_agent(STAY_URL, payload)
activities = await call_agent(ACTIVITIES_URL, payload)
# Log outputs
print("flights:", flights)
print("stay:", stay)
print("activities:", activities)
# Ensure all are dicts before access
flights = flights if isinstance(flights, dict) else {}
stay = stay if isinstance(stay, dict) else {}
activities = activities if isinstance(activities, dict) else {}
return {
"flights": flights.get("flights", "No flights returned."),
"stay": stay.get("stays", "No stay options returned."),
"activities": activities.get("activities", "No activities found.")
}Essa função usa a função auxiliar call_agent() para enviar a carga útil para cada serviço downstream e registra entradas e saídas para visibilidade durante o desenvolvimento. Esse arquivo é essencialmente onde reside a verdadeira lógica de orquestração.
Criando um __main__.py arquivo
O arquivo __main__.py serve como ponto de entrada para o servidor FastAPI que envolve o agente host.
from common.a2a_server import create_app
from .task_manager import run
app = create_app(agent=type("Agent", (), {"execute": run}))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, port=8000)O arquivo principal faz o seguinte:
- Ele usa
create_app()decommon/a2a_server.pypara gerar um aplicativo FastAPI com um ponto de extremidade/runpadronizado. - Em seguida, ele passa um objeto simples do tipo agente com um método
execute()que delega internamente para a funçãotask_manager.run(). - Por fim, ele inicia o servidor FastAPI usando
uvicornem uma porta especificada (geralmente 8000).
Isso alinha a interface do agente host com outros agentes downstream, mantendo a consistência em todo o sistema.
Criando um .well-known/agent.json arquivo
Esse arquivo funciona como um padrão multiagente clássico em que um nó central delega e compõe tarefas.
{
"name": "host_agent",
"description": "Coordinates travel planning among specialized agents."
}Embora seja opcional, é uma ótima prática incluir isso em todos os diretórios de agentes, conforme explicado anteriormente.
Etapa 5: Criando a interface do usuário com o Streamlit
Por fim, vamos criar um aplicativo simples em que os usuários possam inserir suas preferências e receber um itinerário estruturado. Comece criando um arquivo travel_ui.py no diretório raiz e adicione o seguinte código a ele.
import streamlit as st
import requestsImportamos bibliotecas básicas como Streamlit e solicitações de suporte à interface do usuário.
st.set_page_config(page_title="ADK-Powered Travel Planner", page_icon="✈️")
st.title("🌍 ADK-Powered Travel Planner")
origin = st.text_input("Where are you flying from?", placeholder="e.g., New York")
destination = st.text_input("Destination", placeholder="e.g., Paris")
start_date = st.date_input("Start Date")
end_date = st.date_input("End Date")
budget = st.number_input("Budget (in USD)", min_value=100, step=50)
if st.button("Plan My Trip ✨"):
if not all([origin, destination, start_date, end_date, budget]):
st.warning("Please fill in all the details.")
else:
payload = {
"origin": origin,
"destination": destination,
"start_date": str(start_date),
"end_date": str(end_date),
"budget": budget
}
response = requests.post("http://localhost:8000/run", json=payload)
if response.ok:
data = response.json()
st.subheader("✈️ Flights")
st.markdown(data["flights"])
st.subheader("🏨 Stays")
st.markdown(data["stay"])
st.subheader("🗺️ Activities")
st.markdown(data["activities"])
else:
st.error("Failed to fetch travel plan. Please try again.")O aplicativo Streamlit oferece uma interface de usuário fácil de usar para interagir com o planejador de viagens multiagente criado com o ADK. Aqui estão alguns aspectos que abordamos no código acima.
- Ele usa
text_input,date_inputenumber_inputpara coletar origem, destino, datas e orçamento. - Ao clicar em "Plan My Trip" (Planejar minha viagem), ele valida a entrada para garantir que nenhum campo seja deixado em branco.
- Se for válido, ele constrói uma carga útil JSON e envia uma solicitação POST para o site
host_agentemhttp://localhost:8000/run. - O site
host_agentinvoca todos os agentes secundários (voo, estadia, atividade), agrega suas respostas e retorna um plano de viagem unificado. - A resposta é analisada e exibida usando o método
st.markdown()em cabeçalhos separados para voos, estadias e atividades. - Se o backend falhar, uma mensagem de erro de fallback será exibida usando
st.error().
Agora, execute o seguinte comando em seu terminal localmente:
uvicorn agents.host_agent.__main__:app --port 8000 &
uvicorn agents.flight_agent.__main__:app --port 8001 &
uvicorn agents.stay_agent.__main__:app --port 8002 &
uvicorn agents.activities_agent.__main__:app --port 8003 &
streamlit run travel_ui.pyQuando um usuário clica em "Planejar minha viagem", o agente host assume o controle, ativa os agentes e exibe os resultados na interface do usuário:
Sua estrutura geral de arquivos seria semelhante a esta:
ADK_demo/
├── agents/
│ ├── host_agent/
│ │ ├── agent.py
│ │ ├── task_manager.py
│ │ ├── __main__.py
│ │ └── .well-known/
│ │ └── agent.json
│ │
│ ├── flight_agent/
│ │ ├── agent.py
│ │ ├── task_manager.py
│ │ ├── __main__.py
│ │ └── .well-known/
│ │ └── agent.json
│ │
│ ├── stay_agent/
│ │ ├── agent.py
│ │ ├── task_manager.py
│ │ ├── __main__.py
│ │ └── .well-known/
│ │ └── agent.json
│ │
│ └── activities_agent/
│ ├── agent.py
│ ├── task_manager.py
│ ├── __main__.py
│ └── .well-known/
│ └── agent.json
│
├── common/
│ ├── a2a_client.py # Utility to send requests to other agents
│ └── a2a_server.py # Shared FastAPI A2A-compatible server template
│
├── shared/
│ └── schemas.py # Shared Pydantic schema
│
├── streamlit_app.py # Frontend UI for user input and response rendering
├── requirements.txt
└── README.md 
E é isso! Eu reuni tudo o que construímos neste projeto do GitHub.
Conclusão
Com apenas alguns aplicativos FastAPI e agentes ADK, criamos um planejador de viagens colaborativo que:
- Comunica-se por meio do protocolo A2A
- Usa agentes do LLM para voos, estadias e atividades
- Agrega e exibe os resultados em uma interface de usuário Streamlit limpa
Embora todos os agentes usem o mesmo modelo, o comportamento desse sistema é multiagente, ou seja, os agentes têm funções distintas, responsabilidades isoladas e comunicação estruturada.
Aqui estão alguns recursos para você começar:
