LangGraph V1.0.5 文档精炼版

LangGraph V1.0.5 文档精炼版

第1章 LangGraph 概述

1.1 什么是LangGraph

知识点精炼：LangChain团队推出的低级别智能体编排框架与运行时，以图结构建模Agent工作流，核心由节点（计算单元）、边（流程转换逻辑）、状态（全局共享数据）三大组件构成；原生支持持久化执行、人机协作、记忆管理、流式处理，可独立使用或与LangChain/LangSmith无缝集成，提供比LangChain更细粒度的执行流程控制。

1.2 为什么需要LangGraph

知识点精炼：解决传统Agent框架状态管理混乱、执行流程不可控、错误恢复困难的痛点；弥补LCEL线性序列在复杂循环、多条件交互场景的局限性，基于图计算与状态机理念提供生产级AI Agent解决方案。

1.3 使用场景

知识点精炼：核心适用场景包括复杂多智能体系统、需长期记忆的应用、人工审核工作流、后台处理+实时交互任务、需精细控制的定制化智能体编排。

1.4 与LangChain区别

1.5 架构设计

知识点精炼：基于Google Pregel大规模并行图计算算法，核心分为三层：

1. Pregel运行时：执行分Plan（规划本轮执行节点）、Execution（并行执行选中节点）、Update（节点输出更新到通道）三个阶段，每轮执行称为一个「超步」；
2. Actors（PregelNode）：订阅通道、读写数据的计算节点，实现LangChain的Runnable接口；
3. Channels（通道）：节点间通信载体，支持LastValue（最新值存储）、Topic（发布-订阅主题）、BinaryOperatorAggregate（聚合操作）三类。

第2章 快速入门

2.1 环境安装

知识点精炼：通过conda创建Python3.12隔离环境，pip安装langgraph包，通过pip show验证安装结果。
核心代码：

# 创建隔离环境conda create-nlanggraphpython==3.12conda activate langgraph# 安装最新版langgraphpipinstall-Ulanggraph# 验证安装pip show langgraph

2.2 简单示例

知识点精炼：基于StateGraph实现最简工作流，核心步骤为：定义节点函数、创建状态图、添加节点与边、编译图、调用执行。
核心代码：

fromlanggraph.graphimportStateGraph,MessagesState,START,END# 1. 定义节点函数defmock_llm(state:MessagesState):return{"messages":[{"role":"ai","content":"hello world"}]}# 2. 构建状态图graph=StateGraph(MessagesState)graph.add_node(mock_llm)graph.add_edge(START,"mock_llm")graph.add_edge("mock_llm",END)# 3. 编译并调用图graph=graph.compile()response=graph.invoke({"messages":[{"role":"user","content":"hi!"}]})

第3章 Graph API

3.1 State

知识点精炼：State是图的全局共享数据结构，由schema（状态结构）和reducer函数（状态更新规则）组成；是所有节点、边的输入基础，核心载体为StateGraph类。

3.1.1 Schema

知识点精炼：定义状态的结构规范，支持TypedDict/Pydantic模型，分为三类：

1. state_schema：图的完整内部状态，必填，所有节点可访问读写；
2. input_schema：可选，图的入参规范，为state_schema子集；
3. output_schema：可选，图的出参规范，为state_schema子集；
   支持节点间私有状态传递，私有数据仅指定节点可访问，不写入全局状态。
   核心代码：

fromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,END# 输入/输出/全局状态定义classInputState(TypedDict):question:strclassOutputState(TypedDict):answer:strclassOverallState(InputState,OutputState):pass# 节点定义defanswer_node(state:InputState):return{"answer":"你好","question":state["question"]}# 构建图builder=StateGraph(OverallState,input_schema=InputState,output_schema=OutputState)builder.add_node("answer_node",answer_node)builder.add_edge(START,"answer_node")builder.add_edge("answer_node",END)graph=builder.compile()

3.1.2 Reducer

知识点精炼：Reducer是状态更新的核心规则，每个状态键对应独立reducer，控制节点输出与全局状态的合并逻辑，保障并行更新的数据一致性，支持自定义扩展。

核心代码：

# 1. 默认覆盖fromtyping_extensionsimportTypedDictclassDefaultReducerState(TypedDict):foo:intdefnode_default(state:DefaultReducerState):return{"foo":2}# 2. add_messagesfromtypingimportAnnotated,Listfromlanggraph.graph.messageimportadd_messagesclassAddMessagesState(TypedDict):messages:Annotated[List,add_messages]defchat_node(state:AddMessagesState):return{"messages":[("assistant","Hello")]}# 3. operator.addimportoperatorclassListAddState(TypedDict):data:Annotated[List[int],operator.add]classNumberAddState(TypedDict):count:Annotated[int,operator.add]# 4. 自定义ReducerfromtypingimportDict,Anydefcustom_reducer(current_value:Dict[str,Any],new_value:Dict[str,Any])->Dict[str,Any]:result=current_value.copy()result.update(new_value)returnresultclassCustomReducerState(TypedDict):metadata:Annotated[Dict[str,Any],custom_reducer]

