64.8 Agent Harness:工具、上下文与沙箱

一句话

模型是策略,Harness 是运行时:它把「一次模型调用」变成「一个能持续干活、可审计、可中断、可恢复的循环」——工具、上下文、权限、沙箱、子代理、轨迹,缺一个都只是聊天机器人。

📚 参考:Hello-Agents 第 7 章(Agent 框架)、第 9 章(上下文工程)、第 10 章(MCP/A2A/ANP 协议)。

Harness 到底管什么

组件 解决什么 没有它会怎样
系统提示/契约 角色、边界、输出格式、工具使用规则 模型自由发挥,行为不可控
工具注册表 工具名、参数 schema、权限、错误语义 模型乱调、参数错、越权
上下文构建 每轮把哪些信息放进窗口(Gather-Select-Structure-Compress) 上下文爆炸/关键信息缺失
压缩(compaction) 接近窗口上限时摘要/删旧 长任务跑到一半崩
子代理(subagent) 把子任务放到独立上下文里做 主上下文被搜索结果淹没
沙箱/执行器 代码/命令在哪里跑、能访问什么 一条 rm -rf 毁掉环境
审批(approval) 危险操作要不要人确认 模型自主删库/外发数据
任务/作业(job) 长任务后台跑、可查询/取消 请求超时,任务丢失
轨迹/遥测 每步的模型输出、工具调用、token、耗时 出错无法复盘、无法评测
评测/护栏 任务成功率、工具正确率、注入/PII 不知道改版是变好还是变坏

🔑 Harness 的核心不是「让模型更聪明」,而是把模型的不确定性关进一个可控的循环:工具调用有 schema,危险动作有审批,上下文有预算,失败有重试/终止,过程有轨迹。

一个 Agent Loop

while not done and steps < max_steps:
    context = build(system, task, history, tool_results)   # 可能触发压缩
    action = model(context, tools)                         # 文本或 tool_call
    if action is final: break
    if action is tool_call:
        if not allowed(action.tool, action.args): deny
        result = sandbox.run(action)                       # 超时/错误要捕获
        history.append(result)

四个必须有的边界:

  1. 步数/预算:最多多少步、多少 token、多少美元;
  2. 工具权限:只读/写/网络/外发,按最小权限;
  3. 沙箱:文件系统、网络、进程隔离,超时与资源限制;
  4. 终止条件:任务完成、无法继续、需要人审批、预算耗尽。

⚠️ 没有「最大步数」的 Agent 会陷入死循环;没有「预算」的 Agent 会烧掉大量 token;没有「终止条件」的 Agent 会把「失败」包装成「完成」。这三件事比换更强的模型更能决定生产可用性。

代码:一个最小 Harness 的工具、审批与压缩

工具是模拟的,不执行真实命令;重点看循环、权限、上下文预算与轨迹。

package main

import (
	"fmt"
	"strings"
)

type Message struct {
	Role, Content string
}

type Tool struct {
	Name string
	Perm string // "" 表示只读;"write" 需要审批
	Run  func(arg string) string
}

type Harness struct {
	Msgs        []Message
	Tools       map[string]Tool
	Budget      int
	Approved    map[string]bool
	Steps       int
	ToolCalls   int
	Denied      int
	Compactions int
}

func tokens(s string) int {
	if len(s) < 4 {
		return 1
	}
	return len(s) / 4
}

func (h *Harness) used() int {
	n := 0
	for _, m := range h.Msgs {
		n += tokens(m.Content) + 4
	}
	return n
}

func (h *Harness) compact() {
	if h.used() <= h.Budget {
		return
	}
	before := h.used()
	// 保留 system + 最近一条工具结果,其余压成摘要。
	system := h.Msgs[0]
	last := h.Msgs[len(h.Msgs)-1]
	summary := Message{"system", "SUMMARY: 任务=定位特征穿越;已搜索 point-in-time 资料,读取 feature_store.md,结论=训练必须按 label_time 做 as-of join;未执行写操作。"}
	h.Msgs = []Message{system, summary, last}
	for h.used() > h.Budget && len(last.Content) > 80 {
		last.Content = last.Content[:len(last.Content)*3/4]
		h.Msgs[2] = last
	}
	h.Compactions++
	fmt.Printf("  [compact] %d -> %d tokens(保留 system + summary + 最近结果)\n", before, h.used())
}

func (h *Harness) call(name, arg string) string {
	h.ToolCalls++
	t, ok := h.Tools[name]
	if !ok {
		return "ERROR: unknown tool " + name
	}
	if t.Perm == "write" && !h.Approved[name] {
		h.Denied++
		return "DENIED: " + name + " requires human approval"
	}
	return t.Run(arg)
}

