多Agent系统核心协作模式：Lead、Worker与Spawn架构实战解析

1. 项目概述：从零构建多Agent协作系统的核心枢纽

最近在社区里看到不少朋友对多Agent系统的实现跃跃欲试，但往往在搭建好一两个独立的智能体后，就卡在了如何让它们真正“协同工作”这个坎上。大家可能已经用LangChain、AutoGen或者一些开源框架搭建了能处理单一任务的Agent，但当任务变得复杂，需要多个Agent像一支团队一样接力或并行处理时，整个系统就容易变得混乱不堪：消息不知道发给谁，任务状态难以追踪，资源竞争导致死锁……这感觉就像你组建了一个全明星团队，但没给他们配项目经理和协作流程，结果大家各自为战，效率反而更低。

这正是“Harness”要解决的问题。你可以把它理解为多Agent系统的“操作系统内核”或“团队调度中心”。它不替代Agent本身的“大脑”（即大模型推理能力），而是专注于解决协作中的“体力活”和“管理难题”。本次连载聚焦的“Lead”、“Worker”与“Spawn”模式，是Harness中三种经典且核心的协作范式，理解了它们，你就掌握了设计多Agent工作流的关键钥匙。无论你是想实现一个能自动分解复杂任务的智能助手，还是构建一个模拟软件公司各职能部门的数字团队，这套模式都能提供清晰的架构蓝图。接下来，我将结合具体的实现思路和踩坑经验，带你彻底搞懂这三种角色如何运作，并手把手教你搭建一个可运行的原型。

2. 核心理念拆解：Harness是什么，以及为什么需要Lead/Worker/Spawn

在深入代码之前，我们必须先统一思想：Harness到底扮演什么角色？很多人会把它和Agent框架（如LangChain）混淆。简单来说，Agent框架关心的是“单个智能体如何思考与行动”，它封装了与大模型对话、工具调用、记忆管理等能力。而Harness关心的是“多个智能体如何组织与交互”，它负责路由消息、协调任务、管理生命周期和监控状态。

想象一个开发团队：框架给了每个程序员（Agent）编程技能（LLM调用）和工具箱（Tools）。而Harness则是公司的项目管理办公室（PMO）和IT基础设施部，它制定了任务从产品经理（Lead）下发到开发（Worker）的流程，规定了何时需要招聘新人（Spawn），并确保大家不会同时修改同一份代码（资源冲突）。

2.1 三种核心角色的职责界定

基于上述比喻，我们来精确界定三个角色：

1. Lead Agent（领导智能体）：它是工作流的发起者和决策者。其核心职责是任务规划与分解。它接收一个宏观的、模糊的用户请求（如“开发一个个人博客网站”），然后将其分解为一系列具体的、可执行的子任务（如“设计数据库Schema”、“编写用户认证API”、“实现前端文章列表页”）。Lead不亲自执行具体任务，而是扮演“大脑”和“调度器”的角色。

2. Worker Agent（工作智能体）：它是任务的执行者。每个Worker通常具备某一领域的专长（如后端开发、前端开发、测试）。它从任务队列中领取由Lead分解好的具体任务，调用自己的工具和知识去完成，并将结果返回。Worker是系统生产力的直接来源。

3. Spawn（孵化/动态创建）：这是一种特殊的机制，而非一个常驻角色。它指的是系统在运行时，根据当前工作负载或任务特性，动态地创建或销毁Agent实例的能力。例如，当Lead发现需要同时处理10个数据清洗任务时，它可以指令Harness“孵化”出5个专门的数据清洗Worker来并行处理，任务完成后，这些临时Worker可以被回收以释放资源。Spawn机制是实现弹性伸缩和资源优化的关键。

2.2 为什么这种模式优于简单链式调用？

你可能会问，我用LangChain的SequentialChain把几个Agent串起来不行吗？对于简单、线性的流程，当然可以。但面对复杂场景，Lead/Worker/Spawn模式的优势就凸显了：

