DeepResearch这个词前阵子突然就热起来了。用户给一句研究指令，比如“帮忙调研一下2024年主流向量数据库的选型差异”，它能在后台自动拆题、跑几十次搜索、读几十个网页，最后交出一份带引用来源、有条理的完整报告。说实话，第一次看到这个效果时我是有点被震到的，但作为写代码的人，更吸引我的不是这个工具本身，而是背后那套编排逻辑——它到底是怎么用代码把“搜索—阅读—归纳—写报告”这一长串动作串成一条自动化流水线的。
这篇文章不讨论某家产品的商业包装，直接进到代码层面，拆解一个典型DeepResearch系统的核心链路。我会用一个去掉品牌外壳的通用实现来讲解：状态怎么在设计好的节点之间流转，Planner怎么把大问题切成可搜索的子问题，搜索层如何做并发与去重，上下文怎么避免被网页撑爆，以及最终报告如何保证引用不丢、结论不胡编。整篇文章里的代码是一版可以跑通的最小骨架，适合想自己复刻一个简化版DeepResearch的开发者，也适合只想知道这类系统内部秘密的读者。
1. 先搞明白：DeepResearch到底把“深”字落在哪
很多人第一次用DeepResearch类工具，会把它跟“搜索 + ChatGPT总结”混为一谈。我自己一开始也是这么想的，直到动手拆代码才发现：它真正厉害的地方不是某一个模型的智力，而是一套围绕“研究”这个目标设计的Agent工作流。
1.1 传统搜资料和DeepResearch的本质区别
普通人自己查资料的流程大概是：打开搜索引擎，输入关键词，点开几篇看着靠谱的链接，浏览器里开十几个标签页，边读边往文档里复制粘贴，最后再手工整理成一段结论。这个过程有两个很致命的弱点：一是人的短期记忆有限，读到第5个页面时早忘了第1个页面说过什么；二是很少有人做到“交叉验证”，搜索结果第一页说什么就信什么。
DeepResearch在代码层面做的事情，本质上就是把上述流程程序化，而且比人做得更可控制。它不只是把搜索结果塞给大模型做“总结”，而是把一个大的研究主题拆开，每个子问题单独去搜索、单独阅读、单独形成结论，最后再由上一层模块把结论拼接成一篇完整报告。这个“先拆再合”的过程，是代码里最核心的架构思想。
1.2 多智能体状态流：从一句话到一份报告
如果用通俗的说法来概括多智能体状态流，那它就是一条生产线：原料是一条用户指令，经过拆解、检索、提炼、汇总四道工序，最后出厂成品是一篇结构化报告。每一道工序由一个独立模块负责，模块之间的信息通过一个共享的“状态对象”传递。
一个典型的DeepResearch流水线包含下面这些节点：
- 入口节点：接收用户意图，判断这个任务值不值得做深度研究。
- Planner（规划器）：把主题拆成若干个可搜索的子问题，并规划每轮搜索的query。
- Searcher（搜索器）：执行实际搜索，对每个query拿到一批网页结果。
- Processor（处理器）：抓取网页正文，清洗噪声，按子问题归档。
- Analyst（分析师）：针对每个子问题，结合已读材料生成阶段结论。
- Writer（报告器）：把所有子结论整合成最终报告，并挂上引用来源。
- Controller（控制器）：决定是否需要追加一轮“深挖搜索”，判断当前资料是否足够。
这里每个节点都是独立的函数或类，节点之间不直接调用彼此的方法，而是通过读State、写State来通信。这样设计的好处是：想增加新的研究策略时，只需要在Controller里加一个节点，不需要改动其他模块。
1.3 为什么普通RAG那套做法不够用
你可能用过RAG（检索增强生成）类的问答应用：查一个知识库，把最相关的几个片段取出来拼到Prompt里，然后让模型回答。DeepResearch不能用这个思路，核心原因在于研究对象通常是开放域问题，而不是封闭知识库。向量数据库里检索不到实时网页，即便接入了搜索引擎API，一次性把几十个搜索结果全塞进一个Prompt里，上下文窗口也撑不住。
更深层的问题是，RAG是“单轮检索—单轮生成”结构，而DeepResearch是“多轮计划—多轮检索—分层归纳”结构。RAG回答“我的文档里支持哪几种导出格式”很合适，但回答“对比三种方案在不同行业里的落地情况”就力不从心了——因为这个问题本身需要多次搜索不同维度的信息。DeepResearch的代码结构用“有状态的循环”替代了“单次查询”，这就是它显得更智能的最根本原因。
2. 一版能跑通的最小代码骨架：先搭状态流
看DeepResearch代码的第一步，不是去看某个Agent内部写了多少Prompt，而是先把状态流搞明白。只要状态流转是通的，后续替换任何节点都很轻松。
2.1 数据模型：研究任务的State定义
我用Python dataclass来定义State，这是LangGraph这类工作流框架里最常见的写法。State本质上就是一个所有模块共享的数据结构：
from dataclasses import dataclass, field from typing import List, Optional @dataclass class DeepResearchState: instruction: str # 用户原始指令 subtasks: List[dict] = field(default_factory=list) # 已拆分的子问题 search_results: List[dict] = field(default_factory=list) # 每次搜索的原始结果 documents: List[dict] = field(default_factory=list) # 已抓取/清洗的正文 findings: List[dict] = field(default_factory=list) # 每个子问题的阶段结论 report: Optional[str] = None # 最终报告 current_round: int = 0 # 当前研究轮次 max_rounds: int = 3 # 最大轮次控制几个字段的设计考量：
subtasks存的是字典而不是单独的类，好处是灵活，Planner想往里面塞多少附加信息（比如建议的搜索关键词）都不需要改State定义。documents字段把“原文”完整存放，是为了最后Writer阶段写报告时能够引用原文位置，不至于只凭中间结论凭空发挥。current_round和max_rounds是安全阀。DeepResearch最怕的就是Controller判断“资料不够”，然后无限追加搜索。这个字段确保程序有一个明确的终点。
2.2 总控循环：让Planner、Searcher、Analyst轮流干活
有了State之后，整体控制逻辑就是一个循环。这里用最直观的while循环写一版可运行的骨架，避免引入太多框架概念：
def run_deep_research(state: DeepResearchState): # 第一步：生成子任务 if not state.subtasks: state.subtasks = planner.invoke(state.instruction) # 第二步：对每个子任务执行搜索与阅读 for st in state.subtasks: if st.get("status") == "done": continue queries = derive_search_queries(st) # 一个子问题可能对应多个query for q in queries: results = search_web(q) # 搜索引擎API调用 for item in results: doc = fetch_and_clean(item["url"]) # 抓取正文并清洗 if doc: state.documents.append({ "subtask_id": st["id"], "query": q, "url": item["url"], "content": doc }) st["status"] = "done" # 第三步：针对每个子任务产出现阶段结论 grouped = group_documents_by_subtask(state.documents) for st in state.subtasks: docs = grouped.get(st["id"], []) if docs: finding = analyst.analyze(st, docs) state.findings.append(finding) # 第四步：判断是否追加深挖轮 if state.current_round < state.max_rounds: followups = controller.needs_followup(state) if followups: state.subtasks.extend(followups) state.current_round += 1 return run_deep_research(state) # 第五步：撰写最终报告 state.report = writer.invoke(state) return state.report这段代码非常粗糙，但它完美展示了“循环 + 状态累积”的研究范式。每一个递归调用对应一轮深挖；每轮深挖都会把新获取的documents追加到State里，旧结论不会丢失。
实际生产环境里，我一般不用递归，而是用一个显式的for round in range(max_rounds)循环，并在每轮结束时把关键中间结果落一份缓存。这样如果搜索API在第三轮挂了，至少前两轮的结果还能用，不会全盘重来。
2.3 外部依赖与最小运行环境
要跑通这个骨架，你至少需要：
- 一个搜索引擎API（比如Tavily、SerpAPI、Bing Web Search）。我日常调试用Tavily，因为返回结构干净，直接给
url和title。 - 一个支持网页正文提取的库。
BeautifulSoup加requests是最经典组合，但面对反爬站点时经常会拿到空壳。 - 一个大模型API，用于Planner、Analyst、Writer三个节点的LLM调用。
搭建环境就三步：
pip install requests beautifulsoup4 openai tavily-python export OPENAI_API_KEY="your-key" export TAVILY_API_KEY="your-key"后端模型我建议先直接用最强的那个，先把流程跑通，再慢慢换小模型来省成本。因为DeepResearch的大部分质量不是靠单次推理，而是靠整个流程，有时候同一个模型性能差一些，只要流程稳健，结果差距并没有想象中那么大。
3. 核心节点逐个拆：规划、搜索、浓缩、成稿
骨架搭好之后，真正决定成品质量的是节点内部的实现。下面这四个节点是整个系统的灵魂，每一个都有不少值得展开的设计细节。
3.1 Planner：把大主题拆成可搜索的子问题
Planner的输入是用户的一句话，输出是一组“能直接拿去搜索的子问题”。比如给定“调研2024年主流向量数据库的对比”，Planner应该拆出这样的子任务：
- 向量数据库的核心指标有哪些维度（性能、功能、生态）
- 主流的开源方案（Milvus、Qdrant、Weaviate等）各自特点
- 这几家的性能基准测试公开结果
- 企业选型时常见的评估维度
每个子任务还需要带上search_query字段，它决定了Searcher真正去搜什么。一个常见的错误是让Planner直接拿“子问题”当搜索词，比如“向量数据库的核心指标有哪些维度”这种整句丢给搜索引擎，效果很差。实际实现中，我会让Planner先输出子问题，再为每个子问题生成1到3个不同的搜索词变体，比如用“向量数据库 benchmark “milvus vs qdrant”“2024”这种带引号和年份的长尾词。
Prompt设计上有一个很管用的小技巧：给Planner限定子问题数量区间（比如3到6个），并且要求每个子问题必须包含“可检索性”字段。如果模型的第一次输出里有一个子问题明显没法用搜索引擎找到答案，比如“分析人类对数据库的潜在偏见”，你需要在第二轮让模型重写，而不是硬着头皮往下走。
3.2 Searcher：搜索接口抽象与多query并发
Searcher模块是代码里对“外部世界”的接口。它做得很粗糙的话，搜索引擎返回什么就收什么，会给下游带来一堆噪声。我在项目里通常把它封装成一个带重试和并发控制的服务类：
import asyncio from tenacity import retry, stop_after_attempt, wait_exponential class SearchService: def __init__(self, api_key: str, concurrency: int = 4): self._client = TavilyClient(api_key=api_key) self._sem = asyncio.Semaphore(concurrency) @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=30)) async def search(self, query: str, top_k: int = 6): async with self._sem: resp = await self._client.search(query=query, max_results=top_k) return resp["results"]这里有三层设计：
- 信号量并发控制：
Semaphore(concurrency)限制同时进行的搜索请求数量。搜索引擎API是按调用次数计费的，而且大多数有每分钟速率限制，不加并发控制在多query场景下必炸。 - 重试策略：搜索API偶尔返回5xx。加
tenacity重试，用指数退避而不是固定间隔，能有效避开限流周期。 - 结果条数控制：
top_k不要设太大，每个query取6条左右就够了。取20条的结果往往是大量的低质SEO聚合页混进来，你去重和清洗的成本远高于多一次的补充搜索。
3.3 Processor与记忆浓缩：不无限塞页面的关键
搜索拿到URL之后就要抓取正文。这段代码的坑比想象中多。一个成熟的网页正文提取器至少要处理：广告导航、无正文的JS渲染页、乱码字符、重复文本等。我的经验是用trafilatura这个库替代手写的BeautifulSoup逻辑，它在抽取新闻和文章类页面时表现相当稳定。
但抓取只是第一步。真正的难点是怎么把抓回来的海量文本塞进小模型的上下文。一个页面干干净净之后可能还有8000到50000字，两个页面就能顶爆早期模型的窗口。所以Processor之后必须跟着一个“浓缩记忆”步骤。
浓缩的实现路径是分层摘要：
def condense_document(content: str, max_chars: int = 3000): if len(content) <= max_chars: return content chunks = split_into_chunks(content, chunk_size=2000, overlap=200) summaries = [] for i in range(0, len(chunks), 2): batch = chunks[i:i + 2] s = llm.summarize("总结以下片段中的事实性信息，不要臆测", "\n".join(batch)) summaries.append(s) if len("".join(summaries)) > max_chars: return condense_document("".join(summaries), max_chars) return ". ".join(summaries)思路就是不断做4到5个片段的小范围摘要，再把摘要拼起来继续摘要，直到压缩到目标长度。这种“摘要的摘要”在实践里效果还不错，虽然会损失一些细节，但对下游分析师而言够用了。
给每篇网页保留的字段也值得讲究，我最终只存三样：url、title、condensed_content。原始全文就不进State了，只在引用校验时通过URL重新拉取。
3.4 Writer：成稿、引用、结论的强制对齐
Writer是最后一个节点，它做的不是“生成”，更像“拼装与排版”。它读已归档的一组findings，每个finding已经被Analyst归纳成了带引用标的短结论，Writer要做的只是把它们组织成一篇连贯报告，并给每段加上明确的结论。
这里最容易翻车的坑是引用对不上。Analyst的结论里可能写着[source-1]，但到了Writer阶段，State里的documents已经被浓缩过，引用索引早就错乱了。我的解决办法是在findings里直接内嵌引用信息，不做全局索引：
{ "subtask_id": "sub_2", "conclusion": "目前主流测试下，方案A的检索延迟优于方案B，尤其在百万级向量规模下", "evidence": [ {"url": "https://example.com/benchmark", "quote": "方案A p95 latency = 8ms"}, {"url": "https://example.com/compare", "quote": "方案B p95 latency = 35ms"} ] }这样Writer写报告时直接复制evidence里的quote作为脚注，不需要去猜某个引用编号指向哪里。这是一种很朴素的“内容可溯源”设计，虽然没有专门RAG系统做得那么花哨，但对研究报告来说足够扎实。
我在Prompt里还会强制Writer遵循两个规则：第一，每一部分的小结论后面必须挂来源引语；第二，如果某件事没有任何引语支撑，不许写进报告正论部分。第二个规则拦截了大量幻觉输出——宁可报告短一点，也不能让用户误以为一个没有依据的数字是真实的。
4. 不改一行Prompt也能提升质量的工程细节
Prompt写得好不好当然影响质量，但我实际测试下来，有一批工程细节对最终报告质量的影响力不亚于提示词。而且这些细节是“写死的”，不依赖模型的临时发挥，稳定性更好。
4.1 搜索去重与query改写
多轮搜索中，同一个URL被重复抓取的情况非常常见。今天搜“向量数据库 benchmark”，明天加年份搜“2024 向量数据库 benchmark”，第一页里很可能出现了同一个榜单页。如果你的代码不处理，唯一的区别就是State里多了一条重复documents，白白浪费一次LLM解析的时间。
所以Processor入口处维护一个URL集合：
seen_urls = set() for doc in state.documents: seen_urls.add(normalize_url(doc["url"])) # 抓取前先做归一化检查 def normalize_url(url: str) -> str: url = url.split("#")[0] url = url.rstrip("/") if url.startswith("http://"): url = "https://" + url[7:] return url.lower()URL归一化处理掉#锚点、末尾斜杠、http/https混用、大小写差异，这一套组合拳能去掉至少15%的重复抓取成本。
另一个隐蔽问题的是query改写。有些子任务第一轮搜索完全搜不出东西，不是主题冷门，而是query写得太笼统。我在Searcher返回空结果的时候，会触发一个“改写并复搜”的兜底：把原始query和已获得的零散文档交给LLM，让它生成3个更具体的query，再走一次搜索流程。这比直接把“没结果”上报给Controller要高效得多。
4.2 上下文裁剪的两个策略：截断阈值加分层摘要
前面第3.3节已经提到了分层摘要，但真实系统中还需要一个前置硬性截断。你压缩一篇5万字的页面，跟截断一篇5万字的页面，成本完全不同。
实际操作中我的策略是双阈值：
- 单文档硬限制：清洗并提取正文后，超过
max_input_chars（比如15000）的直接硬截断。 - 单子问题软限制：归属同一个子任务的全部文档，累计浓缩目标上限是8000到10000字。
这两个阈值一卡，整个研究流程的最大上下文规模就是可预估的了。假设5个子问题、每个子问题最多10个文档，那Analyst的输入上限大约5万字，再经过一轮分层压缩，全流程的token消耗基本可控。
我自己踩过最狠的坑，是一次跑了个长尾问题，第一轮就抓了60多个网站，单是原始正文就有20MB，直接导致后续所有LLM调用的context堆积到爆炸，API返回400错误。从那之后，我所有进入State的文本都必须过一道长度校验，这是硬规则，不是可选项。
4.3 同源交叉验证与一致性检查
研究报告和普通问答有一个很重要的差别：研究要求结论经得起推敲。代码层面的应对是“结论置信度标记”。
具体做法是在Analyst模块里，为每个子结论生成一个confidence字段和supporting_sources列表。如果同一个结论有两到三个不同站点的来源支撑，置信度标记为high；如果只有一个来源，标记为medium；如果来源之间存在明显矛盾，标记为conflict。
冲突检测不是让模型“看着凭感觉”，而是直接比对证据引语里的数字或时间戳。比如两个来源分别写“方案A成本约10万美元/年”和“方案A年费用5万美元”，Analyst应当被指示输出冲突标记，而不是自行选择一个“更可信”的数字写进报告。
到Writer阶段，conflict结论会单独放一节“存在争议的发现”，把两边观点都列出来。研究报告最重要的品质是可复核性，让用户看到冲突、让用户自己判断，永远好过模型替用户消歧。
4.4 防幻觉：读不到就别编，强制“证据缺失”标记
DeepResearch里最危险的一种失败模式是“一本正经地编造来源”。很多实现为了让报告好看，会在找不到证据时用模型已有的知识顶上去，结果就是报告里出现了搜索中根本不存在的数据。
对付这个问题的代码级手段很简单：在Analyst和Writer的Prompt里都加入一条“硬性要求”，如果结论没有对应证据，必须在结论前写下[证据不足]前缀。这条前缀同时会作为文本特征被下游解析，最终在报告里渲染成一个“待核实”的高亮块。
我见过一些团队用更激进的做法：禁止模型在无引用状态下写具体数字。一旦模型在输出中包含数字，代码会强制校验数字是否能在evidence.quote里找到。找不到就整段重写。这种校验会降低一些文字流畅度，但换来的是研究报告的底线安全，值得。
5. 实测翻车记录：从跑通到跑好的常见坑
这节记录我在实际把DeepResearch代码从“能跑”调到“能看”过程中遇到的最典型的几类问题，每一条都对应一个明确的修复方案，希望能帮你跳过这些阶段。
5.1 搜索API限流与超时
第一轮写代码时没做并发控制，5个子任务同时发出，每个子任务3个query，15个请求在10秒内全部打到搜索API上，然后就是一片429。随后我还犯了个错——用固定等待1秒重试，结果API极限恢复时间到了，所有重试仍然堆在同一个时间窗口里继续429。
最终改法就是第3.2节里那套方案：Semaphore限制并发数到4、tenacity用指数退避重试。另外一个建议是给每次搜索加上超时时间，单次搜索超过15秒直接放弃。研究报告损失一次搜索结果并不可怕，可怕的是整个流程卡死在那里等一个永远等不到的响应。
5.2 Context爆掉：爬虫拿到了6MB网页
有个案例至今印象深刻。某个研究问题涉及一家公司的财报页面，网页里嵌了大量Base64图片和JS变量，抓下来正文提取器把JSON脚本当正文抽了，一个文档就有6MB。压缩函数写的时候还假设输入长度最多几万字，结果直接撑爆了后续的LLM调用。
现在的Processor里会多一步格式嗅探：先看文档里是否包含大段<script>、长base64字符串、CSS花括号块，是的话先做“strip and classify”再走正文通路。硬截断阈值也从原来的max_chars=30000调低到15000。还是那句话，宁可丢一点边角信息，不能让拦截器瘫痪。
5.3 引用对不上：报告里的[source-3]指向了无关URL
早期我的Writing阶段是用“全局引用表”的方式——所有文档按顺序编号，Writer只需引用编号。听起来优雅，实际上Analyst阶段用的文档顺序和Writer阶段列表顺序很容易因为缓存顺序变化而错位。一篇报告里出现“引用[source-3]”，但我检查后发现source-3根本是另一篇不相关页面，这种错误对研究报告是致命的。
后来就改成了第3.4节写到的“evidence内嵌引用”模式。每一步分析结论都带着自己的URL和原文引语，天然不会错位，也简化了Writer的逻辑。这一类改动提醒我：在多Agent工作流里，让信息在自己的最小作用域内携带必要的上下文，远比集中式维护一个大索引靠谱。
5.4 答非所问：用户问“方案对比”，报告变成了流水账
最后一个翻车不是技术问题，而是用户意图理解问题。用户让DeepResearch做“方案对比”，但Planner拆出来的子任务全是“A方案特点”“B方案特点”，Writer最终交出来的报告就是两段平行罗列，完全没有比较表。
修复很简单，就是在Planner的Prompt里增强“研究视角多样性”指令。针对对比类问题，明确要求拆出“差异维度”这类子任务，而不是只拆“各方案介绍”。也就是让子任务配方和用户的原始意图严格对齐。你可以写一个简单的意图匹配逻辑：先让模型识别用户问的是“描述型”“对比型”还是“建议型”，再据此决定子任务模板。这套逻辑不复杂，但对成品形态的影响立竿见影。
最后分享一个我自己的实操体会：DeepResearch这类系统的门槛不在某个模型的API，而在工作流的工程稳定度。只要状态流转清楚、每个节点职责单一、引用链路可控，哪怕每个节点用的都是同一个普通模型，整条流水线跑出来的结果也已经远超“一次性RAG问答”的效果。如果你想实验，我建议先不要加太多花哨策略，就按这篇文章给出的最小骨架跑通一轮，再逐个模块去优化。顺着这个节奏，你很快就能拥有一版专属于自己的DeepResearch工具。
