TL;DR

Agent Harness 是围绕大模型构建的运行时基础设施,负责管理 Agent 的生命周期、上下文状态、工具调用链路与执行安全。本文以 Claude Code 为范本,拆解其七个核心工程机制:执行循环(感知—推理—行动—观测的 while 闭环)、原子化工具集(bash/文件操作按需组合)、动态技能加载(目录常驻+内容按需注入)、三层上下文压缩(微观清理→阈值重置→模型主动压缩)、Human-in-the-loop 审批(高风险操作前插入人工确认节点)、任务编排(会话内 todo 列表 + 跨会话带依赖图的持久化任务系统),以及多智能体协作(一次性 Subagent 上下文隔离 + 持久化 Agent Teams 邮箱通信)。这些机制共同解决了 Agent 工程化落地中最核心的几个问题:任务不丢失、上下文不爆炸、执行可审计、复杂任务可拆解、多 Agent 可协同。

本文所涉及的代码示例全部来自 shareAI-lab/learn-claude-code 仓库,一个按难度递进展示 Agent Harness 核心机制的教学项目。

从工作流到 Harness Engineer

随着大模型能力不断突破,智能体(Agent)的构建正逐步从基于工作流、提示词链路、规则引擎等固化流程的 “巨型 Shell 脚本”,转向真正具备感知、推理、决策与执行能力的自主智能系统。前者是在传统符号系统引入AI能力,后者则是以大模型为核心、具有自主规划、动态决策和环境交互能力的系统,可以解决所有复杂、多步骤、需要判断的工作。

现在的问题是,大模型能力够了却不听话,无法精准匹配任务需求。所以其应用关键,是在最大化模型的开放推理与自主决策能力的同时,提升模型任务执行的精准度与成功率。目前,行业内已形成普遍共识:简洁稳健的支撑框架搭配优秀模型,其效果远比重度流程编排更为灵活和高效。

应对这一需求,业内提出了Agent Harness,一套围绕大语言模型(LLM)构建的运行时基础设施。核心作用是统一管理Agent的生命周期、上下文状态、工具调用链路、执行安全与状态持久化,本质上是为自主智能体提供“可控运行环境”——让不确定性的模型推理,能够在确定性的工程框架内安全、稳定、可复现地落地执行。构成要素包括:

  • Tools(工具):涵盖文件读写、Shell、网络、数据库、浏览器等基础操作工具,为智能体提供执行具体任务的硬件与软件支撑;
  • Knowledge(知识):包含产品文档、领域资料、技能拓展、API规范、风格指南等,为智能体提供专业的知识储备;
  • Observation(观测):可获取执行结果、执行日志、浏览器状态、传感器数据等任务相关信息,使智能体感知外部状态;
  • Action Interfaces(行动接口):支持通过CLI命令、API调用、UI交互等方式,让智能体将决策转化为实际操作;
  • Permissions(权限):通过沙箱隔离、审批流程、信任边界等机制,保障智能体执行过程的安全性与合规性。

这一理念在行业内已有成熟实践,国外的Claude Code[1]、国内的Kimi Code[4]便是典型代表。它们摒弃僵化的工作流设计,不用人工预设的规则干涉模型决策,而是为大模型配齐工具、知识、上下文管理及安全边界,然后充分信任模型、给够自由度,让模型充分自主地完成复杂任务的全流程。

为什么学习 Claude Code 的 Harness

在众多的 Agent 工程实践中,Claude Code 是最值得拆解的范本。它是首个将 “自主规划 + 工具执行 + 多智能体协作 + 工程化保障” 完整落地且始终保持领先的生产级代码 Agent,其架构、机制与工程实践几乎覆盖了当前 Agent 开发的所有关键痛点,且有明确的生产验证与可量化效果。主要特性为:

  1. Agent执行循环:作为整个系统的骨干核心,它实现了“感知—推理—行动—观测”的自主闭环机制,能够让模型根据每一步的执行结果进行实时反馈与迭代优化,确保任务推进的连贯性与准确性。
  2. 原子化能力集:提供Bash、文件读写、Glob搜索、Grep分析等底层原子级工具能力,Agent可通过对这些基础能力的灵活组合来解决复杂任务。支持一次调用多个工具并实现并行执行,有效提升任务处理效率。
  3. 动态技能加载:摒弃传统全量Prompt模式,采用按需加载的思路,根据具体任务需求加载对应领域知识或特定技能(如代码审查、测试生成等),既有效节省了上下文窗口资源,又显著提升了模型的任务专注度,同时具备良好的跨场景可迁移性。
  4. 上下文压缩与管理:通过智能摘要与状态剪枝技术,精准解决长程任务中常见的模型“遗忘”与上下文“污染”问题,确保模型始终聚焦于当前任务的核心信息,同时有效降低token消耗,平衡性能与成本。
  5. 人工审批机制:human-in-the-loop,建立严格的审批流程,在赋予模型一定自主决策自由度的同时,通过人工确认机制筑牢安全底线,避免违规操作与风险输出,保障生产级应用的安全性。
  6. 复杂任务编排:支持对复杂任务进行自动拆解与并行规划,让模型具备处理多步骤工程任务的宏观视野;同时引入带依赖图的任务系统,清晰梳理任务间的关联关系,确保任务推进的有序性与高效性。
  7. 多智能体协作网络:支持主Agent根据任务需求动态派生子Agent(Sub-Agents),并通过异步邮箱机制实现主、子Agent团队间的解耦与高效协同,模拟真实开发团队的分工模式,提升复杂代码任务的处理能力。

Claude Code 的 Harness Engineer

Agent执行循环

