☰
LangGraph多智能体协作:从零搭建科研文献智能助手
2026/10/21 21:54:07 网站建设 项目流程

做科研的人都知道,读论文、建综述、跟踪前沿,这项工作本身有一套完整的流水线。但市场上很多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,下面挂四个Worker:

  • Search 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直接操作大模型API
arxiv学术论文元数据检索
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. 从零构建科研文献助手的四个Worker

3.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( query=query, max_results=max_results, sort_by=arxiv.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, k=top_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(model="gpt-4o-mini", temperature=0.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决策前设置一个breakpoint:

app = graph.compile( checkpointer=checkpointer, 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(checkpointer=YOUR_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执行完回到Supervisor,Supervisor重新决策,如果它不判断“已经完成”的条件,就会一直把同一个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在检索前用一个自我反思步骤:

  1. 第一次检索,跑完看看前5篇的标题
  2. 如果文章相关性低于预期,提取现有结果里的共有关键词,修正查询串
  3. 重新检索一次

这个“两步检索”会多花一次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策略、以及如何把系统做成一个可实际交付给实验团队使用的服务。每个环节单独展开,都能讲很多实操细节,我们后面一篇一篇聊。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询