第 7 章：工具的执行

发表于 更新于 分类于 阅读次数：
LLM 生成工具调用指令后的完整执行流程——function call 的解析与参数提取、ToolGuard 三层安全检查、实际函数调用与返回值回传给 LLM 的完整闭环。
个人公众号
1
2
3
Browser -> HTTP -> FastAPI -> Runner -> Agent -> Prompt -> ReAct -> LLM -> [Tool] -> Response
                                                                            ^
                                                                        you are here

上一章 LLM 返回了思考结果。如果结果里包含工具调用——“我需要调用 get_current_time“——ReAct 循环就进入”行动”阶段。这一章，我们跟着工具调用走一遍：LLM 的工具调用指令是怎么被解析的？安全检查在哪里发生？实际执行做了什么？结果怎么回到 LLM？

问题

LLM 说”调用 read_file，参数是 /etc/passwd“。谁来执行这个调用？如果 LLM 想删除文件怎么办？工具的返回值是怎么变成 LLM 能理解的消息的？

术语其实很简单

术语：Tool Call（工具调用）
想象你给助手一张便签——“帮我去隔壁拿份文件，文件名是 X”。工具调用就是 LLM 写的”便签”——里面写着工具名和参数。LLM 自己不会执行，它只是告诉 Agent 该调用什么。

术语：ToolResponse（工具响应）
助手拿着文件回来了——“这是你要的文件，内容是……”。工具响应就是工具执行后的返回值——告诉 LLM 执行结果是什么。

术语：Guardian（守卫）
想象银行柜台前的安检——不是每个操作都需要审批，但大额转账必须过安检。Guardian 就是工具执行的安检员——检查工具调用是否安全，不安全的拦截。

探索

从 LLM 输出到工具执行——六步流程

上一章 LLM 返回了包含 tool_use 块的响应。ReAct 循环对每个 tool_use 调用 _acting()：

1
2
3
4
5
6
7
8
9
LLM 返回: {"type": "tool_use", "name": "get_current_time", "input": {}}
  |
  v
1. ToolGuardMixin._acting()      # 安全检查
2. _decide_guard_action()          # 决策：放行/拦截/审批
3. 如果放行 -> ReActAgent._acting()  # 正常执行
4. Toolkit.call_tool_function()     # 查找并调用实际函数
5. 工具函数执行 -> ToolResponse     # 返回结果
6. 结果包装成 ToolResultBlock      # 存入记忆，供下次推理使用

第 1-2 步：ToolGuardMixin 的安全关卡

ToolGuardMixin._acting() 是每个工具执行前的安全关卡（第 5 章介绍过）。它的决策逻辑：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 伪代码
def _decide_guard_action(self, tool_call):
    tool_name = tool_call["name"]

    # 检查一：工具是否在黑名单？
    if engine.is_denied(tool_name):
        return "auto_denied"  # 直接拒绝，不可审批

    # 检查二：是否有一次性预审批令牌？
    if has_preapproval_token(tool_name):
        consume_token()
        return "preapproved"  # 用户刚才在 UI 上点过"允许"

    # 检查三：运行安全守卫检查
    result = engine.guard(tool_name, tool_input)

    if result.has_findings():
        return "needs_approval"  # 有风险，需要用户确认
    else:
        return None  # 放行，正常执行

三种结果对应三种处理：

  • auto_denied：返回拒绝消息，LLM 会看到”该操作被安全策略禁止”
  • needs_approval：记录待审批请求，返回等待消息。用户在 UI 确认后，下次 _reasoning() 会重放这个工具调用
  • 放行（None）：调用 super()._acting()，进入正常执行

第 3 步：Guardian——具体的安全检查

ToolGuardEngine 注册了多个 Guardian，每个负责一类安全检查：

1
2
3
4
ToolGuardEngine
  +-- FilePathToolGuardian      # 检查敏感文件路径（如 /etc/shadow）
  +-- RuleBasedToolGuardian      # 正则匹配危险模式
  +-- ShellEvasionGuardian       # 检测 Shell 命令混淆

默认受保护的工具有 8 个：execute_shell_command、read_file、write_file、edit_file、append_file、send_file_to_user、view_text_file、write_text_file。

ShellEvasionGuardian 是最有意思的——它检测 LLM 生成的 Shell 命令中是否包含混淆手段：命令替换 $()、反引号、标志位绕过、反斜杠转义等。LLM 有时会”聪明”地构造看起来无害但实际危险的命令。

第 4 步：Toolkit.call_tool_function——查找并调用

