★ FEATURED ARTICLE 做科研的人都知道读论文、建综述、跟踪前沿这项工作本身有一套完整的流水线。但市场上很多AI工具只能做“单打独斗”的事给个PDF它帮忙总结一下给个关键词它帮你搜索但离真正帮你把一个科研任务从头到尾跑完还差得远。LangGraph 多智能体模式解决的就是这个问题——把“总结一下”“查几篇文献”“对比研究方法”“写个综述草稿”这些能力拆成不同的角色让它们在一个有状态、可控制的流程里协作。这个系列我会带着大家从零开始搭建一个协同式科研文献智能助手第一篇先解决核心骨架怎么设计Agent分工怎么用LangGraph把它们编排起来以及怎么处理各环节之间的信息传递。1. 为什么我不建议用单Agent硬撑而是拆成多个智能体协同1.1 单Agent做科研辅助遇到的三道坎先聊一个很实际的问题既然ChatGPT这种大模型已经能“看起来什么都会”为什么我还要给文献助手专门拆成多Agent我在第一版原型里其实试过用单Agent硬扛。你可以把所有的检索工具、总结工具、写作提示词都丢给同一个Agent让它自己决定用什么工具、按什么顺序执行。表面上看代码简单了但实际跑起来会接连撞上三堵墙。第一堵墙是上下文污染。科研流程每一步对信息的需求是不同的检索时你需要干净的关键词精读时你需要论文全文的局部内容做对比分析时你需要结构化表格写综述时你需要前几步产出的沉淀结论。如果所有中间结果都往同一个对话上下文里塞很快上下文就被各种原始文本填满模型注意力被稀释后面的分析质量直线下降。我实测过处理十篇论文以上的量单Agent模式最后的综述生成基本会丢掉关键对比维度。第二堵墙是状态难以控制。文献助手天然是“多轮工具调用”的场景先查查完要选选完要读读完要归纳归纳完要写。单Agent模式下这个流程由模型自由发挥你很难保证它每一轮都在正确轨道上。它很有可能在第3步读过论文后第4步突然开始写综述完全跳过对比分析——因为大模型并不天然理解科研工作的标准步骤。第三堵墙是排错成本。单Agent出问题时你不知道是检索环节崩了还是总结环节输出格式不对还是模型自己绕晕了。调试就像在一坨面条里找一根线头。1.2 多智能体的三种主流协作模式多Agent协作有很多种模式目前社区里常见的大致是三类。第一种叫Supervisor模式也叫“上级-下属”模式。一个主控Agent负责任务理解和分配其余的Worker Agent只干自己那一摊子活干完交结果给主控。这种模式适合流水线清晰、环节边界明确的场景。第二种叫Peer模式Agent之间地位平等可以互相提问、互相回答。这种模式适合头脑风暴类的场景但实现难度高而且容易出现聊着聊着就跑偏。第三种叫Hierarchical模式类似公司的多层汇报用多个层级的Supervisor去管理大量Agent。这个适合工程上大规模部署但对于一个科研文献助手来说有点杀鸡用牛刀。我选择的是Supervisor模式。原因是科研文献处理本身就是一个天然的流水线查文献 → 精读 → 对比 → 综述每个环节依赖前一个环节的输出边界明确。用Supervisor做总控四个专业化Agent各自专注一件事既符合直觉又方便出现问题时分模块排查。1.3 科研文献助手的整体架构设计这个项目的整体结构是这样的核心是一个Supervisor Agent下面挂四个WorkerSearch Agent负责文献检索输入关键词输出论文列表Review Agent负责单篇精读输入论文ID或文本块输出结构化笔记Compare Agent负责多篇对比输入多篇笔记输出对比表格Writer Agent负责综述生成输入对比结果和笔记输出综述初稿信息流方向是用户输入研究主题 → Supervisor理解需求 → 调用Search → 挑选论文 → 调用Review → 调用Compare → 调用Writer → 最终产出综述初稿。四个Agent之间不直接通信所有信息都通过一个中心化状态State传递。这就是LangGraph相对其他Agent框架的核心优势Agent只是图中的节点数据流是显式的。注意多Agent不是越多越好。我见过有人给文献助手塞了十几个Agent最后光调输出格式格式就花了半天。分4个是因为科研流程的核心环节数出来正好是4个不能多不能少。2. 环境准备与LangGraph的核心概念2.1 项目骨架与依赖安装先说环境的硬性准备。这个项目的核心依赖如下pip install langgraph langchain langchain-openai openai arxiv pypdf chromadb推荐Python版本3.10以上因为后面用到的一些类型标注特性需要新语法支持。我自己用的是Python 3.11。简单解释一下每个包的用处包名用途langgraph核心的状态图编排框架langchain工具封装、输出解析的统一接口langchain-openaiOpenAI模型接入openai直接操作大模型APIarxiv学术论文元数据检索pypdfPDF文本解析chromadb本地向量库用于文献语义检索提示如果本地网络环境访问部分模型接口不稳定建议优先配置好稳定的API端点与代理策略。这一步没做好后面所有Agent都跑不动绕不开这个前提。2.2 理解StateGraph先建模状态再定义流动LangGraph的核心模型是“图”图上有节点节点之间有边。节点的函数是“输入一个状态输出对状态的修改”。图的执行引擎负责按边定义流动状态。它的大白话解释就是整个流程是一张有向图你定义好每个环节做什么、环节之间怎么走框架帮你按图执行。我第一次上手时最容易犯的错是把这个框架当成“调用链”去理解——怎么串联就是线性调用。但LangGraph真正的价值在于支持条件分支conditional edges支持循环比如Review发现信息不足可以回到Search重新查支持人工介入human-in-the-loop支持状态持久化checkpoint这些特性对于科研流程很重要。比如Review Agent读论文时发现检索结果里缺了关键方法需要回去补查这就是循环的本质需求。2.3 先跑一个最小可运行的Demo先看一个最简单的例子理解完整的“画图-编译-运行”流程from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END # 1. 定义状态 class State(TypedDict): query: str result: str # 2. 定义节点函数 def search_node(state: State) - dict: # 模拟检索 return {result: f检索完成{state[query]}} def review_node(state: State) - dict: return {result: f精读完成{state[result]}} # 3. 构建图 graph StateGraph(State) graph.add_node(search, search_node) graph.add_node(review, review_node) graph.add_edge(START, search) graph.add_edge(search, review) graph.add_edge(review, END) # 4. 编译并运行 app graph.compile() output app.invoke({query: graph neural networks}) print(output)这个例子看着简单但它传达了LangGraph最重要的编程范式节点是纯函数只接收state只返回对state的修改不进行显式的相互调用。这种设计的价值在大型项目里才会显现。所有节点只关心“读什么、写什么”不关心“谁在调用我”使得每个Agent的测试成本极低——单独拉出来就能跑。3. 从零构建科研文献助手的四个Worker3.1 Search Agent先解决“怎么找”的问题Search Agent负责的工作是把用户的一句话需求转成具体检索行为跑一个或多个检索工具然后把结构化论文列表写进状态。这个Agent的关键点在于提示词设计。你不能直接把用户的原话丢给搜索接口。比如用户说“我想了解图神经网络在药物发现中的应用”这个自然语言需要用模型转换成适合学术检索的关键词组合必要时拆成多组。我的实现里给它配了两个工具from langchain_core.tools import tool tool def search_arxiv(query: str, max_results: int 10) - list[dict]: 使用arxiv API检索学术论文。 import arxiv search arxiv.Search( queryquery, max_resultsmax_results, sort_byarxiv.SortCriterion.Relevance, ) results [] for paper in search.results(): results.append({ id: paper.entry_id, title: paper.title, summary: paper.summary, published: str(paper.published.date()), authors: [a.name for a in paper.authors], }) return results tool def search_local_index(query: str, top_k: int 5) - list[dict]: 从本地向量知识库中检索相关段落。 # 这里可以替换成你自己的chroma库查询逻辑 results vector_store.similarity_search(query, ktop_k) return [{content: r.page_content, source: r.metadata.get(source, )} for r in results]Search Agent自己并不实现具体的检索算法它只是“决定”要调用哪些工具、传什么参数然后把工具返回的结果整理成统一结构写回状态。3.2 Review Agent建立结构化精读能力Search Agent返回一堆论文之后下一步是精读。Review Agent接收论文ID或全文文本返回结构化笔记。为什么要结构化因为后面的Compare Agent需要横向量化对比它需要的是标准格式字段而不是大段的自然语言。我在状态里规定了精读笔记的JSON结构structured_notes_schema { paper_id: str, title: str, research_problem: str, # 这篇论文解决什么问题 method: str, # 用了什么方法 experiments: list, # 实验设置与数据集 results: list, # 关键数字结论 limitations: str, # 作者自述的局限 key_insight: str, # 你总结的核心贡献 }Review Agent的一次精读成本比较高所以我把“精读”拆成两步先用摘要快速判断相关性只对通过初筛的论文做深度精读。这样能节省大量token消耗。实际实现里Review Agent的节点函数需要把PDF切块后分段送给模型。这里一定要控制chunk大小。经验值是一篇10页的论文控制在500-800词一个文本块分5-8块调用模型聚合成一篇结构化工整笔记。3.3 Compare Agent多篇对比与交叉验证Review Agent产出的是一篇篇独立笔记Compare Agent干的事是把它们放到同一张桌子上做对比。我在设计它的提示词时强制让模型按维度输出而不是按论文输出。很多AI工具生成对比表格容易写成“论文A做了什么论文B做了什么”这种流水账这不是对比。真正的对比应该按维度在方法维度上A与B有什么区别各自的核心设计特点是什么在实验设置上A用了什么数据集B用了什么差异意味着什么在结论上A和B的核心主张是否互相印证comp_result_schema { dimensions: [ { dimension: methodology, findings: [], discrepancies: str, # 分歧点 }, # ... ], summary: str, }Compare Agent真正消耗的是模型的推理能力比Review更“烧脑”。我发现它的输出质量高度依赖上游笔记的规范性。如果Review返回的笔记字段参差不齐Compare就会自己脑补缺失内容导致对比结果不可信.3.4 Writer Agent综述生成最后一步是Writer Agent把前面的结构化分析转成一篇可读性好的综述初稿。这里要特别注意节奏。很多人会用同一个模型同一个提示词去同时做分析、整合、写作结果写出来的综述像拼盘而不是融会贯通。我的做法是给Writer Agent设计两层提示第一层生成大纲第二层基于大纲逐段生成。大纲生成阶段输入的是Compare Agent的维度信息和关键分歧点。模型先规划文章逻辑结构背景引入、方法综述、关键发现、现存挑战、展望。这个阶段不生成任何正文只输出提纲。正文生成阶段按提纲逐段执行每一段都必须引用对应的论文ID。实操中用Output Parser强制模型输出带引注的格式能大幅减少幻觉引用。3.5 Supervisor编排逻辑什么时候走哪条路整个图的执行逻辑是这样定义的from langgraph.graph import StateGraph, START, END graph StateGraph(ResearchState) # 添加节点 graph.add_node(supervisor, supervisor_node) graph.add_node(search_agent, search_agent_node) graph.add_node(review_agent, review_agent_node) graph.add_node(compare_agent, compare_agent_node) graph.add_node(writer_agent, writer_agent_node) # 开始与结束 graph.add_edge(START, supervisor) # 条件路由supervisor根据决策决定下一步去哪个agent graph.add_conditional_edges( supervisor, lambda state: state[next_step], { search: search_agent, review: review_agent, compare: compare_agent, write: writer_agent, finish: END, } ) # worker回到supervisor重新决策 for agent in [search_agent, review_agent, compare_agent, writer_agent]: graph.add_edge(agent, supervisor)关键点在于Supervisor节点里的决策逻辑。我用的不是硬编码规则而是用一个“小模型”充当路由器。它读当前状态里已完成的工作、用户需求和历史决策记录输出next_step字段。这个设计带来一个副产品流程不会死板地“必须走完四步才结束”。如果用户在中间阶段只想查文献不想写综述Supervisor会在Search完成后直接输出finish。这样用户检索完就能直接拿结果走人不需要等后面一连串无用操作跑完。4. 核心实现详解状态设计、Prompt工程与图装配4.1 状态数据结构设计整张图的“数据库”在LangGraph里State结构设计得好整个项目就成功了一半设计得差后面每个节点都会觉得信息“不够用”。我设计的ResearchState结构如下class ResearchState(TypedDict): # 用户需求 user_query: str # 检索阶段 search_keywords: str papers: list[dict] # 精读阶段 selected_paper_ids: list[str] notes: dict[str, dict] # paper_id - 结构化笔记 # 对比阶段 comparison: dict # 写作阶段 outline: str draft: str # 控制信号 next_step: str messages: list[str] # 决策历史记录设计时有个原则状态里的每个字段都必须有明确的“生产者”和“消费者”。比如papers只有search_agent写只有supervisor和review_agent读。这样你调试时能很清楚地知道如果你的review_agent没有拿到论文列表问题一定出在search_agent的返回值上。实操心得避免在State里塞大段原始文本。比如把论文全文塞进State会让graph的每次状态快照都超大启动checkpoint功能后磁盘占用会爆炸。正确做法是存入文件路径或ID需要时再加载。4.2 各Agent的工具绑定与调用链LangChain的工具绑定语法需要和LangGraph的节点模式结合。我们以Review Agent为例展示完整的绑定方式from langchain_openai import ChatOpenAI review_llm ChatOpenAI(modelgpt-4o-mini, temperature0.2) review_llm_with_tools review_llm.bind_tools([load_pdf_tool]) def review_agent_node(state: ResearchState) - dict: llm_response review_llm_with_tools.invoke( [ {role: system, content: REVIEW_AGENT_PROMPT}, {role: user, content: f待读论文{state[selected_paper_ids]}}, ] ) # 处理工具调用 if llm_response.tool_calls: for call in llm_response.tool_calls: if call[name] load_pdf: text load_pdf_tool.invoke(call[args]) note review_llm.invoke(format_note_prompt(text)) notes[call[args][paper_id]] note return {notes: notes, next_step: compare}这段代码展示了一个关键思想Agent节点既可以用大模型自主规划工具调用也可以直接把函数结果拿回来做二次处理。LangGraph不强制你用某一种方式它只是给节点之间提供状态流转的“管道”。4.3 Supervisor Prompt让模型当合格的项目经理Supervisor的提示词设计直接影响整个系统的兜底能力。我给Supervisor的System Prompt长这样你是一个科研文献项目管理器。你的职责是 1. 解析用户的研究目标拆解成当前可执行的具体步骤。 2. 根据当前进度从以下Agent中选择下一步执行search_agent、review_agent、compare_agent、writer_agent。 3. 四位Agent已经完成的服务会在会话历史中标注不要重复调用。 4. 当且仅当所有必要环节完成或者用户明确要求结束后才输出finish。 要求 - 每次决策只输出一个步骤名不要附加任何解释。 - 如果用户只要求检索检索完直接finish不要自作主张进入写作。“不要附加任何解释”这行字是为了让路由输出稳定可解析。我踩过一次坑就是Supervisor输出了一长串解释结果我还要写正则去提取步骤名。直接让它只输出一个词配合结构化输出省心很多。4.4 编译入口与人工审核介入图的编译和调用入口并不复杂关键是要学会使用interrupt机制。真实科研场景里AI检索回来的论文用户不一定都同意选中所以我在“选择论文”这个环节插入了Human-in-the-loop。实现方式是在Supervisor决策前设置一个breakpointapp graph.compile( checkpointercheckpointer, interrupt_before[review_agent], # 在review前暂停 ) # 首次运行后获取未决状态 config {configurable: {thread_id: user-session-001}} result app.invoke(initial_state, config) # 检查是否需要人工介入 if review_agent in app.get_state(config).next: print(等待用户确认论文列表...) # 用户修改selected_paper_ids app.update_state(config, {selected_paper_ids: user_confirmed_ids}) # 继续执行 result app.invoke(None, config)这个机制对科研场景研究价值非常大AI负责干活人类负责拍板。既保持了自动化效率又保住了科研工作者的决策权。4.5 用Checkpointer保存历史会话LangGraph的Checkpointer允许你把每次运行状态归档之后可以随时恢复和回溯。from langgraph.checkpoint.memory import MemorySaver YOUR_SAVER MemorySaver() app graph.compile(checkpointerYOUR_SAVER) config {configurable: {thread_id: thread-123}} app.invoke(initial_state, config)初次做这个项目用MemorySaver就够了它把状态存在内存中速度快。如果要部署成服务建议换SQSSaver或者自定义数据库保存这样才能跨进程恢复。我当时在产品化阶段踩过的坑是忘记配置checkpointer导致invoke每次调用都从空状态开始Supervisor完全不记得之前检索过什么。记住一句话LangGraph在没有checkpointer时是没有记忆的图只是一个纯计算结构。5. 调试实录这几次让我崩溃又涨经验的坑5.1 图的递归深度限制问题我第一次跑通完整流水线时遇到一个特别莫名其妙的报错RecursionLimit: Recursion limit reached in the graph.排查后发现原因是Supervisor和四个Worker之间形成了一个大循环。Worker执行完回到SupervisorSupervisor重新决策如果它不判断“已经完成”的条件就会一直把同一个Worker安排来安排去最终触发图的递归深度限制。加一个简单的决策边界就能解决在State里维护一个已完成任务列表Supervisor在做决策时先检查该任务是否已执行过。如果执行过绝不能重复分配而是转向finish。5.2 Worker返回格式不规范下游直接炸掉Compare Agent最怕上游给它的notes格式不对。比如Review Agent返回的experiments字段有时是字符串有时是列表有时直接缺失。下游解析JSON时就偶发崩溃。解决思路是给每个Worker的返回增加一个独立的数据清洗函数在写入State前强制转换为统规范结构。def normalize_note(raw: dict) - dict: return { paper_id: str(raw.get(paper_id, )), method: str(raw.get(method, )).strip(), results: [str(x) for x in raw.get(results, [])], # ... }5.3 上下文窗口跑长流程时Token直接溢出四步Agent流程跑完后积累了四轮对话记录。这些记录都汇总在messages列表里。到了第三步模型就已经记不住前面的关键信息了。我采取的方案是“摘要化历史记录”与“分批读取”结合。核心原则是状态里只留当前最重要信息的历史摘要原始文本要么入库要么删除。实际开源社区里也广泛采用类似策略。5.4 检索召回不准关键词修改策略Search Agent返回不相关文献是高频事故。我调试后发现最常出现的问题是用户停留在泛泛而谈比如“AI in education”检索出来的都是各种大杂烩。解药是让Search Agent在检索前用一个自我反思步骤第一次检索跑完看看前5篇的标题如果文章相关性低于预期提取现有结果里的共有关键词修正查询串重新检索一次这个“两步检索”会多花一次API调用但能把检索相关性提升不少。5.5 Agent“偷懒”策略大模型最基本的倾向是用最省事的方式解决问题。Supervisor有时会故意跳过Compare直接让Writer基于Review来写综述。因为写综述比做对比分析省力。因此必须硬性约束State中的comparison是writer_agent的必需输入如果comparison为空Supervisor不能分配writer_agent。用数据和规则约束行为而不是仅仅依赖提示词的“苦口婆心”。6. 从Demo到可用几个可以继续扩展的方向6.1 接入真实文献库的路子当前Demo直接用arxiv API这已经能跑通但要好用还需要接数据库层的文献管理。比如你可以把检索结果存入sqlite或postgres然后支持用户管理文献库、打标签、标记星标。实际项目里可以在本地维护一个论文元数据表包含标题、作者、年份、摘要、引注等字段。每次检索先用本地表搜命中率不足再走外部API。6.2 加入人工审核节点科研文献对准确性的要求很高AI生成的综述不能直接投出去。除了前面提到的论文选择环节冒出了人工确认我实操里发现“比较结论”和“综述初稿”两处也值得做人工审核。LangGraph的做法是再加两个interrupt_before断点。每次人工审核后update_state然后继续执行下游节点。这比生成完一次性人工全部审核结果发现第一步就错了导致整篇重做高效得多。6.3 多Agent协同再进阶的路线这个项目的后续版本我计划在演进方向上做三件事。第一让Review Agent具备自动抽取论文的数值结果并生成可复现的对比图表。目前数据都躺在JSON里后续可以输出pandas DataFrame再自动绘图。第二引入“批判性Agent”。它可以专门去Review已有结果里是否存在逻辑漏洞或引用错误。相当于给自己加一个审稿人。第三用更细粒度的工具扩展让Search Agent之间接入更多外部数据源比如预印本、数据集公开库而不只拘泥于论文全文。这样它可以覆盖端到端科研辅助的更多场景。最后说点实际的体会把LangGraph落地到科研文献场景最大的感受是框架学起来其实不难难的是对流程的设计和对状态的治理。真正花时间的不是在写Agent的提示词而是在回答“这个Agent的输出到底是谁在消费格式怎么约定失败了怎么回退”这些问题。做这个项目的过程中我反复调整的是Supervisor的决策逻辑。它既不能太死板把所有环节都硬走一遍又不能太自由动不动就跳过关键步骤。目前用的“已完成任务列表必要条件检查”在稳定性和灵活性之间算是一个比较平衡的落点。第一期的内容就先到这里。这个系列接下来我会逐个拆解各环节的优化方法检索质量调优、精读笔记的抽取稳定性、对比分析层面的Prompt策略、以及如何把系统做成一个可实际交付给实验团队使用的服务。每个环节单独展开都能讲很多实操细节我们后面一篇一篇聊。
阅读完成 · 觉得有帮助？