• 动态性与适应性：Spawn机制允许系统根据需求动态调整“兵力”，应对突发的高负载或处理异构子任务。
• 职责分离与可维护性：Lead只关心“做什么”和“谁来做”，Worker只关心“怎么做”。这种分离使得系统更容易理解、调试和扩展。要新增一个任务类型，往往只需要增加一个新的Worker，而不必改动核心调度逻辑。
• 更好的错误处理与重试：当某个Worker任务失败时，Harness可以在Lead的指导下，将任务重新分配给另一个同类Worker，甚至触发Spawn一个新的实例来重试，实现了任务级别的容错。
• 资源优化：通过Spawn机制，可以实现Agent实例的池化管理，避免大量Agent常驻内存，尤其在使用昂贵的GPU资源运行本地大模型时，这一点至关重要。

理解了这些，我们就从纸上谈兵进入实战环节。下面我将以一个“智能内容创作团队”为例，展示如何从零实现这套系统。

3. 系统架构与核心模块实现

我们将构建一个能够自动完成“调研-撰写-润色-发布”闭环的内容创作多Agent系统。整体架构如下图所示（概念图）：

[用户请求] -> [Lead Agent] -> [任务分解队列] | v [Harness核心] <---> [Worker池： 调研员、撰稿人、润色师、发布员] | v [结果聚合] -> [最终输出]

这个系统中，Lead Agent负责解析如“写一篇关于多Agent系统架构的科普文章”这样的请求，并将其分解为“调研关键词”、“撰写初稿”、“润色校对”、“生成发布格式”等子任务。Harness核心负责管理这些任务的派发、执行和状态同步。各类Worker则各司其职。

3.1 第一步：定义任务与消息协议

多Agent协作的首要问题是通信。我们必须设计一套所有Agent都能理解的消息格式。这里我推荐使用基于Pydantic的模型来定义，它清晰且易于验证。

from enum import Enum from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class TaskStatus(str, Enum): PENDING = "pending" ASSIGNED = "assigned" IN_PROGRESS = "in_progress" COMPLETED = "completed" FAILED = "failed" class Task(BaseModel): """任务单元，Harness调度的基本单位""" task_id: str = Field(..., description="唯一任务ID") task_type: str = Field(..., description="任务类型，如'research', 'write', 'polish'") description: str = Field(..., description="任务详细描述") dependencies: List[str] = Field(default_factory=list, description="前置任务ID列表") status: TaskStatus = Field(default=TaskStatus.PENDING) assigned_worker: Optional[str] = Field(default=None, description="被分配的Worker ID") result: Optional[Dict[str, Any]] = Field(default=None, description="任务执行结果") metadata: Dict[str, Any] = Field(default_factory=dict, description="额外元数据") class AgentMessage(BaseModel): """Agent间通信的基本消息格式""" msg_id: str sender: str # 发送者Agent ID receiver: str # 接收者Agent ID 或 'broadcast' msg_type: str # 如 'task_assignment', 'task_result', 'spawn_request' payload: Dict[str, Any] # 消息内容 timestamp: float

注意：在消息设计中，task_id和msg_id最好使用UUID或具有足够熵的字符串，避免在分布式或高并发环境下产生冲突。dependencies字段是实现有向无环图（DAG）任务流的关键，它让Lead可以描述复杂的任务依赖关系。

3.2 第二步：实现Harness核心——消息总线与任务调度器

Harness的核心是一个消息总线（Message Bus）和一个任务调度器（Task Scheduler）。消息总线负责可靠地传递AgentMessage，而调度器则管理Task的生命周期。

这里我们实现一个简单的基于内存的消息总线和调度器。在生产环境中，你可能需要引入Redis、RabbitMQ等消息队列和数据库。

