08. nanobot 源码解读:tool

08. nanobot 源码解读：tool

文档内容基于 HKUDS/nanobot: "🐈 nanobot: The Ultra-Lightweight Personal AI Agent" 的 main 分支 56ce1816 提交进行说明。

• 信息陈旧：大模型的知识来自训练数据，在训练之后出现的新知识则无法知晓
• 计算不精准：大模型通过概率机制生成下一个 token，容易出现幻觉，如果进行数学计算等需要精确结果的操作时容易出现问题
• 无法执行动作：大模型如果要和外部系统交互，仅凭生成的文本很难操作外部系统

因此有了工具调用（function calling）的诉求。tool，就是能让大模型能够调用外部函数或API的能力。

tool 的抽象

nanobot 在 nanobot/agent/tools/base.py 中定义了描述 tool 的抽象类 Tool。

首先关注 Tool 中与大模型传参相关的抽象方法（已装饰为属性）：

• name：str，工具名称
• description：str，用于描述这个工具的功能和用途，大模型依据此判断是否要调用该工具
• parameters：dict[str, Ant]，描述工具参数的 JSON Schema。nanobot 提供了装饰器来设置 parameters ，所以源码中有部分实现类用装饰器替代了抽象方法实现。

其次关注 Tool 中与 Agent Loop 期间工具执行相关的抽象方法（已装饰为属性）：

• read_only：bool，表示该工具是只读操作，不会产生副作用
• exclusive：bool，表示该工具必须单独执行
• concurrent_safe：bool，默认值为 read_only and not exclusive，表示该工具在并发场景下是否安全

仅 concurrent_safe 属性在 Agent Loop 流程中被使用，用于控制工具是否并发执行。

接下来关注 Tool 自动发现与加载时用到的两个内部属性：

• _plugin_discoverable：bool，默认 True。nanobot 自动发现工具时，若该属性设置为 False 时，则不会被发现。
• _scopes：set[str]，默认 {"core"}，表示该工具的应用范围，可用的值有

• core：Agent Loop 流程中可以被主Agent使用
• subagent：Agent Loop 流程中可以被子Agent使用
• memory：顾名思义应该是在涉及改写记忆文件的操作（如 dream）中使用，但是当前仅在测试代码中发现使用 memory 值

最后重点关注 Tool 执行相关的抽象方法：

• execute：抽象方法，工具执行的实现处，接收大模型参数，执行工具调用，然后返回结果给大模型。nanobot 约定了该方法的返回值应该为 str 或者 list[dict]。

大模型调用工具时，会根据描述工具入参的 JSON Schema 来生成参数。nanobot 在收到大模型生成的参数后，需要将参数转换为适合 Python 函数调用的入参格式 dict，还需要校验参数是否符合 JSON Schema 规范，校验通过之后才会调用 execute。因此，Tool 还提供了关于参数校验、参数转换等方法的实现。

JSON Schema 相关

JSON Schema 是一种用于描述和验证 JSON 数据结构的声明式语言。

简单讲，JSON Schema 是一个描述数据结构的 JSON 格式规范，主要作用是：

• 告诉大模型，工具需要什么样的数据结构
• 校验大模型生成的参数是否符合数据结构规范

比如一个获取天气的工具 def get_weather(city: str) -> str，就需要一个 {"city": "xxx"} 格式的输入，用 JSON Schema 描述这个输入如下：

