Tool Calling 进阶：工具网关、多工具编排与生产级实践

为什么工具调用需要进阶设计

基础的 Function Calling（定义一个 JSON Schema → LLM 选择调用 → 执行返回结果）在 1-5 个工具的简单场景下工作良好。但当 Agent 拥有 20+ 工具时，问题开始暴露：

• 所有工具定义注入 System Prompt → Token 消耗暴涨（每个工具 ~200-500 token）
• LLM 在大量工具中选择的准确率下降（"选择困难症"）
• 工具之间有依赖关系时编排变得复杂
• 不同工具的执行环境、超时、重试策略各不相同
• 工具的权限控制、审计、限流需要统一管理

本文深入讨论工具系统的进阶设计模式。

模式一：工具网关（Tool Gateway）

核心思想：不在 System Prompt 中列出所有工具，而是先做意图分类，只注入相关工具子集。

┌─────────────────────────────────────────────────┐
│                  Agent 循环                       │
│  用户输入 → 意图分类 → 获取相关工具 → LLM 推理     │
│             ↓                      ↓             │
│      Tool Gateway           工具执行 + 结果回传    │
└─────────────────────────────────────────────────┘

# 实现
class ToolGateway:
    """两层路由：分类 → 选工具"""

    # 第一层：意图分类（轻量级，用便宜的模型）
    INTENT_CLASSES = {
        "search": ["web_search", "kb_search", "image_search", "code_search"],
        "data_query": ["sql_query", "api_fetch", "file_read", "csv_analyze"],
        "action": ["send_email", "create_ticket", "deploy", "send_message"],
        "analysis": ["code_review", "data_viz", "text_summary", "sentiment"],
        "knowledge": ["rag_query", "vector_search", "doc_qa"],
    }

    async def route(self, user_input: str, conversation_context: list):
        # Step 1: 意图分类 → 获取工具类别
        intent = await self.classify_intent(user_input, conversation_context)

        # Step 2: 返回该类别下的工具定义（通常 3-8 个）
        candidate_tools = self.INTENT_CLASSES.get(intent, [])

        # Step 3: 按相关性排序，取 top-N
        scored_tools = await self.score_relevance(
            user_input, candidate_tools
        )
        return scored_tools[:8]  # 最多注入 8 个工具到 Prompt

意图分类的两种实现

模式二：工具组合链（Tool Chaining）

当一个任务的完成需要多个工具按顺序执行，且后一个工具的输入依赖前一个工具的输出时，称为工具链。

示例：用户问"帮我查一下最近 3 天关于 AI Agent 的热门 GitHub 仓库，然后发邮件给我"

# 工具链自动编排
Step 1: github_trending_search(query="AI Agent", days=3)
    → [{"repo": "crewAI", "stars": 15000}, {"repo": "AutoGPT", ...}]

Step 2: summarize_repos(repos=[...])  # 依赖 Step 1 的输出
    → "3个热门仓库：crewAI（多Agent框架）、AutoGPT..."

Step 3: send_email(to="user@example.com", body=summary)  # 依赖 Step 2
    → {"status": "sent", "message_id": "abc123"}

自动编排 vs 手动编排

• LLM 自动编排：ReAct 循环中 LLM 自己决定调用顺序。灵活但可能出错。
• DAG 预定义：用有向无环图定义工具依赖关系，Agent 按图执行。精确但不够灵活。
• 混合模式：常用链用 DAG 预定义，未知链交给 LLM 自由组合。实战推荐。

# DAG 预定义示例
tool_dag = {
    "send_report": {
        "steps": [
            {"tool": "query_database", "args": {"sql": "..."}, "depends_on": []},
            {"tool": "generate_chart", "args": {"data": "$query_database.result"},
             "depends_on": ["query_database"]},
            {"tool": "send_email", "args": {"attachment": "$generate_chart.url"},
             "depends_on": ["generate_chart"]},
        ],
        "timeout": 120,
        "on_error": "retry_once"
    }
}

模式三：工具描述工程（Tool Description Engineering）

工具描述的质量直接决定 LLM 能否正确选择和使用工具。这里的"工程"远不止写一段 description 文字。

3.1 命名规范

3.2 Description 的黄金法则

# ❌ 差：描述太泛
"description": "搜索东西"

# ✅ 好：清晰的边界 + 使用指导
"description": (
    "在内部知识库中全文搜索文档。"
    "适用场景：查找技术文档、API 参考、历史决策记录。"
    "不适用场景：实时网页搜索（用 search_web）、SQL 查询（用 query_db）。"
    "返回格式：JSON 数组，每项含 title/snippet/score/url。"
    "提示：短查询（2-5词）比长句子效果好；用英文关键词比中文效果好。"
)

3.3 参数描述的技巧