安全检查通过后，调用链到达 ReActAgent._acting()，它调用 Toolkit.call_tool_function()：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# agentscope Toolkit 核心逻辑
async def call_tool_function(self, tool_call):
    # 1. 查找注册的工具函数
    tool_func = self.tools[tool_call["name"]]

    # 2. 合并参数：预设参数 + LLM 提供的参数
    kwargs = {**tool_func.preset_kwargs, **tool_call.get("input", {})}

    # 3. 调用实际函数
    if inspect.iscoroutinefunction(tool_func.original_func):
        result = await tool_func.original_func(**kwargs)
    else:
        result = tool_func.original_func(**kwargs)

    return result  # ToolResponse

查找是 O(1) 的字典查找。参数合并让预设参数（如工作目录）和 LLM 提供的参数无缝结合。

第 5 步：具体工具函数长什么样？

所有工具函数遵循统一的接口约定：异步函数，接收类型化参数，返回 ToolResponse。

get_current_time（src/qwenpaw/agents/tools/get_current_time.py）：

1
2
3
4
5
6
async def get_current_time() -> ToolResponse:
    """获取当前时间"""
    timezone = 从配置读取用户时区
    now = datetime.now(timezone)
    formatted = now.strftime("%Y-%m-%d %H:%M:%S %Z")
    return ToolResponse(content=[TextBlock(type="text", text=formatted)])

最简单的工具——没有参数，读配置，格式化时间，返回文本。

read_file（src/qwenpaw/agents/tools/file_io.py）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
async def read_file(file_path, start_line=None, end_line=None) -> ToolResponse:
    # 1. 路径解析：相对路径基于工作区目录
    resolved = _resolve_file_path(file_path)

    # 2. 安全读取（多种编码回退）
    content = await read_file_safe(resolved)

    # 3. 按行范围截取
    if start_line or end_line:
        content = 按行切片(content, start_line, end_line)

    # 4. 输出截断（默认 50KB）
    content, truncated = truncate_text_output(content)

    return ToolResponse(content=[TextBlock(type="text", text=content)])

文件操作工具的关键设计：输出截断。truncate_text_output() 在字节级别截断，保持行完整性，截断后附加提示告诉 LLM 用 start_line 参数读取剩余内容。这样 LLM 读大文件时不会撑爆上下文窗口。

execute_shell_command（src/qwenpaw/agents/tools/shell.py）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
async def execute_shell_command(command, timeout=60.0, cwd=None) -> ToolResponse:
    # 1. 预处理：处理 LLM JSON 编码产生的转义字符
    command = _collapse_embedded_newlines(command)

    # 2. 解析工作目录
    cwd = cwd or 从工作区上下文获取

    # 3. 创建子进程（独立进程组）
    proc = await asyncio.create_subprocess_shell(
        command, cwd=cwd, start_new_session=True, ...
    )

    # 4. 带超时等待
    try:
        stdout, stderr = await asyncio.wait_for(
            proc.communicate(), timeout=timeout
        )
    except asyncio.TimeoutError:
        # 杀掉整个进程组
        os.killpg(os.getpgid(proc.pid), signal.SIGTERM)

    return ToolResponse(content=[TextBlock(type="text", text=输出)])

Shell 命令执行的几个关键细节：

  • start_new_session=True 创建独立进程组——超时时可以杀掉整个进程组（包括子进程）
  • _collapse_embedded_newlines() 处理 LLM 在 JSON 里编码换行符的问题
  • 默认 60 秒超时，超时后先 SIGTERM，不行再 SIGKILL

第 6 步：工具结果如何回到 LLM

工具执行完毕后，结果被包装成 ToolResultBlock 存入 Agent 记忆：

1
2
3
4
5
6
7
8
9
10
11
# ReActAgent._acting() 中的关键步骤
result_msg = Msg(
    role="system",
    content=[ToolResultBlock(
        type="tool_result",
        id=tool_call["id"],     # 与 LLM 的 tool_use id 对应
        name=tool_call["name"], # 工具名
        output=chunk.content,   # 工具返回的内容
    )]
)
self.memory.add(result_msg)  # 存入记忆

下一次 _reasoning() 调用时，LLM 会看到完整的历史：用户消息 → 助手说”我需要调用 X 工具” → 工具返回了 Y 结果 → 现在基于这些信息继续推理。

工具注册表——18 个内置工具

第 3 章提到 _create_toolkit() 注册了 18 个内置工具。它们按类别分布：

