1. 从单兵作战到团队协作为什么我们需要多 Agent 系统如果你已经跟着这个系列一路从零开始搭建了自己的第一个 Agent那么恭喜你你已经迈出了从“玩具”走向“工具”的关键一步。但很快你就会遇到一个现实问题单个 Agent 的能力是有限的。它就像一个全栈工程师既要负责与用户对话又要调用工具查询天气、分析数据、生成图表甚至还要处理文件上传。当任务稍微复杂一点比如用户要求“分析这份销售数据找出异常点生成一份报告并给我三个优化建议最后把报告发到我的邮箱”单个 Agent 就会显得手忙脚乱逻辑臃肿响应变慢甚至因为一个环节出错导致整个流程崩溃。这就是多 Agent 系统要解决的问题。它不再是让一个“超人”去完成所有事而是组建一个分工明确的“特种部队”。在这个部队里有负责统筹指挥的“队长”Lead有负责执行具体任务的“队员”Worker还有负责动态创建新队员的“后勤官”Spawn。这种架构带来的好处是显而易见的职责分离、并行处理、容错性增强、以及可扩展性的大幅提升。一个 Agent 挂了不影响其他 Agent 的工作新任务来了可以动态创建合适的 Agent 去处理复杂的流程可以被拆解成清晰的子任务流水线。而Harness就是这套特种部队的“指挥与保障系统”。它不是 Agent 本身不负责具体的推理或工具调用逻辑。你可以把它理解为一套包裹在 Agent 核心逻辑之外的“基础设施层”或“脚手架”。它的职责是管理 Agent 的生命周期创建、调度、销毁、协调 Agent 之间的通信谁把任务交给谁、维护共享状态大家都能访问的公共信息板以及处理可能出现的异常某个队员任务失败了怎么办。没有 Harness你的多个 Agent 就像一盘散沙各自为战无法形成有效的合力。所以当我们谈论“多 Agent 研究 Harness”时我们研究的核心就是如何设计一套高效、可靠、易用的框架来让 Lead、Worker、Spawn 这三种角色协同工作从而构建出能够处理复杂现实任务的智能体团队。这不仅是学术前沿更是工程落地的关键。2. 核心角色拆解Lead、Worker 与 Spawn 的职责边界在多 Agent 系统中清晰的角色定义是协作的基石。Lead、Worker、Spawn 这三个概念并非凭空而来它们是对协作模式中不同职能的高度抽象。理解它们各自的职责和交互方式是设计 Harness 的第一步。2.1 Lead团队的指挥官与决策大脑Lead Agent 是整个多 Agent 系统的核心与起点。它通常是与用户进行直接交互的那个 Agent负责接收原始的用户请求User Query并对其进行任务规划与分解Task Planning Decomposition。它的核心职责包括意图理解与任务解析分析用户的自然语言指令理解其深层意图和最终目标。例如用户说“帮我规划一下下周的旅行”Lead 需要理解这涉及到查询天气、查找机票酒店、制定日程等多个子任务。规划与编排将宏大的目标拆解为一系列具体的、可执行的原子任务Atomic Tasks。这类似于产品经理将需求拆解成一个个开发任务。Lead 需要决定任务的执行顺序串行、并行或有依赖关系并为每个任务分配合适的 Worker。Worker 调度与任务分发Lead 不亲自执行具体任务而是根据任务类型调用或创建具备相应能力的 Worker Agent将原子任务分配给它们。它需要知道“谁能做什么”。结果汇总与决策收集各个 Worker 返回的子任务结果进行综合评估、校验和整合。如果某个 Worker 任务失败或结果不达标Lead 需要决定是重试、换人还是调整方案。最终输出与用户交互将整合后的最终结果以用户友好的方式如自然语言报告、图表、文件等返回给用户并可能进行后续的对话澄清。技术实现要点Lead 通常是一个具备强规划能力和工具调用能力的 Agent。它的“工具库”里最重要的工具可能就是“调用 Worker”或“创建 Worker”通过 Spawn。它的 Prompt 设计需要强调其“管理者”角色包含任务拆解模版、Worker 能力目录等。2.2 Worker专注的执行单元与技能专家Worker Agent 是系统的“手”和“脚”是具体任务的执行者。每个 Worker 通常被设计为只擅长某一类或某几类特定任务追求的是在垂直领域的深度与可靠性。它的核心职责包括接收并理解原子任务从 Lead 或上级 Worker 那里接收一个明确、具体的指令例如“查询北京明天下午的天气”、“计算这份 Excel 表格中 A 列的平均值”。调用专用工具或 APIWorker 内部封装了执行该任务所需的所有工具函数、API 接口或领域知识。一个“数据分析 Worker”可能集成了 Pandas、Matplotlib 等库一个“邮件发送 Worker”则封装了 SMTP 协议。执行并返回结构化结果Worker 执行任务后需要将结果以预先约定好的、结构化的格式如 JSON返回给调用者。这包括成功状态、执行结果、可能的错误信息或日志。保持无状态性理想情况为了更好的可扩展性和容错性Worker 最好设计为无状态的。即它的执行不依赖于之前的会话历史每次任务都是独立的。状态信息应由 Harness 或 Lead 来维护。技术实现要点Worker 的实现可以非常轻量甚至可以是纯函数。它的核心是一个“输入-处理-输出”的管道。在基于大语言模型的架构中Worker 的 Prompt 会非常具体限定其能力范围防止它“越权”处理其他任务。例如给代码生成 Worker 的 Prompt 会明确禁止它去回答历史知识问题。2.3 Spawn动态资源管理者与团队扩容器Spawn 是一个相对特殊但至关重要的角色。它不是一个执行任务的 Agent而是一个管理 Agent 生命周期的机制或服务。你可以把它想象成云平台的“自动伸缩组”或 Kubernetes 的“Deployment”。它的核心职责包括按需创建 Agent当 Lead 发现当前已有的 Worker 都无法满足某个新任务的需求时它可以请求 Spawn 机制根据任务描述动态创建一个新的、具备特定能力的 Worker。例如任务需要“将这段中文翻译成德语”而现有 Worker 中没有翻译专家Lead 就可以请求 Spawn 一个“德语翻译 Worker”。资源管理与回收Spawn 负责管理 Agent 实例所占用的资源如内存、会话。当某个 Worker 长时间闲置或任务完成后Spawn 可以将其销毁以释放资源。这对于运行成本敏感的应用如按 token 计费的大模型调用至关重要。提供 Agent 模板Spawn 机制通常维护着一个“Agent 模板库”。每个模板定义了创建一类 Agent 所需的全部配置基础模型、系统 Prompt、工具列表、温度参数等。当需要创建时只需指定模板名和少量参数即可。技术实现要点Spawn 可以是一个独立的服务也可以是 Harness 框架提供的一个核心函数如harness.spawn_worker(worker_typetranslator, target_languagede)。其内部需要实现 Agent 的实例化、上下文初始化、并可能将其注册到通信总线上。这三者的关系构成了一个经典的“管理者-工作者-工厂”模式。Lead 是管理者Worker 是工作者Spawn 是工厂。Harness 则是整个工厂的运营系统确保这个模式能顺畅、高效、稳定地运行起来。3. Harness 工程之道构建多 Agent 协作的基础设施理解了角色我们就可以深入探讨如何用 Harness 将它们串联起来。Harness 的设计质量直接决定了多 Agent 系统的上限。一个好的 Harness 应该像优秀的中间件一样让开发者感觉不到它的存在却能享受到它带来的所有便利。3.1 通信总线Agent 之间如何“对话”多个 Agent 要协作首先得能互相通信。你不能让每个 Agent 都去直接调用另一个 Agent 的函数那会带来紧耦合和混乱。Harness 需要提供一个统一的通信层通常我们称之为“消息总线”或“事件总线”。通信模式主要有两种直接调用式类似于 RPC远程过程调用。Lead 直接调用某个 Worker 的“执行”方法并同步等待结果。这种方式简单直接适用于串行、强依赖的任务流。# 伪代码示例 weather_result harness.invoke_worker(worker_idweather_001, task查询上海明天温度)其缺点是调用方Lead会被阻塞直到 Worker 执行完毕无法充分利用并行优势。发布订阅式基于消息队列。Lead 将任务以消息的形式发布到总线上并标明任务类型。所有订阅了该类任务的 Worker 都可以消费消息。或者更常见的由 Harness 的路由机制将消息精准投递给某个特定的 Worker。Worker 处理完后将结果作为另一条消息发布回总线Lead 再订阅结果消息。# 伪代码示例 # Lead 发布任务 task_msg {task_id: 123, type: data_analysis, payload: {data: csv_data}} harness.publish(task_queue, task_msg) # 数据分析 Worker 订阅并处理 # Harness 内部将 task_msg 路由给 data_analysis_worker result_msg data_analysis_worker.process(task_msg) harness.publish(result_queue, result_msg) # Lead 订阅并接收结果 final_result harness.consume(result_queue, task_id123)这种方式解耦彻底支持异步和并行是复杂系统的首选。Harness 需要实现消息的序列化/反序列化、持久化防止丢失、以及可靠的投递机制。消息格式标准化Harness 必须定义一套统一的消息格式。一个典型的任务消息可能包含以下字段{ message_id: uuid, conversation_id: session_uuid, from_agent: lead_agent_id, to_agent: target_worker_id, // 或为空由路由决定 task_type: weather_query, payload: { city: 北京, date: 2024-05-20 }, expect_format: json, priority: normal, created_at: timestamp }结果消息格式也类似会包含task_id用于关联以及statussuccess/failure、result、error_info等字段。3.2 状态管理团队的共享记忆与上下文在多步骤、长会话的协作中状态管理至关重要。状态可以分为两类会话状态即当前用户对话的上下文。例如用户最初说“我想去旅游”后来又说“预算一万元”。这两个信息需要被所有参与规划的 AgentLead 和相关的 Worker感知到。任务流程状态即一个复杂任务被拆解后各个子任务的执行进度、中间结果和依赖关系。例如“订机票”任务成功了“订酒店”任务失败了这个状态需要被 Lead 知晓以决定后续动作。Harness 需要提供一个集中的状态存储服务。常见的做法是使用一个共享的键值存储如 Redis 或内存中的字典并以conversation_id或task_flow_id作为命名空间。# 伪代码Harness 提供的状态接口 class HarnessStateManager: def set_state(self, key, value, namespace): # 存储状态如 harness.set_state(user_budget, 10000, session_idabc) pass def get_state(self, key, namespace): # 获取状态所有Agent都可以调用 pass def append_to_history(self, message, namespace): # 追加对话历史维护完整的上下文 passLead 负责维护和更新主要的状态Worker 在需要时可以读取相关状态如“用户偏好”并在执行后更新任务状态。Harness 要确保状态访问的并发安全。3.3 路由与负载均衡把任务交给最合适的“人”当 Lead 发布一个“数据分析”任务时系统里可能有三个数据分析 Worker。Harness 的路由机制需要决定将任务交给哪一个。这就是路由策略。常见的路由策略基于能力的路由最简单的每个 Worker 在启动时向 Harness 注册自己的能力标签[data_analysis, python, pandas]。Harness 维护一个能力- Worker 的映射表。收到任务后根据task_type查找匹配的 Worker。轮询/随机负载均衡在多个相同能力的 Worker 间平均分配任务避免单个 Worker 过载。基于资源的路由选择当前内存/CPU 占用最低的 Worker。基于亲和性的路由如果任务 B 严重依赖任务 A 的中间结果且任务 A 是由 Worker X 执行的那么任务 B 也优先路由给 Worker X可以利用其缓存或上下文。在动态 Spawn 的场景下路由变得更加智能。如果 Harness 发现当前没有能处理“德语翻译”的 Worker它可以先触发 Spawn 机制创建一个然后再将任务路由过去。3.4 容错与监控让系统稳定可靠任何分布式系统都会出错多 Agent 系统也不例外。Harness 必须内置容错机制。超时与重试为每个任务设置超时时间。如果 Worker 在规定时间内没有返回结果Harness 可以标记任务失败并通知 Lead。Lead 可以根据策略决定是否重试可能换一个 Worker。死信队列对于反复失败的任务不要让它无限循环。Harness 可以将其移入死信队列并触发告警让开发者介入排查。健康检查定期向 Worker 发送心跳包检查其是否存活。对于失活的 Worker将其从注册表中移除避免后续任务被路由到“僵尸”节点。分布式追踪为每个用户请求生成一个唯一的trace_id并贯穿整个调用链。在日志中记录每个 Agent 的输入、输出、耗时。当出现问题时可以通过trace_id快速还原整个执行链路定位瓶颈或错误源头。这类似于微服务中的 OpenTelemetry。# 伪代码在Harness中集成追踪 def invoke_with_trace(self, worker_id, task, trace_id): span_id generate_span_id() logger.info(f[Trace-{trace_id}][Span-{span_id}] Start invoking worker {worker_id}) start_time time.time() try: result self._do_invoke(worker_id, task) logger.info(f[Trace-{trace_id}][Span-{span_id}] Worker {worker_id} succeeded in {time.time()-start_time:.2f}s) return result except Exception as e: logger.error(f[Trace-{trace_id}][Span-{span_id}] Worker {worker_id} failed: {e}) raise将这些模块组合起来一个 Harness 的雏形就出现了。它提供了通信、状态、路由、生命周期管理和可观测性让开发者可以专注于设计单个 Agent 的能力而无需操心它们如何组织在一起。4. 实战架构设计一个轻量级多 Agent Harness 的实现蓝图理论讲完了我们来点实际的。如何从零开始设计一个轻量级的、可用于原型验证的多 Agent Harness我们不追求大而全而是追求概念清晰、易于理解和扩展。4.1 核心类设计我们设计几个核心的 Python 类来构建这个框架。1.Message类定义通信的基本单元。from dataclasses import dataclass from typing import Any, Dict, Optional import uuid import time dataclass class Message: 统一的消息格式 msg_id: str None conversation_id: str None from_agent: str None to_agent: Optional[str] None # 为空则代表广播或由路由决定 task_type: str None payload: Dict[str, Any] None expect_format: str json priority: int 0 created_at: float None def __post_init__(self): if self.msg_id is None: self.msg_id str(uuid.uuid4()) if self.created_at is None: self.created_at time.time()2.Agent基类所有 AgentLead, Worker的父类。from abc import ABC, abstractmethod class Agent(ABC): Agent抽象基类 def __init__(self, agent_id: str, harness): self.agent_id agent_id self.harness harness # 持有Harness引用用于通信 self.skills [] # 能力标签列表 harness.register_agent(self) # 自动注册 abstractmethod async def process_message(self, message: Message) - Dict[str, Any]: 处理消息的核心方法由子类实现 pass def send_message(self, to_agent_id: str, task_type: str, payload: Dict): 发送消息的辅助方法 msg Message( from_agentself.agent_id, to_agentto_agent_id, task_typetask_type, payloadpayload ) return self.harness.dispatch_message(msg)3.Worker类继承自Agent代表具体执行者。class Worker(Agent): 工作者Agent def __init__(self, agent_id: str, harness, skills: list, llm_clientNone, toolsNone): super().__init__(agent_id, harness) self.skills skills self.llm llm_client # 可选的LLM客户端用于需要推理的Worker self.tools tools or {} # 该Worker专有的工具函数字典 async def process_message(self, message: Message): # 根据 task_type 调用相应的工具或逻辑 task_handler self.tools.get(message.task_type) if not task_handler: return {status: error, reason: fUnsupported task type: {message.task_type}} try: result await task_handler(**message.payload) return {status: success, result: result} except Exception as e: return {status: error, reason: str(e)}4.Lead类继承自Agent代表指挥者。class Lead(Agent): 领导者Agent def __init__(self, agent_id: str, harness, planner_llm): super().__init__(agent_id, harness) self.planner_llm planner_llm # 用于任务规划的LLM self.skills [planning, orchestration] async def process_user_query(self, user_input: str, conversation_id: str): 处理用户原始输入的主入口 # 1. 使用 planner_llm 进行任务规划 plan await self._create_plan(user_input) # plan 可能是一个任务列表如 # [{type: web_search, params: {query: 北京天气}}, # {type: data_analysis, params: {data: ...}}] # 2. 执行计划 results [] for task in plan: # 2.1 寻找或创建Worker worker_id await self.harness.find_or_spawn_worker_for_task(task[type]) # 2.2 分发任务并等待结果 task_msg Message( conversation_idconversation_id, from_agentself.agent_id, to_agentworker_id, task_typetask[type], payloadtask[params] ) task_result await self.harness.dispatch_and_wait(task_msg, timeout30) results.append(task_result) # 3. 汇总结果生成最终回复 final_output await self._synthesize_results(results) return final_output async def _create_plan(self, user_input): # 调用LLM进行规划这里简化处理 # 实际应用中这里会有复杂的Prompt工程和解析逻辑 pass5.Harness核心类系统的中枢。import asyncio from typing import Dict, List class Harness: 多Agent系统的协调器 def __init__(self): self.agents: Dict[str, Agent] {} # agent_id - Agent 实例 self.agent_skills: Dict[str, List[str]] {} # skill - [agent_id] self.message_queue asyncio.Queue() # 简易内存消息队列 self.state_store {} # 简易内存状态存储key: namespace, value: dict self._worker_templates {} # Worker模板库 def register_agent(self, agent: Agent): 注册一个Agent self.agents[agent.agent_id] agent for skill in agent.skills: self.agent_skills.setdefault(skill, []).append(agent.agent_id) print(f[Harness] Agent {agent.agent_id} registered with skills: {agent.skills}) async def dispatch_message(self, message: Message): 分发消息到目标Agent if message.to_agent: # 直接指定了接收者 target_agent self.agents.get(message.to_agent) if not target_agent: return {status: error, reason: fAgent {message.to_agent} not found} return await target_agent.process_message(message) else: # 需要路由根据 task_type 找到有对应技能的Agent candidates self.agent_skills.get(message.task_type, []) if not candidates: # 没有现成Worker尝试动态创建 new_worker_id await self.spawn_worker(message.task_type) if new_worker_id: candidates [new_worker_id] else: return {status: error, reason: fNo worker for task: {message.task_type}} # 简单负载均衡取第一个可扩展为更复杂的策略 target_agent_id candidates[0] message.to_agent target_agent_id return await self.agents[target_agent_id].process_message(message) async def spawn_worker(self, worker_type: str) - Optional[str]: 动态创建一个指定类型的Worker template self._worker_templates.get(worker_type) if not template: print(f[Harness] No template found for worker type: {worker_type}) return None # 根据模板创建Worker实例 worker_id f{worker_type}_{uuid.uuid4().hex[:8]} new_worker Worker( agent_idworker_id, harnessself, skillstemplate[skills], llm_clienttemplate.get(llm), toolstemplate.get(tools) ) # 新Worker会自动注册到 self.agents 和 self.agent_skills print(f[Harness] Spawned new worker: {worker_id}) return worker_id def set_state(self, key, value, namespacedefault): 设置共享状态 if namespace not in self.state_store: self.state_store[namespace] {} self.state_store[namespace][key] value def get_state(self, key, namespacedefault, defaultNone): 获取共享状态 return self.state_store.get(namespace, {}).get(key, default) async def dispatch_and_wait(self, message: Message, timeout: float): 分发消息并同步等待结果简化版 # 这里是一个简化的同步等待实际应用中可能更复杂涉及回调或Future task asyncio.create_task(self.dispatch_message(message)) try: result await asyncio.wait_for(task, timeouttimeout) return result except asyncio.TimeoutError: return {status: error, reason: Task timeout}4.2 一个完整的协作流程示例让我们用上面的框架模拟一个“旅行规划”的简单场景。import asyncio # 1. 定义工具函数实际Worker的能力 async def search_weather(city: str, date: str): # 模拟网络调用 await asyncio.sleep(0.5) return f{city}在{date}的天气是晴朗25摄氏度。 async def search_flight(from_city: str, to_city: str, date: str): await asyncio.sleep(1) return f找到航班{from_city} - {to_city}{date}价格1200元。 async def book_hotel(city: str, check_in: str, nights: int): await asyncio.sleep(0.8) return f已预订{city}的酒店入住{check_in}{nights}晚总价800元。 # 2. 创建Harness实例 harness Harness() # 3. 预注册一些Worker weather_worker Worker( agent_idweather_master, harnessharness, skills[weather_search], tools{weather_search: search_weather} ) travel_worker Worker( agent_idtravel_expert, harnessharness, skills[flight_search, hotel_booking], tools{flight_search: search_flight, hotel_booking: book_hotel} ) # 4. 注册Worker模板供Spawn使用 harness._worker_templates[weather_search] { skills: [weather_search], tools: {weather_search: search_weather} } harness._worker_templates[flight_search] { skills: [flight_search], tools: {flight_search: search_flight} } # 5. 创建Lead Agent (这里简化了LLM部分用固定规划逻辑) class SimpleLead(Lead): async def _create_plan(self, user_input): # 一个非常简单的、硬编码的规划器 if 北京 in user_input and 上海 in user_input: return [ {type: weather_search, params: {city: 北京, date: 2024-05-20}}, {type: weather_search, params: {city: 上海, date: 2024-05-21}}, {type: flight_search, params: {from_city: 北京, to_city: 上海, date: 2024-05-20}}, ] return [] async def _synthesize_results(self, results): success_results [r for r in results if r.get(status) success] return f规划完成。共执行{len(success_results)}项子任务。详情{success_results} lead SimpleLead(lead_agent, harness, planner_llmNone) # 6. 模拟用户请求 async def main(): user_query 我想下周从北京去上海帮我看看天气和航班。 conversation_id conv_001 final_answer await lead.process_user_query(user_query, conversation_id) print(最终回复:, final_answer) # 运行 asyncio.run(main())这个示例虽然简单但完整地演示了从 Harness 初始化、Agent 注册、任务规划、消息路由包括潜在的 Spawn到结果汇总的整个闭环。你可以看到Harness 如何像一个无形的指挥者在背后协调着一切。5. 进阶议题与避坑指南当你开始着手实现或使用一个多 Agent Harness 时会遇到许多在简单 demo 中不会出现的问题。下面分享一些进阶思考和常见陷阱。5.1 如何设计有效的 Agent 间通信协议消息格式只是通信的一部分协议还包括通信模式。除了前面提到的直接调用和发布订阅还有两种重要模式请求-响应这是最常用的同步模式调用方等待被调用方返回。Harness 需要管理超时和重试。流式响应对于生成文本、流式传输数据等长时间任务Worker 可以分多次返回结果Chunk。Harness 需要支持这种流式消息的转发让 Lead 能实时看到进度。这通常通过 WebSocket 或 Server-Sent Events (SSE) 实现。避坑提示不要在消息中传递过大的数据如图片、大文件。应该传递一个引用如文件 ID 或 URL由接收方按需从共享存储如 S3、数据库中拉取。否则消息队列会不堪重负。5.2 动态 Spawn 的成本与资源管理动态创建 Agent尤其是基于大模型的 Agent听起来很美好但成本很高。每次 Spawn 可能意味着一次新的 LLM 会话初始化产生额外的延迟和费用。最佳实践Worker 池化对于常用类型的 Worker采用池化技术。Harness 维护一个空闲 Worker 池需要时从池中取出用完放回而不是销毁。这类似于数据库连接池。懒加载与预热系统启动时只创建核心的、高概率用到的 Worker。其他 Worker 在第一次被请求时再创建懒加载并可以留在池中一段时间。设置上限为每种类型的 Worker 设置最大实例数防止资源耗尽。使用轻量级 Worker对于简单任务Worker 可以不依赖 LLM而是纯函数或规则引擎这样 Spawn 成本极低。5.3 调试与可观测性是生命线多 Agent 系统调试起来比单体应用复杂得多。一个问题可能出现在 Lead 的规划、某个 Worker 的执行、或是消息传递的途中。必须建立的观测手段结构化日志每个 Agent、每条消息都带上唯一的trace_id和span_id。使用像structlog这样的库将日志输出为 JSON 格式方便接入 ELKElasticsearch, Logstash, Kibana或 Loki 进行聚合查询。关键指标监控监控消息队列长度、各类型任务的平均处理时间、Worker 的错误率、Spawn 频率等。这些指标能帮你提前发现系统瓶颈。可视化追踪集成像 Jaeger 这样的分布式追踪系统。你可以清晰地看到一个用户请求是如何像流水线一样流经各个 Agent 的每个环节耗时多少一目了然。这对于性能优化和故障排查至关重要。5.4 避免“僵尸对话”与状态泄露在多轮对话中Harness 需要管理对话的上下文。一个常见的坑是对话结束了但相关的 Agent 实例和状态没有被及时清理导致内存泄漏。解决方案会话超时为每个conversation_id设置一个超时时间如30分钟无活动。超时后Harness 触发清理流程通知相关 Agent 释放资源清除该会话在状态存储中的所有数据。显式结束信号设计一个特殊的消息类型如session_terminate当客户端如前端断开连接或用户明确结束对话时发送。Harness 广播此消息所有参与该会话的 Agent 进行清理。引用计数对于共享资源如某个被多个会话引用的数据文件使用引用计数当计数归零时再真正删除。5.5 测试策略从单元到集成测试多 Agent 系统需要分层进行单元测试测试单个 Worker 的工具函数是否正确。这是最容易的。组件测试测试 Lead 的规划逻辑。可以 Mock 掉 Harness 和 Worker只验证给定用户输入Lead 生成的计划是否符合预期。集成测试启动一个包含 Harness、Lead 和几个关键 Worker 的完整环境模拟端到端的用户场景。使用真实的 LLM 调用但可以用低配模型或设置低温度以减少成本和不稳定性。混沌测试模拟网络延迟、Worker 进程突然崩溃、消息丢失等情况观察系统的容错和恢复能力。例如在任务执行中途 Kill 掉一个 Worker看 Lead 是否会超时并重试。记住多 Agent 系统本质上是一个分布式系统。分布式系统领域的所有经典问题——网络分区、时钟同步、一致性——在这里都可能以某种形式出现。从简单开始逐步增加复杂性并始终将可观测性放在首位是驾驭它的不二法门。