Orquestar canalizaciones de IA con Airflow

Empezar con prototipos de IA es sencillo. Un cuaderno, una clave de API y unas pocas líneas de Python pueden bastar para clasificar texto, resumir documentos o generar recomendaciones. 

Pero mantener ese prototipo funcionando a diario de forma fiable es otro reto muy distinto.

¿Qué pasa cuando la API del modelo alcanza el límite de peticiones? ¿Y si una solicitud agota el tiempo, la salida no sigue el formato esperado o una base de datos aguas abajo no está disponible temporalmente?

Cuando una canalización de IA pasa a ser un flujo de trabajo recurrente, la fiabilidad, la monitorización y la recuperación se vuelven tan importantes como el propio modelo.

En este artículo, te mostraré cómo crear una canalización completa de IA con Apache Airflow, una plataforma de código abierto para desarrollar, programar y monitorizar flujos de trabajo. 

Nuestra canalización se ejecutará a diario: obtendrá reseñas de clientes, clasificará su sentimiento con un modelo de lenguaje grande (LLM) y almacenará los resultados. ¡Todo de forma fiable en un único flujo!

¿Por qué Airflow para canalizaciones de IA?

Un cuaderno de Jupyter es perfecto para experimentar. A todos nos encantan los notebooks para probar ideas, escribir código rápido y visualizar resultados sobre la marcha.

Un cron suele bastar para ejecutar un script sencillo cada noche. Aunque, siendo sinceros, muchos scripts se ejecutan a mano durante semanas antes de que alguien les asigne un horario en condiciones.

Los problemas empiezan cuando un script se convierte en varios y esos scripts empiezan a depender entre sí. En ese punto ya no tenemos un programa aislado: tenemos un flujo de trabajo.

Ni los notebooks ni los cron jobs están pensados para gestionar flujos de trabajo complejos con fiabilidad. 

Las dependencias suelen quedar ocultas en el código, los fallos requieren investigación manual y reejecutar la canalización puede implicar repetir pasos que ya habían terminado con éxito.

Este tipo de complejidad ya aparece en canalizaciones ETL (Extract-Transform-Load) tradicionales. Un flujo ETL puede necesitar recuperar datos, transformarlos, validarlos y cargarlos en otro sistema. 

Aunque las entradas y transformaciones sean previsibles, cada paso adicional crea un nuevo punto potencial de fallo. Hoy en día, las canalizaciones de IA llevan este reto un paso más allá. 

Si te interesan específicamente las canalizaciones ETL, te recomiendo este curso sobre canalizaciones ETL en Airflow.

Canalizaciones de IA

Piensa en un flujo de trabajo típico de IA: recopilar datos, limpiarlos, llamar a una API de modelo, validar la respuesta, almacenar el resultado y avisar a alguien cuando algo falla. 

Cada paso depende del anterior y cada uno puede fallar de forma distinta: una API puede devolver un error de límite de peticiones. Una solicitud al modelo puede expirar. Una respuesta generada puede ser JSON válido y aun así no pasar un control de calidad. 

Las salidas también pueden estar incompletas, ser inconsistentes o tener un formato distinto al que espera el siguiente paso.

Airflow no elimina estos fallos, pero hace que gestionarlos sea mucho más sencillo. 

Con Airflow, una tarea que llama a un LLM puede reintentarse tras una respuesta temporal de rate limit sin volver a ejecutar la ingesta de datos. Una tarea de validación puede rechazar un resultado mal formado o de baja calidad antes de llegar a producción. Las alertas pueden notificar al equipo solo después de agotar los reintentos configurados.

Airflow hace esto posible representando el flujo de trabajo como un grafo acíclico dirigido (DAG). 

En lugar de esconder toda la canalización dentro de un script largo, un DAG define cada operación como una tarea independiente con dependencias explícitas, políticas de reintentos, logs y estado de ejecución. 

Esta es la principal razón por la que Airflow encaja tan bien con canalizaciones de IA: trata los fallos como una parte esperable del flujo de trabajo y no como una sorpresa excepcional. 

