AI Agent 工程师 Day 32：Tool Executor V2：执行流水线设计

2026-10-21

AI Agent 工程师 Day 32：Tool Executor V2：执行流水线设计

6 个月学习计划 | M2 - Agent Harness + Tool + Environment | Day 32/168

今日主题: Executor 重构

Tool Executor V2：执行流水线设计

开篇结论

Tool Executor 是 Agent 系统中「模型意图」与「真实世界副作用」之间的最后一道闸门。核心认知只有一条：Executor 的价值不在功能多，而在流水线（Pipeline）清晰、接口稳定。一个生产级 Executor 应该是六步流水线——校验、权限、执行、超时、结果归一化、追踪——但对外只暴露一个方法 executor.execute(tool_call, context)。不要拆成 5 个 Manager：Manager 拆分把一条有顺序依赖的因果链切碎，调用方需要理解编排逻辑，而流水线把顺序内聚在一个对象里，出错时也只有一个地方可查。本文将逐层拆解这条流水线的每个环节为什么必须存在、边界在哪、代码怎么落。

背景与动机

在一个 6 个月的 AI Agent 工程师成长路线中，第 32 天落在 M2 阶段的第 1 周。此前你已经完成了：

LLM 调用层：会跑通模型 API，理解 tool calling 协议
基础 Executor V1：一个「拿到 tool_call 就调函数」的朴素实现
若干 Tool 定义：函数注册、schema 描述

V1 的典型问题是：它只在「Happy Path」下工作。模型幻觉出一个不存在的工具名，V1 直接抛异常；LLM 生成的参数缺字段，V1 传进去报 TypeError；工具卡死 30 秒，V1 干等着；同一次调用在不同路径下返回 dict、字符串、异常三种形态，上层 Agent Loop 没法统一处理。

M2 阶段的主题是把 Agent 从「能跑」推向「可靠」。而 Executor 正是可靠性的第一个落点——因为它是所有副作用的必经之路。这里的投入产出比极高：在 Executor 这一个组件里建立防御，等于给所有 Tool 免费上了一层保险。这一周后续的 Tool Registry、错误恢复、沙箱执行，都建立在今天这条流水线之上。

核心内容

一、六步流水线：为什么是这六步（Pipeline Design）

生产级 Tool Executor 的执行流水线（Execution Pipeline）如下：

tool_call
   │
   ▼
┌──────────────┐   拒绝: ToolResult(success=False, error="unknown tool")
│ 1. Validate  │   校验工具存在 + 参数符合 schema
└──────┬───────┘
       ▼
┌──────────────┐   拒绝: ToolResult(success=False, error="permission denied")
│ 2. Permission│   权限/白名单/危险操作确认
└──────┬───────┘
       ▼
┌──────────────┐   异常/超时都被捕获
│ 3. Execute   │   真正调用工具函数
│ (+ 4.Timeout)│   超时是执行的保护壳，不是独立阶段
└──────┬───────┘
       ▼
┌──────────────┐   把任意返回值归一为统一 ToolResult
│ 5. Normalize │
└──────┬───────┘
       ▼
┌──────────────┐   写入 trace,无论成功失败
│ 6. Trace     │
└──────┬───────┘
       ▼
  ToolResult → Agent Loop

这六步不是拍脑袋列的清单，而是一条因果链：

Validate 先行：模型是概率系统，会产出不合法输入。在执行前用 schema 校验拦截，成本是微秒级；执行后才发现，可能是污染了一半的数据。左移校验（Shift-Left Validation）是所有工程系统的通用原则。
Permission 次之：参数合法不等于允许执行。rm -rf /tmp/x 是合法参数，但你可能不允许 Agent 执行删除操作。权限检查必须在副作用发生之前。
Execute + Timeout 合并理解：超时不是流水线的第五个独立站点，而是执行阶段的保护壳。为什么？因为「超时」必须能中断「执行」，两者天然耦合。分开成两个组件反而要处理「谁持有取消权」的问题。
Normalize 兜底：无论上面发生了什么——成功、异常、超时、拒绝——输出必须是同一个结构。这是整个设计里最重要的一条纪律。
Trace 收尾且必达：观测数据必须在所有路径上都被记录，包括被拒绝的调用。安全审计恰恰最需要那些「没执行成功」的记录。

