Skip to content

03.2 — 工具定义与注册

定位: Agent 工具系统的"骨架"——理解如何规范定义工具、如何设计可扩展的工具注册中心、以及工具生命周期管理 面试高频度: ⭐⭐⭐⭐ 考查方式: 设计题(设计工具注册系统)、对比题(不同工具定义方式的优劣)、实践题(工具 Schema 编写)


一、这是什么?为什么需要它?

是什么

工具定义 是描述工具功能、参数、约束的结构化规范。工具注册 是将定义好的工具注册到 Agent 运行时的管理过程。

为什么需要规范的工具定义?

❌ 没有规范定义的工具:
  每个工具的描述风格不一致
  LLM 难以理解工具是做什么的
  参数信息不完整,LLM 不知道传什么
  工具发现困难——不知道有哪些工具可用

✅ 有规范定义的工具:
  LLM 精确理解每个工具的用途
  参数信息完整,调用准确率高
  工具可被发现、可被搜索
  统一管理:版本、权限、监控

核心洞察:工具定义的质量直接决定了 LLM 能否准确调用它。写工具描述,就像在教一个聪明但"没见过这个世界"的人使用你的 API


二、原理拆解

2.1 Tool Schema 规范详解

通用的工具定义格式(JSON Schema):

json
{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "properties": {
    "name": {
      "type": "string",
      "description": "工具名称(唯一标识符)"
    },
    "description": {
      "type": "string",
      "description": "工具描述——LLM 决定是否调用此工具的依据"
    },
    "parameters": {
      "type": "object",
      "description": "工具参数定义(OpenAPI-like JSON Schema)",
      "properties": {},
      "required": []
    }
  }
}

编写高质量工具定义的黄金法则

python
# ❌ 差的定义
def get_weather(city):
    """获取天气"""
    pass

# ✅ 好的定义
get_weather_schema = {
    "name": "get_weather",
    "description": "获取指定城市的实时天气信息,包括温度、天气状况、湿度等",
    "parameters": {
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "城市名,如'北京'、'上海'、'New York'"
            },
            "unit": {
                "type": "string",
                "enum": ["celsius", "fahrenheit"],
                "description": "温度单位,默认 celsius"
            }
        },
        "required": ["city"]
    }
}

工具定义的关键原则

原则解释反例正例
描述清晰说清楚工具做什么"处理数据""从数据库查询用户订单记录"
参数精确每个参数说清楚格式和用途"data""userId (int): 用户唯一标识"
约束明确枚举值、格式、范围"type""type (enum: pdf/csv/json)"
边界说明什么情况不能用"仅支持查询 30 天内数据"
示例提供给 LLM 参考示例"示例: city='北京'"

2.2 工具注册中心设计

python
class ToolRegistry:
    """可扩展的工具注册中心"""
    
    def __init__(self):
        self._tools = {}        # name -> Tool 对象
        self._groups = {}       # group_name -> [tool_names]
        self._middlewares = []  # 工具执行中间件链
    
    def register(self, tool_def: dict, group: str = "default"):
        """注册一个工具"""
        name = tool_def["name"]
        if name in self._tools:
            raise ValueError(f"工具 {name} 已注册")
        
        self._tools[name] = {
            "definition": tool_def,
            "handler": None,    # 稍后绑定
            "enabled": True,
            "metadata": {
                "created_at": now(),
                "call_count": 0,
                "error_count": 0,
            }
        }
        
        if group not in self._groups:
            self._groups[group] = []
        self._groups[group].append(name)
    
    def bind_handler(self, name: str, handler: callable):
        """绑定工具的执行函数"""
        if name not in self._tools:
            raise KeyError(f"工具 {name} 未注册")
        self._tools[name]["handler"] = handler
    
    def get_tools_for_scene(self, scene: str) -> list:
        """根据场景获取可用工具"""
        if scene in self._groups:
            return [
                self._tools[name]["definition"]
                for name in self._groups[scene]
                if self._tools[name]["enabled"]
            ]
        return []
    
    def execute(self, name: str, arguments: dict) -> dict:
        """执行工具"""
        tool = self._tools.get(name)
        if not tool:
            return {"error": f"未知工具: {name}"}
        if not tool["enabled"]:
            return {"error": f"工具 {name} 已禁用"}
        if not tool["handler"]:
            return {"error": f"工具 {name} 未绑定执行函数"}
        
        # 中间件链(权限检查、日志、限流等)
        context = {"tool": tool, "arguments": arguments}
        for middleware in self._middlewares:
            result = middleware(context)
            if result:  # 中间件返回非空 = 阻断
                return result
        
        try:
            tool["metadata"]["call_count"] += 1
            result = tool["handler"](**arguments)
            return {"result": result}
        except Exception as e:
            tool["metadata"]["error_count"] += 1
            return {"error": str(e)}

