1. Introducción
Descripción general
En este lab, irás más allá de los chatbots simples y crearás un sistema multiagente distribuido.
Si bien un solo LLM puede responder preguntas, la complejidad del mundo real a menudo requiere roles especializados. No le pides a tu ingeniero de backend que diseñe la IU, y no le pides a tu diseñador que optimice las consultas de la base de datos. Del mismo modo, podemos crear agentes de IA especializados que se enfoquen en una tarea y se coordinen entre sí para resolver problemas complejos.
Compilarás un sistema de creación de cursos que constará de los siguientes elementos:
- Agente de investigación: Usa
google_searchpara encontrar información actualizada. - Agente de evaluación: Critica la investigación en cuanto a calidad y completitud.
- Agente de Content Builder: Convierte la investigación en un curso estructurado.
- Agente organizador: Administra el flujo de trabajo y la comunicación entre estos especialistas.
Requisitos previos
- Conocimientos básicos de Python
- Conocimientos sobre la consola de Google Cloud
Actividades
- Define un agente que usa herramientas (
researcher) que puede buscar en la Web. - Implementa un resultado estructurado con Pydantic para
judge. - Conéctate a agentes remotos con el protocolo Agent-to-Agent (A2A).
- Construye un
LoopAgentpara crear un ciclo de retroalimentación entre el investigador y el evaluador. - Ejecuta el sistema distribuido de forma local con el ADK.
- Implementa el sistema multiagente en Google Cloud Run.
Principios de arquitectura y organización
Antes de escribir código, comprendamos cómo trabajan juntos estos agentes. Estamos creando una canalización de creación de cursos.
El diseño del sistema
Organización con agentes
Los agentes estándar (como el Investigador) funcionan. Los agentes orquestadores (como LoopAgent o SequentialAgent) administran otros agentes. No tienen sus propias herramientas; su "herramienta" es la delegación.
LoopAgent: Actúa como un buclewhileen el código. Ejecuta una secuencia de agentes de forma repetida hasta que se cumple una condición (o se alcanza la cantidad máxima de iteraciones). Usamos esta información para la Investigación con Loop:- El investigador encuentra información.
- El juez la critica.
- Si Judge dice "Fail", EscalationChecker permite que continúe el bucle.
- Si Judge dice "Pass", EscalationChecker interrumpe el bucle.
SequentialAgent: Actúa como una ejecución de secuencia de comandos estándar. Ejecuta los agentes uno tras otro. Usamos esto para la canalización de alto nivel:- Primero, ejecuta el Circuito de investigación (hasta que finalice con datos adecuados).
- Luego, ejecuta Content Builder (para escribir el curso).
Si combinamos estos elementos, creamos un sistema sólido que puede autocorregirse antes de generar el resultado final.
2. Configuración
Configuración del entorno
Abre Cloud Shell: Abre una pestaña nueva y escribe shell.cloud.google.com.
Obtén el código de partida
- Clona el repositorio de inicio en tu directorio principal:
cd ~ git clone https://github.com/amitkmaraj/prai-roadshow-lab-1-starter.git cd prai-roadshow-lab-1-starter - Ejecuta la secuencia de comandos init para asociar los créditos de la rampa de acceso con la facturación.
chmod +x ./init.sh ./init.sh - Abre esta carpeta en tu editor.
Habilita las APIs
Ahora que tienes un proyecto nuevo, ejecuta el siguiente comando para habilitar los servicios de Google Cloud necesarios:
gcloud services enable \
run.googleapis.com \
artifactregistry.googleapis.com \
cloudbuild.googleapis.com \
aiplatform.googleapis.com \
compute.googleapis.com
Esto podría demorar unos segundos.
Instala las dependencias
Usamos uv para administrar dependencias rápidamente.
- Instala las dependencias del proyecto:
# Ensure you have uv installed: pip install uv uv sync - Establece el ID de tu proyecto de Google Cloud.
- Sugerencia: Puedes encontrar el ID del proyecto en el panel de la consola de Cloud o ejecutando
gcloud config get-value project.
export GOOGLE_CLOUD_PROJECT=$(gcloud config get-value project) - Sugerencia: Puedes encontrar el ID del proyecto en el panel de la consola de Cloud o ejecutando
- Configura las variables de entorno restantes:
export GOOGLE_CLOUD_LOCATION=us-central1 export GOOGLE_GENAI_USE_VERTEXAI=true
3. 🕵️ El agente de investigación
El Investigador es un especialista. Su único trabajo es encontrar información. Para ello, necesita acceder a una herramienta: la Búsqueda de Google.
¿Por qué separar al investigador?
Análisis detallado: ¿Por qué no tener un solo agente que haga todo?
Los agentes pequeños y enfocados son más fáciles de evaluar y depurar. Si la investigación es deficiente, itera sobre la instrucción del investigador. Si el formato del curso es incorrecto, debes iterar en el Creador de contenido. En una instrucción monolítica que "lo hace todo", corregir una cosa a menudo rompe otra.
- Si trabajas en Cloud Shell, ejecuta el siguiente comando para abrir el editor de Cloud Shell:
cloudshell workspace . - Abre
agents/researcher/agent.py. - Verás un esqueleto con un TODO.
- Agrega el siguiente código para definir el agente
researcher:# ... existing imports ... # Define the Researcher Agent researcher = Agent( name="researcher", model=MODEL, description="Gathers information on a topic using Google Search.", instruction=""" You are an expert researcher. Your goal is to find comprehensive and accurate information on the user's topic. Use the `google_search` tool to find relevant information. Summarize your findings clearly. If you receive feedback that your research is insufficient, use the feedback to refine your next search. """, tools=[google_search], ) root_agent = researcher
Concepto clave: Uso de herramientas
Observa que pasamos tools=[google_search]. El ADK controla la complejidad de describir esta herramienta al LLM. Cuando el modelo decide que necesita información, genera una llamada a herramienta estructurada, el ADK ejecuta la función de Python google_search y devuelve el resultado al modelo.
4. ⚖️ El agente juez
El investigador trabaja mucho, pero los LLM pueden ser perezosos. Necesitamos un juez para que revise el trabajo. El juez acepta la investigación y devuelve una evaluación estructurada de aprobado o no aprobado.
Resultado estructurado
Análisis detallado: Para automatizar los flujos de trabajo, necesitamos resultados predecibles. Una opinión de texto divagante es difícil de analizar de forma programática. Al aplicar un esquema JSON (con Pydantic), nos aseguramos de que Judge devuelva un valor booleano pass o fail sobre el que nuestro código pueda actuar de manera confiable.
- Abre
agents/judge/agent.py. - Define el esquema
JudgeFeedbacky el agentejudge.# 1. Define the Schema class JudgeFeedback(BaseModel): """Structured feedback from the Judge agent.""" status: Literal["pass", "fail"] = Field( description="Whether the research is sufficient ('pass') or needs more work ('fail')." ) feedback: str = Field( description="Detailed feedback on what is missing. If 'pass', a brief confirmation." ) # 2. Define the Agent judge = Agent( name="judge", model=MODEL, description="Evaluates research findings for completeness and accuracy.", instruction=""" You are a strict editor. Evaluate the 'research_findings' against the user's original request. If the findings are missing key info, return status='fail'. If they are comprehensive, return status='pass'. """, output_schema=JudgeFeedback, # Disallow delegation because it should only output the schema disallow_transfer_to_parent=True, disallow_transfer_to_peers=True, ) root_agent = judge
Concepto clave: Restricción del comportamiento del agente
Establecemos disallow_transfer_to_parent=True y disallow_transfer_to_peers=True. Esto obliga al juez a solo devolver el JudgeFeedback estructurado. No puede decidir "chatear" con el usuario ni delegar la tarea a otro agente. Esto lo convierte en un componente determinístico en nuestro flujo de lógica.
5. 🧪 Cómo realizar pruebas de forma aislada
Antes de conectarlos, podemos verificar que cada agente funcione. El ADK te permite ejecutar agentes de forma individual.
Concepto clave: El tiempo de ejecución interactivo
adk run inicia un entorno ligero en el que tú eres el "usuario". Esto te permite probar las instrucciones del agente y el uso de herramientas de forma aislada. Si el agente falla aquí (p.ej., no puede usar la Búsqueda de Google), definitivamente fallará en la orquestación.
- Ejecuta Researcher de forma interactiva. Ten en cuenta que apuntamos al directorio del agente específico:
# This runs the researcher agent in interactive mode uv run adk run agents/researcher - En la instrucción del chat, escribe lo siguiente:
Find the population of Tokyo in 2020export GOOGLE_CLOUD_PROJECT=$(gcloud config get-value project) export GOOGLE_CLOUD_LOCATION=us-central1 export GOOGLE_GENAI_USE_VERTEXAI=true - Sal del chat (Ctrl + C).
- Ejecuta Judge de forma interactiva:
uv run adk run agents/judge - En la instrucción del chat, simula la entrada:
Topic: Tokyo. Findings: Tokyo is a city.status='fail'porque los resultados son demasiado breves.
6. ✍️ El agente Creador de contenido
El Creador de contenido es el escritor creativo. Toma la investigación aprobada y la convierte en un curso.
- Abre
agents/content_builder/agent.py. - Define el agente
content_builder.content_builder = Agent( name="content_builder", model=MODEL, description="Transforms research findings into a structured course.", instruction=""" You are an expert course creator. Take the approved 'research_findings' and transform them into a well-structured, engaging course module. **Formatting Rules:** 1. Start with a main title using a single `#` (H1). 2. Use `##` (H2) for main section headings. 3. Use bullet points and clear paragraphs. 4. Maintain a professional but engaging tone. Ensure the content directly addresses the user's original request. """, ) root_agent = content_builder
Concepto clave: Propagación del contexto
Quizás te preguntes: "¿Cómo sabe el Creador de contenido lo que encontró el Investigador?". En el ADK, los agentes de una canalización comparten un session.state. Más adelante, en el orquestador, configuraremos el investigador y el juez para que guarden sus resultados en este estado compartido. La instrucción de Content Builder tiene acceso efectivo a este historial.
7. 🎻 El organizador
El Orchestrator es el administrador de nuestro equipo de varios agentes. A diferencia de los agentes especializados (Investigador, Juez y Creador de contenido) que realizan tareas específicas, el trabajo del Organizador es coordinar el flujo de trabajo y garantizar que la información fluya correctamente entre ellos.
🌐 La arquitectura: Agent-to-Agent (A2A)
En este lab, crearemos un sistema distribuido. En lugar de ejecutar todos los agentes en un solo proceso de Python, los implementamos como microservicios independientes. Esto permite que cada agente se escale de forma independiente y falle sin que se bloquee todo el sistema.
Para que esto sea posible, usamos el protocolo Agent-to-Agent (A2A).
El protocolo A2A
Análisis detallado: En un sistema de producción, los agentes se ejecutan en diferentes servidores (o incluso en diferentes nubes). El protocolo A2A crea una forma estándar para que se descubran y se comuniquen entre sí a través de HTTP. RemoteA2aAgent es el cliente del ADK para este protocolo.
- Abre
agents/orchestrator/agent.py. - Ubica el comentario
# TODO: Define Remote Agentso la sección para las definiciones de agentes remotos. - Agrega el siguiente código para definir las conexiones. Asegúrate de colocarlo después de las importaciones y antes de cualquier otra definición de agente.
# ... existing code ... # Connect to the Researcher (Localhost port 8001) researcher_url = os.environ.get("RESEARCHER_AGENT_CARD_URL", "http://localhost:8001/a2a/agent/.well-known/agent-card.json") researcher = RemoteA2aAgent( name="researcher", agent_card=researcher_url, description="Gathers information using Google Search.", # IMPORTANT: Save the output to state for the Judge to see after_agent_callback=create_save_output_callback("research_findings"), # IMPORTANT: Use authenticated client for communication httpx_client=create_authenticated_client(researcher_url) ) # Connect to the Judge (Localhost port 8002) judge_url = os.environ.get("JUDGE_AGENT_CARD_URL", "http://localhost:8002/a2a/agent/.well-known/agent-card.json") judge = RemoteA2aAgent( name="judge", agent_card=judge_url, description="Evaluates research.", after_agent_callback=create_save_output_callback("judge_feedback"), httpx_client=create_authenticated_client(judge_url) ) # Content Builder (Localhost port 8003) content_builder_url = os.environ.get("CONTENT_BUILDER_AGENT_CARD_URL", "http://localhost:8003/a2a/agent/.well-known/agent-card.json") content_builder = RemoteA2aAgent( name="content_builder", agent_card=content_builder_url, description="Builds the course.", httpx_client=create_authenticated_client(content_builder_url) )
8. 🛑 El verificador de derivaciones
Un bucle necesita una forma de detenerse. Si el juez dice "Aprobar", queremos salir del bucle de inmediato y pasar al Creador de contenido.
Lógica personalizada con BaseAgent
Análisis detallado: No todos los agentes usan LLMs. A veces, necesitas una lógica de Python simple. BaseAgent te permite definir un agente que solo ejecuta código. En este caso, verificamos el estado de la sesión y usamos EventActions(escalate=True) para indicarle a LoopAgent que se detenga.
- Aún en
agents/orchestrator/agent.py. - Busca el marcador de posición TODO de
EscalationChecker. - Reemplázala por la siguiente implementación:
class EscalationChecker(BaseAgent): """Checks the judge's feedback and escalates (breaks the loop) if it passed.""" async def _run_async_impl( self, ctx: InvocationContext ) -> AsyncGenerator[Event, None]: # Retrieve the feedback saved by the Judge feedback = ctx.session.state.get("judge_feedback") print(f"[EscalationChecker] Feedback: {feedback}") # Check for 'pass' status is_pass = False if isinstance(feedback, dict) and feedback.get("status") == "pass": is_pass = True # Handle string fallback if JSON parsing failed elif isinstance(feedback, str) and '"status": "pass"' in feedback: is_pass = True if is_pass: # 'escalate=True' tells the parent LoopAgent to stop looping yield Event(author=self.name, actions=EventActions(escalate=True)) else: # Continue the loop yield Event(author=self.name) escalation_checker = EscalationChecker(name="escalation_checker")
Concepto clave: Control de flujo a través de eventos
Los agentes se comunican no solo con texto, sino también con eventos. Cuando se genera un evento con escalate=True, este agente envía un indicador a su elemento superior (el LoopAgent). El LoopAgent está programado para captar este indicador y finalizar el bucle.
9. 🔁 El ciclo de investigación
Necesitamos un circuito de retroalimentación: Investigación -> Evaluación -> (Falla) -> Investigación -> …
- Aún en
agents/orchestrator/agent.py. - Agrega la definición
research_loop. Coloca este elemento después de la claseEscalationCheckery la instanciaescalation_checker.research_loop = LoopAgent( name="research_loop", description="Iteratively researches and judges until quality standards are met.", sub_agents=[researcher, judge, escalation_checker], max_iterations=3, )
Concepto clave: LoopAgent
El LoopAgent recorre sus sub_agents en orden.
researcher: Busca datos.judge: Evalúa datos.escalation_checker: Decide si se debeyield Event(escalate=True). Si sucedeescalate=True, el bucle se interrumpe antes. De lo contrario, se reinicia en el investigador (hastamax_iterations).
10. 🔗 La canalización final
Por último, une todo.
- Aún en
agents/orchestrator/agent.py. - Define el
root_agenten la parte inferior del archivo. Asegúrate de que este reemplazo se aplique a cualquier marcador de posiciónroot_agent = Noneexistente.root_agent = SequentialAgent( name="course_creation_pipeline", description="A pipeline that researches a topic and then builds a course from it.", sub_agents=[research_loop, content_builder], )
Concepto clave: Composición jerárquica
Ten en cuenta que research_loop es en sí mismo un agente (un LoopAgent). Lo tratamos como cualquier otro agente secundario en SequentialAgent. Esta componibilidad te permite crear lógica compleja anidando patrones simples (bucles dentro de secuencias, secuencias dentro de routers, etcétera).
11. 💻 Ejecuta de forma local
Antes de ejecutar todo, veamos cómo el ADK simula el entorno distribuido de forma local.
Análisis detallado: Cómo funciona el desarrollo local
En una arquitectura de microservicios, cada agente se ejecuta como su propio servidor. Cuando realices la implementación, tendrás 4 servicios de Cloud Run diferentes. Simular esto de forma local puede ser doloroso si tienes que abrir 4 pestañas de terminal y ejecutar 4 comandos.
Esta secuencia de comandos inicia los procesos de uvicorn para el investigador (puerto 8001), el evaluador (8002) y el creador de contenido (8003). Establece variables de entorno, como RESEARCHER_AGENT_CARD_URL, y las pasa al organizador (puerto 8004). Así es exactamente como lo configuraremos en la nube más adelante.
- Ejecuta la secuencia de comandos de orquestación:
./run_local.sh - Prueba lo siguiente:
- Si usas Cloud Shell: Haz clic en el botón Vista previa en la Web (en la parte superior derecha de la terminal) -> Vista previa en el puerto 8080 -> Cambiar puerto a
8000. - Si ejecutas el servidor de forma local: Abre
http://localhost:8000en tu navegador. - Instrucción: "Crea un curso sobre la historia del café".
- Observa: Orchestrator llamará al investigador. El resultado se envía al juez. Si el juez la rechaza, el bucle continúa.
- "Internal Server Error" o errores de autenticación: Si ves errores de autenticación (p.ej., relacionados con
google-auth), asegúrate de haber ejecutadogcloud auth application-default loginsi se ejecuta en una máquina local. En Cloud Shell, asegúrate de que la variable de entornoGOOGLE_CLOUD_PROJECTesté configurada correctamente. - Errores de terminal: Si el comando falla en una nueva ventana de terminal, recuerda volver a exportar tus variables de entorno (
GOOGLE_CLOUD_PROJECT, etcétera).
- Si usas Cloud Shell: Haz clic en el botón Vista previa en la Web (en la parte superior derecha de la terminal) -> Vista previa en el puerto 8080 -> Cambiar puerto a
- Prueba de agentes de forma aislada: Incluso cuando se ejecuta el sistema completo, puedes probar agentes específicos segmentando sus puertos directamente. Esto es útil para depurar un componente específico sin activar toda la cadena.
- Researcher Only (Port 8001):
http://localhost:8001 - Solo para jueces (puerto 8002):
http://localhost:8002 - Solo Content Builder (puerto 8003):
http://localhost:8003 - Organizador (puerto 8004):
http://localhost:8004(acceso directo a la lógica del organizador)
- Researcher Only (Port 8001):
12. 🚀 Implementa en Cloud Run
La validación final se ejecuta en la nube. Implementaremos cada agente como un servicio independiente.
Información sobre la configuración de la implementación
Cuando implementamos agentes en Cloud Run, pasamos varias variables de entorno para configurar su comportamiento y conectividad:
GOOGLE_CLOUD_PROJECT: Garantiza que el agente use el proyecto de Google Cloud correcto para el registro y las llamadas a Vertex AI.GOOGLE_GENAI_USE_VERTEXAI: Indica al framework del agente (ADK) que use Vertex AI para la inferencia del modelo en lugar de llamar directamente a las APIs de Gemini.[AGENT]_AGENT_CARD_URL: Es fundamental para el orquestador. Le indica al orquestador dónde encontrar los agentes remotos. Si configuras este parámetro en la URL de Cloud Run implementada (específicamente, la ruta de la tarjeta del agente), permitimos que el orquestador descubra al investigador, al juez y al creador de contenido, y se comunique con ellos a través de Internet.
- Implementa el investigador:
gcloud run deploy researcher \ --source agents/researcher/ \ --region us-central1 \ --allow-unauthenticated \ --labels dev-tutorial=prod-ready-1 \ --set-env-vars GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT \ --set-env-vars GOOGLE_GENAI_USE_VERTEXAI="true"RESEARCHER_URL=$(gcloud run services describe researcher --region us-central1 --format='value(status.url)') echo $RESEARCHER_URL - Implementa el juez:
gcloud run deploy judge \ --source agents/judge/ \ --region us-central1 \ --allow-unauthenticated \ --labels dev-tutorial=prod-ready-1 \ --set-env-vars GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT \ --set-env-vars GOOGLE_GENAI_USE_VERTEXAI="true"JUDGE_URL=$(gcloud run services describe judge --region us-central1 --format='value(status.url)') echo $JUDGE_URL - Implementa Content Builder:
gcloud run deploy content-builder \ --source agents/content_builder/ \ --region us-central1 \ --allow-unauthenticated \ --labels dev-tutorial=prod-ready-1 \ --set-env-vars GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT \ --set-env-vars GOOGLE_GENAI_USE_VERTEXAI="true"CONTENT_BUILDER_URL=$(gcloud run services describe content-builder --region us-central1 --format='value(status.url)') echo $CONTENT_BUILDER_URL - Implementa el Orchestrator: Usa las variables de entorno capturadas para configurar el Orchestrator.
gcloud run deploy orchestrator \ --source agents/orchestrator/ \ --region us-central1 \ --allow-unauthenticated \ --labels dev-tutorial=prod-ready-1 \ --set-env-vars RESEARCHER_AGENT_CARD_URL=$RESEARCHER_URL/a2a/agent/.well-known/agent-card.json \ --set-env-vars JUDGE_AGENT_CARD_URL=$JUDGE_URL/a2a/agent/.well-known/agent-card.json \ --set-env-vars CONTENT_BUILDER_AGENT_CARD_URL=$CONTENT_BUILDER_URL/a2a/agent/.well-known/agent-card.json \ --set-env-vars GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT \ --set-env-vars GOOGLE_GENAI_USE_VERTEXAI="true"ORCHESTRATOR_URL=$(gcloud run services describe orchestrator --region us-central1 --format='value(status.url)') echo $ORCHESTRATOR_URL - Implementa el frontend:
gcloud run deploy course-creator \ --source app \ --region us-central1 \ --allow-unauthenticated \ --labels dev-tutorial=prod-ready-1 \ --set-env-vars AGENT_SERVER_URL=$ORCHESTRATOR_URL \ --set-env-vars GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT - Prueba la implementación remota: Abre la URL de tu Orchestrator implementado. Ahora se ejecuta por completo en la nube y utiliza la infraestructura sin servidores de Google para escalar tus agentes.Sugerencia: Encontrarás todos los microservicios y sus URLs en la interfaz de Cloud Run.
13. Resumen
¡Felicitaciones! Compilaste e implementaste correctamente un sistema multiagente distribuido listo para producción.
Qué logramos
- Descomposición de una tarea compleja: En lugar de una instrucción gigante, dividimos el trabajo en roles especializados (investigador, juez y creador de contenido).
- Control de calidad implementado: Utilizamos un
LoopAgenty unJudgeestructurado para garantizar que solo la información de alta calidad llegue al paso final. - Creado para la producción: Con el protocolo Agent-to-Agent (A2A) y Cloud Run, creamos un sistema en el que cada agente es un microservicio independiente y escalable. Esto es mucho más sólido que ejecutar todo en una sola secuencia de comandos de Python.
- Orquestación: Usamos
SequentialAgentyLoopAgentpara definir patrones de flujo de control claros.
Próximos pasos
Ahora que tienes la base, puedes extender este sistema:
- Agregar más herramientas: Otorga acceso al investigador a documentos internos o APIs.
- Mejora el juez: Agrega criterios más específicos o incluso un paso de "Human in the Loop".
- Swap Models: Prueba usar diferentes modelos para diferentes agentes (p.ej., un modelo más rápido para el juez y un modelo más potente para el redactor de contenido).
Ahora puedes crear flujos de trabajo complejos y confiables basados en agentes en Google Cloud.