🔗 OpenClaw Agent工具链组合

世界上有一种协作叫工具链，工具A的输出变成工具B的输入，像流水线一样高效...

"帮我分析这个网站的数据"，简单一句话，背后却是：网页抓取 → 数据解析 → 数据清洗 → 数据分析 → 报告生成。五步操作，一步不能少。工具链就是这条生产线。

📋 功能介绍

🎯 工具链的核心价值

💡 工具链示例

• 数据管道 - web_fetch → parse → clean → analyze → report
• 内容生成 - research → outline → draft → edit → publish
• 客服流程 - query → classify → search → answer → feedback
• 代码审查 - read → parse → lint → test → report

🚀 使用方法

1. 串行工具链

# 串行工具链配置
agents:
  data-pipeline:
    name: 数据管道
    
    tools:
      - web_fetch
      - html_parse
      - data_clean
      - data_analyze
      - report_generate
      
    # 工具链示例
    toolchain:
      name: webpage_analysis
      
      steps:
        # Step 1: 抓取网页
        - tool: web_fetch
          input:
            url: "{{user_input.url}}"
          output: raw_html
          
        # Step 2: 解析HTML
        - tool: html_parse
          input:
            html: "{{raw_html}}"
            selector: "table.data"
          output: parsed_data
          
        # Step 3: 清洗数据
        - tool: data_clean
          input:
            data: "{{parsed_data}}"
            rules: [remove_nulls, normalize_dates]
          output: clean_data
          
        # Step 4: 分析数据
        - tool: data_analyze
          input:
            data: "{{clean_data}}"
            analysis: [mean, max, min, trend]
          output: analysis_result
          
        # Step 5: 生成报告
        - tool: report_generate
          input:
            data: "{{clean_data}}"
            analysis: "{{analysis_result}}"
          output: final_report

2. 并行工具链

# 并行执行多个工具
agents:
  multi-source-aggregator:
    name: 多源聚合
    
    toolchain:
      name: parallel_search
      
      # 并行分支
      parallel:
        - branch: source_a
          tool: api_call
          input: {source: "source_a", query: "{{query}}"}
          
        - branch: source_b
          tool: api_call
          input: {source: "source_b", query: "{{query}}"}
          
        - branch: source_c
          tool: web_search
          input: {query: "{{query}}"}
          
      # 汇总结果
      merge:
        tool: data_merge
        input:
          sources: ["{{source_a}}", "{{source_b}}", "{{source_c}}"]
          strategy: dedupe

3. 条件工具链

# 根据结果动态路由
agents:
  smart-processor:
    name: 智能处理器
    
    toolchain:
      name: conditional_flow
      
      steps:
        - tool: classify
          input: "{{user_input}}"
          output: classification
          
        # 条件分支
        - switch: "{{classification.type}}"
          cases:
            order_query:
              tool: order_search
              input: {order_id: "{{classification.order_id}}"}
              
            product_query:
              tool: product_search
              input: {product: "{{classification.product_name}}"}
              
            complaint:
              tool: create_ticket
              input: {issue: "{{user_input}}", priority: high}
              
            default:
              tool: web_search
              input: {query: "{{user_input}}"}

4. 循环工具链

# 循环处理分页数据
agents:
  pagination-fetcher:
    name: 分页抓取
    
    toolchain:
      name: paginated_fetch
      
      # 初始参数
      init:
        page: 1
        all_data: []
        
      # 循环条件
      loop:
        condition: "{{page <= max_page}}"
        max_iterations: 100
        
        steps:
          - tool: api_call
            input: {page: "{{page}}"}
            output: page_data
            
          # 累积数据
          - action: append
            target: all_data
            value: "{{page_data.items}}"
            
          # 下一页
          - action: increment
            target: page
            
      # 循环结束后
      final:
        tool: data_analyze
        input: {data: "{{all_data}}"}

✨ 最佳实践

# 竞品分析完整工具链
agents:
  competitor-analyzer:
    name: 竞品分析师
    
    toolchain:
      name: competitor_analysis
      
      # 阶段1：信息收集（并行）
      parallel:
        - branch: product_info
          toolchain:
            - tool: web_search
              input: {query: "{{product}} 特点"}
            - tool: web_fetch
              input: {url: "{{result.url}}"}
              
        - branch: pricing
          tool: web_search
          input: {query: "{{product}} 价格"}
          
        - branch: reviews
          toolchain:
            - tool: web_search
              input: {query: "{{product}} 评价"}
            - tool: sentiment_analyze
              input: {text: "{{result.content}}"}
              
      # 阶段2：数据整合
      - tool: data_merge
        input:
          sources:
            - "{{product_info}}"
            - "{{pricing}}"
            - "{{reviews}}"
            
      # 阶段3：分析对比
      - tool: compare_analyze
        input:
          our_product: "{{our_product_data}}"
          competitor_data: "{{merged_data}}"
          
      # 阶段4：生成报告
      - tool: report_generate
        input:
          analysis: "{{compare_result}}"
          template: competitor_report.md

💻 代码示例

Python工具链

from openclaw import Agent
from openclaw.toolchain import Chain, Step, Parallel

# 定义工具链
chain = Chain([
    # Step 1: 抓取
    Step("web_fetch", input={"url": "{{url}}"}),
    
    # Step 2: 解析
    Step("html_parse", input={"html": "{{step1.content}}"}),
    
    # Step 3: 分析
    Step("data_analyze", input={"data": "{{step2.data}}"})
])

# 创建Agent
agent = Agent(
    name="data-analyst",
    tools=["web_fetch", "html_parse", "data_analyze"],
    toolchain=chain
)

# 执行
result = agent.run(url="https://example.com/data")
print(result.final)

条件链

from openclaw.toolchain import ConditionalChain

chain = ConditionalChain(
    # 分类
    classify_step=Step("classify"),
    
    # 条件分支
    branches={
        "type_a": Step("process_type_a"),
        "type_b": Step("process_type_b"),
        "default": Step("general_process")
    }
)

result = agent.run("处理这个请求", toolchain=chain)

并行链

from openclaw.toolchain import ParallelChain

# 并行执行
chain = ParallelChain([
    Step("fetch_source_a"),
    Step("fetch_source_b"),
    Step("fetch_source_c")
])

# 执行（三个源并行）
results = agent.run("聚合数据", toolchain=chain)

# 合并结果
merged = chain.merge(results)

错误处理

from openclaw.toolchain import Chain, Step

# 带错误处理的链
chain = Chain([
    Step("web_fetch", 
         retry=3,  # 失败重试3次
         fallback=Step("cache_fetch")),  # 失败后用缓存
         
    Step("parse",
         on_error="skip"),  # 解析失败跳过
         
    Step("analyze",
         timeout=60)  # 超时60秒
])

# 执行并处理错误
try:
    result = chain.execute(context)
except ChainError as e:
    print(f"链执行失败: {e.failed_step}")
    print(f"部分结果: {e.partial_results}")

🔗 相关链接

• OpenClaw工具调用 - 工具基础
• OpenClaw工具编排 - 编排策略
• OpenClaw工具组合 - 组合模式
• OpenClaw工作流自动化 - 自动化流程
• ClawHub入门指南 - 发现更多Skills

📊 工具链性能对比