第34章：工具使用模式

概述

工具（Tool）是 Agent 与外部世界交互的桥梁。Agent 的能力上限，很大程度上取决于它能使用多少工具、如何选择工具、以及如何组合工具。本章将以设计模式（Design Pattern）的视角，系统讲解 Agent 工具使用的各种模式——从最基础的单工具调用，到复杂的工具链编排、动态选择策略、安全沙箱执行和性能优化。掌握这些模式，将帮助你构建出既强大又可靠的 Agent 工具系统。

34.1 工具抽象的本质

34.1.1 从函数到工具

在 Agent 编程中，"工具"不仅仅是函数——它是一个带有丰富语义描述的结构化抽象。工具不仅要能执行，更要能被 LLM 理解和选择。

from pydantic import BaseModel, Field
from typing import Any, Callable

# 普通函数——Agent无法理解
def search(query: str) -> str:
    return api.search(query)

# 工具——Agent可以理解并选择
class SearchTool:
    name = "web_search"
    description = "在互联网上搜索最新信息，适用于需要查找实时数据、新闻、产品信息等场景"
    
    class Args(BaseModel):
        query: str = Field(..., description="搜索关键词，应简洁明确")
        num_results: int = Field(default=5, description="返回结果数量，1-20")
        language: str = Field(default="zh", description="搜索语言：zh/en")
    
    async def execute(self, query: str, num_results: int = 5,
                      language: str = "zh") -> str:
        results = await search_api(query, num_results, language)
        return self._format_results(results)

34.1.2 Function Calling 标准接口