A medida que un experimento crece y se convierte en un proceso de producción recurrente, Airflow aporta la capa de orquestación necesaria para que sea observable, recuperable y fácil de mantener.

Antes de construir nuestra propia canalización, vamos a preparar el entorno.

Configurar el entorno

Tenemos que completar tres pasos breves para dejar listo el entorno. 

Instalar Airflow 3 y el SDK de IA de Airflow

Para este tutorial, ejecutaremos Airflow en local. 

Esta configuración es cómoda para aprender y desarrollar porque un solo comando inicializa la base de metadatos e inicia los servicios necesarios para usar Airflow.

Primero, vamos a crear y activar un entorno virtual para aislar Airflow y sus dependencias del resto de tu instalación de Python:

python3 -m venv ~/airflow-venv
source ~/airflow-venv/bin/activate

A continuación, necesitamos instalar Airflow 3 fijando versión y usando el archivo de constraints oficial:

AIRFLOW_VERSION=3.3.0
PYTHON_VERSION="$(python -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")')"

CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"

pip install "apache-airflow==${AIRFLOW_VERSION}" \
 --constraint "${CONSTRAINT_URL}"

Ten en cuenta que Airflow depende de muchos paquetes de Python, y esos paquetes tienen sus propias dependencias. Sin el archivo de constraints, pip puede elegir versiones más nuevas o incompatibles. 

La instalación podría parecer correcta, pero después Airflow puede fallar al importar providers, al iniciar servicios o al ejecutar tareas. La guía oficial de instalación de Airflow recomienda usar archivos de constraints para instalar un conjunto de dependencias compatibles.

Como nuestro flujo llamará a un LLM, también necesitamos el provider Common AI de Airflow y la dependencia de OpenAI para Pydantic AI:

pip install \
 "apache-airflow-providers-common-ai" \
 "pydantic-ai-slim[openai]"

Ahora vamos a configurar la conexión a OpenAI. Airflow usa conexiones para guardar credenciales y parámetros de conexión fuera del código del DAG. Esto mantiene el archivo de Python más limpio y evita hardcodear secretos en la canalización.

Una vez validada la instalación, toca establecer la clave de OpenAI para permitir la comunicación real con el modelo. Hay muchas opciones para hacerlo. 

Mi opción preferida es:

1. Exportar nuestra clave de API de OpenAI en la sesión de terminal actual:

export OPENAI_API_KEY="your-openai-api-key"

2.  Crear una conexión de Airflow llamada pydanticai_default:

airflow connections add pydanticai_default \
 --conn-type pydanticai \
 --conn-password "$OPENAI_API_KEY" \
 --conn-extra '{"model":"openai:gpt-4o-mini"}'

El provider Common AI utiliza el tipo de conexión pydanticai para comunicarse con proveedores de LLM a través de Pydantic AI. 

Este comando guarda la conexión en la base de metadatos de Airflow, así que después no necesitas mantener el archivo .env ni pasar la clave cada vez que inicies Airflow.

3. Por último, podemos arrancar el entorno local de Airflow con:

airflow standalone

Mantén esta terminal abierta mientras Airflow esté en ejecución. El modo standalone es ideal para desarrollo local y tutoriales, pero no está pensado para despliegues en producción.

Para acceder a la interfaz de Airflow, ahora podemos abrir http://localhost:8080 en el navegador. Lo primero que veremos es la página de inicio de sesión. Pero ¿qué usuario y contraseña debemos usar?

Vista de inicio de sesión de Apache Airflow.

Cuando Airflow arranca en modo standalone, genera automáticamente credenciales de acceso. El usuario y la contraseña se imprimen en la salida de la terminal y también se guardan en $AIRFLOW_HOME/simple_auth_manager_passwords.json.generated.

Por defecto, AIRFLOW_HOME es ~/airflow, así que podemos recuperar las credenciales con:

cat ~/airflow/simple_auth_manager_passwords.json.generated

Con esas credenciales ya podremos iniciar sesión en la interfaz de Airflow.

Explorar la interfaz

Tras iniciar sesión, Airflow abre en la página Home. Esta vista te ofrece un resumen rápido del entorno local de Airflow.