import asyncio import uuid import time from collections import defaultdict from typing import Callable, Dict, List class SimpleMessageBus: """一个简单的内存消息总线，用于演示。生产环境应替换为成熟的消息中间件。""" def __init__(self): self._subscribers: Dict[str, List[Callable]] = defaultdict(list) self._message_queue = asyncio.Queue() async def publish(self, message: AgentMessage): """发布消息到总线""" await self._message_queue.put(message) async def subscribe(self, agent_id: str, callback: Callable[[AgentMessage], None]): """订阅消息。当消息的receiver为该agent_id或为'broadcast'时，触发callback。""" self._subscribers[agent_id].append(callback) async def run(self): """启动消息总线的事件循环""" while True: message = await self._message_queue.get() # 处理广播消息 if message.receiver == "broadcast": for callbacks in self._subscribers.values(): for cb in callbacks: asyncio.create_task(self._safe_callback(cb, message)) # 处理指定接收者的消息 elif message.receiver in self._subscribers: for cb in self._subscribers[message.receiver]: asyncio.create_task(self._safe_callback(cb, message)) self._message_queue.task_done() async def _safe_callback(self, callback, message): try: await callback(message) if asyncio.iscoroutinefunction(callback) else callback(message) except Exception as e: print(f"Error in message callback: {e}") class TaskScheduler: """任务调度器，维护任务状态，并根据策略分配任务给Worker""" def __init__(self, message_bus: SimpleMessageBus): self.tasks: Dict[str, Task] = {} self.message_bus = message_bus self._worker_capabilities: Dict[str, List[str]] = {} # worker_id -> [task_type1, task_type2] def register_worker(self, worker_id: str, capabilities: List[str]): """注册Worker及其能处理的任务类型""" self._worker_capabilities[worker_id] = capabilities print(f"Worker {worker_id} registered with capabilities: {capabilities}") def create_task(self, task_type: str, description: str, dependencies: List[str] = None, **metadata) -> str: """创建一个新任务""" task_id = str(uuid.uuid4()) task = Task( task_id=task_id, task_type=task_type, description=description, dependencies=dependencies or [], metadata=metadata ) self.tasks[task_id] = task print(f"Task created: {task_id} - {task_type}") # 创建任务后，尝试调度 asyncio.create_task(self._schedule_tasks()) return task_id async def _schedule_tasks(self): """调度逻辑：找到可运行（依赖已满足）的PENDING任务，并分配给空闲的Worker""" for task in self.tasks.values(): if task.status != TaskStatus.PENDING: continue # 检查依赖是否全部完成 if all(self.tasks[dep_id].status == TaskStatus.COMPLETED for dep_id in task.dependencies): # 寻找有能力且空闲的Worker（这里简化处理，假设Worker通过心跳表示空闲） # 实际项目中，需要更复杂的负载均衡算法 suitable_workers = [ wid for wid, caps in self._worker_capabilities.items() if task.task_type in caps ] if suitable_workers: # 简单策略：选择第一个可用Worker chosen_worker = suitable_workers[0] task.status = TaskStatus.ASSIGNED task.assigned_worker = chosen_worker # 通过消息总线发送任务分配消息 assignment_msg = AgentMessage( msg_id=str(uuid.uuid4()), sender="scheduler", receiver=chosen_worker, msg_type="task_assignment", payload={"task": task.dict()}, timestamp=time.time() ) await self.message_bus.publish(assignment_msg) print(f"Task {task.task_id} assigned to Worker {chosen_worker}")

实操心得：这个调度器是非常基础的版本。在实际项目中，你需要考虑更多：

1. Worker状态管理：Worker应该定期发送心跳，调度器需要知道Worker是否存活、当前负载如何。
2. 调度策略：不仅仅是“第一个可用”，可能需要考虑Worker的专长权重、历史成功率、当前负载（如果Worker能并行处理多个任务）等。
3. 任务优先级：为Task模型增加priority字段，让高优先级任务优先被调度。
4. 持久化：任务状态必须持久化到数据库，防止系统重启后状态丢失。

3.3 第三步：实现Lead Agent——任务规划与分解者

Lead Agent是整个系统的“指挥官”。它通常由一个能力较强的大模型（如GPT-4）驱动，负责理解用户意图并进行任务分解。

