从零构建 Agent Harness：工具校验、工作区隔离与执行预算

大多数 AI agent 教程的起点是：接一个语言模型，塞给它几个工具，然后循环到它给出答案为止。这足以演示工具调用，但把真正有意思的工程问题都跳过了——工具失败时怎么办？谁校验参数？什么机制阻止 agent 无限跑下去、执行危险命令或把上下文窗口撑爆？又怎么在不依赖模型随机输出的前提下测试这些行为？

这些正是 agent harness 的职责：它夹在语言模型和模型所操作的环境之间，负责把模型生成的「意图」变成受控的「执行」。本文用 TypeScript 从零搭一个与模型供应商无关的小型 harness，以编码 agent 为例，先写最简执行循环，再逐层加上工具校验、工作区隔离、执行预算、上下文管理、可观测性和确定性的失败测试。所有代码都可以在没有 API key、没有网络的情况下跑通。

准备工作

开始之前确认几件事：

• 语言基础：熟悉 TypeScript，并对 AI agent 的基本概念（消息、工具调用、循环）有初步了解。
• 运行时：Node.js 24 与 npm。本文使用 Node 内置的测试运行器（node:test），因此不需要额外安装测试框架。
• 依赖策略：尽量少装依赖。除了 TypeScript 编译工具链，核心逻辑只依赖标准库。
• 目录结构：建议按下面的方式组织，后续每一节都会往对应文件里加内容。

agent-harness/
├── src/
│   ├── core/
│   │   ├── types.ts        # 契约：Message、Tool、Model、AgentResult
│   │   ├── agent.ts        # 执行循环
│   │   └── registry.ts     # 工具注册与参数校验（第 2 节加入）
│   ├── models/
│   │   └── fake.ts         # 脚本化假模型，用于测试
│   └── tools/              # 具体工具实现
├── tests/
│   └── agent.test.ts
├── package.json
└── tsconfig.json

如果你习惯按里程碑推进，可以把每一节做成一个 git tag，用 git switch --detach <tag> 在快照之间切换，这样能对照着看每一步到底改了什么，而不是一上来就面对最终代码。切换后依次运行 npm ci、npm run typecheck、npm test 和 npm run demo 验证环境。

操作步骤

第一步：确定核心契约

整个设计里最重要的一个架构决定是：模型只提出动作，harness 负责执行动作。当模型请求一次工具调用时，它实际上什么都没做，只是生成了一段结构化请求，包含工具名和参数。harness 必须解释这段请求、调用对应实现、再把结果回传给模型。

先在 src/core/types.ts 里把契约写清楚：

export type ToolCall = {
  id: string;
  name: string;
  arguments: unknown;
};

export type Message =
  | { role: 'user'; content: string }
  | { role: 'assistant'; content: string; toolCalls?: ToolCall[] }
  | { role: 'tool'; callId: string; content: string; isError: boolean };

export interface ToolDefinition {
  name: string;
  description: string;
  parameters: Record<string, unknown>; // 给模型看的 JSON Schema
}

export interface Tool extends ToolDefinition {
  execute(argumentsValue: unknown, signal: AbortSignal): Promise<unknown>;
}

export interface Model {
  generate(input: {
    messages: readonly Message[];
    tools: readonly ToolDefinition[];
    signal: AbortSignal;
  }): Promise<{ content: string; toolCalls?: ToolCall[] }>;
}

export type AgentResult = {
  status: 'completed' | 'step_limit' | 'aborted';
  answer?: string;
  messages: Message[];
  steps: number;
};

几个细节值得注意。Model 接口里没有任何 OpenAI、Anthropic 或其他供应商的痕迹，供应商适配器只需在自己的 API 和这些类型之间做转换，harness 本身不用改。Tool 继承 ToolDefinition，意味着模型看到的定义和 harness 持有的可执行实现是同一个对象的两面，但只有 harness 能拿到 execute。

还有一点：工具参数的类型是 unknown，不是 any。模型生成的数据属于不可信输入，必须在使用前校验，这是下一节的主题。

第二步：写出基础执行循环

src/core/agent.ts 里的循环遵循一个固定序列：把对话和可用工具定义发给模型；把模型响应追加进对话；如果它请求了工具，执行并追加结果；重复，直到模型给出最终答案或触达步数上限。

import type { AgentResult, Message, Model, Tool } from './types.ts';

