LangGraph 入门Agent 编排框架到底是什么写在前面最近半年AI Agent 这个概念火得一塌糊涂。从 AutoGPT 到各种自主智能体开发者们发现了一个残酷的现实调用 LLM 很简单让 LLM 靠谱地完成一个多步骤任务难。问题出在哪一个典型 Agent 的工作流长这样收到用户问题 → 判断是否需要查资料 → 调用搜索工具 → 分析结果 → 发现需要更多信息 → 再查一次 → 综合回答每一步都可能出错每一步都可能超时有些步骤需要人确认才能继续。如果你用if-else硬写这个流程代码很快会变成一坨不可维护的意大利面。LangGraph 就是来解决这个问题的。它不是又一个 LLM API 封装而是一个Agent 编排框架——准确地说它是编排框架的底层运行时。这篇文章是 LangGraph 系列的第一篇我们不讲复杂概念先搞清楚LangGraph 是谁、能干什么、怎么上手。1. Agent 编排为什么需要专门一个框架先想想不用框架你怎么写一个 Agentdef agent(user_input: str): messages [{role: user, content: user_input}] max_loops 10 for i in range(max_loops): response llm.invoke(messages) if response.tool_calls: for tc in response.tool_calls: result execute_tool(tc) messages.append({role: tool, content: result}) else: return response.content return 超时了看起来还行那再想想这些需求往上加对话持久化用户离开了回来还能继续人工审批执行某些操作前必须等人确认流式输出逐字显示 LLM 的思考过程故障恢复程序崩溃后自动从断点恢复状态回放出错了回溯到前一步检查状态多 Agent 协作两个 Agent 接力完成任务如果把这些全塞进上面那个for循环里代码量会暴增而且每加一个特性就要改整个结构。这就是编排框架的价值。它不替你写业务逻辑但替你管理状态怎么流转、失败了怎么办、能不能让人插一脚这些脏活。2. LangGraph 在 LangChain 生态中的角色LangGraph 是 LangChain 公司开发的但很多人搞不清楚它和 LangChain 本身有什么区别。我用一张图解释┌─────────────────────────────────────────────────┐ │ Deep AgentsAgent Harness │ │ 规划、子Agent、文件系统、上下文管理 │ ├─────────────────────────────────────────────────┤ │ LangChainAgent Framework │ │ 模型抽象、工具抽象、预置 Agent 架构 │ ├─────────────────────────────────────────────────┤ │ LangGraphOrchestration Runtime【← 我们在这】 │ │ 持久化、流式、Human-in-the-loop、状态管理 │ └─────────────────────────────────────────────────┘三层各干各的层级框架一句话概括HarnessDeep Agents拿来即用的完整 Agent 方案FrameworkLangChain模型/工具/链的乐高积木RuntimeLangGraph⬅️底层编排引擎管理状态和流程LangGraph 是最底层的那块。它不关心你用哪个 LLM、不关心你用什么工具、不替你设计 prompt。它只关心一件事Agent 的状态怎么安全、可靠、可恢复地在节点之间流转。有意思的是LangChain 内置的 Agent比如create_react_agent底层也是用 LangGraph 实现的。所以学 LangGraph 不只是学一个新框架——你在学 LangChain 生态的底层引擎。3. LangGraph 的设计灵感LangGraph 不是凭空造出来的它站在三个巨人的肩膀上Google Pregel—— Google 2010 年发表的图计算论文。Pregel 的核心思想是以节点为中心的图计算每轮迭代称为 Superstep中所有活跃节点并行执行然后同步等待。LangGraph 借用了这个Superstep的概念。Apache Beam—— 统一批处理和流处理的数据管道框架。LangGraph 的流式设计受到了 Beam 的启发。NetworkX—— Python 最流行的图分析库。LangGraph 的 API 命名add_node、add_edge就是向 NetworkX 致敬。所以可以这么理解LangGraph 用 Pregel 的图计算思想 像 NetworkX 一样声明节点和边 专为 LLM Agent 定制。4. 核心概念一句话说清楚LangGraph 就三个概念StateGraph State Node EdgeState状态—— 一张全局黑板所有节点共享。它存着当前对话的消息、Agent 的临时变量、运行计数等。每个节点可以读它也可以往上面写东西。Node节点—— 一个普通函数。接收当前 State干点活调用 LLM、执行工具、查数据库然后返回 State 的更新。Edge边—— 就是箭头。告诉 LangGraph这个节点跑完后下一步去哪个节点。可以是固定的一条路也可以是看 State 情况决定的条件分支。节点干活的边指路的State 记事的。就这么简单。5. 环境搭建pip install -U langgraph如果需要用 LangChain 集成 LLM后面几期会用到pip install -U langgraph langchain langchain-openai安装完成后验证一下python -c import langgraph; print(langgraph.__version__)小提示推荐在虚拟环境里安装。python -m venv lg_env source lg_env/bin/activateLinux/Mac或lg_env\Scripts\activateWindows。6. 第一个 LangGraph 程序Hello World理论讲够了我们直接写代码。这个例子不需要任何 API Key纯本地运行。from langgraph.graph import StateGraph, MessagesState, START, END def mock_llm(state: MessagesState): 假装是个 LLM收到消息就回个 hello world return {messages: [{role: ai, content: hello world}]} # 第 1 步创建图 graph_builder StateGraph(MessagesState) # 第 2 步添加节点 graph_builder.add_node(llm, mock_llm) # 第 3 步添加边 —— 从 START 到 llm再从 llm 到 END graph_builder.add_edge(START, llm) graph_builder.add_edge(llm, END) # 第 4 步编译 graph graph_builder.compile() # 第 5 步运行 result graph.invoke({messages: [{role: user, content: hi!}]}) print(result)跑一下看看输出{ messages: [ HumanMessage(contenthi!, ...), AIMessage(contenthello world, ...) ] }来一行行拆解代码在干什么StateGraph(MessagesState)创建一个图规定状态格式为消息列表add_node(llm, mock_llm)注册一个叫 “llm” 的节点节点函数是mock_llmadd_edge(START, llm)图的入口指向 “llm” 节点add_edge(llm, END)“llm” 节点执行完结束graph.compile()编译检查生成可调用的执行器graph.invoke(...)传入初始状态让图跑起来用你熟悉的工具来类比的话compile()就像是 React 的编译——不是真的编译成机器码而是做一些校验和准备工作。7. 可视化你的图让人看一眼就明白你的 Agent 长什么样LangGraph 自带可视化# 需要有 IPython 环境Jupyter Notebook / VS Code 交互窗口 from IPython.display import Image, display # xrayTrue 会展开子图结构 display(Image(graph.get_graph(xrayTrue).draw_mermaid_png()))你会看到一张类似于这样的图[START] → [llm] → [END]看起来简单那是因为只有一个节点。等到后面有多个节点、条件分支、循环的时候这个可视化会变得非常有价值——你一眼就能看出 Agent 的决策逻辑。8. 再加点逻辑做一个有状态的计数器单节点太没意思了我们加点料。下面这个例子是一个有状态的计数器——每次调用都会自增计数from typing import TypedDict from langgraph.graph import StateGraph, START, END class CounterState(TypedDict): 自定义状态只存一个计数 count: int def increment(state: CounterState) - CounterState: 计数 1 current state.get(count, 0) return {count: current 1} # 构建图 builder StateGraph(CounterState) builder.add_node(add_one, increment) builder.add_edge(START, add_one) builder.add_edge(add_one, END) graph builder.compile() # 测试连续调用 3 次 # 注意这里每次 invoke 都是独立的新调用 for i in range(3): result graph.invoke({count: 0}) print(f第 {i1} 次调用结果: {result})输出第 1 次调用结果: {count: 1} 第 2 次调用结果: {count: 1} 第 3 次调用结果: {count: 1}等等这不是每次调用都从 0 开始吗没错因为每一次invoke都是独立运行的我们传入了{count: 0}。要实现跨调用的记忆需要引入Checkpointer这是第 6 期的内容先留个悬念。9. MessagesState 是什么你可能注意到了第一个例子用了MessagesState第二个例子用了自定义的TypedDict。其实MessagesState本质就是from typing import Annotated from typing_extensions import TypedDict import operator from langchain.messages import AnyMessage class MessagesState(TypedDict): messages: Annotated[list[AnyMessage], operator.add]关键在Annotated[list[AnyMessage], operator.add]这里的operator.add——它是一个Reducer归约器。什么意思呢每次节点返回新的消息时不是覆盖旧消息而是追加。这样你每次调用 LLM 返回的AIMessage都自动追加到messages列表末尾。如果你在这里有点懵没关系第 2 期会专门讲 State 和 Reducer。10. 常见坑点一览坑 1不加 Reducer 会被覆盖class BadState(TypedDict): messages: list # 没加 Annotated # 如果节点返回 {messages: [new_msg]} # 它会盖掉之前的 messages不是追加坑 2忘记 compilebuilder StateGraph(MyState) builder.add_node(...) # builder.compile() ← 忘了这步 # graph.invoke(...) → AttributeError坑 3Node 函数必须返回 dictdef my_node(state): # ✅ 正确 return {key: value} def my_node(state): # ❌ 错误 return hello