1
2
3
4
5
6
7
8
文件操作: read_file, write_file, edit_file, append_file
搜索:    grep_search, glob_search
Shell:   execute_shell_command
浏览器:  browser_use, desktop_screenshot
多媒体:  view_image, view_video, send_file_to_user
时间:    get_current_time, set_user_timezone
Agent:   delegate_external_agent, list_agents, chat_with_agent, ...
监控:    get_token_usage

注册过程在 _create_toolkit() 中，用一个字典映射字符串到函数对象：

1
2
3
4
5
6
tool_functions = {
    "read_file": read_file,
    "write_file": write_file,
    "execute_shell_command": execute_shell_command,
    ...
}

每个工具函数的 docstring 和类型注解被自动解析成 JSON Schema，作为 tools 参数传给 LLM——LLM 据此知道有哪些工具可用、每个工具接受什么参数。

实验

在源码中找到工具注册和执行的关键位置：

  1. 打开 src/qwenpaw/agents/react_agent.py，搜索 _create_toolkit
  2. 打开 src/qwenpaw/agents/tools/get_current_time.py，看最简单的工具实现
  3. 打开 src/qwenpaw/agents/tools/shell.py，看最复杂的工具实现
  4. 打开 src/qwenpaw/agents/tool_guard_mixin.py，搜索 _decide_guard_action

预期结果：能看到工具注册表、工具函数结构、安全检查决策逻辑。

工程权衡

为什么用 docstring + 类型注解生成 JSON Schema？

每个工具的 JSON Schema（告诉 LLM 这个工具接受什么参数）是从函数的 docstring 和类型注解自动生成的。如果手写 Schema，改函数签名时容易忘改 Schema，导致 LLM 用错误的参数调用。自动生成保证 Schema 和代码始终一致。

为什么文件操作要截断输出？

LLM 的上下文窗口有限（通常几十万 Token）。如果一个 10MB 的日志文件被完整读入，会撑爆上下文，Agent 就无法继续思考。50KB 的默认截断限制确保工具返回的内容在可控范围内。截断后附加”用 start_line=200 继续”的提示，LLM 能按需读取剩余内容。

为什么 Shell 命令要独立进程组？

如果 LLM 执行了 find / -name "*.log" 这样的命令，它会生成大量输出且可能运行很久。用独立进程组（start_new_session=True），超时时可以用 os.killpg() 一次杀掉父进程和所有子进程。如果在同一进程组里，子进程（如管道命令）可能成为孤儿进程继续运行。

常见误区

误区：工具执行是同步的，一个接一个？

LLM 可以在一次响应中请求多个工具调用。agentscope 的 ReActAgent 支持 parallel_tool_calls=True——多个独立的工具调用通过 asyncio.gather() 并行执行。但安全检查（_decide_guard_action）加了 _tool_guard_lock，确保并行的工具调用不会同时绕过安全检查。

误区：工具返回的结果直接显示给用户？

不是。工具返回的结果存入 Agent 记忆，供 LLM 下一次推理时使用。用户看到的是 LLM 基于工具结果的最终文字回复。比如 get_current_time 返回 “2026-05-11 10:30:00”，LLM 看到这个结果后生成”现在是 2026 年 5 月 11 日上午 10 点 30 分”——用户看到的是后者。

动手环节

任务：阅读一个简单工具和一个复杂工具的实现，理解工具函数的结构模式。

步骤：

  1. 打开 src/qwenpaw/agents/tools/get_current_time.py，阅读 get_current_time() 函数
  2. 打开 src/qwenpaw/agents/tools/shell.py，搜索 execute_shell_command() 函数
  3. 打开 src/qwenpaw/agents/tools/utils.py，搜索 truncate_text_output() 函数
  4. 打开 src/qwenpaw/security/tool_guard/engine.py，搜索 guard() 方法

预期输出：

  • get_current_time() 只有几行，是最简单的工具
  • execute_shell_command() 有几十行，处理了进程组、超时、编码等问题
  • truncate_text_output() 实现了字节级截断，保持行完整性

自检：

  • 理解了工具函数的统一结构：异步函数 + 类型参数 + 返回 ToolResponse
  • 知道 ToolGuardMixin 在 _acting() 中做安全检查
  • 知道工具结果存入记忆供下次推理使用

工具执行完毕，结果回到了 Agent 记忆中。下一步是 ReAct 循环的最后一轮——LLM 基于工具结果生成最终文字回复。这个回复怎么从 Agent 的内存流到浏览器的屏幕上？下一章我们走”响应的归途”。