Agent工具调用与编排综述 - 从单一工具到复杂工作流的完整指南

Agent工具调用与编排综述 - 从单一工具到复杂工作流的完整指南

研究日期: 2026-03-31
关键词: Tool Calling, Function Calling, Workflow Orchestration, Tool Composition, API Integration, MCP
适用场景: API集成、工作流自动化、复杂任务执行、外部系统交互

目录

一、引言：工具是Agent的手脚

1.1 为什么Agent需要工具？

没有工具的Agent：

1
2
3
4
5
6
7
8
9
纯LLM Agent的局限:
  用户: "帮我查询北京的天气"
  Agent: "我无法访问实时数据"  ❌
  
  用户: "帮我订一张机票"
  Agent: "我无法执行实际操作"  ❌
  
  用户: "帮我分析这个CSV文件"
  Agent: "我无法处理大量数据"  ❌

有工具的Agent：

1
2
3
4
5
6
7
8
9
工具增强的Agent能力:
  用户: "帮我查询北京的天气"
  Agent: [调用天气API] → "北京今天晴，15-25度"  ✅
  
  用户: "帮我订一张机票"
  Agent: [调用订票系统] → "已为您预订..."  ✅
  
  用户: "帮我分析这个CSV文件"
  Agent: [调用数据分析工具] → "分析完成..."  ✅

1.2 工具调用的核心价值

能力 无工具 有工具 价值提升
实时数据 无法获取 API调用 ✅ 打破时间限制
外部系统 无法交互 系统集成 ✅ 打破边界限制
计算能力 有限 专业工具 ✅ 打破能力限制
文件操作 无法处理 文件系统 ✅ 打破存储限制
网络访问 无 爬虫/API ✅ 打破空间限制

1.3 工具调用的演进

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
阶段1: 硬编码工具 (2020-2022)
  ├─ 预定义工具集
  ├─ 简单if-else逻辑
  └─ 无自动选择

阶段2: 函数调用 (2022-2023)
  ├─ OpenAI Function Calling
  ├─ 自动参数填充
  └─ 单工具调用

阶段3: 多工具编排 (2023-2024)
  ├─ 工具链组合
  ├─ 条件分支
  └─ 并行执行

阶段4: 自主工具使用 (2024-2025)
  ├─ 工具自主发现
  ├─ 动态组合
  └─ 自我优化

阶段5: MCP协议 (2025-)
  ├─ 标准化工具接口
  ├─ 跨平台互操作
  └─ 生态系统

1.4 工具调用 vs 相关概念

1
2
3
4
5
6
7
8
9
10
11
12
13
14
工具调用 (Tool Calling)
  └─ 执行单个工具

工具编排 (Tool Orchestration)
  └─ 协调多个工具的执行

工作流 (Workflow)
  └─ 定义工具执行的有向图

技能 (Skill)
  └─ 工具的高层抽象和组合

插件 (Plugin)
  └─ 可动态加载的工具包

二、工具调用形式化定义

2.1 工具定义

定义：工具是一个六元组 $T = (N, D, I, O, P, E)$

  • $N$：工具名称
  • $D$：工具描述
  • $I$：输入参数模式（JSON Schema）
  • $O$：输出模式
  • $P$：前置条件
  • $E$：执行函数

2.2 工具调用过程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
tool_definition = {
    "name": "get_weather",
    "description": "获取指定城市的天气信息",
    "parameters": {
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "城市名称，如：北京、上海"
            },
            "unit": {
                "type": "string",
                "enum": ["celsius", "fahrenheit"],
                "default": "celsius"
            }
        },
        "required": ["city"]
    },
    "returns": {
        "type": "object",
        "properties": {
            "temperature": "number",
            "condition": "string",
            "humidity": "number"
        }
    }
}

# 调用示例
tool_call = {
    "name": "get_weather",
    "arguments": {
        "city": "北京",
        "unit": "celsius"
    }
}

# 执行结果
result = {
    "temperature": 20,
    "condition": "晴",
    "humidity": 45
}

2.3 工具调用生命周期

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
1. 工具发现 (Discovery)
   └─ Agent 了解可用工具

2. 工具选择 (Selection)
   └─ 根据任务选择合适工具

3. 参数填充 (Parameter Binding)
   └─ 从上下文中提取参数

4. 执行 (Execution)
   └─ 调用工具并获取结果

5. 结果处理 (Result Processing)
   └─ 解析结果并整合到响应

6. 错误处理 (Error Handling)
   └─ 处理执行失败的情况

三、工具描述与发现

3.1 工具描述标准

3.1.1 OpenAI Function Calling 格式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
{
  "name": "search_web",
  "description": "搜索互联网获取信息",
  "parameters": {
    "type": "object",
    "properties": {
      "query": {
        "type": "string",
        "description": "搜索查询词"
      },
      "num_results": {
        "type": "integer",
        "description": "返回结果数量",
        "default": 5
      }
    },
    "required": ["query"]
  }
}