{    "type": "object",           # 输入是一个对象    "properties": {             # 这个对象包含以下属性        "city": {                 # 属性 city            "type": "string"        # 类型是 string        }    },    "required": ["city"]        # 对象的必填字段有 city}

如果大模型要获取北京的天气信息，就需要生成参数 {"city": "北京"}。

更多详细规则可以参阅 json-schema.org。

在 nanobot/agent/tools/base.py 中定义了抽象类 Schema：

• to_json_schema(): abstractmethod，生成符合 JSON Schema 格式的 dict
• validate_value(value): 校验传入的参数 value 是否符合自身表示的 JSON Schema
• validate_json_schema_value(val, schema): staticmethod，校验传入的参数 val 是否符合参数 schema 表示的 JSON Schema

nanobot/agent/tools/schema.py 中根据 JSON Schema 规范定义了各个类型对应的 Schema 实现：StringSchema、IntegerSchema、NumberSchema、BooleanSchema、ArraySchema、ObjectSchema。

tool 的执行

tool 的执行重点关注下面两处：

• nanobot/agent/tools/registry.py 中的 ToolRegistry.execute
• nanobot/agent/runner.py 中的 AgentRunner._run_tool

首先查看 ToolRegistry.execute，该方法实现了参数解析和工具执行：

async def execute(self, name: str, params: Any) -> Any:    """Execute a tool by name with given parameters."""    hint = "\n\n[Analyze the error above and try a different approach.]"    # prepare_call 需要做到这几件事:    #    1. 根据 name 找到 tool    #    2. 转换 params 为函数入参形式（dict）并校验是否符合工具参数规范    #    3. 返回 tool、工具入参、错误信息（找不到 tool 或者参数有问题）    tool, params, error = self.prepare_call(name, params)    if error:        return error + hint    try:        assert tool is not None  # guarded by prepare_call()        # Tool.execute 返回值建议为 str 或者 list[dict]        result = await tool.execute(**params)        # 如果有报错，建议返回 str 且以 Error 开头        if isinstance(result, str) and result.startswith("Error"):            return result + hint        return result    except Exception as e:        return f"Error executing {name}: {str(e)}" + hint

然后是 AgentRunner._run_tool，这个函数在 Agent Loop 流程中调用，对比 ToolRegistry.execute，有更多的业务处理逻辑，且增加了事件记录：

async def _run_tool(    self,    spec: AgentRunSpec,    tool_call: ToolCallRequest,    external_lookup_counts: dict[str, int],    workspace_violation_counts: dict[str, int],) -> tuple[Any, dict[str, str], BaseException | None]:    hint = "\n\n[Analyze the error above and try a different approach.]"    # 限制外部工具的重复调用(允许重复两次)：web_fetch(url) 和 web_search(query)    # 参数相同时就返回【让大模型参考之前的答案】    lookup_error = repeated_external_lookup_error(        tool_call.name,        tool_call.arguments,        external_lookup_counts,    )    if lookup_error:        event = {            "name": tool_call.name,            "status": "error",            "detail": "repeated external lookup blocked",        }        if spec.fail_on_tool_error:            return lookup_error + hint, event, RuntimeError(lookup_error)        return lookup_error + hint, event, None    # spec.tools 就是 ToolRegistry 实例，这里将 ToolRegistry.execute 的实现拆分了    # 然后加入事件记录，同时也将报错信息做了符合此处调用的处理    prepare_call = getattr(spec.tools, "prepare_call", None)    tool, params, prep_error = None, tool_call.arguments, None    if callable(prepare_call):        with suppress(Exception):            prepared = prepare_call(tool_call.name, tool_call.arguments)            if isinstance(prepared, tuple) and len(prepared) == 3:                tool, params, prep_error = prepared    if prep_error:        event = {            "name": tool_call.name,            "status": "error",            "detail": prep_error.split(": ", 1)[-1][:120],        }        # 对某些异常生成定制化的错误信息        #  - ssrf攻击：如通过 web_fetch 工具访问内网资源        #  - 工作区越界：如通过 read_file 工具访问没有权限读的文件        # 注意，这需要 tool 返回的错误信息字符串与匹配逻辑一致才可以细分        handled = self._classify_violation(            raw_text=prep_error,            soft_payload=prep_error + hint,            event=event,            tool_call=tool_call,            workspace_violation_counts=workspace_violation_counts,        )        # 有定制化错误信息就使用定制化错误信息        if handled is not None:            return handled        # 没有定制化错误信息就用之前的        return prep_error + hint, event, (            RuntimeError(prep_error) if spec.fail_on_tool_error else None        )    # 需要跟踪文件编辑情况，构造 file_edit_start 事件交给对应处理器处理    # 后面还有 file_edit_error / file_edit_end 事件    # 文件编辑跟踪相关的代码在这里先省略了    try:        if tool is not None:            # 执行工具            # 可以发现，AgentRunner._run_tool 整体和 ToolRegistry.execute 是一样的            # 都是 prepare_call + tool.execute            result = await tool.execute(**params)        else:            # 这块是兜底操作，回到了 ToolRegistry.execute            result = await spec.tools.execute(tool_call.name, params)    except asyncio.CancelledError:        raise    except BaseException as exc:        # 略，file_edit_error 事件 以及 error信息处理        return payload, event, None    # 略，file_edit_end 事件    # 对结果进行下修饰后返回    detail = "" if result is None else str(result)    detail = detail.replace("\n", " ").strip()    if not detail:        detail = "(empty)"    elif len(detail) > 120:        detail = detail[:120] + "..."    return result, {"name": tool_call.name, "status": "ok", "detail": detail}, None

tool 的发现与加载

nanobot 可以手动指定哪些工具需要添加到大模型参数，也可以通过 nanobot/agent/tools/loader.py 中的 ToolLoader 自动发现并加载符合规范的工具。

自动发现有两个来源：

第一个来源是扫描 nanobot/agent/tools/ 包，读取各模块属性，然后筛选符合规范的 Tool 实现类:

• 是 Tool 的实现类: isinstance(attr, type) and issubclass(attr, Tool) and attr is not Tool
• 不能是内部属性: not attr_name.startswith("_")
• 不能是抽象类: not getattr(attr, "__abstractmethods__", None)
• 允许被自动发现: not getattr(attr, "_plugin_discoverable", True) is False
• 首次发现

第二个来源是通过 entry-point 找到 nanobot.tools 组来筛选对应的 Tool 实现类:

• 是 Tool 的实现类: isinstance(cls, type) and issubclass(cls, Tool)
• 不是抽象类: not getattr(cls, "__abstractmethods__", None)
• 允许被自动发现: getattr(cls, "_plugin_discoverable", True)

然后在根据 scope 属性筛选哪些工具能被加载到 Agent Loop 流程中使用。

tool 的实现

需要继承 Tool 并实现抽象方法。用 read_file (ReadFileTool) 工具做个说明：

# 通过装饰器 tool_parameters 就不用实现 Tool 的抽象方法 parameters 了# 当然也可以提供 parameters 方法实现来描述 schema@tool_parameters(    tool_parameters_schema(        path=StringSchema("The file path to read"),        offset=IntegerSchema(            1,            description="Line number to start reading from (1-indexed, default 1)",            minimum=1,),        limit=IntegerSchema(            2000,            description="Maximum number of lines to read (default 2000)",            minimum=1,),        pages=StringSchema("Page range for PDF files, e.g. '1-5' (default: all, max 20 pages)"),        force=BooleanSchema(            description="Bypass same-file read deduplication and return content again.",            default=False,),        required=["path"],)    # 整体来讲，大模型就知道有这么一个函数定义了，且知道各个参数都是干什么的    # def xx(path: str, offset:int=?, limit:int=?, pages:str=?, force:bool=?)    # 这里写 =? 主要是表现这个参数不是必填参数    # 然后函数的名称、函数的作用则需要通过 name、description 来描述了)# ReadFileTool -> _FsTool -> Toolclass ReadFileTool(_FsTool):    _scopes = {"core", "subagent", "memory"}    # 略    @property    def name(self) -> str:        # 让大模型知道这个工具名称是 read_file        return "read_file"    @property    def description(self) -> str:        # 让大模型知道这个工具的用途是读文件        # 尽量写详细点，不然大模型判断起来可能会有失误        return (            "Read a file (text, image, or document). "            "Text output format: LINE_NUM|CONTENT. "            "Images return visual content for analysis. "            "Supports PDF, DOCX, XLSX, PPTX documents. "            "Use find_files/list_dir first when the path is uncertain. "            "Read the relevant range before editing so replacements or patches "            "are based on current content. "            "Use offset and limit for large text files. "            "Use force=true to re-read content even if unchanged. "            "Reads exceeding ~128K chars are truncated."        )    # 略    async def execute(        self,        path: str | None = None,        offset: int = 1,        limit: int | None = None,        pages: str | None = None,        force: bool = False,        **kwargs: Any,) -> Any:        # execute 签名必须要与 parameters 描述的一致        # 也可以看到，工具的默认值在这里放着（前面schema定义的默认值仅有描述意义，虽然在nanobot中也没有把默认值转成json给大模型）        # 具体实现略