大模型学习23 大模型学习 - LangGraph3-节点

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# RunnableConfig：LangGraph标准的配置类型fromlangchain_core.runnablesimportRunnableConfig# StateGraph构建状态图，START起点，END终点fromlanggraph.graphimportStateGraph,START,END# Runtime：LangGraph运行时对象，用来传递context上下文、流式输出writerfromlanggraph.runtimeimportRuntime# 模拟大模型类，代替真实LLMclassMockLLM:# invoke方法，接收prompt，直接返回拼接好的模拟回答definvoke(self,prompt:str):returnf"AI Generated: Answer for '{prompt}'"# 模拟数据库类classMockDatabase:# 根据user_id获取用户信息，简单判断是否是vip用户defget_user_info(self,user_id:str):return{"id":user_id,"role":"vip"if"vip"inuser_idelse"standard"}# 定义客服流程的状态结构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_query=state["query"]# 1、从runtime当中获取context对象，context就是我们外部注入进来的依赖llm=runtime.context['llm']db=runtime.context['db']# 取出config里的configurable配置块configurable=config.get("configurable",{})# 获取用户id，取不到就默认是guest访客user_id=configurable.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_info=db.get_user_info(user_id)user_tier=user_info.get("role")print(f"[Node] 从 DB 获取用户角色:{user_tier}")# 4、使用 runtime.writer 向图外输出自定义流式数据（可以前端展示进度、状态）writer=runtime.stream_writer writer({"status":"thinking","message":f"正在调用 LLM 为{user_tier}用户生成回复..."})# 休眠0.5秒模拟LLM思考耗时time.sleep(0.5)# 5、根据用户角色构建不同的 Prompt，并模拟 LLM 调用prompt=f"User({user_tier}) asks:{user_query}"llm_response=llm.invoke(prompt)# 返回字典，更新state中的response和logreturn{"response":llm_response,"log":[f"Processed by{llm.__class__.__name__}"]}# 这个return永远不会执行，属于冗余代码，可以删掉returnNone# 运行演示的入口函数defrun_demo():# 创建图构造器，绑定状态结构builder=StateGraph(CustomerSupportState)# 注册客服节点builder.add_node("customer_service",node_customer_service)# 连线：起点到客服节点builder.add_edge(START,"customer_service")# 客服节点执行完直接结束builder.add_edge("customer_service",END)# 编译图graph=builder.compile()# 初始化外部依赖对象：模拟LLM、模拟DBmy_llm=MockLLM()my_db=MockDatabase()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参数实现依赖注入result=graph.invoke(initial_state,config=config,context=context)# 打印最终返回的状态结果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返回 """# 从传入的共享状态中取出用户问题query=state['query']# 修改state里的web_result字段state['web_result']=f'{query}的网络搜索结果'# 把整个state字典返回（不推荐的写法，仅用来演示）returnstatedefquery_file(state:MyState)->dict:""" 【节点函数】做本地文件检索，返回文件内检索结果 ⚠️ 和上面一样：直接返回完整 state，属于不推荐写法 """# 获取用户查询问题query=state['query']# 填充文件检索结果到状态state['file_result']=f'{query}的文件搜索结果'# 返回完整statereturnstatedefanswer(state:MyState)->dict:""" 【节点函数】汇总web和文件两路结果，交给LLM生成最终答案 这个节点会在 query_web、query_file 两个节点都跑完之后才执行 """# 读取两路检索结果web_result=state['web_result']file_result=state['file_result']# 拼接生成最终回答final_answer=f'LLM基于{web_result}，{file_result}的最终结果'# 写入状态state['final_answer']=final_answerreturnstate# 实例化状态图，指定状态结构为 MyStategraph=StateGraph(state_schema=MyState)# 向图中注册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_graph=graph.compile()# 调用图，传入初始状态：只有query字段res=compiled_graph.invoke({"query":"什么是Langgraph"})# 打印最终回答print(res['final_answer'])

3.特殊节点

START 和 END 节点都是 LangGraph 当中的特殊节点。
START 它代表着将用户输入发送到图中的节点。引用此节点的主要目的是确定应首先调用哪些节点。
END节点是一个特殊节点，代表终止节点。当想表示哪些边在完成后没有动作时，会引用这个节点。
START 和 END 节点本质上是一个字符串，如下所示：

END=sys.intern("__end__")"""The last (maybe virtual) node in graph-style Pregel."""START=sys.intern("__start__")"""The first (maybe virtual) node in graph-style Pregel."""

START 作为一个特殊 节点，在图执行完START节点之后，也会创建一个 checkpoint

4 节点缓存

LangGraph 支持基于节点输入对节点进行缓存。对于配置了缓存的节点，且缓存结果没有过期，以相同的输入再次调用节点时，可直接从缓存当中读取结果，不需要再进行节点计算。
使用缓存的方法如下：

• （1）编译图时指定缓存存储后端：langgraph.cache 包当中提供了内存级别的缓存后端，Redis 缓存后端，和 Sqlite 缓存后端的具体实现，需要构造相关实例，并在图编译时传入。
• （2）为节点指定缓存策略。每个缓存策略需配置：
  • key_func：用于根据节点的输入生成缓存键，默认情况下是使用 pickle 对输入进行 hash 运算的结果。
  • ttl：缓存的生存时间（以秒为单位）。如果未指定，缓存将永不过期。

importtimefromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraphfromlanggraph.cache.memoryimportInMemoryCachefromlanggraph.typesimportCachePolicyfromlanggraph.constantsimportSTARTfromlanggraph.checkpoint.memoryimportInMemorySaverclassState(TypedDict):x:intresult:intbuilder=StateGraph(State)checkpointer=InMemorySaver()# 1、定义节点：模拟一些有耗时计算的节点defexpensive_node(state:State)->dict[str,int]:# expensive computationprint(f"expensive_node 被调用")time.sleep(5)print(f"expensive_node 计算完成")return{"result":state["x"]*2}# 2、添加节点时，为节点配置缓存策略，这里设置为10秒缓存，缓存键的配置同样使用默认的方式，可以在CachePolicy当中传入key_func配置不同的缓存键生成策略builder.add_node("expensive_node",expensive_node,cache_policy=CachePolicy(ttl=10))builder.add_edge(START,"expensive_node")# 3、图编译时，传入缓存器，此处使用InMemoryCache，也可以使用RedisCache等graph=builder.compile(cache=InMemoryCache(),checkpointer=checkpointer)# 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_counter=0defunstable_api_call(state:State)->Dict[str,Any]:""" 模拟一个不稳定的API调用，有概率抛出异常失败，用来测试重试机制 :param state: 当前图状态 :return: 增量更新的状态字典（LangGraph规范写法，只返回要更新的字段） """globalattempt_counter attempt_counter+=1print(f"尝试调用API，这是第{attempt_counter}次尝试")# 业务逻辑：前2次调用主动抛出异常模拟失败，第3次调用成功ifattempt_counter<3:raiseException(f"模拟API调用失败 (尝试{attempt_counter})")else:# 第三次尝试成功，返回需要更新的状态字段return{"result":f"API调用成功，经过{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_counter=0# ---------------------- 演示1：默认重试策略 ----------------------print("1. 使用默认重试策略:")print(" 默认策略会对除特定异常外的所有异常进行重试")print(" 不会重试的异常包括: ValueError, TypeError, ArithmeticError, ImportError,")print(" LookupError, NameError, SyntaxError, RuntimeError,")print(" ReferenceError, StopIteration, StopAsyncIteration, OSError\n")builder1=StateGraph(State)# 注册节点，绑定重试策略# retry_policy=RetryPolicy(max_attempts=5)：最多总共尝试5次builder1.add_node("unstable_call",unstable_api_call,retry_policy=RetryPolicy(max_attempts=5))# 连线：起点 -> unstable_call -> 结束builder1.add_edge(START,"unstable_call")builder1.add_edge("unstable_call",END)# 编译图graph1=builder1.compile()print("测试默认重试策略:")try:# 执行图，传入初始状态result=graph1.invoke({"result":""})print(f"最终结果:{result}\n")exceptExceptionase:print(f"最终失败:{type(e).__name__}:{e}\n")# ---------------------- 演示2 预留位置：自定义重试策略 ----------------------print("2. 使用自定义重试策略:")print(" 自定义策略只对特定错误进行重试\n")# 重置计数器（这里代码里演示2逻辑是空的，你后续可以在这里补充）attempt_counter=0# ---------------------- 演示3：测试默认不重试的异常（ValueError） ----------------------print("3. 测试不会重试的异常类型:")builder3=StateGraph(State)# 注册抛ValueError的节点，同样配置最多尝试3次builder3.add_node("value_error_call",value_error_call,retry_policy=RetryPolicy(max_attempts=3))# 连线builder3.add_edge(START,"value_error_call")builder3.add_edge("value_error_call",END)graph3=builder3.compile()print("测试 ValueError（默认策略不会重试）:")try:result=graph3.invoke({"result":""})print(f"最终结果:{result}\n")exceptExceptionase:# 只会抛一次异常，不会自动重试print(f"最终失败:{type(e).__name__}:{e}\n")if__name__=="__main__":# 程序入口，运行整套重试demorun_demo()

补充

1. RetryPolicy参数
   • max_attempts：最大尝试次数（包含第一次原始调用）
   • retry_on：自定义要捕获并重试的异常类型列表，就是演示 2 要写的核心参数
   • backoff：退避策略（重试间隔，指数退避等）
2. 默认规则：
   • 普通Exception子类（网络异常、自定义异常）默认会重试
   • ValueError / TypeError / SyntaxError这类内置语法 / 参数类异常，LangGraph 默认判定为代码问题，不是临时故障，不重试

6.图内外数据传递

– 默认情况下，通过 graph.invoke 调用图时，仅在整个图的执行过程都结束之后，我们才能够拿到最终的状态，那么如果想要在图的执行过程当中，想要获取到图内部所产生的数据，应该如何实现？
– 考虑 Agent 的一个场景：首先，我们希望在LLM 输出时，就能够拿到 LLM 所产生的 token，在前端进行展示；其次，如果 Agent 的流程较长，我们希望能够拿到，当前正在执行的节点或者流程是什么。
– 由于 LangGraph 实现了 langchain_core 当中的 Runnable 接口，其为我们提供了 stream 和 astream 方法，也即流式输出方法，通过流式输出，就能解决前面所说到的问题。
Stream方法提供了多种不同的模式，如下表所示：

7.人工审核节点

Agent 在完成用户设定的任务时，有时候我们希望用户参与当中部分重要决策过程。LangGraph 为此提供了一个非常方便的原语：interrupt。
其原理如下图所示：

可以在节点内部任意位置引入 interrupt，当节点执行到interrupt 所在位置时，就会停止执行，并可以得到从图内传递出来的，需要用户审核的数据，在用户进行相关审核完成后，又可通过 graph.invoke，让图继续往下执行。

需要注意的是，使用 interrupt 时，在第二次调用 graph.invoke(command) 继续执行时，有 interrupt 的函数，会从函数起点往下执行，如果在函数当中，有更新数据、调用 API 接口等相关操作时，会造成多次重复执行的情况。因此，一定要保证这种类型操作的幂等性，或者是将其封装到一个单独的节点当中，避免在恢复时再次执行。