export async function runAgent(options: {
  model: Model;
  tools: readonly Tool[];
  prompt: string;
  maxSteps?: number;
  signal?: AbortSignal;
}): Promise<AgentResult> {
  const { model, tools, prompt, maxSteps = 10 } = options;

  if (!Number.isSafeInteger(maxSteps) || maxSteps < 1) {
    throw new RangeError('maxSteps must be a positive safe integer');
  }

  const signal = options.signal ?? new AbortController().signal;
  const messages: Message[] = [{ role: 'user', content: prompt }];

  const registry = new Map(tools.map(tool => [tool.name, tool]));
  if (registry.size !== tools.length) throw new Error('Duplicate tool names');

  const definitions = tools.map(({ name, description, parameters }) => ({
    name, description, parameters,
  }));

  for (let step = 0; step < maxSteps; step++) {
    if (signal.aborted) return { status: 'aborted', messages, steps: step };

    let response;
    try {
      response = await model.generate({ messages: [...messages], tools: definitions, signal });
    } catch (error) {
      if (signal.aborted) return { status: 'aborted', messages, steps: step };
      throw error;
    }

    if (signal.aborted) return { status: 'aborted', messages, steps: step };

    const calls = response.toolCalls ?? [];
    messages.push({
      role: 'assistant',
      content: response.content,
      ...(calls.length ? { toolCalls: calls } : {}),
    });

    if (!calls.length) {
      return { status: 'completed', answer: response.content, messages, steps: step + 1 };
    }

    for (const call of calls) {
      if (signal.aborted) return { status: 'aborted', messages, steps: step + 1 };

      const tool = registry.get(call.name);
      if (!tool) {
        messages.push({
          role: 'tool', callId: call.id,
          content: `Unknown tool: ${call.name}`, isError: true,
        });
        continue;
      }

      try {
        const result = await tool.execute(call.arguments, signal);
        if (signal.aborted) return { status: 'aborted', messages, steps: step + 1 };
        messages.push({
          role: 'tool', callId: call.id,
          content: JSON.stringify(result) ?? 'null', isError: false,
        });
      } catch (error) {
        if (signal.aborted) return { status: 'aborted', messages, steps: step + 1 };
        messages.push({
          role: 'tool', callId: call.id,
          content: error instanceof Error ? error.message : String(error),
          isError: true,
        });
      }
    }
  }

  return { status: 'step_limit', messages, steps: maxSteps };
}

这段代码里有三个行为值得单独说明。第一，没有工具调用的响应会结束整个运行。第二，工具失败不会自动终止 agent，harness 会把错误信息与原始调用 ID 关联后记录下来，让模型在下一轮自行反应——比如换一个路径、换一种方式重试。第三，maxSteps 是硬性上限：模型可以不断请求工作，但它无权决定自己能用多少轮。

第三步：用假模型做确定性测试

不要在测试里调用真实 API。那既不确定，又消耗 token 或算力。用一个脚本化的假模型替代：它按顺序返回预先写好的响应，脚本耗尽就抛错，这样多出来的意外轮次不会静默通过。

// src/models/fake.ts
import type { Model } from '../core/types.ts';

export function scriptedModel(
  responses: Array<Awaited<ReturnType<Model['generate']>>>
): Model {
  let index = 0;
  return {
    async generate() {
      const response = responses[index++];
      if (!response) throw new Error('Scripted model exhausted');
      return response;
    },
  };
}

对应的最小测试放在 tests/agent.test.ts，用 npm test 运行：

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { runAgent } from '../src/core/agent.ts';
import { scriptedModel } from '../src/models/fake.ts';

test('executes a tool and returns the final answer', async () => {
  const result = await runAgent({
    prompt: 'Say hello',
    tools: [{
      name: 'hello',
      description: 'Return a greeting',
      parameters: { type: 'object' },
      async execute() { return 'Hello'; },
    }],
    model: scriptedModel([
      { content: '', toolCalls: [{ id: '1', name: 'hello', arguments: {} }] },
      { content: 'Done' },
    ]),
  });

  assert.equal(result.status, 'completed');
  assert.equal(result.answer, 'Done');
  assert.deepEqual(result.messages[2], {
    role: 'tool', callId: '1', content: '"Hello"', isError: false,
  });
});

同样的模式可以覆盖未知工具、步数上限和取消。到这里，循环能跑了，但它仍然无条件信任模型生成的参数，也没有对已注册工具能做什么施加任何限制。

第四步：在执行前校验工具调用

工具描述只是对模型的一个请求，不是模型会返回合法参数的保证。把最初的 Map 换成 ToolRegistry，由它负责注册和运行时参数校验：

import type { Tool, ToolDefinition } from './types.ts';

export class ToolRegistry {
  private readonly tools = new Map<string, Tool>();

  constructor(tools: readonly Tool[]) {
    for (const tool of tools) {
      if (!/^[a-zA-Z][a-zA-Z0-9_-]*$/.test(tool.name)) {
        throw new Error(`Invalid tool name: ${tool.name}`);
      }
      if (this.tools.has(tool.name)) {
        throw new Error(`Duplicate tool name: ${tool.name}`);
      }
      this.tools.set(tool.name, tool);
    }
  }