# 使用示例
registry = ToolRegistry()

# 1. 定义并注册工具
registry.register({
    "name": "search_docs",
    "description": "搜索技术文档",
    "parameters": {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "搜索关键词"}
        },
        "required": ["query"]
    }
}, group="research")

# 2. 绑定执行函数
def search_docs_handler(query: str) -> str:
    return f"搜索 '{query}' 的结果..."

registry.bind_handler("search_docs", search_docs_handler)

# 3. Agent 获取可用工具
research_tools = registry.get_tools_for_scene("research")
print(f"研究场景可用工具: {[t['name'] for t in research_tools]}")

2.3 工具的分组与场景管理

随着工具数量增长,需要分组管理:

场景分组的核心思想:
  不把所有工具都给 LLM——而是根据当前场景选择合适的工具子集

分组策略:
  ┌─ 按功能 ─────────────────┐
  │  数据查询组: search, query_db  │
  │  文件操作组: read, write, edit │
  │  代码执行组: python, bash     │
  │  网络请求组: fetch, api_call  │
  └────────────────────────────┘

场景与工具组的映射:
  "帮我查资料" → 数据查询组
  "帮我改代码" → 文件操作组 + 代码执行组
  "分析数据"   → 数据查询组 + 代码执行组

动态工具选择策略

python
class DynamicToolSelector:
    """动态选择适合当前任务的工具集"""
    
    def select_tools(self, task: str, available_groups: dict) -> list:
        """让 LLM 自己判断需要哪些工具"""
        groups_desc = "\n".join([
            f"- {group}: {desc}"
            for group, desc in available_groups.items()
        ])
        
        decision = llm.call(f"""
        任务: {task}
        可用工具组:
        {groups_desc}
        
        请选择当前任务最需要的工具组(可多选):
        只回复组名,每行一个。
        """)
        
        selected = []
        for line in decision.strip().split("\n"):
            line = line.strip()
            if line in available_groups:
                selected.extend(available_groups[line])
        
        return selected[:15]  # 限制最多 15 个工具

2.4 工具版本管理

python
class VersionedToolRegistry(ToolRegistry):
    """支持版本管理的工具注册中心"""
    
    def register_version(self, name: str, version: str, tool_def: dict):
        """注册特定版本的工具"""
        if name not in self._tools:
            self._tools[name] = {"versions": {}}
        
        self._tools[name]["versions"][version] = tool_def
        self._tools[name]["current_version"] = version
    
    def rollback(self, name: str, version: str):
        """回滚到指定版本"""
        if version in self._tools[name]["versions"]:
            self._tools[name]["current_version"] = version
            return True
        return False
    
    def get_tool_def(self, name: str) -> dict:
        """获取当前版本的工具定义"""
        tool = self._tools[name]
        ver = tool.get("current_version")
        return tool["versions"][ver]

三、图解全景

┌──────────────────────────────────────────────────────────────┐
│                工具系统架构图                                 │
│                                                              │
│  ┌─ 工具定义层 ──────────────────────────────────────────┐   │
│  │                                                        │   │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐ │   │
│  │  │ 工具 A       │  │ 工具 B       │  │ 工具 C       │ │   │
│  │  │ name: search │  │ name: calc   │  │ name: send   │ │   │
│  │  │ desc: 搜索   │  │ desc: 计算   │  │ desc: 发送   │ │   │
│  │  │ params: {...}│  │ params: {...}│  │ params: {...}│ │   │
│  │  └──────────────┘  └──────────────┘  └──────────────┘ │   │
│  └──────────────────────┬─────────────────────────────────┘   │
│                         │                                      │
│  ┌──────────────────────▼─────────────────────────────────┐   │
│  │  工具注册中心(Registry)                                │   │
│  │                                                        │   │
│  │  ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐         │   │
│  │  │ 元数据  │ │ 场景分组 │ │ 权限   │ │ 监控   │         │   │
│  │  │ 管理   │ │ 映射   │ │ 控制   │ │ 统计   │         │   │
│  │  └────────┘ └────────┘ └────────┘ └────────┘         │   │
│  └──────────────────────┬─────────────────────────────────┘   │
│                         │                                      │
│  ┌──────────────────────▼─────────────────────────────────┐   │
│  │  中间件链                                               │   │
│  │                                                        │   │
│  │  鉴权 → 限流 → 审计 → 参数校验 → 执行 → 结果处理        │   │
│  └──────────────────────┬─────────────────────────────────┘   │
│                         │                                      │
│                         ▼                                      │
│                    ┌──────────┐                               │
│                    │ LLM 推理  │ ← 只看到工具定义,不关心实现   │
│                    └──────────┘                               │
└──────────────────────────────────────────────────────────────┘