3.1.2 JSON Schema 扩展

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
{
  "name": "analyze_data",
  "description": "分析数据集",
  "parameters": {
    "type": "object",
    "properties": {
      "data_source": {
        "type": "string",
        "format": "uri",
        "description": "数据源URL或路径"
      },
      "analysis_type": {
        "type": "string",
        "enum": ["statistical", "trend", "anomaly"],
        "description": "分析类型"
      }
    },
    "required": ["data_source", "analysis_type"]
  },
  "x-execution": {
    "timeout": 30,
    "retries": 3,
    "rate_limit": "10/minute"
  }
}

3.1.3 OpenAPI Specification

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
openapi: 3.0.0
info:
  title: Weather API
  version: 1.0.0

paths:
  /weather:
    get:
      operationId: getWeather
      summary: 获取天气信息
      parameters:
        - name: city
          in: query
          required: true
          schema:
            type: string
          description: 城市名称
      responses:
        '200':
          description: 成功
          content:
            application/json:
              schema:
                type: object
                properties:
                  temperature:
                    type: number
                  condition:
                    type: string

3.2 工具注册表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class ToolRegistry:
    def __init__(self):
        self.tools = {}
        self.categories = defaultdict(list)
    
    def register(self, tool_definition):
        """
        注册工具
        """
        tool_name = tool_definition["name"]
        
        # 存储工具定义
        self.tools[tool_name] = tool_definition
        
        # 分类索引
        category = tool_definition.get("category", "general")
        self.categories[category].append(tool_name)
    
    def discover(self, query):
        """
        发现工具
        """
        # 关键词匹配
        matching_tools = []
        
        for tool_name, tool_def in self.tools.items():
            # 名称匹配
            if query.lower() in tool_name.lower():
                matching_tools.append(tool_def)
                continue
            
            # 描述匹配
            if query.lower() in tool_def.get("description", "").lower():
                matching_tools.append(tool_def)
                continue
            
            # 标签匹配
            tags = tool_def.get("tags", [])
            if any(query.lower() in tag.lower() for tag in tags):
                matching_tools.append(tool_def)
        
        return matching_tools
    
    def get_by_category(self, category):
        """
        按类别获取工具
        """
        tool_names = self.categories.get(category, [])
        return [self.tools[name] for name in tool_names]

3.3 工具动态发现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class DynamicToolDiscovery:
    def __init__(self):
        self.scanners = [
            APIEndpointScanner(),
            PythonModuleScanner(),
            MCPToolScanner()
        ]
    
    def discover_tools(self):
        """
        动态发现可用工具
        """
        discovered = []
        
        for scanner in self.scanners:
            tools = scanner.scan()
            discovered.extend(tools)
        
        return discovered

class APIEndpointScanner:
    def scan(self):
        """
        扫描API端点
        """
        tools = []
        
        # 扫描OpenAPI规范
        openapi_urls = self.find_openapi_docs()
        
        for url in openapi_urls:
            spec = self.fetch_openapi_spec(url)
            api_tools = self.parse_openapi_to_tools(spec)
            tools.extend(api_tools)
        
        return tools

四、工具选择与决策

4.1 基于规则的决策

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class RuleBasedToolSelector:
    def __init__(self):
        self.rules = [
            {
                "condition": lambda q: "天气" in q or "temperature" in q,
                "tool": "get_weather"
            },
            {
                "condition": lambda q: "搜索" in q or "search" in q,
                "tool": "web_search"
            },
            {
                "condition": lambda q: "计算" in q or "calculate" in q,
                "tool": "calculator"
            }
        ]
    
    def select_tool(self, query, available_tools):
        """
        选择工具
        """
        for rule in self.rules:
            if rule["condition"](query):
                tool_name = rule["tool"]
                
                if tool_name in available_tools:
                    return tool_name
        
        return None

4.2 基于语义相似度的决策

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class SemanticToolSelector:
    def __init__(self, embedding_model):
        self.embedding_model = embedding_model
        self.tool_embeddings = {}
    
    def index_tools(self, tools):
        """
        索引工具
        """
        for tool in tools:
            # 工具描述向量化
            description = f"{tool['name']}: {tool['description']}"
            embedding = self.embedding_model.embed(description)
            
            self.tool_embeddings[tool['name']] = embedding
    
    def select_tool(self, query, available_tools):
        """
        选择最相似的工具
        """
        # 查询向量化
        query_embedding = self.embedding_model.embed(query)
        
        # 计算相似度
        scores = {}
        for tool_name in available_tools:
            if tool_name in self.tool_embeddings:
                tool_embedding = self.tool_embeddings[tool_name]
                similarity = cosine_similarity(query_embedding, tool_embedding)
                scores[tool_name] = similarity
        
        # 选择最相似的
        if scores:
            best_tool = max(scores, key=scores.get)
            return best_tool
        
        return None

4.3 基于LLM的决策

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class LLMToolSelector:
    def __init__(self, llm_client):
        self.llm = llm_client
    
    def select_tool(self, query, available_tools):
        """
        使用LLM选择工具
        """
        # 构建提示词
        prompt = f"""
        用户请求: {query}
        
        可用工具:
        {self.format_tools(available_tools)}
        
        请选择最合适的工具来处理用户请求。
        
        输出格式:
        {{
          "tool": "工具名称",
          "reason": "选择理由"
        }}
        """
        
        # LLM决策
        response = self.llm.generate(prompt)
        
        # 解析结果
        decision = json.loads(response)
        
        return decision["tool"]