  definitions(): ToolDefinition[] {
    return [...this.tools.values()].map(({ name, description, parameters }) => ({
      name, description, parameters,
    }));
  }

  get(name: string): Tool | undefined {
    return this.tools.get(name);
  }
}

用 const registry = new ToolRegistry(tools) 创建，用 registry.definitions() 取给模型看的定义。当响应请求某个工具时，先校验参数再调用 execute：

const tool = registry.get(call.name);
if (!tool) {
  messages.push({
    role: 'tool', callId: call.id,
    content: `Unknown tool: ${call.name}`, isError: true,
  });
  continue;
}

const validationErrors = validate(call.arguments, tool.parameters);
if (validationErrors.length) {
  messages.push({
    role: 'tool', callId: call.id,
    content: `Invalid arguments: ${validationErrors.join('; ')}`, isError: true,
  });
  continue;
}

这是一条关键边界：arguments: unknown 在校验通过之前始终不可信。校验失败时工具根本不会运行，模型收到的是一条与原始调用 ID 关联的错误结果。

validate 函数只需支持一个刻意做小的 JSON Schema 子集：原始类型与容器类型检查、对象 properties、required 字段、additionalProperties: false、数组 items、以及 enum。报错时给出类似 $.flags[0] 的路径，方便定位格式错误的输入。不要试图实现完整 JSON Schema——子集越小，行为越可预测，测试越好写。

不是所有错误都同一性质。 未知工具、参数非法、工具内部抛错，这三类都该回传给模型，但语义不同：前两类说明模型这一轮请求本身有问题，第三类说明环境或实现出了问题。把它们都标成 isError: true 是合理的，但错误文案要能区分，否则模型无法判断该改参数还是该换策略。

第五步：给 agent 工具，但不给它整台机器

有了校验还不够。一个能读写任意路径的文件工具，等于把整台机器交出去。这里要做两件事。

把文件操作限制在工作区内。 所有路径先解析成绝对路径，再检查它是否位于工作区根目录之下。注意要处理 .. 穿越和符号链接：只做字符串前缀比较是不够的，应当用 path.resolve 归一化后再比较，必要时用 fs.realpath 解析真实路径。任何越界访问都返回错误结果，而不是抛异常终止整个运行。

让模型选择命令，而不是发明命令。 不要提供一个接收任意字符串并交给 shell 执行的工具。更安全的做法是提供一个受限的命令工具：只接受白名单内的命令名和结构化参数，由 harness 拼装成参数数组后以不经过 shell 的方式启动子进程。这样模型无法通过分号、管道或命令替换注入额外指令。

这两条限制都应该在工具实现内部完成，而不是依赖提示词约束模型。提示词是建议，代码才是边界。

第六步：给执行加上预算

只有 maxSteps 是不够的。一个模型可以在单轮里请求十个工具调用，也可以让某个工具跑上几分钟。把两类限制分开：

• 轮次上限：模型最多能进行多少次生成，对应 maxSteps。
• 工具调用上限：整个运行期间最多执行多少次工具，跨轮累计。

在循环里维护一个计数器，每次真正执行工具前递增并检查；超限时返回一个明确的终止状态，而不是静默截断。同时把 AbortSignal 一路传下去：传给 model.generate，也传给每个 tool.execute。工具实现应当在耗时操作中检查 signal.aborted，让取消和超时能真正中断工作，而不是等它自然结束。

这些限制同样要能测试，而且不必等真实故障发生。用假模型脚本出「连续请求工具直到超限」的序列，断言返回状态是预算耗尽；用 AbortController 在第一次工具调用后立即 abort()，断言状态是 aborted。整个过程毫秒级完成。

第七步：控制上下文规模

上下文窗口是有限资源，而工具输出往往是最容易失控的部分：一次目录遍历、一次日志读取，就可能返回几十万字符。两个措施：

• 限制工具输出：在把结果写入消息之前截断，并明确标注「已截断」，让模型知道信息不完整。截断阈值按字符数或字节数设定，写进配置而不是散落在各处。
• 限制面向模型的历史：当消息总量超过阈值时，从最早的轮次开始裁剪，但保留系统提示与最近若干轮。

裁剪时有一条硬规则：不要把一次工具调用和它的结果拆开。如果 assistant 消息里带着 toolCalls，那么对应的 tool 结果消息必须一起保留或一起移除。否则模型会看到一次没有结果的调用，或者一条没有来源的结果，两种都会让后续推理变得混乱。

第八步：让执行过程可观测

在执行的边界上发出事件，而不是在实现内部随意打日志。典型的事件点包括：运行开始、每轮模型调用前后、每次工具调用前后、运行结束。事件里带上轮次序号、工具名、耗时和结果状态，这样出问题时能还原出完整时间线。

