CrewAI avanzado 01

1. Introducción

1.1. Qué vamos a ver

En un artículo anterior vimos qué era CrewAI, un framework muy interesante para trabajar con sistemas multiagénticos. Retomo el tema para profundizar, ya sí, en la definición de agentes y tareas, así como diversos temas más complejos, como las herramientas o la memoria.

Como son temas muy extensos, dividiré el artículo en dos partes. La primera estará dedicada a los agentes y las tareas y la segunda al resto de features.

Si no se ha leído el primer artículo y se desconoce todo sobre CrewAI, conviene hacerlo antes de continuar, porque este asume familiaridad con esos conceptos y sube el nivel de complejidad.

Para ir viendo en la práctica cómo funciona, vamos a preparar una nueva tripulación: un equipo que ayude a planificar viajes, aunque cada cual puede desarrollar cualquier otra idea. Lo importante es que resulte divertida.

El proyecto se llamará Itinera y consistirá en un planificador de rutas temáticas capaz de diseñar itinerarios personalizados combinando cultura y naturaleza. El ejemplo concreto que usaremos a lo largo del artículo es una ruta de 9 días por el norte de Italia para alguien que ya ha visitado la Toscana y quiere evitar repetir destinos. El sistema recibe la zona, los lugares ya visitados, la duración del viaje, los intereses y el ritmo deseado, y devuelve un itinerario día a día con paradas, contexto y orden geográfico lógico.

Para construirlo veremos algunos conceptos clave como:

1. Herramientas personalizadas.

En el primer artículo usamos herramientas que CrewAI incluye de serie. Aquí crearemos herramientas propias con el decorador @tool, conectadas a APIs reales: Wikipedia para información cultural e histórica, OpenStreetMap/Nominatim para coordenadas y puntos de interés, y Open-Meteo para previsión meteorológica. Las tres son gratuitas y no requieren registro ni clave de API.

2. Proceso jerárquico.

En el proyecto anterior, los agentes trabajaban en secuencia (Process.sequential): uno terminaba y el siguiente tomaba el relevo. Aquí usaremos Process.hierarchical, donde un agente manager llamado Herodoto coordina al resto, decide el orden de actuación y pondera los resultados antes de producir la salida final.

3. Human-in-the-loop.

Antes de coordinar al equipo, Herodoto formulará una pregunta de confirmación al usuario. Esto permite corregir parámetros o añadir matices sin reescribir el código.

4. Outputs estructurados con Pydantic.

En lugar de devolver texto libre, los agentes devolverán objetos Python validados contra un esquema. Esto hace que los datos sean predecibles y fáciles de procesar en fases posteriores.

5. Kickoff asíncrono.

La función kickoff() bloquea el hilo principal hasta que la crew termina. Con kickoff_async() la ejecución se lanza como una corrutina, lo que permite, por ejemplo, lanzar varias crews en paralelo o mantener una interfaz responsiva mientras se procesa.

6. Caché de resultados y memoria.

CrewAI puede almacenar el resultado de una tarea para no repetirla si los inputs no han cambiado. En un sistema que hace varias llamadas a APIs externas, esto ahorra tiempo y reduce el consumo de tokens. Además, veremos cómo añadir una memoria al sistema.

7. Guardrails.

Son validaciones que se ejecutan antes de que la crew empiece a trabajar. Sirven para detectar inputs malformados, incompletos o fuera del dominio del sistema, y devolver un error claro en lugar de dejar que un agente falle a mitad de la ejecución.

En conjunto, creo que no solo nos servirá para aprender a manejar mejor CrewAI, sino también conocer algunos conceptos fundamentales de la arquitectura agéntica como outputs estructurados o guardrails.

1.2. Instalación

Para empezar el proyecto hay que seguir los mismos pasos que en el anterior:

# Instalamos CrewAI
uv tool install crewai

# Creamos el proyecto con el nombre normalizado que queramos,
# en ese caso, itinera
crewai create crew itinera

# Entramos en el proyecto
cd itinera

# Instalamos las dependencias
crewai install

# Añadimos el proveedor que vayamos a usar
uv add "crewai[anthropic]"

Además, vamos a añadir requests, una librería para consumir APIs REST de forma sencilla.

uv add requests

Por último, en .env, indicamos el api key del proveedor y con qué modelo vamos a querer trabajar. Para este tutorial vale cualquiera una miajita potente, como es el caso de haiku 4.5.

MODEL=claude-haiku-4-5-20251001
ANTHROPIC_API_KEY=sk-patatas

2. Diseño del sistema: el equipo

2.1. Un equipo de especialistas

El equipo puede estar formado por cinco agentes. Cuatro son especialistas y uno es el coordinador. Antes de ver el código, conviene entender qué hace cada uno y por qué el sistema está diseñado así.

1. Herodoto será el manager, el coordinador. Recibe los parámetros del viaje, decide qué agentes activan y en qué orden, pondera los resultados y produce la salida final. No tiene personalidad asignada -a diferencia de los agentes del artículo anterior- porque su rol es estructural, no creativo.

