引言：当 AI 工作流需要“记忆”
在构建复杂的 AI 应用时，尤其是涉及多步推理、工具调用或对话交互的场景，一个核心挑战浮现出来：如何让工作流拥有“状态”？传统的大语言模型（LLM）调用是无状态的，每次输入都是独立的。但现实中的任务，如代码生成、多轮客服或自动化报告，需要模型记住上下文、追踪进度并做出连贯决策。LangGraph，作为 LangChain 生态中的新星，正是为解决这一问题而生。它允许开发者以图结构定义有状态的 AI 工作流，让 Agent 能够像人类一样“思考”和“记忆”。
本文将带你从零开始，理解 LangGraph 的核心概念，并通过一个可运行的代码示例，演示如何构建一个具备记忆能力的 AI 工作流。
LangGraph 是什么？
LangGraph 是一个用于构建有状态、多参与者应用的库，它将工作流建模为有向图（Directed Graph）。每个节点（Node）代表一个计算步骤（如调用 LLM、执行函数、处理数据），而边（Edge）则定义了节点间的数据流动和控制逻辑。与传统的链式调用不同，图结构天然支持循环、分支和条件跳转，这使得它非常适合构建具有循环推理能力的 Agent。
核心概念速览
- Graph：整个工作流的蓝图，包含节点和边。
- Node：工作流中的一个步骤。可以是 Python 函数、LLM 调用或任何可调用对象。
- Edge：连接节点的路径。控制数据如何从一个节点流向另一个节点。
- State：贯穿整个工作流的上下文对象。每个节点都可以读取和修改 State，从而保持记忆。
- Compiled Graph：将 Graph 编译成可执行对象，用于实际运行。
实战：构建一个带记忆的问答 Agent
我们来实现一个简单的 Agent：它接收用户问题，判断是否需要查询外部知识（模拟），然后生成最终答案。整个过程会维护一个对话历史 State。
环境准备
首先，确保安装了 LangGraph 和 LangChain：
pip install langgraph langchain-openai定义 State
State 是一个 TypedDict，定义了工作流中可共享的数据结构。
from typing import TypedDict, List, Annotated
import operator
from langgraph.graph import StateGraph, END
class AgentState(TypedDict):
messages: List[dict] # 对话历史
need_knowledge: bool # 是否需要查询知识库
final_answer: str # 最终答案这里 messages 使用了 Annotated 类型，并指定 operator.add 作为归约器（Reducer），表示当多个节点向同一个 State 字段添加数据时，会进行列表拼接而非覆盖。
定义节点函数
每个节点是一个 Python 函数，接收当前 State，返回一个字典用于更新 State。
def call_llm(state: AgentState) -> dict:
"""模拟 LLM 调用：根据消息历史判断是否需要知识"""
# 简单规则：如果问题包含"查询"或"知识"，则设为 True
last_msg = state["messages"][-1]["content"]
need_knowledge = "查询" in last_msg or "知识" in last_msg
return {"need_knowledge": need_knowledge}
def query_knowledge_base(state: AgentState) -> dict:
"""模拟知识库查询"""
state["messages"].append({"role": "system", "content": "已查询知识库，得到结果：LangGraph 是一个图框架"})
return {"messages": state["messages"]}
def generate_answer(state: AgentState) -> dict:
"""生成最终答案"""
context = ""
if state["need_knowledge"]:
context = "根据知识库信息，"
answer = f"{context}你的问题是：{state['messages'][0]['content']}。我是基于 LangGraph 构建的 Agent。"
return {"final_answer": answer}构建图
现在，将节点和边组合起来。我们定义条件边：如果 need_knowledge 为 True，则走知识库查询路径，否则直接生成答案。
# 初始化图
graph = StateGraph(AgentState)
# 添加节点
graph.add_node("llm_call", call_llm)
graph.add_node("knowledge_base", query_knowledge_base)
graph.add_node("answer", generate_answer)
# 设置入口边
graph.set_entry_point("llm_call")
# 添加条件边
graph.add_conditional_edges(
"llm_call",
lambda state: "knowledge_base" if state["need_knowledge"] else "answer",
{"knowledge_base": "knowledge_base", "answer": "answer"}
)
# 添加最终边
graph.add_edge("knowledge_base", "answer")
graph.add_edge("answer", END)
# 编译图
app = graph.compile()运行工作流
现在，我们可以输入一个问题，观察 Agent 如何根据 State 做出决策。
# 初始状态
initial_state = {
"messages": [{"role": "user", "content": "LangGraph 是什么？需要查询知识库"}],
"need_knowledge": False,
"final_answer": ""
}
# 运行图
result = app.invoke(initial_state)
print("最终答案:", result["final_answer"])输出：
最终答案: 根据知识库信息，你的问题是：LangGraph 是什么？需要查询知识库。我是基于 LangGraph 构建的 Agent。工作流可视化
编译后的图对象 app 提供了 get_graph() 方法，可以生成 Mermaid 图或 ASCII 表示。
print(app.get_graph().draw_ascii())输出示例：
+-----------+ +------------------+ +----------+
| llm_call | --> | knowledge_base | --> | answer |
+-----------+ +------------------+ +----------+
| ^
| (else) |
+---------------------------------------------+进阶：处理循环与持久化
LangGraph 的真正威力在于支持循环。例如，你可以让 Agent 在生成答案后，自动检查答案质量，若不满足则重新调用 LLM 修改。这通过添加指向自身或之前节点的边实现。
此外，LangGraph 支持 Checkpointing（检查点），可以将 State 持久化到数据库（如 SQLite、PostgreSQL），使得工作流可以从中断处恢复，这对于长时间运行的任务至关重要。
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string(":memory:")
app = graph.compile(checkpointer=memory)
# 后续调用会自动保存和恢复 State总结与行动建议
LangGraph 为构建有状态、可循环的 AI 工作流提供了优雅的图结构抽象。通过本文，你学习了：
- 如何使用 StateGraph 定义状态
- 如何创建节点和条件边
- 如何运行一个完整的 Agent 工作流
下一步行动建议：
- 尝试复杂场景：将
call_llm节点替换为真实的大模型 API 调用（如 OpenAI），并引入工具调用节点（如搜索、计算器）。 - 引入循环：设计一个自我纠正循环——让 Agent 先回答，然后检查答案，若发现错误则重新生成。
- 探索持久化：将 Checkpointer 集成到你的应用中，实现对话状态的长期保存。
- 观看社区案例：GitHub 上 LangGraph 的
examples目录包含多个复杂示例，如代码 Agent 和 RAG 系统。
LangGraph 仍在快速发展中，它的设计理念——用图表达 Agent 的思维流程——正在成为构建下一代 AI 应用的核心范式。开始动手构建你的第一个有状态工作流吧！