一个关键的设计判断：前两步是同步、快速、无副作用的（纯函数判断），后三步才涉及真实执行。这条分界线意味着 Validate 和 Permission 失败时可以零成本短路返回，连超时计时器都不用启动。

二、统一 ToolResult：唯一的出口契约（Result Normalization）

结果归一化（Result Normalization）的核心是定义一个不可变的输出契约：

from dataclasses import dataclass, field
from typing import Any

@dataclass(frozen=True)
class ToolResult:
    success: bool
    output: Any | None = None        # 成功时的业务数据
    error: str | None = None         # 失败时的人类可读错误
    latency_ms: float = 0.0          # 执行耗时（含超时判定）
    metadata: dict = field(default_factory=dict)  # trace_id, tool_name, attempt 等

    @classmethod
    def failure(cls, error: str, **kw) -> "ToolResult":
        return cls(success=False, error=error, **kw)

四条字段的职责边界必须清晰：

字段	职责	典型内容	反模式
`success`	唯一的成功判定依据	`True/False`	用 `output is not None` 代替
`output`	成功时的返回值	查询结果、文件内容、结构化 dict	把错误信息塞进 output
`error`	失败时的诊断信息	`"timeout after 5s"`, `"param 'path' missing"`	失败时 error 为 None
`latency_ms`	性能观测	`123.4`	只在成功时记录

这里最容易被违反的纪律是：失败时 success=False 且 error 非 None，成功时两者相反，永远不要出现第三态。Agent Loop 拿到 ToolResult 后只需要一个 if result.success 分支就能决策——把错误信息反馈给模型重试，或终止循环。如果输出形态不确定，Agent Loop 就要写一堆防御性判断，而 LLM 拿到格式混乱的工具输出时，行为会显著退化（模型对结构化、一致的反馈的遵循度远高于对随意字符串的遵循度）。

metadata 是扩展位而非杂物箱：放 trace_id、tool_name、重试次数、被拒绝原因码等机器可读字段。原则是：error 给人和模型看，metadata 给下游程序看。

三、stdout / stderr / structured result / error / metadata 的处理

执行一个外部工具（比如 shell 命令或子进程工具）时，输出有五个可能的通道，它们必须被显式归位：

工具进程                    ToolResult
─────────                  ──────────
exit code = 0   ────────►  success = True
exit code != 0  ────────►  success = False
stdout          ────────►  output（主体内容）
stderr          ────────►  error 或 metadata["stderr"]（视 exit code 而定）
JSON 结构化输出  ────────►  output（解析后直接作为结构化 dict）
未捕获异常       ────────►  success = False, error = repr(e)
执行上下文信息   ────────►  metadata

关键决策点：

exit code 决定 success，stderr 不决定。很多工具往 stderr 打进度日志但正常退出，只看 stderr 会产生大量假失败。
stderr 的去向：exit code 非 0 时，stderr 是最有价值的错误线索，拼进 error；exit code 为 0 时，stderr 放进 metadata["stderr"] 备查，不污染主结果。
结构化输出优先。如果工具约定输出 JSON，优先 json.loads 后放入 output——喂给 LLM 的内容，结构化程度越高，模型利用得越好。解析失败则降级为原始文本，并在 metadata 标注 parse_failed: true。
output 大小截断。一个 cat 大文件命令可能产生 10MB stdout，直接塞进上下文会打爆 token 预算。归一化阶段应做截断（如保留头尾各 N KB 并加标记），这是很多新手 Agent 系统在生产环境踩的第一个坑。

四、Executor 接口为什么必须简单（Interface Stability）

Executor V2 的公开接口只有一句话：

1 2	class Executor(Protocol): def execute(self, tool_call: ToolCall, context: ExecutionContext) -> ToolResult: ...

一个方法、两个参数、一个返回值。为什么在内部做了六步流水线之后，对外反而要收得这么紧？

因为接口复杂度是会传染的。如果 Executor 暴露 validate()、check_permission()、run() 三个方法，那么每个调用方（Agent Loop、重试器、测试代码）都必须自己编排顺序、自己处理中间态。一旦某处编排漏了一步——比如重试时跳过了权限检查——就是一个安全漏洞。而单方法接口把编排锁死在 Executor 内部，调用方想绕都绕不过去。安全关键路径上，「不可能被误用」比「灵活」重要一个数量级。

