现在让我们设好番茄钟放一首好听的音乐开始学习吧 🌈 😋
在系列的前两篇中，我们掌握了 LangChain 的核心基础，学会了如何用 LCEL 优雅地搭建一条“单向流水线”。
然而，现实业务往往比流水线复杂得多：如果生成的代码报错了，模型需要循环回来修复；如果搜到的资料不全，模型需要动态决策去换个词再搜。面对这种包含“循环”、“条件分支”和“自主决策”的场景，单纯的“链（Chain）”式结构会变得异常臃肿，难以维护。
今天，我们要聊的主角是 LangGraph。它将带你跳出“单行道”，进入 AI 开发的“立交桥”时代。
一、 引言：从“链(Chain)”到“图(Graph)”的思维跃迁
1. LangChain 线性思维的局限
LangChain 的核心是 Chain。它的本质是一个 DAG（有向无环图）。
- 痛点：它是一条通往终点的单行道。虽然我们可以通过
if-else强行在链里写分支，但一旦涉及“失败重试”、“多轮循环推理”或“多智能体来回博弈”，代码就会迅速变成意大利面条，逻辑极其僵化。
2. LangGraph 的破局机制
LangGraph 将工作流建模为“图（Graph）”。它不仅仅是 LangChain 的升级，更是一种思维方式的进化：
- 循环支持 (Cycles)：允许流程在节点之间形成回路，让模型可以“知错就改”。
- 动态路由：模型可以根据中间结果，实时决定下一站去哪。
- 状态持久化：全局共享一个
State对象，大模型不再是“执行脚本”，而是升级为具有自主决策能力的智能体（Agent）。
二、 核心概念拆解：通俗理解“快递配送系统”
为了理解这些抽象概念，我们可以把 LangGraph 想象成一个“智能快递配送系统”：
1. 状态 (State) —— 贯穿全局的“包裹信息卡”
在 LangGraph 中，State 是流转的灵魂。它是一个由开发者定义的共享对象（通常是 TypedDict）。
比喻：就像每个包裹都贴着一张卡片，记录着发货人、收货地、是否加急、已走过哪些站点。所有站点（节点）都能读取并更新这张卡片，确保信息不丢失。
2. 节点 (Nodes) —— 执行计算的“快递站点”
节点是执行具体逻辑的地方（其实就是普通的 Python 函数）。
比喻：快递系统中的“揽收站”、“分拣中心”、“配送站”。它们互不直接通信，只是接收包裹（State），处理完后更新包裹卡片的信息，然后交给系统进行下一站流转。
3. 边 (Edges) 与 条件边 —— 配送的“路网”
- 普通边 (Edges)：固定的流向。例如：揽收完必须去分拣中心。
- 条件边 (Conditional Edges)：图的灵魂。系统会根据包裹卡片上的信息（如：加急标志）动态决定下一站。
比喻：如果卡片写着“加急”，系统自动将包裹送往“航空件节点”；否则走“陆运节点”。
三、 核心构建机制：StateGraph 与 compile()
构建一个图通常分为三步：
- 初始化：
StateGraph(YourStateClass)，定义这个图里流转的数据结构。 - 添加组件：通过
.add_node()添加处理函数，通过.add_edge()设置连接。 - 编译 (compile)：这是最关键的一步。
划重点：.compile() 的真正含义
与 C++ 等传统编程语言的编译不同，LangGraph 的编译是在运行时动态验证图的合法性。它会检查：从起点能否走到终点？有没有孤立的死节点？有没有死循环？一旦编译通过，这个图就变成了一个标准的 Runnable 对象，可以像 LangChain 里的链一样直接调用。
四、 实战演练 1：非 LLM 基础图（熟悉 API）
我们先不用大模型，用纯 Python 函数跑通一个“智能快递系统”，感受一下图的流转逻辑。
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
# 1. 定义状态 (包裹卡片)
class PackageState(TypedDict):
location: str
priority: str # "normal" 或 "urgent"
status: str
# 2. 编写节点 (快递站点)
def receive_node(state: PackageState):
print("--- 揽收包裹 ---")
return {"status": "已入库"}
def express_delivery(state: PackageState):
print("--- 走航空特快配送 ---")
return {"status": "配送中(飞)"}
def normal_delivery(state: PackageState):
print("--- 走普通陆运配送 ---")
return {"status": "配送中(车)"}
# 3. 编写动态路由函数 (分拣员)
def sorting_logic(state: PackageState) -> Literal["urgent_path", "normal_path"]:
if state["priority"] == "urgent":
return "urgent_path"
return "normal_path"
# 4. 构建图
workflow = StateGraph(PackageState)
workflow.add_node("receive", receive_node)
workflow.add_node("express", express_delivery)
workflow.add_node("normal", normal_delivery)
# 设置连线
workflow.add_edge(START, "receive") # 起点到揽收
# 设置核心：条件边
workflow.add_conditional_edges(
"receive", # 从哪个节点出发进行判断
sorting_logic, # 判断逻辑函数
{
"urgent_path": "express", # 如果返回这个，去 express 节点
"normal_path": "normal" # 如果返回那个，去 normal 节点
}
)
workflow.add_edge("express", END)
workflow.add_edge("normal", END)
# 5. 编译并运行
app = workflow.compile()
app.invoke({"priority": "urgent", "location": "北京"})
五、 实战演练 2：带有大模型的循环智能体 (AI Agent)
理解了图结构，接下来我们玩点真的：用大模型构建一个真正的 AI Agent。
这个 Agent 的目标是：根据用户指令，自主决定是否需要调用搜索工具。如果调用了工具，它会拿着结果回来再次思考，直到给出最终答案。
import operator
from typing import Annotated, TypedDict, Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
# ==========================================
# 1. 定义具体的工具 (Tool)
# ==========================================
# 这里也可以使用一些专门为ai打造的搜索工具比如 Tavily Search
@tool
def search_weather(city: str) -> str:
"""当用户询问某地的天气时，必须调用此工具。需要传入城市名称。"""
print(f"n[🛠️ 系统日志] 正在执行工具：查询 {city} 的天气...")
# 这里使用模拟数据，实际生产中可替换为真实的第三方天气 API
weather_database = {
"北京": "晴天，25°C，非常适合户外活动",
"上海": "小雨，20°C，出门请记得带伞",
"广州": "雷阵雨，30°C，天气闷热"
}
return weather_database.get(city, f"抱歉，没有查到 {city} 的天气数据。")
# 将工具放入列表
tools_list = [search_weather]
# ==========================================
# 2. 定义状态 (State)
# ==========================================
class AgentState(TypedDict):
# operator.add 确保新产生的消息会叠加到旧列表中，而不是覆盖
messages: Annotated[list, operator.add]
# ==========================================
# 3. 设置核心节点 (Nodes)
# ==========================================
# 初始化大模型，并将我们刚才写的工具绑定给它
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0).bind_tools(tools_list)
def llm_call(state: AgentState):
"""大模型推理节点"""
response = model.invoke(state["messages"])
# 将大模型的最新回复包装成列表返回，它会被附加到 State 的 messages 中
return {"messages": [response]}
# ==========================================
# 4. 定义路由逻辑 (Conditional Edges)
# ==========================================
def should_continue(state: AgentState) -> Literal["tools", END]:
"""判断下一步是去调用工具，还是结束对话"""
last_message = state["messages"][-1]
# 如果大模型的最新回复中包含工具调用指令 (tool_calls)
if last_message.tool_calls:
return "tools"
# 如果没有调用工具，说明已经得出最终结论，流程结束
return END
# ==========================================
# 5. 构建与编译图 (Graph)
# ==========================================
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("agent", llm_call)
# 使用 LangGraph 内置的 ToolNode 来自动执行工具列表中的函数
workflow.add_node("tools", ToolNode(tools_list))
# 定义信息流向
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue)
# 核心闭环：工具执行完毕后，必须把结果送回给 agent 重新思考
workflow.add_edge("tools", "agent")
# 编译成可运行的应用
app = workflow.compile()
# ==========================================
# 6. 运行与测试
# ==========================================
if __name__ == "__main__":
print("🤖 Agent 已启动...")
# 用户的初始问题
user_input = "北京今天天气怎么样？"
print(f"n👤 用户: {user_input}")
# 初始化状态
inputs = {"messages": [HumanMessage(content=user_input)]}
# 使用 stream 模式运行，以便我们能看到它在节点间流转的每一步
for output in app.stream(inputs, stream_mode="updates"):
# output 的格式是 {"节点名称": {"状态更新"}}
for node_name, state_update in output.items():
last_msg = state_update["messages"][-1]
if node_name == "agent":
if last_msg.tool_calls:
print(f"🧠 Agent (思考中): 我需要调用工具来获取信息...")
else:
print(f"🤖 Agent (最终回复): {last_msg.content}")
elif node_name == "tools":
print(f"🛠️ 工具返回结果: {last_msg.content}")
代码运行原理解析：
当你运行这段代码并询问“北京今天天气怎么样？”时，LangGraph 的流转路径是这样的：
START->agent：大模型收到问题，不知道北京天气，决定调用search_weather工具。agent->should_continue->tools：条件边检测到了大模型的工具调用意图，将路线切换到tools节点。tools节点执行：触发了我们的 Python 函数，打印出[🛠️ 系统日志]，并将“晴天，25°C”的结果打包成ToolMessage。tools->agent：结果被强行送回给大模型。agent第二次执行：大模型结合历史记录（用户提问 + 工具返回的温度），生成了最终的自然语言回答。agent->should_continue->END：由于这次大模型没有再调用工具，条件边将其导向结束。
六、 总结与下期预告
总结：这一篇我们完成了从“线”到“图”的转变。利用 Node（计算站） 和 Conditional Edge（智能路由），我们轻松实现了复杂的循环判断和反馈闭环。在 LangGraph 的框架下，AI 终于不再是按脚本演戏的演员，而是成为了能根据环境反馈、不断调整策略的“指挥官”。
痛点过渡：
虽然我们的 Agent 变聪明了，但它依然有一个致命弱点：它的记忆转瞬即逝。
如果对话过程中网络中断了，或者 Agent 需要处理一个耗时三天的任务，亦或是我们需要在 AI 执行关键操作（如转账）前让人工介入审批。当前的系统会完全崩溃并丢失进度。
下期预告：
在下一篇文章中，我们将深入探讨 LangGraph 最具企业级价值的核心——持久化机制 (Persistence)。我们将手把手教你如何使用检查点（Checkpoints）让 Agent 拥有长期记忆，并解锁酷炫的 Human-in-the-loop（人工介入） 功能。
敬请期待！