4.4 多工具选择

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class MultiToolSelector:
    def select_tools(self, query, available_tools, max_tools=3):
        """
        选择多个工具（可能需要组合）
        """
        # 分析任务需求
        requirements = self.analyze_requirements(query)
        
        # 为每个需求选择工具
        selected_tools = []
        
        for requirement in requirements:
            tool = self.select_tool_for_requirement(
                requirement,
                available_tools
            )
            
            if tool and tool not in selected_tools:
                selected_tools.append(tool)
        
        # 限制数量
        return selected_tools[:max_tools]
    
    def analyze_requirements(self, query):
        """
        分析任务需求
        """
        # 使用LLM分析
        prompt = f"""
        分析以下任务需要哪些能力:
        
        任务: {query}
        
        输出所需能力列表（如：数据检索、计算、文件操作等）
        """
        
        response = self.llm.generate(prompt)
        requirements = json.loads(response)
        
        return requirements

五、工具执行与管理

5.1 工具执行器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class ToolExecutor:
    def __init__(self):
        self.executors = {
            "api": APIExecutor(),
            "function": FunctionExecutor(),
            "script": ScriptExecutor()
        }
    
    def execute(self, tool_call):
        """
        执行工具调用
        """
        tool_name = tool_call["name"]
        arguments = tool_call["arguments"]
        
        # 获取工具定义
        tool_def = self.registry.get(tool_name)
        
        # 验证参数
        self.validate_arguments(arguments, tool_def)
        
        # 选择执行器
        executor_type = tool_def.get("executor_type", "function")
        executor = self.executors[executor_type]
        
        # 执行
        try:
            result = executor.execute(tool_def, arguments)
            
            return {
                "success": True,
                "result": result
            }
        
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

5.2 参数提取与绑定

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class ParameterBinder:
    def bind_parameters(self, tool_definition, context):
        """
        从上下文中提取并绑定参数
        """
        parameters = tool_definition["parameters"]
        bound_args = {}
        
        # 必需参数
        required = parameters.get("required", [])
        properties = parameters["properties"]
        
        for param_name in required:
            # 从上下文提取
            value = self.extract_from_context(param_name, context)
            
            if value is None:
                raise ValueError(f"缺少必需参数: {param_name}")
            
            # 类型转换和验证
            value = self.validate_and_convert(
                value,
                properties[param_name]
            )
            
            bound_args[param_name] = value
        
        # 可选参数
        for param_name, param_def in properties.items():
            if param_name not in required:
                value = self.extract_from_context(param_name, context)
                
                if value is not None:
                    value = self.validate_and_convert(value, param_def)
                    bound_args[param_name] = value
                elif "default" in param_def:
                    bound_args[param_name] = param_def["default"]
        
        return bound_args
    
    def extract_from_context(self, param_name, context):
        """
        从上下文中提取参数值
        """
        # 从对话历史
        if param_name in context.get("entities", {}):
            return context["entities"][param_name]
        
        # 从用户输入
        user_input = context.get("user_input", "")
        extracted = self.extract_with_llm(param_name, user_input)
        
        return extracted

5.3 执行监控

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class ExecutionMonitor:
    def __init__(self):
        self.metrics = defaultdict(list)
    
    def monitor_execution(self, tool_call, execution_result):
        """
        监控工具执行
        """
        tool_name = tool_call["name"]
        
        # 记录指标
        self.metrics[tool_name].append({
            "timestamp": datetime.now(),
            "success": execution_result["success"],
            "duration": execution_result.get("duration", 0),
            "error": execution_result.get("error")
        })
    
    def get_statistics(self, tool_name):
        """
        获取工具统计信息
        """
        history = self.metrics[tool_name]
        
        if not history:
            return None
        
        total = len(history)
        successful = sum(1 for h in history if h["success"])
        
        avg_duration = sum(h["duration"] for h in history) / total
        
        return {
            "total_calls": total,
            "success_rate": successful / total,
            "avg_duration": avg_duration,
            "recent_errors": [
                h for h in history[-10:]
                if not h["success"]
            ]
        }

六、工作流编排模式

6.1 顺序编排（Sequential）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class SequentialOrchestrator:
    def execute(self, tools_sequence, initial_input):
        """
        顺序执行工具链
        """
        result = initial_input
        
        for tool_call in tools_sequence:
            # 执行工具
            execution = self.executor.execute(tool_call)
            
            if not execution["success"]:
                return {
                    "success": False,
                    "error": f"工具 {tool_call['name']} 执行失败",
                    "failed_at": tool_call
                }
            
            # 传递给下一个工具
            result = execution["result"]
        
        return {
            "success": True,
            "result": result
        }

# 示例
workflow = [
    {"name": "fetch_data", "arguments": {"source": "api"}},
    {"name": "transform_data", "arguments": {}},
    {"name": "save_data", "arguments": {"destination": "database"}}
]

result = orchestrator.execute(workflow, initial_input=None)

6.2 并行编排（Parallel）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import asyncio

