做 AI Agent 开发有一段时间的同学大概率会遇到这样一个尴尬场景单个 Agent 套上一个精心设计的 Prompt确实能完成一些单点任务比如改写文案、抽取数据、写一段代码。可一旦任务变成“先做市场调研再写技术方案然后生成汇报 PPT 大纲最后安排成员跟进”单 Agent 就开始频繁失控——上下文越拖越长中间步骤忘了前面结论工具调用来回调不对最后的输出质量完全不可控。这类问题并不是 Prompt 写得不够好而是架构选型出了问题。真实业务里的大型任务天然适合拆成多个角色、多个步骤、多种工具协同完成这背后就是多智能体协作与异步任务编排的用武之地。本文会围绕 DeepAgents 多智能体协作这条主线从零实现一个可运行的多智能体系统内容包括子智能体设计、工具调用、主管-工人协作模式、异步任务编排以及 Harness 在其中的作用。无论你是刚接触 AI 大模型应用开发还是已经在做 Agent 落地这套代码和思路都可以直接参考。1. 背景与核心概念1.1 单智能体的能力边界先给“智能体”下一个通俗定义Agent 大模型 记忆 工具 循环。大模型提供推理能力记忆保存上下文工具让模型能操作真实世界循环让模型可以不断“观察-思考-行动”直到任务完成。单 Agent 适合做短链路任务比如“总结这段文本”“把这份 JSON 转成表格”。一旦任务链路变长它的问题就暴露得很明显上下文窗口有限前面步骤的内容会被截断或稀释。无法并行所有步骤都串行效率低。一个 Agent 很难同时擅长规划、检索、编码、写作四种能力。单个环节失败可能导致整个任务失败没有降级机制。这些问题催生了一个新思路与其训练一个万能 Agent不如把任务拆开交给多个各司其职的子 Agent 协作完成。于是多智能体系统成为 2026 年 AI 应用开发中最受关注的方向之一。1.2 多智能体协作解决什么问题多智能体协作的核心思想是把一个复杂任务拆解成多个子任务分别交给不同的智能体执行再由编排层统一汇总。可以把它理解成一个项目组产品经理负责拆需求开发工程师负责写代码测试工程师负责验证项目经理负责协调进度。每个角色只做自己最擅长的事通过统一的消息机制沟通最终由负责人整合结果。在工程上这种模式带来的好处非常直接每个 Agent 的上下文更短专注自身任务不容易被无关信息干扰。无依赖的子任务可以并行执行大大缩短整体耗时。单个 Agent 失败时可以重试或重新派发不影响整个链路。系统更易扩展新增一个能力只需要注册一个新的子 Agent。多智能体在 AI 大模型应用开发中的典型场景包括研究报告自动撰写、竞品分析、代码仓库巡检、客服工单流转、农业大模型中的环境监测与决策联动等。1.3 DeepAgents 与 Harness 的关系“DeepAgents”强调的不是“很多个 Agent 各说各话”而是带有深度规划、深度执行和深度反思能力的智能体系统。一个 DeepAgent 会先拆解目标再逐步执行过程中不断检查结果是否需要修正。多个 DeepAgent 协作时需要一套框架负责调度、状态管理、消息路由和错误恢复。这个框架在社区中被称作 Harness。你可以把 Agent 理解成员工Harness 理解成项目管理系统员工负责干活系统负责记录进度、分配任务、处理异常。2026 年Harness 工程已经成为 AI 大模型开发里的基础话题。越来越多的模型团队和应用团队开始强调模型能力固然重要但控制模型运行的 Harness 设计往往才是决定一个 Agent 系统能不能稳定落地的关键。2. 环境准备与版本说明2.1 运行环境本文的示例代码以 Python 为主建议使用 Python 3.10 或更高版本因为后面的异步编排会用到 asyncio 和 dataclass高版本支持更完善。操作系统不限Windows、macOS、Linux 都可以。项目会调用大模型接口所以你需要准备好一个模型服务的 API Key。为了示例通用性代码以“兼容 OpenAI 接口的模型服务”为例Key 和 Base URL 都通过环境变量配置。你在实践中可以换成任何兼容 OpenAI 协议的模型服务包括国内外各类大模型平台。2.2 创建项目并安装依赖先创建一个项目目录并初始化虚拟环境mkdir deepagents-demo cd deepagents-demo python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate本项目的依赖非常少核心只有一个 OpenAI SDK。为了环境变量管理方便再加上 python-dotenv。pip install openai python-dotenv把依赖写入 requirements.txtopenai1.35.0 python-dotenv1.0.02.3 配置模型服务在项目根目录创建.env文件OPENAI_API_KEY你的_API_KEY OPENAI_BASE_URLhttps://你的模型服务地址 OPENAI_MODELgpt-4o-mini这里的OPENAI_MODEL建议按实际模型名填写不同模型能力差异很大小模型适合做简单抽取强模型适合做规划和总结。在代码中统一加载配置# config.py import os from dotenv import load_dotenv load_dotenv() API_KEY os.getenv(OPENAI_API_KEY) BASE_URL os.getenv(OPENAI_BASE_URL) MODEL os.getenv(OPENAI_MODEL, gpt-4o-mini)这里有一个工程建议API Key 千万不要硬编码在代码里也不要提交到 Git 仓库。用环境变量管理密钥配合.env文件并在.gitignore中忽略它是最基本也最有效的安全习惯。2.4 项目结构规划本文最终会实现三个层级的完整示例单个子智能体、多智能体协作、异步任务编排。项目结构建议如下deepagents-demo/ ├── .env ├── .gitignore ├── requirements.txt ├── config.py ├── agent.py ├── multi_agent.py ├── orchestrator.py └── demo_async.py后面每一个文件我都会给出完整代码和说明。3. 核心概念Agent、Message、Tool 与 Harness在进入代码之前先把多智能体系统里最核心的四个概念讲透。理解了它们后面看代码会顺畅很多。3.1 Agent有目标、有工具、有循环的执行单元一个 Agent 的核心不是“调用一次大模型”而是“循环”。标准的 Agent 循环包含四步模型根据当前消息决定调用哪个工具或直接输出最终答案。系统执行工具调用。把工具结果作为新消息返回给模型。模型再次判断任务是否完成。这个循环会一直持续直到模型不再请求调用工具或者达到设定的最大步数。没有这个循环大模型只能“说”不能“做”有了这个循环大模型才能通过工具影响真实系统。3.2 Message智能体之间协作的统一语言无论是单个 Agent 内部的多次调用还是多个 Agent 之间的信息传递底层都统一为消息。一条消息最基本的字段是role消息角色常见的有 system、user、assistant、tool。content消息正文。tool_calls模型发起的工具调用请求包含函数名和参数。多智能体协作的本质就是不同 Agent 之间互相构造、交换、消费 Message。因此在设计系统时消息结构越统一后续扩展越容易。3.3 Tool给智能体装上“手”工具的本质是一个函数外加一份 JSON Schema 描述。描述告诉模型“这个函数叫什么、有哪些参数、什么时候应该用”。模型本身不会执行代码它只生成工具调用的请求真正执行的是我们自己的代码。工具设计的好与坏直接决定 Agent 能不能稳定完成任务。一个常见的坑是工具描述写得太模糊模型不知道该在什么时候调用或者参数设计不合理模型总能生成各种奇怪的参数。3.4 Harness多智能体系统的执行框架Harness 在前文提过这里再展开讲。单个 Agent 需要循环控制多个 Agent 需要任务调度整个系统需要状态管理、错误处理、日志记录。这一层基础设施就是 Harness。可以把 Harness 理解成“智能体操作系统”。它不负责具体的业务推理而是负责决定哪个 Agent 什么时候执行、消息该路由给谁、工具调用失败后怎么重试、任务依赖关系怎么解析、结果怎么汇聚。理解了 Harness 的概念也就理解了异步任务编排在系统中的位置编排器是 Harness 中负责“任务调度”的模块而异步能力让无依赖的任务真正并行起来。4. 实战一实现一个可运行的子智能体先从最基础的子智能体开始。这一节的目标是编写一个具备工具调用循环能力的 Agent 类它能根据用户问题自动决定是否调用工具并最终返回答案。4.1 定义一个模拟工具为了演示工具调用机制这里定义一个“天气查询”工具。为了不依赖外部 API它返回模拟数据但工具定义方式与真实场景完全一致。# tools.py import json def get_weather(city: str, date: str 今天) - str: 查询指定城市在指定日期的天气情况 weather_map { 北京: 晴23℃, 上海: 小雨26℃, 广州: 多云30℃, } weather weather_map.get(city, 晴25℃) return json.dumps({city: city, date: date, weather: weather}, ensure_asciiFalse)这个函数接收城市和日期返回 JSON 字符串。实际项目中工具函数可能是查数据库、调内部接口、操作文件系统但封装思路完全一样输入参数、返回结果把外部能力包装成模型可调用的函数。4.2 把函数转换成模型可识别的工具 SchemaOpenAI 兼容接口要求工具用 JSON Schema 描述。我们可以写一个转换函数也可以直接手写 Schema# tools.py WEATHER_TOOL { type: function, function: { name: get_weather, description: 查询指定城市在指定日期的天气情况。当用户询问天气时使用。, parameters: { type: object, properties: { city: {type: string, description: 城市名称}, date: {type: string, description: 日期默认今天} }, required: [city] } } }注意这里的 description 写得非常关键。模型靠这段描述来决定什么时候调用工具描述越具体行为越稳定。4.3 编写子智能体类现在实现 Agent 类。它的核心是一个循环把消息发给模型如果模型返回工具调用就执行工具、把结果追加到消息列表然后继续循环如果模型返回普通回复就结束循环。# agent.py import json from openai import OpenAI from config import API_KEY, BASE_URL, MODEL class Agent: def __init__(self, name, system_prompt, toolsNone, tool_mapNone, modelNone, max_steps10): self.name name self.system_prompt system_prompt self.tools tools or [] self.tool_map tool_map or {} self.model model or MODEL self.max_steps max_steps self.messages [{role: system, content: system_prompt}] self.client OpenAI(api_keyAPI_KEY, base_urlBASE_URL) def run(self, user_input: str) - str: self.messages.append({role: user, content: user_input}) for step in range(self.max_steps): response self.client.chat.completions.create( modelself.model, messagesself.messages, toolsself.tools if self.tools else None, ) message response.choices[0].message self.messages.append(message) # 没有工具调用说明模型已经给出最终答案 if not message.tool_calls: return message.content # 遍历模型发起的工具调用请求 for tool_call in message.tool_calls: fn_name tool_call.function.name fn_args json.loads(tool_call.function.arguments) result self._call_tool(fn_name, fn_args) self.messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) raise RuntimeError(fAgent {self.name} 达到最大步数 {self.max_steps}任务未完成) def _call_tool(self, fn_name, fn_args): if fn_name not in self.tool_map: return json.dumps({error: f未知工具: {fn_name}}, ensure_asciiFalse) try: result self.tool_map[fn_name](**fn_args) return result if isinstance(result, str) else json.dumps(result, ensure_asciiFalse) except Exception as e: return json.dumps({error: str(e)}, ensure_asciiFalse)这里有几个细节值得注意tools可能为空此时不传 tools 参数避免模型被空工具列表迷惑。message.tool_calls是判断循环是否继续的关键信号。工具执行结果以 role 为 tool 的消息回传OpenAI 接口要求必须带上tool_call_id。工具异常被捕获后转成 JSON 字符串返回给模型这样模型能自己根据错误调整参数而不是整个流程崩溃。4.4 运行与验证写一个入口脚本测试# demo_single.py from agent import Agent from tools import get_weather, WEATHER_TOOL system_prompt 你是一个天气助手。当用户询问天气时调用 get_weather 工具查询然后根据结果回答。 agent Agent( nameweather_bot, system_promptsystem_prompt, tools[WEATHER_TOOL], tool_map{get_weather: get_weather}, ) result agent.run(北京今天天气怎么样) print(result)运行python demo_single.py预期输出类似北京今天天气晴朗气温 23℃体感舒适适合外出活动。整个过程模型经历了“判断需要调用工具 - 生成工具调用参数 - 收到工具结果 - 生成最终回答 ”四个阶段。虽然功能简单但它已经是完整的 Agent 闭环了。5. 实战二多智能体协作单独的 Agent 能跑通后下一步进入多智能体协作实战。5.1 架构设计主管-工人模式多智能体协作的模式有很多比如对话模式、管道模式、层级模式。本文选择最常见、也最容易落地的“主管-工人”模式Supervisor Agent负责任务拆解、结果质量检查和最终汇总。Worker Agent每个 Worker 负责一个具体子任务。整个流程分三步用户提交总任务。Supervisor 将总任务拆成 N 个子任务并指定每个子任务的执行者。Worker 依次或并行执行自己的子任务结果回传给 Supervisor由 Supervisor 合成最终输出。这种模式的好处是职责清晰。规划、执行、汇总分离后续想替换某个 Worker或者新增一个 Worker都不需要改动其他模块。5.2 代码实现新建multi_agent.py。这里复用上一节的Agent类通过不同 system prompt 来区分 Supervisor 和 Worker 的行为。# multi_agent.py import json from agent import Agent class MultiAgentSystem: def __init__(self, supervisor_prompt, worker_prompts, modelNone): self.supervisor Agent( namesupervisor, system_promptsupervisor_prompt, modelmodel, ) self.workers {} for name, prompt in worker_prompts.items(): self.workers[name] Agent( namename, system_promptprompt, modelmodel, ) def process(self, task: str) - str: # 第一步Supervisor 拆分任务输出 JSON 数组 plan_text self.supervisor.run( f请拆分以下任务为子任务每个子任务指定执行者。\n任务{task}\n 输出 JSON 数组格式为 [{\worker\: \worker名\, \task\: \子任务描述\}] ) subtasks self._parse_plan(plan_text) # 第二步逐个子任务分发给 Worker 执行 worker_results [] for item in subtasks: worker_name item.get(worker) sub_task item.get(task) if worker_name not in self.workers: worker_results.append({worker: worker_name, error: 未知执行者}) continue result self.workers[worker_name].run(sub_task) worker_results.append({worker: worker_name, result: result}) # 第三步Supervisor 汇总结果 summary self.supervisor.run( f以下是子任务执行结果请综合成一份完整报告\n f{json.dumps(worker_results, ensure_asciiFalse, indent2)} ) return summary staticmethod def _parse_plan(text: str): try: return json.loads(text) except json.JSONDecodeError: # 如果模型返回了多余的文字尝试截取 JSON 部分 start text.find([) end text.rfind(]) 1 if start -1 or end 0: raise ValueError(f无法解析规划结果: {text}) return json.loads(text[start:end]) if start 0 and end start else []回调解析是这段代码里最值得注意的部分。大模型并不保证输出纯 JSON经常出现“好的以下是拆分结果[...]”这种带有前后缀的情况所以这里提供了容错解析。5.3 运行与验证模拟一个实际场景写一篇关于“AI 技术在农业中的应用”的科普文章。三个 Worker 分别负责引言、核心应用、未来展望。# demo_multi.py from multi_agent import MultiAgentSystem supervisor_prompt ( 你是一个高级项目经理。你会把用户任务拆解为多个子任务 分派给指定的 Worker 执行。Worker 列表writer_intro、writer_core、writer_future。 最后你负责汇总所有 Worker 的结果。 ) worker_prompts { writer_intro: 你是科普文章引言撰写专家负责写文章引言300字以内。, writer_core: 你是农业科技领域专家负责介绍AI在土壤监测、气象预警、智能灌溉施肥中的应用500字以内。, writer_future: 你是行业趋势分析师负责写AI农业的未来展望300字以内。, } system MultiAgentSystem( supervisor_promptsupervisor_prompt, worker_promptsworker_prompts, ) result system.process(写一篇关于AI技术在农业中的应用的科普文章) print(result)python demo_multi.py运行完成后输出会是一篇结构完整的科普文章。通过打印每个 Worker 的返回结果你可以清楚观察到多智能体协作的过程。6. 实战三异步任务编排到了这一步“多智能体协作”已经实现了但目前的 worker 执行是串行的。如果任务一耗时 10 秒、任务二耗时 10 秒、任务三耗时 10 秒整体就要 30 秒。如果这三个任务互不依赖完全可以并行执行总耗时只有 10 秒。这就是异步任务编排的价值。6.1 任务类型分析在真实的智能体系统中任务之间通常存在两类关系无依赖关系可以并行执行。有依赖关系B 必须等 A 完成才能执行。比如“先调研再写方案最后生成 PPT”这是一个链式依赖而“同时调研市场、调研技术、调研竞品”则是并行关系。一个合格的编排器应该同时支持这两种关系。6.2 用 asyncio 实现并行Python 原生的 asyncio 库已经提供了很好的并发支持。最简单的并行执行可以用asyncio.gather# demo_async.py import asyncio async def worker(task_name, duration): print(f[开始] {task_name}) await asyncio.sleep(duration) print(f[完成] {task_name}) return f{task_name}_result async def main(): results await asyncio.gather( worker(任务A, 2), worker(任务B, 3), worker(任务C, 1), ) print(results) if __name__ __main__: asyncio.run(main())运行结果如下[开始] 任务A [开始] 任务B [开始] 任务C [完成] 任务C [完成] 任务A [完成] 任务B [任务A_result, 任务B_result, 任务C_result]注意这里三个任务同时开始总耗时只有约 3 秒而不是 6 秒。6.3 带依赖关系的异步编排器gather只能处理“一批任务全部可以并行”的情况。遇到“任务 D 依赖 B 和 C 的结果”这种场景需要设计一个更通用的编排器。实现思路每个任务声明自己的依赖任务名。编排器维护一个“待执行”集合和“已完成”字典。循环扫描把“依赖已全部完成”的任务取出来用 asyncio 并发执行。执行完成后更新结果直到所有任务完成。# orchestrator.py import asyncio from dataclasses import dataclass, field from typing import Callable, Awaitable dataclass class Task: name: str deps: list field(default_factorylist) coro: Callable[..., Awaitable] None async def execute(self, context: dict): kwargs {dep: context[dep] for dep in self.deps} context[self.name] await self.coro(**kwargs) class AsyncOrchestrator: def __init__(self, tasks: list): self.tasks {task.name: task for task in tasks} self.status {} # name - executed or not self.context {} def validate(self): for task in self.tasks.values(): for dep in task.deps: if dep not in self.tasks: raise ValueError(f任务 {task.name} 依赖了不存在的任务 {dep}) async def _process_ready_tasks(self): ready [] for name, task in self.tasks.items(): if name in self.status: continue if all(dep in self.status for dep in task.deps): ready.append(task) return ready async def run(self): self.validate() while len(self.status) len(self.tasks): ready await self._process_ready_tasks() if not ready: raise RuntimeError(存在循环依赖或无法调度的任务) await asyncio.gather(*[task.execute(self.context) for task in ready]) for task in ready: self.status[task.name] True return self.context这个编排器的核心是_process_ready_tasks每次只找“所有依赖都已完成”的任务并发执行。执行完一批更新状态继续下一批。6.4 测试编排器下面模拟一个多智能体异步编排场景Task A调研需求无依赖耗时 3 秒Task B技术预研依赖 A耗时 2 秒Task C竞品分析无依赖耗时 2 秒Task D生成方案依赖 B 和 C耗时 2 秒# demo_orchestrator.py import asyncio from orchestrator import AsyncOrchestrator, Task async def research_requirement(): await asyncio.sleep(3) return 需求调研报告 async def tech_preview(需求调研报告): await asyncio.sleep(2) return f技术预研基于{需求调研报告} async def competitor_analysis(): await asyncio.sleep(2) return 竞品分析报告 async def final_plan(技术预研, 竞品分析报告): await asyncio.sleep(2) return f最终方案{技术预研} {竞品分析报告} async def main(): orchestrator AsyncOrchestrator([ Task(name调研, deps[], cororesearch_requirement), Task(name预研, deps[调研], corotech_preview), Task(name竞品, deps[], corocompetitor_analysis), Task(name方案, deps[预研, 竞品], corofinal_plan), ]) results await orchestrator.run() print(results) if __name__ __main__: asyncio.run(main())执行过程会充分体现异步编排的优势调研和竞品分析并行启动调研完成后预研开始竞品完成后方案等待预研和竞品都完成最后才执行。总耗时从串行的 9 秒压缩到 7 秒任务越多这种优势越明显。6.5 把异步编排接入多智能体系统回到多智能体场景第 5 节的 Worker 执行是串行的现在可以把每个 Worker 封装成 asyncio 任务交给编排器调度。思路是每个 Worker 的run方法改成异步版本。根据任务依赖关系构造 Task 列表。编排器管理整个执行过程。Supervisor 在最后一个“汇总任务”中接收所有 Worker 结果。这里不重复贴全部代码核心改动是使用asyncio.to_thread把已有的同步 Agent.run 包装成可异步执行的任务asyncio.to_thread(worker_agent.run, sub_task)在真实生产环境中Agent 的run方法往往本身就是异步的因为它内部要调模型接口、调外部工具而这些 I/O 操作天然适合异步。7. 常见问题与排查思路多智能体系统开发中遇到的坑通常集中在循环控制、工具调用、消息管理、异步调度等几个方面。下面整理一份排查清单。问题现象常见原因解决思路Agent 陷入死循环缺少终止条件模型反复调用工具设置 max_steps在 system prompt 中明确“不需要调用工具时就输出最终结果”Agent 不调用工具直接编答案工具 description 不清晰或模型误判自己知道答案强化工具描述触发条件在 prompt 里加 few-shot 示例工具参数解析失败模型返回的 JSON 参数与 schema 不一致捕获 json.JSONDecodeError把错误信息以 tool 消息回传给模型让它重新生成多个 Agent 消息串了多个 Agent 共用了同一个 messages 列表确保每个 Agent 实例持有独立的 messages不要用全局消息列表API 请求超时模型响应过慢或网络不稳定设置 timeout 参数实现指数退避重试长任务改异步异步编排卡死任务存在循环依赖或初始化失败在 run() 开头调用 validate()准备超时机制成本失控工具循环次数过多模型越试越偏限制 max_steps优先使用小模型对大模型调用设置每日预算和日志监控7.1 更深入地看如何调试 Agent 循环调试 Agent 循环最直接的办法是打印每一个步骤的消息内容。在Agent.run中加一行日志就能看到每次模型返回的消息、工具调用、工具结果print(f[{self.name}] step {step}: {message})生产环境建议换成结构化日志记录步骤号、模型名、tokens、工具名称、耗时。7.2 异步编排器常见陷阱异步编排最隐蔽的问题出现在共享状态上。多个并发任务同时读写同一个 dict可能出现数据覆盖。我上面的示例里编排器用一个 context 字典保存所有任务结果但每个任务写入的是自己的 key因此并发安全。如果业务需要多个 Worker 操作同一个全局状态建议引入锁或者使用不可变数据。另一个常见问题是某个任务异常后asyncio.gather会立即抛出异常导致其他正常任务被取消。真实场景中更推荐捕获异常并记录到结果中让错误任务留给上层处理而不是中断整个链路。8. 最佳实践与工程建议代码能跑通只是第一步真正要落地到工程中还需要关注以下这些实践。8.1 让每个 Agent 的职责保持单一不要试图让一个 Agent 什么都干。“写文章”和“查数据”是两种能力“规划”和“执行”是两种职责。一个 Agent 的 system prompt 只描述一个核心目标工具列表也保持精简。工具太多会显著增加模型选错工具的概率。8.2 设计稳定的工具接口工具的参数尽量使用基础类型字符串、数字、布尔值。避免让模型生成复杂嵌套对象。工具的描述写清楚触发条件和参数含义。工具内部做好输入校验不要信任模型生成的任何字段。8.3 日志与可观测性优先多智能体系统比传统 API 服务复杂得多一次用户请求可能触发 10 次以上模型调用。没有日志出了问题几乎无法排查。建议记录每一次模型请求的模型名、prompt 摘要、tokens、耗时。每一次工具调用的函数名、参数、返回值。每一次任务分发的来源和去向。全局链路 ID把一次用户请求的所有日志串起来。8.4 使用模型分级策略强模型贵且慢弱模型便宜且快。工程上最常见的优化是把任务按难度分级规划任务、工具选择、最终汇总使用强模型。简单抽取、格式化、关键词判断使用小模型。一个系统如果全部调用最强模型成本会直线上升如果全部调用小模型拆解任务时又容易出错。8.5 安全边界不可忽视Agent 一旦接了工具就等于把代码执行能力交给了模型。这里有三条底线最小权限原则Agent 的 API Key、数据库账号、文件系统权限只开到完成当前任务所需的最小范围。防止提示注入如果工具会读取外部网页或用户上传文件外部内容里可能包含恶意指令。必须区分“系统指令”和“不可信外部内容”后者只能作为数据处理不能直接作为指令执行。生产环境操作必须审计所有涉及删除、写入、发布的动作记录操作人、操作对象和结果。高风险操作建议加人工确认环节。8.6 从原型到生产关注性能指标原型阶段只需要一个 Agent 循环生产阶段则要考虑吞吐量和延迟。异步编排在原型阶段可能只带来几秒的优化但在生产环境、任务数量较大时是决定性因素。建议在系统设计初期就把异步能力考虑进去。9. 总结与学习路线到这里我们已经从零实现了一个包含子智能体、多智能体协作和异步任务编排的最小系统。回看整条链路核心能力其实就三件事设计好单个 Agent 的循环规划好多个 Agent 之间的分工与消息传递用异步编排把无依赖任务并行起来。配合一个理解“何时调度谁、何时停止循环、失败如何恢复”的 Harness 框架一个可用的多智能体系统就能稳定运行。如果接下来你想继续深入建议按这个顺序学习长期记忆让 Agent 在多次任务之间记住关键信息而不是每次从空白开始。工具调用可靠性研究 structured output、function calling 的进阶用法减少参数解析错误。任务评估给输出结果设计自动化评分机制用评估集验收每次 Prompt 改动。生产级 Harness 框架去阅读 LangGraph、AutoGen、CrewAI 或者各家模型团队推出的 Agent Harness 实现对比它们的调度、重试、状态持久化方案。异步任务队列如果任务量再上台阶把进程内 asyncio 换成消息队列实现分布式多智能体系统。多智能体开发是一个典型的“看起来容易、做起来细节极多”的方向。建议你先跑通本文的代码再自己换一个业务场景比如写技术日报、自动整理会议纪要、做竞品信息汇总把 Supervisor、Worker、异步编排三个模块重新组合一遍。只有亲手改过一轮遇到报错并排查过才能真正理解 Harness 为什么是多智能体系统的核心基础设施。如果在运行过程中遇到问题欢迎在评论区留言交流。