Harness工程实战进阶：团队协作与任务自治，让AI编程更高效（收藏版）

本文深入Harness工程实战，介绍了如何通过s09到s12的机制实现团队协作与任务自治。通过TeammateManager和MessageBus实现持久Agent间的通信与身份管理，利用request_id和FSM协议解决关机与计划审批的结构化协调问题，引入WORK-IDLE双阶段循环让Agent自组织地认领任务，并借助WorktreeManager实现任务的目录隔离。这些机制的核心在于通过叠加工具和注入逻辑在不变的agent_loop基础上实现复杂功能，构建了一个高效、可扩展的AI编程协作体系。

Agent 变得能"记住任务"也能"并行干活"了。但它始终只有一个——一个人扛所有活。

当项目变大，一个人忙不过来：前端和后端可以同时开发，测试和代码可以并行推进。我们需要的是一支团队——多个 Agent 各有分工、互相通信、自主协作。

s09 到 s12 就是为解决这个问题的：

这篇文章记录我学习这四个机制的过程。

s09：一个人干不完，分给队友——持久化 Agent + JSONL 邮箱

问题

回顾之前的机制：Subagent（s04）是一次性的——生成、干活、返回摘要、消亡，没有身份，没有跨调用的记忆。Background Tasks（s08）能异步跑 shell 命令，但做不了 LLM 引导的决策。

真正的团队协作需要三样东西：

1. 持久 Agent

   ——能跨多轮对话存活，有自己的状态

2. 身份和生命周期管理

   ——知道谁在工作、谁在空闲

3. Agent 之间的通信通道

   ——队友之间能发消息

解决方案：TeammateManager + MessageBus

s09 引入了两个核心组件：

TeammateManager管理团队成员的身份和生命周期。每个队友是一个独立线程中运行的完整 agent loop，状态保存在config.json中：

class TeammateManager: def __init__(self, team_dir: Path): self.dir = team_dir self.config_path = self.dir / "config.json" self.config = self._load_config() # {"members": [...]} self.threads = {}

.team/config.json存储团队名册——谁在、什么角色、什么状态：

{ "team_name": "default", "members": [ {"name": "alice", "role": "coder", "status": "working"}, {"name": "bob", "role": "tester", "status": "idle"} ]}

MessageBus提供异步通信——append-only 的 JSONL 收件箱：

.team/inbox/ alice.jsonl ← 每行一条消息，追加写入 bob.jsonl lead.jsonl

关键设计：read_inbox是drain 模式——读取全部消息后立即清空文件。读过的消息不会重复投递。

每个队友都是一个完整的 agent loop

s09 最本质的变化：每个队友在独立线程中运行自己的agent_loop：

def _teammate_loop(self, name, role, prompt): sys_prompt = f"You are '{name}', role: {role}, at {WORKDIR}." messages = [{"role": "user", "content": prompt}] for _ in range(50): # 每轮 LLM 调用前检查收件箱 inbox = BUS.read_inbox(name) for msg in inbox: messages.append({"role": "user", "content": json.dumps(msg)}) response = client.messages.create( model=MODEL, system=sys_prompt, messages=messages, tools=tools, max_tokens=8000) # ... 和 s01 一样的循环结构

和 s01 的循环完全一样——while/for+stop_reason+tool_result。唯一的区别是每轮开头多了一步read_inbox，把收到的消息注入上下文。

这又是一个"在循环上叠加机制"的实例。循环不变，通信能力通过注入实现。

消息类型

s09 定义了 5 种消息类型，为后续课程预留了扩展空间：

领导与队友

s09 采用领导-队友架构。领导（lead）是用户直接交互的主 Agent，通过spawn_teammate工具创建队友，通过send_message和broadcast分配任务：

领导有 9 个工具（基础 4 个 + spawn + list + send + inbox + broadcast），队友有 6 个工具（基础 4 个 + send + inbox）。各自独立运行，通过邮箱协调。

与 s04 Subagent 和 s08 Background Tasks 的对比

Agent 之间通过文件协调，不通过共享内存。JSONL 文件是最简单的异步消息队列——追加写入是原子操作，drain-on-read 保证不重复。

s10：队友之间要有规矩——结构化握手协议

问题

s09 中队友能通信了，但缺少结构化协调：

关机问题：直接杀线程？那会留下写了一半的文件和过期的 config.json。需要握手——领导请求，队友收尾后批准退出（或拒绝继续干）。

