为什么需要 LangGraph
传统 LangChain Agent 把控制权完全交给 LLM，这带来了两个核心问题：一是循环控制困难（Agent 可能无限循环），二是状态管理混乱（中间结果难以传递和检查）。
LangGraph 的核心思路是：用有向图（可以包含环）来显式描述 Agent 的执行流程，让开发者对 Agent 行为拥有精确控制。
LangGraph 的定位
LangGraph 是 LangChain 生态中用于构建有状态、多 Actor 应用的低层库。它不是替代 LangChain，而是为 Agent 编排提供了更底层、更灵活的图执行引擎。
核心概念
StateGraph（状态图）
整个 Agent 的执行蓝图。定义了所有节点（Node）和边（Edge），以及节点之间的转移逻辑。StateGraph 以 State 类型为参数化，确保所有节点共享同一套状态模式。
State（状态）
在图执行期间传递和更新的数据容器，通常定义为 TypedDict。每个节点接收当前 State，返回需要更新的字段。LangGraph 会自动合并（merge）状态更新。
Node（节点）
图中的一个执行单元，对应一个 Python 函数。接收 State，执行逻辑（调用 LLM、调用工具、做判断等），返回状态更新字典。节点是纯函数，便于测试。
Edge（边）
定义节点间的转移关系。普通边（add_edge）是固定跳转；条件边（add_conditional_edges）根据节点输出或状态动态决定下一个节点，这是实现分支和循环的核心机制。
START / END
图的入口和出口特殊节点。START 连接第一个执行节点，END 表示图执行完成。一个图可以有多条通往 END 的路径，实现不同的终止条件。
StateGraph 结构图
LangGraph 状态图执行模型：
┌─────────────────────────────────────────────────────────┐
│ StateGraph │
│ │
│ [START] ──→ [agent_node] ──→ [tool_node] ──→ [END] │
│ ↑ │ │
│ └───────────────┘ │
│ (循环：工具结果反馈给 agent) │
│ │
│ 条件边（router）决定： │
│ agent_node 输出 → 有工具调用？ → tool_node │
│ → 无工具调用？ → END │
└─────────────────────────────────────────────────────────┘
State 在所有节点间共享和传递：
State = {
messages: [HumanMessage, AIMessage, ToolMessage, ...],
tool_calls: [...],
iteration_count: int,
}
第一个 LangGraph Agent
以下代码实现一个完整的 ReAct Agent，能够调用工具并循环迭代：
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
# ── 1. 定义状态 ──────────────────────────────────────────
class AgentState(TypedDict):
# add_messages 是一个 reducer：自动将新消息追加到列表
messages: Annotated[list[BaseMessage], add_messages]
# ── 2. 定义工具 ──────────────────────────────────────────
@tool
def search_web(query: str) -> str:
"""搜索网络获取最新信息。"""
# 真实项目中调用 Tavily、SerpAPI 等
return f"关于'{query}'的搜索结果：[模拟数据]"
@tool
def calculate(expression: str) -> str:
"""计算数学表达式。expression 示例：'2 ** 10 + 100'"""
try:
result = eval(expression, {"__builtins__": {}})
return str(result)
except Exception as e:
return f"计算错误：{e}"
tools = [search_web, calculate]
# ── 3. 创建 LLM 并绑定工具 ───────────────────────────────
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)
# ── 4. 定义节点函数 ──────────────────────────────────────
def agent_node(state: AgentState):
"""LLM 推理节点：决定下一步行动。"""
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
# ToolNode 自动处理工具调用并返回 ToolMessage
tool_node = ToolNode(tools)
# ── 5. 定义路由函数（条件边）─────────────────────────────
def should_continue(state: AgentState) -> str:
"""检查最新消息是否包含工具调用。"""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "use_tools"
return "finish"
# ── 6. 构建图 ────────────────────────────────────────────
graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("tools", tool_node)
graph_builder.add_edge(START, "agent")
# 条件边：agent → (use_tools → tools) 或 (finish → END)
graph_builder.add_conditional_edges(
"agent",
should_continue,
{"use_tools": "tools", "finish": END}
)
# tools 节点执行完后，无条件返回 agent
graph_builder.add_edge("tools", "agent")
# ── 7. 编译并运行 ────────────────────────────────────────
graph = graph_builder.compile()
result = graph.invoke({
"messages": [HumanMessage(content="搜索 LangGraph 最新特性，然后计算 2**16")]
})
for msg in result["messages"]:
print(f"[{msg.__class__.__name__}] {msg.content[:100]}")条件路由与循环控制
LangGraph 的条件边是实现复杂控制流的关键。路由函数返回字符串键，对应到不同的目标节点：
# 多分支路由示例
def complex_router(state: AgentState) -> str:
last_msg = state["messages"][-1]
# 防无限循环：超过最大迭代次数强制终止
if state.get("iteration", 0) >= 10:
return "force_end"
if not last_msg.tool_calls:
return "end"
# 根据工具类型路由到不同节点
tool_name = last_msg.tool_calls[0]["name"]
if tool_name.startswith("code_"):
return "code_executor"
elif tool_name.startswith("search_"):
return "search_tools"
else:
return "general_tools"
graph_builder.add_conditional_edges(
"agent",
complex_router,
{
"end": END,
"force_end": END,
"code_executor": "code_node",
"search_tools": "search_node",
"general_tools": "tools",
}
)State 的 Reducer 机制
LangGraph 的状态更新采用 Reducer 模式：每个字段可以指定如何合并新旧值。
import operator
from typing import Annotated
class AdvancedState(TypedDict):
# add_messages：追加新消息（不替换）
messages: Annotated[list, add_messages]
# operator.add：数值累加（每次 +1）
iteration: Annotated[int, operator.add]
# 无 Reducer：直接替换（最新值覆盖旧值）
current_task: str
is_done: bool
# 自定义 Reducer：合并字典
tool_results: Annotated[dict, lambda a, b: {**a, **b}]
# 节点只需返回需要更新的字段
def my_node(state: AdvancedState):
return {
"iteration": 1, # 会累加：旧值 + 1
"current_task": "搜索", # 会替换
# messages 不返回 = 不修改
}流式执行与检查点
LangGraph 支持流式执行，可以实时观察每个节点的输入输出，这对调试和用户体验都非常关键：
from langgraph.checkpoint.memory import MemorySaver
# 添加检查点（Checkpointer）：支持暂停、恢复、时间旅行
checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)
# 流式执行 - 实时输出每个节点的状态变化
config = {"configurable": {"thread_id": "session_001"}}
for chunk in graph.stream(
{"messages": [HumanMessage(content="你好，帮我搜索 Python 3.13 新特性")]},
config=config,
stream_mode="updates" # 只输出每步的变更
):
for node_name, node_output in chunk.items():
print(f"节点 [{node_name}] 更新：")
if "messages" in node_output:
last = node_output["messages"][-1]
print(f" {last.__class__.__name__}: {last.content[:80]}")
# 多轮对话 - 同一 thread_id 保持历史
graph.invoke(
{"messages": [HumanMessage(content="继续，再搜索 3.14 路线图")]},
config=config # 相同 thread_id，会自动加载历史消息
)
检查点的价值
MemorySaver 将状态保存在内存中，适合开发测试。生产环境应使用 SqliteSaver 或 PostgresSaver，支持跨进程持久化，并能实现"人工审批"（interrupt_before）等高级功能。
与 LangChain 的关系
LangChain 提供了模型封装、提示模板、文档加载器、向量数据库等高层抽象。
LangGraph 在 LangChain 之上提供了图执行引擎，用于构建复杂的 Agent 和多 Actor 工作流。
你可以在 LangGraph 的节点中自由使用任何 LangChain 组件（如 ChatOpenAI、向量检索链、提示模板），二者是互补而非竞争关系。