Vista de la interfaz de Airflow tras el primer inicio de sesión.

En la parte superior, las tarjetas de Stats resumen el estado actual de tus DAGs, incluidos los flujos fallidos, en ejecución y activos. 

Como aún no hemos creado nuestra canalización, es posible que estos valores sigan en cero. 

La sección Health muestra si los servicios principales de Airflow están en marcha. 

Más abajo, la barra Pool Slots muestra la capacidad de ejecución disponible. No configuraremos pools personalizados en este artículo, pero esta área resulta útil para controlar cuántas tareas pueden ejecutarse a la vez. 

Por último, la sección History resume las ejecuciones recientes de DAG. Cuando activemos nuestra canalización de sentimiento de reseñas, esta área empezará a mostrar ejecuciones en cola, en curso, correctas y fallidas.

Estructura del proyecto

Ahora que Airflow está en marcha, vamos a crear los archivos necesarios para nuestra canalización:

mkdir -p ~/airflow/dags
touch ~/airflow/dags/review_sentiment_pipeline.py
touch ~/airflow/requirements.txt

En una instalación local standalone, Airflow usa ~/airflow como AIRFLOW_HOME por defecto. Colocaremos nuestro archivo de DAG dentro del directorio dags y mantendremos un archivo requirements.txt para documentar las dependencias del proyecto. 

Por tanto, el proyecto tendrá la siguiente estructura:

Estructura mínima de un proyecto Airflow.

El archivo eview_sentiment_pipeline.py contendrá la definición del flujo de trabajo, incluidas las tareas para obtener reseñas, clasificar su sentimiento y almacenar los resultados. 

Airflow monitoriza el directorio dags configurado y busca definiciones de DAG válidas dentro de archivos Python. Cuando nuestro archivo contenga un DAG válido, el flujo aparecerá en la página Dags de la interfaz de Airflow.

Si ves que tienes algún problema con la puesta en marcha, te sugiero revisar el tutorial Getting Started with Apache Airflow. .

Construir tu primera canalización con Airflow

Para empezar a trabajar con Airflow, construiremos un job diario que procese reseñas de clientes y clasifique su sentimiento. 

Imagina que acabamos de lanzar una plataforma de e-commerce. Los clientes ya están dejando reseñas y queremos analizarlas automáticamente cada día. 

En lugar de ejecutar un script a mano, podemos usar Airflow para orquestar el flujo de trabajo.

Paso 1: obtener reseñas

Imaginemos que las reseñas en bruto se almacenan en un archivo .jsonl. 

Cada línea contiene una reseña como objeto JSON, y las nuevas reseñas se van añadiendo al final del archivo durante el día. En un sistema de producción, estos datos podrían venir de una API, una base de datos, un object store o una cola de mensajes. Para este tutorial, el archivo JSONL nos da una fuente sencilla y realista de datos en bruto de reseñas.

En este escenario, el primer paso de la canalización es cargar esas reseñas en Airflow. Empezamos con una función normal de Python, como haríamos en un notebook o script:

REVIEWS_FILE = Path.home() / "airflow" / "data" / "reviews.jsonl"

def fetch_reviews() -> list[dict]:
   """Load raw customer reviews from the JSONL landing file."""
   reviews = []

   with REVIEWS_FILE.open("r", encoding="utf-8") as file:
       for line_number, line in enumerate(file, start=1):
           if not line.strip():
               continue

           try:
               reviews.append(json.loads(line))
           except json.JSONDecodeError as error:
               raise ValueError(
                   f"Invalid JSON on line {line_number}"
               ) from error

   print(f"Fetched {len(reviews)} reviews")
   return reviews

Esta función abre el archivo JSONL, lo lee línea a línea, salta las líneas vacías y convierte cada objeto JSON en un diccionario de Python. 

Si una línea contiene JSON no válido, la tarea falla con un mensaje de error claro que indica el número de línea.

Hasta aquí es Python puro. El siguiente paso es convertirlo en una tarea de Airflow y situarla dentro de un DAG.