import openai # 或其他大模型API from typing import List class LeadAgent: def __init__(self, agent_id: str, model_name: str, scheduler: TaskScheduler): self.agent_id = agent_id self.model_name = model_name self.scheduler = scheduler # Lead Agent也需要在消息总线注册，以接收任务完成通知，进行后续规划 # 此处代码省略，详见下文完整流程 async def handle_user_request(self, user_request: str) -> str: """处理用户原始请求，生成任务DAG""" print(f"Lead Agent ({self.agent_id}) received request: {user_request}") # 步骤1: 调用大模型进行任务规划 planning_prompt = f""" 你是一个项目规划专家。请将以下用户请求分解为一系列具体的、可执行的任务。 任务类型必须是以下几种之一：research（调研）、write（撰写）、polish（润色）、format（格式排版）。 请分析任务之间的依赖关系，并以JSON格式输出。 用户请求：{user_request} 输出格式示例： {{ "tasks": [ {{ "task_type": "research", "description": "调研关于XXX的关键概念、最新发展和争议点", "dependencies": [] }}, {{ "task_type": "write", "description": "基于调研结果，撰写一篇关于XXX的科普文章初稿，要求结构清晰", "dependencies": ["research_task_id_placeholder"] }} ] }} """ # 调用大模型API (此处为示意，需替换为实际调用) try: # response = await openai.ChatCompletion.acreate(...) # plan_json = parse_response(response) # 为演示，我们模拟一个固定输出 plan_json = { "tasks": [ {"task_type": "research", "description": "调研多Agent系统的核心架构模式、Lead/Worker/Spawn概念及其应用场景", "dependencies": []}, {"task_type": "write", "description": "撰写一篇介绍多Agent系统Harness中Lead, Worker, Spawn模式的博客文章初稿", "dependencies": ["research_task_1"]}, {"task_type": "polish", "description": "对初稿进行润色，优化逻辑流畅度、技术准确性和语言可读性", "dependencies": ["write_task_1"]}, {"task_type": "format", "description": "将润色后的文章转换为Markdown格式，并添加合适的标题、代码块和列表", "dependencies": ["polish_task_1"]} ] } except Exception as e: print(f"Lead Agent planning failed: {e}") return f"Planning failed: {e}" # 步骤2: 将规划结果转化为具体的Task，并提交给调度器 task_id_mapping = {} created_tasks = [] for i, task_plan in enumerate(plan_json["tasks"]): # 处理依赖关系：将规划中的占位符ID替换为真实的前置任务ID real_dependencies = [] for dep in task_plan.get("dependencies", []): if dep in task_id_mapping: # 假设依赖的是之前已创建任务的ID映射 real_dependencies.append(task_id_mapping[dep]) # 更复杂的实现可能需要解析依赖描述，这里简化处理 task_id = self.scheduler.create_task( task_type=task_plan["task_type"], description=task_plan["description"], dependencies=real_dependencies, original_request=user_request # 将原始请求作为元数据传递下去 ) task_id_mapping[f"{task_plan['task_type']}_task_{i+1}"] = task_id created_tasks.append(task_id) print(f"Lead Agent created tasks: {created_tasks}") return f"Task planning completed. Created {len(created_tasks)} tasks. Master task ID: {created_tasks[0] if created_tasks else 'None'}"

注意事项：Lead Agent的规划能力高度依赖大模型的表现。对于复杂或专业性极强的领域，可能需要提供更详细的系统提示词（System Prompt），甚至让模型以特定的结构化格式（如YAML）输出。同时，规划结果应该有一个验证或确认机制，比如让用户审核生成的任务列表，或者设置一个“审核员”Agent来检查规划的合理性，避免“垃圾进，垃圾出”。

3.4 第四步：实现Worker Agent——任务执行专家

Worker Agent是干实事的。每个Worker都订阅消息总线，等待调度器分配任务，然后调用自己的工具或大模型能力去完成任务。

