做多智能体开发,绕不开 DeepAgents 这类框架,更绕不开 harness 这个编排层。很多人在单 Agent 上跑得很顺,一旦要让多个子智能体协作,马上就会发现任务不知道怎么拆、消息怎么传、结果怎么汇总、长任务怎么异步编排。这篇文章就围绕 DeepAgents 多智能体开发这条线,从子智能体设计讲到异步任务编排,把环境准备、实现思路、参数取舍和排查方法一起拆开。适合有 Python 基础、想从单 Agent 转向多智能体协作的开发者。下面按我实际操作时的顺序来写,先理解问题,再准备环境,最后落地代码。
1. 先搞清楚 DeepAgents 的多智能体模式到底要解决什么问题
1.1 单智能体与多智能体的本质差异
单个 Agent 能完成不少任务,但它的边界很明显:上下文长度有限、工具链单一、一个环节出错整条链路就断。多智能体的核心思路,是把一个大问题拆成多个子任务,交给不同角色的子智能体处理,再由一个编排层统一调度。这个编排层,就是标题里反复出现的 harness。
harness 这个词第一次接触会觉得绕。你可以把它理解成生产线上的控制台:各台机器负责干活,控制台决定流程怎么走、哪台机器先启动、出现故障时怎么切换。在 DeepAgents 这类框架里,harness 不是单独的模型,而是一段编排代码加一组配置。
这里要强调一点:多智能体不是为了看起来高级才存在的。它真正解决的是一类特征明显的任务——内部存在明确分工、不同环节需要并行处理、单个 Agent 的上下文装不下完整流程。如果只是一个简单问题,直接用单 Agent 反而更稳、更快、更省钱。
1.2 子智能体与异步任务编排分别解决什么问题
子智能体解决的是“职责拆分”问题。一个复杂任务如果让一个 Agent 从头干到尾,它很容易在长链路中迷失,前面步骤的结果可能被遗忘,工具调用也可能混在一起。拆成子智能体之后,每个角色只需要关心自己的输入输出,任务边界清晰,出问题时也容易定位责任环节。
异步任务编排解决的是“等待与吞吐”问题。多智能体任务有两个特点:单个任务耗时长,可能几十秒到几分钟;任务数量多,可能有几十上百条。同步执行意味着第一条任务卡住,后面全部排队。哪怕每条任务只花 10 秒,100 条串行就是 1000 秒,基本不可用。异步编排把能并行的子任务放到后台并行跑,用队列控制并发,再统一收集结果,这才是真实场景下的工作方式。
1.3 为什么先理解这些再动手
我在实际开发里见过太多人跳过理解这一步,直接看代码,结果卡在“为什么我的子智能体之间传不了消息”这种问题上。原因很简单:不理解整体架构,就无法判断报错发生在哪一层。
理解 DeepAgents 的多智能体架构,其实只需要记住一条链路:用户请求进入 harness,harness 根据规则调度子智能体,子智能体各自调用大模型和工具,结果返回给 harness,harness 汇总后输出。后面所有代码和配置,都是围绕这条链路展开的。先有这个框架,再往里面填内容,调试时才不会手忙脚乱。
2. 跑起来之前,环境和依赖要确认哪些条件
2.1 系统与 Python 版本
DeepAgents 相关框架大多以 Python 为主。建议至少选一个相对新的 Python 版本,比如 3.10 或更高,避免异步语法和类型注解上的兼容问题。操作系统方面,Windows、macOS、Linux 都能跑,但 Windows 上要注意两点:路径分隔符和某些依赖库的编译环境。
如果条件允许,我更推荐在 Linux 服务器或者 WSL2 里跑。原因不是 Windows 不行,而是多智能体开发经常要装各种原生依赖,Linux 环境下很多坑可以直接避开。学习阶段用 Windows 也没问题,只是遇到编译类报错时,先确认是否有 Visual Studio Build Tools。
2.2 大模型接入方式怎么选
多智能体的每个子智能体背后都要有大模型支撑。接入方式通常分两类:
- 云端 API:调用兼容接口或各类大模型服务,优点是省机器、部署快,缺点是受网络延迟和限流影响。
- 本地部署:用本地模型服务,优点是调用成本低、数据不出内网,缺点是显存占用高、推理速度取决于硬件。
这里没有绝对最优。学习阶段建议先用云端 API 把流程跑通,再考虑本地部署。如果机器只有 8G 显存,本地跑一个大模型会非常吃力,不如先用 API 理解流程。等逻辑清楚了,再按需切换成本地模型。
2.3 依赖安装和版本确认
安装依赖之前,最好先看官方仓库的 README 或安装文档。DeepAgents 这类框架迭代速度不慢,不同版本的接口差异可能很大。原始教程里用的写法,换一个版本之后可能完全跑不通。
我一般按这个顺序处理:
- 创建独立的 Python 虚拟环境,避免污染全局环境。
- 先装最小依赖集合,不要一次性把所有扩展装齐。
- 跑一个最简单的子智能体调用,确认安装成功。
- 再按需添加工具、向量库、队列等扩展。
依赖冲突是最常见的启动失败原因。比如某些框架对 pydantic 版本有硬性要求,安装时没注意,运行时会报出一堆类型校验错误。遇到这类问题,先看错误堆栈里提到的库名,再用 pip 查看当前版本,和官方文档要求做对比。
3. 子智能体怎么设计:角色、工具和消息协议
3.1 子智能体按职责拆,不按功能拆
子智能体的拆分是整个多智能体架构的起点。拆得好,协作顺畅;拆得散,消息满天飞,结果反而比单 Agent 更慢。
我的建议是:按职责拆,不按功能拆。举例来说,一个“整理多智能体框架调研报告”的任务,可以拆成:
- 规划者:把需求拆解成步骤,决定每一步需要什么信息。
- 调研者:负责搜索、查资料,收集必要信息。
- 写作者:基于调研结果生成报告正文。
- 审查者:检查格式、语气、事实一致性,输出修改意见。
每个子智能体只负责一个明确职责,输入输出尽量收敛成结构化字段,而不是一长段自然语言。这样后续编排和排错都会容易很多。如果职责交叉,比如规划者也去搜索,执行者也去审查,日志会非常混乱,排查成本直线上升。
3.2 定义子智能体的核心配置项
用代码定义子智能体时,通常要配置这几项:
| 配置项 | 作用 | 建议 |
|---|---|---|
| name | 子智能体标识,日志里能看到 | 用 agent_planner 这类清晰命名 |
| role / description | 模型理解角色定位的依据 | 写清楚职责边界 |
| instruction | 具体行为准则 | 越明确越好,包含输入输出格式 |
| tools | 允许调用的工具列表 | 只给必要工具,越少越稳 |
| temperature | 控制输出随机性 | 0.2 到 0.7 之间 |
| max_rounds | 单次任务最大循环次数 | 3 到 10 |
这里最容易犯的错,是把 instruction 写得太泛。只写“你是助手”,子智能体根本不知道边界在哪。更好的写法是:“你负责调研,输入是问题列表,输出是带来源的 JSON 数组,不要执行任何写入操作。”这样模型才能准确理解任务边界。
3.3 先跑通一个最小的多智能体流程
不要一上来就搭五个子智能体加异步队列。我建议先跑两条 Agent 的最小链路,再逐步扩展。最小流程通常是:规划者生成任务清单,执行者按清单逐项执行,最后汇总结果。
示意代码如下:
# 示意代码:不同框架 API 名称会有差异,核心结构相同 planner = create_agent( name="planner", role="任务规划", instruction="把用户需求拆成可执行的步骤列表,输出 JSON。", tools=[], ) executor = create_agent( name="executor", role="执行者", instruction="按照步骤执行,每一步返回执行结果。", tools=[search_tool], ) harness = AgentHarness( agents=[planner, executor], max_rounds=5, timeout=120, ) result = harness.run("帮我整理一份多智能体框架的调研报告")这段代码只是示意,重点是确认三个问题:规划者输出的步骤是否符合预期、执行者能否正确调用工具、最终结果能否正确汇总。能跑通这条链路,再考虑添加更多角色和异步编排。
3.4 子智能体之间要有统一的消息格式
多智能体协作最容易出问题的环节,就是消息格式不一致。规划者输出的是一套字段,执行者期待的却是另一套字段,结果执行者读不懂输入,反复重试或者返回空结果。
解决办法是:为所有子智能体定义统一的数据模型,消息结构保持一致。例如:
{ "task_id": "T-001", "status": "todo", "content": "具体任务内容", "metadata": {} }每个子智能体都遵循这一套结构,消息传递就不会乱。定义好之后,先用一小批样例跑通,再放大规模。这比出了错再回头改字段要省时间得多。
4. 异步任务编排:并发、队列与结果收集
4.1 异步编排的三个核心目标
多智能体任务真正落地时,几乎不可避免要面对批量场景。异步编排的核心目标有三个:
- 多条任务并行跑,提高整体吞吐。
- 对并发数做上限控制,避免 API 限流或机器资源耗尽。
- 对失败任务做记录和重试,不影响整个队列继续运行。
同步模式适合教学和验证,异步模式才适合真实场景。如果你只打算跑三五条测试数据,同步和异步差别不大;一旦任务量上到几百条,异步的方案设计就决定了你是等一个小时还是等十分钟。
4.2 用 asyncio 加信号量控制并发
在 Python 里实现异步任务编排,最直接的方式是 asyncio 搭配信号量。信号量的作用,是限制同时运行的任务数量。
示意代码如下:
import asyncio async def process_one(item): result = await harness.arun(item) return result async def main(): sem = asyncio.Semaphore(5) # 同时最多跑 5 条 async def limited(item): async with sem: return await process_one(item) tasks = [limited(item) for item in task_list] results = await asyncio.gather(*tasks, return_exceptions=True) for i, res in enumerate(results): if isinstance(res, Exception): print(f"任务 {i} 失败: {res}") else: print(f"任务 {i} 成功: {res}") asyncio.run(main())Semaphore(5)这个参数很关键。并发开太高,云端 API 会返回限流错误;开太低,吞吐又上不去。不要一上来就开到 20 或 50,先跑一小批观察耗时和错误率,再慢慢往上调。
4.3 任务状态管理和失败重试
批量任务跑完之后,不能只看“是不是都跑完了”,还要看结果完整性和一致性。我一般会做三件事:
- 统计成功、失败、超时的任务数量。
- 对失败任务单独保存错误消息,方便后续重跑。
- 抽查部分成功结果,确认输出质量和预期一致。
重试要区分错误类型。如果是 API 限流,等几秒重试通常有效;如果是输入格式错误,重试多少次都一样,必须先修正数据。如果是子智能体内部逻辑错误,比如工具调用失败,就要看日志定位具体环节。
4.4 生产环境还要补哪些东西
如果只是学习或者内部跑一批小数据,上面的代码够用了。一旦要放到生产环境,我建议补齐这些组件:
- 持久化队列:服务重启后,未完成任务还能继续跑。
- 任务状态表:记录每个任务的输入、输出、错误信息、耗时。
- 超时控制:每个任务、每轮子智能体调用都要有超时时间。
- 结构化日志:每条日志带上 task_id、agent_name、round 编号,方便检索。
这一层不需要自己从零写,很多消息队列和任务框架都可以直接用。重点是理解:异步编排的核心不是“用了 async 关键字”,而是“任务的调度、并发、重试和状态管理”这套完整机制。
5. 实战中常见的坑和排查顺序
5.1 报错不一定是框架问题
多智能体项目里遇到报错,第一反应不应该是质疑框架。我踩过几次之后发现,大量问题的根源其实集中在这几类:
- 输入格式不符合子智能体的预期。
- 环境变量没配好,比如 API Key、模型名称写错。
- 依赖版本冲突,比如 pydantic 或某个工具库版本不兼容。
- 工具调用时权限不足,或文件路径不存在。
- 上下文过长,超过了模型的最大 token 限制。
排查时先看完整的错误堆栈,不要只看第一行。很多框架会把详细报错写到日志文件里,控制台只显示摘要。如果日志里看不到关键信息,先调整日志级别,把 DEBUG 输出打开。
5.2 子智能体之间消息不一致
这个问题在异步编排里尤其明显。两个子智能体对消息结构的理解不一致,后果不是立刻报错,而是执行者返回空结果、重复处理或者产生错误数据。
预防办法是提前定好协议。所有子智能体的输入输出都使用统一结构,并在 instruction 里明确写出格式要求。比如要求输出 JSON 时,要明确指出字段名、字段类型和示例,而不是只说“输出结果”。此外,可以在 harness 层加一个校验函数,消息进出时自动检查结构,不符合就打回重试或报错,避免脏数据流到下一环。
5.3 死循环和超时问题
子智能体如果被赋予过大的自主权,可能出现“反复调用同一个工具”“来回修改同一个结果”的循环。这是多智能体系统里很经典的问题。
预防办法:
- 设置 max_rounds,限制单个子智能体的最大循环轮数。
- 在 harness 层设置总超时,超时直接终止并记录。
- 给工具调用加上限,比如单个任务最多调用搜索工具 5 次。
- 日志里记录每一轮的调用参数和结果摘要。
我见过不少任务卡死,最后发现不是网络问题,而是某个子智能体在一个死循环里反复调工具。这种问题只能靠日志发现,所以开发阶段就要把每轮调用的输入输出都记录下来,不要等到出问题再补日志。
5.4 推荐的排查链路
遇到问题时,我一般按这个顺序查:
- 看现象:是直接报错、任务卡住,还是输出为空。
- 看输入:原始任务内容、子任务输入是否符合预期格式。
- 看日志:找到对应 task_id 的完整调用链,定位哪一步开始异常。
- 看配置:模型名称、API Key、超时时间、并发数是否合理。
- 看资源:内存、CPU、显存、磁盘是否充足,网络是否稳定。
- 看版本:框架版本、依赖版本和官方文档是否一致。
这个顺序能覆盖大多数问题。最忌讳的是跳过日志直接改参数,改一顿之后发现根本不在那个环节。日志永远是第一排查入口,它不会骗你。
6. 落地边界和几点经验坦白
6.1 小任务不需要上多智能体
我不建议为了用多智能体而用多智能体。一个简单问答、一次单次代码生成,单个 Agent 反而更快更稳定。多智能体和异步编排是有成本的:代码更复杂、日志更多、故障面更大、token 消耗更高。只有当任务确实需要分工、并行或者长链路处理时,这套方案才值得上。
判断标准很简单:你把需求拆开看,如果每一步之间没有明显的依赖边界,或者并行收益很低,那就用单 Agent。不要被“多智能体更高级”这种说法带着走。
6.2 低配置环境能学,但要控制规模
如果你的机器配置不高,比如只有 8G 显存,完全可以跑通多智能体 Demo,但不要期望同时跑大并发。建议做法是:模型换成小体量版本,并发数降到 1 到 2,任务样本控制在 10 条以内。
学习阶段最重要的是理解流程,不是压榨性能。先把两个子智能体的协作跑通,再逐步加角色、加工具、加并发。每一步都确认输出符合预期,再往下一步走。跳过验证直接堆功能,后面排错会非常痛苦。
6.3 成本和质量需要提前评估
多智能体上线前要算一笔账:每个子任务都会消耗 token,多智能体协作的整体 token 消耗通常比单 Agent 高出不少。批量任务跑一轮之后,统计一下平均成本和失败重试带来的额外消耗,再决定是不是需要优化流程。
质量方面也要有合理预期。多智能体不是把模型拆成几个之后效果就一定更好,它只是把流程拆开了。如果每个环节的子智能体能力都不足,最终结果反而可能不如一个强模型直接处理。实际项目中,我倾向于关键环节用强模型,辅助环节用小模型或廉价模型,而不是所有子智能体都用同一个配置。
6.4 最值得记住的一句话
多智能体开发的重心不在模型,而在编排。子智能体怎么拆、消息怎么传、并发怎么控、失败怎么处理,这四件事做好了,系统自然稳定。环境、依赖、日志这些前置功夫,才是长期少踩坑的真正关键。先用最小样例跑通流程,再逐步加复杂度,这个节奏比一次性搭完整套架构要稳妥得多。