Encapsular la tarea en un DAG

Para convertir la función en parte de un flujo de trabajo de Airflow, necesitamos tres piezas:

1. Primero, decoramos la función con @task. Esto le indica a Airflow que la función es una tarea.
2. Segundo, definimos un DAG con @dag. El DAG da al flujo un nombre, horario, fecha de inicio y otra configuración.
3. Tercero, llamamos a la tarea dentro de la función del DAG e instanciamos el DAG al final del archivo.

Aquí tienes la versión mínima:

from airflow.sdk import dag

@dag(    fetch_reviews()
   dag_id="fetch_reviews_pipeline",
)

def fetch_reviews_pipeline():
   fetch_reviews()

fetch_reviews_pipeline()

Ten en cuenta que este DAG aún no se ejecutará automáticamente. 

Tal como está escrito, habría que activarlo manualmente desde la UI de Airflow. Para configurar cuándo debe ejecutarse el flujo, podemos añadir parámetros de programación a la definición del DAG. Una opción posible es:

@dag(
   dag_id="review_sentiment_pipeline",
   schedule="@daily",
   start_date=datetime(2026, 1, 1),
   catchup=False,
   tags=["ai", "sentiment"],
)

¡Vamos a desgranarlo!

• dag_id da al DAG un nombre único en Airflow.

• schedule="@daily" ejecuta el DAG una vez al día.
• start_date indica a Airflow cuándo puede empezar a programarlo.
• catchup=False evita que Airflow cree ejecuciones para fechas pasadas.
• tags ayuda a organizar y filtrar el DAG en la UI de Airflow.

Una vez definido el DAG, podemos ir a la UI y comprobar cómo queda en la práctica:

Vista de la UI de Airflow de nuestra canalización review_sentiment_pipeline. Fíjate en los dos tipos de ejecuciones (programadas y manuales) y en el botón Trigger manual.

En la UI vemos que Airflow descubrió automáticamente el DAG llamado review_sentiment_pipeline. 

También vemos que se ejecutó según su horario, y que activamos una ejecución adicional manualmente con el botón Trigger en la esquina superior derecha de la interfaz.

Una vista interesante es la de calendario, donde Airflow registra las ejecuciones de la canalización junto con el estado de cada una:

Vista de calendario de nuestra canalización en Airflow con ejecuciones programadas y manuales.

¡Si quieres, puedes pausar aquí y explorar la UI de Airflow!

Paso 2: clasificación con LLM

Con las reseñas cargadas, el siguiente paso es clasificar su sentimiento. Para ello, usaremos @task.llm, un decorador que convierte una llamada a un LLM en una tarea nativa de Airflow.

Podríamos llamar a la API de OpenAI manualmente dentro de una @task normal, pero @task.llm nos ofrece una integración más limpia. 

Gestiona detalles específicos del LLM como la conexión al modelo, el prompt del sistema, la salida estructurada y la serialización del resultado. Al mismo tiempo, la tarea se beneficia de las funciones estándar de Airflow como reintentos, logging, monitorización y gestión de dependencias.

Este segundo paso tiene algunas piezas más que la tarea en sí, así que vamos a construirlo poco a poco:

1. Primero, definimos la estructura de salida que esperamos del modelo:

from typing import Literal
from pydantic import BaseModel

class ReviewClassification(BaseModel):
   """Structured result returned by the LLM."""

   review_id: str
   sentiment: Literal["positive", "neutral", "negative"]

Esto le indica a la tarea LLM que cada respuesta debe contener el ID de la reseña y una de tres etiquetas de sentimiento: positive, neutral o negative.

2. Después, creamos la función que construye el prompt para una reseña:

@task.llm(
   llm_conn_id="pydanticai_default",
   system_prompt=(
       "Classify the customer review as positive, neutral, or negative. "
       "Use both the written review and the star rating as context. "
       "Return the review ID and sentiment."
   ),
   output_type=ReviewClassification,
   serialize_output=True,
   retries=3,
)
def classify_review(review: dict) -> str:
   """Build the prompt sent to the LLM."""
   return (
       f"Review ID: {review['review_id']}\n"
       f"Product: {review['product']}\n"
       f"Rating: {review['rating']} out of 5 stars\n"
       f"Review: {review['review_text']}"
   )