class ParallelOrchestrator:
    async def execute(self, tools_list, shared_input):
        """
        并行执行多个工具
        """
        tasks = []
        
        for tool_call in tools_list:
            task = self.executor.execute_async(tool_call, shared_input)
            tasks.append(task)
        
        # 等待所有任务完成
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 处理结果
        successful = []
        failed = []
        
        for tool_call, result in zip(tools_list, results):
            if isinstance(result, Exception):
                failed.append({
                    "tool": tool_call["name"],
                    "error": str(result)
                })
            else:
                successful.append({
                    "tool": tool_call["name"],
                    "result": result
                })
        
        return {
            "successful": successful,
            "failed": failed,
            "all_success": len(failed) == 0
        }

# 示例：并行获取多个API
parallel_tools = [
    {"name": "get_weather", "arguments": {"city": "北京"}},
    {"name": "get_weather", "arguments": {"city": "上海"}},
    {"name": "get_weather", "arguments": {"city": "广州"}}
]

result = await orchestrator.execute(parallel_tools, shared_input=None)

6.3 条件编排（Conditional）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class ConditionalOrchestrator:
    def execute(self, workflow_graph, input_data):
        """
        条件分支执行
        """
        current_node = workflow_graph["start"]
        results = {}
        
        while current_node and current_node != "end":
            node = workflow_graph["nodes"][current_node]
            
            if node["type"] == "tool":
                # 执行工具
                result = self.executor.execute(node["tool_call"])
                results[current_node] = result
                
                # 下一个节点
                current_node = node["next"]
            
            elif node["type"] == "condition":
                # 评估条件
                condition_result = self.evaluate_condition(
                    node["condition"],
                    results
                )
                
                # 选择分支
                current_node = node["branches"][str(condition_result)]
        
        return results

# 示例工作流
workflow = {
    "start": "check_stock",
    "nodes": {
        "check_stock": {
            "type": "tool",
            "tool_call": {"name": "check_inventory"},
            "next": "decide"
        },
        "decide": {
            "type": "condition",
            "condition": "results['check_stock']['in_stock']",
            "branches": {
                "True": "ship_order",
                "False": "notify_out_of_stock"
            }
        },
        "ship_order": {
            "type": "tool",
            "tool_call": {"name": "create_shipment"},
            "next": "end"
        },
        "notify_out_of_stock": {
            "type": "tool",
            "tool_call": {"name": "send_notification"},
            "next": "end"
        }
    }
}

6.4 循环编排（Loop）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class LoopOrchestrator:
    def execute(self, tool_call, termination_condition, max_iterations=100):
        """
        循环执行直到满足条件
        """
        iteration = 0
        results = []
        
        while iteration < max_iterations:
            # 执行工具
            result = self.executor.execute(tool_call)
            results.append(result)
            
            # 检查终止条件
            if termination_condition(result):
                break
            
            iteration += 1
        
        return {
            "iterations": iteration,
            "results": results,
            "completed": iteration < max_iterations
        }

# 示例：轮询直到任务完成
result = orchestrator.execute(
    tool_call={"name": "check_job_status", "arguments": {"job_id": "123"}},
    termination_condition=lambda r: r["status"] == "completed",
    max_iterations=60
)

6.5 DAG编排（Directed Acyclic Graph）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class DAGOrchestrator:
    def __init__(self):
        self.dag = nx.DiGraph()
    
    def add_tool(self, tool_id, tool_call, dependencies=None):
        """
        添加工具节点
        """
        self.dag.add_node(tool_id, tool_call=tool_call)
        
        if dependencies:
            for dep in dependencies:
                self.dag.add_edge(dep, tool_id)
    
    def execute(self, input_data):
        """
        按DAG执行
        """
        # 拓扑排序
        execution_order = list(nx.topological_sort(self.dag))
        
        results = {}
        
        for tool_id in execution_order:
            node_data = self.dag.nodes[tool_id]
            tool_call = node_data["tool_call"]
            
            # 收集依赖的结果
            dependencies_results = {
                dep: results[dep]
                for dep in self.dag.predecessors(tool_id)
            }
            
            # 执行工具
            result = self.executor.execute(tool_call, dependencies_results)
            results[tool_id] = result
        
        return results

# 示例DAG
orchestrator = DAGOrchestrator()
orchestrator.add_tool("fetch", {"name": "fetch_data"})
orchestrator.add_tool("analyze1", {"name": "analyze"}, dependencies=["fetch"])
orchestrator.add_tool("analyze2", {"name": "visualize"}, dependencies=["fetch"])
orchestrator.add_tool("report", {"name": "generate_report"}, dependencies=["analyze1", "analyze2"])

result = orchestrator.execute(input_data=None)

七、主流工具调用框架

7.1 OpenAI Function Calling

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
from openai import OpenAI

client = OpenAI()

# 定义工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取城市天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

# 调用
response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "user", "content": "北京今天天气怎么样？"}
    ],
    tools=tools,
    tool_choice="auto"
)

# 处理工具调用
message = response.choices[0].message

if message.tool_calls:
    for tool_call in message.tool_calls:
        function_name = tool_call.function.name
        arguments = json.loads(tool_call.function.arguments)
        
        # 执行工具
        result = execute_tool(function_name, arguments)
        
        # 返回结果给模型
        messages.append(message)
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(result)
        })