四、实战验证

4.1 构建一个完整的多场景工具系统

python
# 构建一个支持多场景的工具系统

# 1. 定义各场景的工具
data_tools = {
    "search": {
        "name": "search",
        "description": "搜索内部知识库",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "limit": {"type": "integer", "default": 5}
            },
            "required": ["query"]
        }
    },
    "query_db": {
        "name": "query_db",
        "description": "执行 SQL 查询",
        "parameters": {
            "type": "object",
            "properties": {
                "sql": {"type": "string", "description": "SQL 语句"}
            },
            "required": ["sql"]
        }
    }
}

code_tools = {
    "read_file": {
        "name": "read_file",
        "description": "读取文件内容",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {"type": "string"}
            },
            "required": ["path"]
        }
    },
    "run_code": {
        "name": "run_code",
        "description": "执行 Python 代码",
        "parameters": {
            "type": "object",
            "properties": {
                "code": {"type": "string"},
                "timeout": {"type": "integer", "default": 30}
            },
            "required": ["code"]
        }
    }
}

# 2. 注册中心
class SimpleRegistry:
    def __init__(self):
        self.tools = {}
        self.scenes = {}
    
    def register_scene(self, scene_name: str, tools: dict):
        self.scenes[scene_name] = tools
        self.tools.update(tools)
    
    def get_scene_tools(self, scene: str):
        return list(self.scenes.get(scene, {}).values())
    
    def get_all_tools(self):
        return list(self.tools.values())

# 3. 使用
registry = SimpleRegistry()
registry.register_scene("data_analysis", data_tools)
registry.register_scene("code_development", code_tools)

# 分析场景只给分析工具
print("数据分析场景工具:", 
      [t["name"] for t in registry.get_scene_tools("data_analysis")])

# 所有工具
print("全部工具:", 
      [t["name"] for t in registry.get_all_tools()])

4.2 工具定义质量检查

bash
# 工具定义质量的自动检查

# 检查点 1: description 不能太短
grep -Po '"description":\s*"\w{1,10}"' tool_definitions.json
# 输出 → 说明 description 过短(少于10字),需要补充

# 检查点 2: 参数要求有 description
grep -Po '"description":\s*"[^"]*"' tool_definitions.json | wc -l
# 对比参数个数,确保每个参数都有描述

# 检查点 3: 枚举值列出所有选项
# 手工审查 type: enum 的参数是否列出了所有值

五、面试视角

追问答案要点
工具注册中心的核心功能?工具注册(名称+Schema+handler)、场景分组(按需提供工具集)、执行路由(调用到具体函数)、中间件(权限/限流/监控)、元数据管理(版本/统计)
如何编写高质量的工具定义?黄金法则:name 唯一且语义化、description 写清用途和边界、参数精确描述格式和约束、提供示例值。目的是让 LLM "一看就懂"
工具数量多了怎么办?分组管理(按功能/场景分组)、动态选择(让 LLM 先选组再选工具)、分层注册(高频工具常驻,低频按需加载)。单次推理工具不超过 15 个
工具注册中心如何保证高可用?工具缓存(减少注册中心查询)、降级策略(注册中心挂掉时使用本地缓存)、健康检查(定期检测工具可用性)、熔断机制(连续失败的工具自动禁用)
如何设计工具的安全管控?三层控制:注册时验证(Schema 合法性)、执行前权限检查(谁可以调用)、执行后审计(记录全部调用日志)。敏感工具需要额外审批

📚 相关链接

  • **函数调用机制** — 函数调用的底层原理
  • **工具执行与容错** — 工具执行的错误处理
  • **工具滥用防御** — 工具安全相关
  • **LangChain 工具系统** — 框架中的工具实现
  • ← 返回 **Agent工具系统索引**

Knowledge4J — Java 知识库