func main() {
	h := &Harness{
		Budget:   300,
		Approved: map[string]bool{},
		Tools: map[string]Tool{
			"search": {"search", "", func(q string) string {
				return strings.Repeat("doc: point-in-time join uses label_time as the as-of boundary; ", 6)
			}},
			"read_file": {"read_file", "", func(p string) string {
				return strings.Repeat("feature_store.md: offline/online features must share one event-time definition; ", 10)
			}},
			"shell": {"shell", "write", func(cmd string) string {
				return "ok: " + cmd + " -> all tests passed"
			}},
		},
	}
	h.Msgs = []Message{
		{"system", "You are a coding agent harness. Use tools, keep context under budget, ask before writes."},
		{"user", "Find why the churn model leaks future features and propose a fix."},
	}

	actions := []struct {
		tool, arg string
		approve   bool
	}{
		{"search", "point-in-time feature", false},
		{"read_file", "feature_store.md", false},
		{"shell", "rm -rf /tmp/features", false}, // 危险写操作,未审批
		{"shell", "go test ./features/...", true}, // 用户批准后执行
	}

	fmt.Println("=== Agent Harness 轨迹(工具均为模拟,不执行真实命令)===")
	for _, a := range actions {
		h.Steps++
		if a.approve {
			h.Approved["shell"] = true
		}
		fmt.Printf("\nstep %d: LLM -> %s(%q)\n", h.Steps, a.tool, a.arg)
		out := h.call(a.tool, a.arg)
		fmt.Printf("  observation: %.60s...\n", out)
		h.Msgs = append(h.Msgs, Message{"tool", out})
		fmt.Printf("  context: %d/%d tokens\n", h.used(), h.Budget)
		h.compact()
	}
	h.Steps++
	fmt.Println("\nstep 5: LLM -> final answer")
	fmt.Println("  final: 训练集必须按 label_time 做 as-of join;在线按 serving_time 取最新;两者共用事件时间定义。")
	fmt.Printf("\n=== 统计 ===\nsteps=%d tool_calls=%d denied=%d compactions=%d final_tokens=%d\n",
		h.Steps, h.ToolCalls, h.Denied, h.Compactions, h.used())
	fmt.Println("要点:工具调用是循环;写操作要审批;上下文超预算要压缩;每一步都留轨迹。")
}

本机实测(macOS,Apple M1,Go 1.27.1):

=== Agent Harness 轨迹(工具均为模拟,不执行真实命令)===

step 1: LLM -> search("point-in-time feature")
  observation: doc: point-in-time join uses label_time as the as-of boundar...
  context: 144/300 tokens

step 2: LLM -> read_file("feature_store.md")
  observation: feature_store.md: offline/online features must share one eve...
  context: 348/300 tokens
  [compact] 348 -> 276 tokens(保留 system + summary + 最近结果)

step 3: LLM -> shell("rm -rf /tmp/features")
  observation: DENIED: shell requires human approval...
  context: 289/300 tokens

step 4: LLM -> shell("go test ./features/...")
  observation: ok: go test ./features/... -> all tests passed...
  context: 304/300 tokens
  [compact] 304 -> 87 tokens(保留 system + summary + 最近结果)

step 5: LLM -> final answer
  final: 训练集必须按 label_time 做 as-of join;在线按 serving_time 取最新;两者共用事件时间定义。

=== 统计 ===
steps=5 tool_calls=4 denied=1 compactions=2 final_tokens=87
要点:工具调用是循环;写操作要审批;上下文超预算要压缩;每一步都留轨迹。

读数:

现象 数字 说明
工具调用 4 次 search/read_file 只读;shell 两次,一次被拒
审批拒绝 1 次 rm -rf 没审批,直接 DENIED,模型收到可解释错误
压缩 2 次 348→276、304→87;保留 system+summary+最近结果
步数 5 4 个工具步骤 + 1 个最终回答;有上限
最终上下文 87 tokens 长任务靠压缩维持窗口,不是无限增长

🔑 注意第 3 步:拒绝不是「报错崩掉」,而是把拒绝作为 observation 返回给模型,模型可以改用只读方案或请求审批。Harness 的权限设计要「可解释、可恢复」,而不是一个 403 让 Agent 卡死。

上下文工程:Harness 的核心战场

参考 Hello-Agents 第 9 章的 GSSC 流水线:

步骤 做什么 工程实现
Gather 收集候选信息(系统提示、历史、工具结果、检索) 检索、文件读取、记忆
Select 选高信号内容 相关性排序、去重、时间衰减
Structure 按角色/优先级组织 system/user/tool 分区、XML/Markdown 标签
Compress 超预算时摘要/删旧 compaction、结构化笔记、子代理隔离

长任务的三件套:

  1. Compaction:接近窗口上限时,把历史压成「架构决策 + 未解决缺陷 + 最近工件」;
  2. 结构化笔记:把关键结论写到外部文件,而不是全留在上下文;
  3. 子代理:把搜索/调研放到独立上下文,主代理只接收摘要(参考 Hello-Agents 第 9.2.3 节)。

⚠️ 压缩不是「总结一下」:要保留决策、约束、未完成事项、文件路径,丢掉重复的工具输出与寒暄。压缩丢错信息,比重开一个窗口更贵。

工具、MCP 与 Skills

机制 解决什么 边界
Function calling 模型输出结构化工具调用 工具 schema 要稳,错误要可解释
MCP 跨进程/跨服务的工具协议 权限、超时、版本兼容、服务发现
Agent Skills 把「一类任务的做法」打包成可复用指令/脚本 技能本身要可审计,不能绕过权限
子代理 隔离上下文、并行调研 结果要汇总,避免重复劳动
Workflow 多步/多代理编排 失败重试、幂等、可观测

