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 三种核心角色的职责界定基于上述比喻我们来精确界定三个角色Lead Agent领导智能体它是工作流的发起者和决策者。其核心职责是任务规划与分解。它接收一个宏观的、模糊的用户请求如“开发一个个人博客网站”然后将其分解为一系列具体的、可执行的子任务如“设计数据库Schema”、“编写用户认证API”、“实现前端文章列表页”。Lead不亲自执行具体任务而是扮演“大脑”和“调度器”的角色。Worker Agent工作智能体它是任务的执行者。每个Worker通常具备某一领域的专长如后端开发、前端开发、测试。它从任务队列中领取由Lead分解好的具体任务调用自己的工具和知识去完成并将结果返回。Worker是系统生产力的直接来源。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_factorylist, description前置任务ID列表) status: TaskStatus Field(defaultTaskStatus.PENDING) assigned_worker: Optional[str] Field(defaultNone, description被分配的Worker ID) result: Optional[Dict[str, Any]] Field(defaultNone, description任务执行结果) metadata: Dict[str, Any] Field(default_factorydict, 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(fError 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(fWorker {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_idtask_id, task_typetask_type, descriptiondescription, dependenciesdependencies or [], metadatametadata ) self.tasks[task_id] task print(fTask 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_idstr(uuid.uuid4()), senderscheduler, receiverchosen_worker, msg_typetask_assignment, payload{task: task.dict()}, timestamptime.time() ) await self.message_bus.publish(assignment_msg) print(fTask {task.task_id} assigned to Worker {chosen_worker})实操心得这个调度器是非常基础的版本。在实际项目中你需要考虑更多Worker状态管理Worker应该定期发送心跳调度器需要知道Worker是否存活、当前负载如何。调度策略不仅仅是“第一个可用”可能需要考虑Worker的专长权重、历史成功率、当前负载如果Worker能并行处理多个任务等。任务优先级为Task模型增加priority字段让高优先级任务优先被调度。持久化任务状态必须持久化到数据库防止系统重启后状态丢失。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(fLead 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(fLead Agent planning failed: {e}) return fPlanning 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_typetask_plan[task_type], descriptiontask_plan[description], dependenciesreal_dependencies, original_requestuser_request # 将原始请求作为元数据传递下去 ) task_id_mapping[f{task_plan[task_type]}_task_{i1}] task_id created_tasks.append(task_id) print(fLead Agent created tasks: {created_tasks}) return fTask 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(fWorker {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_idstr(uuid.uuid4()), senderself.worker_id, receiverscheduler, # 或特定的结果收集器 msg_typetask_result, payload{ task_id: task.task_id, status: TaskStatus.COMPLETED, result: research_result }, timestamptime.time() ) await self.message_bus.publish(result_msg) print(fWorker {self.worker_id} completed task {task.task_id}) # 类似的可以定义 WriteWorker, PolishWorker, FormatWorker # 它们注册不同的能力如 [write], [polish], [format]并实现各自的 execute_task 逻辑。踩坑记录Worker的实现看似简单但有几个关键点容易出错幂等性Worker处理任务必须是幂等的。因为网络问题或调度重试同一个任务可能被分配多次。Worker需要检查任务状态避免重复执行。超时与心跳Worker执行长任务时需要定期向调度器发送心跳证明自己还“活着”。同时调度器应为任务设置超时超时未完成则重新调度。结果格式标准化不同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(fSpawnManager 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(fSpawnManager 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(fUnknown 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(fMaster 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. 常见问题与实战排坑指南在实际搭建和运行过程中你几乎一定会遇到以下问题。这里给出我的排查思路和解决方案。问题现象可能原因排查步骤与解决方案任务卡在PENDING状态1. 没有符合条件的Worker注册。2. 任务依赖未满足。3. 调度器逻辑有Bug。1. 检查_worker_capabilities字典确认有Worker注册了该任务类型。2. 打印任务及其依赖项的状态确认所有前置任务是否为COMPLETED。3. 在_schedule_tasks方法中添加详细日志跟踪其决策过程。Worker收到任务但不执行1. Worker的消息处理回调函数未正确注册或存在错误。2. 消息格式不匹配反序列化失败。3. Worker内部执行逻辑阻塞或抛出未捕获异常。1. 在Worker的_register方法和消息总线的subscribe调用后添加日志确认订阅成功。2. 在handle_task_assignment函数开头打印收到的原始消息检查msg_type和payload结构。3. 用try...except包裹任务执行逻辑并记录异常日志。系统运行一段时间后变慢或卡死1. 消息队列堆积内存泄漏。2. 某个Worker陷入死循环或长时间阻塞。3. 数据库连接未释放。1. 监控消息队列长度和内存使用情况。引入消息TTL和消费者组。2. 为每个任务执行设置超时。使用asyncio.wait_for包装执行代码。3. 确保数据库连接、HTTP会话等资源在使用后正确关闭。使用连接池。Spawn的Worker无法正确注册或通信1. 新Worker进程/容器与主调度器网络不通。2. 注册消息丢失或格式错误。3. Worker工厂函数配置错误。1. 确保网络配置正确如Docker网络、K8s Service。在新Worker启动后首先尝试ping消息总线地址。2. 在新Worker的注册代码中加入重试逻辑和详细的错误日志。3. 简化测试先手动启动一个Worker进程确认它能正常注册和工作再调试Spawn逻辑。Lead Agent规划的任务依赖关系混乱1. 大模型未能理解复杂的依赖。2. 提示词Prompt对依赖关系的描述不够清晰。3. 结果解析JSON Parsing出错。1. 在Prompt中提供更具体、更结构化的依赖关系示例。例如明确要求输出“task_b依赖于task_a”。2. 引入后处理验证检查生成的依赖图中是否存在循环依赖DAG检测。3. 使用Pydantic严格验证大模型返回的JSON结构对解析失败的结果让模型重试。最后再分享一个我踩过的大坑在早期版本中我让Lead Agent在规划时直接生成任务ID然后在创建任务时使用这些ID。这导致了一个隐蔽的问题如果两次规划产生了相同的ID虽然概率低就会发生冲突。最佳实践是Lead只负责描述任务和逻辑依赖如“任务B依赖于任务A的输出”而由Harness核心调度器在创建具体Task实例时生成全局唯一的ID并负责解析和绑定这些逻辑依赖为具体的任务ID引用。这彻底解耦了规划与执行使系统更健壮。构建多Agent系统就像指挥一支交响乐团Harness就是那位指挥家。Lead、Worker、Spawn是乐谱上不同的声部和演奏技巧。从今天这个简单的原型出发你可以逐步加入更复杂的声部更多类型的Agent、更细腻的指挥技巧高级调度算法以及更可靠的乐团管理监控与运维最终演奏出复杂而协调的AI应用交响曲。 多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 三种核心角色的职责界定基于上述比喻我们来精确界定三个角色Lead Agent领导智能体它是工作流的发起者和决策者。其核心职责是任务规划与分解。它接收一个宏观的、模糊的用户请求如“开发一个个人博客网站”然后将其分解为一系列具体的、可执行的子任务如“设计数据库Schema”、“编写用户认证API”、“实现前端文章列表页”。Lead不亲自执行具体任务而是扮演“大脑”和“调度器”的角色。Worker Agent工作智能体它是任务的执行者。每个Worker通常具备某一领域的专长如后端开发、前端开发、测试。它从任务队列中领取由Lead分解好的具体任务调用自己的工具和知识去完成并将结果返回。Worker是系统生产力的直接来源。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_factorylist, description前置任务ID列表) status: TaskStatus Field(defaultTaskStatus.PENDING) assigned_worker: Optional[str] Field(defaultNone, description被分配的Worker ID) result: Optional[Dict[str, Any]] Field(defaultNone, description任务执行结果) metadata: Dict[str, Any] Field(default_factorydict, 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(fError 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(fWorker {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_idtask_id, task_typetask_type, descriptiondescription, dependenciesdependencies or [], metadatametadata ) self.tasks[task_id] task print(fTask 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_idstr(uuid.uuid4()), senderscheduler, receiverchosen_worker, msg_typetask_assignment, payload{task: task.dict()}, timestamptime.time() ) await self.message_bus.publish(assignment_msg) print(fTask {task.task_id} assigned to Worker {chosen_worker})实操心得这个调度器是非常基础的版本。在实际项目中你需要考虑更多Worker状态管理Worker应该定期发送心跳调度器需要知道Worker是否存活、当前负载如何。调度策略不仅仅是“第一个可用”可能需要考虑Worker的专长权重、历史成功率、当前负载如果Worker能并行处理多个任务等。任务优先级为Task模型增加priority字段让高优先级任务优先被调度。持久化任务状态必须持久化到数据库防止系统重启后状态丢失。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(fLead 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(fLead Agent planning failed: {e}) return fPlanning 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_typetask_plan[task_type], descriptiontask_plan[description], dependenciesreal_dependencies, original_requestuser_request # 将原始请求作为元数据传递下去 ) task_id_mapping[f{task_plan[task_type]}_task_{i1}] task_id created_tasks.append(task_id) print(fLead Agent created tasks: {created_tasks}) return fTask 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(fWorker {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_idstr(uuid.uuid4()), senderself.worker_id, receiverscheduler, # 或特定的结果收集器 msg_typetask_result, payload{ task_id: task.task_id, status: TaskStatus.COMPLETED, result: research_result }, timestamptime.time() ) await self.message_bus.publish(result_msg) print(fWorker {self.worker_id} completed task {task.task_id}) # 类似的可以定义 WriteWorker, PolishWorker, FormatWorker # 它们注册不同的能力如 [write], [polish], [format]并实现各自的 execute_task 逻辑。踩坑记录Worker的实现看似简单但有几个关键点容易出错幂等性Worker处理任务必须是幂等的。因为网络问题或调度重试同一个任务可能被分配多次。Worker需要检查任务状态避免重复执行。超时与心跳Worker执行长任务时需要定期向调度器发送心跳证明自己还“活着”。同时调度器应为任务设置超时超时未完成则重新调度。结果格式标准化不同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(fSpawnManager 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(fSpawnManager 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(fUnknown 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(fMaster 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. 常见问题与实战排坑指南在实际搭建和运行过程中你几乎一定会遇到以下问题。这里给出我的排查思路和解决方案。问题现象可能原因排查步骤与解决方案任务卡在PENDING状态1. 没有符合条件的Worker注册。2. 任务依赖未满足。3. 调度器逻辑有Bug。1. 检查_worker_capabilities字典确认有Worker注册了该任务类型。2. 打印任务及其依赖项的状态确认所有前置任务是否为COMPLETED。3. 在_schedule_tasks方法中添加详细日志跟踪其决策过程。Worker收到任务但不执行1. Worker的消息处理回调函数未正确注册或存在错误。2. 消息格式不匹配反序列化失败。3. Worker内部执行逻辑阻塞或抛出未捕获异常。1. 在Worker的_register方法和消息总线的subscribe调用后添加日志确认订阅成功。2. 在handle_task_assignment函数开头打印收到的原始消息检查msg_type和payload结构。3. 用try...except包裹任务执行逻辑并记录异常日志。系统运行一段时间后变慢或卡死1. 消息队列堆积内存泄漏。2. 某个Worker陷入死循环或长时间阻塞。3. 数据库连接未释放。1. 监控消息队列长度和内存使用情况。引入消息TTL和消费者组。2. 为每个任务执行设置超时。使用asyncio.wait_for包装执行代码。3. 确保数据库连接、HTTP会话等资源在使用后正确关闭。使用连接池。Spawn的Worker无法正确注册或通信1. 新Worker进程/容器与主调度器网络不通。2. 注册消息丢失或格式错误。3. Worker工厂函数配置错误。1. 确保网络配置正确如Docker网络、K8s Service。在新Worker启动后首先尝试ping消息总线地址。2. 在新Worker的注册代码中加入重试逻辑和详细的错误日志。3. 简化测试先手动启动一个Worker进程确认它能正常注册和工作再调试Spawn逻辑。Lead Agent规划的任务依赖关系混乱1. 大模型未能理解复杂的依赖。2. 提示词Prompt对依赖关系的描述不够清晰。3. 结果解析JSON Parsing出错。1. 在Prompt中提供更具体、更结构化的依赖关系示例。例如明确要求输出“task_b依赖于task_a”。2. 引入后处理验证检查生成的依赖图中是否存在循环依赖DAG检测。3. 使用Pydantic严格验证大模型返回的JSON结构对解析失败的结果让模型重试。最后再分享一个我踩过的大坑在早期版本中我让Lead Agent在规划时直接生成任务ID然后在创建任务时使用这些ID。这导致了一个隐蔽的问题如果两次规划产生了相同的ID虽然概率低就会发生冲突。最佳实践是Lead只负责描述任务和逻辑依赖如“任务B依赖于任务A的输出”而由Harness核心调度器在创建具体Task实例时生成全局唯一的ID并负责解析和绑定这些逻辑依赖为具体的任务ID引用。这彻底解耦了规划与执行使系统更健壮。构建多Agent系统就像指挥一支交响乐团Harness就是那位指挥家。Lead、Worker、Spawn是乐谱上不同的声部和演奏技巧。从今天这个简单的原型出发你可以逐步加入更复杂的声部更多类型的Agent、更细腻的指挥技巧高级调度算法以及更可靠的乐团管理监控与运维最终演奏出复杂而协调的AI应用交响曲。