作者:hrhcode
目录
核心思想
从 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 TypedDictclass 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),不是完整 State
- LangGraph 会自动合并多个节点的更新
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 TypedDictfrom langgraph.graph import StateGraph, START, ENDclass 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
需求
| LangChain
| LangGraph
| 简单问答机器人
| 适合
| 不太适合
| 快速原型验证
| 适合
| 配置稍多
| 复杂的流程控制
| 困难
| 原生支持
| 需要人工审批
| 需要自己实现
| 内置支持
| 多 Agent 协作
| 困难
| 轻松表达
| 状态持久化
| 需要自己实现
| 内置支持
| 选择建议:
如果你的应用只是简单的"用户提问 → Agent 回答",用 LangChain 就够了。
如果你需要:
- 精确控制执行流程(分支、循环、并行)
- 在关键节点暂停等待人工审批
- 多个 Agent 协作完成复杂任务
- 保存和恢复执行状态
那么 LangGraph 更合适。
运行机制
LangGraph 的执行分为两个阶段:
阶段 1:定义图结构
- 定义 State(数据结构)
- 定义 Node(执行逻辑)
- 定义 Edge(流程控制)
阶段 2:编译执行
- compile() 把图编译成可执行的应用
- invoke() 执行应用
编译的时候,LangGraph 会:
- 验证图结构是否完整(没有孤立节点)
- 优化执行计划
- 配置运行时参数(如 Checkpointer)
为什么用图结构
图结构有几个关键优势:
1. 可视化
图可以画出来,方便理解和调试。
START → A → B → C → END ↑ │ └────┘一看就知道:执行完 B 到 C,C 可能回到 B 形成循环。
2. 并发执行
如果节点之间没有依赖关系,可以并行执行。
┌→ A ─┐START ─┼→ B ─┼→ END └→ C ─┘A、B、C 可以同时执行,最后汇总结果。
3. 状态管理
State 在节点之间传递和更新,天然支持状态追踪和持久化。
4. 流程控制
分支、循环、并行都是图的基本操作,不需要额外实现。
总结
LangGraph 的核心价值:把 Agent 的执行流程从黑盒变成明盒。
- State:数据容器,节点之间传递信息
- Node:执行单元,处理具体逻辑
- Edge:流程控制,决定执行顺序
- Graph:完整应用,编译后执行
原文地址:https://blog.csdn.net/hrh1234h/article/details/160633865 |