好的收到。我现在就围绕“langgraph的一个例子”这个标题用一线从业者的口吻写一篇能直接落地、有思路有代码有坑点的实操型博文。1. 从一个真实需求聊起为什么要用 LangGraph做 LLM 应用开发的朋友大概率都遇到过类似的尴尬单轮对话的玩具 Demo 跑得风生水起一上复杂的业务逻辑——比如“先查资料、再决定要不要追问、最后汇总答案”——就手忙脚乱。你当然可以用一大堆 if/else 把提示词和模型调用串起来但我试过那种代码维护到第三周你自己都不太想打开看。LangGraph 解决的就是这个问题。它的核心思路不是“再包一层提示词”而是把整个应用流程当成一张图来设计和运行。图里有节点、有边、有条件跳转天然适合描述“先做什么、下一步看情况做什么”这类复杂流程。我最近用 LangGraph 重构了一个内部资料问答助手从设计到跑通体验比之前的链式代码舒服太多。这篇文章就用那个项目作为引子带你把一个完整的 LangGraph 例子拆开揉碎。你不仅能看懂节点、状态、条件边这些概念怎么落地还能直接抄走我调试时踩坑换来的经验。不管是做 RAG 问答、Agent 工具调用还是简单的工作流编排这套思路都能复用。2. LangGraph 的核心概念与前置准备2.1 图和链到底有什么区别很多人问过我一个问题LangChain 里的 Chain 不是也能串流程吗为什么非要上 LangGraph我个人的理解是Chain 适合流程确定的线性管道就像流水线一个工位干完传到下一个。但真实业务哪有那么听话你要根据检索结果的相关性决定走哪条分支或者让模型在回答过程中反复调用工具修正信息这种“动态走向”的流程Chain 表达起来特别别扭你得疯狂塞条件判断、拆子链代码可读性直线下降。LangGraph 的思路完全反过来。它把流程建模成一个状态机所有数据放在一个全局的 State 对象里流转每个节点是一段处理逻辑读取 State、处理、再更新 State节点与节点之间的边界可以是有条件的根据 State 里的某个字段决定下一步去哪。说白了这是一张可以自带环路的流程图想回到上一步重新处理很自然就实现了。2.2 环境安装与版本选择这步没什么悬念直接装包。pip install langgraph langchain langchain-openai我建议不要把 langchain 和 langgraph 的版本搞得过于激进优先选择稳定版本。装完之后可以快速验证一下导入是否正常。from langgraph.graph import StateGraph, END print(StateGraph, END)如果这里报错大概率是版本冲突检查一下是否有多个 langchain 核心包混在一起。顺带提一嘴如果你用的模型是 OpenAI 兼容接口langchain-openai 这个包作为模型层适配器很省事下面例子也是基于它写的。3. 资料问答助手的整体设计思路3.1 需求拆解与状态定义我给自己定的目标是做一个“内部资料问答助手”通过自然语言提问就能从一堆工作文档里找到答案并且答案要附上来源方便核对。这个需求如果拆成流程图核心环节大概是这样接收用户问题。先做一次本地知识库的快速检索拿到候选文档片段。判断这些片段是否足够回答用户问题如果不够就调用一个外部工具比如 Web 搜索或者另一个文档库的 API补充信息。最后基于所有可用信息生成最终回复。这个流程最大的特点就是中间有分支而且可能发生“检索不够 → 补充检索 → 判断够不够 → 生成”的循环。用 LangGraph 来表达就非常契合。状态是整个图的心脏。我定义了这样一个 Statefrom typing import TypedDict, List class QAState(TypedDict): question: str # 用户原始问题 retrieved_docs: List[str] # 本地检索到的文档片段 search_query: str # 是否需要外部检索如果需要记录关键词 need_external: bool # 分支条件控制是否需要外部工具 answer: str # 最终生成的答案 tried_retrieve: int # 记录已经外部检索了几次防止死循环这里你可能会问为什么分支条件要单独用一个 bool 字段不直接在节点里 return 一个路径字符串就行吗其实 LangGraph 两种方式都支持但我更倾向于把判断依据写进状态里。这样做的好处是方便调试整个图跑完之后你从 State 里就能完整还原每一步决策依据也方便后续做日志审计。3.2 节点和边的规划我一开始画的图很朴素只有四个节点retrieve本地检索拿相关文档片段。should_search判断是否需要外部搜索这是个带条件的出口。external_search调用外部工具补充信息。generate生成最终答案。节点之间怎么走我用条件边来控制。最开始从 START 进入retrieveretrieve结束之后根据 State 里的need_external字段决定是直接去generate还是先去external_search。而external_search执行完不能让它直接跑到generate还要再回到should_search判断一次这样就形成了一条环能够反复检索直到信息足够。不过这里藏了一个风险如果外部搜索永远返回不够整个流程就会无限循环下去。所以我在 State 里放了一个tried_retrieve计数器在条件判断里限制最多跑两轮外部检索超过次数就强制生成答案。4. 完整实现与代码拆解4.1 构造检索节点第一个节点是普通文档检索。这里我刻意做得简化用关键词匹配模拟向量召回让你先看清 LangGraph 的骨架。真实项目里换成向量库的相似度检索也很容易逻辑就发生在节点内部。fake_docs_db { 网络连接: 设备连接 Wi-Fi 失败时先确认密码是否正确然后重启路由器和网卡。, 打印机故障: 打印机不工作检查驱动版本、数据线和默认打印机设置。, 账号登录: 账号无法登录优先检查网络状态然后尝试找回密码。, } def retrieve_node(state: QAState) - QAState: question state[question] retrieved [] for keyword, doc in fake_docs_db.items(): if keyword in question: retrieved.append(doc) if not retrieved: retrieved.append(本地资料库中未找到直接相关内容需要查看外部资源补充。) # 模拟一个简单的相关性打分这里直接用是否有检索结果来判断 has_result any(未找到 not in d for d in retrieved) return { retrieved_docs: retrieved, need_external: not has_result, }这个节点的输出就是一段新的 State 字典。你需要注意LangGraph 的节点返回的字段会和原有 State 做合并所以这里只需要返回要更新的字段不用把question原样传回来。我第一次写的时候把整个 State 都 return 了结果没问题但代码很冗余而且容易覆盖掉别的节点刚更新的字段。4.2 条件判断节点与外部搜索节点接下来是条件判断节点它会读取 State 里的need_external和tried_retrieve字段决定图的下一个走向。这里我用的是 LangGraph 的条件边机制节点本身只需要返回一个表示路径的字符串。def should_search_node(state: QAState) - QAState: # 如果需要外部搜索而且尝试次数还没超过上限 if state[need_external] and state[tried_retrieve] 2: return {tried_retrieve: state[tried_retrieve] 1, search_query: state[question]} return {tried_retrieve: state[tried_retrieve], search_query: }仔细看这个节点既承担了“更新计数器”的职责也决定了边往哪走。不过 LangGraph 判断边的走向不回看这个返回值而是在add_conditional_edges里根据 State 自己写判断逻辑所以更清晰的写法是让这个节点只更新状态真正的路径选择在构建图时完成。我建议你养成这样的习惯后面逻辑复杂了调试起来清爽很多。外部搜索节点也简单我这里用一个模拟函数假装调用了外部搜索 APIdef external_search_node(state: QAState) - QAState: external_result 外部搜索补充请检查设备固件是否更新到最新版本然后尝试重新配对。 docs state[retrieved_docs] [external_result] return { retrieved_docs: docs, need_external: False, # 假设外部搜索已经补全了信息 }你可能会觉得这一步太“模拟”了。说实在的开发调试阶段用模拟数据非常值得你可以先不管真实 API 的延迟和报错把图本身的逻辑跑通再接入真实工具。这比一上来就接真 API、出问题都不知道是图写错了还是接口写错了要高效得多。4.3 生成答案节点与整体构图生成节点负责把检索到的文档拼进提示词让模型做总结回复。这里我用了一个很朴素的 Prompt实际项目可以换成你自己的模板from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage llm ChatOpenAI(modelgpt-4o-mini, temperature0) def generate_node(state: QAState) - QAState: context \n\n.join(state[retrieved_docs]) prompt f根据以下资料回答问题。 资料 {context} 问题{state[question]} 要求给出答案并尽量引用资料中的内容。 messages [ SystemMessage(content你是一名乐于助人的技术支持助手。), HumanMessage(contentprompt), ] response llm.invoke(messages) return {answer: response.content}到这里四个节点都齐了。现在把它们组合成图并且接上条件边。这是我最喜欢的一部分因为整个流程的走向几乎一眼就能看明白from langgraph.graph import StateGraph, START, END graph StateGraph(QAState) graph.add_node(retrieve, retrieve_node) graph.add_node(should_search, should_search_node) graph.add_node(external_search, external_search_node) graph.add_node(generate, generate_node) graph.add_edge(START, retrieve) graph.add_edge(retrieve, should_search) graph.add_conditional_edges( should_search, lambda state: external_search if state[need_external] and state[tried_retrieve] 2 else generate, {external_search: external_search, generate: generate}, ) graph.add_edge(external_search, should_search) graph.add_edge(generate, END) app graph.compile()注意看external_search回到should_search这条边它就是循环的关键。第一次外部搜索结束后系统会再回到判断节点如果这时need_external仍然为 True而且尝试次数没超就继续搜如果条件不再满足就走generate。4.4 实际运行与输出效果图编译好之后直接invoke传入初始状态就能跑起来。下面是我调试验证时用到的两个问题一快一慢正好覆盖了有分支和无分支的场景。result1 app.invoke({question: 打印机不工作怎么办}) print(result1[answer]) print(result1[tried_retrieve]) print(result1[need_external])因为“打印机故障”这个关键词直接命中了本地文档库所以need_external一开始就是 False流程只是简单地从retrieve一路走到了generate计数器保持 0正好是一条直线。再看一个复杂的result2 app.invoke({question: 设备一直离线无法连接怎么办}) print(result2[answer]) print(result2[tried_retrieve]) print(result2[need_external])这个问题在本地文档库里完全找不到对应词所以need_external为 True流程进入external_search然后再回到should_search发现信息补上了才走向generate。整个过程走了两轮判断计数器是 1。从运行结果来看图的行为完全符合预期。你可以把tried_retrieve和need_external的最终值打出来和流程图对照每一步去向都清清楚楚。5. 常见报错与避坑经验5.1 条件边路径不匹配LangGraph 报这种错误往往是你add_conditional_edges里的路径字典写错了。比如你映射里写了search: external_search但图里根本没有叫search的节点注册过或者你返回的是个动态字符串没有映射到已存在的节点名。我的排查顺序是这样的先检查节点注册名有没有写错再检查字典里的 key 和判断函数实际返回值是否一致。建议你在should_search_node里临时加一句 print把返回值打出来一眼就能定位。5.2 循环次数不可控这是处理 Agent 类应用最让人头疼的问题。外部工具永远返回“信息不足”的时候图就会一直循环。我一开始天真地以为模型总会适可而止结果跑了几轮才意识到必须加硬限制。后来我的通用方案是在状态里放一个计数器字段每进入一次外部搜索节点就加一条件边的判断里强制加上计数上限。如果循环是因为工具调用本身还可以在节点内部检查累计调用次数超过阈值直接抛一个自定义错误配合回调函数做告警。5.3 状态字段被意外覆盖LangGraph 的 State 默认是一个 TypedDict 共享对象所有节点更新都是原地覆盖。如果你的两个节点都在更新同一个列表字段后执行的节点会直接覆盖前面的内容而不是追加。我当时就遇到过一次检索节点本来已经拿到三条候选资料结果外部搜索节点返回的时候只带了一条新内容把之前的结果整个覆盖掉了。解决办法分两种一是像我在代码里写的那样读取旧列表再拼接新的二是给字段指定一个自定义的归约函数比如operator.add让 LangGraph 自动合并列表。from typing import Annotated import operator class QAState(TypedDict): retrieved_docs: Annotated[List[str], operator.add]不过用operator.add要注意如果上一个节点返回空列表add会把空列表和旧列表合并结果是旧列表不变没问题如果某个节点不小心对同一个字段返回了两次更新就会重复。所以要明确自己想表达的是“追加”还是“替换”。5.4 模型调用超时这个锅不全是 LangGraph 的。外部模型接口不稳定会导致节点卡住整个图停在那里。我的做法是在节点内部控制模型调用的超时时间给llm.invoke设置timeout参数或者用asyncio.wait_for包一层超时就返回一条错误备注作为上下文而不是让整个流程死等。6. 更进一步的玩法与扩展思路6.1 加入人机审核节点很多业务场景下AI 生成的内容不能直接放出去需要人工确认。LangGraph 对这类需求支持得很好可以在generate之后加一个review节点走到的位置时暂停等人为修改或者确认后再继续。这比自己在代码里维护一个“挂起会话”要优雅得多因为图本身记住了所处的状态和上下文你只需要调resume继续执行。6.2 多工具 Agent 编排如果把external_search_node替换成“调用多个工具”的 Agent 节点整个图就变成一个多工具编排的自动规划器。你可以在一个节点里让模型自主决定调什么工具、参数是什么然后用另一个路由节点判断是否继续。LangGraph 官方推荐的 Agent 架构就是这么做的很多复杂工作流也都可以拆成这种模式。6.3 状态持久化与流式输出除了最基本的invokeLangGraph 还支持流式输出stream方法会一截一截返回节点结果特别适合做打字机效果。如果你需要对话历史保持可以给编译后的应用配置 Checkpointer 参数让状态持久化到指定存储这样即使重启进程也还能从上次执行位置继续。7. 我对 LangGraph 的个人感受与建议从最早的链式结构到现在的图式编排这种演进不是单纯“新框架有多炫”而是大家逐步认识到LLM 应用的流程天然是动态的、分支的甚至带着反馈循环。用图来表达流程结构变得透明哪里容易出错一眼就能看出来也方便团队协作。每个人都清楚整体流程的骨架和每个节点的职责边界。如果你是第一次接触 LangGraph我建议不要先去啃特别复杂的概念先把我这个最小例子在本地跑通感受一下状态和边的流转然后把例子里的模拟数据替换成真实向量库、真实工具和真实模型。一边替换一边踩坑这套架构的理解深度会比只读文档扎实得多。我个人后续在这个方向上的规划是把业务里几个常见的流程资料检索问答、工单分类转派、周报辅助生成都慢慢迁移到 LangGraph 上来。状态和节点一旦沉淀成公共模块复用性真的会提升很多。希望这篇例子能给你一个还算顺滑的起点。
好的收到。我现在就围绕“langgraph的一个例子”这个标题用一线从业者的口吻写一篇能直接落地、有思路有代码有坑点的实操型博文。1. 从一个真实需求聊起为什么要用 LangGraph做 LLM 应用开发的朋友大概率都遇到过类似的尴尬单轮对话的玩具 Demo 跑得风生水起一上复杂的业务逻辑——比如“先查资料、再决定要不要追问、最后汇总答案”——就手忙脚乱。你当然可以用一大堆 if/else 把提示词和模型调用串起来但我试过那种代码维护到第三周你自己都不太想打开看。LangGraph 解决的就是这个问题。它的核心思路不是“再包一层提示词”而是把整个应用流程当成一张图来设计和运行。图里有节点、有边、有条件跳转天然适合描述“先做什么、下一步看情况做什么”这类复杂流程。我最近用 LangGraph 重构了一个内部资料问答助手从设计到跑通体验比之前的链式代码舒服太多。这篇文章就用那个项目作为引子带你把一个完整的 LangGraph 例子拆开揉碎。你不仅能看懂节点、状态、条件边这些概念怎么落地还能直接抄走我调试时踩坑换来的经验。不管是做 RAG 问答、Agent 工具调用还是简单的工作流编排这套思路都能复用。2. LangGraph 的核心概念与前置准备2.1 图和链到底有什么区别很多人问过我一个问题LangChain 里的 Chain 不是也能串流程吗为什么非要上 LangGraph我个人的理解是Chain 适合流程确定的线性管道就像流水线一个工位干完传到下一个。但真实业务哪有那么听话你要根据检索结果的相关性决定走哪条分支或者让模型在回答过程中反复调用工具修正信息这种“动态走向”的流程Chain 表达起来特别别扭你得疯狂塞条件判断、拆子链代码可读性直线下降。LangGraph 的思路完全反过来。它把流程建模成一个状态机所有数据放在一个全局的 State 对象里流转每个节点是一段处理逻辑读取 State、处理、再更新 State节点与节点之间的边界可以是有条件的根据 State 里的某个字段决定下一步去哪。说白了这是一张可以自带环路的流程图想回到上一步重新处理很自然就实现了。2.2 环境安装与版本选择这步没什么悬念直接装包。pip install langgraph langchain langchain-openai我建议不要把 langchain 和 langgraph 的版本搞得过于激进优先选择稳定版本。装完之后可以快速验证一下导入是否正常。from langgraph.graph import StateGraph, END print(StateGraph, END)如果这里报错大概率是版本冲突检查一下是否有多个 langchain 核心包混在一起。顺带提一嘴如果你用的模型是 OpenAI 兼容接口langchain-openai 这个包作为模型层适配器很省事下面例子也是基于它写的。3. 资料问答助手的整体设计思路3.1 需求拆解与状态定义我给自己定的目标是做一个“内部资料问答助手”通过自然语言提问就能从一堆工作文档里找到答案并且答案要附上来源方便核对。这个需求如果拆成流程图核心环节大概是这样接收用户问题。先做一次本地知识库的快速检索拿到候选文档片段。判断这些片段是否足够回答用户问题如果不够就调用一个外部工具比如 Web 搜索或者另一个文档库的 API补充信息。最后基于所有可用信息生成最终回复。这个流程最大的特点就是中间有分支而且可能发生“检索不够 → 补充检索 → 判断够不够 → 生成”的循环。用 LangGraph 来表达就非常契合。状态是整个图的心脏。我定义了这样一个 Statefrom typing import TypedDict, List class QAState(TypedDict): question: str # 用户原始问题 retrieved_docs: List[str] # 本地检索到的文档片段 search_query: str # 是否需要外部检索如果需要记录关键词 need_external: bool # 分支条件控制是否需要外部工具 answer: str # 最终生成的答案 tried_retrieve: int # 记录已经外部检索了几次防止死循环这里你可能会问为什么分支条件要单独用一个 bool 字段不直接在节点里 return 一个路径字符串就行吗其实 LangGraph 两种方式都支持但我更倾向于把判断依据写进状态里。这样做的好处是方便调试整个图跑完之后你从 State 里就能完整还原每一步决策依据也方便后续做日志审计。3.2 节点和边的规划我一开始画的图很朴素只有四个节点retrieve本地检索拿相关文档片段。should_search判断是否需要外部搜索这是个带条件的出口。external_search调用外部工具补充信息。generate生成最终答案。节点之间怎么走我用条件边来控制。最开始从 START 进入retrieveretrieve结束之后根据 State 里的need_external字段决定是直接去generate还是先去external_search。而external_search执行完不能让它直接跑到generate还要再回到should_search判断一次这样就形成了一条环能够反复检索直到信息足够。不过这里藏了一个风险如果外部搜索永远返回不够整个流程就会无限循环下去。所以我在 State 里放了一个tried_retrieve计数器在条件判断里限制最多跑两轮外部检索超过次数就强制生成答案。4. 完整实现与代码拆解4.1 构造检索节点第一个节点是普通文档检索。这里我刻意做得简化用关键词匹配模拟向量召回让你先看清 LangGraph 的骨架。真实项目里换成向量库的相似度检索也很容易逻辑就发生在节点内部。fake_docs_db { 网络连接: 设备连接 Wi-Fi 失败时先确认密码是否正确然后重启路由器和网卡。, 打印机故障: 打印机不工作检查驱动版本、数据线和默认打印机设置。, 账号登录: 账号无法登录优先检查网络状态然后尝试找回密码。, } def retrieve_node(state: QAState) - QAState: question state[question] retrieved [] for keyword, doc in fake_docs_db.items(): if keyword in question: retrieved.append(doc) if not retrieved: retrieved.append(本地资料库中未找到直接相关内容需要查看外部资源补充。) # 模拟一个简单的相关性打分这里直接用是否有检索结果来判断 has_result any(未找到 not in d for d in retrieved) return { retrieved_docs: retrieved, need_external: not has_result, }这个节点的输出就是一段新的 State 字典。你需要注意LangGraph 的节点返回的字段会和原有 State 做合并所以这里只需要返回要更新的字段不用把question原样传回来。我第一次写的时候把整个 State 都 return 了结果没问题但代码很冗余而且容易覆盖掉别的节点刚更新的字段。4.2 条件判断节点与外部搜索节点接下来是条件判断节点它会读取 State 里的need_external和tried_retrieve字段决定图的下一个走向。这里我用的是 LangGraph 的条件边机制节点本身只需要返回一个表示路径的字符串。def should_search_node(state: QAState) - QAState: # 如果需要外部搜索而且尝试次数还没超过上限 if state[need_external] and state[tried_retrieve] 2: return {tried_retrieve: state[tried_retrieve] 1, search_query: state[question]} return {tried_retrieve: state[tried_retrieve], search_query: }仔细看这个节点既承担了“更新计数器”的职责也决定了边往哪走。不过 LangGraph 判断边的走向不回看这个返回值而是在add_conditional_edges里根据 State 自己写判断逻辑所以更清晰的写法是让这个节点只更新状态真正的路径选择在构建图时完成。我建议你养成这样的习惯后面逻辑复杂了调试起来清爽很多。外部搜索节点也简单我这里用一个模拟函数假装调用了外部搜索 APIdef external_search_node(state: QAState) - QAState: external_result 外部搜索补充请检查设备固件是否更新到最新版本然后尝试重新配对。 docs state[retrieved_docs] [external_result] return { retrieved_docs: docs, need_external: False, # 假设外部搜索已经补全了信息 }你可能会觉得这一步太“模拟”了。说实在的开发调试阶段用模拟数据非常值得你可以先不管真实 API 的延迟和报错把图本身的逻辑跑通再接入真实工具。这比一上来就接真 API、出问题都不知道是图写错了还是接口写错了要高效得多。4.3 生成答案节点与整体构图生成节点负责把检索到的文档拼进提示词让模型做总结回复。这里我用了一个很朴素的 Prompt实际项目可以换成你自己的模板from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage llm ChatOpenAI(modelgpt-4o-mini, temperature0) def generate_node(state: QAState) - QAState: context \n\n.join(state[retrieved_docs]) prompt f根据以下资料回答问题。 资料 {context} 问题{state[question]} 要求给出答案并尽量引用资料中的内容。 messages [ SystemMessage(content你是一名乐于助人的技术支持助手。), HumanMessage(contentprompt), ] response llm.invoke(messages) return {answer: response.content}到这里四个节点都齐了。现在把它们组合成图并且接上条件边。这是我最喜欢的一部分因为整个流程的走向几乎一眼就能看明白from langgraph.graph import StateGraph, START, END graph StateGraph(QAState) graph.add_node(retrieve, retrieve_node) graph.add_node(should_search, should_search_node) graph.add_node(external_search, external_search_node) graph.add_node(generate, generate_node) graph.add_edge(START, retrieve) graph.add_edge(retrieve, should_search) graph.add_conditional_edges( should_search, lambda state: external_search if state[need_external] and state[tried_retrieve] 2 else generate, {external_search: external_search, generate: generate}, ) graph.add_edge(external_search, should_search) graph.add_edge(generate, END) app graph.compile()注意看external_search回到should_search这条边它就是循环的关键。第一次外部搜索结束后系统会再回到判断节点如果这时need_external仍然为 True而且尝试次数没超就继续搜如果条件不再满足就走generate。4.4 实际运行与输出效果图编译好之后直接invoke传入初始状态就能跑起来。下面是我调试验证时用到的两个问题一快一慢正好覆盖了有分支和无分支的场景。result1 app.invoke({question: 打印机不工作怎么办}) print(result1[answer]) print(result1[tried_retrieve]) print(result1[need_external])因为“打印机故障”这个关键词直接命中了本地文档库所以need_external一开始就是 False流程只是简单地从retrieve一路走到了generate计数器保持 0正好是一条直线。再看一个复杂的result2 app.invoke({question: 设备一直离线无法连接怎么办}) print(result2[answer]) print(result2[tried_retrieve]) print(result2[need_external])这个问题在本地文档库里完全找不到对应词所以need_external为 True流程进入external_search然后再回到should_search发现信息补上了才走向generate。整个过程走了两轮判断计数器是 1。从运行结果来看图的行为完全符合预期。你可以把tried_retrieve和need_external的最终值打出来和流程图对照每一步去向都清清楚楚。5. 常见报错与避坑经验5.1 条件边路径不匹配LangGraph 报这种错误往往是你add_conditional_edges里的路径字典写错了。比如你映射里写了search: external_search但图里根本没有叫search的节点注册过或者你返回的是个动态字符串没有映射到已存在的节点名。我的排查顺序是这样的先检查节点注册名有没有写错再检查字典里的 key 和判断函数实际返回值是否一致。建议你在should_search_node里临时加一句 print把返回值打出来一眼就能定位。5.2 循环次数不可控这是处理 Agent 类应用最让人头疼的问题。外部工具永远返回“信息不足”的时候图就会一直循环。我一开始天真地以为模型总会适可而止结果跑了几轮才意识到必须加硬限制。后来我的通用方案是在状态里放一个计数器字段每进入一次外部搜索节点就加一条件边的判断里强制加上计数上限。如果循环是因为工具调用本身还可以在节点内部检查累计调用次数超过阈值直接抛一个自定义错误配合回调函数做告警。5.3 状态字段被意外覆盖LangGraph 的 State 默认是一个 TypedDict 共享对象所有节点更新都是原地覆盖。如果你的两个节点都在更新同一个列表字段后执行的节点会直接覆盖前面的内容而不是追加。我当时就遇到过一次检索节点本来已经拿到三条候选资料结果外部搜索节点返回的时候只带了一条新内容把之前的结果整个覆盖掉了。解决办法分两种一是像我在代码里写的那样读取旧列表再拼接新的二是给字段指定一个自定义的归约函数比如operator.add让 LangGraph 自动合并列表。from typing import Annotated import operator class QAState(TypedDict): retrieved_docs: Annotated[List[str], operator.add]不过用operator.add要注意如果上一个节点返回空列表add会把空列表和旧列表合并结果是旧列表不变没问题如果某个节点不小心对同一个字段返回了两次更新就会重复。所以要明确自己想表达的是“追加”还是“替换”。5.4 模型调用超时这个锅不全是 LangGraph 的。外部模型接口不稳定会导致节点卡住整个图停在那里。我的做法是在节点内部控制模型调用的超时时间给llm.invoke设置timeout参数或者用asyncio.wait_for包一层超时就返回一条错误备注作为上下文而不是让整个流程死等。6. 更进一步的玩法与扩展思路6.1 加入人机审核节点很多业务场景下AI 生成的内容不能直接放出去需要人工确认。LangGraph 对这类需求支持得很好可以在generate之后加一个review节点走到的位置时暂停等人为修改或者确认后再继续。这比自己在代码里维护一个“挂起会话”要优雅得多因为图本身记住了所处的状态和上下文你只需要调resume继续执行。6.2 多工具 Agent 编排如果把external_search_node替换成“调用多个工具”的 Agent 节点整个图就变成一个多工具编排的自动规划器。你可以在一个节点里让模型自主决定调什么工具、参数是什么然后用另一个路由节点判断是否继续。LangGraph 官方推荐的 Agent 架构就是这么做的很多复杂工作流也都可以拆成这种模式。6.3 状态持久化与流式输出除了最基本的invokeLangGraph 还支持流式输出stream方法会一截一截返回节点结果特别适合做打字机效果。如果你需要对话历史保持可以给编译后的应用配置 Checkpointer 参数让状态持久化到指定存储这样即使重启进程也还能从上次执行位置继续。7. 我对 LangGraph 的个人感受与建议从最早的链式结构到现在的图式编排这种演进不是单纯“新框架有多炫”而是大家逐步认识到LLM 应用的流程天然是动态的、分支的甚至带着反馈循环。用图来表达流程结构变得透明哪里容易出错一眼就能看出来也方便团队协作。每个人都清楚整体流程的骨架和每个节点的职责边界。如果你是第一次接触 LangGraph我建议不要先去啃特别复杂的概念先把我这个最小例子在本地跑通感受一下状态和边的流转然后把例子里的模拟数据替换成真实向量库、真实工具和真实模型。一边替换一边踩坑这套架构的理解深度会比只读文档扎实得多。我个人后续在这个方向上的规划是把业务里几个常见的流程资料检索问答、工单分类转派、周报辅助生成都慢慢迁移到 LangGraph 上来。状态和节点一旦沉淀成公共模块复用性真的会提升很多。希望这篇例子能给你一个还算顺滑的起点。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
