结论先行
LangGraph 是一个用于构建有状态、可控的 AI Agent 工作流的编排框架。它把 Agent 的逻辑建模成一张图——节点负责干活,边负责控制流程,状态在整张图中流动。对于初学者,建议从最小的“单节点线形图”开始,逐步加入条件分支和循环,最后再接触持久化和人工中断等高级特性。
一、LangGraph 是什么,为什么需要它
传统 LLM 应用通常是“输入 → 处理 → 输出”的线性链式结构。但真实的 Agent 往往需要循环(比如反复调用工具直到任务完成)、条件分支(根据中间结果走不同路径)、人工审批(在关键步骤暂停等待确认)。这些用线性链很难优雅实现。
LangGraph 的核心思路是:把工作流画成一张图。
每个工作流都由 StateGraph 表示,本质上是一个有向图:
节点:封装具体操作(如调用 LLM、执行工具)
边:定义节点之间的跳转规则
状态:在所有节点间流动的共享数据
这样做的好处是:流程一目了然,加减节点不需要重写协调逻辑,状态有类型约束,循环有内置的终止条件。
二、先跑通第一个例子
在深入概念之前,先安装并运行一个最简单的例子。这一步只需要 30 秒,但能帮你建立最直观的感受。
安装:
pipinstall-Ulanggraph 最小示例——两个节点顺序执行,各自往文本后追加一个字母: python from langgraph.graphimportSTART, StateGraph from typing_extensionsimportTypedDict class State(TypedDict): text: str def node_a(state: State)->dict:return{"text":state["text"]+"a"}def node_b(state: State)->dict:return{"text":state["text"]+"b"}graph=StateGraph(State)graph.add_node("node_a", node_a)graph.add_node("node_b", node_b)graph.add_edge(START,"node_a")graph.add_edge("node_a","node_b")app=graph.compile()print(app.invoke({"text":""}))# 输出: {'text': 'ab'}这段代码做了三件事:
定义状态 State:一个 TypedDict,规定图中流动的数据结构。
定义节点 node_a 和 node_b:接收当前状态,返回要更新的字段。
连接图:START → node_a → node_b,编译后即可调用。
三、四个核心概念
3.1 State(状态)
状态是整个图的共享数据容器。所有节点都能读取它,节点返回的字典会被合并回状态中。
关键规则:节点返回的是增量更新,不是完整状态。比如节点返回 {“text”: “新值”},LangGraph 只把 text 字段覆盖掉,其他字段保持不变。
classAgentState(TypedDict):messages:list# 对话历史iteration:int# 当前轮次final_answer:str# 最终答案默认情况下,节点返回的某个字段会覆盖旧值。如果希望累加(比如消息列表不断追加),需要用 Annotated 指定归并函数:
fromtypingimportAnnotatedfromoperatorimportaddclassState(TypedDict):messages:Annotated[list,add]# 新列表会追加,而非覆盖初学者的常见坑:对话历史字段如果不加 Annotated[list, add],第二条消息会直接丢掉第一条。
3.2 Node(节点)
节点是执行具体任务的函数。接收当前状态,返回要更新的字段。
defprocess_input(state:AgentState)->dict:return{"text":state["text"].upper()}节点可以是普通函数,也可以是 LangChain 的 Runnable。每个节点都是无状态的——它只拿到当前状态的快照,返回更新,实际的状态修改由 LangGraph 在外部完成。
3.3 Edge(边)
边决定执行完一个节点后去哪里。
普通边:固定跳转。
graph.add_edge("node_a","node_b")# node_a 执行完必去 node_b条件边:根据状态动态选择下一个节点。
defshould_continue(state:State)->str:ifstate["count"]<3:return"increment"# 继续循环returnEND# 结束graph.add_conditional_edges("increment",should_continue)路由函数返回的字符串会匹配到目标节点。这使得循环变得自然——只需要让条件边指回前面的节点即可。
3.4 Graph(图)
图是节点和边的集合。构建流程:
builder=StateGraph(State)# 1. 创建图,指定状态结构builder.add_node("name",func)# 2. 添加节点builder.add_edge(START,"name")# 3. 定义边builder.add_conditional_edges(...)# 条件边app=builder.compile()# 4. 编译为可执行对象app.invoke({"text":"hello"})# 5. 运行四、构建一个真正的 Agent 循环
有了上面的基础,就可以实现 Agent 最核心的推理→行动→再推理循环了。
以下是一个简化的 ReAct 风格工作流:
fromlanggraph.graphimportStateGraph,START,ENDfromtypingimportAnnotatedfromoperatorimportaddclassAgentState(TypedDict):messages:Annotated[list,add]# 累加对话iteration:intdefagent_node(state:AgentState)->dict:"""Agent 的“大脑”:决定是调用工具还是直接回答"""# 实际场景中这里会调用 LLMlast=state["messages"][-1]ifstate["messages"]elseNoneiflastand"需要工具"instr(last):return{"messages":["调用工具..."],"iteration":state["iteration"]+1}return{"messages":["任务完成"],"iteration":state["iteration"]+1}deftool_node(state:AgentState)->dict:"""执行工具,返回结果"""return{"messages":["工具执行结果"]}defshould_continue(state:AgentState)->str:"""决定:回到 agent 继续推理,还是结束"""ifstate["iteration"]>=3:# 简化:最多循环 3 次returnEND last_msg=state["messages"][-1]ifstate["messages"]else""if"工具"instr(last_msg):return"tools"returnEND builder=StateGraph(AgentState)builder.add_node("agent",agent_node)builder.add_node("tools",tool_node)builder.add_edge(START,"agent")builder.add_conditional_edges("agent",should_continue,{"tools":"tools",END:END})builder.add_edge("tools","agent")# 关键:工具执行完回到 agent,形成循环app=builder.compile()result=app.invoke({"messages":["需要工具帮忙"],"iteration":0})print(result["messages"])关键设计:tools 节点执行完后指回 agent,形成循环。should_continue 负责在合适的时机跳出循环。
五、进阶但重要的概念
5.1 Checkpointer(检查点):让状态可持久化
默认情况下,每次 invoke 都从头开始,状态用完即弃。加上 Checkpointer 后,每一步的状态都会被保存。
fromlanggraph.checkpoint.memoryimportMemorySaver checkpointer=MemorySaver()# 开发用;生产建议用 SqliteSaver 或 PostgresSaverapp=builder.compile(checkpointer=checkpointer)config={"configurable":{"thread_id":"user-001"}}app.invoke({"messages":["你好"]},config)# 状态被保存app.invoke({"messages":["继续"]},config)# 同一 thread_id,上一轮状态自动恢复thread_id 区分不同会话,实现短期记忆——同一用户的多轮对话上下文得以保持。
5.2 Human-in-the-loop(人工中断)
在关键操作前暂停,等待人工确认:
fromlanggraph.typesimportinterrupt,Commanddefreview_node(state):decision=interrupt({"question":"是否批准?","data":state["draft"]})ifdecision=="approve":return{"status":"approved"}return{"status":"rejected"}# 恢复时传入人工决定app.invoke(Command(resume="approve"),config)5.3 状态归并的细节
默认覆盖:节点返回 {“text”: “新值”},旧值被替换。
归并累加:用 Annotated[list, add] 让列表追加而非覆盖。
强制覆盖:用 Overwrite 绕过归并函数,直接替换。
重要提醒:如果字段用了归并函数(如 add),返回空列表不会清空该字段——空列表会被合并进去,旧值还在。要清空必须用 Overwrite([])。
六、初学者学习路径建议
最关键的一步是亲手写一个包含循环的工作流。LangGraph 的真正价值在于控制流,而不是单次 LLM 调用。当你能让一个图根据中间结果决定“继续循环”还是“退出”时,就已经掌握了它的核心。
七、一张图记住核心概念
┌─────────────────────────────────────────────────────┐ │ StateGraph │ │ │ │ START ──▶ [节点 A] ──▶ [节点 B] ──▶ END │ │ │ │ │ │ │ (条件边) │ │ │ │ ▼ │ │ └────── [节点 C] ◀── 循环 ──────────────┘│ │ │ │ State: 贯穿全图的共享数据,节点返回增量更新 │ │ Node: 执行任务的函数,接收状态,返回更新 │ │ Edge: 普通边(固定跳转) / 条件边(动态路由) │ │ Checkpointer: 状态快照,支持恢复和记忆 │ └─────────────────────────────────────────────────────┘LangGraph 的设计哲学是“低层级、高控制”。它不替你决定 Agent 该怎么想,只负责可靠地执行你画的流程。作为初学者,从简单的图开始,逐步增加复杂度,是最有效的学习方式。