⚙️ Agent Workflow Engine（智能体工作流引擎）详解

背景：妙趣AI的"六轨并行"运营体系就是由Workflow Engine驱动的——每天早上自动生成AI新闻日报。

// OpenClaw Workflow配置
const dailyNewsWorkflow = {
  name: 'daily-ai-news',
  cron: '0 0 * * *',        // 每天午夜执行
  timezone: 'Asia/Shanghai',
  
  steps: [
    {
      id: 'fetch_sources',
      agent: 'news_agent',
      skills: ['rss-fetcher', 'web-scraper', 'trending-checker'],
      parallel: true,        // 并行获取多源
      params: {
        sources: [
          'https://openai.com/blog/rss.xml',
          'https://www.anthropic.com/rss.xml',
          'https://huggingface.co/blog/feed.xml'
        ]
      }
    },
    {
      id: 'curate_news',
      agent: 'curation_agent', 
      depends_on: ['fetch_sources'],
      skills: ['news-ranking', 'de-duplication', 'relevance-check'],
      params: {
        maxItems: 10,
        categories: ['📰新闻动态', '💡创新案例', '📖方法洞察']
      }
    },
    {
      id: 'generate_html',
      agent: 'writing_agent',
      depends_on: ['curate_news'],
      skills: ['html-generator', 'miaoquai-style'],
      params: {
        template: 'news_daily',
        outputPath: '/var/www/miaoquai/news/'
      }
    },
    {
      id: 'publish_update',
      agent: 'publisher_agent',
      depends_on: ['generate_html'],
      skills: ['sitemap-updater', 'discord-poster', 'feishu-notifier'],
      params: {
        discordChannel: '1483699648890802201'
      }
    }
  ],
  
  errorHandling: {
    onStepFailure: 'retry',
    maxRetries: 3,
    onFinalFailure: 'notify_admin'
  }
};

// 执行工作流
const result = await workflowEngine.run(dailyNewsWorkflow);
console.log(`状态: ${result.status}`);  // completed / failed
console.log(`耗时: ${result.duration}ms`);
console.log(`产出: ${result.steps['generate_html'].output.path}`);

问题：用户的问题可能是任意领域，无法用固定流程。

解决方案：让AI Agent作为"路由器"，动态决定下一步。

// 动态路由工作流
const dynamicWorkflow = {
  name: 'smart-router',
  router: {
    agent: 'router_agent',
    skill: 'intent-classification',
    classify: async (query) => {
      const intent = await classifyIntent(query);
      switch (intent.type) {
        case 'coding':
          return 'code_assistance';
        case 'documentation':
          return 'doc_retrieval';
        case 'tool_selection':
          return 'tool_recommendation';
        default:
          return 'general_qa';
      }
    }
  },
  branches: {
    code_assistance: {
      agents: ['code_agent', 'review_agent'],
      skills: ['code-generator', 'debug-analysis', 'test-writer'],
      outputFormat: 'code_with_explanations'
    },
    doc_retrieval: {
      agents: ['search_agent', 'writing_agent'],
      skills: ['web-search', 'doc-summarizer', 'citation-formatter'],
      outputFormat: 'documented_answer'
    },
    tool_recommendation: {
      agents: ['analysis_agent'],
      skills: ['tool-database-query', 'comparison-generator'],
      outputFormat: 'comparison_table'
    }
  }
};

// 用户提问：用Python写一个Web Scraper
const response = await workflowEngine.run(dynamicWorkflow, {
  query: '如何用Python写一个Web Scraper？'
});
// 自动路由到：code_assistance → code_generator + review_agent
// 输出：完整的Python爬虫代码 + 最佳实践建议

问题：某步Agent崩溃或API限流，影响整个工作流。

// 工作流容错配置
const faultTolerantWorkflow = {
  name: 'resilient-processing',
  
  steps: [
    {
      id: 'step1',
      timeout: 30000,          // 30秒超时
      retry: {
        maxAttempts: 3,
        backoff: 'exponential', // 指数退避：1s, 2s, 4s
        onRetry: (attempt, error) => {
          console.log(`第${attempt}次重试，原因: ${error.message}`);
          if (attempt === 2) {
            return { useFallback: true }; // 第3次用降级方案
          }
        }
      },
      fallback: {
        type: 'simplified',    // 降级到简化版
        agent: 'fallback_agent'
      }
    }
  ],
  
  overallTimeout: 120000,      // 整个工作流最多2分钟
  onComplete: async (result) => {
    // 异步回调：发送通知
    await notifyCompletion(result);
  }
};