尧图网页设计 
这次我们来看一套非常契合当下需求的实战内容——Deep-Research 多智能体开发。从技术选型到代码落地完整串一遍多智能体系统怎么做如何用多个 Agent 协作完成深度调研、报告生成和 API 封装。如果你想系统掌握 AI Agent 开发又不满足于只调一个模型接口这篇文章可以直接收藏。先说核心结论Deep-Research 不是一个单一模型而是一整套“规划-检索-写作-审核”的多智能体工作流。传统的单次问答只能拿到一个模型立即生成的回答而 Deep-Research 类系统会先拆解问题再分派多个研究员 Agent 并行查资料、交叉验证最后由写作 Agent 汇总成长文报告。这套能力非常适合行业分析、竞品调研、学术综述、方案生成等场景。本文会从环境准备开始逐步拆解架构设计、角色定义、并行调度、报告生成、API 封装和成本控制。代码部分以 Python 为主接口采用 OpenAI 兼容协议如果你用本地模型或者其他兼容服务只要调整base_url和模型名就能跑通。全程不需要高端显卡CPU 也能完成核心逻辑真正的算力瓶颈在文本生成阶段。1. Deep-Research 多智能体开发核心能力速览能力项说明项目类型Deep-Research 多智能体工作流开发实战核心组件编排器、研究员 Agent、写作 Agent、审核 Agent开发语言Python 3.10依赖框架LangGraph / LangChain / 自研 asyncio 调度模型接入OpenAI 兼容 API 或本地模型服务显存要求不强制使用本地模型时按模型规模评估启动方式Python 脚本直接运行或 FastAPI 封装为服务是否支持 API支持可封装为 HTTP 服务是否支持批量任务支持通过任务队列和并行度控制实现主要能力任务规划、多路并行检索、报告生成、质量审核、结果聚合适合场景行业研究、竞品分析、资料汇总、内容生产、知识工作流需要说明的是这里不限定某一个具体开源项目而是给你一套可以自己实现和扩展的多智能体开发方法论。实际落地时你可以把每个 Agent 替换成不同的模型、不同的工具或不同的知识库接口。2. 适用场景与使用边界2.1 这套多智能体开发适合谁从实际需求来看这套技术最适合三类人群。第一类是知识密集型岗位的技术人员。比如行业分析师、战略研究员、产品经理他们每天需要整理大量资料但真正用来深度思考的时间很少。Deep-Research 多智能体可以把“查资料-读文章-提炼要点-形成报告”这个过程自动化把人从信息搬运中解放出来。第二类是 AI 应用开发者。多智能体不是玩具它是一套可以产品化的架构。你可以把研究员 Agent 换成内部知识库检索把写作 Agent 换成公司报告模板生成器把审核 Agent 换成合规校验器。这就是一个完整的 AI 应用。第三类是技术学习者。如果你想理解 Agent 开发的核心原理从零搭建一个多智能体系统是很好的学习路径。你能看到任务分解、工具调用、上下文管理、并行调度这些关键知识点是如何串联起来的。2.2 使用边界与合规提醒多智能体开发并不是万能的有几个边界要先说清楚。第一信息真实性需要人工复核。Agent 生成的研究报告本质上是基于模型训练数据和检索结果的二次加工可能存在事实偏差、链接失效、数据过时等问题。如果报告用于商业决策、学术发表或公开传播必须在发布前由具备专业能力的人复核。第二版权和引用合规。研究员 Agent 检索到的资料可能有版权限制。生成报告时要保留引用来源商业用途需要确认资料的授权范围。不要直接用多智能体批量扒取并改写他人内容这会带来版权和平台规则风险。第三数据隐私边界。如果研究主题涉及公司内部数据、用户隐私或未公开信息不要直接传给第三方模型 API。建议在这种场景下使用私有化部署的模型或者在传输前做脱敏处理。第四授权问题。如果你要给真实的人脸、声音、作品做分析或生成必须获得明确授权。这在多智能体系统中同样适用不要把技术能力用在侵犯他人权益的方向上。3. 环境准备与前置条件3.1 基础环境要求正式开始写代码之前先确认环境。这套多智能体开发方案不需要 GPU纯 CPU 也能运行因为消耗最大的文本生成是在模型 API 端完成的。你只需要准备一个 Python 环境、一个模型 API 的访问凭证以及足够的磁盘空间存放代码和依赖。推荐环境如下项目推荐配置说明操作系统Windows 10/11、Ubuntu 20.04、macOS 12跨平台核心代码无系统依赖Python3.10 及以上asyncio 和类型注解支持更完整模型服务OpenAI 兼容 API / 本地模型服务只依赖 HTTP 接口不锁模型依赖管理pip / poetry / uv推荐用虚拟环境隔离项目依赖磁盘空间需要 2GB 左右主要是依赖包和日志文件如果你的本地模型是 Ollama、vLLM 或者 Xinference 提供的 OpenAI 兼容接口直接把base_url改成对应服务地址即可。不需要对多智能体代码做额外改动。3.2 安装依赖创建项目目录并初始化虚拟环境mkdir deep-research-agent cd deep-research-agent python3 -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate安装核心依赖pip install openai fastapi uvicorn pydantic python-dotenv这里选的依赖都很轻量。openai用于调用模型接口fastapi和uvicorn用于把多智能体服务封装成 HTTP APIpydantic做数据校验python-dotenv管理环境变量。如果你打算用 LangGraph 做更复杂的图结构控制流可以额外安装pip install langgraph langchain-core不过本文先以自研实现为主LangGraph 方案会在最后提一下。自研实现的好处是你能看清每个环节发生了什么排错更容易。3.3 配置模型接口在项目根目录创建.env文件OPENAI_API_KEYsk-xxxxxxx OPENAI_API_BASEhttps://api.openai.com/v1 MODEL_NAMEgpt-4o-mini然后写一个简单的配置读取模块# config.py import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY) OPENAI_API_BASE os.getenv(OPENAI_API_BASE, https://api.openai.com/v1) MODEL_NAME os.getenv(MODEL_NAME, gpt-4o-mini)如果你用的是本地模型比如 Ollama 启动的qwen2.5:14b只需要改.envOPENAI_API_KEYollama OPENAI_API_BASEhttp://127.0.0.1:11434/v1 MODEL_NAMEqwen2.5:14b这里要注意本地小模型的推理能力会影响整体效果。多智能体系统对模型的指令跟随能力要求比单轮问答更高规划、打分、审查这些环节如果模型太弱很容易输出格式混乱的内容。建议先用强模型跑通流程再逐步替换成更经济的模型。4. 多智能体架构设计角色划分与协作流程多智能体系统的核心不是代码多复杂而是角色怎么划分、任务怎么流转。下面是一套经过实战验证的 Deep-Research 五角色架构。4.1 角色定义角色职责输入输出编排器 Orchestrator接收原始问题拆解研究子任务调度其他 Agent用户问题研究计划、子任务清单研究员 Researcher执行具体检索任务返回带引用的资料摘要子任务结构化资料、来源链接写作专家 Writer汇总所有研究结果生成逻辑连贯的长文报告研究计划 多路研究结果Markdown 报告审核专家 Reviewer检查报告的事实、结构、完整性、引用规范性草稿报告审核意见、修改建议修订器 Reviser根据审核意见修改报告必要时重新调度研究审核意见 草稿最终报告编排器是大脑研究员是手脚写作专家是笔审核专家是质检员修订器是返工工位。整个过程模拟了一个真实团队的协作流程比单个 Agent 一次性输出完整报告要稳定得多。4.2 协作流程我用文字描述一下标准流程代码实现会在下一章给出用户提交一个研究主题比如“梳理 2024 年开源多智能体框架的演进趋势”。编排器先把主题拆成 3 到 5 个子任务例如“梳理主流框架的发布时间和定位”“对比框架的编排方式差异”“分析社区生态活跃度”。多个研究员 Agent 并行执行子任务每个研究员只负责自己的部分输出结构化摘要和来源链接。写作专家拿到所有研究结果按照编排器给出的报告大纲组织内容生成初稿。审核专家检查初稿。如果发现信息缺失、逻辑断裂、引用缺失就输出修改意见。修订器根据审核意见修改或者把缺失部分重新派发给研究员补充。最终报告输出给用户。这个流程里最关键的设计是让每个 Agent 只做一件事。研究员不要写报告写作专家不要检索资料审核专家不要同时修改内容。职责越单一输出质量越稳定排错也越容易。5. 代码实战从单 Agent 到多智能体协作5.1 先实现一个简单的模型调用模块写多智能体之前先确保单次模型调用是通的。创建llm.py# llm.py from openai import OpenAI import config client OpenAI( api_keyconfig.OPENAI_API_KEY, base_urlconfig.OPENAI_API_BASE, ) def chat(messages, temperature0.3, max_tokens4096): 通用的模型调用函数。 messages: [{role: system, content: ...}, {role: user, content: ...}] response client.chat.completions.create( modelconfig.MODEL_NAME, messagesmessages, temperaturetemperature, max_tokensmax_tokens, ) return response.choices[0].message.content先用这个最简函数验证一下链路是否通# test_llm.py from llm import chat result chat([ {role: system, content: 你是一个测试助手只回复一句话。}, {role: user, content: 你好请回复多智能体链路正常。} ]) print(result)python test_llm.py能正常输出内容说明模型接口配置没问题。这一步通过后再往上叠多智能体结构。5.2 定义 Agent 基类多智能体本质上就是多个带独立系统提示词的对话循环。为了方便管理先定义一个基础 Agent 类# agent_base.py from typing import List, Dict, Optional from llm import chat class BaseAgent: 所有 Agent 的基类维护自己的系统提示词和上下文窗口 def __init__(self, name: str, system_prompt: str, temperature: float 0.3): self.name name self.system_prompt system_prompt self.temperature temperature self.messages: List[Dict[str, str]] [ {role: system, content: system_prompt} ] def run(self, task: str) - str: 执行一次任务任务内容会追加到上下文中 self.messages.append({role: user, content: task}) result chat(self.messages, temperatureself.temperature) self.messages.append({role: assistant, content: result}) return result def reset(self): 清空上下文只保留系统提示词 self.messages [{role: system, content: self.system_prompt}]这个基类的优点是简单直观。每个 Agent 实例维护自己的上下文互不干扰。如果你想实现更精细的记忆管理可以在此基础上扩展比如定期裁剪历史消息、只保留最近 N 轮对话、把长期记忆存到向量数据库等。5.3 编排器任务分解与规划编排器是多智能体系统的第一个环节。它不负责具体检索只负责把大问题拆解成可执行的子任务。# orchestrator.py import json from agent_base import BaseAgent ORCHESTRATOR_PROMPT 你是一个深度研究任务的编排器。你的职责是 1. 分析用户给出的研究主题 2. 将其拆解为 3 到 5 个彼此独立、可并行执行的子任务 3. 为每个子任务指定一个研究员角色和检索方向 严格按以下 JSON 格式输出不要输出其他内容 { topic: 研究主题, report_outline: [章节1, 章节2, 章节3], subtasks: [ { id: 1, role: researcher, direction: 检索方向说明, question: 子任务要回答的具体问题 } ] } class Orchestrator(BaseAgent): def __init__(self): super().__init__(orchestrator, ORCHESTRATOR_PROMPT, temperature0.1) def plan(self, topic: str) - dict: raw self.run(topic) # 提取 JSON 部分 start raw.find({) end raw.rfind(}) 1 if start -1 or end 0: raise ValueError(编排器输出不包含有效 JSON) data json.loads(raw[start:end]) return data这里的关键点是使用 JSON 作为中间传输格式。多智能体系统里Agent 之间的通信必须结构化不能直接传自然语言文本否则下游 Agent 解析起来会很痛苦。编排器的输出结构直接决定后续研究员能不能并行执行。5.4 研究员 Agent检索与摘要接下来是研究员。研究员的核心职责是接收一个子任务执行检索把检索结果整理成结构化摘要。# researcher.py import json from agent_base import BaseAgent RESEARCHER_PROMPT 你是一个深度研究员。你的工作流程是 1. 检索与任务相关的资料 2. 提炼关键信息整理成结构化摘要 3. 输出必须包含信息来源 如果没有真实检索工具请基于你训练数据中的知识进行合理整理 但必须在输出中注明基于模型知识整理建议人工核对。 输出格式 { subtask_id: 1, summary: 300字以内的研究摘要, key_points: [要点1, 要点2, 要点3], sources: [来源1, 来源2], confidence: high/medium/low, warning: 如有不确定性或风险在此说明 } class Researcher(BaseAgent): def __init__(self, name: str): super().__init__(name, RESEARCHER_PROMPT, temperature0.3) def research(self, subtask_id: int, question: str) - dict: task f子任务 {subtask_id}: {question} raw self.run(task) start raw.find({) end raw.rfind(}) 1 if start -1 or end 0: raise ValueError(f研究员 {self.name} 输出不包含有效 JSON) data json.loads(raw[start:end]) data[subtask_id] subtask_id return data实际工程中研究员这个角色通常要接入真实的检索工具比如搜索引擎 API、内部知识库、数据库、爬虫等。你可以在RESEARCHER_PROMPT里描述这些工具的使用方式或者在上面的research方法里先调用检索工具再把结果作为上下文传给模型。5.5 写作专家 Agent生成报告多个研究员并行完成子任务后写作专家需要把所有结果汇总成一篇完整的报告。# writer.py import json from agent_base import BaseAgent WRITER_PROMPT 你是一个深度报告写作专家。你的职责是 1. 根据研究计划和多路研究结果生成结构清晰的长文报告 2. 报告必须包含标题、目录、正文、来源引用 3. 所有关键数据必须基于研究员提供的结果不得凭空编造 4. 如果研究结果中存在冲突要明确指出并以最可信来源为准 输出格式直接输出 Markdown 格式报告不需要 JSON 包装。 报告开头需要包含一个信息来源章节列出所有引用过的来源。 class Writer(BaseAgent): def __init__(self): super().__init__(writer, WRITER_PROMPT, temperature0.4) def write_report(self, plan: dict, research_results: list) - str: context f 研究主题: {plan[topic]} 报告大纲: {json.dumps(plan[report_outline], ensure_asciiFalse)} 研究结果如下: {json.dumps(research_results, ensure_asciiFalse, indent2)} return self.run(context)写作专家不需要输出 JSON直接输出 Markdown 报告。因为报告是最终交付物适合给人阅读的格式。5.6 审核专家与修订器质量闭环审核专家检查报告修订器负责返工。这里用两段式实现一个输出审核意见一个根据意见修改。# reviewer.py import json from agent_base import BaseAgent REVIEWER_PROMPT 你是一个严格的质量审核专家。检查以下维度 1. 事实准确性是否有明显的错误或未经验证的断言 2. 结构完整性是否覆盖研究计划的全部章节 3. 引用规范关键数据和结论是否有来源支撑 4. 逻辑连贯章节之间是否有清晰的逻辑关系 5. 格式要求是否为规范的 Markdown 输出 JSON 格式 { passed: true/false, score: 0-100, issues: [ {type: fact|structure|reference|logic|format, description: 问题描述, suggestion: 修改建议} ] } class Reviewer(BaseAgent): def __init__(self): super().__init__(reviewer, REVIEWER_PROMPT, temperature0.1) def review(self, report: str) - dict: raw self.run(report) start raw.find({) end raw.rfind(}) 1 if start -1 or end 0: raise ValueError(审核专家输出不包含有效 JSON) return json.loads(raw[start:end])修订器逻辑如下# revisor.py import json from agent_base import BaseAgent REVISOR_PROMPT 你是一个报告修订专家。根据审核意见修改报告。 要求 1. 只修改审核意见中指出的问题 2. 保持原有报告的整体结构和风格 3. 对不确定的信息用据调研或待核实等措辞标注 直接输出修订后的完整 Markdown 报告不要输出其他内容。 class Revisor(BaseAgent): def __init__(self): super().__init__(revisor, REVISOR_PROMPT, temperature0.4) def revise(self, report: str, review_result: dict) - str: task f 报告草稿如下 {report} 审核意见如下 {json.dumps(review_result, ensure_asciiFalse, indent2)} 请根据审核意见进行修订输出完整报告。 return self.run(task)6. 并行任务调度与结果聚合6.1 用 asyncio 实现并行研究多个研究员 Agent 之间没有依赖关系可以并行执行。Python 的asyncio.gather是这里最直接的方案。# deep_research.py import asyncio from orchestrator import Orchestrator from researcher import Researcher from writer import Writer from reviewer import Reviewer from revisor import Revisor class DeepResearchSystem: def __init__(self, researcher_count: int 3, max_review_rounds: int 2): self.orchestrator Orchestrator() self.researchers [Researcher(fresearcher-{i}) for i in range(researcher_count)] self.writer Writer() self.reviewer Reviewer() self.revisor Revisor() self.max_review_rounds max_review_rounds async def _run_research_task(self, subtask: dict): 单个研究员执行一个子任务 idx subtask[id] # 按 id 取模分配研究员保证每个研究员在同一轮只处理一个任务 researcher self.researchers[idx % len(self.researchers)] return await asyncio.to_thread(researcher.research, idx, subtask[question]) async def research(self, topic: str) - dict: # 1. 编排器规划 plan self.orchestrator.plan(topic) print(f[Orchestrator] 研究主题: {plan[topic]}) print(f[Orchestrator] 拆分为 {len(plan[subtasks])} 个子任务) # 2. 并行研究 research_tasks [ self._run_research_task(subtask) for subtask in plan[subtasks] ] research_results await asyncio.gather(*research_tasks, return_exceptionsTrue) # 3. 处理异常结果 success_results [] for idx, result in enumerate(research_results): if isinstance(result, Exception): print(f[Researcher] 子任务 {idx 1} 执行失败: {result}) continue success_results.append(result) if not success_results: raise RuntimeError(所有研究任务均执行失败) # 4. 写作专家生成报告 print([Writer] 正在生成报告草稿...) draft self.writer.write_report(plan, success_results) print(f[Writer] 草稿长度: {len(draft)} 字符) # 5. 审核与修订循环 report draft for round_idx in range(self.max_review_rounds): print(f[Reviewer] 第 {round_idx 1} 轮审核...) review_result self.reviewer.review(report) print(f[Reviewer] 得分: {review_result.get(score, N/A)} f通过: {review_result.get(passed, False)}) if review_result.get(passed, False): print([Reviewer] 审核通过无需修订) break print(f[Revisor] 根据审核意见修订中问题数: f{len(review_result.get(issues, []))}) report self.revisor.revise(report, review_result) else: print([System] 达到最大修订轮数使用最后一次修订结果) return { topic: topic, plan: plan, research_results: success_results, report: report, }主函数入口# main.py import asyncio from deep_research import DeepResearchSystem async def main(): system DeepResearchSystem(researcher_count3, max_review_rounds2) result await system.research(梳理开源多智能体框架的核心设计模式) print(\n 最终报告摘要 ) print(result[report][:2000]) if __name__ __main__: asyncio.run(main())python main.py如果你的机器是 Windows且 Python 版本较旧遇到事件循环相关的问题可以先升级 Python 到 3.10 以上或者检查是否安装了winloop等兼容库。在大多数情况下直接运行不会有大问题。6.2 任务队列与批量任务如果一次要处理几十个研究主题直接全部并发会打爆模型接口限流也不方便管理失败重试。更好的做法是引入一个任务队列用固定数量的 worker 消费队列。# batch_runner.py import asyncio from deep_research import DeepResearchSystem async def process_topic(semaphore, system, topic): async with semaphore: print(f[Batch] 开始处理: {topic}) try: result await system.research(topic) return {topic: topic, status: success, report: result[report]} except Exception as e: print(f[Batch] 处理失败: {topic}, 错误: {e}) return {topic: topic, status: failed, error: str(e)} async def run_batch(topics, max_concurrency2, researcher_count3): semaphore asyncio.Semaphore(max_concurrency) system DeepResearchSystem(researcher_countresearcher_count) tasks [process_topic(semaphore, system, topic) for topic in topics] results await asyncio.gather(*tasks) return results if __name__ __main__: topics [ 开源 RAG 框架的技术选型对比, 2024 年多智能体框架的发展趋势, 企业级 Agent 应用的落地难点分析, ] results asyncio.run(run_batch(topics, max_concurrency2)) for r in results: print(f{r[topic]} - {r[status]})批量任务里最关键的设计是max_concurrency。这个参数控制同时运行的调研任务数量避免把 API 限流打爆。建议从 2 到 3 开始根据实际响应时间和 API 配额慢慢调大。如果你需要更强的批量任务管理能力可以把任务状态持久化到数据库加上失败重试、进度上报、结果落盘等逻辑。生产环境建议使用 Celery、Arq 或 Temporal 这类成熟任务队列而不是纯 asyncio。7. 接口 API 封装与外部调用多智能体系统的价值在于可以被外部系统调用。用 FastAPI 把DeepResearchSystem封装成 HTTP 服务是最常见的方式。7.1 FastAPI 服务封装# api.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from typing import Optional import asyncio from deep_research import DeepResearchSystem app FastAPI(titleDeep Research Agent API) class ResearchRequest(BaseModel): topic: str Field(..., description研究主题) researcher_count: int Field(3, ge1, le10, description研究员 Agent 数量) max_review_rounds: int Field(2, ge0, le5, description最大审核轮数) class ResearchResponse(BaseModel): topic: str status: str report: Optional[str] None error: Optional[str] None app.get(/health) async def health(): return {status: ok} app.post(/api/research, response_modelResearchResponse) async def research(req: ResearchRequest): try: system DeepResearchSystem( researcher_countreq.researcher_count, max_review_roundsreq.max_review_rounds, ) loop asyncio.get_event_loop() result await loop.run_in_executor( None, lambda: asyncio.run(system.research(req.topic)), ) return ResearchResponse( topicreq.topic, statussuccess, reportresult[report], ) except Exception as e: return ResearchResponse( topicreq.topic, statusfailed, errorstr(e), )启动服务uvicorn api:app --host 127.0.0.1 --port 8000这里要注意asyncio.run和 FastAPI 的事件循环不能直接混用。上面的写法通过run_in_executor把异步任务放到线程池里执行避免了事件循环冲突。如果你的模型调用本身是同步阻塞的这个方案是安全且简单的。7.2 调用测试服务启动后用 curl 验证curl -X POST http://127.0.0.1:8000/api/research \ -H Content-Type: application/json \ -d {topic: 多智能体系统在金融风控中的应用, researcher_count: 3}用 Python 调用import requests url http://127.0.0.1:8000/api/research payload { topic: 开源 Agent 开发框架对比, researcher_count: 3, max_review_rounds: 2, } response requests.post(url, jsonpayload, timeout600) data response.json() print(data[status]) print(data[report][:500] if data[report] else data[error])接口设计上有几个注意事项第一超时时间要设长。Deep-Research 任务涉及多轮模型调用和并行检索耗时通常在几十秒到几分钟不等。前端调用时把超时设置到 5 分钟以上或者直接用异步任务 轮询结果的方式。第二如果服务要部署到公网必须加鉴权。最简单的方案是在中间件里校验Authorization请求头或者使用 API Key。不要裸奔没有认证的服务。第三针对长任务更好的设计是“提交任务-返回任务 ID-轮询结果”两步走。这样即使客户端断线任务也不会丢失。8. 成本控制与性能观察8.1 多智能体系统的 Token 消耗多智能体系统看起来只是调了好几次模型实际上 Token 消耗会比单次问答高出一个数量级。原因是每个 Agent 都有自己的系统提示词和上下文研究员之间互不共享上下文导致信息重复传递。以一次标准研究任务为例环节人/次数单次 Token 估算小计编排器规划1 次500-1000约 1000研究员并行调研3 个子任务每个 1500-3000约 9000写作专家生成1 次2000-5000约 5000审核专家1 次2000-4000约 4000修订器1 次2000-4000约 4000合计--约 23000 Token如果你的模型是gpt-4o-mini这类低成本模型这个量级还可以接受。但如果换成大杯模型一次研究任务的成本就会明显上升。在做预算之前先用一个小主题跑通全流程量一下实际 Token。8.2 如何降低 Token 消耗降低 Token 消耗有几个实用手段。第一精简系统提示词。每个 Agent 的系统提示词不是越长越好。把核心指令控制在 200 字以内能显著减少每轮调用的基础开销。第二限制研究结果的长度。研究员输出的summary字段限制在 300 字以内key_points限制在 5 条以内。结构化摘要比原文引用省 Token 得多。第三审核不通过时才触发修订。max_review_rounds设成 0 可以让系统只生成不审核适合低成本快速产出场景。设成 1 到 2 是质量和成本之间的平衡点。第四做结果缓存。相同或相似的研究主题第一次研究完成后把报告和中间结果缓存到本地。第二次直接读取缓存不再调用模型。import hashlib import json import os CACHE_DIR ./cache os.makedirs(CACHE_DIR, exist_okTrue) def get_cache_key(topic: str) - str: return hashlib.md5(topic.encode(utf-8)).hexdigest() def load_cache(topic: str) - dict | None: key get_cache_key(topic) path os.path.join(CACHE_DIR, f{key}.json) if os.path.exists(path): with open(path, r, encodingutf-8) as f: return json.load(f) return None def save_cache(topic: str, result: dict): key get_cache_key(topic) path os.path.join(CACHE_DIR, f{key}.json) with open(path, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2)8.3 性能观察要点观察多智能体系统性能重点看三个指标。第一个是模型 API 的响应延迟。总耗时的绝大部分都花在模型推理上。你可以把每次调用的耗时、Token 数、模型名记录到日志里。第二个是任务队列积压情况。批量任务跑起来后观察队列里等待的任务数量。如果每天都在积压需要调大并发度或减少任务量。第三个是失败重试率。模型 API 偶发超时、限流、网络抖动都很正常。如果失败率超过 5%优先检查base_url配置、网络连通性和 API 配额而不是代码逻辑。9. 常见问题与排查方法问题现象可能原因排查方式解决方案模型调用报 401/403API Key 错误或过期检查.env文件中的 key 和前导空格重新生成 key确认配置读取正确模型调用报 404base_url 路径不对检查 OpenAI 兼容服务的路径是否包含/v1在 base_url 后补上/v1路径编排器输出 JSON 解析失败模型输出格式不符合要求打印原始输出检查是否包含额外文本在提示词中增强格式约束或使用response_format参数研究员任务超时子任务太复杂或模型响应过慢查看单次任务耗时日志拆细子任务减少单次上下文长度审核结果质量差模型能力不足或审核维度不清晰检查审核 Prompt 是否明确逐条列出审核维度或换更强模型批量任务卡住并发过高触发 API 限流检查 API 配额和错误日志降低 max_concurrency增加退避重试报告内容重复多个研究员输出了相似结果检查子任务是否交叉重叠在编排器 Prompt 中明确子任务边界引用缺失或虚假研究员没有真实检索工具查看 sources 字段是否为空接入真实检索 API或至少让模型标注不确定性服务启动端口被占用8000 端口被其他进程占用用lsof -i:8000或netstat -ano查看更换端口uvicorn api:app --port 8001请求超时任务耗时太长客户端提前断开在客户端设置较长的超时时间改用异步任务 轮询模式排查逻辑其实有规律先看模型接口通不通再看中间 JSON 解析是否成功最后看业务逻辑是否正确。多智能体系统比单 Agent 多了一层“多角色协作”的复杂度所以日志特别重要。每个 Agent 的关键输入输出都要打好日志否则出了问题根本不知道是哪个环节出错。10. 最佳实践与后续扩展建议去视频化的技术文通常会在最后给一点工程化沉淀。这里把实战中比较有用的经验列出来。10.1 从最小闭环开始第一次写多智能体系统不要一上来就接工具、接数据库、接复杂编排。先做一个最小闭环编排器 → 研究员 → 写作专家 → 输出。跑通之后再逐步加入审核、修订、并行调度、API 封装。每一步都验证过再往下走排错成本会低很多。10.2 结构化中间结果是生命线多智能体系统里Agent 之间传递的中间结果必须结构化。用 JSON 定义每个 Agent 的输出格式比让 Agent 自由发挥要稳定得多。我建议给每个输出格式写一个 Pydantic 模型用model_validate_json做校验解析失败就重试一次而不是直接抛异常。from pydantic import BaseModel, Field from typing import List class Subtask(BaseModel): id: int role: str direction: str question: str class Plan(BaseModel): topic: str report_outline: List[str] subtasks: List[Subtask]10.3 接入真实工具链上面演示的研究员 Agent 用的是模型知识没有接入真实检索工具。生产环境要接入搜索引擎 API、Firecrawl、内部知识库或者数据库查询。你可以在研究员的research方法里先调用检索函数再把检索结果注入上下文让模型基于检索结果生成摘要。10.4 换用 LangGraph 做复杂控制流如果你的任务不是简单的“规划-并行-汇总-审核”线性流程而是有条件分支、循环嵌套、人工审批穿插的复杂流程建议用 LangGraph。它把多 Agent 协作建模成一张图节点是 Agent 或工具边是控制流逻辑可观测性和可控性都比自研循环好。10.5 测试时先小后大跑批量任务之前先用一个主题、一个研究员、不加审核跑一遍。确认输出格式、响应时间、Token 消耗都符合预期之后再逐步加大并发和任务数量。不要一上来就 50 个主题并发那不是提效是给自己找麻烦。11. 总结与下一步Deep-Research 多智能体开发的核心不是“调用多个模型”而是把复杂任务拆成多个职责单一的 Agent用结构化中间结果串联协作流程通过并行调度提效通过审核修订保证质量。这套方法不绑定具体框架也不绑定具体模型换模型、换工具、换场景都能复用。建议你上手后最先验证三件事第一编排器能否稳定输出合法 JSON 的子任务列表第二多个研究员并行后能否正确聚合结果第三审核-修订循环能否真正提升报告质量。这三步跑通了整个系统就立住了。最容易踩的坑是过度设计。很多初学者一上来就搭复杂的 Agent 通信协议、消息总线、记忆系统结果跑都跑不起来。先把一个最简单的五角色流程跑通再逐步加复杂度。后续可以继续扩展的方向包括接入真实搜索引擎 API、给研究员 Agent 挂上 RAG 工具、把报告输出接成带数据可视化的网页、加任务队列做成异步批处理平台。每一步扩展都是单独一篇实战文章的量。建议收藏备用动手写的时候对照着搭。 Deep-Research多智能体开发实战：从任务编排到API封装
这次我们来看一套非常契合当下需求的实战内容——Deep-Research 多智能体开发。从技术选型到代码落地完整串一遍多智能体系统怎么做如何用多个 Agent 协作完成深度调研、报告生成和 API 封装。如果你想系统掌握 AI Agent 开发又不满足于只调一个模型接口这篇文章可以直接收藏。先说核心结论Deep-Research 不是一个单一模型而是一整套“规划-检索-写作-审核”的多智能体工作流。传统的单次问答只能拿到一个模型立即生成的回答而 Deep-Research 类系统会先拆解问题再分派多个研究员 Agent 并行查资料、交叉验证最后由写作 Agent 汇总成长文报告。这套能力非常适合行业分析、竞品调研、学术综述、方案生成等场景。本文会从环境准备开始逐步拆解架构设计、角色定义、并行调度、报告生成、API 封装和成本控制。代码部分以 Python 为主接口采用 OpenAI 兼容协议如果你用本地模型或者其他兼容服务只要调整base_url和模型名就能跑通。全程不需要高端显卡CPU 也能完成核心逻辑真正的算力瓶颈在文本生成阶段。1. Deep-Research 多智能体开发核心能力速览能力项说明项目类型Deep-Research 多智能体工作流开发实战核心组件编排器、研究员 Agent、写作 Agent、审核 Agent开发语言Python 3.10依赖框架LangGraph / LangChain / 自研 asyncio 调度模型接入OpenAI 兼容 API 或本地模型服务显存要求不强制使用本地模型时按模型规模评估启动方式Python 脚本直接运行或 FastAPI 封装为服务是否支持 API支持可封装为 HTTP 服务是否支持批量任务支持通过任务队列和并行度控制实现主要能力任务规划、多路并行检索、报告生成、质量审核、结果聚合适合场景行业研究、竞品分析、资料汇总、内容生产、知识工作流需要说明的是这里不限定某一个具体开源项目而是给你一套可以自己实现和扩展的多智能体开发方法论。实际落地时你可以把每个 Agent 替换成不同的模型、不同的工具或不同的知识库接口。2. 适用场景与使用边界2.1 这套多智能体开发适合谁从实际需求来看这套技术最适合三类人群。第一类是知识密集型岗位的技术人员。比如行业分析师、战略研究员、产品经理他们每天需要整理大量资料但真正用来深度思考的时间很少。Deep-Research 多智能体可以把“查资料-读文章-提炼要点-形成报告”这个过程自动化把人从信息搬运中解放出来。第二类是 AI 应用开发者。多智能体不是玩具它是一套可以产品化的架构。你可以把研究员 Agent 换成内部知识库检索把写作 Agent 换成公司报告模板生成器把审核 Agent 换成合规校验器。这就是一个完整的 AI 应用。第三类是技术学习者。如果你想理解 Agent 开发的核心原理从零搭建一个多智能体系统是很好的学习路径。你能看到任务分解、工具调用、上下文管理、并行调度这些关键知识点是如何串联起来的。2.2 使用边界与合规提醒多智能体开发并不是万能的有几个边界要先说清楚。第一信息真实性需要人工复核。Agent 生成的研究报告本质上是基于模型训练数据和检索结果的二次加工可能存在事实偏差、链接失效、数据过时等问题。如果报告用于商业决策、学术发表或公开传播必须在发布前由具备专业能力的人复核。第二版权和引用合规。研究员 Agent 检索到的资料可能有版权限制。生成报告时要保留引用来源商业用途需要确认资料的授权范围。不要直接用多智能体批量扒取并改写他人内容这会带来版权和平台规则风险。第三数据隐私边界。如果研究主题涉及公司内部数据、用户隐私或未公开信息不要直接传给第三方模型 API。建议在这种场景下使用私有化部署的模型或者在传输前做脱敏处理。第四授权问题。如果你要给真实的人脸、声音、作品做分析或生成必须获得明确授权。这在多智能体系统中同样适用不要把技术能力用在侵犯他人权益的方向上。3. 环境准备与前置条件3.1 基础环境要求正式开始写代码之前先确认环境。这套多智能体开发方案不需要 GPU纯 CPU 也能运行因为消耗最大的文本生成是在模型 API 端完成的。你只需要准备一个 Python 环境、一个模型 API 的访问凭证以及足够的磁盘空间存放代码和依赖。推荐环境如下项目推荐配置说明操作系统Windows 10/11、Ubuntu 20.04、macOS 12跨平台核心代码无系统依赖Python3.10 及以上asyncio 和类型注解支持更完整模型服务OpenAI 兼容 API / 本地模型服务只依赖 HTTP 接口不锁模型依赖管理pip / poetry / uv推荐用虚拟环境隔离项目依赖磁盘空间需要 2GB 左右主要是依赖包和日志文件如果你的本地模型是 Ollama、vLLM 或者 Xinference 提供的 OpenAI 兼容接口直接把base_url改成对应服务地址即可。不需要对多智能体代码做额外改动。3.2 安装依赖创建项目目录并初始化虚拟环境mkdir deep-research-agent cd deep-research-agent python3 -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate安装核心依赖pip install openai fastapi uvicorn pydantic python-dotenv这里选的依赖都很轻量。openai用于调用模型接口fastapi和uvicorn用于把多智能体服务封装成 HTTP APIpydantic做数据校验python-dotenv管理环境变量。如果你打算用 LangGraph 做更复杂的图结构控制流可以额外安装pip install langgraph langchain-core不过本文先以自研实现为主LangGraph 方案会在最后提一下。自研实现的好处是你能看清每个环节发生了什么排错更容易。3.3 配置模型接口在项目根目录创建.env文件OPENAI_API_KEYsk-xxxxxxx OPENAI_API_BASEhttps://api.openai.com/v1 MODEL_NAMEgpt-4o-mini然后写一个简单的配置读取模块# config.py import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY) OPENAI_API_BASE os.getenv(OPENAI_API_BASE, https://api.openai.com/v1) MODEL_NAME os.getenv(MODEL_NAME, gpt-4o-mini)如果你用的是本地模型比如 Ollama 启动的qwen2.5:14b只需要改.envOPENAI_API_KEYollama OPENAI_API_BASEhttp://127.0.0.1:11434/v1 MODEL_NAMEqwen2.5:14b这里要注意本地小模型的推理能力会影响整体效果。多智能体系统对模型的指令跟随能力要求比单轮问答更高规划、打分、审查这些环节如果模型太弱很容易输出格式混乱的内容。建议先用强模型跑通流程再逐步替换成更经济的模型。4. 多智能体架构设计角色划分与协作流程多智能体系统的核心不是代码多复杂而是角色怎么划分、任务怎么流转。下面是一套经过实战验证的 Deep-Research 五角色架构。4.1 角色定义角色职责输入输出编排器 Orchestrator接收原始问题拆解研究子任务调度其他 Agent用户问题研究计划、子任务清单研究员 Researcher执行具体检索任务返回带引用的资料摘要子任务结构化资料、来源链接写作专家 Writer汇总所有研究结果生成逻辑连贯的长文报告研究计划 多路研究结果Markdown 报告审核专家 Reviewer检查报告的事实、结构、完整性、引用规范性草稿报告审核意见、修改建议修订器 Reviser根据审核意见修改报告必要时重新调度研究审核意见 草稿最终报告编排器是大脑研究员是手脚写作专家是笔审核专家是质检员修订器是返工工位。整个过程模拟了一个真实团队的协作流程比单个 Agent 一次性输出完整报告要稳定得多。4.2 协作流程我用文字描述一下标准流程代码实现会在下一章给出用户提交一个研究主题比如“梳理 2024 年开源多智能体框架的演进趋势”。编排器先把主题拆成 3 到 5 个子任务例如“梳理主流框架的发布时间和定位”“对比框架的编排方式差异”“分析社区生态活跃度”。多个研究员 Agent 并行执行子任务每个研究员只负责自己的部分输出结构化摘要和来源链接。写作专家拿到所有研究结果按照编排器给出的报告大纲组织内容生成初稿。审核专家检查初稿。如果发现信息缺失、逻辑断裂、引用缺失就输出修改意见。修订器根据审核意见修改或者把缺失部分重新派发给研究员补充。最终报告输出给用户。这个流程里最关键的设计是让每个 Agent 只做一件事。研究员不要写报告写作专家不要检索资料审核专家不要同时修改内容。职责越单一输出质量越稳定排错也越容易。5. 代码实战从单 Agent 到多智能体协作5.1 先实现一个简单的模型调用模块写多智能体之前先确保单次模型调用是通的。创建llm.py# llm.py from openai import OpenAI import config client OpenAI( api_keyconfig.OPENAI_API_KEY, base_urlconfig.OPENAI_API_BASE, ) def chat(messages, temperature0.3, max_tokens4096): 通用的模型调用函数。 messages: [{role: system, content: ...}, {role: user, content: ...}] response client.chat.completions.create( modelconfig.MODEL_NAME, messagesmessages, temperaturetemperature, max_tokensmax_tokens, ) return response.choices[0].message.content先用这个最简函数验证一下链路是否通# test_llm.py from llm import chat result chat([ {role: system, content: 你是一个测试助手只回复一句话。}, {role: user, content: 你好请回复多智能体链路正常。} ]) print(result)python test_llm.py能正常输出内容说明模型接口配置没问题。这一步通过后再往上叠多智能体结构。5.2 定义 Agent 基类多智能体本质上就是多个带独立系统提示词的对话循环。为了方便管理先定义一个基础 Agent 类# agent_base.py from typing import List, Dict, Optional from llm import chat class BaseAgent: 所有 Agent 的基类维护自己的系统提示词和上下文窗口 def __init__(self, name: str, system_prompt: str, temperature: float 0.3): self.name name self.system_prompt system_prompt self.temperature temperature self.messages: List[Dict[str, str]] [ {role: system, content: system_prompt} ] def run(self, task: str) - str: 执行一次任务任务内容会追加到上下文中 self.messages.append({role: user, content: task}) result chat(self.messages, temperatureself.temperature) self.messages.append({role: assistant, content: result}) return result def reset(self): 清空上下文只保留系统提示词 self.messages [{role: system, content: self.system_prompt}]这个基类的优点是简单直观。每个 Agent 实例维护自己的上下文互不干扰。如果你想实现更精细的记忆管理可以在此基础上扩展比如定期裁剪历史消息、只保留最近 N 轮对话、把长期记忆存到向量数据库等。5.3 编排器任务分解与规划编排器是多智能体系统的第一个环节。它不负责具体检索只负责把大问题拆解成可执行的子任务。# orchestrator.py import json from agent_base import BaseAgent ORCHESTRATOR_PROMPT 你是一个深度研究任务的编排器。你的职责是 1. 分析用户给出的研究主题 2. 将其拆解为 3 到 5 个彼此独立、可并行执行的子任务 3. 为每个子任务指定一个研究员角色和检索方向 严格按以下 JSON 格式输出不要输出其他内容 { topic: 研究主题, report_outline: [章节1, 章节2, 章节3], subtasks: [ { id: 1, role: researcher, direction: 检索方向说明, question: 子任务要回答的具体问题 } ] } class Orchestrator(BaseAgent): def __init__(self): super().__init__(orchestrator, ORCHESTRATOR_PROMPT, temperature0.1) def plan(self, topic: str) - dict: raw self.run(topic) # 提取 JSON 部分 start raw.find({) end raw.rfind(}) 1 if start -1 or end 0: raise ValueError(编排器输出不包含有效 JSON) data json.loads(raw[start:end]) return data这里的关键点是使用 JSON 作为中间传输格式。多智能体系统里Agent 之间的通信必须结构化不能直接传自然语言文本否则下游 Agent 解析起来会很痛苦。编排器的输出结构直接决定后续研究员能不能并行执行。5.4 研究员 Agent检索与摘要接下来是研究员。研究员的核心职责是接收一个子任务执行检索把检索结果整理成结构化摘要。# researcher.py import json from agent_base import BaseAgent RESEARCHER_PROMPT 你是一个深度研究员。你的工作流程是 1. 检索与任务相关的资料 2. 提炼关键信息整理成结构化摘要 3. 输出必须包含信息来源 如果没有真实检索工具请基于你训练数据中的知识进行合理整理 但必须在输出中注明基于模型知识整理建议人工核对。 输出格式 { subtask_id: 1, summary: 300字以内的研究摘要, key_points: [要点1, 要点2, 要点3], sources: [来源1, 来源2], confidence: high/medium/low, warning: 如有不确定性或风险在此说明 } class Researcher(BaseAgent): def __init__(self, name: str): super().__init__(name, RESEARCHER_PROMPT, temperature0.3) def research(self, subtask_id: int, question: str) - dict: task f子任务 {subtask_id}: {question} raw self.run(task) start raw.find({) end raw.rfind(}) 1 if start -1 or end 0: raise ValueError(f研究员 {self.name} 输出不包含有效 JSON) data json.loads(raw[start:end]) data[subtask_id] subtask_id return data实际工程中研究员这个角色通常要接入真实的检索工具比如搜索引擎 API、内部知识库、数据库、爬虫等。你可以在RESEARCHER_PROMPT里描述这些工具的使用方式或者在上面的research方法里先调用检索工具再把结果作为上下文传给模型。5.5 写作专家 Agent生成报告多个研究员并行完成子任务后写作专家需要把所有结果汇总成一篇完整的报告。# writer.py import json from agent_base import BaseAgent WRITER_PROMPT 你是一个深度报告写作专家。你的职责是 1. 根据研究计划和多路研究结果生成结构清晰的长文报告 2. 报告必须包含标题、目录、正文、来源引用 3. 所有关键数据必须基于研究员提供的结果不得凭空编造 4. 如果研究结果中存在冲突要明确指出并以最可信来源为准 输出格式直接输出 Markdown 格式报告不需要 JSON 包装。 报告开头需要包含一个信息来源章节列出所有引用过的来源。 class Writer(BaseAgent): def __init__(self): super().__init__(writer, WRITER_PROMPT, temperature0.4) def write_report(self, plan: dict, research_results: list) - str: context f 研究主题: {plan[topic]} 报告大纲: {json.dumps(plan[report_outline], ensure_asciiFalse)} 研究结果如下: {json.dumps(research_results, ensure_asciiFalse, indent2)} return self.run(context)写作专家不需要输出 JSON直接输出 Markdown 报告。因为报告是最终交付物适合给人阅读的格式。5.6 审核专家与修订器质量闭环审核专家检查报告修订器负责返工。这里用两段式实现一个输出审核意见一个根据意见修改。# reviewer.py import json from agent_base import BaseAgent REVIEWER_PROMPT 你是一个严格的质量审核专家。检查以下维度 1. 事实准确性是否有明显的错误或未经验证的断言 2. 结构完整性是否覆盖研究计划的全部章节 3. 引用规范关键数据和结论是否有来源支撑 4. 逻辑连贯章节之间是否有清晰的逻辑关系 5. 格式要求是否为规范的 Markdown 输出 JSON 格式 { passed: true/false, score: 0-100, issues: [ {type: fact|structure|reference|logic|format, description: 问题描述, suggestion: 修改建议} ] } class Reviewer(BaseAgent): def __init__(self): super().__init__(reviewer, REVIEWER_PROMPT, temperature0.1) def review(self, report: str) - dict: raw self.run(report) start raw.find({) end raw.rfind(}) 1 if start -1 or end 0: raise ValueError(审核专家输出不包含有效 JSON) return json.loads(raw[start:end])修订器逻辑如下# revisor.py import json from agent_base import BaseAgent REVISOR_PROMPT 你是一个报告修订专家。根据审核意见修改报告。 要求 1. 只修改审核意见中指出的问题 2. 保持原有报告的整体结构和风格 3. 对不确定的信息用据调研或待核实等措辞标注 直接输出修订后的完整 Markdown 报告不要输出其他内容。 class Revisor(BaseAgent): def __init__(self): super().__init__(revisor, REVISOR_PROMPT, temperature0.4) def revise(self, report: str, review_result: dict) - str: task f 报告草稿如下 {report} 审核意见如下 {json.dumps(review_result, ensure_asciiFalse, indent2)} 请根据审核意见进行修订输出完整报告。 return self.run(task)6. 并行任务调度与结果聚合6.1 用 asyncio 实现并行研究多个研究员 Agent 之间没有依赖关系可以并行执行。Python 的asyncio.gather是这里最直接的方案。# deep_research.py import asyncio from orchestrator import Orchestrator from researcher import Researcher from writer import Writer from reviewer import Reviewer from revisor import Revisor class DeepResearchSystem: def __init__(self, researcher_count: int 3, max_review_rounds: int 2): self.orchestrator Orchestrator() self.researchers [Researcher(fresearcher-{i}) for i in range(researcher_count)] self.writer Writer() self.reviewer Reviewer() self.revisor Revisor() self.max_review_rounds max_review_rounds async def _run_research_task(self, subtask: dict): 单个研究员执行一个子任务 idx subtask[id] # 按 id 取模分配研究员保证每个研究员在同一轮只处理一个任务 researcher self.researchers[idx % len(self.researchers)] return await asyncio.to_thread(researcher.research, idx, subtask[question]) async def research(self, topic: str) - dict: # 1. 编排器规划 plan self.orchestrator.plan(topic) print(f[Orchestrator] 研究主题: {plan[topic]}) print(f[Orchestrator] 拆分为 {len(plan[subtasks])} 个子任务) # 2. 并行研究 research_tasks [ self._run_research_task(subtask) for subtask in plan[subtasks] ] research_results await asyncio.gather(*research_tasks, return_exceptionsTrue) # 3. 处理异常结果 success_results [] for idx, result in enumerate(research_results): if isinstance(result, Exception): print(f[Researcher] 子任务 {idx 1} 执行失败: {result}) continue success_results.append(result) if not success_results: raise RuntimeError(所有研究任务均执行失败) # 4. 写作专家生成报告 print([Writer] 正在生成报告草稿...) draft self.writer.write_report(plan, success_results) print(f[Writer] 草稿长度: {len(draft)} 字符) # 5. 审核与修订循环 report draft for round_idx in range(self.max_review_rounds): print(f[Reviewer] 第 {round_idx 1} 轮审核...) review_result self.reviewer.review(report) print(f[Reviewer] 得分: {review_result.get(score, N/A)} f通过: {review_result.get(passed, False)}) if review_result.get(passed, False): print([Reviewer] 审核通过无需修订) break print(f[Revisor] 根据审核意见修订中问题数: f{len(review_result.get(issues, []))}) report self.revisor.revise(report, review_result) else: print([System] 达到最大修订轮数使用最后一次修订结果) return { topic: topic, plan: plan, research_results: success_results, report: report, }主函数入口# main.py import asyncio from deep_research import DeepResearchSystem async def main(): system DeepResearchSystem(researcher_count3, max_review_rounds2) result await system.research(梳理开源多智能体框架的核心设计模式) print(\n 最终报告摘要 ) print(result[report][:2000]) if __name__ __main__: asyncio.run(main())python main.py如果你的机器是 Windows且 Python 版本较旧遇到事件循环相关的问题可以先升级 Python 到 3.10 以上或者检查是否安装了winloop等兼容库。在大多数情况下直接运行不会有大问题。6.2 任务队列与批量任务如果一次要处理几十个研究主题直接全部并发会打爆模型接口限流也不方便管理失败重试。更好的做法是引入一个任务队列用固定数量的 worker 消费队列。# batch_runner.py import asyncio from deep_research import DeepResearchSystem async def process_topic(semaphore, system, topic): async with semaphore: print(f[Batch] 开始处理: {topic}) try: result await system.research(topic) return {topic: topic, status: success, report: result[report]} except Exception as e: print(f[Batch] 处理失败: {topic}, 错误: {e}) return {topic: topic, status: failed, error: str(e)} async def run_batch(topics, max_concurrency2, researcher_count3): semaphore asyncio.Semaphore(max_concurrency) system DeepResearchSystem(researcher_countresearcher_count) tasks [process_topic(semaphore, system, topic) for topic in topics] results await asyncio.gather(*tasks) return results if __name__ __main__: topics [ 开源 RAG 框架的技术选型对比, 2024 年多智能体框架的发展趋势, 企业级 Agent 应用的落地难点分析, ] results asyncio.run(run_batch(topics, max_concurrency2)) for r in results: print(f{r[topic]} - {r[status]})批量任务里最关键的设计是max_concurrency。这个参数控制同时运行的调研任务数量避免把 API 限流打爆。建议从 2 到 3 开始根据实际响应时间和 API 配额慢慢调大。如果你需要更强的批量任务管理能力可以把任务状态持久化到数据库加上失败重试、进度上报、结果落盘等逻辑。生产环境建议使用 Celery、Arq 或 Temporal 这类成熟任务队列而不是纯 asyncio。7. 接口 API 封装与外部调用多智能体系统的价值在于可以被外部系统调用。用 FastAPI 把DeepResearchSystem封装成 HTTP 服务是最常见的方式。7.1 FastAPI 服务封装# api.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from typing import Optional import asyncio from deep_research import DeepResearchSystem app FastAPI(titleDeep Research Agent API) class ResearchRequest(BaseModel): topic: str Field(..., description研究主题) researcher_count: int Field(3, ge1, le10, description研究员 Agent 数量) max_review_rounds: int Field(2, ge0, le5, description最大审核轮数) class ResearchResponse(BaseModel): topic: str status: str report: Optional[str] None error: Optional[str] None app.get(/health) async def health(): return {status: ok} app.post(/api/research, response_modelResearchResponse) async def research(req: ResearchRequest): try: system DeepResearchSystem( researcher_countreq.researcher_count, max_review_roundsreq.max_review_rounds, ) loop asyncio.get_event_loop() result await loop.run_in_executor( None, lambda: asyncio.run(system.research(req.topic)), ) return ResearchResponse( topicreq.topic, statussuccess, reportresult[report], ) except Exception as e: return ResearchResponse( topicreq.topic, statusfailed, errorstr(e), )启动服务uvicorn api:app --host 127.0.0.1 --port 8000这里要注意asyncio.run和 FastAPI 的事件循环不能直接混用。上面的写法通过run_in_executor把异步任务放到线程池里执行避免了事件循环冲突。如果你的模型调用本身是同步阻塞的这个方案是安全且简单的。7.2 调用测试服务启动后用 curl 验证curl -X POST http://127.0.0.1:8000/api/research \ -H Content-Type: application/json \ -d {topic: 多智能体系统在金融风控中的应用, researcher_count: 3}用 Python 调用import requests url http://127.0.0.1:8000/api/research payload { topic: 开源 Agent 开发框架对比, researcher_count: 3, max_review_rounds: 2, } response requests.post(url, jsonpayload, timeout600) data response.json() print(data[status]) print(data[report][:500] if data[report] else data[error])接口设计上有几个注意事项第一超时时间要设长。Deep-Research 任务涉及多轮模型调用和并行检索耗时通常在几十秒到几分钟不等。前端调用时把超时设置到 5 分钟以上或者直接用异步任务 轮询结果的方式。第二如果服务要部署到公网必须加鉴权。最简单的方案是在中间件里校验Authorization请求头或者使用 API Key。不要裸奔没有认证的服务。第三针对长任务更好的设计是“提交任务-返回任务 ID-轮询结果”两步走。这样即使客户端断线任务也不会丢失。8. 成本控制与性能观察8.1 多智能体系统的 Token 消耗多智能体系统看起来只是调了好几次模型实际上 Token 消耗会比单次问答高出一个数量级。原因是每个 Agent 都有自己的系统提示词和上下文研究员之间互不共享上下文导致信息重复传递。以一次标准研究任务为例环节人/次数单次 Token 估算小计编排器规划1 次500-1000约 1000研究员并行调研3 个子任务每个 1500-3000约 9000写作专家生成1 次2000-5000约 5000审核专家1 次2000-4000约 4000修订器1 次2000-4000约 4000合计--约 23000 Token如果你的模型是gpt-4o-mini这类低成本模型这个量级还可以接受。但如果换成大杯模型一次研究任务的成本就会明显上升。在做预算之前先用一个小主题跑通全流程量一下实际 Token。8.2 如何降低 Token 消耗降低 Token 消耗有几个实用手段。第一精简系统提示词。每个 Agent 的系统提示词不是越长越好。把核心指令控制在 200 字以内能显著减少每轮调用的基础开销。第二限制研究结果的长度。研究员输出的summary字段限制在 300 字以内key_points限制在 5 条以内。结构化摘要比原文引用省 Token 得多。第三审核不通过时才触发修订。max_review_rounds设成 0 可以让系统只生成不审核适合低成本快速产出场景。设成 1 到 2 是质量和成本之间的平衡点。第四做结果缓存。相同或相似的研究主题第一次研究完成后把报告和中间结果缓存到本地。第二次直接读取缓存不再调用模型。import hashlib import json import os CACHE_DIR ./cache os.makedirs(CACHE_DIR, exist_okTrue) def get_cache_key(topic: str) - str: return hashlib.md5(topic.encode(utf-8)).hexdigest() def load_cache(topic: str) - dict | None: key get_cache_key(topic) path os.path.join(CACHE_DIR, f{key}.json) if os.path.exists(path): with open(path, r, encodingutf-8) as f: return json.load(f) return None def save_cache(topic: str, result: dict): key get_cache_key(topic) path os.path.join(CACHE_DIR, f{key}.json) with open(path, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2)8.3 性能观察要点观察多智能体系统性能重点看三个指标。第一个是模型 API 的响应延迟。总耗时的绝大部分都花在模型推理上。你可以把每次调用的耗时、Token 数、模型名记录到日志里。第二个是任务队列积压情况。批量任务跑起来后观察队列里等待的任务数量。如果每天都在积压需要调大并发度或减少任务量。第三个是失败重试率。模型 API 偶发超时、限流、网络抖动都很正常。如果失败率超过 5%优先检查base_url配置、网络连通性和 API 配额而不是代码逻辑。9. 常见问题与排查方法问题现象可能原因排查方式解决方案模型调用报 401/403API Key 错误或过期检查.env文件中的 key 和前导空格重新生成 key确认配置读取正确模型调用报 404base_url 路径不对检查 OpenAI 兼容服务的路径是否包含/v1在 base_url 后补上/v1路径编排器输出 JSON 解析失败模型输出格式不符合要求打印原始输出检查是否包含额外文本在提示词中增强格式约束或使用response_format参数研究员任务超时子任务太复杂或模型响应过慢查看单次任务耗时日志拆细子任务减少单次上下文长度审核结果质量差模型能力不足或审核维度不清晰检查审核 Prompt 是否明确逐条列出审核维度或换更强模型批量任务卡住并发过高触发 API 限流检查 API 配额和错误日志降低 max_concurrency增加退避重试报告内容重复多个研究员输出了相似结果检查子任务是否交叉重叠在编排器 Prompt 中明确子任务边界引用缺失或虚假研究员没有真实检索工具查看 sources 字段是否为空接入真实检索 API或至少让模型标注不确定性服务启动端口被占用8000 端口被其他进程占用用lsof -i:8000或netstat -ano查看更换端口uvicorn api:app --port 8001请求超时任务耗时太长客户端提前断开在客户端设置较长的超时时间改用异步任务 轮询模式排查逻辑其实有规律先看模型接口通不通再看中间 JSON 解析是否成功最后看业务逻辑是否正确。多智能体系统比单 Agent 多了一层“多角色协作”的复杂度所以日志特别重要。每个 Agent 的关键输入输出都要打好日志否则出了问题根本不知道是哪个环节出错。10. 最佳实践与后续扩展建议去视频化的技术文通常会在最后给一点工程化沉淀。这里把实战中比较有用的经验列出来。10.1 从最小闭环开始第一次写多智能体系统不要一上来就接工具、接数据库、接复杂编排。先做一个最小闭环编排器 → 研究员 → 写作专家 → 输出。跑通之后再逐步加入审核、修订、并行调度、API 封装。每一步都验证过再往下走排错成本会低很多。10.2 结构化中间结果是生命线多智能体系统里Agent 之间传递的中间结果必须结构化。用 JSON 定义每个 Agent 的输出格式比让 Agent 自由发挥要稳定得多。我建议给每个输出格式写一个 Pydantic 模型用model_validate_json做校验解析失败就重试一次而不是直接抛异常。from pydantic import BaseModel, Field from typing import List class Subtask(BaseModel): id: int role: str direction: str question: str class Plan(BaseModel): topic: str report_outline: List[str] subtasks: List[Subtask]10.3 接入真实工具链上面演示的研究员 Agent 用的是模型知识没有接入真实检索工具。生产环境要接入搜索引擎 API、Firecrawl、内部知识库或者数据库查询。你可以在研究员的research方法里先调用检索函数再把检索结果注入上下文让模型基于检索结果生成摘要。10.4 换用 LangGraph 做复杂控制流如果你的任务不是简单的“规划-并行-汇总-审核”线性流程而是有条件分支、循环嵌套、人工审批穿插的复杂流程建议用 LangGraph。它把多 Agent 协作建模成一张图节点是 Agent 或工具边是控制流逻辑可观测性和可控性都比自研循环好。10.5 测试时先小后大跑批量任务之前先用一个主题、一个研究员、不加审核跑一遍。确认输出格式、响应时间、Token 消耗都符合预期之后再逐步加大并发和任务数量。不要一上来就 50 个主题并发那不是提效是给自己找麻烦。11. 总结与下一步Deep-Research 多智能体开发的核心不是“调用多个模型”而是把复杂任务拆成多个职责单一的 Agent用结构化中间结果串联协作流程通过并行调度提效通过审核修订保证质量。这套方法不绑定具体框架也不绑定具体模型换模型、换工具、换场景都能复用。建议你上手后最先验证三件事第一编排器能否稳定输出合法 JSON 的子任务列表第二多个研究员并行后能否正确聚合结果第三审核-修订循环能否真正提升报告质量。这三步跑通了整个系统就立住了。最容易踩的坑是过度设计。很多初学者一上来就搭复杂的 Agent 通信协议、消息总线、记忆系统结果跑都跑不起来。先把一个最简单的五角色流程跑通再逐步加复杂度。后续可以继续扩展的方向包括接入真实搜索引擎 API、给研究员 Agent 挂上 RAG 工具、把报告输出接成带数据可视化的网页、加任务队列做成异步批处理平台。每一步扩展都是单独一篇实战文章的量。建议收藏备用动手写的时候对照着搭。