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 开发的所有关键痛点,且有明确的生产验证与可量化效果。主要特性为:
Agent执行循环:作为整个系统的骨干核心,它实现了“感知—推理—行动—观测”的自主闭环机制,能够让模型根据每一步的执行结果进行实时反馈与迭代优化,确保任务推进的连贯性与准确性。
原子化能力集:提供Bash、文件读写、Glob搜索、Grep分析等底层原子级工具能力,Agent可通过对这些基础能力的灵活组合来解决复杂任务。支持一次调用多个工具并实现并行执行,有效提升任务处理效率。
动态技能加载:摒弃传统全量Prompt模式,采用按需加载的思路,根据具体任务需求加载对应领域知识或特定技能(如代码审查、测试生成等),既有效节省了上下文窗口资源,又显著提升了模型的任务专注度,同时具备良好的跨场景可迁移性。
上下文压缩与管理:通过智能摘要与状态剪枝技术,精准解决长程任务中常见的模型“遗忘”与上下文“污染”问题,确保模型始终聚焦于当前任务的核心信息,同时有效降低token消耗,平衡性能与成本。
人工审批机制:human-in-the-loop,建立严格的审批流程,在赋予模型一定自主决策自由度的同时,通过人工确认机制筑牢安全底线,避免违规操作与风险输出,保障生产级应用的安全性。
复杂任务编排:支持对复杂任务进行自动拆解与并行规划,让模型具备处理多步骤工程任务的宏观视野;同时引入带依赖图的任务系统,清晰梳理任务间的关联关系,确保任务推进的有序性与高效性。
多智能体协作网络:支持主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 , ) messages.append({"role" : "assistant" , "content" : response.content}) if response.stop_reason != "tool_use" : return results = [] for block in response.content: 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})
整个循环只有四步,但恰恰是这四步构成了 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_file、write_file、edit_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 : return " - pdf: Process PDF files...\n - code-review: Review code..." def get_content (self, name: str ) -> str : 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 {"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" ]}} {"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.py 、middleware.py 、utils.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 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] PARENT_TOOLS = CHILD_TOOLS + [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("" ) return messages
消息类型共五种,覆盖从普通通信到协调决策的全部场景:
类型
用途
message
普通点对点消息
broadcast
向所有 Teammate 广播
shutdown_request
请求某个 Teammate 优雅关闭
shutdown_response
确认/拒绝关闭请求
plan_approval_response
审批/拒绝某个计划
团队持久化
Teammate 的配置和状态保存在 .team/config.json,跨会话不丢失。每个成员有 name、role、status 三个字段,状态在 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 {"name" : "spawn_teammate" , "input_schema" : {"properties" : {"name" : {"type" : "string" }, "role" : {"type" : "string" }, "prompt" : {"type" : "string" }}, "required" : ["name" , "role" , "prompt" ]}} {"name" : "list_teammates" , "input_schema" : {"properties" : {}}} {"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" ]}} {"name" : "read_inbox" , "input_schema" : {"properties" : {}}} {"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_message、read_inbox,但没有 spawn_teammate 和 broadcast ,无法再派生新成员或广播,避免团队规模失控。
以下是一次完整的多 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