计划审批：领导说"重构认证模块"，队友立刻开干。高风险变更应该先过审。

解决方案：request_id 关联的 FSM

s10 发现了两个问题背后是同一个模式——请求-响应握手。用request_id关联，用有限状态机（FSM）追踪状态：

一个 FSM，两种用途。关机握手和计划审批共享同一个pending → approved | rejected状态机。

关机握手

领导发起关机请求，生成唯一request_id，通过邮箱发送：

def handle_shutdown_request(teammate: str) -> str: req_id = str(uuid.uuid4())[:8] shutdown_requests[req_id] = {"target": teammate, "status": "pending"} BUS.send("lead", teammate, "Please shut down gracefully.", "shutdown_request", {"request_id": req_id}) return f"Shutdown request {req_id} sent (status: pending)"

队友收到后，用同一个request_id响应：

if tool_name == "shutdown_response": req_id = args["request_id"] approve = args["approve"] shutdown_requests[req_id]["status"] = "approved" if approve else "rejected" BUS.send(sender, "lead", args.get("reason", ""), "shutdown_response", {"request_id": req_id, "approve": approve})

队友批准后，自己把should_exit设为True，下一轮循环退出。线程干净终止，config.json 状态更新为shutdown。

计划审批

完全相同的模式，方向相反——队友提交计划给领导审批：

def handle_plan_review(request_id, approve, feedback=""): req = plan_requests[request_id] req["status"] = "approved" if approve else "rejected" BUS.send("lead", req["from"], feedback, "plan_approval_response", {"request_id": request_id, "approve": approve, "feedback": feedback})

request_id 的精妙之处

为什么用request_id而不是直接用消息内容匹配？因为：

1. 并发安全

   ——同时有多个关机请求或计划审批，靠 request_id 区分

2. 状态追踪

   ——shutdown_requests和plan_requests两个 dict 按 ID 查询，O(1) 复杂度

3. 可扩展

   ——任何新的请求-响应协议都可以复用这个模式

# 两个追踪器，同一套 FSMshutdown_requests = {} # {req_id: {target, status}}plan_requests = {} # {req_id: {from, plan, status}}

相对 s09 的变更

用 request_id 关联的 FSM 统一所有请求-响应协议。一个pending → approved | rejected的状态机可以套用在关机、审批、甚至未来的任何协商场景上。

s11：没人分配也能自己找活干——自治 Agent

问题

s09-s10 中，队友只在被明确指派时才动。领导得给每个队友写 prompt，任务看板上 10 个未认领的任务得手动分配。

这扩展不了。

真正的自治：队友自己扫描任务看板，认领没人做的任务，做完再找下一个。不需要领导当"工头"。

解决方案：WORK-IDLE 双阶段循环

s11 给每个队友的 agent loop 增加了空闲阶段：

核心代码结构：

def _loop(self, name, role, prompt): while True: # -- WORK 阶段：标准 agent loop -- for _ in range(50): response = client.messages.create(...) if response.stop_reason != "tool_use": break # 执行工具... if idle_requested: break # -- IDLE 阶段：轮询收件箱 + 任务看板 -- self._set_status(name, "idle") resume = self._idle_poll(name, messages) if not resume: self._set_status(name, "shutdown") return self._set_status(name, "working")

空闲轮询——自组织的引擎

空闲阶段每 5 秒轮询一次，持续 60 秒。两个检查条件，满足任一就恢复工作：