2. Habrá dos investigadores, uno cultural y otro de naturaleza, que trabajan en paralelo cuando Herodoto los convoca. Separarlos en dos agentes distintos permite calibrar el peso de cada disciplina según los intereses del usuario: un viaje orientado a naturaleza dará más protagonismo al investigador de naturaleza; uno mixto los equilibrará. Un agente único que lo hiciera todo no permitiría ese control fino.

3. La planificadora toma los resultados de ambos investigadores y diseña la ruta día a día con criterio geográfico: agrupa destinos cercanos, evita backtracking y respeta el ritmo solicitado. Aclaro, por cierto, que el género no influye en las respuesta. Si asigno personajes masculinos y femeninos es solo por enriquecer el lore del asunto.

4. El redactor, por su parte, recibe esa ruta estructurada y la convierte en el documento final que el usuario se lleva: añade contexto, sugiere horarios y da forma narrativa al itinerario. Planificadora y redactor son agentes separados por la misma razón que los dos investigadores: la planificadora optimiza estructura y logística; el redactor optimiza legibilidad y atractivo. Mezclar ambos roles en uno produciría un agente con objetivos en tensión.

2.2. Anatomía de un agente

Pero antes de seguir, vamos a ver a vuelapluma algunos conceptos básicos sobre los agentes. No hace falta ni por asomo aprenderse estos nombres que siguen de memoria, pero sí entender más o menos qué se puede llegar a definir si se necesita.

Según definen en la documentación oficial de CrewAI, un agente es una unidad autónoma que puede realizar tareas específicas, tomar decisiones basadas en su rol y objetivo, usar herramientas para lograr objetivos, comunicarse y colaborar con otros agentes, mantener memoria de las interacciones y delegar tareas cuando se le permita. La idea central es pensar en cada agente como un miembro especializado de un equipo, con habilidades y responsabilidades concretas.

Cuando se construye un agente se pueden definir un montón y medio de parámetros. Los más importantes son los relativos a la identidad, los que hemos estado viendo.

1. Identidad. Definen el qué y el porqué.

role (rol): Su función y especialización. Debe ser muy específico (ej. “Especialista en UX para análisis de entrevistas”).
goal (objetivo): La meta individual que guía sus decisiones. Debe ser clara y medible.
backstory (trasfondo): Da contexto, personalidad y profundidad. Sirve para enriquecer las interacciones y la calidad del resultado, como se vio en el artículo anterior.

Estos tres atributos tienen que contar la misma historia: un goal ambicioso con un backstory de novato no resulta creíble para el modelo.

Otros relevantes son:

2. Control de ejecución. Gestionan los límites para evitar errores o bucles infinitos.

max_iter (máx. iteraciones): Número máximo de intentos que hará el agente antes de dar su mejor respuesta. Valor por defecto: 20. Útil para tareas que podrían requerir varios pasos.
max_execution_time (tiempo máximo): Límite en segundos para ejecutar una tarea. Esencial para evitar que un agente se quede “atascado”.
max_rpm (máx. peticiones por minuto): Evita superar las tasas de límite de la API del LLM.
cache (caché): Activar (true por defecto) mejora el rendimiento en tareas repetitivas al recordar resultados de herramientas.

3. Control del comportamiento y la calidad. Controlan cómo el agente ejecuta las tareas y maneja los recursos.

llm (modelo de lenguaje): Define el modelo que lo potencia (ej. “gpt-4o”, “anthropic/claude-sonnet-4-5”). Para tareas sencillas no tiene sentido usar modelos complejos. En el yaml esto se define como un string, en Python (vd. infra) puedes pasar una instancia LLM(…) con configuración avanzada (temperature, top_p, etc.).
tools (herramientas): Lista de capacidades que el agente puede usar (búsqueda web, análisis de datos, etc.). Sin las herramientas adecuadas, el agente está muy limitado. Este listado de herramientas disponibles hay que definirlo en Python.
reasoning (razonamiento): Si se activa (true), el agente reflexiona y crea un plan antes de actuar. Es ideal para tareas complejas que requieren estrategia, pero, ojo, hay que tener cuidado con los tokens. No hay que usarlo si no se necesita.

4. Colaboración. Cómo el agente interactúa con el resto de la Crew.

allow_delegation (permitir delegación): Si es false (por defecto), el agente trabaja solo. Si es true, puede pasar tareas a otros agentes la Crew.

5. Debugging y observabilidad. Para entender qué está pasando por dentro.

verbose: logs detallados de ejecución (default false)
step_callback: para hooks de monitorización.

6. Memoria y contexto. Sirven para que el agente recuerde y maneje información a lo largo del tiempo.

memory: mantiene historial de interacciones.
respect_context_window (respetar ventana de contexto): Este es un parámetro de seguridad y robustez muy importante. Cuando está activo (true, por defecto), evita que el agente falle al exceder el límite de tokens del LLM, resumiendo automáticamente el contexto. Si es false, el proceso se detendrá con un error.
knowledge_sources — bases de conocimiento específicas del dominio. Van en el Python.
embedder: configuración del modelo de embeddings. Va también en el Python