Agent 执行循环是骨架,用一句话概括就是:把工具执行结果不断喂回给模型,直到模型主动停下来

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def agent_loop(messages: list):
while True:
response = client.messages.create(
model=MODEL, system=SYSTEM, messages=messages,
tools=TOOLS, max_tokens=8000,
) # 1. 调用 LLM
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use": # 2. 模型不再调用工具,退出
return
results = []
for block in response.content: # 3. 模型可能输出多个工具调用指令,依次执行每个工具
if block.type == "tool_use":
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
results.append({"type": "tool_result", ...})
messages.append({"role": "user", "content": results}) # 4. 结果追加到上下文,继续循环

整个循环只有四步,但恰恰是这四步构成了 Agent 的"感知—推理—行动—观测"闭环:

  • 感知:每轮循环开始,LLM 接收完整的对话历史(包含之前所有工具的执行结果),感知当前任务状态;
  • 推理:LLM 根据上下文推理出下一步需要执行什么操作,以 tool_use 的形式表达出来;
  • 行动:Harness 解析 tool_use 请求,调用真实的 bash 命令执行,获取输出;
  • 观测:将执行结果以 tool_result 的形式追加到消息队列,下一轮 LLM 就能"看到"执行结果。

循环的终止条件只有一个:stop_reason != "tool_use",即模型不再发起工具调用,说明它认为任务已经完成(或者无法继续),此时正常退出并输出最终回复。

这个模式极简却极强——无论任务有多少步骤,只要模型一直有新的工具调用产生,循环就会持续推进;一旦模型认为工作完成,循环自然结束。生产级 Agent 在此基础上叠加策略(最大轮次限制、token 预算、错误重试)、钩子(pre/post tool 审批)和状态持久化,但核心骨架始终是这个 while 循环。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
sequenceDiagram
participant Agent
participant LLM
participant Tool

Agent->>LLM: messages(含用户请求)
LLM-->>Agent: tool_use(如 bash 命令)

loop 工具调用循环
Agent->>Tool: 执行命令
Tool-->>Agent: 执行结果(stdout/stderr)
Agent->>LLM: messages + tool_result
alt 继续调用工具
LLM-->>Agent: tool_use(下一步操作)
Agent->>Tool: 执行命令
Tool-->>Agent: 执行结果
else 任务完成
LLM-->>Agent: stop_reason = end_turn
end
end

Agent->>Agent: 输出最终回复

原子化能力集

工具的设计有一条反直觉的原则:单个工具越简单,整体能力越强

bash 为例,它的接口只有一个字段 command,却能执行任意 shell 命令——编译、测试、搜索、网络请求、进程管理,全部覆盖。浏览器工具同理,一个"打开页面并返回内容"的接口,就能让模型访问整个互联网。这类工具本身不内置任何业务逻辑,只是把底层系统能力暴露给模型,再由模型自己决定怎么组合使用。

相比之下,如果给模型一个"查询 GitHub Issues 并按优先级排序后写入报告"的专用工具,虽然功能明确,但模型只能用在这一个场景;换个任务,工具就失效了。原子化工具的价值正在于此:粒度小、通用性强,模型通过多步组合就能覆盖任意复杂场景,而不需要为每个场景预先设计一个工具。

扩展能力也因此变得极低成本——循环本身不需要改动,只需往工具列表里加一条记录,模型就能立刻使用新能力。read_filewrite_fileedit_file 相比 bash 提供了更精细的文件操作语义,让模型可以精准替换某段文本,而不是用 shell 命令拼接 sed,降低了出错概率,也让执行意图更清晰可审计。

1
2
3
4
5
6
7
# 派发表:工具名 → 处理函数
TOOL_HANDLERS = {
"bash": lambda **kw: run_bash(kw["command"]),
"read_file": lambda **kw: run_read(kw["path"], kw.get("limit")),
"write_file": lambda **kw: run_write(kw["path"], kw["content"]),
"edit_file": lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),
}

工具派发的逻辑极简:一张字典,键是工具名,值是对应的处理函数。循环里只需一行就能路由:

1
2
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown tool: {block.name}"

这种设计有两个好处:一是扩展零成本,新增工具只需往字典里插一条记录;二是调用方与实现方解耦,LLM 只知道工具的名字和 JSON Schema,不知道背后是哪个函数,可以随时替换实现而不影响模型行为。

工具本身的设计遵循"原子化"原则——每个工具只做一件事,粒度足够小、足够通用。这样模型可以自由组合:先用 read_file 读取文件,再用 edit_file 精准替换某段文本,最后用 bash 跑测试验证结果,整个过程不需要任何硬编码的流程控制。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
sequenceDiagram
participant Agent
participant LLM
participant Dispatch as Tool Dispatch
participant Tool

Agent->>LLM: messages + TOOLS 定义(4 个工具的 JSON Schema)
LLM-->>Agent: tool_use { name: "read_file", input: {path: "..."} }

loop 工具调用循环
Agent->>Dispatch: tool_name + input
alt bash
Dispatch->>Tool: run_bash(command)
else read_file
Dispatch->>Tool: run_read(path, limit)
else write_file
Dispatch->>Tool: run_write(path, content)
else edit_file
Dispatch->>Tool: run_edit(path, old_text, new_text)
end
Tool-->>Dispatch: 执行结果
Dispatch-->>Agent: output
Agent->>LLM: tool_result
LLM-->>Agent: 下一步 tool_use 或 end_turn
end

动态技能加载

随着工具和知识越来越多,最直觉的做法是把所有说明都塞进 System Prompt——“你会处理 PDF、你懂代码审查、你熟悉数据库迁移……”。但这条路走不远:System Prompt 越长,模型越容易分心,成本越高,任务专注度越低。