class ResearchWorker: """调研员Worker""" def __init__(self, worker_id: str, message_bus: SimpleMessageBus, scheduler: TaskScheduler): self.worker_id = worker_id self.message_bus = message_bus self.scheduler = scheduler self._register() def _register(self): """向调度器注册自己，并订阅消息""" self.scheduler.register_worker(self.worker_id, ["research"]) # 订阅任务分配消息 asyncio.create_task(self.message_bus.subscribe(self.worker_id, self.handle_task_assignment)) async def handle_task_assignment(self, message: AgentMessage): if message.msg_type != "task_assignment": return task_data = message.payload["task"] task = Task(**task_data) print(f"Worker {self.worker_id} received task: {task.task_id} - {task.description}") # 更新任务状态为进行中 task.status = TaskStatus.IN_PROGRESS # 这里应该通知调度器更新状态，简化处理，直接修改scheduler中的任务对象 # 实际项目需要通过消息总线反馈状态更新 self.scheduler.tasks[task.task_id].status = TaskStatus.IN_PROGRESS # 执行调研任务（模拟） await asyncio.sleep(2) # 模拟耗时操作 research_result = { "content": f"关于'{task.description}'的调研摘要：Lead/Worker/Spawn是三种核心协作角色...", "sources": ["模拟来源1", "模拟来源2"], "key_points": ["角色分离", "动态伸缩", "消息驱动"] } # 任务完成，发送结果消息 result_msg = AgentMessage( msg_id=str(uuid.uuid4()), sender=self.worker_id, receiver="scheduler", # 或特定的结果收集器 msg_type="task_result", payload={ "task_id": task.task_id, "status": TaskStatus.COMPLETED, "result": research_result }, timestamp=time.time() ) await self.message_bus.publish(result_msg) print(f"Worker {self.worker_id} completed task {task.task_id}") # 类似的，可以定义 WriteWorker, PolishWorker, FormatWorker # 它们注册不同的能力，如 `["write"]`, `["polish"]`, `["format"]`，并实现各自的 `execute_task` 逻辑。

踩坑记录：Worker的实现看似简单，但有几个关键点容易出错：

1. 幂等性：Worker处理任务必须是幂等的。因为网络问题或调度重试，同一个任务可能被分配多次。Worker需要检查任务状态，避免重复执行。
2. 超时与心跳：Worker执行长任务时，需要定期向调度器发送心跳，证明自己还“活着”。同时，调度器应为任务设置超时，超时未完成则重新调度。
3. 结果格式标准化：不同Worker返回的结果格式必须事先约定好，以便下游Worker（如润色Worker需要处理撰写Worker的输出）能够正确解析。可以在Task的metadata中定义结果Schema。

3.5 第五步：实现Spawn机制——动态资源管理

Spawn是Harness弹性的体现。我们可以在调度器中实现一个简单的逻辑：当某种类型的任务积压超过阈值时，自动创建新的Worker。

