如果你已经写过几个能跑的 Agent 脚本大概率会撞上同一堵墙让一个 Agent 把一件多步骤的事从头做到尾它总会在中途迷失方向。任务越复杂上下文越长模型越容易忘记最初的目标工具调用也开始互相干扰。我后来把目光转向了 Harness Engineering——把 Agent 外围的控制逻辑当成真正的工程来设计而不是在 prompt 里打补丁。这篇文章就是我从零手写一个支持 Sub-Agent 的 AI Agent 框架的完整记录包括架构规划、核心循环、子任务派发、上下文隔离和事件追踪适合已经跑通基础 Agent 调用、接下来想搞清多 Agent 协作到底该怎么落地的开发者。1. 先说清楚Harness 到底在编排什么1.1 模型只是引擎Harness 才是驾驶舱很多人对 Agent 有个误解Agent 就是那个特别聪明的大模型。真正跑过几天复杂任务的人都知道模型只负责生成“下一步动作”而决定什么时候调用工具、工具结果往哪里放、上下文保留多少、任务失败之后怎么处理、任务要不要拆给其他 Agent 去并行处理——这些全部发生在模型外面的一层控制逻辑里。业界最近把这一层叫 Harness。Harness Engineering 的核心就是把这一层控制逻辑当作工程来做而不是在 prompt 里碰运气。我习惯用一个类比模型像刚入职的员工能力很强但完全不知道公司的汇报线、审批流程和项目管理系统。你给他一个好脑子流程混乱他照样把事情办砸。Harness 就是那套流程它不负责具体干活负责让干活的人始终知道自己在哪个项目里、该调用什么资源、交付给谁、交付标准是什么。所以你会发现两个团队用同一个模型一个做出的是玩具 demo另一个做出的是稳定运行的生产级系统差异几乎全在 Harness 这一层。1.2 为什么单 Agent 会撞墙上下文稀释拿一个实际任务举例让一个 Agent 做某新品市场调研。它需要搜索竞品、抓取用户评论、整理价格区间、最后写一份报告。如果只有一个 Agent 和一个上下文窗口流程会是用户下达主任务 → 调用搜索工具 → 搜索结果进入上下文 → 再调用网页抓取 → 抓取内容进入上下文 → 再总结网页内容……十几轮之后最早的全局目标已经在上下文里被挤到边缘位置模型继续产出的内容越来越碎。更麻烦的是工具返回的长文本会持续侵占有限的窗口等到真正生成报告时模型可能已经忘记了最初要回答的问题。用数字估算一下假设模型上下文窗口是 8K初始指令占 1K每轮工具调用占 0.5K每轮工具返回结果占 2K。跑到第 6 轮的时候常驻内容已经到了 6K 左右留给模型集中推理新信息的空间只剩 2K。任务越长单 Agent 的表现下降得越明显这不是模型能力不够是你让它在一个越来越脏的房间里找东西。Sub-Agent 要解决的核心问题就在这里把一个大任务拆成边界清晰的小任务交给相互隔离的子 Agent 去执行。每个 Sub-Agent 有自己的独立上下文窗口只处理一个子问题父 Agent 拿回来的是干净的摘要而不是一大坨原始对话。这就是“支持 Sub-Agent 的 AI Agent 框架”存在的根本理由。2. 从需求到图纸框架的四层结构2.1 模块划分先别写抽象类先定目录我见过不少人一上来就设计一堆抽象基类结果越写越复杂最后自己都不知道哪个类该负责什么。我的经验是先明确框架必须做的四件事对话循环、任务模型、工具管理、事件分发。对应下来就是 runtime、task、tools、events 四个核心模块。agent_harness/ ├── runtime.py # Agent 主循环、执行入口 ├── agent.py # BaseAgent / ReActAgent / SubAgent ├── task.py # Task / TaskState / AgentResult ├── message.py # Message / ToolCall / ToolResult ├── tools.py # ToolRegistry / ToolSpec / call_sub_agent ├── context.py # ContextBuilder、预算分配 ├── events.py # EventBus / Event └── cli.py # 最小演示入口这个框架不依赖任何重型的 Agent 编排库只依赖模型 API 和 Python 标准库。每个文件都保持在 100 到 200 行核心逻辑加起来一千行左右。目录定了之后你会发现后续加功能都有明确的位置想加记忆模块就塞进 context.py想加日志追踪就挂在 events.py 上不会出现改一个功能要动三个文件的混乱局面。2.2 核心数据模型把 Task 当作一等公民做聊天机器人的人习惯把一切建模成“消息”但做 Multi-Agent 框架时如果只有消息父 Agent 和子 Agent 之间的边界很难表达。我建议引入 Task 对象作为一等公民它要携带自己的目标、层级、可用工具白名单、父任务 ID 和汇总结果。class TaskState(Enum): PENDING pending RUNNING running SUCCEEDED succeeded FAILED failed CANCELLED cancelled dataclass class Task: task_id: str parent_task_id: str | None goal: str depth: int tool_names: list[str] state: TaskState TaskState.PENDING result: AgentResult | None None为什么一开始就要定义这些因为父 Agent 拿到的不是原始对话轮次而是一个结构化结果。没有这个结构后面做汇总、重试、日志记录都会变得很难受。另外消息本身也需要结构化系统、用户、助手、工具结果四类消息要区分开工具调用用 ToolCall 对象表示包含 id、name、arguments 三个字段。模型返回的是结构化的动作描述而不是编好的字符串这样 Harness 可以直接解析执行不需要从自然语言里猜。3. 编排循环主 Agent 如何派发和回收 Sub-Agent3.1 最小可运行的主循环Runtime 的核心是一个 while 循环每轮做四步组装该 Agent 的上下文 → 请求模型 → 解析模型输出中的行动计划 → 执行工具或将结果注入消息。最关键的设计决策是把 Sub-Agent 的派发实现成一个特殊工具调用这样主循环本身不需要感知“子 Agent”的细节它只需要执行一个名为 call_sub_agent 的工具。async def run(self, task: Task) - AgentResult: messages self.ctx_builder.build(task) for step in range(self.max_steps): resp await self.llm.chat(messages, toolsself.tool_specs) messages.append(resp.message) if resp.tool_calls: for tc in resp.tool_calls: result await self.dispatcher.dispatch(tc, task) messages.append(result.to_message()) elif resp.finish_reason finish: return self._finalize(task, messages, resp.text) self._check_budget(task, messages) return AgentResult(statusmax_steps_exceeded, summary..., ...)这个循环只负责单个 Agent 的执行。子 Agent 的创建发生在 dispatcher 里当它看到 tool call 的 name 是 call_sub_agent 时就创建一个新的 Task 对象开一个新的 runtime 去执行执行完毕后返回格式化结果。主循环完全不用关心子任务的内部细节这让整个框架的复杂度下降了一个量级。3.2 call_sub_agent把派生子任务包装成普通工具给模型的工具 schema 长这样{ name: call_sub_agent, description: 创建一个子代理来执行一个边界清晰的子任务。适用于需要独立调研、计算或抓取的任务。, parameters: { goal: { type: string, description: 子代理要完成的具体目标需包含背景与验收标准 }, tool_names: { type: array, items: {type: string}, description: 允许子代理使用的工具白名单 }, expected_output: { type: string, description: 期望返回的摘要格式例如结论/证据/建议 } } }这里有个容易忽略的设计模型并不会直接调用子 Agent它只是向 Harness 描述“我希望有一个子代理去做某事”由 Harness 决定允许不允许、怎么执行。Agent 内部永远不直接 new 一个 runtime而是通过工具接口暴露能力。这样做有三个好处权限控制可以集中在 dispatcher 里做深度限制可以做在这里后续想把子 Agent 放到线程池或独立进程里执行也只需要改 dispatcher不需要动模型和 prompt。3.3 回收与降级父 Agent 只拿到干净的摘要Sub-Agent 跑完之后要把结果带回父上下文但必须带加工过的信息而不是原始消息堆。我设计的 AgentResult 长这样dataclass class AgentResult: status: TaskState summary: str # 给父 Agent 看的结论 artifacts: dict # URL、表格、文件路径等 token_usage: TokenUsage error: str | None None父 Agent 在下一次循环时会收到一条 tool result内容是“子任务 8aa1 已完成summary 是……artifacts 是……”。Prompt 层面还应该明确告诉父 Agentsummary 是二手信息如果需要原始细节可以要求子 Agent 补充但绝不能凭空脑补。我实测中发现模型有时候会强行脑补子任务没有返回的细节所以我在 prompt 里加了一句不得编造未出现在子代理结果中的数据。这句话虽然简单但确实避免了报告里出现假数据。4. 上下文与工具隔离让 Sub-Agent 不被父任务噪音干扰4.1 上下文预算每一层该带什么多 Agent 框架最大的隐形问题不是功能缺失而是 token 成本失控。我的做法是每次请求前先算清楚预算让每一层只带该带的东西。层级固定内容可变内容大致预算父 Agent全局目标、当前计划、已有子任务摘要最新工具结果3~4KSub-Agent子任务说明、验收标准、允许工具列表每轮回传的工具结果2~4K汇总层所有子结果摘要 artifacts 路径模型输出的最终报告4~6K举个例子三个 Sub-Agent每个跑 4 轮每轮上下文约 2K 结果和 0.5K 往返总开销大约是 3 × 4 × 2.5 30K token。如果不用 Sub-Agent把全部中间结果堆给父 Agent父 Agent 上下文到第 4 轮就可能到 30K而模型单次只能处理 8K任务直接失败。所以预算先行不是优化而是必要设计。我在 ContextBuilder 里放了 token_budget 参数超出预算后强制子 Agent 停止并返回已完成部分的摘要至少保住阶段性成果。4.2 工具白名单不是所有工具都能下放如果子 Agent 能看到全部工具会出现两个问题一是 prompt 里 tool 描述占用的 token 成倍增长二是安全边界模糊。一个会写文件的工具被负责网页调研的子 Agent 拿到它可能在意外中覆盖本地文件。所以工具注册进来时就要带上权限元数据。dataclass class ToolSpec: name: str description: str parameters: dict safe_for_sub_agent: bool False # 默认不允许下放 permission: str | None None # read_only / network / fs_write在创建 Sub-Agent 时它的 tool_names 白名单会被严格过滤只保留 safe_for_sub_agentTrue 的工具。父 Agent 需要显式在 call_sub_agent 参数里写出子任务允许的工具名。我的默认策略是所有工具都不下放宁可让父 Agent 多调几次也不要让子 Agent 越权。这个策略后来救过我很多次尤其是那些带文件写入或删除能力的工具一旦被子 Agent 误调用后果很难预判。5. 事件系统与可观测性没有它 Debug 会疯掉5.1 为什么每个关键节点都要发事件写单个 Agent 循环时调试靠 print 可以勉强撑住但变成父子结构之后print 完全不够。子 Agent 可能在协程里跑也可能跑在独立线程里父 Agent 在等待结果时任何一个环节失败没有追踪信息根本不知道是哪一层出了问题。我的做法是全局引入一个事件总线所有关键节点都发事件task.created、task.started、task.completed、task.failedsub_agent.requested、sub_agent.launched、sub_agent.summary_readytool.call_started、tool.call_finished、tool.errorcontext.over_budget、loop.step事件统一结构{type, agent_id, task_id, trace_id, ts, payload}。trace_id 在创建 Sub-Agent 时继承父 Agent 的 id并在子 Agent 里追加自己的路径比如 root/sub_8aa1。这样日志天然形成一棵树从顶层任务展开每个子任务都有独立分支。排查问题时顺着 trace_id 往下走基本能在几分钟内定位到具体是哪一层、哪个工具出的问题。5.2 事件系统的工程量其实很小很多人一听事件系统就觉得是大工程其实不是。用 asyncio.Queue 实现一个极简 EventBus 也就五十行左右class EventBus: def __init__(self): self._subscribers: list[Callable[[Event], Awaitable[None]]] [] def subscribe(self, handler): self._subscribers.append(handler) async def publish(self, event: Event): for handler in self._subscribers: await handler(event)CLI 和后续想做的 Web UI 都挂在事件总线上。CLI 收到 sub_agent.launched 就打一行缩进日志收到 sub_agent.summary_ready 就打印摘要收到 task.completed 就输出最终报告。业务层完全不关心谁在监听只需要按事件类型过滤。将来如果要把任务状态实时推给前端也只需要增加一个 EventBus 的消费者把事件翻译成 WebSocket 消息。这套设计最大的收益在于框架内部出问题时可以快速定位。我有一回遇到子任务卡死父 Agent 一直等不到 summary看日志发现是子 Agent 命中了 tool.error 之后没有把错误写入 result而是直接把异常抛了出来事件系统当场就暴露了这个薄弱点。6. 踩坑实录递归失控、上下文串味和错误传染6.1 Sub-Agent 无限递归深度要先写死第一次跑出多 Agent 方案时我让子 Agent 也能调用 call_sub_agent结果出现了 A 派 B、B 派 C、C 派 D 的链式反应最深跑到 7 层token 烧掉一大半才被我手动打死。原因是模型看到“可以派子任务”就倾向于不断往下拆分越拆越细没有停止机制。之后我加了两条规则max_depth 默认 3超过深度时 call_sub_agent 工具直接返回“不允许继续派生子任务请自行完成”每次派发前校验 depth超限直接拒绝。注意深度计数要从父链路累计而不是只算当前这一层否则 A 派 B 算一层B 派 C 又算一层实际链路早就绕晕了。6.2 上下文串味不共享可变对象是底线有一段时间我为了让所有 Agent 访问方便让它们共享一个全局 context 对象。结果子 Agent A 往里面写了一条临时搜索结果子 Agent B 在另一个任务里读到这条数据当成自己的调研结果写进了报告。这个坑极其隐蔽因为最终报告表面上看起来没有问题数据来源却是错的。修复方案是每个 Task 创建时从父级 context 生成一份独立的初始上下文之后的写入只发生在自己的 context 切片里父 Agent 只接收结构化结果。简单说子 Agent 之间是邮箱式通信不能是共享黑板式通信。只读的全局常量可以共享因为不会变化任何可变的部分必须隔离。6.3 错误传染子任务失败不应拖垮整个编排子 Agent 调搜索工具超时或者某个目标网站抓取失败都是正常情况不应该导致整个框架崩溃。我在 runtime 里做了严格的异常捕获包装任何工具异常都被转换成 ToolResult(statuserror, error_message...)喂回给模型框架本身不往上抛异常。如果子 Agent 连续失败 N 次我一般设 2 到 3 次就把它标记为 FAILED把失败摘要回传给父 Agent由父 Agent 决定是重试、跳过该子任务继续还是让整个任务降级完成。状态机里必须显式处理 FAILED 这个分支否则你会在运行日志里看到一堆没有定义的状态残留特别难排查。6.4 Token 成本审计上线前先跑一次估算跑多了自然会关注钱。我在每次任务结束时会打印 token 小计和总计并允许设置 max_total_tokens比如 60K。超过之后框架进入只读总结模式所有 Agent 不再允许调用昂贵工具只允许基于已有信息生成最终报告。这个机制帮我省了不少不必要的开支。计算公式不复杂总消耗约等于每一轮的输入上下文 tokens 加输出 tokens 加工具结果 tokens这些在模型 API 返回的 usage 字段里都有按 trace_id 汇总即可。我把这个统计挂在事件系统里每轮 step 结束更新一次CLI 上可以实时看到成本变化。如果让我给一个最小起步路径我建议这样第一周先不写任何抽象类用 100 行代码跑通单 Agent 加工具循环第二周把 Task 状态机补上第三周再实现 call_sub_agent 的派发和回收最后再补事件系统。每一步都能跑不需要等整个框架写完才看到效果。我自己也是踩完递归失控、上下文串味和错误传染这三个坑之后才重新整理出这套结构的。如果你正在往这个方向走希望这篇能帮你少烧几万 token。