7.2 LangChain Tools

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from langchain.tools import Tool
from langchain.agents import initialize_agent

# 定义工具
def get_weather(city: str) -> str:
    """获取天气信息"""
    # 实现天气查询
    return f"{city}今天晴，20度"

tools = [
    Tool(
        name="Weather",
        func=get_weather,
        description="获取城市天气信息，输入城市名称"
    )
]

# 创建Agent
agent = initialize_agent(
    tools,
    llm,
    agent="zero-shot-react-description",
    verbose=True
)

# 执行
result = agent.run("北京今天天气怎么样？")

7.3 Anthropic Tool Use

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import anthropic

client = anthropic.Anthropic()

# 定义工具
tools = [
    {
        "name": "get_weather",
        "description": "获取城市天气",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称"
                }
            },
            "required": ["city"]
        }
    }
]

# 调用
response = client.messages.create(
    model="claude-3-opus-20240229",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "北京天气怎么样？"}
    ],
    tools=tools
)

# 处理响应
for block in response.content:
    if block.type == "tool_use":
        tool_name = block.name
        tool_input = block.input
        
        # 执行工具
        result = execute_tool(tool_name, tool_input)

7.4 框架对比

特性 OpenAI LangChain Anthropic
工具定义 JSON Schema Python函数 JSON Schema
自动填充 ✅ ✅ ✅
多工具 ✅ ✅ ✅
并行调用 ✅ ✅ ✅
流式响应 ✅ ✅ ✅
易用性 高 中 高

八、MCP协议详解

8.1 MCP简介

MCP (Model Context Protocol) 是Anthropic推出的标准化工具协议

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# MCP工具定义
{
  "name": "mcp_weather",
  "description": "获取天气信息",
  "inputSchema": {
    "type": "object",
    "properties": {
      "city": {"type": "string"}
    },
    "required": ["city"]
  },
  "transport": {
    "type": "stdio",
    "command": "python",
    "args": ["weather_server.py"]
  }
}

8.2 MCP服务器实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from mcp import Server, Tool

# 创建MCP服务器
server = Server("weather-service")

# 定义工具
@server.tool("get_weather")
async def get_weather(city: str) -> dict:
    """
    获取天气信息
    
    Args:
        city: 城市名称
    
    Returns:
        天气信息字典
    """
    # 调用天气API
    weather_data = await fetch_weather(city)
    
    return {
        "temperature": weather_data["temp"],
        "condition": weather_data["condition"],
        "humidity": weather_data["humidity"]
    }

# 运行服务器
server.run()

8.3 MCP客户端使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from mcp import Client

# 创建客户端
client = Client()

# 连接MCP服务器
client.connect("weather-service")

# 列出工具
tools = client.list_tools()

# 调用工具
result = await client.call_tool("get_weather", {"city": "北京"})

print(result)
# {"temperature": 20, "condition": "晴", "humidity": 45}

8.4 MCP优势

  1. 标准化: 统一的工具接口
  2. 可组合: 工具可以组合使用
  3. 跨平台: 支持多种传输方式
  4. 生态系统: 丰富的预构建工具

九、工具编排最佳实践

9.1 工具设计原则

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# 原则1: 单一职责
class GoodTool:
    def get_user_profile(user_id):
        """只做一件事：获取用户档案"""
        pass

class BadTool:
    def get_user_and_orders_and_recommendations(user_id):
        """做太多事情"""
        pass

# 原则2: 清晰描述
good_description = """
获取指定城市的当前天气信息

参数:
- city: 城市名称（中文或英文）
- unit: 温度单位（celsius或fahrenheit）

返回:
- temperature: 当前温度
- condition: 天气状况
- humidity: 湿度百分比
"""

bad_description = "获取天气"

# 原则3: 容错设计
def robust_tool(param):
    try:
        result = api_call(param)
        return {"success": True, "data": result}
    except APIError as e:
        return {"success": False, "error": str(e)}

9.2 参数验证

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from pydantic import BaseModel, validator

class WeatherQuery(BaseModel):
    city: str
    unit: str = "celsius"
    
    @validator("city")
    def validate_city(cls, v):
        if not v or len(v) < 2:
            raise ValueError("城市名称至少2个字符")
        return v
    
    @validator("unit")
    def validate_unit(cls, v):
        if v not in ["celsius", "fahrenheit"]:
            raise ValueError("单位必须是celsius或fahrenheit")
        return v

# 使用
def get_weather(query: WeatherQuery):
    validated = WeatherQuery(**query)
    # 安全使用

9.3 结果格式化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class ToolResult:
    def __init__(self, success, data=None, error=None, metadata=None):
        self.success = success
        self.data = data
        self.error = error
        self.metadata = metadata or {}
    
    def to_dict(self):
        return {
            "success": self.success,
            "data": self.data,
            "error": self.error,
            "metadata": self.metadata
        }
    
    def to_string(self):
        if self.success:
            return json.dumps(self.data, ensure_ascii=False)
        else:
            return f"错误: {self.error}"

十、错误处理与容错

10.1 错误类型

