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_factorylist) # 已拆分的子问题 search_results: List[dict] field(default_factorylist) # 每次搜索的原始结果 documents: List[dict] field(default_factorylist) # 已抓取/清洗的正文 findings: List[dict] field(default_factorylist) # 每个子问题的阶段结论 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_KEYyour-key export TAVILY_API_KEYyour-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_keyapi_key) self._sem asyncio.Semaphore(concurrency) retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max30)) async def search(self, query: str, top_k: int 6): async with self._sem: resp await self._client.search(queryquery, max_resultstop_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_size2000, overlap200) 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个query15个请求在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_chars30000调低到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工具。