class ToolExecutor:
    def __init__(self, registry, permission_checker,
                 default_timeout: float = 30.0, tracer=None):
        self.registry = registry
        self.permission = permission_checker
        self.default_timeout = default_timeout
        self.tracer = tracer

    def execute(self, tool_call, context) -> ToolResult:
        trace = {"tool": tool_call.name, "trace_id": context.trace_id}
        try:
            # 1. Validate
            tool = self.registry.get(tool_call.name)
            if tool is None:
                return self._fail(f"unknown tool: {tool_call.name}", trace)
            err = tool.validate_params(tool_call.arguments)
            if err:
                return self._fail(f"invalid arguments: {err}", trace)

            # 2. Permission
            if not self.permission.check(tool_call, context):
                return self._fail("permission denied", trace)

            # 3+4. Execute with timeout
            start = time.monotonic()
            raw = self._run_with_timeout(tool, tool_call, self.default_timeout)
            latency = (time.monotonic() - start) * 1000
            if isinstance(raw, TimeoutError):
                return self._fail(f"timeout after {self.default_timeout}s", trace)

            # 5. Normalize
            return self._normalize(raw, latency, trace)
        except Exception as e:
            return self._fail(f"executor error: {e!r}", trace)
        finally:
            # 6. Trace —— 永远执行
            if self.tracer:
                self.tracer.record(trace)

    def _fail(self, msg, trace):
        trace["success"] = False
        trace["error"] = msg
        return ToolResult.failure(msg)

注意三个实现细节：_run_with_timeout 内部用 concurrent.futures 或 asyncio.wait_for 实现（同步函数需要线程池包裹，因为 Python 没有强制杀死线程的手段）；最外层的 except Exception 是最后防线，保证任何意外异常也变成合法的 ToolResult；finally 保证 Trace 在所有路径上落盘。

五、为什么不拆成 5 个 Manager（Cohesion over Decomposition）

重构时最诱人的错误方向是：把六步拆成 ValidationManager、PermissionManager、TimeoutManager、ResultManager、TraceManager 五个类。表面上看是单一职责（SRP），实际是把一条强顺序耦合的因果链人为打碎了。对比一下：

维度	单 Executor + 内部流水线	5 个 Manager 拆分
编排责任	Executor 内部锁定	泄漏给每个调用方
漏调一步的风险	结构上不可能	完全依赖调用方自觉
出错排查	一个栈、一个文件	跨 5 个文件追踪调用链
新增中间步骤	改一个方法	新增 Manager + 改所有调用方
测试	对 `execute()` 做黑盒测试即可	需 mock 4 个依赖对象
复用性	通过组合注入（registry、permission 可替换）	形式上复用，实际更难拼装

澄清一个常见误解：不拆 Manager ≠ 不解耦。解耦的正确姿势是依赖注入——registry、permission_checker、tracer 都是构造时注入的可替换组件，流水线的骨架内聚在 Executor 里。SRP 的本意是「一个模块只对一个变化原因负责」，而这里的变化原因恰好只有一个：执行策略的变化。六步流水线就是一个内聚的整体策略。

经验法则：当拆分导致调用方需要理解执行顺序时，拆分就是错的。顺序即业务逻辑，顺序必须内聚。

实践练习与思考题

练习 1（基础）：实现 ToolExecutor，包含至少三个工具：read_file、web_search（可用 mock）、run_shell。要求：未知工具、参数缺失、权限拒绝、正常执行、人为超时五条路径全部返回合法 ToolResult。

练习 2（进阶）：给 run_shell 实现 stdout/stderr/exit code 的完整归一化，包括大输出截断（保留首尾各 2000 字符 + [truncated] 标记）和 JSON 输出的结构化解析。

练习 3（测试思维）：写一个属性测试（hypothesis 或穷举分支），断言 Executor 在任何输入下返回值都满足 ToolResult 契约——即 success 与 error 的互斥性、latency_ms >= 0、metadata 是 dict。

思考题：

Validate 和 Permission 的顺序能否调换？什么场景下先查权限更合理？（提示：想想权限规则本身也可能配置错误��以及「信息泄露」——让无权限的调用方通过参数校验错误探测到工具的存在）
重试逻辑应该放在 Executor 内部还是外层 Agent Loop？重试时 Trace 该怎么记？
超时后，被中断的工具的副作用（比如已经写了一半的文件）怎么处理？这在今天的流水线里没有答案，什么阶段该解决它？

