【LangGraph】一.LangGraph 到底是什么

目录核心思想从 LangChain 的局限说起LangGraph 的解决方案四个核心概念1. State状态2. Node节点3. Edge边4. Graph图实际例子带循环的检索LangGraph vs LangChain运行机制为什么用图结构总结核心思想LangGraph 是一个用图结构来编排 Agent 执行的框架。为什么需要图结构因为传统的 Agent 执行流程是黑盒——模型自己决定先做什么、后做什么。当你需要精确控制流程时比如先检索文档如果不够就再检索一次然后生成答案黑盒就不够用了。LangGraph 把 Agent 拆分成节点Node和边Edge让你可以像画流程图一样设计 Agent 的执行逻辑。从 LangChain 的局限说起LangChain 的 Agent 工作方式是while True: # 1. 调用模型 response model.invoke(messages) # 2. 模型决定是否调用工具 if response.tool_calls: tool_result execute_tool(response.tool_calls[0]) messages.append(tool_result) else: # 3. 模型说结束就结束 break这个模型的问题在于流程完全由模型控制。你想实现这样的逻辑先检索文档如果检索结果少于 3 篇换个关键词再检索一次然后生成答案最后检查答案质量如果不好就重新生成这在 LangChain 里很难实现因为你没法在流程中设置明确的检查点和分支逻辑。LangGraph 的解决方案LangGraph 让你显式定义流程START → 检索节点 → 判断节点 → 生成节点 → END ↑ │ └──────────┘每个方框是一个节点Node箭头是边Edge。节点执行具体操作边决定执行顺序。四个核心概念1. State状态State 是整个图共享的数据容器。所有节点都能读取和修改 State。from typing import TypedDict class State(TypedDict): question: str # 用户问题 documents: list # 检索到的文档 answer: str # 生成的答案为什么需要 State因为节点之间需要传递数据。检索节点检索到的文档需要传递给生成节点使用。State 就是这个数据传递的媒介。设计原则只放跨节点共享的数据字段名要清晰表达含义不要放临时变量临时变量应该在节点内部处理2. Node节点节点就是一个 Python 函数接收当前 State返回 State 的更新。def retrieve_node(state: State) - dict: # 从 State 读取问题 question state[question] # 执行检索 docs search_database(question) # 返回更新只返回变化的字段 return {documents: docs}关键点节点函数接收完整的 State返回的是 State 的更新dict不是完整 StateLangGraph 会自动合并多个节点的更新3. Edge边边决定节点的执行顺序。有两种类型普通边固定跳转到下一个节点graph.add_edge(retrieve_node, generate_node)条件边根据条件选择下一个节点def should_continue(state: State) - str: # 根据文档数量决定 if len(state[documents]) 3: return retrieve_again # 再去检索 else: return generate # 直接生成 graph.add_conditional_edges( retrieve_node, # 从哪个节点 should_continue, # 条件函数 { retrieve_again: retrieve_node, generate: generate_node, } )4. Graph图图就是节点和边的集合。定义好图之后编译成可执行的应用。from langgraph.graph import StateGraph, START, END # 1. 创建图 graph StateGraph(State) # 2. 添加节点 graph.add_node(retrieve, retrieve_node) graph.add_node(generate, generate_node) # 3. 设置入口 graph.set_entry_point(retrieve) # 4. 添加边 graph.add_edge(retrieve, generate) graph.add_edge(generate, END) # 5. 编译 app graph.compile() # 6. 执行 result app.invoke({question: 什么是 LangGraph})实际例子带循环的检索假设我们要实现检索文档如果不够就再检索一次。from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): question: str documents: list attempts: int # 检索次数 def retrieve(state: State) - dict: 检索文档 docs search(state[question]) return { documents: docs, attempts: state.get(attempts, 0) 1 } def check_docs(state: State) - str: 检查文档是否足够 if len(state[documents]) 3 and state[attempts] 2: return need_more else: return enough def generate(state: State) - dict: 生成答案 answer llm.generate(state[documents]) return {answer: answer} # 组装图 graph StateGraph(State) graph.add_node(retrieve, retrieve) graph.add_node(generate, generate) graph.set_entry_point(retrieve) graph.add_conditional_edges( retrieve, check_docs, { need_more: retrieve, # 循环 enough: generate, } ) graph.add_edge(generate, END) app graph.compile() result app.invoke({question: 什么是 LangGraph})执行流程从retrieve节点开始检索文档执行check_docs如果文档少于 3 篇且检索次数少于 2 次回到retrieve否则执行generate生成答案结束LangGraph vs LangChain需求LangChainLangGraph简单问答机器人适合不太适合快速原型验证适合配置稍多复杂的流程控制困难原生支持需要人工审批需要自己实现内置支持多 Agent 协作困难轻松表达状态持久化需要自己实现内置支持选择建议如果你的应用只是简单的用户提问 → Agent 回答用 LangChain 就够了。如果你需要精确控制执行流程分支、循环、并行在关键节点暂停等待人工审批多个 Agent 协作完成复杂任务保存和恢复执行状态那么 LangGraph 更合适。运行机制LangGraph 的执行分为两个阶段阶段 1定义图结构定义 State数据结构定义 Node执行逻辑定义 Edge流程控制阶段 2编译执行compile()把图编译成可执行的应用invoke()执行应用编译的时候LangGraph 会验证图结构是否完整没有孤立节点优化执行计划配置运行时参数如 Checkpointer为什么用图结构图结构有几个关键优势1. 可视化图可以画出来方便理解和调试。START → A → B → C → END ↑ │ └────┘一看就知道执行完 B 到 CC 可能回到 B 形成循环。2. 并发执行如果节点之间没有依赖关系可以并行执行。┌→ A ─┐ START ─┼→ B ─┼→ END └→ C ─┘A、B、C 可以同时执行最后汇总结果。3. 状态管理State 在节点之间传递和更新天然支持状态追踪和持久化。4. 流程控制分支、循环、并行都是图的基本操作不需要额外实现。总结LangGraph 的核心价值把 Agent 的执行流程从黑盒变成明盒。State数据容器节点之间传递信息Node执行单元处理具体逻辑Edge流程控制决定执行顺序Graph完整应用编译后执行