LangGraph 中控制流、上下文管理和执行控制工具

一、RuntimeRuntime是 LangGraph 提供的一个运行时上下文对象可以把它理解为一个“工具箱”在节点node或工具tool中通过它来获取执行所需的各种信息和控制能力。它解决了之前需要通过config[configurable]来取配置的问题让代码更清晰且支持类型检查。context存放静态的、一次性的配置数据如当前用户ID、数据库连接、API密钥等。store用于跨会话的长期记忆可以把用户偏好、历史记录等持久化保存实现记忆功能。writer用于通过custom流模式发送自定义数据方便在节点执行过程中实时向客户端推送进度或中间结果。control用于节点的协作性取消比如在节点中调用control.requestDrain()来请求图在下一个执行边界停止适合长时间运行或需要优雅停止的任务。signal一个AbortSignal对象当图运行被外部取消时该信号会被触发节点可以根据它来响应并提前结束。使用Runtime从 LangGraph 的新版本开始推荐用Runtime来替代原有的config模式。在定义节点时直接在函数签名中声明它from dataclasses import dataclass from langgraph.runtime import Runtime dataclass class ContextSchema: user_id: str def my_node(state: State, runtime: Runtime[ContextSchema]): # 通过 runtime.context 获取上下文类型安全IDE 有提示 uid runtime.context.user_id # 通过 runtime.store 操作长期记忆 # runtime.store.put(...) # 通过 runtime.writer 发送自定义流 # runtime.writer(进度信息) return {result: f处理用户 {uid}}调用图时通过context参数传入数据即可graph.invoke(input_data, context{user_id: 123})二、SendSend用于“Map-Reduce”模式特别是在需要把一个任务拆分成多个子任务并行处理时。它的作用是动态地、有条件地将一个任务发送给另一个节点并为每个子任务指定独立的状态。Send的构造函数是Send(node_name: str, state: dict)其中node_name是要执行的目标节点state是该子任务的状态。使用它通常和条件边搭配使用由一个节点返回一个Send对象列表。from langgraph.types import Send # 这个节点负责决定如何拆分任务并返回 Send 列表 def continue_to_jokes(state: OverallState): # 为每个主题创建一个子任务发送到 generate_joke 节点 return [Send(generate_joke, {subject: s}) for s in state[subjects]] # 在构建图时添加条件边 builder.add_conditional_edges( generate_topics, # 源节点 continue_to_jokes, # 路由函数返回 Send 列表 [generate_joke] # 目标节点列表 )注意Send与Runtime的传播问题这是目前存在的一个已知限制使用Send创建的子任务如上面的generate_joke节点在 LangGraph Server 中默认情况下不会自动继承根节点的Runtime上下文。这意味着如果generate_joke节点中需要访问runtime.context中的数据在本地直接运行graph.invoke()可能没问题但部署到 LangGraph Server 后这种写法会报错RuntimeError: Runtime context is not available。Send只传递state参数无法通过常规方式直接传递整个Runtime对象。如果在 Server 环境下需要为子任务传递上下文可能需要在state中显式复制所需字段或者留意 LangGraph 后续版本对此问题的修复。三、CommandCommand提供了更灵活的控制流主要有两个用途结合状态更新进行跳转在节点中返回Command可以同时更新当前状态并指定下一个要执行的节点而不是通过add_edge预定义固定路径。中断执行等待用户输入即interrupt函数用于实现“人机协同”场景当节点需要用户输入时调用interrupt()中断图执行并返回一个中断信息等待用户通过Command(resume...)恢复。使用Command场景一跳转与更新from langgraph.types import Command def my_node(state: State): # 可以在返回中同时指定下一个节点和更新状态 if state[some_condition]: return Command(gotofast_node, update{result: 快速路径}) else: return Command(gotoslow_node, update{result: 慢速路径})场景二中断与恢复在节点中调用interrupt()会触发中断并抛出GraphInterrupt异常它返回用户传入的resume值或抛出异常等待恢复。from langgraph.types import interrupt def my_node(state: State): # 执行到这里会中断并返回 请确认是否继续 这个信息 user_confirmation interrupt(请确认是否继续) if user_confirmation 是: return {confirmed: True} else: return {confirmed: False}恢复执行时通过Command传入resume值# 在调用处使用 Command(resume...) 恢复 graph.invoke(Command(resume是), configconfig)