El cuerpo de la función devuelve un prompt de texto, mientras que el decorador @task.llm se encarga de enviarlo al modelo usando la conexión pydanticai_default que configuramos antes. 

El system_prompt da la instrucción al modelo, mientras que output_type=ReviewClassification solicita una respuesta estructurada. También establecemos serialize_output=True para que Airflow almacene el resultado como un diccionario simple, lo que facilita pasarlo a tareas posteriores.

¿Te interesan tareas que requieran inferencia con LLM? Échale un vistazo a este curso de LangChain para aplicaciones más especializadas.

Visualizar la clasificación del modelo

La tarea de clasificación devuelve un resultado estructurado, pero durante el desarrollo es útil ver qué produjo el modelo sin esperar al paso final de almacenamiento. 

De hecho, un paso de la canalización puede componerse de más de una tarea de Airflow. 

En este caso, añadiremos una pequeña tarea que imprima cada clasificación en los logs de Airflow:

@task
def show_classification(classification: dict) -> dict:
   """Print one model classification to the Airflow task logs."""
   print(
       f"Review {classification['review_id']}: "
       f"{classification['sentiment']}"
   )

   return classification

Esta tarea no cambia el resultado. Simplemente registra la salida del modelo y devuelve el mismo diccionario, de modo que la clasificación se pueda seguir pasando al siguiente paso de la canalización.

 3. Por último, tenemos que conectar las nuevas tareas dentro de la función del DAG:

@dag(
   dag_id="review_sentiment_pipeline",
   schedule="@daily",
   start_date=datetime(2026, 1, 1),
   catchup=False,
   tags=["ai", "sentiment"],
)
def review_sentiment_pipeline():
   reviews = fetch_reviews()

   classifications = classify_review.expand(
       review=reviews
   )

   displayed_classifications = show_classification.expand(
       classification=classifications
   )

review_sentiment_pipeline()

Nota: En esta canalización usamos mapeo dinámico de tareas con .expand(review=reviews). Esto le dice a Airflow que cree una instancia de tarea classify_review por cada reseña, de forma que cada una se envía al LLM en una solicitud separada.

Esto suele ser preferible a enviar toda la lista de reseñas a una única tarea LLM. Una llamada como classify_reviews(reviews) procesaría todas las reseñas juntas, lo que dificulta depurar fallos, hace los reintentos menos precisos y aumenta la probabilidad de chocar con límites de contexto del modelo o de la API.

Antes de terminar este segundo paso de la canalización, podemos volver a la UI de Airflow y visualizar cómo ha evolucionado el flujo. 

En este punto, el DAG contiene tres tareas: fetch_reviews, classify_review y show_classification:

Vista de la UI de Airflow con las tres tareas de nuestro DAG review_sentiment_pipeline.

Solución de problemas: gestionar fallos temporales del LLM

Como usamos mapeo dinámico de tareas, Airflow envía una solicitud al LLM por reseña. Es útil porque cada reseña se sigue de manera independiente:

La UI de Airflow mostrando reseñas clasificadas de forma independiente. Algunas tareas ya han terminado o fallado, mientras otras siguen ejecutándose.

Las solicitudes a LLM pueden fallar por límites de API, problemas de red o timeouts del modelo. 

En lugar de fallar toda la canalización al instante, podemos pedirle a Airflow que reintente automáticamente la tarea afectada.

¡Añadamos esta característica!

from datetime import timedelta

@task.llm(
   llm_conn_id="pydanticai_default",
   system_prompt=(
       "Classify the customer review as positive, neutral, or negative. "
       "Use both the written review and the star rating as context. "
       "Return the review ID and sentiment."
   ),
   output_type=ReviewClassification,
   serialize_output=True,
   retries=3,
   retry_delay=timedelta(seconds=30),
   retry_exponential_backoff=True,
   max_retry_delay=timedelta(minutes=5),
)

