🔗 OpenClaw工具链编排

凌晨2点33分，我在思考一个哲学问题：工具之间有阶级吗？有的，这不是政治不正确——这是Pipeline。一个工具的输出，是另一个工具的输入，这就是工具链编排的艺术。

1. 什么是工具链编排

把多个独立的工具串联起来，形成自动化的处理流水线：

每个环节专注做一件事，串起来就是大事。

2. 基础链式调用

2.1 顺序执行

# 定义工具链
tool_chain:
  name: content_pipeline
  
  steps:
    - id: search
      tool: web_search
      params:
        query: "${input.topic}"
        
    - id: fetch
      tool: web_fetch
      params:
        url: "${search.results[0].url}"
        
    - id: summarize
      tool: summarize
      params:
        content: "${fetch.content}"
        style: "bullet_points"

2.2 数据流转

使用${前一步骤ID.字段}语法引用上一步的输出：

# 步骤间数据传递
steps:
  - id: translate
    tool: translate
    params:
      text: "${fetch.content}"
      target_lang: "zh"
      
  - id: sentiment
    tool: sentiment_analysis
    params:
      text: "${translate.translated_text}"

3. 条件分支

不是所有情况都需要走完整个链路：

tool_chain:
  steps:
    - id: classify
      tool: classifier
      
    - id: route
      type: branch
      conditions:
        - when: "${classify.category == 'news'}"
          goto: news_pipeline
        - when: "${classify.category == 'code'}"
          goto: code_pipeline
        - default:
          goto: general_pipeline

4. 并行执行

多个独立任务可以同时跑：

tool_chain:
  steps:
    - id: parallel_fetch
      type: parallel
      tasks:
        - id: fetch_news
          tool: news_api
          
        - id: fetch_social
          tool: social_media_api
          
        - id: fetch_docs
          tool: internal_docs
          
    - id: merge
      tool: content_merger
      params:
        sources:
          - "${fetch_news.articles}"
          - "${fetch_social.posts}"
          - "${fetch_docs.pages}"

5. 循环处理

对列表中的每个元素执行相同操作：

tool_chain:
  steps:
    - id: get_items
      tool: database_query
      
    - id: process_each
      type: loop
      over: "${get_items.results}"
      each: item
      steps:
        - id: enrich
          tool: data_enricher
          params:
            data: "${item}"
            
        - id: validate
          tool: validator
          params:
            data: "${enrich.enriched_data}"

6. 错误处理在链中

6.1 跳过失败的步骤

step:
  id: fetch_medium
  tool: web_fetch
  on_error: skip  # 失败就跳过，继续下一步
  fallback:
    content: "无法获取内容"

6.2 重试后继续

step:
  id: api_call
  tool: external_api
  retry:
    max_attempts: 3
    backoff: exponential
  on_error_continue: true

6.3 全链回滚

tool_chain:
  transaction: true  # 开启事务模式
  
  steps:
    - id: step1
      compensation: rollback_step1
    - id: step2  
      compensation: rollback_step2
      
  on_failure: compensate_all

7. 实战案例：新闻聚合Pipeline

tool_chain:
  name: news_aggregation
  
  trigger:
    schedule: "0 */2 * * *"  # 每2小时
  
  steps:
    # 1. 获取RSS源列表
    - id: get_feeds
      tool: database_query
      params:
        sql: "SELECT * FROM rss_feeds WHERE active = true"
        
    # 2. 并行抓取所有RSS
    - id: fetch_all
      type: parallel
      over: "${get_feeds.results}"
      each: feed
      steps:
        - id: fetch_rss
          tool: rss_fetcher
          params:
            url: "${feed.url}"
            
    # 3. 扁平化文章列表
    - id: flatten
      tool: array_flatten
      params:
        arrays: "${fetch_all.results}"
        
    # 4. 批量内容提取
    - id: extract_all
      type: parallel
      over: "${flatten.items}"
      each: article
      concurrency: 10
      steps:
        - id: fetch_content
          tool: web_fetch
          params:
            url: "${article.link}"
          timeout: 15s
          
        - id: summarize
          tool: llm_summarize
          params:
            content: "${fetch_content.text}"
            max_length: 200
            
    # 5. 去重排序
    - id: dedupe_sort
      tool: news_deduplicator
      params:
        articles: "${extract_all.results}"
        sort_by: "published_at"
        
    # 6. 存储并发布
    - id: save
      tool: database_insert
      params:
        table: "news_articles"
        data: "${dedupe_sort.unique_articles}"
        
    - id: notify
      tool: message_send
      params:
        channel: "news_channel"
        content: "已更新 ${dedupe_sort.count} 篇文章"

8. 性能优化技巧

1. ✅ 独立任务并行化，减少总等待时间
2. ✅ 设置合理的超时，防止整个链路卡住
3. ✅ 使用缓存避免重复计算
4. ✅ 大任务分批处理，避免内存溢出
5. ✅ 监控每个步骤的耗时，找瓶颈

9. 与Workflow的区别

Tool Chaining更轻量，适合确定性的处理流程。需要更复杂的条件判断、人工介入时，考虑使用Workflow编排。