Hay más, pero con los mencionados basta por ahora para entender la configuración de los agentes. En cambio, sí me interesa insistir en esa diferencia entre el yaml y el Python para que entendamos dónde se define cada tipo de parámetro.

En general, en el yaml, conviene definir la configuración estática y reutilizable, la definición de quién es el agente. Aquí se define todo lo declarativo y estático del agente: quién es, sus límites, sus flags. Por el contrario, en Python se definen al instanciar el agente los parámetros de control de ejecución (max_iter, max_execution_time, respect_context_window, reasoning, etc.).

Y ese merge se realiza también en el Python.

@agent
def my_agent(self) -> Agent:
    agent = Agent(
        config=self.agents_config['my_agent'],
        # --- Lo que NO puede ir en yaml porque son objetos ---
        tools=[SerperDevTool(), MiToolCustom()],
        knowledge_sources=[mi_fuente_pdf],
        step_callback=mi_funcion_de_logging,
    )
    return agent

2.4. Buenas prácticas

No hay que obsesionarse demasiado con con el agente en sí, sino, más bien, invertir tiempo en escribir tareas muy claras, con propósito, formato y pasos definidos. El foco debería estar en uno 80/20 según indica el equipo de CrewAI; pero aquí van unas recomendaciones que indican en la documentación oficial.

1. La triada Role-Goal-Backstory es la más importante. Cada uno de estos atributos cumple una función específica.

Role: la función especializada.
- Hay que ser específico, no genérico: “Especialista en documentación técnica” es mejor que “Escritor”.
- Conviene que sean arquetipos profesionales reconocibles. Escritor mejor que “Buscador de las musas”.
- Es recomendable incluir el dominio de la experiencia: “Analista financiero especializado en tendencias de mercado”.
Goal: el propósito y la motivación.
- Orientado a resultados, hay que definir qué se intenta lograr, no solo la actividad.
- Conviene incluir estándares de calidad, expectativas sobre cómo debe ser el trabajo
- Algo más complejo de hacer, pero igual de recomendable es incorporar criterios de éxito, que el agente sepa qué es lo bueno.
Backstory: la experiencia y perspectiva.
- Establece su trayectoria profesional.
- Definir estilo de trabajo y valores, cómo aborda los problemas.
- Hay que crear una personaje coherente: que role, goal y backstory deben contar la misma historia.

2. Hay que especializar a los agentes al máximo en su rol, pero mantenerlos versátiles en su aplicación práctica. Lo que decía antes, en lugar de usar roles genéricos como “Escritor”, hay usar que perfiles concretos como “Especialista en física cuántica para audiencias no técnicas”. Un agente enfocado produce outputs más precisos, consistentes y alineados con la tarea, además de tomar mejores decisiones en su dominio de expertise. El equilibrio está en evitar definiciones excesivamente estrechas que los vuelvan inservibles ante variaciones naturales de la tarea, y en diseñar especializaciones que se complementen entre los agentes de un mismo equipo.

3. Hay que prepararlos para que colaboren. Cuando se diseña agentes que trabajarán en equipo, hay que concentrarse en tres cosas: uno, crear habilidades claramente complementarias, que encajen como piezas de un rompecabezas; dos, definir puntos de entrega (handoffs) explícitos donde un agente sepa exactamente qué formato y contenido debe pasar al siguiente; y, tres, introducir cierta tensión constructiva: agentes con perspectivas ligeramente distintas que dialoguen y se desafíen entre sí, como un investigador meticuloso y un escritor didáctico, ya que suelen generar resultados más ricos que uno solo perfectamente alineado.

Ahora que ya sabemos la base teórica, podemos definir nuestros agentes.

2.5. Los agentes de Itinera

Una vez que hemos definido el equipo, sea este o el que cada cual se haya planteado, hay que trasladarlo al código de CrewAI. Los elementos que definirán a nuestros agentes estarán repartidos entre la configuración yaml y el código Python. Vamos a empezar con el yaml, con src/itinera/config/agents.yaml, que en este ejemplo podría ser algo parecido a esto.

Hay que fijarse, por cierto, que hay parámetros que no van a fuego, sino indicados entre llaves para que los defina el usuario.

Otro detalle importante es que Herodoto lleva allow_delegation en true.

herodoto:
  role: >
    Coordinador de planificación de viajes
  goal: >
    Coordinar al equipo para producir un itinerario de {days} días por {zone}
    que equilibre cultura y naturaleza según los intereses del usuario
  backstory: >
    Eres un coordinador experto en viajes culturales y de naturaleza.
    Tu trabajo es orquestar al equipo, no investigar por tu cuenta.
    Delegas las búsquedas a tus especialistas y ponderas sus resultados
    antes de pasarlos al siguiente paso.
  verbose: true
  allow_delegation: true

cultural_researcher:
  role: >
    Investigadora de cultura, historia y gastronomía
  goal: >
    Identificar los destinos culturales más relevantes de {zone},
    excluyendo los lugares ya visitados: {visited}
  backstory: >
    Llevas años especializándote en patrimonio histórico, arquitectura
    y gastronomía local de {zone}. Sabes qué merece la pena
    y qué es trampa para turistas.
  verbose: true

