Agent 工具调用架构：从 Function Calling 到动态工具编排的工程实践

一、工具膨胀带来的选择难题

在 AI Agent 的落地过程中，工具调用（Tool Use / Function Calling）是最核心的能力之一。没有工具的 Agent 只是一个聊天机器人，有了工具才真正成为"能做事"的系统。然而，当工具数量从 5 个增长到 50 个、再到 500 个时，一个看似简单的问题变得极其棘手：Agent 该如何从数百个工具中，精准选出当前需要的那一个？

这个问题的严重性往往被低估。某金融风控团队给 Agent 配置了 120 个工具，涵盖数据库查询、规则引擎调用、外部 API 对接等。上线后发现，Agent 的工具选择准确率从 5 个工具时的 92% 暴跌到 38%。更糟糕的是，错误工具调用不仅浪费 Token，还可能触发危险操作——比如把"查询用户余额"误选为"冻结用户账户"。

工具选择失败的根因有三个：第一，工具描述的语义重叠导致模型混淆；第二，Prompt 中塞入过多工具描述，超出模型的有效注意力范围；第三，静态工具列表无法适应上下文变化，某些工具只在特定场景下才有意义。解决这些问题，需要从架构层面重新设计工具调用的全链路。

二、工具调用的底层机制：从 Prompt 构造到动态路由

Function Calling 的本质，是在大模型的 Prompt 中注入工具的 JSON Schema 描述，让模型输出结构化的工具调用指令。但这个"注入"过程远比想象中复杂。下面这张图展示了完整的工具调用架构：

graph TD
    A[用户请求] --> B[上下文分析器]
    B -->|提取意图与实体| C[工具路由器]
    C -->|语义匹配| D[向量索引: 工具嵌入]
    D -->|Top-K 候选| E[候选工具集]
    E -->|Schema 注入| F[LLM Prompt 构造器]
    F -->|带工具的 Prompt| G[大模型推理]
    G -->|工具调用指令| H[参数校验器]
    H -->|校验通过| I[工具执行器]
    H -->|校验失败| J[错误反馈 → 重试]
    I -->|执行结果| K[结果解析与汇总]
    K -->|最终响应| L[用户输出]

    C -->|规则匹配| M[静态规则表]
    M --> E

    style A fill:#e1f5fe
    style L fill:#e8f5e9
    style D fill:#fff3e0
    style J fill:#ffebee

核心机制分为三层：

第一层：工具注册与索引。每个工具在注册时，除了 JSON Schema，还需要生成一段语义描述的向量嵌入（Embedding）。这段描述不是简单的工具名，而是工具的功能摘要、适用场景和典型用例的组合文本。例如，"查询用户余额"工具的语义描述可能是"根据用户 ID 查询账户当前余额，适用于账单核对、风控检查等场景，输入为用户唯一标识"。

第二层：工具路由。当用户请求到达时，先通过上下文分析器提取意图和实体，然后将意图向量与工具嵌入做相似度检索，取 Top-K 候选。同时，静态规则表提供基于关键词的精确匹配作为补充。两层结果合并后，形成当前请求的候选工具集。

第三层：Prompt 构造与执行。将候选工具集的 Schema 注入 Prompt，交由大模型决策。模型输出工具调用指令后，参数校验器检查类型、必填项和值域，校验通过才执行。

三、生产级代码实现：动态工具路由与安全执行

下面是一个完整的动态工具路由系统实现，包含工具注册、向量检索、安全执行三个核心模块：

import asyncio
import hashlib
import json
from dataclasses import dataclass, field
from typing import Any, Callable, Optional

import numpy as np
from pydantic import BaseModel, ValidationError


# ---- 工具定义 ----
class ToolSchema(BaseModel):
    """工具的完整定义，包含元信息与参数 Schema"""
    name: str
    description: str  # 语义描述，用于向量检索
    use_cases: list[str]  # 适用场景列表
    parameters: dict  # JSON Schema 格式的参数定义
    danger_level: int = 0  # 0=只读, 1=写入, 2=危险操作
    requires_confirmation: bool = False