def _idle_poll(self, name, messages): for _ in range(IDLE_TIMEOUT // POLL_INTERVAL): # 60/5 = 12次 time.sleep(POLL_INTERVAL) # 条件1：收到消息 inbox = BUS.read_inbox(name) if inbox: messages.append({"role": "user", "content": f"<inbox>{inbox}</inbox>"}) return True # 条件2：发现未认领任务 unclaimed = scan_unclaimed_tasks() if unclaimed: claim_task(unclaimed[0]["id"], name) messages.append({"role": "user", "content": f"<auto-claimed>Task #{unclaimed[0]['id']}: " f"{unclaimed[0]['subject']}</auto-claimed>"}) return True return False # 超时 → 关机

任务看板扫描

扫描.tasks/目录，找pending状态、无owner、无blockedBy的任务：

def scan_unclaimed_tasks() -> list: unclaimed = [] for f in sorted(TASKS_DIR.glob("task_*.json")): task = json.loads(f.read_text()) if (task.get("status") == "pending" and not task.get("owner") and not task.get("blockedBy")): unclaimed.append(task) return unclaimed

三个条件缺一不可：待做（pending）、没人认（无 owner）、能做（无阻塞）。

认领操作加了_claim_lock——两个队友同时看中同一个任务，只有一个能成功：

def claim_task(task_id: int, owner: str) -> str: with _claim_lock: task = json.loads(path.read_text()) if task.get("owner"): return f"Error: Task {task_id} already claimed" task["owner"] = owner task["status"] = "in_progress" path.write_text(json.dumps(task, indent=2))

身份重注入——对抗上下文遗忘

一个容易被忽略的细节：Agent 经过上下文压缩（s06）后可能忘了自己是谁。

s11 的解决方案——检测 messages 过短（说明发生了压缩），在开头插入身份块：

if len(messages) <= 3: messages.insert(0, { "role": "user", "content": f"<identity>You are '{name}', role: {role}, " f"team: {team_name}. Continue your work.</identity>" }) messages.insert(1, { "role": "assistant", "content": f"I am {name}. Continuing." })

为什么要同时插 user + assistant 两条？因为 API 要求 role 交替。身份声明（user）+ 确认回应（assistant）构成一个完整的对话对，后续消息可以正常追加。

相对 s10 的变更

Agent 应该自己找活干，而不是等分配。WORK-IDLE 双阶段循环让每个 Agent 成为自组织节点。空闲时轮询，发现工作就认领，做完继续找。超时就关机——资源不浪费。

s12：各干各的目录，互不干扰——Worktree 任务隔离

问题

到 s11，Agent 已经能自主认领和完成任务。但所有任务共享一个目录。

两个 Agent 同时重构不同模块——A 改config.py，B 也改config.py。未提交的改动互相污染，谁也没法干净回滚。

任务板管"做什么"但不管"在哪做"。需要给每个任务一个独立的执行空间。

解决方案：控制面 + 执行面双平面架构

s12 引入了WorktreeManager——用 git worktree 给每个任务创建独立目录：

控制面 (.tasks/) 执行面 (.worktrees/)+------------------+ +------------------------+| task_1.json | | auth-refactor/ || status: in_progress <---> | branch: wt/auth-refactor|| worktree: "auth-refactor" | task_id: 1 |+------------------+ +------------------------+| task_2.json | | ui-login/ || status: pending <---+ | branch: wt/ui-login || worktree: "ui-login" | | task_id: 2 |+------------------+ +------------------------+ | index.json (注册表) | | events.jsonl (生命周期) |

任务管目标，worktree 管目录，用 task_id 把两边关联起来。

四步操作流程

第一步：创建任务。把目标持久化到控制面：

TASKS.create("Implement auth refactor")# -> .tasks/task_1.json status=pending worktree=""

第二步：创建 worktree 并绑定任务。传入task_id，自动推进状态：

WORKTREES.create("auth-refactor", task_id=1)# -> git worktree add -b wt/auth-refactor .worktrees/auth-refactor HEAD# -> index.json 新增条目# -> task_1.json: status=in_progress, worktree="auth-refactor"

绑定逻辑同时写入两侧：

def bind_worktree(self, task_id, worktree): task = self._load(task_id) task["worktree"] = worktree if task["status"] == "pending": task["status"] = "in_progress" # 自动推进 self._save(task)

第三步：在 worktree 中执行命令。cwd指向隔离目录：

subprocess.run(command, shell=True, cwd=worktree_path, ...)

第四步：收尾。两种选择：

• worktree_keep(name)

  —— 保留目录供后续使用

• worktree_remove(name, complete_task=True)

  —— 一个调用完成拆除 + 标记任务完成

def remove(self, name, force=False, complete_task=False): self._run_git(["worktree", "remove", wt["path"]]) if complete_task and wt.get("task_id") is not None: self.tasks.update(wt["task_id"], status="completed") self.tasks.unbind_worktree(wt["task_id"]) self.events.emit("task.completed", ...)

事件流——生命周期可观测

s12 新增了EventBus，每个生命周期步骤写入.worktrees/events.jsonl：

{"event": "worktree.create.before", "task": {"id": 1}, "worktree": {"name": "auth-refactor"}, "ts": 1730000000}{"event": "worktree.create.after", "task": {"id": 1}, "worktree": {"name": "auth-refactor", "status": "active"}, "ts": 1730000001}{"event": "worktree.remove.before", "task": {"id": 1}, "worktree": {"name": "auth-refactor"}, "ts": 1730000100}{"event": "task.completed", "task": {"id": 1, "status": "completed"}, "worktree": {"name": "auth-refactor"}, "ts": 1730000101}

8 种事件类型覆盖完整生命周期：create.before/after/failed、remove.before/after/failed、keep、task.completed。

为什么用事件流而不是日志 print？因为事件流是结构化的 JSON，可以被其他程序消费。日志是人看的，事件是机器读的。

崩溃恢复

因为任务状态在.tasks/、worktree 索引在.worktrees/index.json，事件流在events.jsonl——全部在磁盘上。

会话记忆是易失的，磁盘状态是持久的。崩溃后从这三个数据源重建现场。

相对 s11 的变更

任务管目标，目录管执行，用 ID 把两边绑起来。控制面（任务板）和执行面（worktree）通过 task_id 关联，各管各的领域。崩溃恢复只需要读磁盘。

循环变了吗？

回到第一篇的核心问题。s09-s12 加了这么多机制——团队、协议、自治、隔离——agent_loop的核心结构变了吗？

还是没变。

s09：循环前加了read_inbox，消息注入后照样进入 LLM 调用。

s10：加了 shutdown_request/resp 和 plan_approval 工具，handler 多了几个，循环不变。

s11：加了 WORK-IDLE 双阶段，但 WORK 阶段就是标准循环，IDLE 阶段是 Harness 层的轮询逻辑，不影响循环结构。

s12：回到单 Agent，工具从 14 个变成 16 个（task 6 个 + worktree 7 个 + 基础 3 个），循环不变。

循环是常量，工具是变量。所有新能力——团队、协议、自治、隔离——都是通过加工具和加注入实现的。

全系列架构回顾

每个阶段的依赖关系清晰：

s09 需要 s07 的任务系统和 s08 的线程机制。s10 在 s09 上加协议。s11 在 s10 上加自治。s12 把任务系统和目录隔离结合起来。

每一个都是在前一个的基础上叠加，没有重写，没有推翻。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线科技企业深耕十二载，见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事，早已在效率与薪资上形成代际优势，我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套AI 大模型突围资料包：

• ✅ 从零到一的 AI 学习路径图
• ✅ 大模型调优实战手册（附医疗/金融等大厂真实案例）
• ✅ 百度/阿里专家闭门录播课
• ✅ 大模型当下最新行业报告
• ✅ 真实大厂面试真题
• ✅ 2026 最新岗位需求图谱

所有资料 ⚡️ ，朋友们如果有需要《AI大模型入门+进阶学习资源包》，下方扫码获取~

① 全套AI大模型应用开发视频教程

（包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点）

② 大模型系统化学习路线

作为学习AI大模型技术的新手，方向至关重要。 正确的学习路线可以为你节省时间，少走弯路；方向不对，努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划，带你从零基础入门到精通！

③ 大模型学习书籍&文档

学习AI大模型离不开书籍文档，我精选了一系列大模型技术的书籍和学习文档（电子版），它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。

④ AI大模型最新行业报告

2025最新行业报告，针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

⑤ 大模型项目实战&配套源码

学以致用，在项目实战中检验和巩固你所学到的知识，同时为你找工作就业和职业发展打下坚实的基础。

⑥ 大模型大厂面试真题

面试不仅是技术的较量，更需要充分的准备。在你已经掌握了大模型技术之后，就需要开始准备面试，我精心整理了一份大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

以上资料如何领取？

为什么大家都在学大模型？

最近科技巨头英特尔宣布裁员2万人，传统岗位不断缩减，但AI相关技术岗疯狂扩招，有3-5年经验，大厂薪资就能给到50K*20薪！

不出1年，“有AI项目经验”将成为投递简历的门槛。

风口之下，与其像“温水煮青蛙”一样坐等被行业淘汰，不如先人一步，掌握AI大模型原理+应用技术+项目实操经验，“顺风”翻盘！

这些资料真的有用吗？

这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理，现任上海殷泊信息科技CEO，其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证，服务航天科工、国家电网等1000+企业，以第一作者在IEEE Transactions发表论文50+篇，获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

以上全套大模型资料如何领取？