Tool Chaining 工具链

发布时间：2026-03-23 | 分类：Agent工具系统

定义

世界上有一种能力叫做Tool Chaining，它就像一个瑞士军刀——每个工具单独都很实用，但组合起来才能应对各种复杂场景...

Tool Chaining是将多个工具串联使用，前一个工具的输出作为下一个工具的输入，形成自动化流水线的技术。与Prompt Chaining不同，Tool Chaining侧重于工具层面的串联。

核心原理

1. 工具串联模式


┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐
│ Tool A  │ → │ Tool B  │ → │ Tool C  │ → │ Result  │
│  搜索   │    │  获取   │    │  处理   │    │  最终   │
└─────────┘    └─────────┘    └─────────┘    └─────────┘
    ↓              ↓              ↓
  data1         data2          data3
    ↓              ↓              ↓
         作为下一个工具的输入

2. 串联类型

类型	特点	示例
线性串联	A→B→C 顺序执行	搜索→获取→总结
并行分支	A→[B,C] 同时执行	获取→并行处理多数据源
扇出扇入	A→[B,C,D]→E 聚合	搜索多个→合并结果
循环迭代	A→B→A 反复执行	处理→检查→重处理

3. 数据流处理

格式转换：工具输出转换为目标格式
字段映射：提取需要的字段传递给下一个工具
错误传递：某工具失败时的异常处理
状态传递：累积上下文信息

OpenClaw实战应用

案例：新闻摘要自动化

用户: "帮我总结今天的AI新闻"

工具链执行:

1. web_search (搜索新闻)
   输入: "AI新闻 2026"
   输出: [{title, url, snippet}, ...]
   
2. web_fetch (获取详情) × 并行
   输入: [url1, url2, url3, url4, url5]
   输出: [article1, article2, article3, article4, article5]
   
3. llm_summarize (生成摘要)
   输入: [articles...]
   输出: 摘要文本
   
4. write (保存结果)
   输入: {content: 摘要, path: "news/2026-03-23.html"}
   输出: 文件保存成功

OpenClaw内置工具链


# OpenClaw典型工作流

信息收集链:
  web_search → web_fetch → 提取关键信息
  
内容处理链:
  read → edit → write
  
多源验证链:
  [web_search(A), web_search(B)] → 比对 → 验证
  
自动化发布链:
  生成内容 → 上传文件 → 发送通知

工具链配置


# tool_chain.yaml
chains:
  research_summary:
    description: "研究摘要自动化"
    steps:
      - tool: web_search
        params:
          query: "{query}"
          count: 5
        output_var: search_results
        
      - tool: web_fetch
        params:
          url: "{search_results[0].url}"
        output_var: article_1
        
      - tool: web_fetch
        params:
          url: "{search_results[1].url}"
        output_var: article_2
        
      - tool: llm_summarize
        params:
          content: "{article_1} {article_2}"
        output_var: summary
        
      - tool: write
        params:
          path: "output/summary.md"
          content: "{summary}"

代码示例

基础工具链实现


class ToolChain:
    """工具链执行器"""
    
    def __init__(self, tools: dict):
        self.tools = tools
        self.context = {}
    
    async def execute(self, steps: list) -> dict:
        """执行工具链"""
        for step in steps:
            tool_name = step["tool"]
            params = self._resolve_params(step["params"])
            
            # 执行工具
            tool = self.tools[tool_name]
            result = await tool.execute(**params)
            
            # 保存到上下文
            output_var = step.get("output_var", "result")
            self.context[output_var] = result
        
        return self.context
    
    def _resolve_params(self, params: dict) -> dict:
        """解析参数，支持变量引用"""
        resolved = {}
        for key, value in params.items():
            if isinstance(value, str) and value.startswith("{"):
                # 变量引用
                var_path = value[1:-1]
                resolved[key] = self._get_var(var_path)
            else:
                resolved[key] = value
        return resolved
    
    def _get_var(self, path: str):
        """获取上下文变量"""
        keys = path.split(".")
        value = self.context
        for key in keys:
            value = value[key]
        return value

# 使用示例
chain = ToolChain(tools)
result = await chain.execute([
    {"tool": "web_search", "params": {"query": "AI"}, "output_var": "search"},
    {"tool": "web_fetch", "params": {"url": "{search[0].url}"}, "output_var": "article"},
    {"tool": "llm_summarize", "params": {"content": "{article}"}, "output_var": "summary"}
])

LangChain工具链


from langchain.tools import tool
from langgraph.graph import StateGraph

# 定义工具
@tool
def search_news(query: str):
    """搜索新闻"""
    return web_search(query)

@tool
def fetch_article(url: str):
    """获取文章内容"""
    return web_fetch(url)

@tool  
def summarize(text: str):
    """总结内容"""
    return llm.summarize(text)

# 创建图
workflow = StateGraph(NewsState)

workflow.add_node("search", lambda state: {"articles": search_news(state["query"])})
workflow.add_node("fetch", lambda state: {"content": fetch_article(state["articles"])})
workflow.add_node("summarize", lambda state: {"summary": summarize(state["content"])})

workflow.set_entry_point("search")
workflow.add_edge("search", "fetch")
workflow.add_edge("fetch", "summarize")
workflow.set_finish_point("summarize")

app = workflow.compile()
result = await app.ainvoke({"query": "AI news"})

OpenClaw工具链API


// OpenClaw 中的工具链调用
async function executeToolChain(chainConfig) {
    const context = {};
    
    for (const step of chainConfig.steps) {
        // 获取工具
        const tool = getTool(step.tool);
        
        // 解析参数（支持变量替换）
        const params = resolveParams(step.params, context);
        
        // 执行工具
        const result = await tool.execute(params);
        
        // 保存结果
        context[step.output_var || 'result'] = result;
        
        // 检查错误
        if (result.error && step.required) {
            throw new Error(`Tool ${step.tool} failed: ${result.error}`);
        }
    }
    
    return context;
}

// 调用示例
const result = await executeToolChain({
    steps: [
        { tool: 'web_search', params: { query: 'AI 2026' }, output_var: 'news' },
        { tool: 'web_fetch', params: { url: '{news[0].url}' }, output_var: 'article' },
        { tool: 'llm_summarize', params: { text: '{article}' }, output_var: 'summary' },
        { tool: 'write', params: { path: 'summary.md', content: '{summary}' } }
    ]
});

最佳实践

✅ 定义清晰接口：工具输入输出格式标准化
✅ 处理边界情况：空结果、网络错误、超时等
✅ 添加重试机制：临时失败自动重试
✅ 实现熔断保护：防止错误链式传播
✅ 记录执行日志：便于调试和问题排查
❌ 避免链路过深：增加延迟和失败风险
❌ 不要硬编码参数：使用变量引用提高灵活性