nature_researcher:
  role: >
    Investigador de naturaleza, senderismo y paisaje
  goal: >
    Identificar los espacios naturales más relevantes de {zone}
    adaptados al ritmo {pace} del viajero
  backstory: >
    Conoces cada parque, lago y ruta de montaña de {zone}.
    Sabes qué es accesible para un turista sin equipamiento especial
    y qué requiere preparación.
  verbose: true

planner:
  role: >
    Planificadora de rutas día a día
  goal: >
    Diseñar un itinerario de {days} días geográficamente coherente
    combinando los destinos culturales y naturales identificados
  backstory: >
    Tu obsesión es la eficiencia geográfica: nunca propones rutas
    que obliguen a volver sobre los propios pasos. Agrupas destinos
    cercanos y distribuyes el ritmo de forma equilibrada.
  verbose: true

writer:
  role: >
    Redactor de itinerarios de viaje
  goal: >
    Presentar la ruta planificada de forma atractiva y clara,
    con contexto suficiente para que el viajero sepa qué esperar
    en cada parada
  backstory: >
    Escribes sobre viajes desde hace años. Sabes que un buen itinerario
    no es una lista de sitios: es una historia con ritmo, contexto
    y momentos que el viajero recordará.
  verbose: true

Vamos ahora con las tareas.

3. Las tareas

3.1. Anatomía de una tarea

El agente puede ser bueno o regular; lo que de verdad determina la calidad del output es lo bien que esté escrita la tarea, la task. En CrewAI, una tarea es una asignación específica completada por un Agent. Las tareas proporcionan todos los detalles necesarios para la ejecución, como una descripción, el agente responsable, las herramientas requeridas, etcétera. Y es muy importante que estén enfocadas a un solo propósito y a producir una única salida clara.

Una tarea debe tener definidos tres parámetros fundamentales:

description. Aquí se define el proceso, qué hacer, cómo hacerlo, con qué inputs, en qué orden.
expected_output. El entregable, cómo debe verse el resultado final, el formato, la estructura, las secciones, la longitud, etcétera.
agent. el agente responsable de ejecutarla. Es opcional si se usa proceso jerárquico, donde es el coordinador quien decide.

Además, se pueden indicar otros muchos parámetros opcionales. Agrupados por categorías:

1. Asignación y ejecución

name: identificador opcional de la tarea
async_execution: ejecución asíncrona (default false). Útil para tareas largas que no bloquean las siguientes
context: lista de tareas cuyas salidas se usarán como input. Es la pieza clave para encadenar trabajo. Vuelvo sobre esto más adelante.

2. Herramientas

tools: lista de herramientas específicas para esta tarea. Las que se definen aquí, se imponen sobre las del agente, lo que es útil para limitar al agente solo a ciertas tools en una tarea concreta.

3. Formato de salida

markdown: si es true, instruye al agente a formatear el output en Markdown (default false).
output_file: ruta del fichero donde guardar el output.
create_directory: si queremos crear o no los directorios intermedios automáticamente (default true).
output_json: modelo Pydantic para estructurar como JSON.
output_pydantic: modelo Pydantic para el output (más usado que output_json). Solo se puede establecer un tipo de output estructurado por tarea. Debe ser output_json u output_pydantic, pero no los dos.

4. Validación y control de calidad

guardrail: función o string que valida el output antes de pasarlo a la siguiente tarea.
guardrails: lista de guardrails que se ejecutan secuencialmente.
guardrail_max_retries: reintentos máximos cuando el guardrail falla (default 3).

5. Interacción y callbacks

human_input: si es true, un humano revisa el output final (default false). Para human-in-the-loop.
callback: función que se ejecuta tras completar la tarea (útil para logs, notificaciones, etc.).
config: diccionario de configuración específica de la tarea.

Este listado ha cambiado y seguirá cambiando con el tiempo. No hace falta, insisto, aprendérselo de memoria, al igual que la paramétrica de los agentes, pero sí entender qué se puede definir, en qué consiste la tarea, cómo se encadenan unas con otras, etcétera.

También en este caso se pueden definir unos parámetros en el yaml y otros en Python. La lógica es la misma: lo serializable al yaml, los objetos a Python. Por ejemplo:

# tasks.yaml
research_task:
  description: >
    Investiga las últimas tendencias en {topic}
  expected_output: >
    Una lista de 10 bullets con la información más relevante
  agent: researcher
  markdown: true
  output_file: research.md

Y en las funciones:

# crew.py
@task
def research_task(self) -> Task:
  return Task(
    config=self.tasks_config['research_task'],
    # --- Lo que no puede ir en yaml ---
    tools=[SerperDevTool()],
    output_pydantic=ResearchReport,
    guardrail=validate_min_sources,
    callback=notify_slack,
  )

3.3. Entradas y salidas

Una de las grandes dificultades en la construcción de sistemas multiagénticos es ver cómo encandenar las entradas y las salidas para que todo fluya.