class SpawnManager: """管理Worker的动态创建与销毁""" def __init__(self, scheduler: TaskScheduler, message_bus: SimpleMessageBus, worker_factory: Callable): self.scheduler = scheduler self.message_bus = message_bus self.worker_factory = worker_factory # 一个能创建特定类型Worker的函数 self.active_workers: Dict[str, asyncio.Task] = {} # worker_id -> 运行任务 self._monitor_task = None async def start_monitoring(self): """启动监控，根据任务队列情况动态调整Worker数量""" self._monitor_task = asyncio.create_task(self._monitor_loop()) async def _monitor_loop(self): while True: await asyncio.sleep(10) # 每10秒检查一次 # 分析任务队列中每种类型PENDING任务的数量 pending_counts = {} for task in self.scheduler.tasks.values(): if task.status == TaskStatus.PENDING: pending_counts[task.task_type] = pending_counts.get(task.task_type, 0) + 1 # 简单的Spawn策略：如果某类任务积压超过3个，且当前活跃Worker少于5个，就Spawn一个 for task_type, count in pending_counts.items(): if count > 3: # 计算当前该类Worker的数量（简化：通过注册的能力判断） current_workers = sum(1 for caps in self.scheduler._worker_capabilities.values() if task_type in caps) if current_workers < 5: await self.spawn_worker(task_type) async def spawn_worker(self, worker_type: str): """动态创建一个指定类型的新Worker""" worker_id = f"{worker_type}_worker_{int(time.time())}" print(f"SpawnManager is spawning a new {worker_type} worker: {worker_id}") # 使用工厂函数创建Worker实例 worker_instance = self.worker_factory(worker_id, worker_type, self.message_bus, self.scheduler) # 保存对Worker的引用（如果需要后续管理） # self.active_workers[worker_id] = worker_instance # 在实际场景中，可能需要在新进程中启动Worker，这里仅作演示 async def terminate_worker(self, worker_id: str): """终止一个Worker（例如，在空闲一段时间后）""" if worker_id in self.active_workers: # 发送终止信号或取消任务 # self.active_workers[worker_id].cancel() del self.active_workers[worker_id] print(f"SpawnManager terminated worker: {worker_id}") # 同时需要从调度器注销该Worker的能力（简化处理）

核心逻辑解析：Spawn机制的核心是一个监控循环，它持续观察系统状态（如任务队列长度、Worker负载），并根据预设的策略（如阈值规则、预测算法）做出决策。这里的策略极其简单，实际项目中可能需要更复杂的算法，例如基于响应时间预测的弹性伸缩，或者考虑创建Worker的成本（如启动延迟、资源消耗）。

4. 完整工作流串联与系统启动

现在，我们把所有模块组装起来，形成一个可以运行的原型系统。

import asyncio async def main(): # 1. 初始化核心组件 message_bus = SimpleMessageBus() scheduler = TaskScheduler(message_bus) # 2. 初始化Spawn管理器（需要先定义worker工厂函数） def worker_factory(wid, wtype, bus, sched): # 根据类型创建不同的Worker if wtype == "research": return ResearchWorker(wid, bus, sched) elif wtype == "write": return WriteWorker(wid, bus, sched) # 假设已实现 # ... 其他类型 else: raise ValueError(f"Unknown worker type: {wtype}") spawn_manager = SpawnManager(scheduler, message_bus, worker_factory) # 3. 预创建一些初始Worker（预热） initial_workers = [ ResearchWorker("research_1", message_bus, scheduler), WriteWorker("write_1", message_bus, scheduler), PolishWorker("polish_1", message_bus, scheduler), FormatWorker("format_1", message_bus, scheduler), ] # 4. 初始化Lead Agent lead = LeadAgent("lead_1", "gpt-4", scheduler) # 5. 启动消息总线监听循环（在后台运行） bus_task = asyncio.create_task(message_bus.run()) # 启动Spawn监控循环 spawn_task = asyncio.create_task(spawn_manager.start_monitoring()) # 6. 模拟用户请求 user_request = "写一篇关于多Agent系统Harness中Lead, Worker, Spawn模式的科普文章，要求通俗易懂，并附带简单代码示例。" master_task_id = await lead.handle_user_request(user_request) print(f"Master task initiated: {master_task_id}") # 7. 等待一段时间，观察任务执行（在实际系统中，这里可能是等待最终结果的事件） await asyncio.sleep(30) # 8. 检查任务完成状态 completed_tasks = [t for t in scheduler.tasks.values() if t.status == TaskStatus.COMPLETED] print(f"\n=== 执行结果摘要 ===") print(f"总任务数: {len(scheduler.tasks)}") print(f"已完成: {len(completed_tasks)}") for task in scheduler.tasks.values(): print(f" - [{task.status}] {task.task_id}: {task.description[:50]}...") # 9. 清理（在实际应用中应有更优雅的关闭逻辑） bus_task.cancel() spawn_task.cancel() if __name__ == "__main__": asyncio.run(main())

