做AI研究助手这个需求很多团队都踩过同一个坑给一个聊天机器人接上大模型和检索API就以为能做研究了。实测下来单Agent在“查资料—梳逻辑—出综述—验质量”这条长链路上很快就会崩——上下文被冲散、角色混乱、生成结果看起来像样但一问引用就露馅。后来我用LangGraph把整个流程拆成四个智能体协作的流水线从零搭了一个能跑通的专业AI研究助手。这篇文章就把我的完整实现思路、核心代码和踩过的坑整理出来适合手里有基础大模型API、想搞懂多智能体编排的开发者也适合想在项目里快速落地“研究/综述类”功能的人。1. 为什么研究助手非要“多智能体”不可1.1 单Agent跑长任务的三种典型崩法我早期做研究助手Demo时就是一个大模型加一个搜索API在系统提示词里塞满指令让模型自己“边查边写”。表面看流程简单但实际用起来翻车翻得很有规律。第一次翻车是上下文被打穿。比如研究“对比三种向量数据库在中文场景的落地表现”搜索API会返回几十个页面的摘要我把这些摘要全塞进上下文还没写到结论Token额度就爆了模型开始把前面的内容记错甚至复读。第二次翻车是角色指令互相污染——我在提示词里既要求“客观收集信息”又要求“生成有观点的综述”结果模型在收集阶段就开始下结论在综述阶段又去补充搜索逻辑线全乱了。第三次翻车最致命引用失真。模型会凭空生成看起来很像样的参考文献标题、作者、摘要、出处都对得上但一查全是幻觉。这三次翻车不是调提示词就能解决的本质是“一个人在长链条上既要当侦察兵又要当分析师又要当质检员”这件事本身就不现实。研究助手的关键难点不在“让模型产出文字”而在于“让研究过程可控、可验证、可定位故障点”。1.2 角色分工取代“一个人干所有事”多智能体架构的价值就是把整个研究任务拆成规划、检索、综述、审查四个环节每个环节由独立Agent负责环节之间通过结构化的状态对象传递信息。上一环节产出什么、下一环节消费什么都是明确定义的。这就像现实里的研究团队定选题的人、查文献的人、写综述的人、审稿的人各有分工。任何一个环节出问题都可以单独定位而不是一锅粥。LangGraph在这个架构里扮演的角色是任务调度中枢它把各个Agent定义为图上的节点把条件判断定义为边让流程不仅能按顺序执行还能在稿子质量不达标时自动绕回写手那边重写形成可控循环。我之所以选择LangGraph而不是自己写编排代码是因为图结构本身就表达了业务流程状态对象是唯一的数据通道不需要再维护一套“队列任务状态机”的胶水代码。下面这张对比表能直观看出差别对比维度单Agent方案LangGraph多智能体方案上下文管理全部塞进一个上下文易超限各节点只消费所需状态字段故障定位整个流程一个黑盒精确到某个节点质量控制靠提示词碰运气独立审查节点循环重写可扩展性改提示词等于重构增删节点即可调整流水线调试难度日志混乱可逐节点检查状态快照2. 四个角色的研究流水线规划、检索、综述、审查2.1 角色职责与“窄接口、宽实现”的边界设计系统里我设计了四个Agent角色Planner规划者接收用户的研究主题拆解成一组结构化子问题并为每个子问题生成检索关键词。Researcher检索者根据子问题逐项调用检索服务收集原始资料输出结构化证据列表。Writer写手基于证据列表生成一篇结构化综述草案。Critic审稿人对照标准检查综述输出质量判断和修改意见如果合格放行不合格退回Writer重写。这里有一个关键设计原则Researcher不改写资料Writer不自行检索Critic不做除了判决之外的事。每个节点的输入输出都被收窄到一个很小的范围调试和替换组件都变得容易。我把这套设计叫“窄接口、宽实现”——节点之间只通过状态对象的特定字段通信每个Agent内部用什么样的提示词模板和模型参数可以独立调优互相不干扰。这个隔离设计带来的直接好处是如果检索服务换了我只需要改Researcher内部逻辑Planner、Writer、Critic完全不用动。如果模型供应商切换我只需要在统一客户端层替换配置图的拓扑结构一字不改。2.2 为什么四个角色是这个项目的最小闭环有人会问为什么不是两个Agent一个干活一个检查或者六个Agent再拆出数据清洗和排版。我实测下来的感受是少于四个规划和执行混在一起容易跑偏检索和写作混在一起会导致生成内容不受证据约束多于四个编排复杂度和Token开销增长非常快小团队根本维护不过来。规划、检索、综述、审查这四个环节正好对应研究流程中四个物理上分离的工序组成了最小完整闭环。我之前试验过把“检索”再拆成“查询生成”和“内容抓取”两个子Agent结果查询生成环节的收益微乎其微反而因为多一轮模型调用每次检索慢了几秒还多花几千Token最终把这个拆分砍掉了。这个教训就是多智能体不是越多越好而是越能对应真实工序分工越好。3. 开跑前的状态设计决定了整个图好不好写3.1 安装依赖版本别追太新LangGraph本身是一个Python库安装很简单。不过在动手之前我建议先把依赖范围钉住不要无脑装最新版。我在本地用的组合是langgraph配合langchain-core模型客户端用通用的接口封装这样后续换模型不影响流水线代码。pip install langgraph langchain-core这里提醒一句LangGraph的API迭代比较快跨大版本之间节点返回的格式、StateGraph的导入路径都可能有变化。如果你照着网上老教程写突然报ImportError大概率是版本不一致。建议固定主版本号再调试比如写成langgraph0.2,0.4这种范围。3.2 TypedDict状态对象图上流转的唯一数据通道LangGraph用状态对象管理节点之间的数据流转。我定义了一个ResearchState使用了typing.TypedDict每个字段标明类型让图上的数据流一目了然from typing import Any, Dict, List, TypedDict class ResearchState(TypedDict, totalFalse): topic: str # 用户输入的研究主题 sub_questions: List[str] # Planner 拆解出的子问题 search_keywords: List[str] # Planner 生成的检索关键词 evidence: List[Dict[str, Any]] # Researcher 收集的证据列表 draft: str # Writer 生成的综述草案 review: Dict[str, str] # Critic 的审查结果verdict comment revision_count: int # 记录重写次数防止死循环这段状态设计有几个细节值得说明。totalFalse表示字段可以逐步补齐图执行时从一个只有topic的字典开始每个节点往里填新字段。evidence用List[Dict[str, Any]]承载异构数据是因为检索结果的字段经常不固定写成强类型反而要处理一堆兼容逻辑。状态对象的另一大作用是断点调试。LangGraph支持检查点checkpoint可以打印出每个节点执行后的完整状态快照。哪一步产生了脏数据、哪一步没拿到预期字段看一眼状态就明白了这是自研状态机很难做到的。3.3 节点函数的第一原则只消费状态、只返回更新LangGraph节点本质上是普通Python函数接收当前状态字典返回一个“部分状态更新”字典。我会特别强调这个原则——不要直接在节点内部修改原状态对象而是返回一个新的字典LangGraph自动合并进状态。def sample_node(state: ResearchState) - Dict[str, Any]: # 只读取 state 的字段 topic state[topic] # 做点什么 result do_something(topic) # 返回需要更新的字段未涉及的字段保持不变 return {new_field: result}这种函数式风格让每个节点可以独立测试你先手动构造一个状态字典喂给节点看返回值是否符合预期再放进图里跑。所以我把每个节点的核心逻辑都写成纯函数模型调用和API调用包在外层方便单测。4. 四个核心节点的实现与提示词细节4.1 Planner把主题拆成可检索的JSONPlanner的任务是把宏观主题变成“可以用搜索引擎直接查”的具体问题。我让模型以JSON格式输出然后用json.loads解析这样可以避免在长字符串里用正则抠字段。import json def planner_node(state: ResearchState) - Dict[str, Any]: prompt f你是研究规划者。用户的研究主题是{state[topic]} 你的任务 1. 将主题拆解为 3~5 个互不重叠的子问题 2. 为每个子问题给出 1~2 组中文检索关键词 3. 子问题要具体到可以被搜索引擎直接检索 严格输出 JSON不要输出其他内容 {{sub_questions: [子问题1, 子问题2], search_keywords: [关键词组1, 关键词组2]}} resp llm_client.chat(prompt, response_formatjson) data json.loads(resp) return { sub_questions: data[sub_questions], search_keywords: data[search_keywords], }这里需要两个提示词技巧。第一数量要封顶我明确写“3~5个”否则模型会给你拆出一大堆子问题后面的检索和Token开销都会爆炸。第二关键词要独立于子问题子问题是给人看的关键词是给检索API用的两者数量不一定一一对应所以我在状态里分开存两个列表。我在测试中发现一个常见问题模型倾向于把子问题写得太宏观比如“介绍向量数据库的原理”这种问题搜出来的结果泛而浅。解决办法是在提示词里加一句“子问题要具体到可以被搜索引擎直接检索”模型就会自动拆成“对比HNSW和IVF索引的召回率差异”这种颗粒度。4.2 Researcher检索、去重、截断的细节Researcher是四个节点里最容易写出性能问题的环节。如果不对检索结果做数量和长度限制后面的Writer会瞬间被打爆上下文。我给每个子问题最多取3条结果每条摘要截断到300字符以内同时用URL做全局去重def researcher_node(state: ResearchState) - Dict[str, Any]: new_evidence [] seen_urls set() for q, kw in zip(state[sub_questions], state[search_keywords]): results search_client.search(kw, top_k3) for item in results: url item.get(url, ) if url in seen_urls: continue seen_urls.add(url) snippet item.get(snippet, )[:300] new_evidence.append({ title: item.get(title, ), snippet: snippet, url: url, question: q, }) return {evidence: new_evidence}检索器的输出我特意加了一个question字段把证据跟子问题关联起来。这样Writer在引用时能知道每个证据对应的上下文Critic在审查时也能检查“每个子问题是否都有证据支撑”。这是一个小改动但对后两个节点的质量提升很关键。有一个实现细节需要提醒很多搜索API返回的snippet包含HTML标签或特殊符号最好在存入状态之前做一次基本清洗否则写综述时模型会把这些噪声带进正文。4.3 Writer用编号引用倒逼模型不编造Writer的核心难点不是“写得好不好”而是“写的内容有没有出处”。我用最笨但最有效的方法把证据列表拼成带编号的上下文强制模型引用编号。def writer_node(state: ResearchState) - Dict[str, Any]: evidence state[evidence] context_blocks [] for i, e in enumerate(evidence, 1): context_blocks.append( f[{i}] 标题{e[title]}\n内容{e[snippet]}\n来源{e[url]} ) context \n\n.join(context_blocks) prompt f以下是检索到的资料请基于这些资料撰写一篇结构化综述。 要求 1. 只依据引号内的资料撰写禁止编造数据、机构或来源 2. 关键论断后标注来源编号例如 [1]、[2] 3. 结构包含概述、分论点、结论 4. 如果资料不足明确写“当前资料不足以支撑此论点”不要强行补充 资料 {context} 研究主题{state[topic]} 综述 draft llm_client.chat(prompt, max_tokens2500) return {draft: draft}为什么编号引用能防止模型编造因为模型在生成时会把“事实来源”从隐性的参数记忆变成了显性的上下文约束它必须引用给定编号范围内的资料如果它引用了一个不存在的[8]就明显违反了结构约束。当然这不是万无一失的但比不编号时幻觉率下降非常多。另一个关键设计是第4条要求——“如果资料不足明确写当前资料不足以支撑此论点”。模型为了输出完整综述很容易拿训练语料里泛泛的背景知识来凑数这一条能有效逼出资料缺口让Critic有据可查。4.4 Critic质量关卡和修改反馈Critic是我觉得最有价值的节点。它不做重写而是扮演一个带明确验收标准的审稿人逐项检查并输出结构化判决。def critic_node(state: ResearchState) - Dict[str, Any]: topic state[topic] sub_questions state[sub_questions] draft state[draft] evidence state[evidence] prompt f你是严格的审稿人。请检查以下研究综述是否符合标准。 审查维度 1. 覆盖度是否回答了全部子问题 {sub_questions} 2. 证据度每个关键论断是否有对应的引用编号引用编号是否在资料总数范围内 3. 一致性是否存在自相矛盾 4. 结构性是否包含概述、分论点、结论 综述全文 {draft} 现有资料总数{len(evidence)} 严格按以下 JSON 输出不要输出其他内容 {{verdict: pass 或 fail, comment: 具体的修改意见}} resp llm_client.chat(prompt, response_formatjson) review json.loads(resp) return { review: review, revision_count: state[revision_count] (1 if review[verdict] fail else 0), }Critic的提示词里有两个细节容易被忽略。第一是引用编号范围校验我会把现有资料总数传给Critic它就能核对论文里的引用编号是否在范围内这一步极大压缩了幻觉引用的生存空间。第二是comment必须具体只写“质量不行”等于没说所以我要求它输出“哪个子问题没覆盖到”“哪里缺少引用编号”这样重写的Writer才有方向。5. 用StateGraph组装流程并让条件路由决定“重写”还是“收工”5.1 图的构建节点注册与顺序边基础执行顺序是一个线性链路Planner - Researcher - Writer - Critic。我用StateGraph构建from langgraph.graph import StateGraph, END builder StateGraph(ResearchState) builder.add_node(planner, planner_node) builder.add_node(researcher, researcher_node) builder.add_node(writer, writer_node) builder.add_node(critic, critic_node) builder.set_entry_point(planner) builder.add_edge(planner, researcher) builder.add_edge(researcher, writer) builder.add_edge(writer, critic)这里梳理一下语义节点代表“一段可复用的处理单元”边代表“前一个节点的输出状态作为后一个节点的输入状态”。LangGraph执行时会沿着边把状态对象传给下一个节点整个过程对开发者来说是透明的——你可以随时把中间状态打印出来看。5.2 条件路由的循环边界设计如果Critic给了fail流程需要回到Writer重写而不是直接结束。这就要用到add_conditional_edges根据状态字段决定下一步走哪条边MAX_REVISION 2 def router_after_critic(state: ResearchState) - str: if state[review][verdict] fail and state[revision_count] MAX_REVISION: return writer return end builder.add_conditional_edges( critic, router_after_critic, {writer: writer, end: END}, )这里最容易被忽视的是“中止循环”的边界条件。如果只有verdict fail就返回writer当Critic一直不满意时流程会无限循环白白烧掉大量Token。所以我用revision_count做计数器最多重写两次之后无论Critic是否通过都强制结束。我一开始没有这个计数器结果是测试数据里有一条需要大量专业背景的综述Critic连续判了四次fail每次重写都会拉高全文质量却始终达不到它“完美主义”的标准——因为模型和审判模型是同一个它的标准会跟着内容水涨船高。加计数器不是妥协而是让系统在成本和上限之间取得平衡。5.3 完整可运行的运行入口编译并执行app builder.compile() if __name__ __main__: result app.invoke({ topic: 对比几种主流的关键词提取方法在中文学术场景下的表现, }) print(最终综述\n, result[draft]) print(\n审查结果\n, result[review])这个入口只是示意。实际接入服务时我会把app.compile()包装成一个异步函数用astream做流式输出这在下一节详述。6. 实测中的四个坑Streaming、死循环、Token配额、引用可信度6.1 没有Streaming时用户以为服务挂了第一个版本我直接用invoke同步调用结果前端等待提示一直是“研究中…”用户经常以为服务挂了。研究任务本身就慢因为要拆问题、多次检索、写长文、审查一次完整调用可能在10到30秒之间。用户等30秒什么反馈都没有体验非常差。我用astream_events方案解决了这个问题把每个节点的开始与结束事件推到前端前端展示一个流程状态面板——“正在拆解子问题”“正在检索资料”“正在撰写综述”“正在审查”。这让等待过程变成了产品体验的一部分。实现上不是直接调用invoke而是用astream迭代状态变化每步拿到对应节点的名字async for event in app.astream(initial_state, stream_modeupdates): for node_name, updated_values in event.items(): print(f节点 {node_name} 完成更新了字段 {list(updated_values.keys())})这个改动对项目上线意义很大。即使你不做前端调试时用stream_modeupdates也能实时看到每个节点跑到了哪一步出问题能立刻定位到具体节点。6.2 Critic死循环修订次数和降级策略前面提到过计数器问题。这里补充一个更细的坑Critic会因重复审查同一个文本而放大自己的标准。我观察到一个现象同一份综述如果第一次Critic给fail第二次重写后内容质量提升了但Critic的新标准也跟着提升了于是又给fail。到第三次重写模型已经在大改结构了但Critic还是不满意。加MAX_REVISION 2之后流程会在第二次重写后强制收尾。但如果此时综述仍然不合格就直接给用户交一份劣质结果也不行。我的策略是两阶段降级前两轮走严格审查如果重写两轮仍fail就切换到宽松模式让Critic只检查“是否存在伪造来源”和“是否包含完整结构”不再追求完美表述然后作为最终稿提交。但实测这个降级策略有个前提必须明确在输出里注明“该结果未经完全审查通过”否则用户可能把低质量内容当成研究结论。我在最终返回上加了一个warning字段由前端展示。6.3 给Writer留Token给Researcher省Token研究任务的Token开销比普通对话高一个量级因为检索结果上下文很长。我的预算分配原则是Token要花在必要的地方检索结果必须被压缩写作能力必须给足。节点默认模型档位最大Token限制说明Planner低成本快模型500只生成JSON不需要大幅生成Researcher不调用模型0纯API检索靠截断控制体积Writer高性能模型2500生成主体内容token要给足Critic低成本快模型500只输出JSON判决检索结果我限定每条摘要最多300字符每个子问题最多取3条总体证据列表控制在10条以内。这样Writer的输入上下文基本在6000字符以内加上生成内容一轮完整调用大约消耗12000到15000 Token。如果检索摘要不截断这条数字很容易翻倍甚至翻三倍。6.4 引用可信度宁可丢证据不要伪来源研究类产品的底线是引用真实可信。大模型生成的引用即使有编号约束依然可能在语义上错配比如引用编号对应的资料其实不支撑该论断。为此我在Critic之外加了第三个细节在后处理阶段把所有引用编号解析回证据列表再验证编号存在性。def verify_citations(draft: str, evidence_count: int) - bool: import re cited re.findall(r\[(\d)\], draft) valid all(1 int(n) evidence_count for n in cited) return valid, set(int(n) for n in cited)如果查出来有编号不存在我不会要求模型重写而是直接在返回层过滤掉非法引用并给用户标注“部分引用待核实”。因为到了这一步让模型重修的边际收益很低而再次校验的成本却不低。倒不如把不可信的部分标记出来保持产品透明。7. 我后续想扩展的三个方向记忆、工具和成本7.1 用检查点机制把状态持久化当前实现里每次invoke都是无状态执行的用户不能追问“刚才那个主题的结论依据是什么”。要做多轮对话必须把状态对象持久化。LangGraph官方推荐的做法是用检查点器把状态写入数据库比如内存、Redis或SQLite。from langgraph.checkpoint import InMemorySaver saver InMemorySaver() app builder.compile(checkpointersaver)这样每一次执行都有独立线程ID再次调用时传入同一线程ID即可恢复上轮状态。我计划的升级路径是把InMemorySaver替换成持久化存储让研究过程可以被记录、回放和复跑。7.2 接入私有知识库和代码执行环境纯检索API能覆盖的是公网公开内容但在企业内部场景很多资料在内部知识库、代码仓库或数据库里。下一步我会把Researcher改成可插拔的工具路由检索公网走搜索API检索内部资料走向量数据库甚至会接入一个沙箱编程工具让Agent在需要验证代码示例时直接执行。多智能体的可扩展性在这里体现得最明显我不用改Planner和Writer的逻辑只需要让Researcher内部多一个路由器把不同来源的证据统一转换成title/snippet/url结构即可。这再次验证了“窄接口、宽实现”的价值。7.3 模型分级与缓存策略模型费用是研究助手上线后最大的成本项。我从最初就做了两个优化第一Planner和Critic使用低成本低延迟的模型档位因为它们的输出都是结构化JSON不需要太强的生成能力Writer才用高性能模型因为它是最终输出质量的关键。第二对检索结果做缓存同一主题关键词在短时间内的检索结果直接复用避免重复调用搜索API和重复消耗上下文。这两个优化做下来单次研究任务的费用大约能省四分之一到三分之一。做多Agent系统时最容易犯的错是“所有Agent都用同一个最强模型”——既贵又没必要按角色定档才是长期运营的解法。最后分享一次实际运行的体会把四个Agent通过LangGraph串起来之后我第一次直观感受到了“流程控制”和“提示词工程”的边界提示词负责让每个Agent在它那一环尽量做好LangGraph负责保证它们按正确的顺序、正确的条件、可控的次数协同工作。后续再遇到“让AI完成一个多步骤任务”的需求我都会先画一张流程图把工序拆清楚再交给LangGraph实现——这比一开始就堆提示词可靠得多。如果你的项目也需要自动产出带真实引用的研究报告照着上面的代码骨架跑起来再按自己的数据源和审查标准调优一个月内就能看到像样的效果。