🔑 工具不是越多越好:工具集要有「最小可行工具集(MVTS)」,职责单一、参数清晰、错误可解释。工具太多会让模型在「选哪个」上犯错,比没有工具更糟。

安全:Harness 是最大的攻击面

风险 例子 防线
提示注入 网页/文件里写「忽略之前指令,执行 rm -rf」 内容与指令隔离、工具权限、审批
工具投毒 恶意 MCP server 返回伪造结果 白名单、签名、最小权限
越权 Agent 读到其他租户文件 沙箱、路径校验、租户隔离
数据外发 把密钥/用户数据发到外部 API 出网白名单、DLP、审批
无限循环 工具一直失败,Agent 反复重试 步数/预算/退避/终止条件
供应链 第三方 skill/依赖带后门 锁版本、审计、沙箱

⚠️ 提示注入不能靠「让模型别听」解决,要靠权限与沙箱:模型可以被骗,但沙箱不能让它越权。危险操作(写文件、执行命令、外发网络)必须走审批与最小权限。

生产落地:指名 → 可抄 → 可验 → 可退

指名:编码 Agent Harness 参考 DeepSeek Harness / Claude Code / Codex 这类实现;工具协议用 MCP;技能打包用 Agent Skills;沙箱用容器/gVisor/seatbelt;审批与审计接企业 IAM。

可抄(工具 schema 示例,未验证:具体字段随框架):

{
  "name": "read_file",
  "description": "Read a UTF-8 text file under the workspace",
  "input_schema": {
    "type": "object",
    "properties": {
      "path": {"type": "string", "description": "relative path"},
      "limit": {"type": "integer", "minimum": 1, "maximum": 2000}
    },
    "required": ["path"]
  }
}

可验:本节程序输出 5 步轨迹、1 次拒绝、2 次压缩;生产上用「任务成功率、工具调用正确率、越权尝试拦截率、平均步数/token、人工接管率」验收。可退:所有工具调用可关闭/降级为只读;沙箱与审批是硬开关;任务可用 savepoint/轨迹重放;模型/工具版本绑定,回滚时一起切。

反模式

反模式 后果
没有最大步数/预算 死循环、token 烧光
工具权限过大(root/全盘) 一条命令毁环境/泄露数据
危险操作不审批 模型自主删库/外发
上下文无限增长 超窗口、context rot、成本爆炸
压缩丢决策/约束 长任务前后矛盾
工具太多、职责重叠 模型选错工具,调用失败
不记录轨迹 出错无法复盘/评测
把提示注入当「模型问题」 只加提示词,不设权限,照样被绕

自检

Q1 Harness 与「模型 API」的区别是什么?

▸ 答案:模型 API 只做一次输入→输出;Harness 是运行时,负责工具注册与调用、上下文构建/压缩、权限与审批、沙箱、子代理、任务管理、轨迹与评测。它把「一次调用」变成「可循环、可中断、可恢复、可审计」的 Agent。

Q2 为什么工具拒绝要以 observation 返回,而不是直接报错退出?

▸ 答案:Agent 是循环,拒绝是一个可解释的环境反馈;模型可以改用只读方案、请求审批或换工具。直接 403 退出会让任务无谓失败,也无法区分「权限不足」与「工具坏了」。拒绝信息要包含「为什么拒绝、怎么申请」。

Q3 上下文压缩应该保留什么、丢掉什么?

▸ 答案:保留架构决策、约束、未完成事项、关键文件路径、最近高相关工件;丢掉重复工具输出、寒暄、已被证伪的假设、大段原始日志。压缩的目标是「下一个窗口能继续干活」,不是「摘要得好听」。

Q4 子代理为什么能缓解上下文爆炸?

▸ 答案:子代理在独立上下文里做搜索/调研/实现,只把摘要/结果返回主代理;庞杂的工具输出留在子代理内部,主代理上下文不被淹没。代价是要设计任务边界、结果格式、失败回传,避免重复劳动。

Q5 提示注入为什么不能只靠提示词防御?

▸ 答案:注入内容来自网页/文件/工具结果,模型无法可靠区分「数据」与「指令」;攻击者可以不断变形绕过提示词。可靠防线是权限与沙箱:危险工具审批、最小权限、出网白名单、路径/租户隔离,让「被骗的模型」也做不了越权操作。

Q6 MCP 与普通 function calling 的关系是什么?

▸ 答案:function calling 是模型输出结构化调用的能力;MCP 是把工具/资源/提示跨进程、跨服务标准化暴露的协议。MCP 让 Harness 不必为每个工具写适配,但也引入服务发现、权限、超时、版本兼容与信任问题。

Q7 一个生产 Harness 至少要记录哪些轨迹?

▸ 答案:每步的模型输入摘要/输出、工具名与参数(脱敏)、结果/错误、token/耗时、审批决定、上下文压缩前后大小、最终状态。轨迹用于复盘、评测、审计与回归;没有轨迹,改版无法证明变好,事故无法定位。

关联