Aquí, retries=3 permite hasta tres intentos adicionales tras el primer fallo. 

El primer reintento espera 30 segundos y el backoff exponencial aumenta la espera entre intentos posteriores, hasta un máximo de cinco minutos.

Lo mejor es que Airflow reintenta solo la tarea mapeada que falló. 

Si 49 reseñas se clasificaron correctamente y una solicitud falló, solo esa reseña fallida se reprograma para reintento. El resto de la canalización no necesita repetir trabajo que ya ha salido bien. 

Tras un tiempo, las clasificaciones fallidas volverán a estar en cola para reintento:

UI de Airflow mostrando todas las tareas fallidas de clasificación a la espera de reintentos.

Una vez que la clasificación de la reseña tiene éxito, cada instancia mapeada de show_classification() muestra la respuesta del modelo para una reseña en sus propios logs. 

Esto significa que el paso de visualización también es independiente: puedes abrir, inspeccionar o depurar la salida de una reseña con independencia del estado de las demás.

Vista de la UI de Airflow con la clasificación de sentimiento del LLM para la reseña 47.

Paso 3: almacenar resultados

Imprimir cada clasificación es útil al desarrollar el DAG porque nos permite inspeccionar rápidamente las respuestas del modelo en los logs de Airflow. 

Sin embargo, los logs no son un lugar fiable para almacenar resultados. Son más difíciles de consultar, reutilizar e integrar con sistemas posteriores.

El último paso, por tanto, es recopilar las clasificaciones y escribirlas en una salida estructurada. 

En este artículo, retiraremos la llamada a show_classification() y guardaremos los resultados en un archivo JSON local. 

En una canalización de producción, esta tarea normalmente escribiría en una base de datos, un data warehouse o un object store. 

Una posible implementación para esta nueva @task sería:

OUTPUT_FILE = (
   Path.home()
   / "airflow"
   / "output"
   / "classified_reviews.json"
)

@task
def store_results(results: list[dict]) -> str:
   """Write all classified reviews to a local JSON file."""
   OUTPUT_FILE.parent.mkdir(parents=True, exist_ok=True)

   classifications = list(results)

   with OUTPUT_FILE.open("w", encoding="utf-8") as file:
       json.dump(
           classifications,
           file,
           indent=2,
           ensure_ascii=False,
       )

   print(
       f"Stored {len(classifications)} classifications "
       f"in {OUTPUT_FILE}"
   )

   return str(OUTPUT_FILE)

Esta tarea es nuestro paso de carga. 

Toma las clasificaciones producidas por el LLM, crea una carpeta de salida si hace falta y escribe todo en un archivo JSON local. 

La pequeña conversión list(results) es útil porque las clasificaciones provienen de tareas mapeadas. 

Airflow recopila esas salidas mapeadas por nosotros, y esta tarea las convierte en una lista normal antes de guardarlas. Ahora solo queda conectar la tarea dentro del DAG llamando a store_results(classifications).

Con esta línea, Airflow sabe que store_results() depende del paso de clasificación. Es decir, el archivo de salida se escribe solo cuando todas las clasificaciones han terminado correctamente.

Ejecutar la canalización

Con los tres pasos conectados, el DAG completo ahora obtiene las reseñas, crea una tarea LLM por reseña y almacena los resultados combinados. 

Dentro de la función del DAG, la canalización queda así:

@dag(
   dag_id="review_sentiment_pipeline",
   schedule="@daily",
   start_date=datetime(2026, 1, 1),
   catchup=False,
   tags=["ai", "sentiment"],
)
def review_sentiment_pipeline():
   reviews = fetch_reviews()
   classifications = classify_review.expand(
       review=reviews
   )
   store_results(classifications)

review_sentiment_pipeline()

Después de guardar el archivo, Airflow tardará unos segundos en analizarlo y descubrir el DAG completo. Una vez que review_sentiment_pipeline aparezca en la lista de DAGs, actívalo con el conmutador y pulsa Trigger para iniciar una ejecución manual. 

