2026/10/23 11:55:03

用LangGraph构建多智能体AI研究助手:架构设计与实践

用LangGraph构建多智能体AI研究助手:架构设计与实践 我最早产生“用 LangGraph 自己搭一个研究助手”这个念头是在一次被深度调研折磨到崩溃的深夜。当时要给一个项目做技术选型调研光是收集资料就花了两天读了大几十篇文档和讨论帖最后整理出来的报告依然觉得漏了很多东西。那时候市面上其实已经有不少“AI 搜索助手”类的产品有的能自动联网检索有的能生成报告但用下来的感觉总是不对劲——要么太封闭你不能控制它的搜索策略和思考过程要么太脆弱对话稍微长一点就开始丢上下文要么干脆就是套了个壳本质还是单次问答根本没有“做研究”该有的那种渐进式探索和迭代深化。后来我开始调研 LangGraph越看越觉得这东西才对路子。它不是一个现成的问答机器人而是一个工作流编排框架——你不用被 LLM 的单轮对话能力卡住而是可以把“研究”这件事拆成一个个带状态的节点规划、搜索、阅读、分析、写作、校验。每个节点可以由一个专门的大模型指令控制节点之间用图结构连接支持条件分支、循环重试、并行执行。换句话说你不再是写一段 prompt 调一次 API而是在搭建一条完整的研究流水线。这篇文章我就完整分享一下我是怎么从零用 LangGraph 实现一个专业 AI 研究助手的包括架构设计的思路、完整可运行的代码、以及我在实际调试中踩过的那些坑。1. 为什么研究助手必须做成多智能体结构有朋友问过我一个问题现在 GPT 类模型调用联网搜索的能力已经很强了为什么还要用多智能体直接在系统提示词里要求模型“请搜索后再回答”不就行了吗说实话单智能体方案在简单场景下确实够用。但一旦落到“做研究”这种任务上差距立刻就出来了。先看单智能体面临的问题。研究这个动作天然是分阶段的一开始你对问题域一无所知需要先做宽泛的探索找到关键词、概念边界和主要流派然后要深入具体资料提炼核心论点和关键数据接着要把不同来源的信息交叉验证判断哪些可信哪些不可信最后才是组织成结构化的报告。这些阶段的信息处理方式完全不同对模型的上下文窗口占用、推理深度、工具调用频率的要求也完全不一样。如果这些任务全部压在一个对话 session 里完成很快就撞上两个瓶颈一是 token 消耗爆炸二是指令冲突严重——既要模型保持“探索心态”去搜资料又要它进入“批判性思维”模式去交叉验证还要它切换成“严谨学术语体”去写报告单条系统提示词根本顾不过来。多智能体方案解决的就是这个“角色分离”的问题。每个智能体只负责一个稳定的子任务拥有自己独立的系统提示词、自己的工作记忆和工具调用逻辑。规划智能体不用写报告写作智能体不用碰搜索工具这样每条 prompt 都能做到高度专项化模型不需要在多重身份之间反复横跳输出质量和稳定性自然就上去了。第二个关键点在于 Graph 这个结构本身。LangGraph 的核心抽象是“图”节点是智能体或工具调用边是流转逻辑。这个抽象之所以适合做研究流程是因为研究本来就不是一步到位的线性过程。一次搜索往往无法覆盖复杂问题你需要根据初步结果判断“要不要补充搜索”阅读完几篇文档后可能需要重新规划方向分析过程中发现了新的关键词又得回到搜索节点。这些都是典型的分支、循环、回退逻辑。如果用传统链式流程Chain来做就得把这种回环逻辑硬编码成状态机的判断代码但用 LangGraph你只需要定义节点和条件边图本身就把这种动态路由能力天然支持了。还有一个现实原因促使我必须用多智能体——并行需求。一个合格的调研报告往往需要同时覆盖多个子主题。比如研究“大模型在边缘设备上的推理优化”至少涉及模型压缩、硬件加速、量化算法、推理框架好几个方向。如果让一个智能体串行去查时间会拖得很长。多智能体架构天然可以把这个探索过程并行化——每个子主题分配一个搜索分析 Agent单独跑一条路径最后汇总到主 Agent 做整合。这种并行能力既提升了速度也保证了每个子主题的调研深度不会因为前面的环节超时而被牺牲。2. 整体架构设计先画图再写代码动手写代码前我花了不少时间在架构设计上。很多初学者习惯打开编辑器就开始写 prompt 调 API结果写到一半发现流程理不清这里缺个状态那里缺个变量反复重构。LangGraph 的好处恰恰在于它逼着你先想清楚流程——因为节点和边是先在图上画出来的。我的研究助手架构由五个核心 Agent 组成外加两个辅助性的普通节点。2.1 五大核心智能体的职责划分规划智能体Planner Agent。这是整个流程的总指挥负责接收用户输入的研究问题把它拆解为若干个子研究课题。拆解的规则我要求它必须覆盖三个维度问题本身是什么、围绕问题有哪些关键子议题、每个子议题应该用什么搜索关键词去试探。它输出的是一个 JSON 结构包含子任务列表、每个子任务的搜索关键词集合以及任务之间的依赖顺序说明。搜索智能体Researcher Agent。这个对应多个实例因为子任务是并发的——规划完成后每个子任务会触发一个独立的搜索 Agent。它负责调用搜索工具收集搜索结果并抓取相关页面正文。这个 Agent 的指令重点不在于“思考”而在于“稳准狠”——如何构造查询词、如何筛选高价值来源、如何截取页面有效内容而不被噪音带偏。分析智能体Analyst Agent。负责深度阅读搜索返回的文档。它要提炼每个文档的核心观点、关键数据、局限性和与当前研究课题的关系输出结构化笔记。我把分析结果设定为一个列表每个文档对应一个条目标明来源链接、核心观点摘要、证据强度评分和潜在偏差提示。这些笔记是后续写作 Agent 的弹药库。写作智能体Writer Agent。它接收所有分析笔记整合为最终研究报告。它的系统提示词要求非常高必须按研究报告的结构组织要有摘要、正文分级标题、对比分析、结论建议还要求所有观点必须附带上来源引用编号。为了强制这个要求我在后处理环节加了格式校验函数不满足引用规范的输出会被打回重写。校验智能体Reviewer Agent。这是我个人最看重的一个环节也是让我这套系统真正区别于普通“搜索总结”工具的关键。它的职责是模拟一个挑剔的读者检查报告是否存在以下问题事实性错误的风险、子课题之间的衔接断裂、引用标注缺失、结论与证据不匹配。如果发现需要修正的地方它会把报告连同修改意见一起打回写作 Agent 迭代重写。这个“评委”角色的存在对整个系统质量有决定性的提升。2.2 为什么“带状态”的图结构如此关键画完角色分之后第二步就是设计它们的协作方式也就是状态在节点之间的流转规则。LangGraph 的核心概念有两个状态State和流转函数Node / Edge。状态是一个贯穿整张图的共享数据容器任何节点都可以读取和写入。我的状态定义得非常明确包括研究问题、子任务列表、搜索原文列表、分析笔记列表、报告当前草稿、校验意见、最大迭代次数等字段。每个节点处理完毕后往状态里写入自己的产出物图引擎根据条件边决定下一步该唤醒谁。为什么“状态驱动”这件事这么重要因为研究过程最忌讳的是失忆。单智能体对话聊到后来经常忘记前文说过什么经常出现结论引用了一个前面没有出现过的信息源。但在图结构里所有信息都在 State 里沉淀着每个节点只基于状态中的数据进行工作不会出现信息凭空消失的问题。在实际构造图的时候我用到了三种边普通边表示固定顺序流转比如规划完成之后必然进入搜索阶段“条件边”则根据当前状态决定走哪个分支比如校验未通过就走回写作 Agent还有一种是终点判断当连续迭代超过设定次数或者校验通过就进入结束节点。2.3 兜底机制失败处理和无限循环防御任何跟外部工具打交道的系统都必须面对失败。搜索 API 可能超时、某篇网页可能返回 404、大模型输出可能是截断的。这套架构里我每个节点都做了异常捕获搜索失败就换个关键词重试模型输出 JSON 解析失败就用正则提取代码块重试如果连续重试两次依然失败就把当前任务标记为“降级任务”跳过它但记入最终报告的“调研限制说明”部分。无限循环是图结构编排最大的隐性风险。最经典的坑是写作 Agent 和校验 Agent 互相不服气——写作改完校验还是觉得不行打回再改改完再打回最坏情况下两个 Agent 会一直迭代到 token 耗尽甚至死循环。我的解法是给状态加了一个 revision_count 字段每次打回写作节点时 1一旦超过上限、也就是最多允许打回两次就由条件边强制走向“输出报告”节点只是带上一个提示说明“该版本未通过终审请注意风险”。这个兜底机制救了我很多次。3. 环境准备与工程化要点架构想清楚了接下来就是动手搭环境。这里我尽量给到贴脸级别的细节包括版本和具体安装命令。3.1 依赖安装与版本选型我用的 Python 环境是 3.10 版本虚拟环境工具用的 conda但这套东西也可以直接用 venv没有特殊依赖。核心依赖一共四个类别LangGraph 本体负责图编排LangChain 生态提供 Prompt 模版和模型接入Tavily 负责提供搜索 API另外还要一个 SSE 流式输出相关的库用于终端展示进度。conda create -n research_assistant python3.10 conda activate research_assistant pip install langgraph langchain-openai langchain-community pip install tavily-python pip install python-dotenv关于模型选择我个人实际用的是 OpenAI 的接口因为 LangChain 的适配层封得很好切换成本低。文中代码示例统一用gpt-4o和gpt-4o-mini做模型调度——需要深度推理的节点用大模型比如规划、写作、校验偏机械执行的节点用 mini 版本比如单篇文档的摘要抽取因为这类任务逻辑简单用小模型能省不少成本实际效果差异也不大。还有一个必不可少的环节是设置环境变量。在项目根目录建一个.env文件写入需要的所有密钥OPENAI_API_KEYyour_openai_api_key TAVILY_API_KEYyour_tavily_api_key然后在入口文件顶部用dotenv加载。这里有个经验之谈密钥千万别写进代码文件里更别推到仓库我当时为了省事把 key 直接硬编码到代码里后来调试过程中不小心分享过屏幕结果半小时内就被盗刷了一笔从那以后再也不敢了。3.2 Tavily 搜索工具接入搜索工具我用的 Tavily原因很直接它专门为 AI Agent 场景设计返回的是已经清洗过的网页内容不用自己再写页面抓取和正文抽取的代码。在 LangChain 里面TavilySearchResults 这个类已经封装好了可以直接当工具挂给 Agent。初始化方式是tavily TavilySearchResults(max_results5)参数控制每次搜索返回的文档数量。我实测下来做研究调研时每次返回 5 篇文档是一个性价比比较高的值——太少了经常漏信息太多了模型在处理时又难免注意力稀释。关于搜索结果的正文长度问题我的做法是只保留前 2500 个字符。原始网页内容往往非常长如果全量塞给模型上下文窗口会被无关导航和广告文案快速挤占。LangGraph 的状态设计里文档是逐条迭代处理的每次只处理一段文档的摘要这样上下文压力会小很多。3.3 LangGraph 版本适配心得写这篇文章的时候LangGraph 的 API 又经历了一次迭代现在还有一些新的写法在逐渐推广。我代码里使用的是经典的StateGraph对象加上add_node/add_edge/add_conditional_edges这套稳定 API也是目前教程资料最多的写法。如果你属于追新版本的人看到新版的Graph对象要知道那是同一个框架的下一代接口底层逻辑没有变化状态、节点、边的概念完全一致只是写法和一些工具函数的命名做了调整。我给的代码在新版下稍作适配就可以跑核心思路不过时。4. 核心代码实现节点构建与图编排现在进入正题核心代码。按照工程化的拆法我把代码分成五个模块状态定义、节点函数、图构建、主入口调用、辅助工具函数。每个模块我讲解关键逻辑同时附上完整代码。4.1 状态类定义所有节点的共享内存from typing import TypedDict, List, Dict, Optional, Annotated import operator class ResearchState(TypedDict): # 用户输入的研究问题 research_topic: str # 规划智能体输出的子任务列表 subtasks: List[str] # 每个子任务对应的搜索关键词 search_queries: Dict[str, List[str]] # 搜索后收集到的原始文档含标题、链接、正文片段 documents: List[Dict] # 分析智能体输出的结构化笔记 analysis_notes: List[Dict] # 写作智能体产出的报告草稿 report_draft: str # 校验智能体的修改意见和反馈 review_feedback: str # 记录图已经执行到第几个节点用于日志分析 node_log: List[str] # 打回复写次数计数器 revision_count: int # 是否已经完成最终校验 passed_review: bool这里有一个 LangGraph 推荐的分层聚合写法特别提一下。比如 documents 这个字段会被多个搜索 Agent 实例同时写入需要合并而不是覆盖。此时可以在类型定义时用Annotated[list, operator.add]这样的写法LangGraph 会自动用累加操作符把多个源写入的列表拼接起来。实际使用里面我把 documents 定义成Annotated[List[Dict], operator.add]这样多个并行搜索 Agent 的结果就会自动汇合在一起不用手写 merge 函数。4.2 Planner 节点从研究问题到子任务拆分from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI import json planner_prompt ChatPromptTemplate.from_messages([ (system, 你是一位严谨的学术研究规划专家。用户的原始问题是{topic} 你需要将这个问题拆解为多个关键子任务。请以严格的JSON格式输出格式如下 {{ subtasks: [ {{ id: 1, title: 子任务的标题, objective: 这个子任务要解决的具体研究目标, search_keywords: [关键词1, 关键词2, 关键词3] }} ], recommended_order: [子任务1, 子任务2] }} 要求 1. 子任务数量至少3个。 2. 搜索关键词必须具体、可执行不要过于宽泛。 3. 考虑问题中可能存在的分支方向和互相矛盾的观点确保子任务覆盖足够的视角。 ) ]) def planner_node(state: ResearchState) - dict: model ChatOpenAI(modelgpt-4o, temperature0.2) chain planner_prompt | model response chain.invoke({topic: state[research_topic]}) raw_text response.content # LLM 输出偶尔会带 markdown 代码块标记这里做一层格式清洗 if in raw_text: raw_text raw_text.strip(json).replace(, ).strip() try: parsed json.loads(raw_text) except json.JSONDecodeError: # 如果解析失败重新发一次把只输出JSON强调得更加严格 strict_prompt ChatPromptTemplate.from_messages([ (system, 只输出一个有效的JSON对象不要输出任何其他内容包括markdown标记。) ]) chain strict_prompt | model response chain.invoke({input: raw_text}) raw_text response.content.replace(json, ).replace(, ).strip() parsed json.loads(raw_text) subtask_titles [t[title] for t in parsed[subtasks]] search_queries {t[title]: t[search_keywords] for t in parsed[subtasks]} return { subtasks: subtask_titles, search_queries: search_queries, node_log: state[node_log] [planner] }规划节点的核心是“JSON 输出清洗”。这一步我必须强调因为这是做 LLM Agent 最容易踩坑的地方没有之一。大模型嘴上答应输出 JSON实际很多时候它会把 JSON 包在代码块标记里或者前面加一句“好的这是你的 JSON”这些杂质不处理掉 json.loads 必然报错。我处理了两层先剥掉代码块标记再剥掉前后说明性的文字。如果还是解析失败我就让模型重新输出一次但把系统提示词换成“只输出 JSON不要任何多余内容”——这两次重试基本能解决九成以上的异常情况。4.3 Researcher 节点并行搜索与内容抓取from langchain_community.tools import TavilySearchResults def researcher_node(state: ResearchState) - dict: tavily TavilySearchResults(max_results4) documents list(state.get(documents, [])) # 遍历规划好的每个子任务的每个关键词逐个执行搜索 for subtask in state[subtasks]: for keyword in state[search_queries].get(subtask, []): try: results tavily.invoke(keyword) for res in results: # 参数topic用来记录这篇文档归属于哪个子任务后续分析时有用 documents.append({ subtask: subtask, keyword: keyword, title: res.get(title, ), url: res.get(url, ), content: res.get(content, )[:2500], }) except Exception as e: # 搜索失败不致命记录下来继续下一个关键词 documents.append({ subtask: subtask, keyword: keyword, title: f搜索失败: {e}, url: , content: }) return {documents: documents, node_log: state[node_log] [researcher]}关于并行搜索我要说明一下上面这段代码为了确保读者能直接跑通我用了串行循环的执行方式。实际生产环境中每个子任务的搜索循环是可以并行化的——用 Python 的concurrent.futures.ThreadPoolExecutor或者 asyncio 并发去跑多个 Tavily 调用时间可以从分钟级降到十几秒。我后期的版本就是这么干的因为 Tavily API 本身是网络 IO 操作线程池带来的速度提升非常可观。不过并行版本有一个注意点多个线程同时往 documents 列表里 append 会出现并发竞争问题。建议用线程池返回值汇总后再合并或者用我们前面提到的Annotated[List, operator.add]让 LangGraph 自动合并多个节点的输出。这个细节就是 LangGraph 这套框架设计得比较聪明的地方。4.4 Analyst 节点逐篇文档深度阅读analyst_prompt ChatPromptTemplate.from_messages([ (system, 你是研究助理。请阅读下面这篇文档并提炼出与当前研究课题关联最紧密的信息。用中文输出JOSON格式 {{ doc_title: 文档标题, doc_url: 原始链接, core_idea: 这篇文档的核心论点是什么用2-3句话概括, key_evidence: [两个关键证据或数据如果有的话], limitations: [这篇文档的潜在偏差或局限如果没有则写[]], relevance_score: 0-10之间的数字表示与研究的关联度 }} 只输出JSON。) ]) def analyst_node(state: ResearchState) - dict: model ChatOpenAI(modelgpt-4o-mini, temperature0.1) analysis_notes [] for doc in state.get(documents, []): if not doc.get(content): continue # 构造单文档分析的输入提示里面附带了研究主问题让模型有参照物 chain_input { input: doc[content], research_topic: state[research_topic] } try: response model.invoke(analyst_prompt.format_messages(**chain_input)) raw_text response.content.replace(json, ).replace(, ).strip() parsed json.loads(raw_text) parsed[subtask] doc[subtask] analysis_notes.append(parsed) except Exception as e: # 单篇文档解析失败跳过但记录 analysis_notes.append({ doc_title: doc.get(title, ), doc_url: doc.get(url, ), core_idea: f解析失败: {str(e)}, key_evidence: [], limitations: [该文档在分析节点未能成功处理], relevance_score: 0 }) return {analysis_notes: analysis_notes, node_log: state[node_log] [analyst]}分析节点用 gpt-4o-mini 是刻意选择因为这里的任务模式稳定、目标单一输入一篇文档输出结构固定的笔记。它不需要深度推理需要的是快速且省成本。我在实测中发现即使换成更大的模型这个环节的产出质量也不会有质的提升。真正影响输出质量的变量是文档正文的截断长度——2500 个字符对一篇信息密度高的技术文章来说可能不够很多关键细节都在后半截。这里我后来加了一个启发式策略如果文档正文一眼看过去超过 2500 个字符就截取开头 1500 加上结尾 1000因为多数文章的开头是摘要和背景结尾是结论都比被埋在中间的具体实施细节对研究摘要更有用。4.5 Writer 节点从笔记到完整报告writer_prompt ChatPromptTemplate.from_messages([ (system, 你是一名资深研究分析师。下面是一组关于研究课题{topic}的结构化分析笔记可能来自多个主题方向。 请基于这些笔记撰写一份完整的研究报告。要求 1. 报告必须包括 - 课题简介与研究目标 - 关键发现章节使用小标题分块 - 子专题对比分析表格 - 存在的矛盾观点或不确定性 - 参考资料列表用[数字]标注在正文中并统一列在末尾 2. 所有关键论述必须引用笔记中的来源链接引用格式为[资料1][资料2]。 3. 报告中出现的所有观点必须来自分析笔记不要凭空增加笔记中没有的信息。 4. 语言风格专业、克制、信息密度高避免浮夸宣传性用语。 ), (user, 分析笔记如下\n{notes}) ]) def writer_node(state: ResearchState) - dict: model ChatOpenAI(modelgpt-4o, temperature0.3) notes_json json.dumps(state.get(analysis_notes, []), ensure_asciiFalse, indent2) response model.invoke(writer_prompt.format_messages( topicstate[research_topic], notesnotes_json )) new_report response.content # 在这里做了一个本地规则兜底如果报告里完全没有出现参考资料标记打回重写一次 if [资料 not in new_report: forcing_prompt 你遗漏了分析笔记中的参考资料引用。请务必在正文中使用[资料编号]格式标注来源并在报告末尾列出完整参考列表。请重新撰写。 response model.invoke([ {role: system, content: forcing_prompt writer_prompt.messages[0].content}, {role: user, content: notes_json} ]) new_report response.content return {report_draft: new_report, node_log: state[node_log] [writer]}写作节点的核心因子是把可用信息卷到极致。分析笔记里每一条都包含了 doc_url 和 relevance_score模型需要具备足够的信息密度才能写出一篇有对比、有侧重、有来源的报告。这里有两个容易忽略的点。一个是引用格式的统一问题。我要求的格式是[资料1]这种方式而不是 markdown 链接是刻意为之。因为[资料N]格式在纯文本里面最稳定模型生成失误率远低于它在正文中正确生成超链接的概率。报告最后我会放一个“参考资料”章节把编号与 URL 一一对应这样人工速查和脚本校验都很方便。另一个是强制重写逻辑。如果草稿里一个引用标记都没有说明模型大概率偷懒了直接重写是经济的选择。这种本地规则 LLM 判断的双重保险是我在这套系统里面反复使用的模式——把 LLM 最不擅长的一致性问题用确定性代码兜住。4.6 Reviewer 节点模拟权威评审的打回逻辑reviewer_prompt ChatPromptTemplate.from_messages([ (system, 你是一名挑剔的研究报告评审专家。你的任务是对草稿提出具体的修改意见。 检查以下维度 1. 事实性风险报告中的观点是否有足够的资料保证 2. 结构完整性是否覆盖了所有子课题有没有明显空洞 3. 引用规范性是否每条关键观点都有[资料N]标注引用列表是否完整 4. 连贯性章节之间是否有逻辑跳跃 请输出JSON {{ passed: true或false, issues: [问题1, 问题2], suggestions: [建议1, 建议2] }} ), (user, 研究报告草稿\n{draft}\n\n原始研究问题{topic}) ]) def reviewer_node(state: ResearchState) - dict: model ChatOpenAI(modelgpt-4o, temperature0.2) response model.invoke(reviewer_prompt.format_messages( draftstate.get(report_draft, ), topicstate[research_topic] )) raw_text response.content.replace(json, ).replace(, ).strip() parsed json.loads(raw_text) # 本地兜底规则报告没有主体内容直接打回 if len(state.get(report_draft, )) 500: parsed[passed] False parsed[issues].append(报告生成失败或内容过短) return { review_feedback: json.dumps(parsed, ensure_asciiFalse), passed_review: parsed[passed], revision_count: state[revision_count] 1 if not parsed[passed] else state[revision_count], node_log: state[node_log] [reviewer] }评审节点的核心价值在于“复审回路”。它和写作节点组成的循环是整个架构里最接近人类真实研究流程的地方——不经过几轮修改的报告拿不出手。在流程控制上我用条件边来做路由判断passed_review为 True 就走结束节点为 False 则检查revision_count如果当前次数已经超过 2直接放行但有警告标记如果没有超限继续回写作节点。这里还有一个细节值得分享我在 reviewer 的 system 里要求每条问题写“具体可执行的修改建议”而不是“泛泛的问题描述”。实践中发现写作节点面对“内容不够深入”这种模糊意见时改出来的版本往往和原来一个样。但面对“第二章节缺少与 LLaMA 量化方案的对比数据需要补充 1-2 篇该方向的资料来分析”这种精确指令时改写效果立竿见影。4.7 图构建与主程序入口把所有节点串成流水线from langgraph.graph import StateGraph, END from langgraph.graph.graph import START def build_graph(): graph StateGraph(ResearchState) # 注册所有节点 graph.add_node(planner, planner_node) graph.add_node(researcher, researcher_node) graph.add_node(analyst, analyst_node) graph.add_node(writer, writer_node) graph.add_node(reviewer, reviewer_node) # 定死主线顺序规划 - 搜索 - 分析 - 写作 - 评审 graph.add_edge(START, planner) graph.add_edge(planner, researcher) graph.add_edge(researcher, analyst) graph.add_edge(analyst, writer) # 条件边评审未通过且修订次数允许就回写作否则到 END graph.add_conditional_edges( reviewer, lambda state: rewrite if (not state[passed_review] and state[revision_count] 2) else finish, { rewrite: writer, finish: END } ) return graph.compile()主入口调用逻辑def run_research(topic: str) - str: app build_graph() initial_state { research_topic: topic, subtasks: [], search_queries: {}, documents: [], analysis_notes: [], report_draft: , review_feedback: , node_log: [], revision_count: 0, passed_review: False, } final_state app.invoke(initial_state) # 组装最终结果返回报告正文 调研元信息 result { report: final_state[report_draft], review_feedback: final_state[review_feedback], retry_count: final_state[revision_count], sources_count: len(final_state.get(documents, [])), trace_log: final_state[node_log] } return result if __name__ __main__: ex_topic 多模态大模型在医疗影像分析中的应用现状 result run_research(ex_topic) print(result[report])启动之后它会按 planner - researcher - analyst - writer - reviewer -可能回环 - END 的顺序执行。这里提醒一下第一次跑的朋友整个过程因为涉及多次大模型调用和搜索请求跑完一轮研究可能需要五到十分钟控制台看起来像是卡住了其实不是是在按序执行节点。我建议在节点函数中加入print日志或者接一个简单的进度显示至少能知道现在跑到哪一步了。后面我贴出的 node_log 字段就是为了这个目的服务的——每次节点返回时把自己追加到日志里最后可以从 final_state 里取出完整的执行路径调试时非常有用。5. 真实场景实测让系统跑一个完整研究问题为了验证这套系统确实能干活我拿“多模态大模型在医疗影像分析中的应用现状”这个具体问题跑了一遍完整流程。以下是我当时实际记录的执行过程与产出的关键标注。规划节点输出的子任务结构分成了四路主流模型技术路线比较几种多模态架构在影像分析上的技术方案与评测指标数据资源与标注现状医疗影像数据集的规模、标注成本与可得性落地部署挑战模型在临床环境部署时的算力、隐私和监管问题尚存在的争议与局限性论文中反复提到的假阳性、黑盒问题和临床信任问题四路子任务对应了“技术-数据-工程-社会影响”四个层面覆盖面比我最初手动想的调研大纲还要完整。搜索阶段一共触发了 10 个关键词收集了 25 条文档。分析节点最终产出 18 条有效结构化笔记另外有 3 条因为搜索失败或内容为空被标记为降级。写作节点生成报告用时约 40 秒生成的报告大概 1800 字结构上有摘要、现状分析、四个子专题章节和一个参考列表。第一次评审打回了这份报告主要问题是“第三章节的临床落地挑战部分引用的资料只有一倍来源且缺少与真实临床案例的对照”。打回后写作节点按意见补充了两篇部署相关的资料分析重新梳理了该章节的论证结构。第二次评审通过最终报告带 28 条参考资料索引完整度基本达到了能直接参赛的水准。对了远程跑这个流程的时候注意时刻关注 API 花费。我用 gpt-4o 跑规划、写作、评审用 gpt-4o-mini 跑文档分析这一整套流程走下来所有 token 加起来的一次成本大约是两块钱人民币左右。考虑到产出报告质量和人工搜集对比这个成本是可以接受的。付费 API 没有余额时记得及时充值或处理账单。6. 常见问题与排查技巧实录这套系统跑通以后我整理了一份踩坑笔记基本覆盖了新手最常遇到的问题。6.1 JSON 解析失败是头号敌人任何在 LangGraph 里扔json.loads()的人都免不了被模型输出搞崩几次。模型在系统提示词里答应得好好的“只输出 JSON”实际输出时还是会时不时地带上 markdown 标记、注释、甚至“当然可以”这类废话前缀。我的处理办法有三层先用字符串替换清掉代码块标记再尝试直接解析失败以后把提示词换成“只输出一个 JSON 对象”重新调一次再失败就跳过该条文档并降级标记。这套电脑兜底之后整体流程的稳定性从 85% 提到了 98% 以上。6.2 上下文填满与 token 超限研究任务天然消耗 token。文档正文、分析笔记、报告草稿都会撑大上下文经常做到第三个子任务的时候模型就报 context length。我给出的三个解决方案是文档正文按需截断到 2500 字符分析笔记每个条目用 gpt-4o-mini 生成并只保留关键字段不保留原文散文写作阶段重新聚合笔记只把与当前报告最相关的 notes 传给 Writer其余的信息源文档留在 State 里不进入 Prompt。这一套做下来token 消耗降低了约六成。6.3 模型对话回环死循环前面提到过写作与评审之间偶尔会卡死。这个问题在调试文本格式问题时最常见比如写作 Agent 一直输出 markdown 表格方式评审 Agent 一直要求换一种格式双方互不退让能迭代到世界末日。我的上限策略是 revision_count 一到 2 次必须放行宁要一份有瑕疵的报告也不要一套死循环跑不完的流程。另外还有一个技巧每个节点函数的第一行就打日志追踪到具体哪一轮迭代出问题方便对着报文判断是写法问题还是内容问题。6.4 多智能体并发下的状态冲突LangGraph 对状态更新的处理默认情况下是整体替换如果有两个并行分支同时往同一个字段写值后写的一方会覆盖先写的一方。所以并行搜索场景我强烈建议把聚合字段定义成Annotated[List, operator.add]这样每次写入都会走累加逻辑而不是覆盖。此外每个节点最好只修改自己关心的字段不要顺手把别的节点的中间产物也改掉否则并行分支之间会互相污染。7. 进阶优化让研究助手更“专业”的三个方向基础版本能跑通以后我后续做了以下几方面的增强按收益从高到低排个序你可以根据自己的需要挑着做。第一个方向是把搜索策略升级成“两阶段检索”。第一次搜索发现关键词不匹配时基于初步结果做一轮二次关键词生成再搜。这套系统能有效应对冷门话题——原来的版本经常因为初始关键词不对导致搜索结果跟主题相关性差后来加了二次检索覆盖率提升不少。第二个方向是把状态扩展为带记忆的持久化结构。LangGraph 有专门用于持久化的接口把状态存到数据库里面。这样可以做到一次询问生成报告后用户能够追问“关于第三部分再深入一点”系统能定位到对应的中期状态并根据既有资料继续研究而不是所有任务都从头跑一遍。第三个方向是让报告输出支持多种格式比如 markdown、PDF、幻灯片结构。写作节点的尾部接一个格式转换节点用报告原文生成摘要版、演示版或图文版。这一层虽然不算核心技术但对下游消费报告的人来说体验差异是巨大的。此外还有一个小技巧对提升报告可信度很有帮助在搜索节点里对每个域名做来源类型标记。学术域名的检索结果权重高商业博客次之论坛讨论仅作参考。把这些可验证的来源信息写入笔记写作节点就拥有了“哪些证据更可靠”的判别依据写出来的报告也不再是一锅乱炖。我自己实际用下来最大的体会是LangGraph 构建多智能体系统时最核心的并不是搭好节点和边而是想清楚“状态里到底放什么”。所有智能体的协作本质上是状态的生产与消费——规划 Agent 产生子任务搜索 Agent 消费关键词、产生文档分析 Agent 消费文档、产生笔记……每个节点都是状态转化器。你把状态字段定得越清晰每一个 Agent 的职责就越容易写清楚后续调试和扩展的路也越顺。作为一个喜欢把主动权握在自己手里的开发者用 LangGraph 搭建这套研究助手给我最大的感触是“终于不用被黑盒 API 牵着走了”。从规划路径到搜索策略到评审标准每一步都可以自定义出了任何问题都能顺着 node_log 一路追查。如果你现在也面临经常需要做大量资料调研的工作非常推荐照着这套框架动手搭一个属于自己的研究助手先在本地小规模跑通再根据自己关注的领域把搜索源、评审标准和报告模板逐步调成自己工作的形状。这个过程本身就是一次很有意思的多智能体工程实践。