运行这个脚本，你将看到Lead Agent创建任务，调度器分配任务，各个Worker依次执行，并且当任务积压时，SpawnManager可能会尝试创建新的Worker。这就完成了一个最基本的多Agent Harness系统的演示。

5. 进阶考量与生产级优化

上面的原型揭示了核心原理，但距离一个健壮的生产系统还有巨大差距。以下是几个必须考虑的进阶方向：

5.1 通信可靠性：从内存总线到消息队列

内存消息总线无法持久化，进程崩溃消息就丢了。生产环境必须使用如Redis Streams、RabbitMQ、Apache Kafka或NATS等成熟的消息中间件。它们提供了持久化、确认机制、死信队列等企业级特性。

# 伪代码示例：使用Redis作为消息总线 import redis.asyncio as redis import json class RedisMessageBus: def __init__(self, redis_url): self.redis = redis.from_url(redis_url) self.pubsub = self.redis.pubsub() async def publish(self, channel: str, message: AgentMessage): await self.redis.publish(channel, message.json()) async def subscribe(self, channel: str, callback): async for message in self.pubsub.listen(): if message['type'] == 'message': msg_data = json.loads(message['data']) await callback(AgentMessage(**msg_data))

5.2 状态持久化与可观测性

所有Task和关键事件的状态必须持久化到数据库（如PostgreSQL, MongoDB）。这不仅是为了容错，更是为了可观测性。你需要能回答：

• 当前系统中有多少任务？各自状态如何？
• 每个Worker的处理速度和成功率是多少？
• 任务的平均端到端延迟是多少？
• Spawn机制触发的频率和效果如何？

建议集成像Prometheus和Grafana这样的监控系统，为关键指标（任务队列长度、Worker数量、任务处理耗时）设置仪表盘。

5.3 更智能的调度与Spawn策略

• 基于技能的调度：Worker注册时不仅声明能处理的任务类型，还可以声明技能水平或权重。调度器可以将复杂任务分配给技能值高的Worker。
• 基于负载预测的Spawn：使用时间序列分析预测未来几分钟的任务负载，提前Spawn Worker，避免冷启动延迟影响用户体验。
• 成本感知调度：如果系统混合使用了不同成本的计算资源（如CPU Worker和昂贵的GPU Worker），调度器应优先将任务分配给成本更低的资源。

5.4 错误处理与补偿机制

• 任务重试与退避：任务失败后不应立即无限重试。应实现指数退避重试机制，并在重试一定次数后，将任务标记为FAILED并通知Lead Agent或人工介入。
• Worker健康检查与隔离：定期检查Worker的健康状况。连续失败多次的Worker应被标记为不健康并从调度池中隔离，防止其继续接收任务。
• 全局事务与回滚：对于涉及多个步骤的复杂工作流，可能需要实现类似Saga的模式。如果后续步骤失败，需要有能力触发前面步骤的补偿操作（虽然这在AI生成内容等场景中较难实现）。

6. 常见问题与实战排坑指南

在实际搭建和运行过程中，你几乎一定会遇到以下问题。这里给出我的排查思路和解决方案。

最后再分享一个我踩过的大坑：在早期版本中，我让Lead Agent在规划时直接生成任务ID，然后在创建任务时使用这些ID。这导致了一个隐蔽的问题：如果两次规划产生了相同的ID（虽然概率低），就会发生冲突。最佳实践是：Lead只负责描述任务和逻辑依赖（如“任务B依赖于任务A的输出”），而由Harness核心（调度器）在创建具体Task实例时生成全局唯一的ID，并负责解析和绑定这些逻辑依赖为具体的任务ID引用。这彻底解耦了规划与执行，使系统更健壮。

构建多Agent系统就像指挥一支交响乐团，Harness就是那位指挥家。Lead、Worker、Spawn是乐谱上不同的声部和演奏技巧。从今天这个简单的原型出发，你可以逐步加入更复杂的声部（更多类型的Agent）、更细腻的指挥技巧（高级调度算法）以及更可靠的乐团管理（监控与运维），最终演奏出复杂而协调的AI应用交响曲。