Hay dos aspectos clave que debemos tener en cuenta:

toda tarea devuelve un objeto TaskOutput con varios formatos disponibles:

raw: string con la salida en bruto (default).
pydantic: modelo Pydantic parseado (si lo configuramos en el output_pydantic).
json_dict: diccionario JSON (si configuraste output_json).
agent: qué agente la ejecutó.
summary: resumen auto-generado de las primeras 10 palabras de la description.

Las tareas se pueden encadenar de dos formas:

Implícita (proceso secuencial): el output de cada tarea pasa automáticamente a la siguiente.
Explícita con context: se define qué tareas concretas alimentan a otra, lo que es útil cuando la dependencia no es la inmediatamente anterior y una tarea necesita el output de varias tareas previas.

Por ejemplo:

write_blog = Task(
  description="Escribe un post sobre dinosaurios",
  expected_output="Post de 4 párrafos",
  agent=writer,
  context=[research_dinosaurs, research_ops]  # espera a las dos
)

A partir de aquí, podemos complicarnos cuanto queramos con salidas estructuradas y guardrails. Veamos qué es eso.

3.4. Salidas estructuradas

Por defecto, una tarea devuelve texto plano (raw), tal cual lo que el LLM haya escrito, y eso funciona si el output es para que lo lea un humano, pero puede ser un problema cuando el output tiene que ser consumido por código, otra tarea, una API, una base de datos…

Para garantizar que la salida tendrá una forma predecible -campos concretos, tipos concretos, validación automática, whatever- podemos apoyarnos en el campo output_json, en el que, entre otros formatos, podemos definir un modelo Pydantic.

from pydantic import BaseModel
from typing import List
from crewai import Task

# 1. Definimos el molde del output
class BlogPost(BaseModel):
  title: str
  content: str
  tags: List[str]
  word_count: int

# 2. Se lo pasamos a la tarea
blog_task = Task(
  description="Escribe un post sobre dinosaurios de unas 200 palabras",
  expected_output="Un post con título, contenido, tags y conteo de palabras",
  agent=blog_agent,
  output_pydantic=BlogPost  # <--- aquí la clave
)

Luego, cuando ejecutamos el crew, podemos acceder al output de varias formas:

result = crew.kickoff()

# Acceso tipo diccionario
title = result["title"]
content = result["content"]

# Acceso al objeto Pydantic
title = result.pydantic.title
tags = result.pydantic.tags

# Convertir a dict
output_dict = result.to_dict()

El structured output te garantiza, en suma, que la salida tendrá una forma predecible: campos concretos, tipos concretos, validación automática, etcétera. Un patrón que funciona muy bien en un crew secuencial es que las tareas intermedias vayan con output_pydantic con esquemas estrictos para garantizar que los datos fluyan limpios entre agentes; pero la tarea final sea un output libre (markdown, prosa), para que el usuario vea texto natural.

3.5. Guardrails

Por último, hay que hablar de guardrail, que son un mecanismo de control o protección que se utiliza para limitar, validar o supervisar el comportamiento de un sistema.

En CrewAI, los guardrails validan el output antes de pasarlo a la siguiente tarea, y si falla, el agente reintenta hasta guardrail_max_retries veces. Pueden ser de dos tipos:

1. **Function-based (Python)**: control total y determinismo.

def validate_word_count(result: TaskOutput) -> Tuple[bool, Any]:
  words = len(result.raw.split())
  if words > 200:
      return (False, f"Demasiado largo: {words} palabras")
  return (True, result.raw)

LLM-based (string): validación con lenguaje natural, ideal para criterios subjetivos.

guardrail="El post debe ser menor de 200 palabras y sin jerga técnica"

Y lo bueno es que con guardrails=[…] podemos encadenar validaciones, ya que se ejecutan en orden, cada una recibe el output de la anterior. Podemos incluso mezclar funciones Python con strings LLM:

guardrails=[
  validate_word_count,                          # función: precisión
  "El contenido debe ser apto para todo público", # LLM: subjetivo
  format_output,                                # función: transforma
]

Importante: los guardrails LLM no comparten el contexto de interpolación de la tarea. Tienen que ser autocontenidos y validar el output por sí mismo, sin depender de los inputs originales del usuario. Si necesitas validar contra inputs concretos (como “no debe coincidir con la lista de visitados”), eso es trabajo para un guardrail Python, donde sí puedes acceder al contexto completo.

Bueno, pues con esto más o menos, podemos ya preparar las tareas de nuestra Crew, que no olvidemos, es lo que determina la calidad del output.

4. Tareas de Itinera

4.1. La idea general

Como acabamos de ver, el agente puede ser bueno o regular; lo que de verdad determina la calidad del output es lo bien que esté escrita la tarea, que debe ser de un solo propósito y producir una única salida clara.

En Itinera podríamos definir cuatro tareas en cadena, coordinadas por Herodoto, que es el manager:

research_culture, para la investigadora cultural, busca destinos de la zona, excluyendo visitados.
research_nature, para el investigador de naturaleza, busca espacios naturales adaptados al ritmo.
plan_route, para la planificadora, combina ambos resultados en un itinerario día a día.
write_itinerary, para el redactor, convierte el itinerario en un documento Markdown narrativo.

Respecto a las entradas y las salidas, la idea es estructurar y validar los datos que fluyen entre agentes y dejar libre donde los datos llegan al humano. Además, no irán una tras otra, como en el ejemplo anterior, que definimos el orden como process.sequential, sino que será Herodoto quien decida el orden y que hay que delegar en cada caso.

Sin embargo, como la planificadora necesita que las tareas anteriores estén terminadas, indicaremos un context explícito en plan_route, que además tendrá una salida determinística definida en un output estructurado.

Para validar las salidas usaremos dos guardrails diferenciados según el tipo de validación:

LLM (string) en los investigadores, que validarán criterios subjetivos, como la relevancia, no tópicos.
Python (función) en plan_route, con criterios objetivos: número de días, sin repetidos, sin vacío

En código queda así

4.2. tasks.yaml

Lo primero es el yaml. Obsérvese que algunas llevan definidos ya un guardrail, en cuyo caso hay que definir también un agent.

research_culture:
  description: >
    Investiga los destinos culturales más relevantes de {zone} para un viaje
    de {days} días. Excluye obligatoriamente los siguientes lugares ya visitados:
    {visited}. Considera los intereses del viajero: {interests}.
    Para cada destino, busca información histórica y cultural en Wikipedia
    y obtén sus coordenadas con Nominatim.
    Devuelve entre 8 y 12 destinos ordenados por relevancia.
  expected_output: >
    Lista de destinos culturales con nombre, descripción breve (2-3 frases),
    coordenadas (latitud y longitud) y categoría (historia, arte, gastronomía...).
    Sin destinos repetidos con {visited}.
   agent: cultural_researcher
  guardrail: >
    Verifica que los destinos propuestos sean culturalmente relevantes
    y específicos de la zona indicada en la descripción de la tarea,
    no tópicos turísticos genéricos. Rechaza si hay menos de 8 destinos.

research_nature:
  description: >
    Investiga los espacios naturales más relevantes de {zone} para un viaje
    de {days} días con ritmo {pace}. Ten en cuenta los intereses del
    viajero: {interests}.
    Para cada espacio, obtén sus coordenadas con Nominatim y consulta
    la previsión meteorológica con Open-Meteo para indicar si la época
    es adecuada para visitarlo.
    Devuelve entre 6 y 10 espacios ordenados por accesibilidad.
  expected_output: >
    Lista de espacios naturales con nombre, descripción breve (2-3 frases),
    coordenadas, nivel de dificultad (bajo / medio / alto) y nota
    meteorológica si la época es relevante.
  agent: nature_researcher
  guardrail: >
    Verifica que los espacios naturales sean accesibles para turistas
    sin equipamiento técnico especializado. Rechaza si hay menos de
    6 espacios o si alguno requiere alpinismo, escalada o material
    de montaña avanzado.

plan_route:
  description: >
    Con los destinos culturales y naturales identificados, diseña un itinerario
    de {days} días por {zone} para un viajero con ritmo {pace}.
    Agrupa los destinos geográficamente para evitar backtracking.
    Distribuye la carga diaria según el ritmo: tranquilo (1-2 paradas/día),
    moderado (2-3 paradas/día), intenso (3-4 paradas/día).
    Equilibra cultura y naturaleza según los intereses: {interests}.
  expected_output: >
    Itinerario estructurado día a día. Para cada día: número de día,
    zona geográfica, lista de paradas en orden lógico y tipo de cada
    parada (cultural / natural). Sin días vacíos ni paradas repetidas.

write_itinerary:
  description: >
    Toma el itinerario estructurado y conviértelo en el documento final
    que el viajero se llevará. Sé conciso: contexto justo y suficiente,
    sin florituras. Para cada parada añade una descripción breve,
    un horario orientativo y un consejo práctico cuando aporte valor.
    El tono debe ser claro y evocador, como una buena guía de viaje.
  expected_output: >
    Documento en Markdown con título y una sección por día.
    Cada sección incluye: encabezado con día y zona, y 2-4 paradas.
    Cada parada: nombre en negrita, horario orientativo y descripción
    de 1-2 frases. Sin introducción ni cierre largos. Máximo 1500 palabras
  markdown: true
  output_file: itinerary.md

4.3. Salidas estructuradas

Los modelos de las salidas estructuradas pueden ir por ejemplo en un paquete models (en el que conviene añadir un init aunque ya no sea estrictamente necesario). Un modelo por archivo.

# src/itinera/models/route.py

from pydantic import BaseModel
from typing import List, Literal

class Stop(BaseModel):
  name: str
  type: Literal["cultural", "natural"]
  description: str

class DayPlan(BaseModel):
  day: int
  zone: str
  stops: List[Stop]

class RouteOutput(BaseModel):
  zone: str
  total_days: int
  days: List[DayPlan]

4.5. Guardrails

El guardrail que va por código lo podemos guardar en ./guardrails. Los otros, recordemos que iban en el yaml.

