这次我们不开箱玩具直接聊一个真正决定 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 什么是 DeepAgentsDeepAgents 是基于大模型的多智能体协作开发方案核心思路是让一个“主智能体”负责理解用户任务、拆解子目标、调用不同的“子智能体”完成具体步骤最后统一汇总输出。它强调的不是单个模型有多强而是多个模型和工具如何被组织成一条可靠的任务链路。3.2 什么是子智能体子智能体是承担单一职责的智能体实例每个实例可以有自己的系统提示词、工具集、模型参数和上下文窗口管理方式。比如“信息采集子智能体”只负责搜索和抓取“报告撰写子智能体”只负责根据结构化素材生成文字。子智能体不应该堆砌任务而是要“小而专”方便复用和组合。3.3 什么是 harnessharness 在直译上是“背带、控制装置”在多智能体系统里通常指智能体的运行外壳和执行环境。你可以把大模型看作“大脑”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_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL, https://api.openai.com/v1), ) def call_llm(messages, modelgpt-4o-mini, temperature0.3): response client.chat.completions.create( modelmodel, messagesmessages, temperaturetemperature, ) 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, toolsNone, modelgpt-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, modelself.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_factorylambda: 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(payloadpayload) 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, retries3, delay2): 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, timeout600): resp requests.post(f{base_url}/api/tasks, json{payload: payload}, timeout30) task_id resp.json()[task_id] start time.time() while time.time() - start timeout: status requests.get(f{base_url}/api/tasks/{task_id}, timeout30).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-gpuindex,memory.used,memory.total,utilization.gpu --formatcsv不过需要再次强调显存占用取决于模型大小、量化方式、上下文长度和并发数不能凭经验拍脑袋。建议用小并发和小上下文先压测再逐步放大。10.3 如何降低资源占用优先使用云模型 API把显存压力转移给服务端。子智能体之间只传递必要信息不要整段复制大文本。对历史消息做截断或摘要控制上下文膨胀。限制并发执行器数量避免队列任务同时爆发。批量任务分批提交每批完成后再提交下一批。10.4 进程残留与端口冲突开发时反复启动服务容易遇到端口被占用的现象。Linux 和 macOS 下可以用lsof -i :8000找到占用进程后按需结束进程或更换端口uvicorn main:app --host 127.0.0.1 --port 800111. 常见问题与排查方法问题现象可能原因排查方式解决方案依赖安装失败Python 版本不符或镜像源问题查看 pip 报错信息切换 Python 版本或配置国内镜像源模型接口返回 401API Key 无效或无权限检查环境变量重新配置 API Key 或检查服务权限子智能体输出为空模型提示词不当或上下文超限打印原始返回结果精简提示词缩短上下文异步任务一直 pending执行器未启动或队列卡住检查 worker 进程和日志启动 worker增加超时退出批量任务中断任务无持久化检查队列是否内存态切换到 Redis/RabbitMQ端口冲突上一次服务未关闭lsof 检查端口换端口或结束旧进程结果质量不稳定链路太长误差累积记录每个子智能体的输出增加中间结果人工抽检本地显存不足模型过大或并发过高nvidia-smi 观察占用换小模型、量化或降并发12. 最佳实践与使用建议第一版不要追求复杂。先跑通主智能体加两个子智能体的小链路确认模型路由和结果传递没有断层再扩展更多子智能体。每个子智能体必须有明确的输入输出格式约定。无论是 Markdown 标题还是 JSON 结构都要固定下来否则后续无法可靠解析。中间结果一定要落盘。每一步的输出保存到本地既能调试也能在最终结果出错时回溯是哪一步出的问题。批量任务必须加日志和失败重试。按批次写日志记录任务 ID、输入摘要、输出摘要、耗时和错误信息。API 服务对外暴露时要做好访问限制。内网环境也要加简单的 Token 鉴权避免任意机器都能提交任务。涉及真实人物、品牌、版权内容时务必确认授权。多智能体系统只是把流程自动化不代表内容可以随意使用。商用之前必须有复核机制。让第二套模型或人工对最终输出做质量检查不能把智能体结果直接发布。13. 总结与下一步DeepAgents 多智能体协作最值得花时间的地方不是搭几个子智能体那么简单而是把 harness 调度、异步队列、失败重试、结果汇总做成一套稳定的工程结构。建议你先从“主智能体 两三个子智能体 串行流水线”开始跑通之后再引入并发和队列最后再根据业务需要增加 HTTP 接口和批量任务。最容易踩的坑是子智能体职责不清、消息上下文混用、没有失败重试、中间结果不落盘。这四个问题几乎会在所有项目里遇到尽早用工程手段解决后面的开发会顺利很多。下一步可以往两个方向扩展一是把自定义工具接入 harness比如让子智能体调用搜索、数据库、代码执行器等外部能力二是尝试 MCP 协议来统一工具接入方式让多智能体系统的工具生态更标准。等这两个方向都打通你会发现多智能体已经从“演示项目”变成了真正能承载业务流程的生产工具。建议收藏备用动手调试时对照本文的章节定位问题。 DeepAgents多智能体协作开发实战：从子智能体到异步任务编排
这次我们不开箱玩具直接聊一个真正决定 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 什么是 DeepAgentsDeepAgents 是基于大模型的多智能体协作开发方案核心思路是让一个“主智能体”负责理解用户任务、拆解子目标、调用不同的“子智能体”完成具体步骤最后统一汇总输出。它强调的不是单个模型有多强而是多个模型和工具如何被组织成一条可靠的任务链路。3.2 什么是子智能体子智能体是承担单一职责的智能体实例每个实例可以有自己的系统提示词、工具集、模型参数和上下文窗口管理方式。比如“信息采集子智能体”只负责搜索和抓取“报告撰写子智能体”只负责根据结构化素材生成文字。子智能体不应该堆砌任务而是要“小而专”方便复用和组合。3.3 什么是 harnessharness 在直译上是“背带、控制装置”在多智能体系统里通常指智能体的运行外壳和执行环境。你可以把大模型看作“大脑”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_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL, https://api.openai.com/v1), ) def call_llm(messages, modelgpt-4o-mini, temperature0.3): response client.chat.completions.create( modelmodel, messagesmessages, temperaturetemperature, ) 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, toolsNone, modelgpt-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, modelself.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_factorylambda: 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(payloadpayload) 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, retries3, delay2): 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, timeout600): resp requests.post(f{base_url}/api/tasks, json{payload: payload}, timeout30) task_id resp.json()[task_id] start time.time() while time.time() - start timeout: status requests.get(f{base_url}/api/tasks/{task_id}, timeout30).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-gpuindex,memory.used,memory.total,utilization.gpu --formatcsv不过需要再次强调显存占用取决于模型大小、量化方式、上下文长度和并发数不能凭经验拍脑袋。建议用小并发和小上下文先压测再逐步放大。10.3 如何降低资源占用优先使用云模型 API把显存压力转移给服务端。子智能体之间只传递必要信息不要整段复制大文本。对历史消息做截断或摘要控制上下文膨胀。限制并发执行器数量避免队列任务同时爆发。批量任务分批提交每批完成后再提交下一批。10.4 进程残留与端口冲突开发时反复启动服务容易遇到端口被占用的现象。Linux 和 macOS 下可以用lsof -i :8000找到占用进程后按需结束进程或更换端口uvicorn main:app --host 127.0.0.1 --port 800111. 常见问题与排查方法问题现象可能原因排查方式解决方案依赖安装失败Python 版本不符或镜像源问题查看 pip 报错信息切换 Python 版本或配置国内镜像源模型接口返回 401API Key 无效或无权限检查环境变量重新配置 API Key 或检查服务权限子智能体输出为空模型提示词不当或上下文超限打印原始返回结果精简提示词缩短上下文异步任务一直 pending执行器未启动或队列卡住检查 worker 进程和日志启动 worker增加超时退出批量任务中断任务无持久化检查队列是否内存态切换到 Redis/RabbitMQ端口冲突上一次服务未关闭lsof 检查端口换端口或结束旧进程结果质量不稳定链路太长误差累积记录每个子智能体的输出增加中间结果人工抽检本地显存不足模型过大或并发过高nvidia-smi 观察占用换小模型、量化或降并发12. 最佳实践与使用建议第一版不要追求复杂。先跑通主智能体加两个子智能体的小链路确认模型路由和结果传递没有断层再扩展更多子智能体。每个子智能体必须有明确的输入输出格式约定。无论是 Markdown 标题还是 JSON 结构都要固定下来否则后续无法可靠解析。中间结果一定要落盘。每一步的输出保存到本地既能调试也能在最终结果出错时回溯是哪一步出的问题。批量任务必须加日志和失败重试。按批次写日志记录任务 ID、输入摘要、输出摘要、耗时和错误信息。API 服务对外暴露时要做好访问限制。内网环境也要加简单的 Token 鉴权避免任意机器都能提交任务。涉及真实人物、品牌、版权内容时务必确认授权。多智能体系统只是把流程自动化不代表内容可以随意使用。商用之前必须有复核机制。让第二套模型或人工对最终输出做质量检查不能把智能体结果直接发布。13. 总结与下一步DeepAgents 多智能体协作最值得花时间的地方不是搭几个子智能体那么简单而是把 harness 调度、异步队列、失败重试、结果汇总做成一套稳定的工程结构。建议你先从“主智能体 两三个子智能体 串行流水线”开始跑通之后再引入并发和队列最后再根据业务需要增加 HTTP 接口和批量任务。最容易踩的坑是子智能体职责不清、消息上下文混用、没有失败重试、中间结果不落盘。这四个问题几乎会在所有项目里遇到尽早用工程手段解决后面的开发会顺利很多。下一步可以往两个方向扩展一是把自定义工具接入 harness比如让子智能体调用搜索、数据库、代码执行器等外部能力二是尝试 MCP 协议来统一工具接入方式让多智能体系统的工具生态更标准。等这两个方向都打通你会发现多智能体已经从“演示项目”变成了真正能承载业务流程的生产工具。建议收藏备用动手调试时对照本文的章节定位问题。