• 给出具体示例值而非抽象描述："日期格式 YYYY-MM-DD，如 2026-06-10"
• 使用 enum 约束可选值，减少参数幻觉
• 用 minimum/maximum 限定数值范围
• 标记 required 字段，让 LLM 知道哪些参数是必需的
• 互斥参数用 description 说明："callback_url 和 polling 二选一，不要同时提供"

模式四：并行工具调用（Parallel Tool Calls）

现代 LLM（GPT-4、Claude 3.5+）支持单次 Response 返回多个工具调用，应用层可以并行执行，大幅提升效率。

适用场景

# 用户问："比较北京、上海、广州、深圳今天的天气"
# LLM 一次返回 4 个工具调用：
tool_call_1: get_weather("北京")
tool_call_2: get_weather("上海")
tool_call_3: get_weather("广州")
tool_call_4: get_weather("深圳")

# 应用层并行执行 4 个 API 调用 → 汇总结果 → LLM 生成对比回答

并行调用的陷阱

模式五：错误处理与重试策略

工具调用不可能 100% 成功。好的错误处理能让 Agent 在失败时优雅降级而非崩溃。

class ToolExecutor:
    async def execute(self, tool_name: str, args: dict,
                       retry_config: RetryConfig) -> ToolResult:

        for attempt in range(retry_config.max_retries + 1):
            try:
                result = await asyncio.wait_for(
                    self.tools[tool_name].arun(**args),
                    timeout=retry_config.timeout
                )
                return ToolResult(success=True, data=result)

            except asyncio.TimeoutError:
                if attempt < retry_config.max_retries:
                    continue  # 重试
                return ToolResult(
                    success=False,
                    error=f"操作超时（>{retry_config.timeout}s），已重试{retry_config.max_retries}次",
                    error_type="timeout"
                )

            except ValidationError as e:
                # 参数错误 → 不重试，直接返回明确的错误信息
                return ToolResult(
                    success=False,
                    error=f"参数错误: {str(e)}\n"
                          f"正确的参数格式: {json.dumps(tool.schema, indent=2)}",
                    error_type="validation"
                )

            except PermissionError:
                # 权限错误 → 不重试
                return ToolResult(
                    success=False,
                    error="没有执行此操作的权限，请检查授权",
                    error_type="permission"
                )

错误信息设计原则

工具返回的错误信息是给 LLM 看的，LLM 需要根据错误信息决定下一步行动。因此错误信息必须：

• 明确错误类型：超时 / 参数错误 / 权限不足 / 资源不存在 / 服务不可用
• 给出可行建议："请检查日期格式，正确格式为 YYYY-MM-DD"
• 告知是否可重试："此操作不可重试，请尝试其他方式"
• 不要暴露内部细节：返回 "数据库查询失败" 而不是 "PostgreSQL connection pool exhausted at 10.0.1.5:5432"

模式六：工具执行的安全沙箱

LLM 生成的工具调用本质上是"不可信的输入"。工具执行层必须有安全边界：

class SecureToolExecutor:
    """在执行前对工具调用做多层校验"""

    async def execute(self, tool_call: ToolCall, context: SessionContext):
        # 层1: 权限检查
        if tool_call.name not in context.allowed_tools:
            raise PermissionError(f"无权限使用 {tool_call.name}")

        # 层2: 参数白名单校验
        validated_args = self.validate_args(tool_call)

        # 层3: 速率限制（按工具 + 用户维度）
        await self.rate_limiter.check(tool_call.name, context.user_id)

        # 层4: SQL 注入防护（如果涉及数据库查询）
        if tool_call.name == "sql_query":
            validated_args["sql"] = self.sanitize_sql(validated_args["sql"])

        # 层5: 文件路径沙箱（限制访问范围）
        if tool_call.name == "file_read":
            validated_args["path"] = self.sandbox_path(validated_args["path"],
                                                         context.workspace)

        # 层6: 审计日志
        self.audit_log.record(context.user_id, tool_call, validated_args)

        # 执行
        return await self.tools[tool_call.name].arun(**validated_args)

工具调用的 Token 成本优化

工具调用是 Agent 系统的主要 Token 消耗来源之一，优化策略：

1. 压缩工具定义：精简 description，去掉不必要的格式说明。每个工具定义从 500 token 压缩到 200 token，20 个工具就能节省 6000 token
2. 按需注入：用 Tool Gateway 只注入相关工具，RAG 场景同理
3. 缓存工具定义：如果模型提供商支持 Prompt Caching（如 Claude），工具定义部分可以被缓存复用
4. 截断工具结果：工具返回的结果可能很长（如搜索返回 10 篇长文档），截断到必要部分（如前 2000 字符）
5. 摘要化工具结果：对特别长的工具结果（如完整的 API 响应），先用小模型摘要再传回主模型

何时用 MCP 替代原生 Function Calling

← 返回首页