# src/itinera/guardrails/route.py

from typing import Tuple, Any
from crewai.tasks.task_output import TaskOutput
from itinera.models.route import RouteOutput

def validate_route(result: TaskOutput) -> Tuple[bool, Any]:
    route: RouteOutput = result.pydantic

    if len(route.days) != route.total_days:
        return (False, f"Se esperaban {route.total_days} días, hay {len(route.days)}")

    for day in route.days:
        if not day.stops:
            return (False, f"El día {day.day} no tiene paradas")

    all_stops = [stop.name for day in route.days for stop in day.stops]
    if len(all_stops) != len(set(all_stops)):
        return (False, "Hay paradas repetidas en la ruta")

    return (True, route)

5. El motor de la aplicación

Una vez definidos los agentes y las tareas, es el turno del motor de la aplicación, la pieza central del proyecto, donde uniremos todo lo que hemos definido por separado.

Es la clase decorada con CrewBase que hay definida en src/itinera/crew.py, en la que se incorporan:

La configuración de los yaml.
La instancia los agentes como objetos Agent, añadiéndoles las herramientas que necesiten.
La instancia las tareas como objetos Task, añadiéndoles output_pydantic, context, guardrail…
El ensamblaje de una Crew, definiendo el proceso (jerárquico), el manager y el modo de ejecución.

En código queda muy largo, pero creo que es mejor verlo corrido que fragmentado en diversos snippets.

#src/itinera/crew.py
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai.agents.agent_builder.base_agent import BaseAgent

from itinera.models.route import RouteOutput
from itinera.guardrails.route import validate_route

@CrewBase
class Itinera():
    """Itinera crew"""

    agents: list[BaseAgent]
    tasks: list[Task]

    # ------------------------------------------------------------------
    # AGENTES
    # ------------------------------------------------------------------
    # Cada método decorado con @agent crea un agente a partir de su
    # configuración yaml. En proceso jerárquico, el manager (Herodoto)
    # NO se incluye en la lista self.agents: se pasa aparte como
    # manager_agent al construir la Crew.
    # ------------------------------------------------------------------

    @agent
    def herodoto(self) -> Agent:
        return Agent(
            config=self.agents_config['herodoto'],
        )

    @agent
    def cultural_researcher(self) -> Agent:
        return Agent(
            config=self.agents_config['cultural_researcher'],
        )

    @agent
    def nature_researcher(self) -> Agent:
        return Agent(
            config=self.agents_config['nature_researcher'],
        )

    @agent
    def planner(self) -> Agent:
        return Agent(
            config=self.agents_config['planner'],
        )

    @agent
    def writer(self) -> Agent:
        return Agent(
            config=self.agents_config['writer'],
            context=[self.plan_route()],   # le decimos de dónde viene su input
        )

    # ------------------------------------------------------------------
    # TAREAS
    # ------------------------------------------------------------------
    # En proceso jerárquico las tareas NO llevan agent asignado en el
    # yaml: es Herodoto quien decide en tiempo de ejecución qué agente
    # ejecuta cada una.
    # ------------------------------------------------------------------

    @task
    def research_culture(self) -> Task:
        # El guardrail LLM (string) está definido en el yaml.
        return Task(
            config=self.tasks_config['research_culture'],
        )

    @task
    def research_nature(self) -> Task:
        # Guardrail LLM definido en el yaml.
        return Task(
            config=self.tasks_config['research_nature'],
        )

    @task
    def plan_route(self) -> Task:
      # Tres cosas importantes aquí:
      # 1. context explícito: depende de DOS tareas anteriores,
      #    no solo de la inmediatamente previa.
      # 2. output_pydantic: la salida se valida contra RouteOutput,
      #    para que write_itinerary reciba datos limpios.
      # 3. guardrail Python: valida criterios objetivos
      #    (nº de días, paradas no repetidas, días no vacíos).
    
        return Task(
            config=self.tasks_config['plan_route'],
            context=[self.research_culture(), self.research_nature()],
            output_pydantic=RouteOutput,
            guardrail=validate_route,
        )

    @task
    def write_itinerary(self) -> Task:
      # Tarea final. Salida libre en Markdown para el usuario.
      # No lleva output_pydantic ni guardrail: lo que sale aquí
      # es lo que verá el viajero.
        return Task(
            config=self.tasks_config['write_itinerary'],
        )

    # ------------------------------------------------------------------
    # CREW
    # ------------------------------------------------------------------

    @crew
    def crew(self) -> Crew:
        return Crew(
            # No usamos self.agents: en proceso jerárquico el manager
            # desaparecería de la lista y los especialistas no serían
            # encontrados. Los listamos explícitamente.
            agents=[
                self.cultural_researcher(),
                self.nature_researcher(),
                self.planner(),
                self.writer(),
            ],
            tasks=self.tasks,               # generado por @task
            process=Process.hierarchical,
            manager_agent=self.herodoto(),   # <-- aquí es donde se le da el rol de manager
            verbose=True,
            output_log_file="execution_log.md", # para reconstruir la ejecución
        )

