这次我们不开箱玩具，直接聊一个真正决定 AI 工程上限的方向：多智能体协作。单个智能体解决单点问题很容易，但要让大模型真正承担一条业务流程，必须把多个智能体组织起来，让它们各自负责一段任务，再通过异步编排把结果汇总成最终产出。DeepAgents 就是围绕这个思路展开的多智能体协作开发框架，本文会从子智能体设计、harness 运行机制、异步任务编排三个层面，给出可落地的开发路径。
先给结论：这套方案的重点不是“多调用几次大模型”，而是把智能体之间的通信、调度、失败重试、并行执行和结果汇总做成一套工程化结构。文章里会包含核心概念梳理、通用代码模板、接口调用示例、批量任务设计，以及一套常见问题排查清单。如果你是做大模型应用开发、智能体平台、自动化流程建设的工程师，这篇文章可以直接收藏，按章节对照落实验证。
1. 核心能力速览
| 能力项 | 说明 |
|---|---|
| 项目类型 | 多智能体协作开发框架与应用方案 |
| 核心概念 | 子智能体、harness、异步任务编排、工具调用 |
| 主要功能 | 多角色智能体构建、任务拆分、并行调度、结果汇总、API 服务 |
| 模型要求 | 通常需要接入具备工具调用能力的 LLM，如 DeepSeek、GPT 系列、Qwen 系列等 |
| 硬件需求 | 调用云端 API 时无 GPU 硬性要求；本地部署需按模型大小评估显存 |
| 启动方式 | 命令行脚本启动 / WebUI / API 服务，取决于具体实现 |
| 是否支持 API | 支持，可通过 HTTP 接口提交任务并获取结果 |
| 是否支持批量任务 | 支持，可设计任务队列与并发执行器 |
| 适合场景 | 内容生产、数据分析报告、代码审查、多步骤流程自动化、知识库问答管线 |
| 使用边界 | 需要合法授权、数据合规、人工复核关键输出 |
需要注意，DeepAgents 的生态仍在快速变化，不同开源实现的目录结构、配置格式和接口路由可能存在差异。下面所有命令和代码均以“通用实现模板”给出，实际使用时请以你下载到的项目 README 为准，替换对应的路径、端口和模型名。
2. 适用场景与使用边界
2.1 这个方案适合谁
如果你手头有这些需求，多智能体协作就是值得投入的方向：
- 内容生产流水线：采集素材、提炼要点、写稿、审校可以由不同智能体分工完成。
- 数据分析报告：数据读取、指标计算、图表描述、报告生成逐级传递。
- 代码仓库审查：一个智能体扫描变更，另一个智能体检查安全风险，再有一个智能体汇总修改建议。
- 客服工单处理：意图识别、知识检索、回复生成、人工复核提示分层处理。
- 知识库问答：检索增强、上下文压缩、答案生成、引用溯源分开管理。
2.2 不适合什么场景
- 单次请求延迟要求极高的场景不适合把链路拆得太长，多智能体协作必然带来多次模型调用，端到端耗时会明显高于单次问答。
- 完全离线且硬件受限的环境需要谨慎，本地跑多个模型需要足够的显存和内存。
- 输出内容直接商用且不经过复核的场景风险很高，智能体生成内容不代表事实正确，必须有人工质检环节。
2.3 合规边界
涉及图像、声音、人物肖像、版权素材时，必须确认授权；涉及用户隐私数据时，要注意脱敏和访问控制；涉及金融、医疗、法律等领域结论时，绝不能把智能体输出当作最终依据。多智能体系统只是执行工具，责任边界在系统设计和业务方。
3. 先厘清四个关键概念
在写代码之前，把几个容易混淆的词说清楚，后面实操才不会卡壳。
3.1 什么是 DeepAgents
DeepAgents 是基于大模型的多智能体协作开发方案，核心思路是让一个“主智能体”负责理解用户任务、拆解子目标、调用不同的“子智能体”完成具体步骤，最后统一汇总输出。它强调的不是单个模型有多强，而是多个模型和工具如何被组织成一条可靠的任务链路。
3.2 什么是子智能体
子智能体是承担单一职责的智能体实例，每个实例可以有自己的系统提示词、工具集、模型参数和上下文窗口管理方式。比如“信息采集子智能体”只负责搜索和抓取，“报告撰写子智能体”只负责根据结构化素材生成文字。子智能体不应该堆砌任务，而是要“小而专”，方便复用和组合。
3.3 什么是 harness
harness 在直译上是“背带、控制装置”，在多智能体系统里通常指智能体的运行外壳和执行环境。你可以把大模型看作“大脑”，harness 就是负责把大脑的输入输出变成可执行动作的“身体”。它处理模型调用循环、工具解析、错误重试、上下文累积、状态管理等底层逻辑。开发多智能体应用时，大部分工程复杂度其实集中在 harness 层，而不是模型本身。
3.4 什么是异步任务编排
异步任务编排是指把多个智能体任务放入队列，通过调度器决定执行顺序、并发数量、依赖关系和结果回收，而不是同步地“等一个完成再开始下一个”。典型结构是：主智能体拆分任务 -> 任务入队 -> 工作进程并发处理 -> 结果按依赖关系合并 -> 最终输出。异步编排解决的是生产效率问题，多智能体系统的吞吐量上限往往由这一层决定。
4. 环境准备与前置条件
4.1 语言与运行时
多智能体开发最常见的语言是 Python，其次是 TypeScript。以下环境清单以 Python 为例：
- Python 3.10 或更高版本。
- pip 包管理器。
- 能够访问大模型 API，并已准备好 API Key。
- 本机可以访问外网，或者已配置可用的模型代理服务。
4.2 建议安装的核心依赖
pip install openai pydantic httpx python-dotenv如果项目本身提供 requirements.txt，直接执行：
pip install -r requirements.txt依赖安装失败时，优先检查 Python 版本和 pip 镜像源配置，不要贸然升级系统级 Python。
4.3 目录结构建议
一套清晰的项目目录可以避免后续批量任务和模型文件管理混乱：
deepagents-demo/ ├── agents/ # 子智能体定义 ├── core/ # harness 与编排核心 ├── tools/ # 自定义工具 ├── tasks/ # 任务队列与批量任务 ├── config/ # 模型配置、提示词配置 ├── outputs/ # 输出结果 └── main.py # 入口脚本4.4 硬件门槛
如果使用云端模型 API，普通开发机即可运行，没有 GPU 硬性要求。如果要在本地部署模型并作为智能体后端，需要按模型参数量评估显存，建议先跑 7B 级别以下模型验证流程，再决定是否升级硬件。显存占用必须以实际砸测为准，不同量化方式、上下文长度和并发数差别很大。
5. 从单智能体到子智能体：先跑通最小流程
5.1 最小单智能体示例
下面这段代码建立一个最基础的单智能体循环。它做的事情是：接收用户输入、调用大模型、返回结果。
import os from openai import OpenAI client = OpenAI( api_key=os.getenv("LLM_API_KEY"), base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"), ) def call_llm(messages, model="gpt-4o-mini", temperature=0.3): response = client.chat.completions.create( model=model, messages=messages, temperature=temperature, ) return response.choices[0].message.content.strip() if __name__ == "__main__": messages = [ {"role": "system", "content": "你是一个专业的写作助手。"}, {"role": "user", "content": "用三句话总结什么是异步任务编排。"}, ] result = call_llm(messages) print(result)这个流程很简单，但它已经具备一个智能体的基本骨架：系统提示词决定角色，用户消息输入目标，模型返回输出。
5.2 把单智能体包装成子智能体
实际开发中，不能每次都手动拼 messages。建议把子智能体抽象成一个类，内部维护系统提示词和上下文策略。
class SubAgent: def __init__(self, name, system_prompt, tools=None, model="gpt-4o-mini"): self.name = name self.system_prompt = system_prompt self.tools = tools or [] self.model = model self.messages = [{"role": "system", "content": system_prompt}] def run(self, user_message): self.messages.append({"role": "user", "content": user_message}) # 这里调用 call_llm，可扩展到工具调用解析 answer = call_llm(self.messages, model=self.model) self.messages.append({"role": "assistant", "content": answer}) return answer这里的关键是：每个子智能体都有自己的消息上下文。不要把不同角色的历史消息混在一起，否则角色边界会失效。
5.3 注册多个子智能体
假设我们要构建一个“行业调研报告生成器”，可以拆出三个子智能体：
collector = SubAgent( name="信息采集", system_prompt="你负责从给定资料中提取事实、数据和关键观点，输出结构化条目。", ) analyst = SubAgent( name="数据分析", system_prompt="你负责对结构化数据进行趋势判断、风险识别和结论归纳。", ) writer = SubAgent( name="报告撰写", system_prompt="你负责根据分析结果撰写正式报告，语言专业、结构清晰。", )到这一步，三个子智能体还彼此独立。接下来要做的是通过一个主智能体或 harness 把这几个子智能体串起来。
6. 编写核心 harness：让智能体真正“动起来”
6.1 harness 要解决什么问题
harness 的核心工作是维护执行循环。一个最简单的 harness 需要做到：
- 接收总任务。
- 根据任务类型路由到对应子智能体。
- 把子智能体的输出整理成结构化结果。
- 保存执行历史，供后续调试和重试。
6.2 路由式 harness 示例
这里给出一个简单的路由式 harness，它根据任务关键词决定调用哪个子智能体。
class Harness: def __init__(self): self.agents = {} def register(self, agent: SubAgent): self.agents[agent.name] = agent def route(self, task_description: str): if "采集" in task_description or "搜索" in task_description: return self.agents["信息采集"] if "分析" in task_description or "趋势" in task_description: return self.agents["数据分析"] if "报告" in task_description or "撰写" in task_description: return self.agents["报告撰写"] raise ValueError("未匹配到合适的子智能体") def execute(self, task_description: str): agent = self.route(task_description) return agent.run(task_description)这种路由方式适合任务类型固定的场景。更复杂的场景需要让“主智能体”自己决定调用哪个子智能体，也就是“LLM 路由”。LLM 路由需要模型输出结构化指令，例如 JSON 格式：
{ "agent": "数据分析", "input": "分析近三个月的销售趋势，判断是否进入下降周期" }harness 解析这个 JSON，再调用对应子智能体。这样主智能体就成了“调度大脑”，子智能体成了“执行者”。
6.3 harness 与 agent 的区别
很多人问 harness 和 agent 到底什么关系。简单讲：agent 是具备推理和行动能力的智能体，harness 是承载 agent 运行的基础设施。一个完整的多智能体系统里，你可以有多个 agent，但通常只有一个 harness 负责调度、记录、重试和上下文压缩。开发时不要把这些职责混在同一个类里，拆分越清楚，越容易测试和扩展。
7. 多智能体协作开发：从串行到并发
7.1 串行流水线
最简单的多智能体协作是串行流水线，前一个智能体的输出作为后一个智能体的输入。
raw_material = collector.run("提取以下新闻中的关键数据：……") analysis_result = analyst.run(raw_material) report = writer.run(analysis_result)串行方式逻辑清晰，适合步骤之间有严格依赖的场景。缺点是整体耗时长，如果某个环节失败，整条链路中断。
7.2 并行执行
有些子任务彼此独立，可以并发执行。比如报告需要同时包含市场分析、竞品分析和用户反馈三部分，三个子智能体可以并行处理。
import asyncio async def run_parallel(): tasks = [ collector.run("收集市场数据"), collector.run("收集竞品信息"), collector.run("收集用户反馈"), ] results = await asyncio.gather(*tasks) return results results = asyncio.run(run_parallel())并行执行能显著提升吞吐量，但要注意模型 API 的并发限制和本地显存压力。
7.3 主智能体编排模式
更接近生产环境的模式是：主智能体先分析任务，生成执行计划，然后并行调度子智能体，最后收集结果并汇总。
class Orchestrator: def __init__(self, harness: Harness): self.harness = harness self.execution_log = [] def run(self, user_request: str): plan = self._generate_plan(user_request) partial_results = [] for step in plan["steps"]: result = self.harness.execute( f"{step['instruction']}\n上下文：{user_request}" ) partial_results.append(result) self.execution_log.append({ "step": step["agent"], "input": step["instruction"], "output": result, }) final = self._summarize(partial_results) return final这里_generate_plan和_summarize同样可以交给大模型完成，但不要忘记给模型提供清晰的输出格式约束。
8. 异步任务编排：队列、调度与失败重试
8.1 为什么需要异步任务编排
多智能体系统一旦进入生产环境，就不适合“同步等结果”了。用户提交任务后，系统应该立刻返回一个任务 ID，后台异步执行，执行完成后通过回调或查询接口返回结果。这种方式对长时间的批量任务尤其重要。
8.2 基于队列的任务编排模型
一个生产可用的异步编排模型包含这几个组件：
- 任务提交接口：接收用户请求，生成任务 ID，入队。
- 任务队列：保存待执行任务，可以是 Redis、RabbitMQ 或简单的内存队列。
- 执行器：从队列拉取任务，调度子智能体执行。
- 结果存储：保存执行状态和最终结果。
- 查询接口：通过任务 ID 查询进度和结果。
8.3 内存队列快速原型
下面是一个基于 asyncio 的内存队列实现，演示多智能体任务如何异步执行。
import asyncio import uuid from dataclasses import dataclass, field @dataclass class AsyncTask: id: str = field(default_factory=lambda: str(uuid.uuid4())) payload: str = "" status: str = "pending" result: str = "" class AsyncOrchestrator: def __init__(self): self.queue = asyncio.Queue() self.tasks = {} async def submit(self, payload: str) -> str: task = AsyncTask(payload=payload) self.tasks[task.id] = task await self.queue.put(task) return task.id async def worker(self): while True: task = await self.queue.get() try: task.status = "running" task.result = await self._run_pipeline(task.payload) task.status = "completed" except Exception as err: task.status = "failed" task.result = str(err) finally: self.queue.task_done() async def _run_pipeline(self, payload: str): raw = await asyncio.to_thread(collector.run, payload) analysis = await asyncio.to_thread(analyst.run, raw) report = await asyncio.to_thread(writer.run, analysis) return report def get_task(self, task_id: str): return self.tasks.get(task_id)启动异步编排服务的入口可以这么写：
async def main(): orch = AsyncOrchestrator() workers = [asyncio.create_task(orch.worker()) for _ in range(3)] task_id = await orch.submit("生成新能源汽车市场调研报告") while orch.get_task(task_id).status == "pending": await asyncio.sleep(1) print(orch.get_task(task_id)) asyncio.run(main())需要注意，内存队列只适合开发和单机测试。生产环境建议换成 Redis/RabbitMQ，并加上持久化，避免进程重启导致任务丢失。
8.4 失败重试与超时控制
异步编排必须考虑失败重试。常见策略是：
- 单个子智能体调用失败时，最多重试 2 到 3 次，重试间隔按指数退避。
- 超过任务总超时时间后，标记任务失败，不再继续后续环节。
- 记录每次重试的输入输出，方便排查是模型问题还是工具问题。
async def call_with_retry(func, *args, retries=3, delay=2): for attempt in range(retries): try: return await func(*args) except Exception as err: if attempt == retries - 1: raise await asyncio.sleep(delay * (2 ** attempt))8.5 批量任务的目录设计
如果要做批量任务，比如一次性处理 100 篇文档，建议目录结构按批次隔离：
outputs/ ├── batch_20260912/ │ ├── task_001/ │ ├── task_002/ │ └── logs/ └── batch_20260913/每个批量任务单独建目录，输入、中间结果、最终输出、日志分开存，即使某个任务失败也不会污染其他任务的结果。
9. 接口 API 与批量任务集成
9.1 提供 HTTP 接口
异步编排系统必须提供两个核心接口：提交任务和查询状态。下面是用 FastAPI 实现的通用模板。
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() orch = AsyncOrchestrator() class TaskRequest(BaseModel): payload: str @app.post("/api/tasks") async def create_task(req: TaskRequest): task_id = await orch.submit(req.payload) return {"task_id": task_id, "status": "pending"} @app.get("/api/tasks/{task_id}") async def get_task(task_id: str): task = orch.get_task(task_id) if not task: return {"error": "task not found"} return {"task_id": task.id, "status": task.status, "result": task.result}启动服务：
uvicorn main:app --host 127.0.0.1 --port 80009.2 curl 调用示例
提交任务：
curl -X POST http://127.0.0.1:8000/api/tasks \ -H "Content-Type: application/json" \ -d '{"payload": "生成2026年智能驾驶行业趋势报告"}'查询任务：
curl http://127.0.0.1:8000/api/tasks/{task_id}接口能跑通之后，就能很方便地接进自己的内部系统、企业微信机器人或自动化脚本里。
9.3 Python 调用示例
import requests import time base_url = "http://127.0.0.1:8000" def submit_and_wait(payload, timeout=600): resp = requests.post(f"{base_url}/api/tasks", json={"payload": payload}, timeout=30) task_id = resp.json()["task_id"] start = time.time() while time.time() - start < timeout: status = requests.get(f"{base_url}/api/tasks/{task_id}", timeout=30).json() if status["status"] in ("completed", "failed"): return status time.sleep(3) return {"task_id": task_id, "status": "timeout"} result = submit_and_wait("整理过去一年AI Agent领域的融资事件") print(result)这里要注意，不同项目的接口路径和字段名可能不同，务必根据实际后端代码调整。
10. 资源占用与性能观察
10.1 需要观察哪些指标
多智能体系统的性能瓶颈通常不在“模型推理有多快”，而在以下几个方面：
- 模型 API 调用延迟：每个子智能体都要调用一次模型，串行链路下总延迟是多次调用的叠加。
- 上下文长度：多个智能体传递结果时，消息累积会导致上下文变长，吞吐量下降。
- 并发数量：并发过高会被模型服务限流，并发过低则浪费资源。
- 队列堆积数量：观察队列待处理数，能判断执行器是否成为瓶颈。
10.2 如何观察显存占用
如果本地部署模型，可以用nvidia-smi观察显存占用。
nvidia-smi --query-gpu=index,memory.used,memory.total,utilization.gpu --format=csv不过需要再次强调，显存占用取决于模型大小、量化方式、上下文长度和并发数，不能凭经验拍脑袋。建议用小并发和小上下文先压测，再逐步放大。
10.3 如何降低资源占用
- 优先使用云模型 API，把显存压力转移给服务端。
- 子智能体之间只传递必要信息，不要整段复制大文本。
- 对历史消息做截断或摘要，控制上下文膨胀。
- 限制并发执行器数量，避免队列任务同时爆发。
- 批量任务分批提交，每批完成后再提交下一批。
10.4 进程残留与端口冲突
开发时反复启动服务，容易遇到端口被占用的现象。Linux 和 macOS 下可以用：
lsof -i :8000找到占用进程后，按需结束进程或更换端口：
uvicorn main:app --host 127.0.0.1 --port 800111. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 依赖安装失败 | Python 版本不符或镜像源问题 | 查看 pip 报错信息 | 切换 Python 版本或配置国内镜像源 |
| 模型接口返回 401 | API Key 无效或无权限 | 检查环境变量 | 重新配置 API Key 或检查服务权限 |
| 子智能体输出为空 | 模型提示词不当或上下文超限 | 打印原始返回结果 | 精简提示词，缩短上下文 |
| 异步任务一直 pending | 执行器未启动或队列卡住 | 检查 worker 进程和日志 | 启动 worker，增加超时退出 |
| 批量任务中断 | 任务无持久化 | 检查队列是否内存态 | 切换到 Redis/RabbitMQ |
| 端口冲突 | 上一次服务未关闭 | lsof 检查端口 | 换端口或结束旧进程 |
| 结果质量不稳定 | 链路太长，误差累积 | 记录每个子智能体的输出 | 增加中间结果人工抽检 |
| 本地显存不足 | 模型过大或并发过高 | nvidia-smi 观察占用 | 换小模型、量化或降并发 |
12. 最佳实践与使用建议
- 第一版不要追求复杂。先跑通主智能体加两个子智能体的小链路，确认模型路由和结果传递没有断层，再扩展更多子智能体。
- 每个子智能体必须有明确的输入输出格式约定。无论是 Markdown 标题还是 JSON 结构，都要固定下来，否则后续无法可靠解析。
- 中间结果一定要落盘。每一步的输出保存到本地，既能调试，也能在最终结果出错时回溯是哪一步出的问题。
- 批量任务必须加日志和失败重试。按批次写日志，记录任务 ID、输入摘要、输出摘要、耗时和错误信息。
- API 服务对外暴露时要做好访问限制。内网环境也要加简单的 Token 鉴权，避免任意机器都能提交任务。
- 涉及真实人物、品牌、版权内容时，务必确认授权。多智能体系统只是把流程自动化，不代表内容可以随意使用。
- 商用之前必须有复核机制。让第二套模型或人工对最终输出做质量检查，不能把智能体结果直接发布。
13. 总结与下一步
DeepAgents 多智能体协作最值得花时间的地方，不是搭几个子智能体那么简单，而是把 harness 调度、异步队列、失败重试、结果汇总做成一套稳定的工程结构。建议你先从“主智能体 + 两三个子智能体 + 串行流水线”开始，跑通之后再引入并发和队列，最后再根据业务需要增加 HTTP 接口和批量任务。
最容易踩的坑是：子智能体职责不清、消息上下文混用、没有失败重试、中间结果不落盘。这四个问题几乎会在所有项目里遇到，尽早用工程手段解决，后面的开发会顺利很多。
下一步可以往两个方向扩展：一是把自定义工具接入 harness，比如让子智能体调用搜索、数据库、代码执行器等外部能力；二是尝试 MCP 协议来统一工具接入方式，让多智能体系统的工具生态更标准。等这两个方向都打通，你会发现多智能体已经从“演示项目”变成了真正能承载业务流程的生产工具。建议收藏备用，动手调试时对照本文的章节定位问题。