Esto nos permite probar toda la canalización al instante, sin esperar a la próxima ejecución diaria programada. Cuando la ejecución termine con éxito, el archivo JSON final estará disponible localmente en ~/airflow/output/classified_reviews.json.

El archivo JSON de salida es el resultado duradero del flujo, que puede reutilizarse fuera del DAG o pasarse a sistemas posteriores más adelante. 

En cambio, la UI de Airflow nos da la vista operativa de la canalización: qué tareas se ejecutaron, cuáles fallaron o tuvieron éxito y dónde inspeccionar los logs. 

La canalización final se muestra a continuación:

Vista de la UI de Airflow con las tres tareas de nuestra canalización. El panel inferior derecho muestra las propiedades de cada tarea, incluido su nombre, regla de activación, operador o decorador y número de reintentos.

Finalmente, en los próximos días, la UI de Airflow empezará a llenarse con las ejecuciones diarias automáticas, dándonos un historial claro de cuándo se ejecutó la canalización y si cada ejecución tuvo éxito:

Vista de la UI de Airflow con las ejecuciones de la canalización y sus estados. Tras el Paso 3, la canalización deja de usar show_classification y ejecuta store_results en su lugar.

Más allá: funciones de IA en Airflow 3 que explorar

Nuestra canalización ahora obtiene reseñas, las clasifica con un LLM y guarda los resultados en una planificación diaria. 

Airflow 3 también admite patrones más avanzados que pueden hacer que los flujos de IA sean más reactivos y fáciles de supervisar.

Planificación impulsada por eventos

En lugar de ejecutar la canalización a una hora fija cada día, Airflow puede iniciarla cuando ocurre un evento externo. 

Por ejemplo, la canalización de reseñas podría ejecutarse siempre que se suba un nuevo archivo de reseñas al almacenamiento de objetos o llegue un mensaje a una cola.

Airflow 3 admite la planificación basada en eventos mediante assets y triggers compatibles con eventos. Esto evita tener que comprobar repetidamente si hay datos nuevos y permite que la canalización reaccione en cuanto se detecta un evento.

Human-in-the-loop con HITLOperator

No todas las decisiones del modelo deben aceptarse automáticamente. 

Algunas reseñas pueden ser ambiguas, sensibles o requerir juicio humano. Los operadores human-in-the-loop de Airflow pueden pausar un flujo en ejecución y solicitar input a través de la interfaz de Airflow. 

Una persona revisor/a puede aprobar o rechazar un resultado, aportar información o elegir qué rama del flujo debe continuar.

El provider Common AI también admite patrones de revisión interactiva para flujos agentic, donde quien revisa puede aprobar, rechazar o solicitar cambios a un resultado generado por un LLM.

Juntas, la planificación por eventos y la revisión humana permiten crear canalizaciones de IA que reaccionan rápido sin renunciar a la supervisión humana donde importa. 

Para saber más sobre las novedades de Airflow 3.0, te recomiendo consultar nuestra guía aparte. 

Reflexiones finales

En este artículo, construimos una canalización completa de IA con Apache Airflow. 

La canalización obtiene reseñas de clientes, clasifica cada reseña con un LLM y guarda los resultados finales en un archivo JSON estructurado.

Por el camino, vimos cómo Airflow convierte funciones de Python en tareas programadas y observables. Esto es especialmente útil en flujos de IA, donde pueden darse límites de API, timeouts de modelo y salidas inesperadas. 

Con dependencias entre tareas, mapeo dinámico, reintentos, logs y la UI de Airflow, podemos hacer que las canalizaciones de IA sean más fáciles de monitorizar, depurar y recuperar.

El ejemplo usó archivos locales para simplificar, pero la misma estructura se puede extender a APIs, bases de datos, flujos impulsados por eventos y revisión human-in-the-loop. 

Si quieres seguir explorando, te recomiendo empezar con nuestro curso Introduction to Apache Airflow in Python y después pasar a Building Data Pipelines with Airflow para practicar de forma más práctica. 

Para una ruta de aprendizaje más amplia, el itinerario Data Engineer in Python cubre el conjunto de habilidades necesario para construir y gestionar canalizaciones de datos en producción.