30秒快速回答: LangGraph 是一个用「图(Graph)」来编排 AI 应用的框架。你定义节点(Node,每个节点是一个函数或 LLM 调用)和边(Edge,控制流转方向),图可以循环、分支、暂停等待人工输入。它解决了普通 Chain「只能线性执行、无法回环修正」的痛点,是构建复杂 Agent 的工业级方案。
为什么需要 LangGraph?
普通 LangChain Chain 是线性的:
Chain: Input → LLM → Output
问题:无法「检查→修正→再检查」的循环
但真实 AI 应用需要更复杂的控制流:
Agent 工作流:
┌──→ 生成 → 检查 → 不合格 ──┐
│ │
└────────────────────────────┘
↓ 合格
输出
LangGraph 用「状态图」表达这种复杂流程:
| 能力 | 普通 Chain | LangGraph |
|---|---|---|
| 执行方式 | 线性 | 图(可循环/分支) |
| 状态管理 | 无 | 持久化 State |
| 人工干预 | 不支持 | 支持(interrupt) |
| 循环控制 | 不支持 | 支持(条件边) |
| 多 Agent 协作 | 复杂 | 原生支持 |
LangGraph 的核心概念
1. State(状态)
所有节点共享一个状态对象,节点读取并修改它:
from typing import TypedDict, Annotated
import operator
class AgentState(TypedDict):
messages: Annotated[list, operator.add] # 消息列表(累加)
query: str # 用户查询
search_results: Annotated[list, operator.add] # 搜索结果
analysis: str # 分析结果
report: str # 最终报告
2. Node(节点)
每个节点是一个函数,接收 State,返回要更新的部分:
def search_node(state: AgentState):
"""搜索节点"""
results = web_search(state["query"])
return {"search_results": [results]} # 只返回要更新的字段
def analyze_node(state: AgentState):
"""分析节点"""
analysis = llm.analyze(state["search_results"])
return {"analysis": analysis}
3. Edge(边)
控制节点之间的流转:
# 普通边:A → B
workflow.add_edge("search", "analyze")
# 条件边:根据 State 决定走哪条路
def should_continue(state: AgentState) -> str:
if state["analysis"] == "需要更多信息":
return "search" # 回到搜索节点(循环!)
return "write_report" # 进入报告节点
workflow.add_conditional_edges(
"analyze",
should_continue,
{"search": "search", "write_report": "write_report"}
)
实战:构建一个带循环的 Agent
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
# 1. 定义状态
class WritingState(TypedDict):
topic: str
draft: str
review_comments: str
iteration: int
# 2. 定义节点
def write_node(state: WritingState):
llm = ChatOpenAI(model="gpt-4o")
prompt = f"写一篇关于「{state['topic']}」的文章(第 {state['iteration']} 稿)"
if state.get("review_comments"):
prompt += f"\n根据修改意见改进:{state['review_comments']}"
draft = llm.invoke(prompt).content
return {"draft": draft, "iteration": state["iteration"] + 1}
def review_node(state: WritingState):
llm = ChatOpenAI(model="gpt-4o")
prompt = f"审查以下文章,给出修改意见(如果没问题就说'通过'):\n{state['draft']}"
comments = llm.invoke(prompt).content
return {"review_comments": comments}
# 3. 条件边:检查是否通过
def check_approval(state: WritingState) -> str:
if "通过" in state["review_comments"]:
return "end"
elif state["iteration"] >= 3: # 最多改 3 次
return "end"
else:
return "rewrite"
# 4. 构建图
workflow = StateGraph(WritingState)
workflow.add_node("write", write_node)
workflow.add_node("review", review_node)
workflow.set_entry_point("write")
workflow.add_edge("write", "review")
workflow.add_conditional_edges(
"review",
check_approval,
{"rewrite": "write", "end": END}
)
app = workflow.compile()
# 5. 执行
result = app.invoke({
"topic": "AI Agent 的未来趋势",
"iteration": 1
})
print(result["draft"])
高级特性
持久化与断点恢复
from langgraph.checkpoint.sqlite import SqliteSaver
# 使用 SQLite 持久化状态
memory = SqliteSaver.from_conn_string(":memory:")
app = workflow.compile(checkpointer=memory)
# 每次调用带 thread_id,可恢复
config = {"configurable": {"thread_id": "user_123"}}
app.invoke({"topic": "..."}, config)
# 中断后恢复
app.invoke(None, config) # 从断点继续
人工干预(Human-in-the-loop)
from langgraph.types import interrupt
def sensitive_action(state):
# 暂停等待人工确认
user_input = interrupt({"question": "确认要删除这个文件吗?"})
if user_input == "yes":
delete_file()
return {"status": "done"}
多 Agent 协作
# 用 Supervisor 模式协调多个 Agent
from langgraph.prebuilt import create_supervisor
supervisor = create_supervisor(
agents=[research_agent, writer_agent, reviewer_agent],
model="gpt-4o"
)
LangGraph vs 其他框架
| 框架 | 特点 | 适合场景 |
|---|---|---|
| LangGraph | 图编排、状态管理、生产级 | 复杂多步骤 Agent |
| LangChain | 链式调用、工具丰富 | 简单 RAG/Chain |
| CrewAI | 角色扮演、多 Agent 协作 | 模拟团队分工 |
| AutoGen | 对话式多 Agent | 研究/实验 |
| LlamaIndex | 文档索引 + Agent | 知识库问答 |
常见问题
Q: LangGraph 和 LangChain 是什么关系?
LangGraph 是 LangChain 生态的一部分,但更底层、更灵活。LangChain 适合「线性 Chain」,LangGraph 适合「有状态、有循环的 Agent」。两者可以混用。
Q: LangGraph 学习曲线陡吗?
比普通 Chain 陡,但比手写状态机简单。核心概念只有三个:State、Node、Edge。掌握后构建复杂 Agent 的效率远超手写代码。
Q: LangGraph 能用于生产环境吗?
能。LangGraph 提供了 LangGraph Platform(原 LangSmith)用于部署、监控和调试。许多企业已在生产环境使用。
下一步建议: 用本文的「写作 Agent」代码作为起点,加入你自己的节点(如「查资料」「配图」),体验用图编排复杂 AI 工作流的感觉。