LangGraph V1.0.5 核心文档精炼版第1章 LangGraph 概述1.1 什么是LangGraph核心知识点：LangChain团队推出的低级别智能体编排框架与运行时，基于图模型构建有状态Agent，核心由**节点（计算单元）、边（流程转换）、状态（全局共享数据）**三大组件构成。原生支持持久化执行、人机协作、长短时记忆管理、流式处理与生产级部署，可独立使用，也可与LangChain/LangSmith无缝集成，相比LangChain提供更细粒度的流程控制。核心资源GitHub：https://github.com/langchain-ai/langgraph官方文档：https://docs.langchain.com/oss/python/langgraph/overview1.2 为什么需要LangGraph核心知识点：传统Agent框架/LCEL线性序列，存在状态管理混乱、执行流程不可控、错误恢复困难的问题，无法满足复杂循环、多条件交互的Agent需求。LangGraph通过图计算模型+状态机理念，解决线性编排的局限性，适配生产级AI Agent构建。1.3 核心使用场景核心知识点：复杂多智能体系统、需长期记忆的应用、人工审核工作流、后台处理+实时交互任务、需精细控制的定制化智能体编排。1.4 与LangChain 核心区别特性LangGraphLangChain抽象级别低级，细粒度流程控制高级，开箱即用的链式封装状态管理内置状态机+检查点，原生支持需开发者自行实现执行模型基于图的并行执行线性链式执行持久化原生内置需额外开发实现适用场景复杂、有状态的智能体应用简单链式调用开发1.5 架构设计核心知识点：Pregel架构：运行时基于Google Pregel并行图计算模型，执行分3个阶段，每轮执行称为一个超步(super-step)：Plan：确定本轮要执行的节点Execution：并行执行所有选中节点Update：将节点输出更新到通信通道Actors(PregelNode)：订阅通道、读写数据的节点，实现LangChain Runnable接口。Channels：节点间通信载体，分为三类：LastValue：存储最后发送的值Topic：可配置的发布-订阅主题BinaryOperatorAggregate：用于聚合操作第2章 快速入门2.1 环境安装核心知识点：通过conda创建隔离环境，pip安装langgraph核心包。核心代码# 1. 创建环境conda create-n langgraph python==3.12conda activate langgraph# 2. 安装核心包pip install-U langgraph# 3. 验证安装pip show langgraph2.2 最小可用示例核心知识点：基于StateGraph完成图的定义、节点/边添加、编译与调用，基于MessagesState实现最简对话流。核心代码fromlanggraph.graphimportStateGraph,MessagesState,START,END# 1. 定义节点函数（LLM调用/自定义逻辑）defmock_llm(state:MessagesState):return{"messages":[{"role":"ai","content":"hello world"}]}# 2. 初始化状态图graph=StateGraph(MessagesState)# 3. 添加节点与边graph.add_node(mock_llm)graph.add_edge(START,"mock_llm")graph.add_edge("mock_llm",END)# 4. 编译图graph=graph.compile()# 5. 调用执行response=graph.invoke({"messages":[{"role":"user","content":"hi!"}]})print(response)第3章 Graph API3.1 State核心知识点：State是图的全局共享数据结构，由**Schema（结构定义）和Reducer（更新逻辑）**组成，是所有节点/边的输入基础，核心类为StateGraph。3.1.1 Schema核心知识点：定义图的状态结构，支持TypedDict/Pydantic定义，分为三类：state_schema：图的完整内部状态，必填，所有节点可读写input_schema：输入校验规则，state_schema的子集，可选output_schema：输出过滤规则，state_schema的子集，可选支持节点间私有状态传递，不写入全局state_schema核心代码fromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,END# 1. 定义输入/输出/全局状态classInputState(TypedDict):question:strclassOutputState(TypedDict):answer:strclassOverallState(InputState,OutputState):pass# 2. 定义节点defanswer_node(state:InputState):answer="再见"if"bye"instate["question"].lower()else"你好"return{"answer":answer,"question":state["question"]}# 3. 构建图并指定输入输出Schemabuilder=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()# 4. 调用result=graph.invoke({"question":"你好"})3.1.2 Reducer核心知识点：定义状态字段的更新合并逻辑，每个字段独立配置，决定节点返回值如何与全局状态合并，核心类型如下：Reducer类型核心作用默认行为未指定时，直接覆盖更新字段值add_messages消息列表专用，追加LangChain消息对象operator.add列表追加、字符串拼接、数值累加operator.mul数值相乘，原生设计存在默认值bug，需自定义实现自定义Reducer开发者自定义合并逻辑核心代码# 1. 默认覆盖Reducerfromtyping_extensionsimportTypedDictclassDefaultState(TypedDict):foo:intdefnode_default(state:DefaultState):return{"foo":2}# 2. add_messages ReducerfromtypingimportAnnotated,Listfromlanggraph.graph.messageimportadd_messagesclassMsgState(TypedDict):messages:Annotated[List,add_messages]defchat_node(state:MsgState):return{"messages":[("assistant","Hello")]}# 3. operator.add ReducerimportoperatorclassAddState(TypedDict):data:Annotated[List[int],operator.add]count:Annotated[int,operator.add]defadd_node(state:AddState):return{"data":[1,2],"count":5}# 4. 自定义Reducerdefcustom_reducer(current_value:dict,new_value:dict)-dict:result=current_value.copy()result.update(new_value)returnresultclassCustomState(TypedDict):metadata:Annotated[dict,custom_reducer]defcustom_node(state:CustomState):return{"metadata":{"version":"1.0"}}3.2 Nodes核心知识点：节点是承载计算逻辑的同步/异步函数，入参包含state、config、runtime，通过add_node添加到图中，支持丰富的高级配置。3.2.1 特殊节点核心知识点：START：虚拟起始节点，定义图的入口END：虚拟终止节点，定义图的结束3.2.2 节点缓存核心知识点：基于节点输入实现结果缓存，避免重复执行耗时计算，支持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调用、数据库查询等不稳定场景，可指定最大重试次数、自定义重试异常条件。核心代码fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportRetryPolicyfromtyping_extensionsimportTypedDictclassState(TypedDict):result:strdefunstable_api_call(state:State):# 模拟不稳定API调用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,retry_on=custom_retry_on))builder.add_edge(START,"unstable_call")builder.add_edge("unstable_call",END)graph=builder.compile()3.2.4 延迟节点执行核心知识点：通过defer=True设置节点延迟执行，确保该节点等待所有并行分支任务完成后再执行，适配分支长度不一致的汇总场景。核心代码importoperatorfromtypingimportAnnotatedfromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassState(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"]}# 构建图，设置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核心知识点：边定义节点间的路由与执行顺序，分为4类核心类型，支持循环创建与递归限制。3.3.1 普通边核心知识点：无条件固定连接两个节点，源节点执行完成后，固定跳转到目标节点。核心代码fromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassGraphState(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 条件边核心知识点：根据当前状态动态决定下一个执行节点，通过路由函数返回目标节点名称，支持路由别名映射。核心代码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