class StandardTool:
    """标准化的工具接口"""
    
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description
        self._parameters: list[dict] = []
    
    def param(self, name: str, type_: str, description: str,
              required: bool = True, default: Any = None,
              enum: list[str] | None = None) -> 'StandardTool':
        """链式定义参数"""
        param_def = {
            "type": type_,
            "description": description,
        }
        if not required:
            param_def["default"] = default
        if enum:
            param_def["enum"] = enum
        self._parameters.append({"name": name, **param_def})
        return self
    
    def to_openai_schema(self) -> dict:
        """转换为OpenAI Function Calling格式"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": {
                        p["name"]: {
                            "type": p["type"],
                            "description": p["description"],
                            **({"enum": p["enum"]} if p.get("enum") else {})
                        }
                        for p in self._parameters
                    },
                    "required": [
                        p["name"] for p in self._parameters 
                        if "default" not in p
                    ]
                }
            }
        }

# 使用示例
search = (StandardTool("search", "搜索互联网获取最新信息")
    .param("query", "string", "搜索关键词")
    .param("num_results", "integer", "结果数量", required=False, default=5)
    .param("region", "string", "搜索区域", required=False, 
           enum=["cn", "us", "eu", "global"]))

34.2 基础工具模式

34.2.1 单一工具调用

最简单的模式：Agent 识别用户意图，选择并调用单个工具。

class SingleToolAgent:
    """单工具Agent"""
    
    def __init__(self, tool, llm):
        self.tool = tool
        self.llm = llm
    
    async def run(self, user_input: str) -> str:
        # 1. 判断是否需要工具
        decision = await self._should_use_tool(user_input)
        
        if not decision["use_tool"]:
            return await self.llm.generate(user_input)
        
        # 2. 提取工具参数
        args = await self._extract_args(user_input, self.tool)
        
        # 3. 执行工具
        result = await self.tool.execute(**args)
        
        # 4. 格式化输出
        return await self._format_response(user_input, result)

34.2.2 参数提取模式

class ParameterExtractor:
    """工具参数提取器"""
    
    async def extract(self, user_input: str, tool: StandardTool,
                      llm) -> dict:
        """从自然语言中提取工具参数"""
        prompt = f"""
        用户输入: {user_input}
        工具: {tool.name}
        描述: {tool.description}
        参数: {json.dumps(tool._parameters, ensure_ascii=False)}
        
        请从用户输入中提取工具参数，以JSON格式返回。
        如果某个参数无法从输入中提取，使用默认值或null。
        """
        response = await llm.generate(prompt)
        
        try:
            params = json.loads(response)
            return self._validate_params(params, tool)
        except json.JSONDecodeError:
            return self._fallback_extraction(user_input, tool)

34.3 工具链模式

34.3.1 Sequential Chain（顺序链）

class ToolChain:
    """工具顺序链"""
    
    def __init__(self):
        self._steps: list[tuple[StandardTool, str]] = []
    
    def add_step(self, tool: StandardTool, 
                 input_mapping: str = "$prev_result") -> 'ToolChain':
        self._steps.append((tool, input_mapping))
        return self
    
    async def execute(self, initial_input: dict) -> str:
        current_result = initial_input
        
        for i, (tool, mapping) in enumerate(self._steps):
            step_input = self._resolve_mapping(mapping, current_result)
            current_result = await tool.execute(**step_input)
        
        return current_result

34.3.2 DAG执行模式

import networkx as nx

class DAGExecutor:
    """有向无环图工具执行器"""
    
    def __init__(self):
        self.graph = nx.DiGraph()
        self._tools: dict[str, StandardTool] = {}
    
    def add_tool(self, tool: StandardTool, 
                 dependencies: list[str] | None = None):
        self._tools[tool.name] = tool
        self.graph.add_node(tool.name)
        if dependencies:
            for dep in dependencies:
                self.graph.add_edge(dep, tool.name)
    
    async def execute(self, initial_inputs: dict = None) -> dict:
        if not nx.is_directed_acyclic_graph(self.graph):
            raise ValueError("工具依赖图包含循环！")
        
        results = initial_inputs or {}
        execution_order = list(nx.topological_sort(self.graph))
        
        for tool_name in execution_order:
            tool = self._tools[tool_name]
            args = {}
            for pred in self.graph.predecessors(tool_name):
                args[pred] = results.get(pred, "")
            results[tool_name] = await tool.execute(**args)
        
        return results

34.3.3 条件分支

class ConditionalToolChain:
    """条件分支工具链"""
    
    def when(self, condition: str, tools: list[StandardTool]):
        """定义条件分支"""
        self._conditions.append({"condition": condition, "tools": tools})
        return self
    
    async def execute(self, input_data: dict) -> str:
        for cond in self._conditions:
            if self._evaluate(cond["condition"], input_data):
                for tool in cond["tools"]:
                    result = await tool.execute(**input_data)
                    input_data["last_result"] = result
                return result
        return "没有匹配的条件分支"

34.4 工具选择策略

34.4.1 基于语义的自动选择

class SemanticToolSelector:
    """语义工具选择器"""
    
    def __init__(self, tools: list[StandardTool], embedder):
        self.tools = tools
        self.embedder = embedder
        self._tool_embeddings: dict[str, list[float]] = {}
    
    async def build_index(self):
        for tool in self.tools:
            index_text = f"{tool.name}: {tool.description}"
            self._tool_embeddings[tool.name] = \
                await self.embedder.embed(index_text)
    
    async def select(self, query: str, top_k: int = 3) -> list[StandardTool]:
        query_embedding = await self.embedder.embed(query)
        
        scored_tools = []
        for tool in self.tools:
            similarity = cosine_similarity(
                [query_embedding], [self._tool_embeddings[tool.name]]
            )[0][0]
            scored_tools.append((tool, similarity))
        
        scored_tools.sort(key=lambda x: x[1], reverse=True)
        return [t for t, s in scored_tools[:top_k]]

34.4.2 ReAct式动态选择

class ReActToolSelector:
    """ReAct模式下的动态工具选择"""
    
    async def think_and_act(self, task: str, tools: list[StandardTool],
                            llm, max_steps: int = 10) -> str:
        tools_desc = "\n".join(
            f"- {t.name}: {t.description}" for t in tools
        )
        
        context = f"任务: {task}\n可用工具:\n{tools_desc}"
        history = []
        
        for step in range(max_steps):
            thought_prompt = f"""
            {context}
            历史步骤: {self._format_history(history)}
            
            请思考下一步。格式：
            思考: [推理过程]
            行动: [工具名(参数)] 或 "最终答案: [回答]"
            """
            response = await llm.generate(thought_prompt)
            action = self._parse_action(response)
            
            if action["type"] == "final_answer":
                return action["content"]
            
            tool = next(t for t in tools if t.name == action["tool_name"])
            result = await tool.execute(**action["args"])
            history.append({"action": f"{action['tool_name']}", "observation": str(result)[:500]})
            context += f"\n\n观察: {str(result)[:500]}"
        
        return "未能在限定步骤内完成任务。"

34.5 工具组合与编排

34.5.1 Composite Tools（组合工具）

class CompositeTool(StandardTool):
    """组合工具：将多个工具封装为一个高级工具"""
    
    def __init__(self, name: str, description: str):
        super().__init__(name, description)
        self._sub_tools: list[StandardTool] = []
    
    def add_sub_tool(self, tool: StandardTool):
        self._sub_tools.append(tool)
        return self
    
    async def execute(self, **kwargs) -> str:
        intermediate = kwargs
        for tool in self._sub_tools:
            result = await tool.execute(**intermediate)
            intermediate["prev_result"] = result
            intermediate["input"] = str(result)
        return intermediate.get("prev_result", "")

# 智能搜索 = 搜索 + 摘要
smart_search = CompositeTool("smart_search", "智能搜索：搜索并摘要")
smart_search.add_sub_tool(StandardTool("search", "搜索").param("query", "string", "查询"))
smart_search.add_sub_tool(StandardTool("summarize", "摘要").param("text", "string", "文本"))

34.6 工具错误处理

34.6.1 超时与重试

class ResilientTool:
    """带弹性策略的工具包装器"""
    
    def __init__(self, tool: StandardTool, 
                 max_retries: int = 3, timeout: float = 30.0):
        self.tool = tool
        self.max_retries = max_retries
        self.timeout = timeout
    
    async def execute(self, **kwargs) -> dict:
        try:
            result = await asyncio.wait_for(
                self._retry_execute(**kwargs),
                timeout=self.timeout
            )
            return {"success": True, "data": result}
        except asyncio.TimeoutError:
            return {"success": False, "error": f"工具 {self.tool.name} 超时"}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    async def _retry_execute(self, **kwargs):
        for attempt in range(self.max_retries):
            try:
                return await self.tool.execute(**kwargs)
            except Exception:
                if attempt == self.max_retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)

34.6.2 Graceful Degradation（优雅降级）

class DegradableTool:
    """可降级的工具"""
    
    def __init__(self, primary: StandardTool,
                 fallback: StandardTool = None,
                 fallback_response: str = None):
        self.primary = primary
        self.fallback = fallback
        self.fallback_response = fallback_response
    
    async def execute(self, **kwargs) -> Any:
        try:
            return await self.primary.execute(**kwargs)
        except Exception as e:
            if self.fallback:
                return await self.fallback.execute(**kwargs)
            if self.fallback_response:
                return self.fallback_response
            raise

34.7 自定义工具开发

34.7.1 工具开发规范

工具命名：动词_名词（如 search_web、calculate_expression）

描述模板：

• 一句话功能描述
• 适用场景（2-3个）
• 不适用场景
• 输入要求和输出格式

参数设计规则：

• 必需参数不超过3个
• 使用具体类型（date 而非 string）
• 有限选项用 enum 约束
• 可选参数提供合理默认值

34.7.2 完整工具示例

class FileAnalysisTool(StandardTool):
    """文件分析工具"""
    
    def __init__(self):
        super().__init__("analyze_file", "分析上传的文件内容")
        self.param("file_path", "string", "文件路径或URL")
        self.param("analysis_type", "string", "分析类型",
                  enum=["summary", "statistics", "schema", "quality"])
        self.param("max_rows", "integer", "最大分析行数",
                  required=False, default=1000)
    
    async def execute(self, file_path: str, analysis_type: str,
                      max_rows: int = 1000) -> dict:
        data = await self._load_file(file_path)
        analyzers = {
            "summary": self._summarize,
            "statistics": self._statistics,
            "schema": self._infer_schema,
            "quality": self._check_quality,
        }
        return analyzers[analysis_type](data, max_rows)

34.8 工具安全模式

34.8.1 输入验证与沙箱

class SecureToolWrapper:
    """安全工具包装器"""
    
    def __init__(self, tool: StandardTool):
        self.tool = tool
        self._validators = []
    
    def add_validator(self, validator):
        self._validators.append(validator)
        return self
    
    async def execute(self, **kwargs) -> Any:
        for validator in self._validators:
            is_valid, error = validator(kwargs)
            if not is_valid:
                raise ValueError(f"输入验证失败: {error}")
        return await self.tool.execute(**kwargs)

# SQL注入检测
def no_sql_injection(params: dict) -> tuple[bool, str]:
    dangerous = ["DROP", "DELETE", "INSERT", "UPDATE", "UNION", "--", ";"]
    for key, value in params.items():
        if isinstance(value, str):
            if any(kw in value.upper() for kw in dangerous):
                return False, f"参数 {key} 包含可疑SQL关键字"
    return True, ""

34.9 工具性能模式

34.9.1 异步并发

class AsyncToolExecutor:
    """异步工具执行器"""
    
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def execute_many(self, calls: list[tuple[StandardTool, dict]]) -> list[dict]:
        tasks = [self._execute_one(tool, args) for tool, args in calls]
        return await asyncio.gather(*tasks, return_exceptions=True)
    
    async def _execute_one(self, tool: StandardTool, args: dict) -> dict:
        async with self.semaphore:
            start = time.time()
            try:
                result = await tool.execute(**args)
                return {"tool": tool.name, "success": True, "result": result,
                        "duration_ms": (time.time() - start) * 1000}
            except Exception as e:
                return {"tool": tool.name, "success": False, "error": str(e)}

34.9.2 结果缓存

class CachedTool:
    """带缓存的工具"""
    
    def __init__(self, tool: StandardTool, cache_ttl: int = 300):
        self.tool = tool
        self._cache: dict[str, tuple[float, Any]] = {}
        self.ttl = cache_ttl
    
    async def execute(self, **kwargs) -> Any:
        key = hashlib.md5(json.dumps(kwargs, sort_keys=True).encode()).hexdigest()
        if key in self._cache:
            ts, val = self._cache[key]
            if time.time() - ts < self.ttl:
                return val
        
        result = await self.tool.execute(**kwargs)
        self._cache[key] = (time.time(), result)
        return result

最佳实践

1. 工具描述即API文档：description 是 LLM 选择工具的唯一依据，必须准确详尽
2. 参数约束要明确：用 enum 限制选项，用 type 指定格式
3. 错误信息要友好：返回给 Agent 的错误信息应包含修复建议
4. 工具粒度适中：5-15个工具为最佳范围
5. 监控工具使用：记录调用频率、成功率、延迟，持续优化

常见陷阱

1. 描述模糊："处理数据" → 应该写 "查询SQL数据库并返回JSON结果"
2. 参数过多：超过5个必需参数的工具调用失败率极高
3. 忽略幂等性：非幂等工具重试会产生副作用
4. 缺少降级：主工具失败时没有备选方案
5. 无限流和缓存：高频调用导致成本爆炸

小结

工具是 Agent 能力的核心载体。本章从设计模式的视角系统讲解了工具抽象、调用、选择、编排、错误处理、安全约束和性能优化等模式。掌握这些模式，能够帮助你构建出既强大又安全的 Agent 工具系统。好的工具设计 = 清晰的接口 + 健壮的错误处理 + 合理的降级策略。

延伸阅读

1. MCP协议规范: https://modelcontextprotocol.io/
2. 论文: "ReAct: Synergizing Reasoning and Acting in Language Models"
3. 论文: "Toolformer: Language Models Can Teach Themselves to Use Tools"
4. OpenAI Function Calling: https://platform.openai.com/docs/guides/function-calling
5. Anthropic Tool Use: https://docs.anthropic.com/claude/docs/tool-use