为什么需要LangGraph,当链式调用不够用的时候
前面写了很多LangChain的链式调用。LLMChain、SequentialChain、路由链,能处理大部分场景。
但你迟早会遇到一个问题。有些任务,用链式调用怎么写都别扭。
任务有分支,根据中间结果决定走哪条路。任务有循环,做完了检查一下,不满意要重来。任务有状态,中间产生的信息要在多个步骤之间传递和修改。
这些场景,链式调用的"一条直线走到底"的模式就不够用了。你需要图。
这一篇我们讲为什么需要LangGraph,它跟LangChain的链式调用有什么区别,什么时候该用它。
链式调用的局限
先看链式调用为什么不够用。
链式调用的本质是流水线。输入进来,经过一串固定的步骤,输出结果。每一步的输出是下一步的输入,像管道一样,一路向前。
这种模式简单清晰,适合线性的任务。写文章、翻译、摘要,这些任务从头到尾一条线,用链式调用很自然。
但真实世界的任务不总是线性的。
条件分支。Agent执行一个任务,中间需要做判断。如果搜索到了相关资料就走分析路径,没搜到就走提问路径。链式调用不支持"如果…就走A路径,否则走B路径"。
循环重试。Agent写了一段代码,测试发现报错了,需要修改后重新测试。这个"写-测-改-再测"的过程是循环的。链式调用只能走一遍,不能循环。
状态管理。复杂任务中间会产生很多状态信息。当前进行到第几步、已经收集了哪些信息、哪些步骤完成了哪些没完成。这些状态需要在多个步骤之间共享和更新。链式调用没有显式的状态管理机制。
并行执行。有些任务可以并行。同时搜索三个不同的关键词,同时分析多个文档。链式调用是串行的,不方便做并行。
LangGraph就是用来解决这些问题的。
LangGraph是什么
LangGraph是LangChain团队推出的一个扩展库,专门用来构建有状态的、多步骤的复杂Agent工作流。
核心概念是图。把Agent的工作流表示成一个有向图。图的节点是处理步骤,图的边定义了步骤之间的跳转关系。
跟链式调用的区别在于,图可以表达更复杂的控制流。
图可以有条件边。从节点A出来,根据条件决定走向节点B还是节点C。这就是条件分支。
图可以有环。从节点B走到节点C,再从节点C走回节点B。这就是循环。
图有全局状态。所有节点共享一个状态对象,每个节点可以读取和修改状态。这就是状态管理。
图可以有并行节点。多个节点同时执行,等全部完成后再继续。这就是并行。
什么时候用LangGraph
不是所有场景都需要LangGraph。简单的任务用LangChain的链式调用就够了,没必要上LangGraph。
LangGraph适合这些场景。
多步骤决策任务。任务需要多个步骤,步骤之间有依赖关系和条件判断。比如"调研一个技术方案"这个任务,需要先搜索资料,然后判断资料够不够,够了就分析总结,不够就换个关键词重新搜。
需要反思和重试的任务。Agent做完一步以后要检查结果,不对就重来。比如写代码-测试-修bug-再测试的循环。
多Agent协作。多个Agent分工合作,有的负责规划,有的负责执行,有的负责审查。Agent之间有消息传递和状态共享。
复杂的状态管理。任务执行过程中需要维护复杂的状态,比如对话历史、任务进度、中间结果。需要在多个步骤之间共享和更新。
如果你的任务是线性的、单步的、不需要循环和分支的,就用LangChain的链式调用。更简单、更好维护。
如果你的任务有分支、有循环、有复杂状态,就上LangGraph。
一个直观的例子
来看一个LangGraph能做但链式调用做不了的例子。
假设我们要做一个研究Agent。用户给一个研究主题,Agent先搜索资料,然后判断资料够不够。够了就写总结,不够就换个角度再搜。写完总结后自我检查,质量不行就重写。
这个工作流是这样的。
搜索 -> 判断够不够 -> 够了 -> 写总结 -> 检查质量 -> 不行 -> 重写总结
-> 不够 -> 换关键词 -> 搜索(回到第一步)
这里有条件判断(够不够、质量行不行),有循环(重新搜索、重写总结),有状态管理(搜索到的资料、总结的内容)。
用链式调用没法表达这种流程。用LangGraph就可以。
fromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict,List# 定义状态classResearchState(TypedDict):topic:strsearch_results:List[str]summary:strquality_check:strsearch_count:int# 定义节点defsearch_node(state):# 搜索资料topic=state["topic"]# ... 执行搜索results=state.get("search_results",[])results.append(f"关于{topic}的搜索结果...")return{"search_results":results,"search_count":state.get("search_count",0)+1}defjudge_node(state):# 判断资料够不够results=state["search_results"]iflen(results)>=3:return{"quality_check":"enough"}return{"quality_check":"not_enough"}defsummarize_node(state):# 写总结results=state["search_results"]summary=f"基于{len(results)}条资料的总结..."return{"summary":summary}defcheck_node(state):# 检查质量summary=state["summary"]iflen(summary)>20:return{"quality_check":"good"}return{"quality_check":"bad"}# 定义条件边defshould_search_or_summarize(state):ifstate["quality_check"]=="enough":return"summarize"return"search"defshould_finish_or_rewrite(state):ifstate["quality_check"]=="good":returnENDreturn"summarize"# 构建图workflow=StateGraph(ResearchState)# 添加节点workflow.add_node("search",search_node)workflow.add_node("judge",judge_node)workflow.add_node("summarize",summarize_node)workflow.add_node("check",check_node)# 设置入口workflow.set_entry_point("search")# 添加边workflow.add_edge("search","judge")workflow.add_conditional_edges("judge",should_search_or_summarize)workflow.add_edge("summarize","check")workflow.add_conditional_edges("check",should_finish_or_rewrite)# 编译app=workflow.compile()# 运行result=app.invoke({"topic":"AI Agent开发","search_results":[],"summary":"","quality_check":"","search_count":0,})print(result["summary"])print(f"搜索次数:{result['search_count']}")这段代码定义了一个有分支、有循环的研究Agent工作流。搜索完判断够不够,不够就回去再搜,够了就写总结。写完检查质量,不行就重写。
用链式调用是写不出这种流程的。这就是LangGraph的价值。
LangGraph和LangChain的关系
很多人搞不清LangGraph和LangChain的关系。
LangChain是基础框架,提供了Chain、Agent、Tool、Memory、Retriever这些核心组件。适合大部分Agent开发场景。
LangGraph是LangChain的扩展,专门处理复杂的工作流。它在LangChain的基础上,增加了图结构的状态管理能力。
两者的关系是互补的。LangChain管基础组件,LangGraph管复杂流程。实际项目里经常一起用,LangGraph的节点里调用LangChain的组件。
安装也很简单。
pipinstalllanggraph下一篇我们深入LangGraph的核心概念。节点、边、状态、条件边,每个概念怎么用,有什么讲究。