1
2
3
4
5
6
7
class ToolErrorType(Enum):
    VALIDATION_ERROR = "validation_error"      # 参数验证失败
    EXECUTION_ERROR = "execution_error"        # 执行错误
    TIMEOUT_ERROR = "timeout_error"            # 超时
    RATE_LIMIT_ERROR = "rate_limit_error"      # 速率限制
    NETWORK_ERROR = "network_error"            # 网络错误
    PERMISSION_ERROR = "permission_error"      # 权限不足

10.2 重试策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import asyncio
from functools import wraps

def retry(max_retries=3, backoff_factor=2):
    """
    重试装饰器
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                
                except (NetworkError, TimeoutError) as e:
                    if attempt == max_retries - 1:
                        raise
                    
                    # 指数退避
                    wait_time = backoff_factor ** attempt
                    await asyncio.sleep(wait_time)
                
                except Exception as e:
                    # 不可重试的错误
                    raise
            
            return None
        
        return wrapper
    
    return decorator

# 使用
@retry(max_retries=3)
async def call_external_api(params):
    # API调用
    pass

10.3 熔断器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open
    
    def call(self, func, *args, **kwargs):
        """
        通过熔断器调用
        """
        if self.state == "open":
            # 检查是否可以尝试恢复
            if self.should_attempt_reset():
                self.state = "half-open"
            else:
                raise CircuitBreakerOpen("熔断器打开")
        
        try:
            result = func(*args, **kwargs)
            self.on_success()
            return result
        
        except Exception as e:
            self.on_failure()
            raise
    
    def on_success(self):
        """
        成功时重置
        """
        self.failure_count = 0
        self.state = "closed"
    
    def on_failure(self):
        """
        失败时计数
        """
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        
        if self.failure_count >= self.failure_threshold:
            self.state = "open"
    
    def should_attempt_reset(self):
        """
        是否应该尝试恢复
        """
        if self.last_failure_time is None:
            return False
        
        elapsed = (datetime.now() - self.last_failure_time).seconds
        return elapsed >= self.timeout

10.4 降级策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class FallbackStrategy:
    def __init__(self):
        self.fallback_handlers = {}
    
    def register_fallback(self, tool_name, fallback_func):
        """
        注册降级处理
        """
        self.fallback_handlers[tool_name] = fallback_func
    
    def execute_with_fallback(self, tool_call):
        """
        带降级的执行
        """
        tool_name = tool_call["name"]
        
        try:
            # 尝试正常执行
            return self.executor.execute(tool_call)
        
        except Exception as e:
            # 检查是否有降级方案
            if tool_name in self.fallback_handlers:
                fallback = self.fallback_handlers[tool_name]
                return fallback(tool_call, e)
            
            # 没有降级方案
            raise

# 示例
def weather_fallback(tool_call, error):
    """天气API降级：返回缓存或默认值"""
    return {
        "temperature": "未知",
        "condition": "数据暂时不可用",
        "error": str(error)
    }

十一、评估与优化

11.1 工具调用评估指标

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class ToolCallMetrics:
    def evaluate(self, tool_calls_history):
        """
        评估工具调用
        """
        # 成功率
        success_rate = sum(1 for tc in tool_calls_history if tc["success"]) / len(tool_calls_history)
        
        # 平均延迟
        avg_latency = sum(tc["duration"] for tc in tool_calls_history) / len(tool_calls_history)
        
        # 工具选择准确率
        selection_accuracy = self.evaluate_selection_accuracy(tool_calls_history)
        
        # 参数填充准确率
        parameter_accuracy = self.evaluate_parameter_accuracy(tool_calls_history)
        
        return {
            "success_rate": success_rate,
            "avg_latency": avg_latency,
            "selection_accuracy": selection_accuracy,
            "parameter_accuracy": parameter_accuracy
        }

11.2 性能优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class ToolCallOptimizer:
    def optimize(self, tool_calls):
        """
        优化工具调用
        """
        # 1. 批量处理
        batched = self.batch_similar_calls(tool_calls)
        
        # 2. 并行化
        parallelized = self.identify_parallelizable(batched)
        
        # 3. 缓存
        cached = self.apply_caching(parallelized)
        
        # 4. 去重
        deduped = self.deduplicate(cached)
        
        return deduped
    
    def batch_similar_calls(self, tool_calls):
        """
        批量相似调用
        """
        # 按工具名分组
        grouped = defaultdict(list)
        
        for tc in tool_calls:
            grouped[tc["name"]].append(tc)
        
        # 批量执行
        batched = []
        for tool_name, calls in grouped.items():
            if self.supports_batch(tool_name):
                batch_result = self.execute_batch(tool_name, calls)
                batched.append(batch_result)
            else:
                batched.extend(calls)
        
        return batched

十二、工程实践案例

12.1 案例1：数据分析工作流

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class DataAnalysisWorkflow:
    def __init__(self):
        self.tools = {
            "fetch_data": FetchDataTool(),
            "clean_data": CleanDataTool(),
            "analyze": AnalyzeTool(),
            "visualize": VisualizeTool(),
            "report": ReportTool()
        }
    
    def analyze_dataset(self, data_source):
        """
        分析数据集
        """
        # DAG工作流
        workflow = [
            {
                "id": "fetch",
                "tool": "fetch_data",
                "args": {"source": data_source}
            },
            {
                "id": "clean",
                "tool": "clean_data",
                "args": {},
                "depends_on": ["fetch"]
            },
            {
                "id": "analyze_stats",
                "tool": "analyze",
                "args": {"type": "statistical"},
                "depends_on": ["clean"]
            },
            {
                "id": "analyze_trends",
                "tool": "analyze",
                "args": {"type": "trend"},
                "depends_on": ["clean"]
            },
            {
                "id": "visualize",
                "tool": "visualize",
                "args": {},
                "depends_on": ["analyze_stats", "analyze_trends"]
            },
            {
                "id": "report",
                "tool": "report",
                "args": {},
                "depends_on": ["visualize"]
            }
        ]
        
        # 执行工作流
        orchestrator = DAGOrchestrator()
        
        for step in workflow:
            orchestrator.add_tool(
                step["id"],
                {"name": step["tool"], "arguments": step["args"]},
                step.get("depends_on", [])
            )
        
        results = orchestrator.execute({})
        
        return results["report"]

12.2 案例2：客服机器人工具链

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class CustomerServiceTools:
    def __init__(self):
        self.registry = ToolRegistry()
        
        # 注册工具
        self.registry.register({
            "name": "query_order",
            "description": "查询订单状态",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"}
                },
                "required": ["order_id"]
            }
        })
        
        self.registry.register({
            "name": "create_ticket",
            "description": "创建工单",
            "parameters": {
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"},
                    "issue": {"type": "string"},
                    "priority": {"type": "string", "enum": ["low", "medium", "high"]}
                },
                "required": ["customer_id", "issue"]
            }
        })
        
        self.registry.register({
            "name": "check_inventory",
            "description": "检查库存",
            "parameters": {
                "type": "object",
                "properties": {
                    "product_id": {"type": "string"}
                },
                "required": ["product_id"]
            }
        })
    
    def handle_inquiry(self, user_message):
        """
        处理客户咨询
        """
        # LLM选择工具
        selected_tools = self.tool_selector.select_tools(
            user_message,
            self.registry.tools.keys()
        )
        
        # 执行工具
        results = []
        for tool_name in selected_tools:
            tool_def = self.registry.tools[tool_name]
            
            # 参数填充
            args = self.parameter_binder.bind_parameters(
                tool_def,
                {"user_input": user_message}
            )
            
            # 执行
            result = self.executor.execute({
                "name": tool_name,
                "arguments": args
            })
            
            results.append(result)
        
        # 生成响应
        response = self.generate_response(user_message, results)
        
        return response

12.3 案例3：代码助手工具编排

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class CodeAssistantTools:
    def __init__(self):
        self.tools = {
            "read_file": ReadFileTool(),
            "write_file": WriteFileTool(),
            "execute_code": ExecuteCodeTool(),
            "search_web": SearchWebTool(),
            "install_package": InstallPackageTool()
        }
    
    def implement_feature(self, feature_description):
        """
        实现功能
        """
        # 步骤1: 搜索相关资料
        search_results = self.tools["search_web"].execute({
            "query": feature_description
        })
        
        # 步骤2: 读取现有代码
        existing_code = self.tools["read_file"].execute({
            "path": "main.py"
        })
        
        # 步骤3: 生成代码
        new_code = self.generate_code(
            feature_description,
            search_results,
            existing_code
        )
        
        # 步骤4: 写入文件
        self.tools["write_file"].execute({
            "path": "main.py",
            "content": new_code
        })
        
        # 步骤5: 执行测试
        test_result = self.tools["execute_code"].execute({
            "code": "python -m pytest tests/"
        })
        
        return {
            "code": new_code,
            "test_result": test_result
        }

十三、关键挑战与解决方案

13.1 挑战1：工具选择错误

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class ToolSelectionValidator:
    def validate_selection(self, query, selected_tool, available_tools):
        """
        验证工具选择
        """
        # 计算相关性分数
        relevance_score = self.calculate_relevance(query, selected_tool)
        
        if relevance_score < 0.5:
            # 可能选择错误，建议替代
            alternatives = self.suggest_alternatives(query, available_tools)
            
            return {
                "valid": False,
                "reason": "相关性分数低",
                "alternatives": alternatives
            }
        
        return {"valid": True}

13.2 挑战2：参数缺失

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class MissingParameterHandler:
    def handle_missing_param(self, param_name, tool_def, context):
        """
        处理缺失参数
        """
        # 策略1: 从上下文推断
        inferred = self.infer_from_context(param_name, context)
        
        if inferred:
            return inferred
        
        # 策略2: 使用默认值
        default = tool_def["parameters"]["properties"][param_name].get("default")
        
        if default is not None:
            return default
        
        # 策略3: 询问用户
        return self.ask_user(param_name)

13.3 挑战3：工具执行超时

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class TimeoutHandler:
    def execute_with_timeout(self, tool_call, timeout_seconds):
        """
        带超时的执行
        """
        try:
            result = asyncio.wait_for(
                self.executor.execute_async(tool_call),
                timeout=timeout_seconds
            )
            return result
        
        except asyncio.TimeoutError:
            # 超时处理
            return {
                "success": False,
                "error": "执行超时",
                "timeout": timeout_seconds
            }

13.4 挑战4：工具依赖冲突

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class DependencyResolver:
    def resolve_conflicts(self, tool_calls):
        """
        解决依赖冲突
        """
        # 检测循环依赖
        graph = self.build_dependency_graph(tool_calls)
        
        if self.has_cycle(graph):
            # 打破循环
            graph = self.break_cycle(graph)
        
        # 排序
        execution_order = self.topological_sort(graph)
        
        return execution_order

十四、未来趋势与研究方向

14.1 趋势1：自主工具发现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
class AutonomousToolDiscovery:
    def discover_and_use(self, task):
        """
        自主发现并使用工具
        """
        # 分析任务需求
        requirements = self.analyze_requirements(task)
        
        # 搜索可用工具
        candidate_tools = self.search_tool_sources(requirements)
        
        # 测试工具
        tested_tools = self.test_tools(candidate_tools)
        
        # 选择最佳工具
        best_tool = self.select_best(tested_tools)
        
        # 使用工具
        result = self.use_tool(best_tool, task)
        
        return result

14.2 趋势2：工具学习与优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class ToolLearning:
    def learn_from_usage(self, usage_history):
        """
        从使用历史学习
        """
        # 分析成功案例
        successful = [h for h in usage_history if h["success"]]
        
        # 提取模式
        patterns = self.extract_patterns(successful)
        
        # 优化工具选择策略
        self.update_selection_policy(patterns)
        
        # 优化参数填充
        self.update_parameter_binding(patterns)

14.3 趋势3：工具组合生成

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class ToolCompositionGenerator:
    def generate_composite_tool(self, task_requirements):
        """
        生成组合工具
        """
        # 识别需要的原子工具
        atomic_tools = self.identify_atomic_tools(task_requirements)
        
        # 设计组合方式
        composition = self.design_composition(atomic_tools)
        
        # 生成代码
        composite_code = self.generate_code(composition)
        
        # 测试
        tested = self.test_composite(composite_code)
        
        return tested

十五、总结与行动指南

15.1 核心要点回顾

维度 关键点 建议
工具描述 JSON Schema/OpenAPI 清晰准确
工具选择 规则/语义/LLM 混合策略
工具执行 参数填充/验证/监控 容错设计
编排模式 顺序/并行/条件/DAG 按需选择
错误处理 重试/熔断/降级 多层防护
框架选择 OpenAI/LangChain/MCP 根据场景

15.2 不同场景的推荐方案

场景1：简单API集成

1
2
3
4
推荐方案: OpenAI Function Calling
- 实现: 简单
- 性能: 好
- 成本: 低

场景2：复杂工作流

1
2
3
4
推荐方案: LangChain + DAG编排
- 实现: 中等
- 性能: 好
- 成本: 中

场景3：企业级工具生态

1
2
3
4
推荐方案: MCP协议 + 自建编排
- 实现: 复杂
- 性能: 很好
- 成本: 高

15.3 实施路线图

阶段1：基础工具调用（1-2周）

1
2
3
1. 定义工具（Function Calling）
2. 实现参数填充
3. 基本错误处理

阶段2：工具编排（2-4周）

1
2
3
1. 实现顺序/并行执行
2. 添加条件分支
3. 优化性能

阶段3：智能编排（1-2月）

1
2
3
1. 动态工具发现
2. 自主工具选择
3. 持续学习优化

15.4 避坑指南

  1. ❌ 工具描述不清：导致选择错误
  2. ❌ 缺少验证：参数错误
  3. ❌ 无超时控制：长时间等待
  4. ❌ 无降级方案：单点故障
  5. ❌ 忽视监控：问题难以定位

15.5 工具推荐

工具 用途 链接
OpenAI Function Calling 官网
LangChain 工具编排 GitHub
MCP 标准化协议 GitHub
FastAPI API开发 官网

15.6 最终建议

  1. 从简单开始：先实现单工具调用
  2. 重视描述：工具描述要准确
  3. 多层容错：重试+熔断+降级
  4. 监控一切：建立完整的可观测性
  5. 持续优化：从使用中学习
  6. 标准化：考虑使用MCP协议

参考资料

核心论文

  1. 工具调用

  2. 工具编排

文档

  1. OpenAI Function Calling: https://platform.openai.com/docs/guides/function-calling
  2. Anthropic Tool Use: https://docs.anthropic.com/claude/docs/tool-use
  3. LangChain Tools: https://python.langchain.com/docs/modules/tools/

开源项目

  1. LangChain: https://github.com/langchain-ai/langchain
  2. MCP: https://github.com/anthropics/mcp
  3. AutoGPT: https://github.com/Significant-Gravitas/AutoGPT

作者: 来顺 (AI Assistant)
生成时间: 2026-03-31
阅读时长: ~50分钟
适用读者: AI工程师、系统架构师、工具开发者

💡 核心观点: 工具调用是Agent连接外部世界的关键能力，从单一工具到复杂工作流，需要清晰的描述、智能的选择、可靠的执行和灵活的编排。在实践中不断优化，建立健壮的工具生态系统。

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

程序员，关于技术我所知甚少，唯有探索，方得真知。