1.节点在 LangGraph 中节点Node是图里执行具体工作的基本单元。结合状态与 Pregel 算法节点就是那个读取状态、执行逻辑、然后返回状态更新的“执行者”。核心定义与签名从代码层面看节点的本质是一个函数或可调用对象其签名约定为接收当前状态返回部分更新。输入节点拿到的是当前步骤的状态快照。输出节点不需要返回完整的状态只需返回它想要修改的字段Partial Update。LangGraph 会负责将这些更新合并回全局状态中。节点的常见类型与实现形式你可以把任何逻辑封装成节点常见的实现形式包括普通函数最直接的方式。你可以写一个 Python 或 JS 函数在里面调用大模型、查询数据库或运行工具然后返回结果。LangChain Runnable任何遵循 LangChain Runnable 接口的对象如 Chain、Retriever都可以直接作为节点使用。可调用类包含__call__方法的类实例也可以作为节点适合需要维护内部状态或复杂初始化的场景。异步节点支持async def定义的异步函数用于处理并发的 I/O 操作如并行调用多个 API。特殊节点与高级特性除了用户自定义的普通节点LangGraph 还提供了一些内置或特殊类型的节点预构建节点LangGraph 提供了开箱即用的节点例如ToolNode。它专门用于自动执行工具调用你无需自己编写工具调用的循环逻辑直接将其加入图中即可。子图作为节点一个编译后的图本身也是一个 Pregel 实例可以被嵌套在另一个图的节点中运行。这允许你将复杂的流程模块化构建分层级的图结构。特殊路由节点Command虽然通常由边来决定路由但节点也可以通过返回Command对象来同时更新状态并决定下一步跳转到哪个节点从而将控制流逻辑内聚在节点内部。与 Pregel 算法的关系对于 Pregel 超级步模型节点就是那个在“执行阶段”被调度的对象。在一个超级步中被选中的所有节点会并行执行它们各自基于同一个状态快照进行计算执行完毕后它们返回的更新会在“更新阶段”被统一提交。2.节点的输入输出2.1 节点输入– 在LangGraph中一般来讲节点都是 Python 函数可以是同步的也可以是异步的它们接受以下参数state图的状态代表了具体的业务数据config一个RunnableConfig对象包含诸如thread_id之类的配置信息在调用图时也可以传递用户定义的其他配置runtime一个Runtime对象包含运行时context可自定context在调用图时传入即可以及其他信息如store和stream_writer等2– 以上参数会在运行过程中自动被LangGraph运行时注入。– 节点定义config和runtime入参如下所示在运行时这两个参数LangGraph名称会通过关键字传参方式进行传参因此两个参数位置可以调换。具体示例如下# 导入time模块用来模拟处理耗时importtime# TypedDict定义状态结构Any任意类型List列表注解fromtypingimportTypedDict,Any,List# RunnableConfigLangGraph标准的配置类型fromlangchain_core.runnablesimportRunnableConfig# StateGraph构建状态图START起点END终点fromlanggraph.graphimportStateGraph,START,END# RuntimeLangGraph运行时对象用来传递context上下文、流式输出writerfromlanggraph.runtimeimportRuntime# 模拟大模型类代替真实LLMclassMockLLM:# invoke方法接收prompt直接返回拼接好的模拟回答definvoke(self,prompt:str):returnfAI Generated: Answer for {prompt}# 模拟数据库类classMockDatabase:# 根据user_id获取用户信息简单判断是否是vip用户defget_user_info(self,user_id:str):return{id:user_id,role:vipifvipinuser_idelsestandard}# 定义客服流程的状态结构classCustomerSupportState(TypedDict):query:str# 用户原始提问response:str# 最终给用户的客服回复log:List[str]# 流程处理日志记录关键步骤# 客服业务节点函数# state状态config配置放user_id等runtime运行时对象承载context和stream writerdefnode_customer_service(state:CustomerSupportState,config:RunnableConfig,runtime:Runtime)-dict: 客服节点展示如何从 config 中获取注入的依赖 (LLM, DB) 并使用 runtime 推送进度。 # 取出用户问题user_querystate[query]# 1、从runtime当中获取context对象context就是我们外部注入进来的依赖llmruntime.context[llm]dbruntime.context[db]# 取出config里的configurable配置块configurableconfig.get(configurable,{})# 获取用户id取不到就默认是guest访客user_idconfigurable.get(user_id,guest)print(f\n[Node] 开始处理User ID:{user_id})# 2、验证依赖是否存在做简单容错ifnotllmornotdb:return{response:System Error: Dependencies not injected!,log:[Error]}# 3、使用db对象查看用户角色user_infodb.get_user_info(user_id)user_tieruser_info.get(role)print(f[Node] 从 DB 获取用户角色:{user_tier})# 4、使用 runtime.writer 向图外输出自定义流式数据可以前端展示进度、状态writerruntime.stream_writer writer({status:thinking,message:f正在调用 LLM 为{user_tier}用户生成回复...})# 休眠0.5秒模拟LLM思考耗时time.sleep(0.5)# 5、根据用户角色构建不同的 Prompt并模拟 LLM 调用promptfUser({user_tier}) asks:{user_query}llm_responsellm.invoke(prompt)# 返回字典更新state中的response和logreturn{response:llm_response,log:[fProcessed by{llm.__class__.__name__}]}# 这个return永远不会执行属于冗余代码可以删掉returnNone# 运行演示的入口函数defrun_demo():# 创建图构造器绑定状态结构builderStateGraph(CustomerSupportState)# 注册客服节点builder.add_node(customer_service,node_customer_service)# 连线起点到客服节点builder.add_edge(START,customer_service)# 客服节点执行完直接结束builder.add_edge(customer_service,END)# 编译图graphbuilder.compile()# 初始化外部依赖对象模拟LLM、模拟DBmy_llmMockLLM()my_dbMockDatabase()print()print(场景: 依赖注入演示)print(演示目标: 将 LLM 和 DB 对象通过 runtime.context 注入到节点中)print()# 准备初始状态用户的提问initial_state{query:如何升级会员}# user_id为一个配置信息将其放入到config当中config{configurable:{user_id:vip_user_999,}}print([System] 开始运行图并注入依赖对象...)# 组装要注入的上下文依赖会放进runtime.context给节点使用context{llm:my_llm,db:my_db}# 执行图额外传入context参数实现依赖注入resultgraph.invoke(initial_state,configconfig,contextcontext)# 打印最终返回的状态结果print(result)if__name____main__:run_demo()运行结果 :2.2 节点输出节点的输出为当前节点对状态的增量更新而不能将接收到的整个状态实例返回出去。LangGraph 底层会将所有节点输出的状态都作为增量状态并尝试和当前的全局状态做一次合并操作。如果将整个状态都输出对于没有配置任何 reducer 的状态键langgraph 底层无法完成合并会抛出异常对于配置了 reducer 的状态键如果节点输出了不属于该节点更新的状态也会导致数据产生问题。以下演示错误返回了整个状态的问题# 导入类型注解工具 TypedDict用来定义状态结构fromtypingimportTypedDict# 导入LangGraph内置常量START 图起点、END 图终点fromlanggraph.constantsimportSTART,END# 导入状态图核心类 StateGraphfromlanggraph.graphimportStateGraph# 定义图的状态结构体 MyState继承 TypedDict# 状态里保存整个流程流转过程中所有需要共享的数据classMyState(TypedDict):query:str# 用户原始提问file_result:str# 本地文件检索结果web_result:str# 联网搜索结果final_answer:str# LLM 融合信息后的最终回答defquery_web(state:MyState)-dict: 【节点函数】做网络搜索返回搜索结果 ⚠️ 错误演示写法直接return整个state而非只返回需要增量修改的字段 LangGraph 的设计理念节点只返回【要更新的部分状态】而不是把完整state返回 # 从传入的共享状态中取出用户问题querystate[query]# 修改state里的web_result字段state[web_result]f{query}的网络搜索结果# 把整个state字典返回不推荐的写法仅用来演示returnstatedefquery_file(state:MyState)-dict: 【节点函数】做本地文件检索返回文件内检索结果 ⚠️ 和上面一样直接返回完整 state属于不推荐写法 # 获取用户查询问题querystate[query]# 填充文件检索结果到状态state[file_result]f{query}的文件搜索结果# 返回完整statereturnstatedefanswer(state:MyState)-dict: 【节点函数】汇总web和文件两路结果交给LLM生成最终答案 这个节点会在 query_web、query_file 两个节点都跑完之后才执行 # 读取两路检索结果web_resultstate[web_result]file_resultstate[file_result]# 拼接生成最终回答final_answerfLLM基于{web_result}{file_result}的最终结果# 写入状态state[final_answer]final_answerreturnstate# 实例化状态图指定状态结构为 MyStategraphStateGraph(state_schemaMyState)# 向图中注册3个节点节点名默认和函数名一致graph.add_node(answer)graph.add_node(query_web)graph.add_node(query_file)# 【并行分支】START起点同时指向 query_web 和 query_file# LangGraph 这里会并行执行这两个节点双分支并行检索graph.add_edge(START,query_web)graph.add_edge(START,query_file)# 两个分支都执行完成之后才进入 answer 节点汇聚节点graph.add_edge(query_web,answer)graph.add_edge(query_file,answer)# 编译图生成可执行的runnable对象compiled_graphgraph.compile()# 调用图传入初始状态只有query字段rescompiled_graph.invoke({query:什么是Langgraph})# 打印最终回答print(res[final_answer])3.特殊节点START 和 END 节点都是 LangGraph 当中的特殊节点。START 它代表着将用户输入发送到图中的节点。引用此节点的主要目的是确定应首先调用哪些节点。END节点是一个特殊节点代表终止节点。当想表示哪些边在完成后没有动作时会引用这个节点。START 和 END 节点本质上是一个字符串如下所示ENDsys.intern(__end__)The last (maybe virtual) node in graph-style Pregel.STARTsys.intern(__start__)The first (maybe virtual) node in graph-style Pregel.START 作为一个特殊 节点在图执行完START节点之后也会创建一个 checkpoint4 节点缓存LangGraph 支持基于节点输入对节点进行缓存。对于配置了缓存的节点且缓存结果没有过期以相同的输入再次调用节点时可直接从缓存当中读取结果不需要再进行节点计算。使用缓存的方法如下1编译图时指定缓存存储后端langgraph.cache 包当中提供了内存级别的缓存后端Redis 缓存后端和 Sqlite 缓存后端的具体实现需要构造相关实例并在图编译时传入。2为节点指定缓存策略。每个缓存策略需配置key_func用于根据节点的输入生成缓存键默认情况下是使用 pickle 对输入进行 hash 运算的结果。ttl缓存的生存时间以秒为单位。如果未指定缓存将永不过期。importtimefromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraphfromlanggraph.cache.memoryimportInMemoryCachefromlanggraph.typesimportCachePolicyfromlanggraph.constantsimportSTARTfromlanggraph.checkpoint.memoryimportInMemorySaverclassState(TypedDict):x:intresult:intbuilderStateGraph(State)checkpointerInMemorySaver()# 1、定义节点模拟一些有耗时计算的节点defexpensive_node(state:State)-dict[str,int]:# expensive computationprint(fexpensive_node 被调用)time.sleep(5)print(fexpensive_node 计算完成)return{result:state[x]*2}# 2、添加节点时为节点配置缓存策略这里设置为10秒缓存缓存键的配置同样使用默认的方式可以在CachePolicy当中传入key_func配置不同的缓存键生成策略builder.add_node(expensive_node,expensive_node,cache_policyCachePolicy(ttl10))builder.add_edge(START,expensive_node)# 3、图编译时传入缓存器此处使用InMemoryCache也可以使用RedisCache等graphbuilder.compile(cacheInMemoryCache(),checkpointercheckpointer)# 4、第一次调用传入的状态为5print(graph.invoke({x:5},config{configurable:{thread_id:1}}))# 5、第二次调用传入相同的状态由于缓存策略会直接从缓存中返回结果而不会重新调用节点print(graph.invoke({x:5},config{configurable:{thread_id:2}}))5.节点重试– 在很多使用场景中有些节点由于客观原因限制导致其执行过程是不稳定的。因此我们可能希望节点拥有自定义的重试策略例如在调用API、查询数据库或调用大语言模型LLM等情况下。– 为节点添加重试策略需要在 add_node 中设置 retry_policy 参数。retry_policy 参数接受一个 RetryPolicy 命名元组对象。– RetryPolicy 对象有两个属性max_attempts 和 retry_on参数前者定义了总共重试次数后者定义了对于哪些异常类型进行重试。示例代码如下importrandomfromtypingimportDict,Anyfromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportRetryPolicy# 定义图状态 # State节点之间传递的状态结构只有一个 result 字段保存接口返回结果classState(TypedDict):result:str# 全局计数器用来记录 unstable_api_call 已经尝试调用多少次attempt_counter0defunstable_api_call(state:State)-Dict[str,Any]: 模拟一个不稳定的API调用有概率抛出异常失败用来测试重试机制 :param state: 当前图状态 :return: 增量更新的状态字典LangGraph规范写法只返回要更新的字段 globalattempt_counter attempt_counter1print(f尝试调用API这是第{attempt_counter}次尝试)# 业务逻辑前2次调用主动抛出异常模拟失败第3次调用成功ifattempt_counter3:raiseException(f模拟API调用失败 (尝试{attempt_counter}))else:# 第三次尝试成功返回需要更新的状态字段return{result:fAPI调用成功经过{attempt_counter}次尝试}defvalue_error_call(state:State)-Dict[str,Any]: 模拟抛出 ValueError 的节点 ✅ 重点LangGraph 默认重试策略【不会重试ValueError】这类内置异常 print(调用会抛出 ValueError 的节点)raiseValueError(模拟 ValueError 异常)defrun_demo():print( LangGraph 节点重试策略演示 \n)# 重置全局计数器保证每个演示之间互不干扰globalattempt_counter attempt_counter0# ---------------------- 演示1默认重试策略 ----------------------print(1. 使用默认重试策略:)print( 默认策略会对除特定异常外的所有异常进行重试)print( 不会重试的异常包括: ValueError, TypeError, ArithmeticError, ImportError,)print( LookupError, NameError, SyntaxError, RuntimeError,)print( ReferenceError, StopIteration, StopAsyncIteration, OSError\n)builder1StateGraph(State)# 注册节点绑定重试策略# retry_policyRetryPolicy(max_attempts5)最多总共尝试5次builder1.add_node(unstable_call,unstable_api_call,retry_policyRetryPolicy(max_attempts5))# 连线起点 - unstable_call - 结束builder1.add_edge(START,unstable_call)builder1.add_edge(unstable_call,END)# 编译图graph1builder1.compile()print(测试默认重试策略:)try:# 执行图传入初始状态resultgraph1.invoke({result:})print(f最终结果:{result}\n)exceptExceptionase:print(f最终失败:{type(e).__name__}:{e}\n)# ---------------------- 演示2 预留位置自定义重试策略 ----------------------print(2. 使用自定义重试策略:)print( 自定义策略只对特定错误进行重试\n)# 重置计数器这里代码里演示2逻辑是空的你后续可以在这里补充attempt_counter0# ---------------------- 演示3测试默认不重试的异常ValueError ----------------------print(3. 测试不会重试的异常类型:)builder3StateGraph(State)# 注册抛ValueError的节点同样配置最多尝试3次builder3.add_node(value_error_call,value_error_call,retry_policyRetryPolicy(max_attempts3))# 连线builder3.add_edge(START,value_error_call)builder3.add_edge(value_error_call,END)graph3builder3.compile()print(测试 ValueError默认策略不会重试:)try:resultgraph3.invoke({result:})print(f最终结果:{result}\n)exceptExceptionase:# 只会抛一次异常不会自动重试print(f最终失败:{type(e).__name__}:{e}\n)if__name____main__:# 程序入口运行整套重试demorun_demo()补充RetryPolicy参数max_attempts最大尝试次数包含第一次原始调用retry_on自定义要捕获并重试的异常类型列表就是演示 2 要写的核心参数backoff退避策略重试间隔指数退避等默认规则普通Exception子类网络异常、自定义异常默认会重试ValueError / TypeError / SyntaxError这类内置语法 / 参数类异常LangGraph 默认判定为代码问题不是临时故障不重试6.图内外数据传递– 默认情况下通过 graph.invoke 调用图时仅在整个图的执行过程都结束之后我们才能够拿到最终的状态那么如果想要在图的执行过程当中想要获取到图内部所产生的数据应该如何实现– 考虑 Agent 的一个场景首先我们希望在LLM 输出时就能够拿到 LLM 所产生的 token在前端进行展示其次如果 Agent 的流程较长我们希望能够拿到当前正在执行的节点或者流程是什么。– 由于 LangGraph 实现了 langchain_core 当中的 Runnable 接口其为我们提供了 stream 和 astream 方法也即流式输出方法通过流式输出就能解决前面所说到的问题。Stream方法提供了多种不同的模式如下表所示模式描述values每一个执行后流式输出完整的状态updates图执行过程中每一步执行后流式输出增量更新。如果在同一个步当中产生了多个增量更新这些增量更新会分别流式输出。custom流式输出节点内部的自定义数据。messages在任何调用了LLM的节点当中流式输出两元组数据LLM Tokenmetadatadebug流式输出所有能输出的信息混合模式流模式传入列表在列表当中添加多种不同的模式可以得到多种流式输出7.人工审核节点Agent 在完成用户设定的任务时有时候我们希望用户参与当中部分重要决策过程。LangGraph 为此提供了一个非常方便的原语interrupt。其原理如下图所示可以在节点内部任意位置引入 interrupt当节点执行到interrupt 所在位置时就会停止执行并可以得到从图内传递出来的需要用户审核的数据在用户进行相关审核完成后又可通过 graph.invoke让图继续往下执行。需要注意的是使用 interrupt 时在第二次调用 graph.invoke(command) 继续执行时有 interrupt 的函数会从函数起点往下执行如果在函数当中有更新数据、调用 API 接口等相关操作时会造成多次重复执行的情况。因此一定要保证这种类型操作的幂等性或者是将其封装到一个单独的节点当中避免在恢复时再次执行。