@dataclass
class RegisteredTool:
    """注册后的工具对象，包含运行时信息"""
    schema: ToolSchema
    handler: Callable
    embedding: Optional[np.ndarray] = None
    call_count: int = 0
    error_count: int = 0


class ToolRouter:
    """动态工具路由器：基于语义检索 + 规则匹配的混合路由"""

    def __init__(self, embed_dim: int = 1536, top_k: int = 8):
        self._tools: dict[str, RegisteredTool] = {}
        self._embeddings: np.ndarray = np.empty((0, embed_dim))
        self._name_index: list[str] = []
        self._rule_table: dict[str, list[str]] = {}  # 关键词 → 工具名列表
        self._top_k = top_k

    def register(self, schema: ToolSchema, handler: Callable):
        """注册工具，自动生成语义嵌入与规则索引"""
        # 生成语义嵌入（实际生产中调用 Embedding API）
        embedding = self._compute_embedding(schema)
        tool = RegisteredTool(schema=schema, handler=handler, embedding=embedding)
        self._tools[schema.name] = tool

        # 更新向量索引
        if self._embeddings.size == 0:
            self._embeddings = embedding.reshape(1, -1)
        else:
            self._embeddings = np.vstack([self._embeddings, embedding])
        self._name_index.append(schema.name)

        # 更新规则索引：从 use_cases 提取关键词
        for case in schema.use_cases:
            for keyword in self._extract_keywords(case):
                self._rule_table.setdefault(keyword, []).append(schema.name)

    def route(self, intent: str, context: dict) -> list[ToolSchema]:
        """混合路由：语义检索 + 规则匹配，返回候选工具集"""
        # 语义检索
        query_emb = self._compute_text_embedding(intent)
        scores = self._embeddings @ query_emb
        top_indices = np.argsort(scores)[-self._top_k:][::-1]
        semantic_candidates = {self._name_index[i] for i in top_indices}

        # 规则匹配
        rule_candidates = set()
        for keyword in self._extract_keywords(intent):
            rule_candidates.update(self._rule_table.get(keyword, []))

        # 合并去重，语义检索结果优先
        merged = list(semantic_candidates)
        for name in rule_candidates:
            if name not in semantic_candidates:
                merged.append(name)

        return [self._tools[n].schema for n in merged if n in self._tools]

    def _compute_embedding(self, schema: ToolSchema) -> np.ndarray:
        """将工具的语义信息编码为向量（生产环境替换为真实 Embedding 调用）"""
        text = f"{schema.name}: {schema.description} | 场景: {', '.join(schema.use_cases)}"
        # 这里用哈希模拟，实际应调用 text-embedding-3-small 等
        hash_bytes = hashlib.sha256(text.encode()).digest()
        return np.frombuffer(hash_bytes[:24], dtype=np.float32) / 128.0

    def _compute_text_embedding(self, text: str) -> np.ndarray:
        """将查询文本编码为向量"""
        hash_bytes = hashlib.sha256(text.encode()).digest()
        return np.frombuffer(hash_bytes[:24], dtype=np.float32) / 128.0

    @staticmethod
    def _extract_keywords(text: str) -> list[str]:
        """简易关键词提取（生产环境用 jieba + 停用词过滤）"""
        return [w for w in text.split() if len(w) > 1]