6. Punto de entrada

Con todo ya configurado y la Crew montada, solo falta definir el punto de entrada, el archivo main.py. Dos ideas clave al repecto:

Este fichero es el punto de entrada local de la crew. La idea es mantenerlo limpio: solo definir los inputs y lanzar la crew. Toda la lógica de agentes, tareas y orquestación vive en crew.py.
Las claves del diccionario inputs deben coincidir con las variables {...} declaradas en los yaml de agentes y tareas. Si en el yaml hay {zone}, aquí debe existir “zone”. Un typo silencioso aquí es la causa más común de errores al arrancar.

Aunque pongo el código completo, ahora solo hay que fijarse en el método run().

import sys
import warnings

from datetime import datetime

from itinera.crew import Itinera

# Silenciamos un warning conocido de pysbd (una dependencia interna
# de CrewAI) que no afecta a la ejecución pero ensucia la salida.
warnings.filterwarnings("ignore", category=SyntaxWarning, module="pysbd")

def run():
    """
    Lanza la crew con los parámetros del viaje.

    Estos inputs alimentan las variables {zone}, {days}, {visited},
    {interests} y {pace} que aparecen en agents.yaml y tasks.yaml.

    En la vida real se recibirían de algún input del usuario.
    """
    inputs = {
        "zone": "norte de Italia",
        "days": 9,
        "visited": ["Florencia", "Siena", "San Gimignano", "Pisa"],
        "interests": ["arte renacentista", "gastronomía", "lagos alpinos"],
        "pace": "moderado",
        "current_year": str(datetime.now().year),
    }

    try:
        # kickoff() arranca la crew y bloquea el hilo hasta que termina.
        # Más adelante veremos kickoff_async() para ejecuciones no
        # bloqueantes.
        Itinera().crew().kickoff(inputs=inputs)
    except Exception as e:
        raise Exception(f"An error occurred while running the crew: {e}")


def train():
    """
    Entrena la crew durante N iteraciones y guarda el resultado en un
    fichero. Útil para refinar el comportamiento de los agentes con
    feedback humano.

    Uso: crewai train <n_iteraciones> <fichero_salida>
    """
    inputs = {
        "zone": "norte de Italia",
        "days": 9,
        "visited": ["Florencia", "Siena"],
        "interests": ["arte renacentista", "gastronomía"],
        "pace": "moderado",
        "current_year": str(datetime.now().year),
    }
    try:
        Itinera().crew().train(
            n_iterations=int(sys.argv[1]),
            filename=sys.argv[2],
            inputs=inputs,
        )
    except Exception as e:
        raise Exception(f"An error occurred while training the crew: {e}")


def replay():
    """
    Reproduce la ejecución a partir de una task_id concreta, sin tener
    que volver a ejecutar todo desde cero. Muy útil para depurar la
    última tarea sin gastar tokens en las anteriores.

    Uso: crewai replay <task_id>
    """
    try:
        Itinera().crew().replay(task_id=sys.argv[1])
    except Exception as e:
        raise Exception(f"An error occurred while replaying the crew: {e}")


def test():
    """
    Ejecuta la crew en modo test usando un LLM como evaluador.
    Devuelve métricas sobre la calidad de la ejecución.

    Uso: crewai test <n_iteraciones> <modelo_evaluador>
    """
    inputs = {
        "zone": "norte de Italia",
        "days": 9,
        "visited": ["Florencia"],
        "interests": ["arte renacentista"],
        "pace": "moderado",
        "current_year": str(datetime.now().year),
    }

    try:
        Itinera().crew().test(
            n_iterations=int(sys.argv[1]),
            eval_llm=sys.argv[2],
            inputs=inputs,
        )
    except Exception as e:
        raise Exception(f"An error occurred while testing the crew: {e}")


def run_with_trigger():
    """
    Lanza la crew a partir de un payload JSON externo, en lugar de
    inputs hardcodeados. Pensado para integraciones donde otro sistema
    (webhook, cola de mensajes, etc.) dispara la ejecución.

    Uso: python main.py '{"zone": "...", "days": 9, ...}'
    """
    import json

    if len(sys.argv) < 2:
        raise Exception("No trigger payload provided. Please provide JSON payload as argument.")

    try:
        trigger_payload = json.loads(sys.argv[1])
    except json.JSONDecodeError:
        raise Exception("Invalid JSON payload provided as argument")

    # crewai_trigger_payload es la clave estándar que CrewAI usa para
    # exponer el payload completo a los agentes. Las demás claves se
    # dejan vacías porque, en este modo, los valores vienen del payload.
    inputs = {
        "crewai_trigger_payload": trigger_payload,
        "zone": "",
        "days": "",
        "visited": "",
        "interests": "",
        "pace": "",
    }

    try:
        result = Itinera().crew().kickoff(inputs=inputs)
        return result
    except Exception as e:
        raise Exception(f"An error occurred while running the crew with trigger: {e}")

Y ya solo faltaría correr el invento con

crewai run

Bueno, otro día seguimos con la segunda parte.