通过 Nanobot 源码学习架构---（8）Tools

可以使用一个退出条件控制整个流程。循环持续运行, 直到模型不再调用工具。-------- ------- --------- | User | --- | LLM | --- | Tool | | prompt | | | | execute | -------- ------ -------- ^ | | tool_result | ---------------- (loop until stop_reason ! tool_use)我们可以进一步完善使得一次分发就可以调用到正确的工具加工具不需要改循环。-------- ------- ------------------ | User | --- | LLM | --- | Tool Dispatch | | prompt | | | | { | -------- ------ | bash: run_bash | ^ | read: run_read | | | write: run_wr | ----------- edit: run_edit | tool_result | } | ------------------ The dispatch map is a dict: {tool_name: handler_function}. One lookup replaces any if/elif chain.1.3 执行流程Tools 调用不是简单的一问一答而是决策-执行-再推理的协作闭环是一个多轮交互的闭环过程。模型负责动脑应用负责动手两者通过结构化消息JSON 指令与自然语言完成多轮协同。第一次模型调用应用程序首先向大模型发起请求请求体包含两个核心要素用户问题User Query当前需要解决的任务工具清单Available Tools模型可调用的函数列表及其 Schema 定义这相当于给模型一本操作手册——告诉它有哪些能力可用但暂不执行。接收模型的工具调用决策模型接收到请求后会进行意图判断与工具选择场景模型响应后续动作需要外部工具返回JSON 格式的工具调用指令包含工具名称与入参进入步骤 3执行工具无需外部工具返回自然语言格式的直接回复流程结束直接呈现给用户模型在此阶段扮演决策者角色它只生成调用指令并不实际执行工具。在应用端执行工具应用程序接收到 JSON 指令后进入工具执行层解析指令中的工具名称与参数在应用端或沙箱环境运行对应工具捕获工具输出结果成功返回数据或失败返回错误信息第二次模型调用获取工具输出后应用程序需要将结果回注上下文再次发起模型调用。此时的消息序列Messages结构为[用户问题] → [模型工具调用指令] → [工具执行结果]这相当于告诉模型你刚才要求执行的操作已完成这是结果请基于这些信息继续推理。接收模型的最终响应模型整合以下信息进行最终推理原始用户问题工具输出结果中间推理过程Chain-of-Thought最终生成自然语言格式的回复直接呈现给用户完成整个交互闭环。流程本质总结轮次角色动作第一轮模型决策是否需要工具 规划调用哪个工具应用端应用执行实际运行工具第二轮模型整合基于结果生成回复我们从 GitHub - shareAI-lab/claw0: 0 - 1 learn OpenClaw: sections to build an claw-AI agent from scratch · GitHub 的文档中也可以窥见端倪。工具 数据 (schema) 处理函数映射表. 模型选一个名字, 你查表执行.TOOLS: JSON schema 字典列表, 告诉模型有哪些工具可用.TOOL_HANDLERS:dict[str, Callable], 将工具名映射到 Python 函数.process_tool_call(): 字典查找 **kwargs分发.内层循环: 模型可能连续调用多个工具, 然后才生成文本.工具结果放在 user 消息中(Anthropic API 的要求).User Input | v messages[] -- LLM API (toolsTOOLS) | stop_reason? / \ end_turn tool_use | | Print for each tool_use block: TOOL_HANDLERS[name](**input) | tool_result | messages[] -- {role:user, content:[tool_result]} | back to LLM -- may chain more tools or end_turn -- Print0x02 核心代码作用与特色总结Nanobot Agent 工具系统的核心实现包括Tool抽象基类定义了 Agent 工具的标准化接口规范名称、描述、参数 Schema、执行逻辑提供通用的参数校验能力基于 JSON Schema和 OpenAI 函数 Schema 转换能力是所有自定义工具的 “模板”保证了工具体系的一致性和可扩展性。所有工具只需实现指定抽象方法即可接入 Agent无需修改核心逻辑同时基类内置通用能力参数校验、Schema 转换减少重复开发。ExecTool工具类基于Tool基类实现的具体工具为 Agent 提供安全可控的 Shell 命令执行能力支持超时控制、工作目录限制、危险命令拦截、路径遍历防护、工作目录限制、允许列表等安全机制既满足 Agent 与系统交互的核心需求又规避了 Shell 执行的典型安全风险。具体如下图所示。0x03 实现3.1 整体逻辑关系图架构图如下工具调用流程如下3.2 TOOLS.mdTOOLS.md 是本地工具提示。脚本存放在哪里哪些命令可用。这样 Agent 就不需要去猜而是确切知道。AGENTS.md 定义行为流程TOOLS.md 定义能力边界。简单说它是智能体的工具箱说明书告诉智能体可以使用哪些工具、怎么用、什么时候用。# Tool Usage Notes Tool signatures are provided automatically via function calling. This file documents non-obvious constraints and usage patterns. ## exec — Safety Limits - Commands have a configurable timeout (default 60s) - Dangerous commands are blocked (rm -rf, format, dd, shutdown, etc.) - Output is truncated at 10,000 characters - restrictToWorkspace config can limit file access to the workspace ## cron — Scheduled Reminders - Please refer to cron skill for usage.3.3 Tool 抽象基类Tool抽象基类是工具体系的基础骨架。3.3.1 Tool 基类核心流程参数校验validate_params() 是工具参数校验入口方法校验传入参数是否符合当前工具的Schema规范。Schema 转换to_schema()是工具Schema转换方法将当前工具转换为OpenAI函数调用的Schema格式这样可以让LLM能识别工具的调用格式兼容OpenAI Function Call生态。3.3.2 代码class Tool(ABC): Abstract base class for agent tools. Tools are capabilities that the agent can use to interact with the environment, such as reading files, executing commands, etc. # 定义JSON Schema类型到Python原生类型的映射字典用于参数类型校验 # 核心作用将Schema中定义的类型如string转换为Python可识别的类型str方便后续类型检查 _TYPE_MAP { string: str, # Schema字符串类型对应Python str integer: int, # Schema整数类型对应Python int number: (int, float), # Schema数字类型对应Python int/float兼容整数和浮点数 boolean: bool, # Schema布尔类型对应Python bool array: list, # Schema数组类型对应Python list object: dict, # Schema对象类型对应Python dict } # 定义抽象属性工具名称必须由子类实现 # 作用指定工具在函数调用中的唯一标识如execLLM通过该名称调用对应工具 property abstractmethod def name(self) - str: Tool name used in function calls. pass # 定义抽象属性工具描述必须由子类实现 # 作用向LLM说明工具的功能帮助LLM判断何时调用该工具描述需清晰易懂 property abstractmethod def description(self) - str: Description of what the tool does. pass # 定义抽象属性工具参数Schema必须由子类实现 # 作用定义工具入参的JSON Schema规范用于参数校验和向LLM声明参数格式 property abstractmethod def parameters(self) - dict[str, Any]: JSON Schema for tool parameters. pass # 定义抽象方法工具执行逻辑必须由子类实现 # 作用实现工具的核心功能接收参数并返回执行结果async标识异步执行适配Agent异步架构 abstractmethod async def execute(self, **kwargs: Any) - str: Execute the tool with given parameters. Args: **kwargs: Tool-specific parameters. Returns: String result of the tool execution. pass # 工具参数校验入口方法校验传入参数是否符合当前工具的Schema规范 # 返回值校验错误信息列表空列表表示校验通过 def validate_params(self, params: dict[str, Any]) - list[str]: Validate tool parameters against JSON schema. Returns error list (empty if valid). # 获取当前工具的参数Schema若未定义则默认为空字典 schema self.parameters or {} # 校验Schema的顶层类型必须是object因为参数本质是键值对否则抛出异常 if schema.get(type, object) ! object: raise ValueError(fSchema must be object type, got {schema.get(type)!r}) # 调用内部递归校验方法传入待校验参数、完整Schema、空路径用于定位错误参数 return self._validate(params, {**schema, type: object}, ) # 内部递归校验方法核心参数校验逻辑支持嵌套类型如对象、数组的递归校验 # val待校验的参数值schema当前层级的Schema规范path参数的路径用于精准定位错误如command.working_dir def _validate(self, val: Any, schema: dict[str, Any], path: str) - list[str]: # 获取当前Schema定义的类型和参数路径标签用于错误提示 t, label schema.get(type), path or parameter # 第一步基础类型校验——若Schema类型在映射表中检查参数值是否为对应Python类型 if t in self._TYPE_MAP and not isinstance(val, self._TYPE_MAP[t]): return [f{label} should be {t}] # 初始化错误列表用于收集所有校验错误 errors [] # 第二步枚举值校验——若Schema定义了enum检查参数值是否在枚举列表中 if enum in schema and val not in schema[enum]: errors.append(f{label} must be one of {schema[enum]}) # 第三步数值类型整数/数字范围校验 if t in (integer, number): # 最小值校验若定义了minimum检查参数值是否大于等于最小值 if minimum in schema and val schema[minimum]: errors.append(f{label} must be {schema[minimum]}) # 最大值校验若定义了maximum检查参数值是否小于等于最大值 if maximum in schema and val schema[maximum]: errors.append(f{label} must be {schema[maximum]}) # 第四步字符串类型长度校验 if t string: # 最小长度校验若定义了minLength检查字符串长度是否达标 if minLength in schema and len(val) schema[minLength]: errors.append(f{label} must be at least {schema[minLength]} chars) # 最大长度校验若定义了maxLength检查字符串长度是否超限 if maxLength in schema and len(val) schema[maxLength]: errors.append(f{label} must be at most {schema[maxLength]} chars) # 第五步对象类型校验支持嵌套对象 if t object: # 获取对象的属性定义和必填属性列表 props schema.get(properties, {}) # 必填属性校验检查所有必填属性是否存在于参数中 for k in schema.get(required, []): if k not in val: errors.append(fmissing required {path . k if path else k}) # 递归校验对象的每个属性值若属性在Schema中定义则递归校验其值 for k, v in val.items(): if k in props: errors.extend(self._validate(v, props[k], path . k if path else k)) # 第六步数组类型校验支持数组元素的递归校验 if t array and items in schema: # 遍历数组每个元素递归校验元素是否符合items定义的Schema for i, item in enumerate(val): errors.extend(self._validate(item, schema[items], f{path}[{i}] if path else f[{i}])) # 返回所有校验错误 return errors # 工具Schema转换方法将当前工具转换为OpenAI函数调用的Schema格式 # 作用让LLM能识别工具的调用格式兼容OpenAI Function Call生态 def to_schema(self) - dict[str, Any]: Convert tool to OpenAI function schema format. return { type: function, function: { name: self.name, # 工具名称对应OpenAI函数名 description: self.description, # 工具描述帮助LLM理解工具用途 parameters: self.parameters, # 工具参数Schema声明入参格式 } }3.4 ExecTool Shell 执行工具ExecTool是基于 Tool 基类的具体实现。3.4.1 Agent 如何运行 grep 命令LLM 通过结合系统提示中的工具功能描述执行 shell 命令、用户请求的意图分析搜索文件内容以及上下文信息来决定使用 exec 工具来执行 grep 命令Agent 接收用户请求 search for keyword in historyLLM 分析用户请求识别出需要执行 shell 命令的意图当检测到类似 search for, find in history, look up 等请求时并且上下文中提到 memory/HISTORY.md 文件时LLM 识别需要执行 grep 命令根据常识LLM 构造适当的参数grep -i keyword memory/HISTORY.mdLLM 会选择 ExecTool 作为执行工具根据工具描述和功能决策流程如下用户请求 ─ 包含搜索/查找意图 ─ 是 ─ 涉及文件操作 ─ 是 → 使用 exec 工具执行 grep ─ 否 → 继续判断 ─ 否 → 判断其他工具需求3.4.2 ExecTool 核心流程Shell命令执行流程如下3.4.3 代码Shell execution tool. class ExecTool(Tool): Tool to execute shell commands. # 工具初始化方法配置Shell执行的安全参数和运行参数 # 参数说明 # - timeout命令执行超时时间默认60秒 # - working_dir默认工作目录若未指定则使用当前目录 # - deny_patterns危险命令正则黑名单默认内置常见危险命令 # - allow_patterns命令允许列表白名单为空则不启用 # - restrict_to_workspace是否限制命令仅能访问指定工作目录防止越权访问 # - path_append追加到环境变量PATH的路径用于指定命令查找路径 def __init__( self, timeout: int 60, working_dir: str | None None, deny_patterns: list[str] | None None, allow_patterns: list[str] | None None, restrict_to_workspace: bool False, path_append: str , ): self.timeout timeout # 初始化命令超时时间 self.working_dir working_dir # 初始化默认工作目录 # 初始化危险命令黑名单默认拦截删除、格式化、系统关机、fork炸弹等破坏性命令 self.deny_patterns deny_patterns or [ r\brm\s-[rf]{1,2}\b, # 匹配rm -r/rm -rf/rm -fr递归删除文件 r\bdel\s/[fq]\b, # 匹配del /f/del /q强制/静默删除文件Windows r\brmdir\s/s\b, # 匹配rmdir /s删除目录及子目录Windows r(?:^|[;|]\s*)format\b, # 匹配format命令格式化磁盘仅匹配独立命令 r\b(mkfs|diskpart)\b, # 匹配磁盘操作命令mkfs格式化文件系统、diskpart磁盘分区 r\bdd\sif, # 匹配dd命令磁盘写入if指定输入文件 r\s*/dev/sd, # 匹配写入磁盘设备如 /dev/sda破坏性操作 r\b(shutdown|reboot|poweroff)\b, # 匹配系统关机/重启/断电命令 r:\(\)\s*\{.*\};\s*:, # 匹配fork炸弹无限创建进程导致系统崩溃 ] self.allow_patterns allow_patterns or [] # 初始化命令允许列表默认空不启用 self.restrict_to_workspace restrict_to_workspace # 初始化工作目录限制开关 self.path_append path_append # 初始化PATH环境变量追加路径 # 实现抽象属性工具名称固定为execLLM通过该名称调用 property def name(self) - str: return exec # 实现抽象属性工具描述说明工具功能并提醒谨慎使用 property def description(self) - str: return Execute a shell command and return its output. Use with caution. # 实现抽象属性工具参数Schema定义exec工具的入参规范 property def parameters(self) - dict[str, Any]: return { type: object, # 顶层类型为对象键值对 properties: { # 命令参数必填字符串类型描述要执行的Shell命令 command: { type: string, description: The shell command to execute }, # 工作目录参数可选字符串类型描述命令执行的工作目录 working_dir: { type: string, description: Optional working directory for the command } }, required: [command] # 声明command为必填参数 } # 实现抽象方法工具核心执行逻辑异步执行Shell命令 # 参数command要执行的命令、working_dir临时工作目录、其他扩展参数 async def execute(self, command: str, working_dir: str | None None, **kwargs: Any) - str: # 确定命令执行的工作目录优先使用传入的working_dir其次是工具默认的working_dir最后是当前目录 cwd working_dir or self.working_dir or os.getcwd() # 执行命令安全校验检查命令是否包含危险内容、是否越权访问路径等 guard_error self._guard_command(command, cwd) # 若安全校验失败直接返回错误信息不执行命令 if guard_error: return guard_error # 复制当前环境变量避免修改全局环境变量 env os.environ.copy() # 若配置了PATH追加路径将其添加到环境变量PATH中 if self.path_append: env[PATH] env.get(PATH, ) os.pathsep self.path_append try: # 异步创建Shell子进程执行命令 # - stdout/stderr重定向到管道用于捕获输出 # - cwd指定工作目录 # - env指定环境变量 process await asyncio.create_subprocess_shell( command, stdoutasyncio.subprocess.PIPE, stderrasyncio.subprocess.PIPE, cwdcwd, envenv, ) try: # 等待命令执行完成并捕获输出设置超时时间防止命令挂起 stdout, stderr await asyncio.wait_for( process.communicate(), timeoutself.timeout ) except asyncio.TimeoutError: # 命令超时终止进程 process.kill() # 等待进程完全终止释放文件描述符等资源 try: await asyncio.wait_for(process.wait(), timeout5.0) except asyncio.TimeoutError: pass # 返回超时错误信息 return fError: Command timed out after {self.timeout} seconds # 初始化输出列表用于拼接标准输出、标准错误、退出码 output_parts [] # 若有标准输出解码为字符串utf-8无法解码的字符替换并添加到输出列表 if stdout: output_parts.append(stdout.decode(utf-8, errorsreplace)) # 若有标准错误解码后添加到输出列表标注STDERR if stderr: stderr_text stderr.decode(utf-8, errorsreplace) if stderr_text.strip(): # 仅当标准错误非空时添加 output_parts.append(fSTDERR:\n{stderr_text}) # 若命令退出码非0执行失败添加退出码信息到输出列表 if process.returncode ! 0: output_parts.append(f\nExit code: {process.returncode}) # 拼接所有输出部分若无输出则返回(no output) result \n.join(output_parts) if output_parts else (no output) # 截断超长输出限制最大长度为10000字符避免返回结果过大 max_len 10000 if len(result) max_len: result result[:max_len] f\n... (truncated, {len(result) - max_len} more chars) # 返回命令执行结果 return result except Exception as e: # 捕获其他执行异常如进程创建失败返回错误信息 return fError executing command: {str(e)} # 命令安全防护方法校验命令是否安全返回错误信息None表示安全 # 核心作用拦截危险命令、越权路径访问是ExecTool的核心安全机制 def _guard_command(self, command: str, cwd: str) - str | None: Best-effort safety guard for potentially destructive commands. # 去除命令首尾空格便于正则匹配 cmd command.strip() # 转换为小写正则匹配不区分大小写 lower cmd.lower() # 第一步黑名单校验——检查命令是否匹配危险模式 for pattern in self.deny_patterns: if re.search(pattern, lower): return Error: Command blocked by safety guard (dangerous pattern detected) # 第二步白名单校验——若配置了允许列表命令必须匹配其中一个模式 if self.allow_patterns: if not any(re.search(p, lower) for p in self.allow_patterns): return Error: Command blocked by safety guard (not in allowlist) # 第三步工作目录限制校验——若开启限制拦截路径遍历和越权访问 if self.restrict_to_workspace: # 拦截路径遍历字符../ 或 ..\防止访问上级目录 if ..\\ in cmd or ../ in cmd: return Error: Command blocked by safety guard (path traversal detected) # 解析工作目录的绝对路径用于后续路径校验 cwd_path Path(cwd).resolve() # 提取命令中的Windows绝对路径如C:\Users\test win_paths re.findall(r[A-Za-z]:\\[^\\\], cmd) # 提取命令中的POSIX绝对路径如/usr/bin仅匹配独立的绝对路径避免误匹配 posix_paths re.findall(r(?:^|[\s|])(/[^\s\]), cmd) # 遍历所有提取的绝对路径检查是否超出工作目录范围 for raw in win_paths posix_paths: try: # 解析路径为绝对路径 p Path(raw.strip()).resolve() except Exception: # 路径解析失败如非法路径跳过校验 continue # 若路径是绝对路径且不在工作目录及其子目录中拦截命令 if p.is_absolute() and cwd_path not in p.parents and p ! cwd_path: return Error: Command blocked by safety guard (path outside working dir) # 所有校验通过返回None命令安全 return None3.5 沙箱3.5.1 两层实现nanobot的沙箱实现分两层第一层命令守卫(软防护)。ExecTool._guard_command()在执行前用正则表达式检查命令deny_patterns默认屏蔽rm-rf、dd、format、shutdown、forkbomb等危险模式allow_patterns可选白名单只有匹配的命令才放行restrict_to_workspace若开启绝对路径必须在workspace目录内(防止路径逃逸)内网URL检测调用security.network.contains_internal_url() 屏蔽访问内网地址保护内部文件屏蔽直接写入history.jsonl/.dream_cursor(防止LLM篡改记忆)第二层bubblewrap(硬隔离Linux 容器)。sandbox.py的 _bwrap()把命令包裹在bwrap 沙箱里只读挂载/usr、/bin、系统库用tmpfs隐藏workspace 的父目录(config.json 所在目录)workspace读写挂载media目录只读挂载新进程组(--new-session)、进程死亡传播(--die-with-parent)环境变量隔离_build_env()只传递最小环境(HOME/LANG/TERM)屏蔽所有API key等敏感变量(allowed_env_keys 白名单除外)。3.5.2 流程我们加入 沙箱之后重新走以下完整的运行流程从接受到用户消息开始一直到最后。3.5.3 代码bwrap(bubblewrap)是Linux的非特权容器工具nanobot的_bwrap()函数用它构造如下命令bubblewrap的实现如下def _bwrap(command: str, workspace: str, cwd: str) - str: 将命令包裹在 bubblewrap 沙箱中执行 (需要容器内存在 bwrap)。 只有 workspace 目录以读写方式挂载其父目录(存放 config.json 的位置) 被一个全新的 tmpfs 覆盖隐藏。media 目录以只读方式挂载供 exec 命令 读取用户上传的附件。 安全属性: - config.json (含 API key) 在沙箱内不可见 - workspace 的父目录整体被替换为空 tmpfs。 - 文件系统写操作被限制在 workspace 内其他路径均为只读或不可见。 - 进程运行在新会话 (--new-session) 中外部 TTY 的信号无法传入 父进程退出时沙箱进程自动被回收 (--die-with-parent)防止孤儿进程。 ws Path(workspace).resolve() media get_media_dir().resolve() # 确保 cwd 始终在沙箱内。如果传入的路径已逃出 workspace 根目录 # (例如传了绝对路径)则回退到 workspace 根防止 bwrap 收到 # 沙箱外的 --chdir 参数。 try: sandbox_cwd str(ws / Path(cwd).resolve().relative_to(ws)) except ValueError: sandbox_cwd str(ws) # 必须存在于宿主机的路径 (bind 挂载失败会报错)。 required [/usr] # 可选路径--ro-bind-try 在路径不存在时静默跳过。 optional [/bin, /lib, /lib64, /etc/alternatives, /etc/ssl/certs, /etc/resolv.conf, /etc/ld.so.cache] args [ bwrap, --new-session, # 脱离调用方的 TTY/会话防止 SIGHUP、SIGINT # 等信号从宿主机泄漏到沙箱进程。 --die-with-parent, # nanobot 退出时自动回收沙箱进程, 防止孤儿进程残留。 ] # 挂载运行 shell 和常用 CLI 工具所需的最小只读系统目录树。 for p in required: args [--ro-bind, p, p] for p in optional: args [--ro-bind-try, p, p] # 路径不存在时静默跳过。 args [ --proc, /proc, # 许多工具 (ps、top、/proc/self/...) 依赖 /proc。 --dev, /dev, # /dev/null、/dev/urandom 等设备节点所必需。 --tmpfs, /tmp, # 隔离的 /tmp, 与宿主机不共享。 # 将 workspace 的*父目录*替换为空 tmpfs。 # 这会隐藏 config.json (含 API key) 以及同级目录 # (如 ~/.nanobot/sessions、~/.nanobot/media 的上层), # 使沙箱内进程无法访问这些文件。 --tmpfs, str(ws.parent), # 在刚刚创建的空 tmpfs 下重建 workspace 挂载点, # 再将真实的 workspace 目录以读写方式绑定挂载进去。 # LLM 可以在 workspace 内自由读写文件。 --dir, str(ws), --bind, str(ws), str(ws), # 将 media 上传目录以只读方式暴露给沙箱, # 使 agent 可以读取用户附件, 但无法修改或删除。 --ro-bind-try, str(media), str(media), --chdir, sandbox_cwd, # 在正确的工作目录中启动命令。 --, sh, -c, command, ] return shlex.join(args)3.6 并行 vs 串行nanoboot 中工具默认是串行 (sequential)执行。并行是可选特性, 且受工具属性限制。即使开启, 也不是所有工具一起跑而是按批次(batch)执行批次内并行批次间串行。3.6.1 并行执行规则并行执行规则 (_partition_tool_batches)如下