class SafeToolExecutor:
    """安全工具执行器：参数校验 + 超时控制 + 危险操作拦截"""

    def __init__(self, router: ToolRouter, timeout: float = 30.0):
        self._router = router
        self._timeout = timeout

    async def execute(self, tool_name: str, arguments: dict) -> dict:
        """执行工具调用，包含完整的安全检查链"""
        if tool_name not in self._router._tools:
            return {"error": f"工具 '{tool_name}' 未注册"}

        tool = self._router._tools[tool_name]
        schema = tool.schema

        # 危险操作拦截
        if schema.danger_level >= 2 and not schema.requires_confirmation:
            return {"error": f"工具 '{tool_name}' 为危险操作，需人工确认后执行"}

        # 参数校验：利用 Pydantic 验证 JSON Schema
        try:
            self._validate_params(schema.parameters, arguments)
        except ValueError as e:
            return {"error": f"参数校验失败: {e}"}

        # 超时执行
        try:
            result = await asyncio.wait_for(
                self._call_handler(tool.handler, arguments),
                timeout=self._timeout,
            )
            tool.call_count += 1
            return {"result": result}
        except asyncio.TimeoutError:
            tool.error_count += 1
            return {"error": f"工具 '{tool_name}' 执行超时 ({self._timeout}s)"}
        except Exception as e:
            tool.error_count += 1
            return {"error": f"工具执行异常: {type(e).__name__}: {e}"}

    @staticmethod
    async def _call_handler(handler: Callable, args: dict) -> Any:
        """统一处理同步/异步 handler"""
        if asyncio.iscoroutinefunction(handler):
            return await handler(**args)
        return handler(**args)

    @staticmethod
    def _validate_params(schema: dict, params: dict):
        """基于 JSON Schema 的参数校验（核心字段检查）"""
        required = schema.get("required", [])
        for field_name in required:
            if field_name not in params:
                raise ValueError(f"缺少必填参数: {field_name}")

        properties = schema.get("properties", {})
        for key, value in params.items():
            if key not in properties:
                raise ValueError(f"未知参数: {key}")
            # 类型检查简化版，生产环境用 jsonschema 库完整校验
            expected_type = properties[key].get("type")
            if expected_type == "string" and not isinstance(value, str):
                raise ValueError(f"参数 '{key}' 应为字符串类型")
            if expected_type == "integer" and not isinstance(value, int):
                raise ValueError(f"参数 '{key}' 应为整数类型")

这段代码的核心设计思路有三点：第一，工具注册时同时构建向量索引和规则索引，实现语义检索与精确匹配的混合路由；第二，执行器对危险操作分级拦截，danger_level >= 2 的工具必须经过人工确认；第三，所有工具调用都有超时保护和异常计数，防止单个工具故障拖垮整个 Agent。

四、架构权衡：检索精度、延迟与安全性的三角博弈

动态工具路由并非银弹，它在三个维度上存在不可回避的 Trade-off：

精度 vs 延迟。向量检索的 Top-K 越大，召回率越高，但注入 Prompt 的工具描述越多，模型的推理延迟和 Token 消耗也越大。实测数据：Top-5 时工具选择准确率 85%，P95 延迟 1.2s；Top-15 时准确率 94%，P95 延迟 2.8s。对于实时对话场景，Top-8 到 Top-10 是一个合理的平衡点。

动态路由 vs 静态绑定。动态路由的灵活性是以额外的 Embedding 计算和检索延迟为代价的。如果工具数量少于 10 个，静态绑定（把所有工具都放进 Prompt）反而更简单高效。动态路由的收益在工具数量超过 20 个时才显著体现。

安全拦截 vs 执行效率。危险操作分级和参数校验增加了每次调用的开销。在高吞吐场景下（如批量文档处理），逐条校验可能成为瓶颈。一种折中方案是：对只读工具（danger_level=0）跳过参数校验，仅对写入和危险操作启用完整校验链。

此外，工具描述的质量直接决定路由效果。一段模糊的描述（如"查询数据"）会导致语义检索无法区分"查询用户信息"和"查询订单记录"。工具注册时必须强制要求填写结构化的语义描述，包括功能、输入、输出和适用场景四个维度。

五、总结

Agent 工具调用架构的核心挑战，不是"如何让模型调用工具"，而是"如何在工具规模膨胀时保持选择精度与执行安全"。本文给出的解法是三层架构：语义嵌入 + 规则索引的混合路由解决选择精度问题，参数校验 + 危险分级的安全执行器解决执行安全问题。

落地路线建议：第一阶段，工具数量小于 10 个时，直接使用静态绑定，所有工具 Schema 全量注入 Prompt；第二阶段，工具数量 10-50 个时，引入基于关键词的规则路由，按场景分组工具；第三阶段，工具数量超过 50 个时，部署完整的向量检索路由，配合 Embedding 模型实现语义匹配。每个阶段都要建立工具调用的监控指标——选择准确率、执行成功率、P95 延迟——用数据驱动架构演进，而非凭直觉提前过度设计。