3.2 Nodes

知识点精炼：节点是承载计算逻辑的Python函数（支持同步/异步），入参为state、config、runtime，通过add_node添加到图中，是图的核心计算单元。

3.2.1 START/END Node

知识点精炼：START是虚拟起始节点，定义图的执行入口；END是虚拟终止节点，定义图的执行终点，均无实际计算逻辑。

3.2.2 Node Caching

知识点精炼：支持基于节点输入的结果缓存，可自定义缓存键生成规则与TTL，避免重复执行耗时计算。
核心代码：

fromlanggraph.graphimportStateGraphfromlanggraph.cache.memoryimportInMemoryCachefromlanggraph.typesimportCachePolicyfromtyping_extensionsimportTypedDictclassState(TypedDict):x:intresult:intdefexpensive_node(state:State):return{"result":state["x"]*2}# 构建图并配置缓存builder=StateGraph(State)builder.add_node("expensive_node",expensive_node,cache_policy=CachePolicy(ttl=10))builder.set_entry_point("expensive_node")builder.set_finish_point("expensive_node")graph=builder.compile(cache=InMemoryCache())

3.2.3 添加重试策略

知识点精炼：为节点配置自定义重试策略，指定最大重试次数、重试异常类型，适用于API调用、数据库查询等不稳定操作；默认策略排除ValueError、TypeError等非可重试异常。
核心代码：

fromlanggraph.graphimportStateGraphfromlanggraph.typesimportRetryPolicyfromtyping_extensionsimportTypedDictclassState(TypedDict):result:strdefunstable_api_call(state:State):raiseException("模拟API调用失败")# 自定义重试条件defcustom_retry_on(exception:Exception)->bool:return"模拟API调用失败"instr(exception)# 构建图builder=StateGraph(State)# 默认重试策略builder.add_node("unstable_call",unstable_api_call,retry_policy=RetryPolicy(max_attempts=5))# 自定义重试策略builder.add_node("custom_retry_call",unstable_api_call,retry_policy=RetryPolicy(max_attempts=5,retry_on=custom_retry_on))

3.2.4 延迟节点执行

知识点精炼：通过defer=True设置节点延迟执行，确保该节点等待所有并行分支任务完成后再执行，适用于分支长度不一致的汇总场景。
核心代码：

fromlanggraph.graphimportStateGraph,START,ENDfromtyping_extensionsimportTypedDictimportoperatorfromtypingimportAnnotatedclassState(TypedDict):aggregate:Annotated[list,operator.add]defa(state:State):return{"aggregate":["A"]}defb(state:State):return{"aggregate":["B"]}defb_2(state:State):return{"aggregate":["B_2"]}defc(state:State):return{"aggregate":["C"]}# 延迟执行的汇总节点defd(state:State):return{"aggregate":["D"]}# 构建图builder=StateGraph(State)builder.add_node("a",a)builder.add_node("b",b)builder.add_node("b_2",b_2)builder.add_node("c",c)builder.add_node("d",d,defer=True)# 标记延迟执行# 构建边builder.add_edge(START,"a")builder.add_edge("a","b")builder.add_edge("a","c")builder.add_edge("b","b_2")builder.add_edge("b_2","d")builder.add_edge("c","d")builder.add_edge("d",END)graph=builder.compile()

3.3 Edges

知识点精炼：边定义节点间的路由规则与执行流程，分为普通边、条件边、入口点、条件入口点四类；一个节点可有多条出边，目标节点将并行执行。

3.3.1 Normal Edges

知识点精炼：普通边为节点间的无条件跳转，源节点执行完成后，直接跳转到目标节点执行。
核心代码：

fromlanggraph.graphimportStateGraph,START,ENDfromtyping_extensionsimportTypedDictclassGraphState(TypedDict):value:intdefnode_a(state:GraphState):return{"value":state["value"]+1}defnode_b(state:GraphState):return{"value":state["value"]*2}builder=StateGraph(GraphState)builder.add_node("node_a",node_a)builder.add_node("node_b",node_b)# 添加普通边builder.add_edge(START,"node_a")builder.add_edge("node_a","node_b")builder.add_edge("node_b",END)graph=builder.compile()

3.3.2 Conditional Edges

知识点精炼：条件边根据当前状态动态决定下一个执行节点，通过路由函数返回目标节点标识，支持自定义路由映射。
核心代码：

fromtypingimportLiteralfromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassGraphState(TypedDict):value:intdefnode_a(state:GraphState):return{"value":state["value"]+1}defnode_b(state:GraphState):return{"value":state["value"]*2}defnode_c(state:GraphState):return{"value":state["value"]-1}# 路由函数defroute_condition(state:GraphState)->Literal["node_b","node_c"]:return"node_b"ifstate["value"]%2==0else"node_c"# 构建图builder=StateGraph(GraphState)builder.add_node("node_a",node_a)builder.add_node("node_b",node_b)builder.add_node("node_c",node_c)builder.add_edge(START,"node_a")# 添加条件边builder.add_conditional_edges