De tu primer agente en Python con Gemini y el Agent Development Kit, hasta una arquitectura de producción con agentes secuenciales, paralelos y balanceo de carga en Vertex AI Agent Engine.
0 / 10 completado
Nivel 1 · Fundamentos
Tu primer agente con el ADK
Objetivo: instalar el Agent Development Kit (ADK) de Google, configurar Gemini y correr tu primer agente.
El Agent Development Kit (ADK) es el framework open source y code-first de Google para construir agentes con Gemini (o cualquier modelo vía LiteLLM), pensado para escalar directo a Vertex AI en producción.
1. Instala el ADK
TERMINAL
pip install google-adk
export GOOGLE_API_KEY="tu-gemini-api-key"
# o, si usás Vertex AI directamente:
# export GOOGLE_GENAI_USE_VERTEXAI=true
# export GOOGLE_CLOUD_PROJECT="tu-proyecto-gcp"
2. Tu primer agente
agent.py
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
agent = LlmAgent(
name="asistente",
model="gemini-2.5-flash",
instruction="Respondé de forma breve y clara.",
)
session_service = InMemorySessionService()
runner = Runner(agent=agent, app_name="mi_app", session_service=session_service)
async def main():
session = await session_service.create_session(app_name="mi_app", user_id="u1")
contenido = types.Content(role="user", parts=[types.Part(text="¿Qué es un agente de IA, en una frase?")])
async for evento in runner.run_async(user_id="u1", session_id=session.id, new_message=contenido):
if evento.is_final_response():
print(evento.content.parts[0].text)
A diferencia de otros SDKs, el ADK separa explícitamente el Agent (la definición) del Runner (quién lo ejecuta) y el SessionService (dónde vive el estado) — esta separación es clave cuando más adelante quieras desplegar en Agent Engine.
Reto
Corré el mismo agente con adk web desde la terminal para probarlo en la interfaz de desarrollo local del ADK.
Nivel 2 · Personalidad
Instrucciones y parámetros del modelo
Objetivo: controlar el comportamiento del agente con instruction y la configuración de generación.
agent.py
from google.adk.agents import LlmAgent
from google.genai import types
agent = LlmAgent(
name="soporte",
model="gemini-2.5-flash",
instruction=(
"Sos un asistente técnico de soporte. Respondé en español, "
"breve y con pasos numerados. Si no sabés algo, decilo."
),
generate_content_config=types.GenerateContentConfig(
temperature=0.3,
max_output_tokens=400,
),
)
El campo instruction equivale al system prompt de otros frameworks. Podés usar un string o una función que genera la instrucción dinámicamente según el estado de la sesión.
Reto
Cambiá instruction por una función que incluya la hora actual en el prompt de sistema, para que el agente pueda razonar sobre "ahora".
Nivel 3 · Herramientas
Tools: funciones Python normales
Objetivo: darle herramientas al agente usando funciones Python comunes, sin decoradores.
Una particularidad agradable del ADK: cualquier función Python con type hints y docstring ya es una tool válida — no hace falta un decorador @tool, solo pasarla en la lista tools=[].
agent.py
def clima_actual(ciudad: str) -> dict:
"""Devuelve el clima actual reportado para una ciudad dada.
Args:
ciudad: nombre de la ciudad, ej. "Ciudad de Guatemala"
Returns:
Un diccionario con el estado del clima.
"""
datos = {"Ciudad de Guatemala": "22°C, parcialmente nublado"}
return {"clima": datos.get(ciudad, "sin datos")}
agent = LlmAgent(
name="asistente_clima",
model="gemini-2.5-flash",
instruction="Ayudás a consultar el clima de ciudades.",
tools=[clima_actual],
)
Reto
Agregá una segunda función buscar_producto(nombre: str) -> dict y confirmá que Gemini elige la herramienta correcta según la pregunta.
Nivel 4 · Memoria
Sesiones y estado
Objetivo: mantener contexto entre turnos usando el mismo session_id, y entender el estado de sesión (session.state).
chat_con_memoria.py
session = await session_service.create_session(app_name="mi_app", user_id="cliente-42")
async def hablar(texto):
contenido = types.Content(role="user", parts=[types.Part(text=texto)])
async for evento in runner.run_async(user_id="cliente-42", session_id=session.id, new_message=contenido):
if evento.is_final_response():
return evento.content.parts[0].text
print(await hablar("Busco una laptop para diseño gráfico"))
print(await hablar("¿Cuál es la más barata de las que me mencionaste?"))
# Ambas llamadas comparten el mismo session.id, así que el historial se mantiene
Más allá del historial de mensajes, session.state es un diccionario que tus tools pueden leer y escribir — útil para guardar preferencias del usuario o resultados intermedios entre agentes (lo vas a usar en el Nivel 8).
Reto
Guardá en session.state el nombre del usuario la primera vez que lo mencione, y hacé que el agente lo use en respuestas posteriores.
Nivel 5 · Salida estructurada
output_schema con Pydantic
Objetivo: forzar una respuesta JSON validada con output_schema.
agent.py
from pydantic import BaseModel
from google.adk.agents import LlmAgent
class TicketSoporte(BaseModel):
categoria: str
urgencia: str
resumen: str
agent = LlmAgent(
name="clasificador",
model="gemini-2.5-flash",
instruction="Clasificá el ticket de soporte que te llega.",
output_schema=TicketSoporte,
output_key="ticket_clasificado", # se guarda en session.state
)
Ojo con esta particularidad del ADK: cuando definís output_schema, el agente ya no puede usar tools en esa misma llamada — el modelo se enfoca exclusivamente en producir el JSON. Si necesitás tools y salida estructurada a la vez, separá el trabajo en dos agentes (uno que investiga con tools, otro que estructura el resultado final).
Reto
Armá un flujo de 2 agentes: uno con tools que resuelve la tarea, y un segundo con output_schema que solo formatea la respuesta del primero.
Nivel 6 · Streaming
Eventos en tiempo real con run_async
Objetivo: procesar eventos del agente a medida que ocurren, no solo la respuesta final.
runner.run_async() ya devuelve un stream de eventos — lo que hiciste en niveles anteriores filtrando solo is_final_response() es en realidad descartar eventos intermedios útiles (llamadas a tools, respuestas parciales).
stream.py
async for evento in runner.run_async(user_id="u1", session_id=session.id, new_message=contenido):
if evento.get_function_calls():
for llamada in evento.get_function_calls():
print(f"[tool] llamando a {llamada.name}({llamada.args})")
elif evento.content and evento.content.parts:
for parte in evento.content.parts:
if parte.text:
print(parte.text, end="", flush=True)
Para audio/voz en tiempo real, el ADK también expone run_live(), pensado para conversaciones bidireccionales de baja latencia.
Reto
Envolvé este loop de eventos en un generador async y exponelo vía WebSocket para un chat en vivo.
Nivel 7 · Multi-agente
sub_agents: tu primer equipo de agentes
Objetivo: componer varios agentes especializados usando sub_agents, con delegación dinámica.
El ADK soporta multi-agente nativamente: un agente raíz con sub_agents puede transferir el control a uno de ellos según el modelo decida — el sub-agente correcto se activa según su description, igual que una tool.
equipo_agentes.py
from google.adk.agents import LlmAgent
agente_ventas = LlmAgent(
name="ventas",
model="gemini-2.5-flash",
description="Especialista en preguntas de precios y productos.",
instruction="Respondé preguntas de ventas de forma concisa.",
)
agente_soporte = LlmAgent(
name="soporte",
model="gemini-2.5-flash",
description="Especialista en errores técnicos y bugs.",
instruction="Ayudá a resolver problemas técnicos paso a paso.",
)
agente_raiz = LlmAgent(
name="recepcion",
model="gemini-2.5-flash",
instruction="Analizá cada mensaje y delegá al sub-agente correcto.",
sub_agents=[agente_ventas, agente_soporte],
)
El agente_raiz nunca resuelve preguntas de dominio él mismo — su única responsabilidad es decidir a cuál de sus sub_agents transferir el turno.
Reto
Agregá un tercer sub-agente de facturación y probá una conversación donde el agente raíz transfiera correctamente según la intención detectada.
Nivel 8 · Orquestador
SequentialAgent y ParallelAgent
Objetivo: usar orquestación determinista con agentes de flujo de trabajo, en vez de dejar que el modelo decida el orden.
Cuando el orden de ejecución no debe depender del razonamiento del modelo (por ejemplo: siempre investigar → redactar → revisar, en ese orden exacto), el ADK ofrece agentes de workflow deterministas: SequentialAgent ejecuta sus hijos en orden fijo, ParallelAgent los corre al mismo tiempo.
orquestador.py
from google.adk.agents import LlmAgent, SequentialAgent, ParallelAgent
investigador = LlmAgent(
name="investigador", model="gemini-2.5-flash",
instruction="Investigá datos técnicos concisos sobre el tema.",
output_key="investigacion", # queda disponible en session.state["investigacion"]
)
redactor = LlmAgent(
name="redactor", model="gemini-2.5-flash",
instruction="Usá {investigacion} de session.state para escribir un párrafo claro.",
output_key="borrador",
)
revisor = LlmAgent(
name="revisor", model="gemini-2.5-flash",
instruction="Corregí gramática y claridad de {borrador} de session.state.",
)
pipeline = SequentialAgent(
name="pipeline_contenido",
sub_agents=[investigador, redactor, revisor],
)
# Cada agente pasa su output al siguiente automáticamente vía session.state
Con ParallelAgent, varios sub-agentes corren en threads separados pero comparten el mismo session.state — por eso cada uno debe escribir en una output_key distinta, para evitar condiciones de carrera.
Reto
Reemplazá el SequentialAgent por un ParallelAgent para investigador y un segundo investigador (otra fuente), y compará tiempos de ejecución.
Nivel 9 · Producción
Evaluación y observabilidad
Objetivo: evaluar la calidad del agente y trazar su comportamiento antes de desplegarlo.
TERMINAL
adk eval mi_paquete_agente ruta/al/set_de_evaluacion.json
# Corre el agente contra casos de prueba definidos y compara
# la respuesta real vs. la esperada, con métricas de calidad
El ADK trae un framework de evaluación integrado (adk eval) para correr regresiones antes de cada release del agente
La interfaz de desarrollo local (adk web) muestra cada llamada a tool y cada paso del razonamiento para depurar
Guardrails básicos: limitar max_output_tokens, validar inputs de usuario antes de pasarlos al agente, y usar before_model_callback / after_model_callback para inspeccionar o bloquear contenido
Reto
Escribí un before_model_callback que bloquee cualquier mensaje que contenga una palabra de una lista prohibida, antes de que llegue al modelo.
Nivel 10 · Arquitectura final
Vertex AI Agent Engine + balanceo de carga
Objetivo: desplegar el pipeline completo a Vertex AI Agent Engine, con escalado automático y carga distribuida.
Google recomienda desplegar agentes del ADK directamente en Vertex AI Agent Engine: un runtime completamente administrado que se encarga del escalado horizontal — no arma vos mismo el load balancer, se lo indicás como configuración de despliegue.
1. Desplegar el pipeline a Agent Engine
deploy.py
from vertexai import agent_engines
app = agent_engines.create(
agent_engine=pipeline, # el SequentialAgent del Nivel 8
requirements=["google-adk", "google-cloud-aiplatform"],
display_name="pipeline-contenido-prod",
min_instances=2, # nunca menos de 2 réplicas activas (evita cold starts)
max_instances=20, # techo de auto-escalado ante picos de tráfico
)
print(app.resource_name)
2. Balanceo de carga entre despliegues regionales
Para arquitecturas multi-región, ponés varios despliegues de Agent Engine (uno por región) detrás de un Google Cloud Load Balancer global, que enruta cada request a la región con menor latencia/carga disponible:
Con esto tenés el camino completo: de un LlmAgent de pocas líneas en el Nivel 1, a un pipeline multi-agente desplegado en múltiples regiones, con auto-escalado y evaluación continua de calidad.
Reto final
Desplegá el pipeline del Nivel 8 en Agent Engine con min_instances=2 y medí el tiempo de cold-start comparado con min_instances=0.