关于用量统计要克制：只记录你真正拿得到的数据。如果供应商适配器返回了 token 用量，就透传；如果没有，就不要估算一个看起来合理的数字填进去。编造的用量指标比没有指标更糟，因为它会让成本分析建立在错误前提上。

第九步：测试失败路径

正常路径的测试只能证明一切顺利时能跑。真正需要覆盖的是失败：

• 在信任模型响应之前先校验它：假模型可以返回结构不合法的响应，比如缺少 content、toolCalls 不是数组、调用缺少 id。harness 应当拒绝这类响应并给出明确错误，而不是让 undefined 一路渗透到下游。
• 被拒绝的批次不要进入对话记录：如果一轮响应整体不合法，不要把它追加进 messages。否则后续轮次会带着一条残缺的 assistant 消息继续，模型看到的历史本身就是坏的。
• 用可复现的变体覆盖分支：同一个测试文件里，用不同的脚本序列分别触发未知工具、参数非法、工具抛错、步数超限、预算超限、中途取消。每个用例都断言 status、steps 和关键消息内容。

一个完整示例

下面把前面的部件串成一个能跑通的最小例子：一个受限的 read_file 工具、一个脚本化模型、一次成功调用，以及一次参数校验失败。

import { test } from 'node:test';
import assert from 'node:assert/strict';
import path from 'node:path';
import { readFile } from 'node:fs/promises';
import { runAgent } from '../src/core/agent.ts';
import { ToolRegistry } from '../src/core/registry.ts';
import { scriptedModel } from '../src/models/fake.ts';

const workspace = path.resolve('/tmp/agent-workspace');

function insideWorkspace(target: string): string {
  const resolved = path.resolve(workspace, target);
  if (resolved !== workspace && !resolved.startsWith(workspace + path.sep)) {
    throw new Error(`Path escapes workspace: ${target}`);
  }
  return resolved;
}

const readFileTool = {
  name: 'read_file',
  description: 'Read a UTF-8 text file inside the workspace',
  parameters: {
    type: 'object',
    properties: { path: { type: 'string' } },
    required: ['path'],
    additionalProperties: false,
  },
  async execute(args: unknown) {
    const { path: target } = args as { path: string };
    return await readFile(insideWorkspace(target), 'utf8');
  },
};

test('reads a file, then reports a validation error', async () => {
  const registry = new ToolRegistry([readFileTool]);

  const model = scriptedModel([
    { content: '', toolCalls: [{ id: 'c1', name: 'read_file', arguments: { path: 'notes.txt' } }] },
    { content: '', toolCalls: [{ id: 'c2', name: 'read_file', arguments: { path: 42 } }] },
    { content: 'Finished' },
  ]);

  const result = await runAgent({
    model,
    tools: [readFileTool],
    prompt: 'Read notes.txt and summarize it',
    maxSteps: 5,
  });

  assert.equal(result.status, 'completed');
  assert.equal(result.answer, 'Finished');

  const first = result.messages.find(m => m.role === 'tool' && m.callId === 'c1');
  assert.equal(first?.isError, false);

  const second = result.messages.find(m => m.role === 'tool' && m.callId === 'c2');
  assert.equal(second?.isError, true);
  assert.match(second!.content, /Invalid arguments/);
});

这个例子里，第一次调用正常返回文件内容，第二次因为 path 不是字符串被校验拦下，工具根本没有执行，模型收到一条带 $.path 路径的错误信息，然后在第三轮给出最终答案。整个测试不访问网络、不消耗 token，结果完全确定。

注意事项

• 参数校验是边界，不是装饰：unknown 在校验通过前一律不可信。校验失败时不要「尽力而为」地调用工具，直接返回错误结果。
• 工具失败不终止运行：把错误与调用 ID 关联后回传，让模型有机会调整策略。但预算耗尽和取消应当终止运行，并返回明确状态。
• 路径检查要归一化：只做字符串前缀比较挡不住 .. 穿越。先 path.resolve，必要时再解析真实路径。
• 不要提供任意 shell 执行工具：白名单命令加结构化参数，以不经过 shell 的方式启动子进程。
• 裁剪历史时保持调用与结果成对：拆开会让模型看到自相矛盾的历史。
• 不要编造用量数据：拿不到就不记录，估算值会污染成本分析。
• 假模型脚本耗尽要抛错：否则多出来的意外轮次会静默通过测试。
• 版本与依赖信息以官网当前信息为准：Node.js 版本、TypeScript 版本和各类配额都可能变化，落地前请核对官方文档。

把真实语言模型接进来是最后一步，也是最简单的一步：写一个实现 Model 接口的适配器，把供应商的请求与响应映射到 Message、ToolDefinition 和 ToolCall 上。harness 本身不需要任何改动——这正是把契约和实现分开的价值所在。