如果说最近 AI 应用里哪个方向最值得开发者关注Deep-Research 类型的产品一定排在前列。用户输入一个问题系统能自动拆解任务、搜索资料、交叉验证、输出带引用的研究报告这类体验已经超出了普通聊天机器人的范畴。但如果你想自己动手做一个类似的系统很快会发现一个残酷的事实真正难的不是调用大模型 API而是“多智能体”这套工程体系怎么搭。Deep-Research 表面上是模型能力强背后其实是把复杂的调研任务拆成搜索、分析、综述、校验等多个环节再交给不同的 Agent 去协作完成。这篇文章就按照从零上手的实战思路带你拆解 Deep-Research 多智能体开发的完整链路。看完后你能掌握架构设计、核心概念、代码实现、运行验证和常见坑点并在此基础上搭建属于自己的深度研究型 Agent。1. Deep-Research 多智能体开发到底在解决什么问题先问一个问题为什么单 Agent 做不好深度研究如果你让一个传统聊天机器人写一份“2025 年主流 AI Agent 框架对比”的报告通常会得到一份看起来很完整、但无法验证的宏观回答。它不会主动去搜索最新信息不会对多个来源做交叉验证更不会告诉你哪些结论来自哪个文档。这不是模型不够聪明而是任务形态不匹配。深度研究类任务有三个特点信息源是动态的模型训练数据里没有最新内容必须实时检索。结论需要可追溯不能只给结论还要给依据和出处。过程是分阶段的先收集素材再分析提炼最后汇总成文中间还有验证迭代。这三个特点决定了“一个 Agent 包办所有事”的结构性不足。你当然可以在一个 Agent 里把所有逻辑写进去但很快会遇到上下文过长、工具调用混乱、阶段性结果不可复用、错误发生后难以定位等问题。多智能体的思路是把“一个人干所有事”改成“一个项目组协作”。Coordinator 负责拆解任务和调度Search Agent 负责检索Analyst Agent 负责分析和信息抽取Writer Agent 负责最终报告。每个 Agent 职责单一、输入输出明确、可独立测试出了问题也能快速定位。这才是 Deep-Research 类应用真正的技术门槛所在不是模型能力而是多智能体的组织和工程落地。2. 核心概念Deep-Research、Agent、多智能体与 MCP在进入代码之前先把几个高频概念理清楚。很多初学者被这些词绕晕其实它们对应的是不同层面的东西。2.1 Deep-Research 是一种应用形态Deep-Research 可以理解为“深度研究型 Agent 应用”。它的核心流程是用户提出一个研究问题。系统把问题拆解成若干子任务。子任务分发给不同 Agent每个 Agent 可能调用外部工具。结果汇聚后经过验证和整理输出一份带引用的报告。它不是某个模型的名字而是一类应用的统称。只要符合“自动拆解 多阶段处理 可追溯输出”这几点都可以叫 Deep-Research 应用。2.2 Agent有工具调用能力的任务执行单元Agent 在本文语境下是“具备大模型推理能力、并能调用外部工具完成特定任务的程序单元”。它和普通 API 调用的区别在于普通 API 调用输入问题返回文本。Agent输入目标自己决定用什么工具、按什么顺序做、如何基于工具返回结果继续推理。Deep-Research 里的 Search Agent 就是典型例子。它不只是“搜索一下”而是先判断需要哪些检索词再调用搜索工具最后对搜索结果做初步筛选。2.3 多智能体一套任务编排体系多智能体是把多个 Agent 组织起来让它们围绕同一个目标协作。这里要记住一件事多智能体不等于多个大模型调用而是多个“有角色分工的 Agent 实例”的协作关系。常见交互模式有四种模式特点适用场景主从模式Leader 拆任务Worker 执行适合 Deep-ResearchCoordinator 主导一切流水线模式前一个 Agent 输出是后一个输入搜索 → 分析 → 写作阶段清晰对等协作模式多个 Agent 平等讨论互相补充多角度头脑风暴、观点合成竞争评审模式多个 Agent 生成结果评审者打分结果优化、质量评估、方案选优大部分 Deep-Research 系统是“主从模式 流水线模式”的混合体。Coordinator 作为 Leader把任务拆解后按流水线的方式派发给 Worker最终汇总输出。2.4 MCPAgent 接入工具的标准协议你会在多智能体开发中频繁看到 MCP 这个词。MCP 全称 Model Context Protocol解决的问题是Agent 怎么标准化地访问外部工具和数据源。可以把它理解成 Agent 世界的 USB-C 接口。没有统一协议时每个 Agent 接入搜索服务、网页抓取、数据库都要自己写一套工具调用逻辑换一个 Agent 就要重新适配。有了 MCP工具提供方实现一个 MCP ServerAgent 通过统一协议调用即可。在 Deep-Research 场景里搜索、网页抓取、知识库查询都很适合封装成 MCP Server。后面代码示例中我会用函数模拟工具调用但在生产环境更推荐通过 MCP 协议统一管理工具层。3. 系统架构、交互模式与基础环境准备3.1 一个最小 Deep-Research 多智能体架构为了不过度设计本文示例采用“1 个 Coordinator 3 个 Worker Agent”的架构用户提问 | v Coordinator(协调者) ----- 任务拆解、流程调度、结果汇总 | | | v v v Search Analyst Writer Agent Agent AgentSearch Agent接收查询目标生成检索词调用搜索工具返回原始素材和来源链接。Analyst Agent接收原始素材提炼关键信息剔除噪声输出结构化分析结果。Writer Agent接收分析结果组织语言输出带引用的完整研究报告。Coordinator 是大脑但它不直接执行搜索和写作只负责任务拆解、状态管理和结果聚合。3.2 技术选型与版本说明本文示例使用 Python 3.10依赖以下核心库openai调用兼容 OpenAI 规范的大模型接口。httpx发起异步 HTTP 请求模拟搜索和网页抓取。pyyaml读取 Agent 配置文件。python-dotenv管理 API Key 等环境变量。注意示例代码中的模型配置以 DeepSeek 这类兼容 OpenAI 规范的模型服务为参考实际接入时请以模型服务商官方文档为准。版本号不写死安装时使用当前稳定版本即可。3.3 环境初始化建议使用虚拟环境隔离依赖mkdir deep-research-agent cd deep-research-agent python -m venv .venv source .venv/bin/activate创建依赖文件requirements.txtopenai1.30.0 httpx0.27.0 pyyaml6.0.1 python-dotenv1.0.1安装依赖pip install -r requirements.txt创建.env文件存放密钥# 模型服务商 API Key LLM_API_KEYyour-api-key # 搜索 API Key按需配置 SEARCH_API_KEY提醒一句.env文件不要提交到 Git 仓库在.gitignore中加上它。4. 核心流程拆解从提问到研究报告在写代码前先把流程拆清楚。Deep-Research 系统通常包含四个阶段。4.1 任务理解与分解Coordinator 拿到用户问题后第一件事不是搜索而是理解问题结构。示例中我们让 Coordinator 调用模型生成研究目标需要检索的关键词列表需要回答的子问题。这一步的价值在于把用户一句模糊的问题变成后续 Agent 可以执行的明确任务。4.2 搜索与素材收集Search Agent 拿到检索词后有两个任务对每个检索词调用搜索工具获取搜索结果页 URL 和摘要。对重要页面抓取正文内容做初步截断。这里要注意不要把所有网页原文直接塞给后续 Agent。上下文长度有限原始网页包含大量导航、广告等噪声需要先做内容抽取和截断。4.3 分析与信息抽取Analyst Agent 接收 Search Agent 传回的素材输出结构化信息每条素材的核心观点关键数据指标可信度评估来源链接。结构化输出非常重要。如果 Analyst Agent 返回的是大段口语化文本Writer Agent 根本无法可靠引用。4.4 报告生成与验证Writer Agent 基于 Analyst Agent 的结构化输出撰写研究报告。报告要求每条主要结论都标注来源链接。Coordinator 在这一阶段还要做一次校验检查报告中是否有结论缺少引用来源如果有回到对应阶段补充证据。5. 完整代码实现一个可运行的 Deep-Research 多智能体现在进入正题。下面的代码是完整可运行的并且支持--fake参数在没有真实 API Key 的情况下也能跑通整个多智能体流程。5.1 配置文件先创建config.yamlcoordinator: model: deepseek-chat max_rounds: 3 agents: search: name: 搜索Agent model: deepseek-chat max_concurrency: 2 analyst: name: 分析Agent model: deepseek-chat writer: name: 综述Agent model: deepseek-chat三个 Agent 使用同一个模型但角色提示词和任务不同。真实项目里可以给不同 Agent 配置不同模型例如分析用更强模型搜索用更快的模型。5.2 主程序代码创建main.pyimport asyncio import json import logging import sys from dataclasses import dataclass, field from typing import Optional import httpx import yaml from dotenv import load_dotenv from openai import AsyncOpenAI load_dotenv() logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s - %(message)s, ) logger logging.getLogger(deep-research-agent) # ---------- 工具层 ---------- async def search_web(httpx_client: httpx.AsyncClient, query: str) - list[dict]: 模拟搜索工具。 真实项目中可替换为 SerpAPI、Bing Search API 或 MCP 协议的搜索工具。 这里返回两条模拟结果保证流程可以独立跑通。 logger.info(f[Tool] 搜索关键词: {query}) return [ { title: f{query} - 资料A, url: fhttps://example.com/a?q{query}, snippet: f这是关于 {query} 的第一条摘要信息来自资料A。, }, { title: f{query} - 资料B, url: fhttps://example.com/b?q{query}, snippet: f这是关于 {query} 的第二条摘要信息来自资料B。, }, ] async def fetch_page(url: str) - str: 模拟网页抓取。 真实项目中需要处理 robots 协议、反爬策略、HTML 正文抽取。 await asyncio.sleep(0.1) return f这是 {url} 的正文内容包含与问题相关的若干关键段落。 # ---------- LLM 客户端 ---------- class LLMClient: def __init__(self, model: str, fake: bool False): self.model model self.fake fake self.client AsyncOpenAI() async def chat(self, system: str, user: str) - str: if self.fake: # 无 API Key 时返回模拟结果便于验证多智能体流程 return f[模拟输出] 模型 {self.model} 已完成任务。 resp await self.client.chat.completions.create( modelself.model, messages[ {role: system, content: system}, {role: user, content: user}, ], temperature0.3, ) return resp.choices[0].message.content or # ---------- Agent 定义 ---------- dataclass class AgentResult: agent_name: str task_name: str output: str meta: dict field(default_factorydict) class ResearchAgent: 通用研究型 Agent 基类子类实现 run 方法。 def __init__(self, name: str, role: str, llm: LLMClient): self.name name self.role role self.llm llm async def run(self, task: str) - AgentResult: raise NotImplementedError def _sys_prompt(self) - str: return self.role class SearchAgent(ResearchAgent): 搜索 Agent将任务转化为检索词调用搜索工具返回素材列表。 async def run(self, task: str) - AgentResult: logger.info(f[{self.name}] 开始执行搜索任务) async with httpx.AsyncClient() as client: raw_results await search_web(client, task) collected [] for item in raw_results[:3]: page await fetch_page(item[url]) collected.append({ title: item[title], url: item[url], snippet: item[snippet], content: page[:500], }) output json.dumps(collected, ensure_asciiFalse, indent2) return AgentResult( agent_nameself.name, task_namesearch, outputoutput, meta{source_count: len(collected)}, ) class AnalystAgent(ResearchAgent): 分析 Agent从素材中提炼关键信息输出结构化分析结果。 async def run(self, task: str) - AgentResult: logger.info(f[{self.name}] 开始分析素材) system ( 你是一名资深信息分析专家。请从素材中提炼核心观点、关键数据和结论 标注每条信息对应的来源URL。如果素材不足请明确说明信息缺口。 输出格式为 JSON。 ) prompt f素材如下\n{task} result_text await self.llm.chat(system, prompt) # 防止模型直接返回 JSON 代码块做一次清理 cleaned result_text.strip() if cleaned.startswith(): cleaned cleaned.strip() if cleaned.startswith(json): cleaned cleaned[4:].strip() return AgentResult( agent_nameself.name, task_nameanalyze, outputcleaned, meta{format: json}, ) class WriterAgent(ResearchAgent): 综述 Agent将分析结果整合为带引用的研究报告。 async def run(self, task: str) - AgentResult: logger.info(f[{self.name}] 开始生成报告) system ( 你是一名专业报告撰写者。请基于分析结果撰写结构化研究报告。 要求开头给出核心结论正文分点展开每条结论必须标注引用来源URL 结尾列出参考资料清单。 ) prompt f分析结果如下\n{task} result_text await self.llm.chat(system, prompt) return AgentResult( agent_nameself.name, task_namewrite, outputresult_text, meta{format: markdown}, ) # ---------- Coordinator ---------- class Coordinator: 协调者负责任务分解、Agent 调度和结果汇总。 def __init__(self, config: dict, fake: bool False): self.config config self.model config[coordinator][model] self.llm LLMClient(self.model, fakefake) self.agents { search: SearchAgent( nameconfig[agents][search][name], role你是一个搜索专家负责检索资料。, llmLLMClient(config[agents][search][model], fakefake), ), analyst: AnalystAgent( nameconfig[agents][analyst][name], role你是一个信息分析专家。, llmLLMClient(config[agents][analyst][model], fakefake), ), writer: WriterAgent( nameconfig[agents][writer][name], role你是一个研究报告撰写专家。, llmLLMClient(config[agents][writer][model], fakefake), ), } self.trace: list[dict] [] async def run(self, user_query: str) - dict: logger.info( 开始执行 Deep-Research 任务 ) logger.info(f用户问题: {user_query}) # 阶段1任务分解 task_id task_ str(hash(user_query) 0xffff) self.trace.append({phase: init, query: user_query, task_id: task_id}) # 阶段2搜索 search_result await self.agents[search].run(user_query) self.trace.append({ phase: search, agent: search_result.agent_name, output_preview: search_result.output[:200], }) # 阶段3分析 analyst_result await self.agents[analyst].run(search_result.output) self.trace.append({ phase: analyze, agent: analyst_result.agent_name, output_preview: analyst_result.output[:200], }) # 阶段4写作 writer_result await self.agents[writer].run(analyst_result.output) self.trace.append({ phase: write, agent: writer_result.agent_name, output_preview: writer_result.output[:200], }) logger.info( Deep-Research 任务执行完成 ) return { task_id: task_id, query: user_query, report: writer_result.output, trace: self.trace, } # ---------- 入口 ---------- async def main(): fake --fake in sys.argv with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) coordinator Coordinator(config, fakefake) query 多智能体系统的主流架构和关键技术 result await coordinator.run(query) print(\n 最终报告 \n) print(result[report]) print(\n 执行追踪 \n) for step in result[trace]: print(json.dumps(step, ensure_asciiFalse)) if __name__ __main__: asyncio.run(main())这段代码覆盖了 Deep-Research 多智能体的完整闭环Coordinator 拆解任务Search Agent 收集素材Analyst Agent 提炼分析Writer Agent 生成报告并且全程记录 trace 用于审计。如果不想申请 API Key直接运行python main.py --fake如果想真正调用大模型去掉--fake并确保.env中配置了LLM_API_KEYpython main.py6. 运行结果与效果验证6.1 预期输出使用--fake参数运行时Search Agent 会从模拟的搜索工具拿回两条结果Analyst Agent 和 Writer Agent 返回模拟文本。最终输出 最终报告 [模拟输出] 模型 deepseek-chat 已完成任务。 执行追踪 {phase: init, query: 多智能体系统的主流架构和关键技术, task_id: task_xxxx} {phase: search, agent: 搜索Agent, output_preview: ...} {phase: analyze, agent: 分析Agent, output_preview: ...} {phase: write, agent: 综述Agent, output_preview: ...}这个结果说明整个多智能体调度链路已经跑通。注意--fake模式只验证流程不能作为真实研究工作使用。6.2 如何判断系统是否正常判断标准有三条日志中四个阶段start、search、analyze、write依次出现没有异常报错。trace中每个阶段都有非空输出。最终报告的task_id在所有阶段日志中保持一致。如果哪个阶段没有出现先看日志里最后一次正常输出的阶段定位中断位置。6.3 接入真实模型后的验证接入真实模型后建议随手做一次“引用可追溯”验证打开最终报告随机抽一条主要结论。检查这条结论是否带有来源 URL。访问该 URL确认内容与报告中的描述一致。这一步是 Deep-Research 系统区别于普通聊天机器人的关键。如果报告里的结论无法追溯说明分析或写作阶段的提示词还不够严格需要增加约束。7. 常见问题与排查方法问题现象可能原因排查方式解决方案请求被限流返回 429调用频率超过模型服务商限制查看日志中 HTTP 状态码和错误信息用asyncio.Semaphore控制并发增加指数退避重试搜索结果质量差检索词太抽象一搜全是不相关内容打印 Search Agent 实际使用的检索词增加 Query Rewriter先在 Coordinator 中生成多个具体检索词报告结论与引用来源不符Writer Agent 没有被约束必须基于引用写作检查写入报告的 Prompt强制 Writer Agent 输出每个观点后紧跟来源标记没有来源不得下结论网页正文太长超出上下文限制直接把完整网页原文传给后续 Agent查看传给 Analyst 的文本长度抓取后先做正文抽取和截断控制在 500 字以内API Key 泄漏到 Git 仓库.env没有加入.gitignore检查仓库历史中的敏感文件第一时间吊销并轮换 Key补充.gitignore多个 Agent 职责重叠互相“踢皮球”提示词里没有明确输入输出边界逐个检查 Agent 的系统提示词每个 Agent 只负责一个明确产出输出格式写成 JSON 结构任务执行时间过长串行执行搜索任务轮次过多查看 trace 中每个阶段的耗时搜索阶段用 asyncio.gather 并行合理设置 Coordinator 最大轮数8. 最佳实践与工程建议代码跑通只是第一步。真实项目里下面这些经验能让你少踩很多坑。8.1 一个 Agent 只做一件事Dissecting 多智能体最忌讳的是“一个 Agent 什么都能干”。Search Agent 只负责搜索和素材收集不要让它顺便写报告Writer Agent 只负责写作不要让它去搜索。职责边界越清晰系统越容易调试和维护。8.2 用结构化中间产物传递数据Search Agent 给 Analyst Agent 输出的是 JSONAnalyst Agent 给 Writer Agent 输出的也是 JSON。不要用大段口语化文本作为 Agent 之间的传递格式。这样做有三个好处后续 Agent 可以直接解析减少花在理解文本上的 token。中间产物的质量更容易评估。可以把中间结果落库供团队审计和优化。8.3 每个 Agent 都要有验证约束Deep-Research 最害怕的是“一本正经地胡说八道”。建议在提示词里给每个 Agent 增加硬性约束Search Agent必须返回 URL 和摘要不允许编造来源。Analyst Agent只分析素材中已有的信息素材不足时明确说明。Writer Agent每条结论必须引用来源 URL无来源的内容不得成文。8.4 给 Coordinator 设置预算多智能体系统容易陷入“无限迭代”。Coordinator 必须有预算意识最大搜索轮数默认 23 轮每次搜索最多返回结果数单个 Agent 的最大 token 消耗。在 Coordinator 里把预算写到配置文件中而不是写死在代码里方便灰度调整。8.5 工具层优先考虑 MCP 协议如果团队内部已经准备接入搜索、数据库、知识库等多个数据源强烈建议把工具层封装成 MCP Server。这样设计的价值在于Agent 通过统一协议调用工具不需要为每个数据源写一套封装。新增工具时只需要增加一个 MCP Server不修改 Agent 核心代码。不同 Agent 之间可以复用同一套工具目录避免重复建设。8.6 日志和追踪要带上任务 IDDeep-Research 系统里一个用户请求会触发多个 Agent、多次工具调用。排查问题时如果没有全局任务 ID所有日志会混在一起。示例代码中的task_id就是为了解决这个问题。生产环境中建议把每一条 Agent 日志、工具调用日志、模型调用日志都绑定task_id并用 JSON 结构化输出方便接入日志平台。8.7 先跑通最小系统再上编排框架不要一上来就堆 LangGraph、AutoGen、CrewAI 这类框架。先用本文这个最小实现理解多智能体的调用关系再逐步替换成框架。原因很简单框架解决的问题是“编排复杂度”但在没有理解多智能体基本流程之前框架带来的抽象反而会成为理解障碍。9. 总结与后续学习方向从零上手 Deep-Research 多智能体开发核心并不是学会某一个框架而是建立起“任务拆解 角色分派 工具调用 流程验证”的工程思维。本文通过一个最小系统把这条链路完整跑通了。现在你可以按照这个思路做三件事第一把代码中的模拟搜索工具替换成真实搜索 API 或 MCP Server接上真实模型跑一次真实的研究报告生成。第二给系统增加一个 Judge Agent让它在 Writer Agent 出稿后做一轮质量评审检查结论是否有引用、结构是否完整。这就是前面提到的“竞争评审模式”。第三把所有中间产物落库用真实用户请求积累数据优化每个 Agent 的提示词和工具调用策略。多智能体开发的边界非常宽但入口其实不复杂从一个问题开始拆成若干任务交给不同角色的 Agent 协作完成最后回到一个可验证的结论。把这套流程打磨好Deep-Research 类应用的雏形就已经出来了。