从零实现 Harness Agent：设计多工具并发执行器

系列导航：系列目录 | 上一篇：从零实现 Harness Agent：实现安全的局部编辑工具 | 下一篇：从零实现 Harness Agent：构建 Skill 感知上下文引擎

本节目标

导读：本篇属于第二部分「工具与安全边界」，讨论工具调度：并发不是单纯性能优化，而是由工具语义决定的执行策略。

本节要实现的是 ToolExecutor 的多工具调度能力：当模型同一轮返回多个 tool call 时，系统可以并发执行安全的只读工具，同时保持有副作用工具顺序执行。

完成这一节后，系统会具备下面这些能力：

同一轮连续多个 read 调用可以进入并发执行组。
write、edit、bash 等副作用工具会保持顺序执行。
并发执行后的 tool observation 仍按模型原始 tool call 顺序返回。
单个工具失败不会吞掉同组其他工具结果。
CLI 或 Feishu 等 channel 可以继续收到工具调用和工具结果事件。

这一节的关键目标是把“并发”作为工具语义的一部分，而不是简单的性能开关。

摘要

现代模型可能在同一轮响应中返回多个 tool call。tiny-claw 的 ToolExecutor 将并行安全的 read 调用批量并发执行，同时保持 write、edit、bash 等副作用工具顺序执行。本文介绍这一执行策略如何在效率和安全之间取得平衡。

背景与问题

当模型需要理解一个项目时，它可能同时读取多个文件。例如同一轮请求中返回：

1
2
3

read pyproject.toml
read README.md
read src/tiny_claw/cli.py

如果全部串行执行，效率较低；如果全部并发执行，又可能让写文件、编辑文件、执行命令等副作用互相干扰。

工具执行器需要回答两个问题：

哪些工具可以并发？
并发后的 observation 顺序如何保持和模型 tool call 对齐？

ToolExecutor 采用保守策略：只把 read 视为并行安全，其余工具默认顺序执行。

设计目标

效率：多个连续 read 可以并发。
安全：副作用工具不并发，避免写冲突和命令竞争。
顺序稳定：返回 observation 的顺序与模型 tool call 顺序一致。
错误隔离：一个工具失败不影响同组其他工具 observation。
可观测：每个工具调用和结果都能通过 channel 和日志输出。
可扩展：后续可以引入工具级并发声明。

整体方案

ToolExecutor.run_tool_calls() 会扫描模型返回的 tool calls，把连续的并行安全工具收集成 group。遇到非并行安全工具时，先执行已有并行 group，再顺序执行当前工具。

flowchart TD
  A["tool calls"] --> B{"是否 read?"}
  B -->|是| C["加入 parallel group"]
  B -->|否| D["先执行 parallel group"]
  D --> E["顺序执行副作用工具"]
  C --> F{"遇到下一个非 read 或结束?"}
  F -->|是| G["ThreadPoolExecutor 并发执行 read group"]
  G --> H["按原顺序返回 observations"]
  E --> H

核心实现

核心文件是 src/tiny_claw/_internal/engine/tool_executor.py。

当前并行安全工具白名单只有 read：

1	PARALLEL_SAFE_TOOL_NAMES = {"read"}

执行入口：

def run_tool_calls(
    self,
    tool_calls: tuple[ToolCall, ...],
    *,
    channel: Channel | None = None,
) -> tuple[Message, ...]:

并行 group 使用 ThreadPoolExecutor，并限制最大 worker 数：

1
2
3

max_workers = min(self.max_parallel_tools, len(tool_calls))
with ThreadPoolExecutor(max_workers=max_workers) as executor:
    observations = tuple(executor.map(self._execute_one, tool_calls))

executor.map() 会按输入顺序产出结果，因此即使并发执行完成顺序不同，返回给模型的 observation 顺序仍然稳定。

对于单个工具调用，执行器会通知 channel：

1
2
3

notify_channel(partial(channel.on_tool_call, tool_call))
observation = self._execute_one(tool_call)
notify_channel(partial(channel.on_tool_result, call=tool_call, result=observation))

这让 CLI、Feishu 或未来其他外部通道可以展示进度。