第4章 节点 Node 详解

节点是图的基本计算单元。本章讲清节点的完整输入输出契约它能接收哪些参数state / config / runtime、返回值必须遵循什么规则增量更新再介绍两个生产级能力——节点缓存CachePolicy与节点重试RetryPolicy以及 START / END 这两个特殊节点的本质。学完本章你写的节点就能达到生产代码的标准依赖可注入、输出可预期、失败可重试、重复调用可缓存。4.1 节点的输入输出节点输入三个可选注入参数在 LangGraph 中节点都是 Python 函数可以同步也可以异步它们可以接受以下参数state图的状态代表具体的业务数据config一个RunnableConfig对象包含 thread_id 之类的配置信息调用图时也可以传递用户自定义的其他配置runtime一个Runtime对象包含运行时 context可自定义 context调用图时传入以及其他信息如 store、stream_writer 等。以上参数会在运行过程中自动被 LangGraph 运行时注入——你不需要也不能手动传它们。由于是按关键字传参config和runtime的参数位置可以调换。三个参数的分工可以用一句话区分state 管业务数据config 管调用配置runtime 管运行时服务。依赖注入完整示例下面的客服节点演示了三个参数的典型用法从runtime.context获取注入的 LLM 与数据库依赖从config获取用户身份用runtime.stream_writer向图外推送进度importtimefromtypingimportTypedDict,Any,Listfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.runtimeimportRuntime# --- 模拟外部依赖 ---classMockLLM:definvoke(self,prompt:str):returnfAI Generated: Answer for {prompt}classMockDatabase:defget_user_info(self,user_id:str):return{id:user_id,role:vipifvipinuser_idelsestandard}classCustomerSupportState(TypedDict):query:str# 用户问题response:str# 客服回复log:List[str]# 处理日志defnode_customer_service(state:CustomerSupportState,config:RunnableConfig,runtime:Runtime)-dict: 客服节点展示如何从 config 中获取配置、 从 runtime.context 中获取注入的依赖LLM、DB并用 runtime 推送进度。 user_querystate[query]# 1、从 runtime 当中获取 context 对象依赖注入llmruntime.context[llm]dbruntime.context[db]# 2、从 config 当中获取调用方传入的配置configurableconfig.get(configurable,{})user_idconfigurable.get(user_id,guest)print(f\n[Node] 开始处理User ID:{user_id})# 3、验证依赖是否存在ifnotllmornotdb:return{response:System Error: Dependencies not injected!,log:[Error]}# 4、使用 db 对象查看用户角色user_infodb.get_user_info(user_id)user_tieruser_info.get(role)print(f[Node] 从 DB 获取用户角色:{user_tier})# 5、使用 runtime.stream_writer 向图外输出自定义数据writerruntime.stream_writer writer({status:thinking,message:f正在调用 LLM 为{user_tier}用户生成回复...})time.sleep(0.5)# 6、根据用户角色构建不同的 Prompt并模拟 LLM 调用promptfUser({user_tier}) asks:{user_query}llm_responsellm.invoke(prompt)return{response:llm_response,log:[fProcessed by{llm.__class__.__name__}]}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()# 初始化外部依赖my_llmMockLLM()my_dbMockDatabase()# user_id 属于配置信息放入 configconfig{configurable:{user_id:vip_user_999,}}# LLM、DB 属于运行时依赖放入 context随 invoke 注入context{llm:my_llm,db:my_db}resultgraph.invoke({query:如何升级会员},configconfig,contextcontext)print(result)if__name____main__:run_demo()逐段解析runtime.context依赖注入节点内部没有任何MockLLM()/MockDatabase()的实例化代码——依赖完全由调用方在graph.invoke(..., context...)处注入。好处显而易见单测时注入 Mock 对象生产时注入真实客户端节点代码一行不改config[configurable]调用配置user_id是「每次调用可能不同」的信息放 config它与 context 的分界是——context 放服务/对象config 放数值型配置runtime.stream_writer进度外送writer({...})发出的数据不出现在状态里、不影响图的执行只出现在stream_modecustom的流中第 5 章详解。这是节点向外界汇报「我正在干嘛」的正规通道比在节点里 print 干净得多返回值依然是增量 dict只写response和log不碰query。节点输出必须返回增量节点的输出为当前节点对状态的增量更新而不能将接收到的整个状态实例返回出去。LangGraph 底层会把所有节点输出的状态都作为增量状态尝试与当前的全局状态做一次合并操作。如果把整个状态都输出会有两类问题对于没有配置任何 reducer 的状态键LangGraph 底层无法完成合并会抛出异常对于配置了 reducer 的状态键如果节点输出了不属于该节点更新的状态会导致数据错乱比如重复累加。错误示范返回整个 statefromtypingimportTypedDictfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphclassMyState(TypedDict):query:strfile_result:strweb_result:strfinal_answer:strdefquery_web(state:MyState)-dict:做网络搜索返回搜索结果# ★ 错误演示直接修改并 return 整个 statequerystate[query]state[web_result]f{query}的网络搜索结果returnstatedefquery_file(state:MyState)-dict:做文件搜索返回搜索结果querystate[query]state[file_result]f{query}的文件搜索结果returnstatedefanswer(state:MyState)-dict:返回最终的答案web_resultstate[web_result]file_resultstate[file_result]final_answerfLLM基于{web_result}{file_result}的最终结果state[final_answer]final_answerreturnstate graphStateGraph(state_schemaMyState)graph.add_node(answer)graph.add_node(query_web)graph.add_node(query_file)graph.add_edge(START,query_web)graph.add_edge(START,query_file)graph.add_edge(query_web,answer)graph.add_edge(query_file,answer)compiled_graphgraph.compile()rescompiled_graph.invoke({query:什么是LangGraph})print(res[final_answer])错误分析query_web和query_file并行执行两个函数都拿到{query: ..., 其余为 None}的初始状态快照各自往自己手里的 state 写入结果后返回了整个 state。合并时灾难发生了query_web返回的完整 state 里file_result是它没有权利写的字段值还是 Nonequery_file返回的完整 state 里web_result同理默认覆盖 reducer 下两个「完整状态」互相覆盖——后到者把先到者写入的键抹回 None于是answer节点读到的web_result或file_result可能是 None拼出的final_answer出现 “None” 字样甚至直接 KeyError。正确写法每个节点只返回自己负责的键defquery_web(state:MyState)-dict:return{web_result:f{state[query]}的网络搜索结果}defquery_file(state:MyState)-dict:return{file_result:f{state[query]}的文件搜索结果}defanswer(state:MyState)-dict:return{final_answer:fLLM基于{state[web_result]}{state[file_result]}的最终结果}一句话原则节点是「领任务、交产出」的工人只交自己那条生产线的产出别把整车间都搬回来。4.2 特殊节点START 与 ENDSTART和END是 LangGraph 当中的特殊节点START代表「将用户输入发送到图中」的节点。引用它的主要目的是确定应首先调用哪些节点从 START 引出的边就是入口边END代表终止节点。当想表示哪些边在完成后没有动作时会引用这个节点指向 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.解析sys.intern把字符串放入全局驻留表保证全进程里START是同一个对象——比较时可以直接用is既省内存又加速称其为「virtual node虚拟节点」很准确它们不代表任何计算只是通道系统里的标记锚点——START 的「执行」就是把用户输入写进入口通道END 的「到达」就是让所有通道不再产生新更新START 也是一个会写检查点的节点在图执行完 START 之后会先创建一个初始 checkpoint再进入第一个超步。这就是get_state_history里第一个快照values为初始输入的来源。4.3 节点缓存CachePolicyLangGraph 支持基于节点输入对节点进行缓存对于配置了缓存的节点且缓存结果未过期时以相同的输入再次调用节点可直接从缓存读取结果不再执行节点计算。使用缓存分两步编译图时指定缓存存储后端langgraph.cache包提供内存缓存InMemoryCache、Redis 缓存、Sqlite 缓存后端构造实例后在 compile 时传入为节点指定缓存策略CachePolicy可配置key_func根据节点输入生成缓存键默认是对输入做 pickle 后的哈希值ttl缓存生存时间秒不指定则永不过期。示例代码importtimefromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraphfromlanggraph.cache.memoryimportInMemoryCachefromlanggraph.typesimportCachePolicyfromlanggraph.constantsimportSTARTfromlanggraph.checkpoint.memoryimportInMemorySaverclassState(TypedDict):x:intresult:intbuilderStateGraph(State)checkpointerInMemorySaver()# 1、定义节点模拟耗时计算defexpensive_node(state:State)-dict[str,int]:print(expensive_node 被调用)time.sleep(5)# 模拟 5 秒的重计算print(expensive_node 计算完成)return{result:state[x]*2}# 2、添加节点时配置缓存策略10 秒过期缓存键用默认生成方式builder.add_node(expensive_node,expensive_node,cache_policyCachePolicy(ttl10))builder.add_edge(START,expensive_node)# 3、编译时传入缓存器也可用 RedisCache 等graphbuilder.compile(cacheInMemoryCache(),checkpointercheckpointer)# 4、第一次调用x5真实执行耗时 5 秒print(graph.invoke({x:5},config{configurable:{thread_id:1}}))# 5、第二次调用输入相同 → 直接命中缓存几乎瞬时返回不打印被调用print(graph.invoke({x:5},config{configurable:{thread_id:2}}))逐段解析命中判定的粒度是「节点输入」第二次调用换了 thread_id2但节点输入状态相同x5照样命中——缓存与 thread 无关只看输入ttl1010 秒内相同输入直接返回缓存超过 10 秒后缓存过期重新计算。适合「数据有时效但不要求强实时」的场景如汇率、推荐结果key_func定制默认对整个输入做哈希。如果输入里包含时间戳、request_id 这类每次都变的字段默认策略会永远 miss需要自定义key_func剔除易变字段典型场景图里有个「调用外部 LLM 改写 query」的节点相同 query 反复进来时缓存能省掉真金白银的 token 费用注意事项节点内部若依赖状态之外的外部变量如全局计数器、当前时间缓存结果可能「过时」——被缓存节点的逻辑必须是输入决定输出的纯函数才有正确性保证。4.4 节点重试RetryPolicy很多场景下节点由于客观原因限制执行过程并不稳定——调用 API、查询数据库、调用 LLM都可能出现瞬时故障。LangGraph 允许为节点配置自定义重试策略。为节点添加重试策略需在add_node中设置retry_policy参数。它接受一个RetryPolicy对象两个关键属性max_attempts总尝试次数上限含首次retry_on对哪些异常类型进行重试。示例代码importrandomfromtypingimportDict,Anyfromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportRetryPolicyclassState(TypedDict):result:str# 用全局变量跟踪尝试次数attempt_counter0defunstable_api_call(state:State)-Dict[str,Any]:模拟一个不稳定的 API 调用前两次失败第三次成功globalattempt_counter attempt_counter1print(f尝试调用API这是第{attempt_counter}次尝试)ifattempt_counter3:raiseException(f模拟API调用失败 (尝试{attempt_counter}))else:return{result:fAPI调用成功经过{attempt_counter}次尝试}defvalue_error_call(state:State)-Dict[str,Any]:模拟抛出 ValueError 的节点默认策略不会重试它print(调用会抛出 ValueError 的节点)raiseValueError(模拟 ValueError 异常)defrun_demo():globalattempt_counter# ---- 演示 1默认重试策略可重试的异常 ----print(1. 使用默认重试策略对瞬时异常重试:)attempt_counter0builder1StateGraph(State)builder1.add_node(unstable_call,unstable_api_call,retry_policyRetryPolicy(max_attempts5),# 最多尝试 5 次)builder1.add_edge(START,unstable_call)builder1.add_edge(unstable_call,END)graph1builder1.compile()try:resultgraph1.invoke({result:})print(f最终结果:{result}\n)exceptExceptionase:print(f最终失败:{type(e).__name__}:{e}\n)# ---- 演示 2默认策略不会重试的异常类型 ----print(2. 测试不会重试的异常类型:)builder3StateGraph(State)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()try:resultgraph3.invoke({result:})print(f最终结果:{result}\n)exceptExceptionase:print(f最终失败:{type(e).__name__}:{e}\n)if__name____main__:run_demo()逐段解析演示 1 的运行轨迹unstable_api_call第 1、2 次抛普通Exception重试机制自动重跑节点第 3 次成功返回。控制台会看到三次「尝试调用API」最终拿到结果——图层面没有任何报错瞬时故障被透明地消化掉了默认策略的重试黑名单默认策略对绝大多数异常重试但对以下异常不重试它们通常意味着代码本身有 bug重试没有意义ValueError, TypeError, ArithmeticError, ImportError, LookupError, NameError, SyntaxError, RuntimeError, ReferenceError, StopIteration, StopAsyncIteration, OSError演示 2 的运行轨迹value_error_call抛出ValueError虽然在黑名单里——即使max_attempts3也只执行一次就立即把异常抛给图的调用方。这就是「策略性放弃」参数校验错误重试一百次还是错retry_on定制如果你的业务判断「ValueError 也值得重试」比如上游服务用 ValueError 报过载可以显式传retry_onValueError覆盖默认黑名单与 checkpointer 的配合重试发生在节点内部超步尚未结束不会写入中间检查点——重试全部失败后本次 invoke 报错但之前超步的检查点仍在可以配合第 3 章的invoke(None)做故障恢复。两个机制一个管「瞬时抖动」一个管「持久故障」层次互补。缓存与重试的定位对比能力解决的问题命中/触发条件语义CachePolicy重复计算浪费节点输入相同且缓存未过期跳过执行直接取旧结果RetryPolicy瞬时故障节点抛出可重试异常同一超步内自动重跑节点两者可以同时配在同一个节点上先查缓存未命中再执行执行失败再重试。4.5 本章小结节点可注入三个参数state业务数据、config调用配置如 user_id、runtime运行时服务context 依赖注入 stream_writer 进度外送节点必须返回增量状态——返回整个 state 在并行场景下会互相覆盖是新手第一大坑START / END 本质是sys.intern的特殊字符串虚拟节点START 执行后即落第一个检查点CachePolicy(ttl...)基于节点输入缓存结果省掉重复的重计算/LLM 调用RetryPolicy(max_attempts..., retry_on...)对瞬时异常自动重试默认黑名单里的确定性异常不重试缓存管「重复」、重试管「抖动」、checkpointer 恢复管「持久故障」三者互补构成节点的生产级可靠性。下一章讲图与外界的交互流式输出的五种模式以及 interrupt 人工审核。