📅 更新于 2026-05-29 | OpenClawAgentic Workflow自动化

🔄 Agentic Workflow 详解

📖 什么是 Agentic Workflow？

Agentic Workflow（智能体工作流）是由AI Agent驱动的动态工作流，与传统固定流程不同，Agent可以根据中间结果自主决策下一步操作，甚至修改整个流程。它结合了Agent的自主性和工作流的可靠性。

传统工作流 vs Agentic Workflow：

🧠 核心原理

1. 工作流模式

Agentic Workflow的常见模式：

• Sequential（顺序）：按步骤执行，但每步可由Agent调整
• Conditional（条件）：Agent根据结果决定下一步
• Iterative（迭代）：循环执行直到满足条件
• Parallel（并行）：多个分支同时执行
• Adaptive（自适应）：根据执行情况动态调整流程

2. 工作流定义

Agentic Workflow可以用代码或声明式定义：

# 声明式定义（YAML）
workflow: seo-content-pipeline
steps:
  - name: research
    agent: researcher
    task: "搜索热门术语"
    output: terms
  
  - name: generate
    agent: writer
    task: "为每个术语生成页面"
    input: terms
    output: pages
    parallel: true  # 并行生成
  
  - name: publish
    agent: publisher
    task: "发布页面并更新sitemap"
    input: pages

3. Agent决策点

工作流中的关键决策点：

工作流执行：
Step 1: 搜索术语 → 成功，得到10个术语
Step 2: 生成页面 → Agent检查：术语是否重复？
         → 跳过已存在的3个，生成7个新页面
Step 3: 更新sitemap → Agent检查：是否有死链？
         → 发现2个死链，先修复再更新
Step 4: 通知 → 任务完成，发送飞书通知

🛠️ OpenClaw 实战应用

场景一：妙趣AI的术语生成工作流

本轮任务就是一个Agentic Workflow：

# 妙趣AI 术语生成工作流
workflow: marketing-glossary-pages
trigger: cron "0 20 * * *"

steps:
  # Step 1: 搜索热门术语
  - task: "搜索OpenClaw热门术语"
    tools: [web_search, coze-web-fetch]
    output: term_list
  
  # Step 2: 过滤已存在的
  - task: "检查/glossary/目录，过滤已存在术语"
    input: term_list
    output: new_terms
  
  # Step 3: 生成页面（并行）
  - task: "为每个术语生成页面"
    input: new_terms
    tools: [write, edit]
    parallel: true
    maxParallel: 5
  
  # Step 4: 更新sitemap
  - task: "更新sitemap.xml"
    tools: [exec, edit]
  
  # Step 5: 通知
  - task: "发送飞书通知"
    tools: [message]
    condition: "new_terms.length > 0"

场景二：自适应工作流（错误恢复）

工作流执行中遇到错误，Agent自动调整：

# 执行术语生成时
try:
    page = await generatePage(term)
    if page.hasError:
        # Agent自主决策：重试 or 跳过
        if page.error === 'timeout':
            await retryWithLongerTimeout(term)
        elif page.error === 'skills_not_found':
            # 换一种方式：直接用LLM生成，不依赖Skill
            await generatePageWithoutSkill(term)
        else:
            logError(page.error)
            # 继续下一个术语
except Exception as e:
    # Agent决定：记录错误，不中断整个工作流
    logError(e)
    continue

场景三：条件分支工作流

# 根据时间决定工作流
if (hour >= 0 && hour < 8) {
  // 早班：生成新闻日报
  workflow = 'ai-news-daily';
} else if (hour >= 8 && hour < 20) {
  // 白班：社区运营
  workflow = 'community-engagement';
} else {
  // 晚班：SEO内容生成
  workflow = 'seo-content-pipeline';
}

await runWorkflow(workflow);

💻 代码示例

示例1：简单的Agentic Workflow引擎

// Agentic Workflow Engine
class AgenticWorkflow {
  constructor(agent) {
    this.agent = agent;
    this.steps = [];
    this.context = {};
  }
  
  addStep(step) {
    this.steps.push(step);
  }
  
  async execute() {
    for (const step of this.steps) {
      // 检查条件
      if (step.condition && !this.evaluateCondition(step.condition)) {
        continue;
      }
      
      // Agent执行步骤
      const result = await this.agent.execute({
        task: step.task,
        input: this.resolveInput(step.input),
        tools: step.tools
      });
      
      // 保存输出到上下文
      if (step.output) {
        this.context[step.output] = result;
      }
      
      // 检查是否继续
      if (result.shouldStop) {
        break;
      }
    }
    
    return this.context;
  }
  
  resolveInput(inputRef) {
    if (!inputRef) return undefined;
    // 解析输入引用，从上下文中获取
    return inputRef.split('.').reduce((obj, key) => obj?.[key], this.context);
  }
  
  evaluateCondition(condition) {
    // 简化：实际应使用表达式解析器
    return eval(condition);  // ⚠️ 注意安全
  }
}

// 使用
const workflow = new AgenticWorkflow(agent);
workflow.addStep({
  name: 'research',
  task: '搜索热门OpenClaw术语',
  output: 'terms'
});
workflow.addStep({
  name: 'generate',
  task: '生成术语页面',
  input: 'terms',
  output: 'pages',
  parallel: true
});
await workflow.execute();

示例2：并行步骤执行

// 支持并行的Workflow步骤
async function executeParallel(steps, context) {
  const promises = steps.map(step => 
    this.agent.execute({
      task: step.task,
      input: context[step.input],
      tools: step.tools
    })
  );
  
  const results = await Promise.all(promises);
  
  // 汇总结果
  results.forEach((result, index) => {
    const step = steps[index];
    if (step.output) {
      context[step.output] = context[step.output] || [];
      context[step.output].push(result);
    }
  });
  
  return context;
}

示例3：工作流定义（YAML格式）

# seo-pipeline.yaml
name: seo-pipeline
description: SEO内容生成流水线

steps:
  - id: search
    task: 搜索热门关键词
    tools: [web_search]
    output: keywords
  
  - id: analyze
    task: 分析关键词难度
    input: keywords
    tools: [seo_analyzer]
    output: analyzed_keywords
  
  - id: generate
    task: 生成SEO页面
    input: analyzed_keywords
    tools: [content_generator, write]
    output: pages
    parallel: true
    maxParallel: 3
  
  - id: optimize
    task: SEO优化
    input: pages
    tools: [seo_optimizer]
  
  - id: publish
    task: 发布页面
    input: pages
    tools: [git, exec]
    condition: "pages.length > 0"

✅ 最佳实践

📊 工作流框架对比

🎬 周星驰式总结

就像《功夫》里的"星爷练功"——不是死板地按套路打，而是根据实际情况灵活应变。Agentic Workflow就是让工作流"活"起来，该快则快，该慢则慢，该绕路就绕路。记住：死流程是地图，活流程是导航——能绕开堵车的才是好导航！

🔗 相关链接

• Agent Workflow 详解
• Agent Skill Composition 详解
• Workflow Engine 详解
• Plan-and-Execute 详解
• OpenClaw 官方文档