Guía para la creación de flujos de trabajo automatizados en OpenFang: De un solo agente a un equipo de agentes

Slug: openfang-workflow-automation-guide Categoría: usage-guides (Guías de uso) Palabras clave objetivo: Flujo de trabajo OpenFang, procesos automatizados OpenFang, orquestación Multi-Agent OpenFang Intención de búsqueda: Informativa (el usuario desea construir procesos complejos de automatización con agentes) Longitud objetivo: ~2000 palabras Idioma: es

¿Por qué necesitamos flujos de trabajo?

Las tareas que un solo agente puede completar son limitadas. Cuando necesitas un proceso integral como "investigar un tema → redactar un artículo → verificar hechos → publicar → monitorear retroalimentación", es necesario orquestar múltiples agentes en un flujo de trabajo.

El motor de flujos de trabajo de OpenFang admite tres modos de orquestación: tubería serial (Pipeline), distribución paralela (Fan-out) y ramificación condicional (Conditional). Puedes combinar Hands y agentes como si fueran piezas de LEGO para crear procesos de automatización complejos.

Este artículo te llevará a dominar la creación de flujos de trabajo en OpenFang, desde los más simples hasta los más avanzados, a través de cuatro casos prácticos.

Conceptos clave de los flujos de trabajo

Antes de profundizar en los casos, conozcamos algunos conceptos fundamentales:

Concepto	Descripción
Task	La unidad mínima de ejecución de un flujo de trabajo; una tarea corresponde a un Hand o un agente
Pipeline	Conecta varias tareas en serie, donde la salida de una tarea se convierte automáticamente en la entrada de la siguiente
Fan-out	Distribuye un trabajo a múltiples agentes para procesamiento paralelo y luego consolida los resultados
Trigger	La fuente de eventos que activa el flujo de trabajo: programación temporal, Webhook, cambios en archivos, comandos de mensajes
Context	Estado compartido que se transmite entre los pasos del flujo de trabajo, similar al alcance de las variables en los lenguajes de programación

Caso 1: Boletín informativo diario (Pipeline simple)

Este es el flujo de trabajo Pipeline más sencillo: generar automáticamente un boletín informativo cada mañana:

[[workflows]]name = "daily_industry_briefing"description = "Genera un boletín de noticias de la industria cada mañana a las 8:00"[workflows.trigger]type = "schedule"cron = "0 8 * * 1-5"             # 8:00 AM en días laborables[[workflows.tasks]]id = "monitor"hand = "collector"config = { keywords = ["AI agent", "Rust framework"], sources = ["news", "github"] }timeout = 300[[workflows.tasks]]id = "summarize"hand = "researcher"config = { depth = "shallow", focus = "summary" }depends_on = ["monitor"]timeout = 600[[workflows.tasks]]id = "notify"channel = "slack"config = { channel = "#industry-news", format = "rich_text" }depends_on = ["summarize"][workflows.output]format = "slack_message"save_to = "workspace/briefings/{{date}}.md"

La ejecución del flujo es: Recolección por el Collector → Resumen por el Researcher → Notificación en Slack. depends_on define la relación de dependencia entre las tareas.

Caso 2: Monitoreo y alertas de competencia (Paralelo + Consolidación)

Este flujo de trabajo monitorea simultáneamente la actividad de varios competidores y consolida los resultados para su análisis:

[[workflows]]name = "competitor_monitor"description = "Monitoreo paralelo de 5 sitios web de la competencia, con alertas tras consolidar cambios"[workflows.trigger]type = "schedule"cron = "0 */6 * * *"# Etapa 1: Recolección paralela (Fan-out)[[workflows.tasks]]id = "check_competitor_a"hand = "collector"config = { target = "competitor-a.com", mode = "diff" }[[workflows.tasks]]id = "check_competitor_b"hand = "collector"config = { target = "competitor-b.com", mode = "diff" }[[workflows.tasks]]id = "check_competitor_c"hand = "collector"config = { target = "competitor-c.com", mode = "diff" }# Etapa 2: Consolidación y análisis (espera a que todas las recolecciones terminen)[[workflows.tasks]]id = "aggregate"hand = "researcher"config = { action = "merge_and_prioritize" }depends_on = ["check_competitor_a", "check_competitor_b", "check_competitor_c"]# Etapa 3: Evaluación condicional — solo notificar si hay cambios importantes[[workflows.tasks]]id = "check_significance"hand = "predictor"config = { action = "assess_impact" }depends_on = ["aggregate"]# Etapa 4: Notificación condicional[[workflows.tasks]]id = "alert_high"channel = "telegram"config = { chat_id = "-100xxx", priority = "high" }depends_on = ["check_significance"]condition = "ctx.significance_score > 0.7"[[workflows.tasks]]id = "alert_low"channel = "email"config = { to = "[email protected]", priority = "low" }depends_on = ["check_significance"]condition = "ctx.significance_score <= 0.7"

Los patrones clave aquí son:

Fan-out: Las 5 tareas de recolección de la Etapa 1 se ejecutan en paralelo, sin necesidad de depends_on.
Barrier: La tarea Aggregate de la Etapa 2 depende de que las 5 recolecciones finalicen.
Conditional: La Etapa 4 selecciona diferentes canales de notificación según la puntuación de importancia.

Caso 3: Línea de producción de contenido (Colaboración multi-agente)

Este es el escenario clásico de colaboración multi-agente: redacción, revisión y creación de imágenes mediante IA:

[[workflows]]name = "content_pipeline"description = "Línea de producción de contenido totalmente automatizada, desde el tema hasta la publicación"[workflows.trigger]type = "webhook"endpoint = "/webhook/content/new"secret = "${WEBHOOK_SECRET}"# Etapa 1: Investigación[[workflows.tasks]]id = "research"hand = "researcher"config = {  depth = "deep",  source_types = ["academic", "news", "social"],  target_length = "3000_words"}timeout = 900# Etapa 2: Redacción (usando diferentes modelos)[[workflows.tasks]]id = "draft"hand = "researcher"              # Reutiliza el Hand Researcher, pero con un prompt diferenteconfig = {  model = "claude-opus-4-8",  instruction = "draft_article",  style = "professional_yet_approachable"}depends_on = ["research"]# Etapa 3: Verificación de hechos[[workflows.tasks]]id = "fact_check"hand = "researcher"config = {  model = "claude-haiku-4-5",   # Usa un modelo rápido para la verificación  action = "verify_claims",  strictness = "high"}depends_on = ["draft"]# Etapa 4: Generación de imágenes[[workflows.tasks]]id = "generate_cover"hand = "clip"config = {  action = "generate_cover_image",  style = "modern_tech",  resolution = "1280x720"}depends_on = ["draft"]           # No depende de la verificación de hechos, puede ser paralelo# Etapa 5: Optimización SEO[[workflows.tasks]]id = "seo_optimize"hand = "researcher"config = { action = "seo_optimize", target_keyword = "{{.keyword}}" }depends_on = ["fact_check"]# Etapa 6: Publicación[[workflows.tasks]]id = "publish"channel = "webhook"config = { url = "{{.cms_endpoint}}", method = "POST" }depends_on = ["seo_optimize", "generate_cover"]# Etapa 7: Promoción[[workflows.tasks]]id = "promote"channel = "twitter"config = { action = "post_thread", hashtags = ["#AI", "#Automation"] }depends_on = ["publish"]

Nota que generate_cover en la Etapa 4 solo depende de draft y no de fact_check; esto significa que la generación de imágenes y la verificación de hechos se ejecutan en paralelo, reduciendo el tiempo total de espera.

Caso 4: Clasificación y escalado de soporte al cliente (Enrutamiento condicional)

Este es un flujo de trabajo de atención al cliente inteligente que enruta automáticamente según el tipo de consulta:

[[workflows]]name = "support_triage"description = "Atención al cliente inteligente: Clasificación → Enrutamiento → Procesamiento → Escalado"[workflows.trigger]type = "message"channel = "whatsapp"pattern = "/help *"# Clasificación[[workflows.tasks]]id = "classify"hand = "researcher"config = {  model = "claude-haiku-4-5",  action = "classify_intent",  categories = ["billing", "technical", "feature_request", "complaint"]}# Facturación → Consulta al sistema[[workflows.tasks]]id = "handle_billing"channel = "webhook"config = { url = "https://billing-api.company.com/lookup" }depends_on = ["classify"]condition = "ctx.intent == 'billing'"# Técnico → Búsqueda en base de conocimientos[[workflows.tasks]]id = "search_kb"hand = "researcher"config = { action = "search_knowledge_base", max_results = 3 }depends_on = ["classify"]condition = "ctx.intent == 'technical'"# Solicitud de función → Registro en Product Board[[workflows.tasks]]id = "log_feature"channel = "webhook"config = { url = "https://productboard-api.company.com/notes" }depends_on = ["classify"]condition = "ctx.intent == 'feature_request'"# Queja → Escalado inmediato a humano[[workflows.tasks]]id = "escalate"channel = "slack"config = { channel = "#urgent-support", mention = "@oncall" }depends_on = ["classify"]condition = "ctx.intent == 'complaint'"# Respuesta final[[workflows.tasks]]id = "reply"channel = "whatsapp"config = { reply_to = "{{.original_message.id}}" }depends_on = ["handle_billing", "search_kb", "log_feature", "escalate"]

Depuración y pruebas de flujos de trabajo

Antes de la implementación en producción, se recomienda usar el modo Dry Run para probar el flujo:

# Dry run (no ejecuta acciones externas reales)openfang workflow run competitor_monitor --dry-run# Ejecutar solo hasta una tarea específicaopenfang workflow run content_pipeline --until draft# Ver el historial de ejecución del flujoopenfang workflow history competitor_monitor --limit 10# Ver registros detallados de una ejecución específicaopenfang workflow inspect competitor_monitor --run-id abc123

Mejores prácticas para flujos de trabajo

Establece tiempos de espera razonables: Cada tarea debe tener un timeout para evitar que un paso bloquee todo el proceso.
Añade mecanismos de reintento: Configura retry = 3 y retroceso exponencial para tareas de red.
Usa condiciones: Evita la ejecución innecesaria de tareas para ahorrar costos de llamadas a la API.
Registra logs de ejecución: Configura log_level = "debug" para facilitar la resolución de problemas.
Monitorea la salud del flujo: Integra las métricas del flujo de trabajo con Prometheus/Grafana.

Preguntas frecuentes

¿Cuántas tareas puede tener un flujo de trabajo como máximo?

No hay un límite estricto. Sin embargo, se recomienda mantener el número de tareas por debajo de 20 para garantizar la mantenibilidad. Los procesos más complejos deben dividirse en sub-flujos.

¿Cómo manejar el fallo de una tarea?

Cada tarea puede configurar su propia estrategia de fallo:

``toml [workflows.tasks.fallback] on_failure = "continue" # continue / retry / abort / escalate retry_count = 3 retry_delay_seconds = 60 fallback_task = "manual_review" # Tarea alternativa tras el fallo``

¿Pueden los flujos de trabajo llamarse entre sí?

Sí. Usando trigger.type = "workflow", un flujo puede activar a otro. Esto permite implementar una lógica de anidamiento compleja.

¿Cómo pasar parámetros dinámicos en un flujo de trabajo?

Utiliza la sintaxis de plantillas {{.variable_name}}. Las variables de contexto provienen de datos del disparador, salidas de tareas anteriores y variables de configuración global.

Siguientes pasos

Guía de configuración de Hands de OpenFang: Aprende la configuración detallada de cada Hand.
Mejores prácticas de seguridad en OpenFang: Protege tus flujos de trabajo automatizados.