与 Agent Engineering 的关联

在真实 Agent 系统中，这条流水线是多个核心机制的挂载点：

Self-Correction（自我纠错）：Agent Loop 拿到 success=False, error="invalid arguments: 'path' is required" 后，把错误信息拼回上下文让模型修正重试。错误信息的质量直接决定重试成功率——这就是为什么 error 要写得具体、面向模型可理解。
可观测性（Observability）：Trace 层记录每次调用的工具名、参数、耗时、成败，是调试「Agent 为什么做了那件事」的唯一依据，也是事后审计合规的基础。被权限拒绝的记录尤其重要。
安全边界（Safety Boundary）：Permission 阶段是 Agent 与危险操作之间唯一的强制关卡。LangChain 的工具白名单、OpenAI 的 function calling 权限控制，本质都是这条流水线的第 2 步。
成本控制：latency 数据积累后可以驱动工具选型（快工具优先）和超时预算分配；输出截断直接控制 token 成本。

一句话：Agent 的智能在模型，但 Agent 的可靠在 Executor。模型每轮可能产出千奇百怪的意图，Executor 的职责就是让这些意图以可预测的形状作用于世界。

FAQ

Q1：工具抛出的异常应该往上抛还是吞掉变成 ToolResult？
必须在 Executor 边界内捕获并转为 ToolResult。上抛意味着把异常处理责任推给每个调用方，且 Agent Loop 通常没有能力处理任意异常。唯一的例外是 Executor 自身的程序性 Bug（如配置错误），那类错误应该 fail-fast 而不是被伪装成「工具失败」。

Q2：异步工具（async def）怎么纳入这条流水线？
流水线结构不变，接口提供一个 async execute() 版本即可（或单独的 AsyncToolExecutor，共享归一化逻辑）。超时改用 asyncio.wait_for，比线程池方案更干净。不要为了统一而用线程池硬跑协程。

Q3：Trace 应该记录参数原文吗？参数里可能有敏感信息？
生产环境应记录参数但做脱敏（redact）：对标记为敏感的字段（密码、token）替换为 ***。完全不记参数会让 Trace 失去排障价值，全记则有合规风险。

Q4：六步会不会太重？简单工具也要走全流程吗？
会走，但每一步对简单工具的代价都是微秒级（字典查找、一次布尔判断）。流水线的价值恰恰在于对慢路径的防御不依赖开发者记得加。如果某步确实无意义（如纯只读工具跳过权限），用配置声明跳过，而不是写一个绕过流水线的快路径。

Q5：ToolResult 里放 retry 建议吗？比如「可重试」标记？
可以放进 metadata["retryable"]，由工具或归一化层判断（如网络超时可重试、参数错误不可重试）。但不建议放顶层字段——重试策略属于上层编排的决策，Executor 只提供事实。

参考资料

Anthropic — Building effective agents（工具执行与错误反馈设计的官方实践）：https://www.anthropic.com/research/building-effective-agents
OpenAI — Function Calling Guide（tool_call 协议与参数校验来源）：https://platform.openai.com/docs/guides/function-calling
LangChain — Tool Calling 文档（Tool 执行封装与异常处理的参考实现）：https://python.langchain.com/docs/concepts/tool_calling/
Python 官方文档 — concurrent.futures（超时实现的基础设施）：https://docs.python.org/3/library/concurrent.futures.html
Microsoft — Semantic Kernel Tool/Plugin 设计（企业级工具权限与流水线参考）：https://learn.microsoft.com/en-us/semantic-kernel/agents/

📋 本日知识点清单

Validate → Permission → Execute → Timeout → Result Normalize → Trace 的完整流水线
统一 ToolResult(success, output, error, latency_ms)
stdout/stderr/structured result/error/metadata 处理
Executor 接口保持简单：executor.execute(tool_call, context)
不拆成 5 个新 Manager

📝 实践练习

重构 Executor，实现 6 步流水线，保持接口简单。

本文是「AI Agent 工程师」6 个月学习计划的一部分。学习路线：LLM → Inference → Runtime → Harness → Memory → Environment → Evaluation → Production。目标：Agent / AI Systems Engineer。