动态技能加载的核心洞察是:不要在开始时把所有知识都给模型,而是让模型在需要时自己来取

实现上分两层:

  • Layer 1(目录层,常驻):System Prompt 里只放技能的名字和一句话描述,约 100 tokens/技能,成本极低;
  • Layer 2(内容层,按需):模型认为需要某个技能时,主动调用 load_skill 工具,Agent 将完整的技能说明以 tool_result 的形式注入上下文。
1
2
3
4
5
skills/
pdf/
SKILL.md ← frontmatter(name, description)+ 完整操作指南
code-review/
SKILL.md
1
2
3
4
5
6
7
8
class SkillLoader:
def get_descriptions(self) -> str:
# Layer 1:只返回名字和描述,注入 System Prompt
return " - pdf: Process PDF files...\n - code-review: Review code..."

def get_content(self, name: str) -> str:
# Layer 2:返回完整 skill body,通过 tool_result 注入上下文
return f'<skill name="{name}">\n{skill["body"]}\n</skill>'

模型收到 System Prompt 后,知道有哪些技能可用,但并不知道细节。当任务需要处理 PDF 时,它会先调用 load_skill("pdf"),读取完整指南后再动手。对于不涉及的技能,内容永远不会进入上下文,窗口资源被精准保留给当前任务。

这个模式的另一个好处是技能可以独立维护。每个 SKILL.md 是一个独立文件,可以随时更新、添加或删除,不需要改动任何代码逻辑,Agent 启动时自动扫描加载。Claude Code 里的可安装第三方 skill,正是这套机制的延伸。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
sequenceDiagram
participant Agent
participant LLM
participant SkillLoader

Note over Agent,LLM: 启动阶段
Agent->>SkillLoader: get_descriptions()
SkillLoader-->>Agent: 技能目录(名称 + 一句话描述)
Agent->>LLM: System Prompt(含技能目录)+ 用户任务

Note over Agent,LLM: 执行阶段
LLM-->>Agent: tool_use { load_skill("pdf") }
Agent->>SkillLoader: get_skill_content("pdf")
SkillLoader-->>Agent: 完整 SKILL.md 内容
Agent->>LLM: tool_result(完整技能说明)

LLM-->>Agent: tool_use(基于技能说明执行具体操作)
Agent->>Agent: 执行工具调用
Agent->>LLM: tool_result
LLM-->>Agent: end_turn

上下文压缩与管理

长程任务是 Agent 的天然杀手。每轮对话把工具输出、中间结果、历史消息全部保留,上下文窗口很快就会被塞满,要么截断丢失关键信息,要么每次调用都花费巨量 token。解决思路是有策略地遗忘:不是随机丢弃,而是按照"越老越不重要"的原则,分三层逐步压缩上下文,让 Agent 可以无限期地持续工作。

Layer 1:micro_compact(每轮静默执行)

工具输出有一个天然的时间局部性:模型引用某次工具结果,绝大多数发生在该工具调用之后的紧邻几步。执行 read_file 读到的文件内容,会在随后的 edit_file 里用到;跑完测试的输出,会在下一步的修复决策里用到。随着任务推进,早期的工具输出早已被"消化"进了后续的推理和操作结果里,模型几乎不会再回头查阅原始输出。保留它们只是在浪费上下文窗口。

因此,每次调用 LLM 之前,把距当前超过 3 步的 tool_result 内容替换为一行占位符:

1
result["content"] = f"[Previous: used {tool_name}]"

压缩前后对比(假设当前已执行 5 次工具调用,保留最近 3 次):

1
2
3
4
5
6
7
8
9
10
11
12
13
# 压缩前
{"role": "user", "content": [{"type": "tool_result", "content": "file1.py\nfile2.py\n...(200行输出)"}]} # 第1步 ← 清理
{"role": "user", "content": [{"type": "tool_result", "content": "def foo():\n pass\n...(500行)"}]} # 第2步 ← 清理
{"role": "user", "content": [{"type": "tool_result", "content": "Tests passed: 42/42"}]} # 第3步 ← 保留
{"role": "user", "content": [{"type": "tool_result", "content": "Wrote 1024 bytes to main.py"}]} # 第4步 ← 保留
{"role": "user", "content": [{"type": "tool_result", "content": "all tests pass"}]} # 第5步 ← 保留

# 压缩后
{"role": "user", "content": [{"type": "tool_result", "content": "[Previous: used bash]"}]} # 第1步
{"role": "user", "content": [{"type": "tool_result", "content": "[Previous: used read_file]"}]} # 第2步
{"role": "user", "content": [{"type": "tool_result", "content": "Tests passed: 42/42"}]} # 第3步(完整保留)
{"role": "user", "content": [{"type": "tool_result", "content": "Wrote 1024 bytes to main.py"}]} # 第4步(完整保留)
{"role": "user", "content": [{"type": "tool_result", "content": "all tests pass"}]} # 第5步(完整保留)

这一步零成本、无感知,只清理"已经不太可能再被引用"的旧输出,保留最近 3 次工具结果的完整内容供模型参考。

Layer 2:auto_compact(token 超阈值自动触发)

micro_compact 只做局部清理,对话本身的骨架——每一轮的 assistant 推理、用户指令、工具调用记录——仍然在积累。任务足够长时,这些内容会把上下文窗口撑满。此时继续追加已无意义:模型看到的上下文越来越长,注意力被稀释,推理质量下降,每次调用的 token 费用也线性攀升。此时可以发起主动清理,把过去发生的一切压缩成一段摘要,以最小的体积保留最关键的信息,然后从这个新起点继续执行。

比如,当上下文估算超过 50,000 tokens 时,触发全量压缩:

1
2
if estimate_tokens(messages) > THRESHOLD:
messages[:] = auto_compact(messages)

auto_compact 做三件事:先把完整对话记录保存到 .transcripts/ 目录,再用以下 prompt 让 LLM 将整个对话压缩成一段摘要,最后把 transcript 文件路径也一并注入到新的上下文里。这样模型在需要时可以通过 read_file 工具读取完整历史,压缩是有损的,但原始记录随时可查:

1
2
3
4
5
Summarize this conversation for continuity. Include:
1) What was accomplished
2) Current state
3) Key decisions made
Be concise but preserve critical details.

压缩后的上下文只剩两条消息,token 消耗减到最低:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 压缩后
{"role": "user", "content": "[Conversation compressed. Transcript: .transcripts/transcript_1234567890.jsonl]

## Accomplished
- 分析了项目结构,确认入口为 main.py
- 修复了 parse_config() 中的 KeyError,原因是缺少默认值
- 所有 42 个单元测试已通过

## Current State
- 当前在 feature/config-fix 分支,改动尚未提交
- main.py 已编辑,config.py 未动

## Key Decisions
- 使用 dict.get() 替代直接下标访问,保持向后兼容"}

{"role": "assistant", "content": "Understood. I have the context from the summary. Continuing."}

原本可能有几十轮对话、数百条工具调用记录,压缩后变成一段结构化摘要,模型从这个新起点继续执行,行为上与压缩前保持连贯。

Layer 3:compact 工具(模型主动触发)

token 阈值是一个被动的工程保险,但它感知不到语义层面的混乱。有些情况下,上下文虽然还没超过阈值,但模型已经在处理一个全新的子任务,前面积累的大量调试日志、中间结果对接下来的工作毫无用处,反而是干扰。更极端的情况是:任务切换了方向,之前的推理路径整体作废,继续带着它执行只会让模型产生错误的"惯性"。这类情况只有模型自己能感知,阈值触发帮不上忙。因此 Agent 把压缩能力也暴露为一个工具,让模型在认为有必要时主动发起:

1
2
3
4
5
6
7
8
{"name": "compact",
"description": "Trigger manual conversation compression.",
"input_schema": {
"type": "object",
"properties": {
"focus": {"type": "string", "description": "What to preserve in the summary"}
}
}}

focus 参数让模型可以指定摘要时重点保留哪些信息——比如"保留当前文件的修改状态和测试结果",而不是让 LLM 自行决定什么重要。调用触发后,执行与 auto_compact 完全相同的流程:保存 transcript、生成摘要、替换消息列表。

三层机制形成一个递进的压缩梯队:微观清理 → 自动重置 → 模型自主,在 token 效率与信息完整性之间找到平衡点。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
sequenceDiagram
participant Agent
participant LLM
participant Disk as .transcripts/

loop 每一轮
Agent->>Agent: micro_compact(清理 3 步前的旧 tool_result)
Agent->>Agent: estimate_tokens(messages)

alt tokens > 50000
Agent->>Disk: 保存完整 transcript
Agent->>LLM: 请压缩对话为摘要
LLM-->>Agent: summary
Agent->>Agent: messages = [summary, ACK]
end

Agent->>LLM: messages(已压缩)
LLM-->>Agent: tool_use 或 end_turn

opt 模型主动调用 compact
Agent->>Disk: 保存完整 transcript
Agent->>LLM: 请压缩对话为摘要
LLM-->>Agent: summary
Agent->>Agent: messages = [summary, ACK]
end
end

人工审批机制

Agent 能自主执行的操作越来越多,随之而来的问题是:人怎么知道它在做什么?怎么确认它做的是对的?

读文件、搜索代码、跑测试,这些操作即便出错也没什么损失,回滚容易,可以放手让模型自己来。但有一类操作不同——覆盖文件、删除内容、执行 git commit、调用外部 API——这些操作一旦执行,影响立刻落地,撤销成本很高,甚至不可逆。对于这类高风险操作,在执行前停下来问一句人,是最低成本的安全保障。

这就是 Human-in-the-loop 的核心思想:不是限制模型的能力,而是在关键决策点上插入一个人工确认节点,确保人始终对 Agent 的行为保持感知和控制权。

具体实现上,新增一个 ask_human 工具:

1
2
3
4
5
6
7
8
9
10
11
12
{"name": "ask_human",
"description": "Ask the human for confirmation or input before taking a sensitive action.",
"input_schema": {
"type": "object",
"properties": {
"question": {"type": "string", "description": "What to ask the human"},
"action": {"type": "string", "description": "The action you are about to take"},
"options": {"type": "array", "description": "Choices to present to the human",
"items": {"type": "string"}}
},
"required": ["question", "action"]
}}

模型在准备写入文件、提交代码、执行破坏性命令之前,先调用 ask_human,把即将执行的操作和候选选项呈现给用户。用户选择后,结果以 tool_result 的形式返回给模型,模型再决定下一步动作。整个过程完全在 Agent 循环内完成:

1
2
3
4
5
6
7
8
模型调用 ask_human:
question: "即将提交代码,请确认操作"
action: "git commit -m 'fix: handle missing config key'"
options: ["yes, commit", "no, cancel", "edit commit message first"]

用户选择:yes, commit

模型:[继续执行 bash("git commit ...")]

这套机制的意义不只是安全,更在于建立信任。用户看到模型在做什么、为什么做,每一步高风险操作都经过自己的确认,Agent 的行为从黑盒变成透明的协作过程。长期下来,用户对 Agent 的能力边界有了清晰认知,能放心地把更复杂的任务交给它,信任也在这个过程中逐步建立。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
sequenceDiagram
participant User
participant Agent
participant LLM
participant Tool

Agent->>LLM: messages
LLM-->>Agent: tool_use { ask_human: "即将执行 git commit,是否继续?" }
Agent->>User: 展示操作内容,等待确认

alt 用户确认
User-->>Agent: yes
Agent->>LLM: tool_result("yes")
LLM-->>Agent: tool_use { bash: "git commit ..." }
Agent->>Tool: 执行
Tool-->>Agent: 执行结果
else 用户拒绝
User-->>Agent: no
Agent->>LLM: tool_result("no")
LLM-->>Agent: 调整方案或终止
end

复杂任务编排

简单任务只需要一个执行循环,但真实的工程任务往往包含十几个步骤——分析代码结构、修改多个文件、跑测试、处理报错、提交代码。模型在执行过程中很容易"迷失":做到第五步时忘记第三步还没完成,或者在一个细节里钻太深,忽略了全局进度。

解决方案是让模型自己管理自己的任务状态,而不是靠外部脚本来调度。这里有两套递进的机制:轻量的 todo 列表用于单次会话的任务追踪,带依赖图的持久化任务系统用于跨会话的复杂工程编排。

todo 工具:会话内的任务追踪

todo 工具给了模型一个持续可见的进度看板。工具的接口很简单:每次调用传入完整的任务列表(不是增量更新), 用新列表整体替换旧列表,并把渲染后的看板作为 tool_result 返回。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 工具定义:模型看到的 JSON Schema
{"name": "todo",
"description": "Update task list. Track progress on multi-step tasks.",
"input_schema": {
"type": "object",
"properties": {
"items": {
"type": "array",
"items": {
"type": "object",
"properties": {
"id": {"type": "string"},
"text": {"type": "string"},
"status": {"type": "string", "enum": ["pending", "in_progress", "completed"]}
},
"required": ["id", "text", "status"]
}
}
},
"required": ["items"]
}}

模型收到任务后,第一步通常是调用 todo 规划拆解:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 模型调用示例
tool_use { todo: items=[
{"id": "1", "text": "读取 config.py 理解当前结构", "status": "in_progress"},
{"id": "2", "text": "修复 parse_config() 的 KeyError", "status": "pending"},
{"id": "3", "text": "补充单元测试", "status": "pending"},
{"id": "4", "text": "更新 README", "status": "pending"}
]}

# tool_result 返回
[>] #1: 读取 config.py 理解当前结构
[ ] #2: 修复 parse_config() 的 KeyError
[ ] #3: 补充单元测试
[ ] #4: 更新 README
(0/4 completed)

完成 #1 后,模型再次调用 todo,把 #1 改为 completed、#2 改为 in_progress,Agent 验证后返回新看板。通过这种"每步更新"的方式,进度始终对模型可见。

1
2
3
4
5
6
7
# todo 列表示例
[x] #1: 读取 config.py 理解当前结构
[>] #2: 修复 parse_config() 的 KeyError ← 当前正在执行
[ ] #3: 补充单元测试
[ ] #4: 更新 README

(1/4 completed)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
sequenceDiagram
participant Agent
participant LLM
participant TodoManager

Agent->>LLM: 用户任务
LLM-->>Agent: tool_use { todo: [任务1~4 pending] }
Agent->>TodoManager: update(items)
TodoManager-->>Agent: tool_result: 4 pending

loop 执行循环
LLM-->>Agent: tool_use { bash / read_file / ... }
Agent->>Agent: 执行工具
Agent->>LLM: tool_result

alt 连续 3 轮未更新 todo
Agent->>LLM: tool_result + reminder("Update your todos.")
LLM-->>Agent: tool_use { todo: #1 completed, #2 in_progress, ... }
Agent->>TodoManager: update(items)
TodoManager-->>Agent: tool_result: 1 completed, 1 in_progress, 2 pending
end
end

任务系统:带依赖图的持久化编排

todo 列表有两个根本局限。

第一,它活在上下文里。一旦触发 auto_compact,列表就随着对话历史一起被摘要掉了。对于跨越多次会话、需要几天才能完成的工程任务,每次重启会话都要重新规划,进度无法延续。

第二,它表达不了依赖关系。真实工程任务的步骤之间往往有先后约束:"写测试"必须等"功能实现"完成,“部署"必须等"所有测试通过”。平铺的列表没有这层语义,模型只能凭上下文记住顺序,一旦被压缩就会失去这些约束。

任务系统解决这两个问题的方式直接:把状态移出上下文,写到磁盘上。每个任务是 .tasks/ 目录下的一个 JSON 文件,对话结束、上下文压缩,文件都还在。每次会话开始,模型调用 task_list 就能完整恢复进度,不依赖任何历史对话记录。

任务系统把状态彻底移出上下文,以 JSON 文件形式持久化到磁盘:

1
2
3
4
.tasks/
task_1.json {"id": 1, "subject": "实现 parse_config", "status": "completed", "blockedBy": [], "blocks": [2]}
task_2.json {"id": 2, "subject": "补充单元测试", "status": "pending", "blockedBy": [1], "blocks": [3]}
task_3.json {"id": 3, "subject": "更新文档", "status": "pending", "blockedBy": [2], "blocks": []}

每个任务有 blockedBy(被谁阻塞)和 blocks(阻塞谁)两个字段,构成双向依赖图。依赖关系是双向自动维护的——设置 task1 blocks task2 时,task2 的 blockedBy 也会同步更新。举个例子,task1 完成时,Agent 调用 TaskManager 把它从所有任务的 blockedBy 中移除,此时模型调用 task_list 时能看到它变为可执行状态。

模型通过四个工具操作任务系统:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# 创建任务
{"name": "task_create",
"description": "Create a new task.",
"input_schema": {"type": "object",
"properties": {"subject": {"type": "string"},
"description": {"type": "string"}},
"required": ["subject"]}}

# 更新状态或依赖关系(addBlocks/addBlockedBy 自动双向同步)
{"name": "task_update",
"description": "Update a task's status or dependencies.",
"input_schema": {"type": "object",
"properties": {"task_id": {"type": "integer"},
"status": {"type": "string",
"enum": ["pending", "in_progress", "completed"]},
"addBlockedBy": {"type": "array", "items": {"type": "integer"}},
"addBlocks": {"type": "array", "items": {"type": "integer"}}},
"required": ["task_id"]}}

# 查看全局进度,会话恢复的入口
{"name": "task_list",
"description": "List all tasks with status summary.",
"input_schema": {"type": "object", "properties": {}}}

# 读取单个任务的完整详情
{"name": "task_get",
"description": "Get full details of a task by ID.",
"input_schema": {"type": "object",
"properties": {"task_id": {"type": "integer"}},
"required": ["task_id"]}}

下面是一次完整的任务系统使用流程,覆盖全部四个工具:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# 任务规划和创建
task_create { subject: "实现 parse_config" }
→ {"id": 1, "status": "pending", "blockedBy": [], "blocks": []}
task_create { subject: "补充单元测试" }
→ {"id": 2, "status": "pending", "blockedBy": [], "blocks": []}
task_create { subject: "更新文档" }
→ {"id": 3, "status": "pending", "blockedBy": [], "blocks": []}

# 建立依赖:task1 完成才能做 task2,task2 完成才能做 task3
task_update { task_id: 1, addBlocks: [2] } # 同时自动给 task2 追加 blockedBy: [1]
task_update { task_id: 2, addBlocks: [3] } # 同时自动给 task3 追加 blockedBy: [2]

# 设置 task#1 的状态为进行中
task_update { task_id: 1, status: "in_progress" }

# 查看整个计划的状态
task_list
→ [>] #1: 实现 parse_config
[ ] #2: 补充单元测试 (blocked by: 1)
[ ] #3: 更新文档 (blocked by: 2)
(0/3 completed)

# 想了解 task2 的具体内容
task_get { task_id: 2 }
→ {"id": 2, "subject": "补充单元测试", "status": "pending", "blockedBy": [1], "blocks": [3]}

# task1 完成,自动解除 task2 的阻塞
task_update { task_id: 1, status: "completed" }

task_list
→ [x] #1: 实现 parse_config
[ ] #2: 补充单元测试
[ ] #3: 更新文档 (blocked by: 2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
sequenceDiagram
participant Agent
participant LLM
participant TaskManager

Note over LLM,TaskManager: 任务规划和创建
LLM-->>Agent: tool_use { task_create × 3 }
Agent->>TaskManager: create(subject)
TaskManager-->>Agent: task JSON × 3

Note over LLM,TaskManager: 建立依赖
LLM-->>Agent: tool_use { task_update addBlocks }
Agent->>TaskManager: update(blockedBy 双向同步)
TaskManager-->>Agent: 更新后的 task JSON

Note over LLM,TaskManager: 设置 task1 为进行中
LLM-->>Agent: tool_use { task_update 1 in_progress }
Agent->>TaskManager: update(1, in_progress)
TaskManager-->>Agent: 更新后的 task JSON

Note over LLM,TaskManager: 查看整个计划的状态
LLM-->>Agent: tool_use { task_list }
Agent->>TaskManager: list_all()
TaskManager-->>Agent: in_progress 1, pending 2

Note over LLM,TaskManager: 了解 task2 的具体内容
LLM-->>Agent: tool_use { task_get 2 }
Agent->>TaskManager: get(2)
TaskManager-->>Agent: task2 详情

Note over LLM,TaskManager: task1 完成,自动解除 task2 的阻塞
LLM-->>Agent: tool_use { task_update 1 completed }
Agent->>TaskManager: update(1, completed)
TaskManager->>TaskManager: _clear_dependency(1)
TaskManager-->>Agent: task2.blockedBy 已清空

LLM-->>Agent: tool_use { task_list }
Agent->>TaskManager: list_all()
TaskManager-->>Agent: completed 1, pending 2

多智能体协作网络

Subagent:上下文隔离

复杂任务往往可以分解为若干相对独立的子任务。这些子任务之间并不需要共享全部上下文——它们只需要知道自己该做什么,以及把结果交回给谁。如果所有子任务都由同一个 Agent 串行执行,所有的中间过程都会堆进同一个上下文窗口,相互干扰,越到后期模型的注意力越分散。

举个例子,假设父任务是"重构整个认证模块",其中一步需要深入分析现有代码的调用关系,这个分析过程需要读十几个文件、反复搜索,产生大量中间输出。如果让父 Agent 亲自完成这个分析,几十条工具调用记录会直接堆进父上下文——这些内容对父任务的后续步骤毫无用处,却要一直占着上下文窗口,干扰父 Agent 对整体任务的判断。但实际上,父任务真正需要的只是结论:“认证逻辑分散在 auth.pymiddleware.pyutils.py 三处,入口是 verify_token()。”

所以更适合的工作模式是分工:父 Agent 通过 task 工具把具体子任务外包给独立的子 Agent,子 Agent 在完全独立的上下文里执行,只把最终结论返回给父 Agent,中间过程全部丢弃、不暴露执行细节。这样每个 Agent 的上下文始终保持聚焦,主 Agent 不会被子任务的执行噪声干扰。

具体实现上,父 Agent 多了一个 task 工具:

1
2
3
4
5
6
{"name": "task", 
"description": "Spawn a subagent with fresh context. It shares the filesystem but not conversation history.",
"input_schema": {"type": "object",
"properties": {"prompt": {"type": "string"},
"description": {"type": "string", "description": "Short description of the task"}},
"required": ["prompt"]}}

父 Agent 调用 task 时,独立启动一个子 Agent,给它一个全新的 messages=[],让它独立跑完整个执行循环,最后只取最后一条文本消息作为 tool_result 返回给父 Agent:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def run_subagent(prompt: str) -> str:
sub_messages = [{"role": "user", "content": prompt}] # 全新上下文
for _ in range(30):
response = client.messages.create(
model=MODEL, system=SUBAGENT_SYSTEM, messages=sub_messages,
tools=CHILD_TOOLS, max_tokens=8000,
)
sub_messages.append(...)
if response.stop_reason != "tool_use":
break
# 执行工具,继续循环...

# 只有最终文本返回父 Agent,子上下文全部丢弃
return "".join(b.text for b in response.content if hasattr(b, "text"))

子 Agent 的工具集故意不包含 task,无法再派生孙 Agent,防止无限递归:

1
2
CHILD_TOOLS  = [bash, read_file, write_file, edit_file]   # 无 task
PARENT_TOOLS = CHILD_TOOLS + [task] # 有 task

子 Agent 上下文独立,但共享同一个文件系统。对于需要传递复杂结构化结果的场景,可以让子 Agent 把报告写入文件,父 Agent 再用 read_file 读取,比纯文本摘要更可靠。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
sequenceDiagram
participant User
participant Parent as Parent Agent
participant Child as Subagent
participant FS as 文件系统

User->>Parent: 重构认证模块
Parent->>Parent: 规划任务

Note over Parent,Child: 派发子任务
Parent->>Child: task { prompt: "分析认证模块调用关系" }
Note over Child: 全新 messages=[], 无 task 工具

loop 子 Agent 执行循环
Child->>FS: read_file / bash 探索代码
FS-->>Child: 文件内容 / 命令输出
end

Child->>FS: write_file "analysis.md"
Child-->>Parent: 摘要文本(子上下文丢弃)

Note over Parent,Child: 父上下文保持干净
Parent->>FS: read_file "analysis.md"
FS-->>Parent: 详细分析结果
Parent->>Parent: 基于分析结果继续执行

Agent Teams:持久化协作网络

Subagent 解决了单次任务的上下文隔离问题,但它是一次性的——任务完成即销毁,无法跨任务保持状态,也无法和其他 Agent 通信。

真实的工程团队不是这样工作的。一个后端工程师和一个测试工程师会持续并行工作、互相发消息确认接口、等待对方完成后再推进下一步。Agent Teams 把这种工作模式带入 Agent 系统:每个 Teammate 是一个持久存在的 Agent,有自己的角色和独立的执行线程,空闲时等待消息,收到消息后继续工作,直到被主动关闭。

Subagent vs Teammate 的核心区别:

1
2
Subagent:  spawn → execute → return summary → 销毁       (一次性)
Teammate: spawn → work → idle → work → ... → shutdown (持久化)

邮箱通信机制

Agent 之间通过文件系统上实现基于邮箱的通信总线。每个 Teammate 对应 .team/inbox/<name>.jsonl 一个文件,Agent 发消息就是在文件中追加一行 JSON 消息体,读消息就是读取全部内容并清空 => 追加写入、一次性消费,天然的异步消息队列:

1
2
3
4
5
6
7
8
9
class MessageBus:
def send(self, sender, to, content, msg_type="message"):
msg = {"type": msg_type, "from": sender, "content": content, ...}
open(f"inbox/{to}.jsonl", "a").write(json.dumps(msg) + "\n")

def read_inbox(self, name):
messages = [json.loads(l) for l in inbox_file.read_text().splitlines()]
inbox_file.write_text("") # drain:读完即清空
return messages

消息类型共五种,覆盖从普通通信到协调决策的全部场景:

类型 用途
message 普通点对点消息
broadcast 向所有 Teammate 广播
shutdown_request 请求某个 Teammate 优雅关闭
shutdown_response 确认/拒绝关闭请求
plan_approval_response 审批/拒绝某个计划

团队持久化

Teammate 的配置和状态保存在 .team/config.json,跨会话不丢失。每个成员有 namerolestatus 三个字段,状态在 working / idle / shutdown 之间流转:

1
2
3
4
5
6
.team/config.json
{"team_name": "default",
"members": [
{"name": "alice", "role": "coder", "status": "working"},
{"name": "bob", "role": "tester", "status": "idle"}
]}

idle 的 Teammate 可以被重新 spawn,继续工作;shutdown 表示已优雅退出。

Lead 的工具集

Lead Agent 拥有五个团队管理工具,叠加在原有的文件/bash 工具之上:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 派生一个持久化 Teammate,在独立线程中启动
{"name": "spawn_teammate",
"input_schema": {"properties": {"name": {"type": "string"},
"role": {"type": "string"},
"prompt": {"type": "string"}},
"required": ["name", "role", "prompt"]}}

# 查看所有 Teammate 及其当前状态
{"name": "list_teammates",
"input_schema": {"properties": {}}}

# 向某个 Teammate 的邮箱发消息
{"name": "send_message",
"input_schema": {"properties": {"to": {"type": "string"},
"content": {"type": "string"},
"msg_type": {"type": "string",
"enum": ["message","broadcast","shutdown_request",
"shutdown_response","plan_approval_response"]}},
"required": ["to", "content"]}}

# 读取并清空 Lead 自己的邮箱
{"name": "read_inbox",
"input_schema": {"properties": {}}}

# 向所有 Teammate 广播消息
{"name": "broadcast",
"input_schema": {"properties": {"content": {"type": "string"}},
"required": ["content"]}}

Lead 的执行循环在每轮开始前会先检查自己的邮箱,把新消息注入上下文,再调用 LLM:

1
2
3
4
5
6
7
8
def agent_loop(messages):
while True:
inbox = BUS.read_inbox("lead")
if inbox:
messages.append({"role": "user", "content": f"<inbox>{json.dumps(inbox)}</inbox>"})
messages.append({"role": "assistant", "content": "Noted inbox messages."})
response = client.messages.create(...)
...

Teammate 的工具集是 Lead 的子集:有文件操作、send_messageread_inbox,但没有 spawn_teammatebroadcast,无法再派生新成员或广播,避免团队规模失控。

以下是一次完整的多 Agent 协作流程:Lead 派生 Alice(coder)和 Bob(tester),通过邮箱协调完成"实现功能 + 测试验证"的全流程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# Lead:派生团队,分配任务
spawn_teammate { name: "alice", role: "coder",
prompt: "实现 parse_config(),支持默认值,写入 config.py" }
spawn_teammate { name: "bob", role: "tester",
prompt: "等待 alice 通知后,对 config.py 跑单元测试并汇报结果" }

send_message { to: "alice", content: "开始实现,完成后通知 bob 测试" }

# Alice(独立线程,收到消息后开始工作)
read_inbox
→ [{"type": "message", "from": "lead", "content": "开始实现,完成后通知 bob 测试"}]

write_file { path: "config.py", content: "def parse_config(path, defaults=None): ..." }
send_message { to: "bob", content: "config.py 已完成,请跑测试",
msg_type: "message" }

# Bob(独立线程,收到 Alice 消息后开始工作)
read_inbox
→ [{"type": "message", "from": "alice", "content": "config.py 已完成,请跑测试"}]

bash { command: "python -m pytest tests/test_config.py -v" }
→ "3 passed in 0.12s"

send_message { to: "lead", content: "测试通过:3/3,config.py 实现正确" }

# Lead(下一轮循环轮询到 Bob 的消息)
read_inbox
→ [{"type": "message", "from": "bob", "content": "测试通过:3/3,config.py 实现正确"}]

整个过程中,Lead 的上下文只看到任务分发和最终结论,Alice 与 Bob 之间的协作细节完全发生在各自的线程和邮箱里。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
sequenceDiagram
participant User
participant Lead
participant Alice as Teammate Alice
participant Bob as Teammate Bob
participant FS as 文件系统

User->>Lead: 开发并测试 parse_config
Lead->>Alice: spawn_teammate { role: coder }
Lead->>Bob: spawn_teammate { role: tester }
Note over Alice,Bob: 各自在独立线程中运行

Lead->>FS: send_message → alice.jsonl
Note over FS: {"from":"lead","content":"开始实现,完成后通知 bob"}

Alice->>FS: read_inbox(drain)
Alice->>FS: write_file config.py
Alice->>FS: send_message → bob.jsonl
Note over FS: {"from":"alice","content":"config.py 已完成,请跑测试"}

Bob->>FS: read_inbox(drain)
Bob->>FS: bash pytest
Bob->>FS: send_message → lead.jsonl
Note over FS: {"from":"bob","content":"测试通过:3/3"}

Lead->>FS: read_inbox(轮询)
Lead->>User: 功能完成,测试全部通过

个人思考

Agent Harness 并非要完全取代工作流,而是解决工作流难以应对的复杂、开放性问题。工作流的价值在于确定性——当任务步骤明确、边界清晰、需要稳定复现时,预编排的流程依然是最可靠的选择。审批流、发布检查清单、标准化的数据 Pipeline,这些场景不需要模型"自由发挥",反而要求每一步都可预期、可追溯。

工作流真正的局限不在于形式本身,而是它把决策权前置给了设计者。当任务复杂度超出预设分支的覆盖范围,或者执行路径需要根据实时反馈动态调整时,硬编码的流程就会僵化。Agent Harness 的价值,是让模型在运行时自主决策,把人类从"写死每一个 if-else"中解放出来,转为设定边界、提供工具、监督结果。

2026 年 2 月 OpenAI 发布的博客提出了 “Harness Engineering” 这一概念[5],标志着行业认知的转向。他们在实验中让 3-7 名工程师用 Codex Agents 在 5 个月内生成约 100 万行代码和 1500 个 PR,期间开发人员没有手写一行代码,实现了约 10 倍效率提升。这个实验的核心方法论是 Agent-First——工程师的角色从直接编写代码转变为设计和监督 AI 智能体。

这也揭示了 AI 时代放权的本质:优先阐述 What 与 Why,让模型自主决策 How。不是放弃控制,而是把控制点从"每一行代码怎么写"上移到"任务边界是什么、成功标准是什么、哪些操作需要人工确认"。Harness 的责任是把这些约束清晰地传递给模型,并在关键节点拦截风险。

Agent Harness 的终极目标,不是打造一个无人干预的黑盒,而是建立一个可监督、可干预、可回滚的协作框架。在这个框架里,模型获得足够的自由度去应对复杂性,人类保留对关键决策的最终把控——这才是人与 AI 在工程领域长期共存的合理形态。

参考内容

[1] anthropics/claude-code - Github
[2] shareAI-lab/learn-claude-code - Github
[3] affaan-m/everything-claude-code - Github
[4] MoonshotAI/kimi-cli - Github
[5] 工程技术:在智能体优先的世界中利用 Codex - OpenAI