【LangGraph】一.LangGraph 到底是什么
目录
核心思想
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),不是完整 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 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
|
需求 |
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:完整应用,编译后执行
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐

所有评论(0)