工具执行器 —— 管理和执行所有工具。 它做两件事: 1. 注册工具(把工具的名称和执行函数绑定起来) 2. 执行工具(根据名称找到函数,传入参数,返回结果) 对应源码: conversation.rs 中的 ToolExecutor trait 以及 tools/src/lib.rs 中的 StaticToolExecutor
| 249 | # 这样加新工具的时候,Agentic Loop 的代码完全不用改! |
| 250 | |
| 251 | class ToolExecutor: |
| 252 | """ |
| 253 | 工具执行器 —— 管理和执行所有工具。 |
| 254 | |
| 255 | 它做两件事: |
| 256 | 1. 注册工具(把工具的名称和执行函数绑定起来) |
| 257 | 2. 执行工具(根据名称找到函数,传入参数,返回结果) |
| 258 | |
| 259 | 对应源码: conversation.rs 中的 ToolExecutor trait |
| 260 | 以及 tools/src/lib.rs 中的 StaticToolExecutor |
| 261 | """ |
| 262 | |
| 263 | def __init__(self): |
| 264 | # _handlers 是一个字典:工具名 → 执行函数 |
| 265 | # 就像一本电话簿:名字 → 电话号码 |
| 266 | self._handlers: dict[str, Callable[[dict], str]] = {} |
| 267 | # _specs 保存工具的说明书 |
| 268 | self._specs: dict[str, ToolSpec] = {} |
| 269 | |
| 270 | def register(self, spec: ToolSpec, handler: Callable[[dict], str]) -> "ToolExecutor": |
| 271 | """ |
| 272 | 注册一个工具。 |
| 273 | |
| 274 | 参数: |
| 275 | spec: 工具说明书 |
| 276 | handler: 工具的执行函数 |
| 277 | 返回: |
| 278 | self(返回自身是为了支持链式调用,后面演示) |
| 279 | """ |
| 280 | self._handlers[spec.name] = handler |
| 281 | self._specs[spec.name] = spec |
| 282 | return self # 返回自身,这样可以连续调用 .register().register() |
| 283 | |
| 284 | def execute(self, tool_name: str, input_json: str) -> tuple[str, bool]: |
| 285 | """ |
| 286 | 执行一个工具。 |
| 287 | |
| 288 | 参数: |
| 289 | tool_name: 工具名称 |
| 290 | input_json: 参数(JSON 字符串) |
| 291 | 返回: |
| 292 | (输出结果, 是否出错) |
| 293 | """ |
| 294 | if tool_name not in self._handlers: |
| 295 | return (f"Unknown tool: {tool_name}", True) |
| 296 | |
| 297 | try: |
| 298 | params = json.loads(input_json) if input_json else {} |
| 299 | except json.JSONDecodeError: |
| 300 | return (f"Invalid JSON input: {input_json}", True) |
| 301 | |
| 302 | try: |
| 303 | result = self._handlers[tool_name](params) |
| 304 | return (result, False) |
| 305 | except Exception as e: |
| 306 | return (f"Tool execution error: {e}", True) |
| 307 | |
| 308 | def get_specs(self) -> list[ToolSpec]: |