Guia de Configuração de Fluxos de Trabalho Automatizados no OpenFang: De um Único Agente a uma Equipe de Agentes

Slug: openfang-workflow-automation-guide Categoria: usage-guides (Guias de Uso) Palavras-chave alvo: Fluxo de trabalho OpenFang, Processos de automação OpenFang, Orquestração Multi-Agent OpenFang Intenção de busca: Informativa (usuários que desejam construir processos complexos de automação com agentes) Contagem de palavras alvo: ~2000 palavras Idioma: pt

Por que precisamos de fluxos de trabalho?

As tarefas que um único agente pode realizar são limitadas. Quando você precisa de um processo de ponta a ponta, como "pesquisar um tópico → escrever um artigo → verificar fatos → publicar → monitorar feedback", você precisa orquestrar vários agentes em um fluxo de trabalho.

O motor de fluxo de trabalho do OpenFang suporta três modos de orquestração: Pipeline serial, distribuição paralela (Fan-out) e ramificação condicional (Conditional). Você pode combinar Hands e agentes como peças de LEGO para criar processos de automação complexos.

Este artigo o guiará pelo domínio da configuração de fluxos de trabalho no OpenFang, do simples ao complexo, através de quatro casos práticos.

Conceitos principais de fluxo de trabalho

Antes de mergulhar nos casos, entenda alguns conceitos fundamentais:

Conceito	Descrição
Task	A unidade mínima de execução de um fluxo de trabalho; uma Task corresponde a um Hand ou a um agente
Pipeline	Conecta várias Tasks em série, onde a saída de uma Task torna-se automaticamente a entrada da próxima
Fan-out	Distribui uma tarefa para vários agentes processarem em paralelo, consolidando os resultados ao final
Trigger	A fonte de evento que dispara o fluxo de trabalho: agendamento, Webhook, alteração de arquivo, comando de mensagem
Context	Estado compartilhado passado entre as etapas do fluxo de trabalho, semelhante ao escopo de variáveis em linguagens de programação

Caso 1: Resumo diário do setor (Pipeline simples)

Este é o fluxo de trabalho Pipeline mais simples — gera automaticamente um resumo do setor todas as manhãs:

[[workflows]]name = "daily_industry_briefing"description = "Gera um resumo das notícias do setor todas as manhãs às 8h"[workflows.trigger]type = "schedule"cron = "0 8 * * 1-5"             # 8h da manhã em dias úteis[[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"

A execução do fluxo é: Coleta pelo Collector → Resumo pelo Researcher → Notificação no Slack. depends_on define a relação de dependência entre as Tasks.

Caso 2: Monitoramento e alertas de concorrentes (Paralelo + Consolidação)

Este fluxo de trabalho monitora vários concorrentes simultaneamente e consolida os resultados para análise:

[[workflows]]name = "competitor_monitor"description = "Monitora 5 sites de concorrentes em paralelo e envia alertas após consolidar as mudanças"[workflows.trigger]type = "schedule"cron = "0 */6 * * *"# Estágio 1: Coleta 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" }# Estágio 2: Consolidação (aguarda a conclusão de todas as coletas)[[workflows.tasks]]id = "aggregate"hand = "researcher"config = { action = "merge_and_prioritize" }depends_on = ["check_competitor_a", "check_competitor_b", "check_competitor_c"]# Estágio 3: Julgamento condicional — notificar apenas se houver mudanças importantes[[workflows.tasks]]id = "check_significance"hand = "predictor"config = { action = "assess_impact" }depends_on = ["aggregate"]# Estágio 4: Notificação 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"

Os padrões chave aqui são:

Fan-out: As 5 Tasks de coleta do Estágio 1 rodam em paralelo, sem necessidade de depends_on
Barreira: A Task Aggregate no Estágio 2 depende da conclusão de todas as 5 coletas
Condicional: O Estágio 4 escolhe canais de notificação diferentes com base na pontuação de importância

Caso 3: Pipeline de produção de conteúdo (Colaboração multi-agente)

Este é o cenário clássico de colaboração multi-agente: redação, revisão e criação de imagens por IA:

[[workflows]]name = "content_pipeline"description = "Pipeline de produção de conteúdo totalmente automatizado, do tópico à publicação"[workflows.trigger]type = "webhook"endpoint = "/webhook/content/new"secret = "${WEBHOOK_SECRET}"# Estágio 1: Pesquisa[[workflows.tasks]]id = "research"hand = "researcher"config = {  depth = "deep",  source_types = ["academic", "news", "social"],  target_length = "3000_words"}timeout = 900# Estágio 2: Redação (usando modelos diferentes)[[workflows.tasks]]id = "draft"hand = "researcher"              # Reutiliza o Researcher Hand, mas com um prompt diferenteconfig = {  model = "claude-opus-4-8",  instruction = "draft_article",  style = "professional_yet_approachable"}depends_on = ["research"]# Estágio 3: Verificação de fatos[[workflows.tasks]]id = "fact_check"hand = "researcher"config = {  model = "claude-haiku-4-5",   # Usa um modelo rápido para verificação  action = "verify_claims",  strictness = "high"}depends_on = ["draft"]# Estágio 4: Geração de imagem de capa[[workflows.tasks]]id = "generate_cover"hand = "clip"config = {  action = "generate_cover_image",  style = "modern_tech",  resolution = "1280x720"}depends_on = ["draft"]           # Não depende da verificação de fatos, pode rodar em paralelo# Estágio 5: Otimização SEO[[workflows.tasks]]id = "seo_optimize"hand = "researcher"config = { action = "seo_optimize", target_keyword = "{{.keyword}}" }depends_on = ["fact_check"]# Estágio 6: Publicação[[workflows.tasks]]id = "publish"channel = "webhook"config = { url = "{{.cms_endpoint}}", method = "POST" }depends_on = ["seo_optimize", "generate_cover"]# Estágio 7: Promoção[[workflows.tasks]]id = "promote"channel = "twitter"config = { action = "post_thread", hashtags = ["#AI", "#Automation"] }depends_on = ["publish"]

Observe que a Task generate_cover no Estágio 4 depende apenas de draft e não de fact_check — isso significa que a geração da imagem e a verificação de fatos serão executadas em paralelo, reduzindo o tempo total de espera.

Caso 4: Triagem e escalonamento de suporte ao cliente (Roteamento condicional)

Este é um fluxo de trabalho de suporte inteligente que roteia automaticamente para diferentes processos com base no tipo de problema:

[[workflows]]name = "support_triage"description = "Suporte inteligente: Triagem → Roteamento → Processamento → Escalonamento"[workflows.trigger]type = "message"channel = "whatsapp"pattern = "/help *"# Triagem[[workflows.tasks]]id = "classify"hand = "researcher"config = {  model = "claude-haiku-4-5",  action = "classify_intent",  categories = ["billing", "technical", "feature_request", "complaint"]}# Faturamento → Consulta ao 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 → Pesquisa na base de conhecimento[[workflows.tasks]]id = "search_kb"hand = "researcher"config = { action = "search_knowledge_base", max_results = 3 }depends_on = ["classify"]condition = "ctx.intent == 'technical'"# Solicitação de recurso → Registro no 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'"# Reclamação → Escalonamento imediato para humano[[workflows.tasks]]id = "escalate"channel = "slack"config = { channel = "#urgent-support", mention = "@oncall" }depends_on = ["classify"]condition = "ctx.intent == 'complaint'"# Resposta final[[workflows.tasks]]id = "reply"channel = "whatsapp"config = { reply_to = "{{.original_message.id}}" }depends_on = ["handle_billing", "search_kb", "log_feature", "escalate"]

Depuração e teste de fluxos de trabalho

Antes da implantação em produção, recomendamos usar o modo Dry Run para testar o fluxo:

# Dry run (não executa ações externas reais)openfang workflow run competitor_monitor --dry-run# Executa apenas até uma Task específicaopenfang workflow run content_pipeline --until draft# Visualiza o histórico de execução do fluxoopenfang workflow history competitor_monitor --limit 10# Visualiza logs detalhados de uma execução específicaopenfang workflow inspect competitor_monitor --run-id abc123

Melhores práticas para fluxos de trabalho

Defina tempos limite razoáveis: Cada Task deve ter um timeout configurado para evitar que uma única etapa trave todo o processo.
Adicione mecanismos de repetição: Tasks de requisição de rede devem ter retry = 3 e backoff exponencial.
Use condições: Evite a execução desnecessária de Tasks para economizar custos de chamadas de API.
Registre logs de execução: Configure log_level = "debug" para facilitar a resolução de problemas.
Monitore a saúde do fluxo: Integre as métricas do fluxo de trabalho ao Prometheus/Grafana.

Perguntas frequentes

Quantas Tasks um fluxo de trabalho pode suportar no máximo?

Não há limite rígido. No entanto, recomendamos manter o número de Tasks em um único fluxo abaixo de 20 para garantir a manutenibilidade. Processos mais complexos devem ser divididos em sub-fluxos.

Como lidar com falhas na execução de uma Task?

Cada Task pode ter uma estratégia de falha configurada individualmente:

``toml [workflows.tasks.fallback] on_failure = "continue" # continue / retry / abort / escalate retry_count = 3 retry_delay_seconds = 60 fallback_task = "manual_review" # Task de reserva após falha``

Fluxos de trabalho podem chamar uns aos outros?

Sim. Usando trigger.type = "workflow", você pode fazer com que um fluxo dispare outro. Combinado com condições, é possível implementar lógicas de aninhamento complexas.

Como passar parâmetros dinâmicos em um fluxo de trabalho?

Use a sintaxe de template {{.variable_name}}. As fontes de variáveis de contexto incluem: dados carregados pelo gatilho, saídas de Tasks anteriores e variáveis de configuração global.

Próximos passos

Guia de Configuração de Hands do OpenFang: Entenda a configuração detalhada de cada Hand
Melhores Práticas de Segurança no OpenFang: Proteja seus fluxos de trabalho automatizados