PregelProtocol是LangGraph架构中的关键抽象层。它通过定义一套标准的执行体契约将图的定义与图的执行解耦。这使得LangGraph能够灵活地支持多种执行后端本地Pregel引擎、远程API等同时为上层应用提供了稳定、统一的编程接口是 LangGraph 实现其灵活性和扩展性的基石。1. 配置绑定通过前面的内容我们会发现RunnableConfig这个对象几乎时无所不在我们在调用Pregel对象的时候可以将它作为参数用来提供用于控制其执行行为比如迭代限制并发控制等的配置。执行引擎还将它作为容器用来在上下游组件之间传递一些对象和信号所以前面的演示实例才可以在节点处理函数中从注入的RunnableConfig中提取像Runtime、PregelScratchpad、Checkpoint的命名空间和静态上下文这样对象和信息。对于单纯Pregel节点RunnableConfig是唯一可以注入到处理函数中的参数所以除了输入参数其他所需的信息只能从它里面提取。with_config方法赋予了执行体与配置绑定的能力。除了提供RunnableConfig对象我们还可以利用关键词参数提供待绑定的配置。由于RunnableConfig本质上就是一个TypedDict对象提供的关键字参数组成的键值对可以直接转换成RunnableConfig对象。with_config方法会将两者合并生成一个新的RunnableConfig对象绑定到执行体上。classPregelProtocol(Runnable[InputT,Any],Generic[StateT,ContextT,InputT,OutputT]):abstractmethoddefwith_config(self,config:RunnableConfig|NoneNone,**kwargs:Any)-Self:...2. 可视化呈现PregelProtocol是LangGraph对图的抽象这里的图是图论的概念但是若真能将它的结构呈现在一张图片中这无疑是非常有意义的。毕竟代码仅仅是面向程序员的语言图片不但直观还没有受众限制。LangGraph专门定义了如下这个Graph类型来表示面向可视化呈现的图。一个Graph对象表示的图依然由节点和边构成。它的每个节点都有一个唯一标识我们可以调用next_id方法为下一个待添加的节点生成此标识。我们不仅可以调用add_node、remove_node和add_edge这样的方法以添加/移除节点和边来构建图还可以调用extend方法将另一个Graph的所有节点和边添加进来。dataclassclassGraph:nodes:dict[str,Node]field(default_factorydict)edges:list[Edge]field(default_factorylist)defnext_id(self)-strdefadd_node(self,data:type[BaseModel]|RunnableType|None,id:str|NoneNone,*,metadata:dict[str,Any]|NoneNone,)-Nodedefremove_node(self,node:Node)-Nonedefadd_edge(self,source:Node,target:Node,data:Stringifiable|NoneNone,conditional:boolFalse,# noqa: FBT001,FBT002)-Edgedefextend(self,graph:Graph,*,prefix:str)-tuple[Node|None,Node|None]:defreid(self)-Graph:deffirst_node(self)-Node|Nonedeflast_node(self)-Node|Nonedeftrim_first_node(self)-Nonedeftrim_last_node(self)-Nonedefto_json(self,*,with_schemas:boolFalse)-dict[str,list[dict[str,Any]]]defdraw_ascii(self)-str:defprint_ascii(self)-None:overloaddefdraw_png(self,output_file_path:str,fontname:str|NoneNone,labels:LabelsDict|NoneNone,)-None:...overloaddefdraw_png(self,output_file_path:None,fontname:str|NoneNone,labels:LabelsDict|NoneNone,)-bytes:...defdraw_png(self,output_file_path:str|NoneNone,fontname:str|NoneNone,labels:LabelsDict|NoneNone,)-bytes|Nonedefdraw_mermaid(self,*,with_styles:boolTrue,curve_style:CurveStyleCurveStyle.LINEAR,node_colors:NodeStyles|NoneNone,wrap_label_n_words:int9,frontmatter_config:dict[str,Any]|NoneNone,)-strdefdraw_mermaid_png(self,*,curve_style:CurveStyleCurveStyle.LINEAR,node_colors:NodeStyles|NoneNone,wrap_label_n_words:int9,output_file_path:str|NoneNone,draw_method:MermaidDrawMethodMermaidDrawMethod.API,background_color:strwhite,padding:int10,max_retries:int1,retry_delay:float1.0,frontmatter_config:dict[str,Any]|NoneNone,base_url:str|NoneNone,proxies:dict[str,str]|NoneNone,)-bytes调用reid方法可以返回一个新的Graph对象它会尽量保留途中可读性的元素但是节点的ID会重新生成。Graph的first_node和last_node方法返回第一个和最后一个节点。如果我们希望删除第一个只有单一输出边或者最后一个只有单一输入边的节点可以调用trim_first_node或者trim_last_node方法。构建好的Graph可以采用不同的呈现方式。Graph提供了五个绘图方法draw_ascii和print_ascii采用ascii码字符的呈现方式前者返回具体的ascii码字符串后者则直接在终端将图绘制出来这种方法不依赖其他的绘图相关的包。draw_mermaid和draw_mermaid_png 采用Mermaid图表的呈现方式Mermaid是一种基于文本的流程图定义语言广泛支持于 GitHub、Notion和各种编辑器中。draw_mermaid返回图标文本而draw_mermaid_png则直接将图表进一步渲染成PNG图片draw_png渲染成PNG图片该方法最终会Graphviz一个开源的图可视化软件来布局和渲染图片。再回到PregelProtocol类型的定义上它定义了如下所示的get_graph/aget_graph方法它们的返回类型DrawableGraph正是上述Graph类型的别名。该方法除了可以传入RunnableConfig对象作为可选配置外还具有一个名为xray的参数。xrayX光参数决定了你在查看图结构时到底能看多深。它专门用于处理子图的展开显示。如果设置为False默认值图将以黑盒模式显式如果你的图中包含子图它只会显示为一个单一的节点。你看不见子图内部的任何节点、边或逻辑。反之将会采用全展开模式它会像X光一样穿透所有层级将所有嵌套子图内部的节点和连线全部平铺出来。fromlangchain_core.runnables.graphimportGraphasDrawableGraphclassPregelProtocol(Runnable[InputT,Any],Generic[StateT,ContextT,InputT,OutputT]):abstractmethoddefget_graph(self,config:RunnableConfig|NoneNone,*,xray:int|boolFalse,)-DrawableGraph:...abstractmethodasyncdefaget_graph(self,config:RunnableConfig|NoneNone,*,xray:int|boolFalse,)-DrawableGraph:...在第一个演示实例中我们创建了一个作为笑话生成器的Agent现在我们将它简化看看由它生成的Graph如何将图的结构以可视化的形式呈现出来。如下面的代码片段所示我们利用StateGraph作为Builder构建了一张由两个节点组成的图它们和Start和End之间有四条边。fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.pregel.protocolimportPregelProtocolfromPILimportImageasPILImageimportiofromlanggraph.checkpoint.memoryimportMemorySaverdefgenerate_joke(state):passdefregenerate_joke(state):passbuilder(StateGraph(dict).add_node(generate_joke,generate_joke).add_node(regenerate_joke,regenerate_joke))builder.add_edge(START,generate_joke)builder.add_edge(regenerate_joke,END)builder.add_conditional_edges(generate_joke,lambda_:bad,{good:END,bad:regenerate_joke})app:PregelProtocolbuilder.compile(MemorySaver())graphapp.get_graph()graph.print_ascii()bytesgraph.draw_mermaid_png()PILImage.open(io.BytesIO(bytes)).show()在将StateGraph编译成Pregel对象后我们调用其get_graph方法得到对应的Graph对象。我们以两种形式呈现其结构前者通过调用print_ascii方法以ASCII字符的形式输出图结构后者调用draw_mermaid_png方法生成一张PNG图片。ascii呈现方式----------- | __start__ | ----------- * * * --------------- | generate_joke | --------------- ... ... .. .. .. .. ----------------- .. | regenerate_joke | .. ----------------- .. *** ... ** .. ** .. --------- | __end__ | ---------PNG图片3. 持久化为了支持中断/恢复的执行方式同时为时间旅行提供支持图必须利用持久化的方式将执行过程的重要时刻的状态保存下来。LangGraph采用基于Checkpoint的持久化形式对于指定的每个任务不论是执行成功针对通道的写入意图还是抛出异常、人为中断或者提供的Resume Value都会以Pending Write的形式被记录下来。当Superstep成功完成针对通道的写入被成功应用这些Pending Write被丢弃换来一个Checkpoint来描述当前的状态。作为LangGraph执行体的抽象PregelProtocol定义了get_state/aget_state方法用于读取在某个Superstep由Checkpoint对于最后一个未完成的Superstep还包括Pending Write构建的状态快照该快照体现为一个StateSnapshot对象。get_state_history/aget_state_history返回由这些快照谱写的一段历史。classPregelProtocol(Runnable[InputT,Any],Generic[StateT,ContextT,InputT,OutputT]):abstractmethoddefget_state(self,config:RunnableConfig,*,subgraphs:boolFalse)-StateSnapshot:...abstractmethodasyncdefaget_state(self,config:RunnableConfig,*,subgraphs:boolFalse)-StateSnapshot:...abstractmethoddefget_state_history(self,config:RunnableConfig,*,filter:dict[str,Any]|NoneNone,before:RunnableConfig|NoneNone,limit:int|NoneNone,)-Iterator[StateSnapshot]:...abstractmethoddefaget_state_history(self,config:RunnableConfig,*,filter:dict[str,Any]|NoneNone,before:RunnableConfig|NoneNone,limit:int|NoneNone,)-AsyncIterator[StateSnapshot]:...abstractmethoddefbulk_update_state(self,config:RunnableConfig,updates:Sequence[Sequence[StateUpdate]],)-RunnableConfig:...abstractmethodasyncdefabulk_update_state(self,config:RunnableConfig,updates:Sequence[Sequence[StateUpdate]],)-RunnableConfig:...abstractmethoddefupdate_state(self,config:RunnableConfig,values:dict[str,Any]|Any|None,as_node:str|NoneNone,)-RunnableConfig:...abstractmethodasyncdefaupdate_state(self,config:RunnableConfig,values:dict[str,Any]|Any|None,as_node:str|NoneNone,)-RunnableConfig:...持久化存储的Checkpoint不仅使我们可以回顾历史还可以提供时间旅行使我们可以从某个历史时刻重新执行后面的流程。不仅如此PregelProtocol还提供了update_state/bulk_update_state/abulk_update_state可以直接修改状态。但是它们并非篡改历史只是基于某个在某个历史时刻开启了另一段平行宇宙而已。4. 两种调用方式PregelProtocol的invoke/ainvoke和stream/astream方法体现了针对LangGraph 执行体两种调用方式。前者采用简单的请求/回复消息交换模式客户端需要等整个流程结束之后采用得到结果。如果整个处理流程比较复杂或者涉及一些耗时的操作过长的等待会带来糟糕的体验。后者采用流式处理使客户端可以实施得到处理的中间结果或者感知到处理的进度。我们将在后续部分对流式处理进行单独介绍。classPregelProtocol(Runnable[InputT,Any],Generic[StateT,ContextT,InputT,OutputT]):abstractmethoddefstream(self,input:InputT|Command|None,config:RunnableConfig|NoneNone,*,context:ContextT|NoneNone,stream_mode:StreamMode|list[StreamMode]|NoneNone,interrupt_before:All|Sequence[str]|NoneNone,interrupt_after:All|Sequence[str]|NoneNone,subgraphs:boolFalse,)-Iterator[dict[str,Any]|Any]:...abstractmethoddefastream(self,input:InputT|Command|None,config:RunnableConfig|NoneNone,*,context:ContextT|NoneNone,stream_mode:StreamMode|list[StreamMode]|NoneNone,interrupt_before:All|Sequence[str]|NoneNone,interrupt_after:All|Sequence[str]|NoneNone,subgraphs:boolFalse,)-AsyncIterator[dict[str,Any]|Any]:...abstractmethoddefinvoke(self,input:InputT|Command|None,config:RunnableConfig|NoneNone,*,context:ContextT|NoneNone,interrupt_before:All|Sequence[str]|NoneNone,interrupt_after:All|Sequence[str]|NoneNone,)-dict[str,Any]|Any:...abstractmethodasyncdefainvoke(self,input:InputT|Command|None,config:RunnableConfig|NoneNone,*,context:ContextT|NoneNone,interrupt_before:All|Sequence[str]|NoneNone,interrupt_after:All|Sequence[str]|NoneNone,)-dict[str,Any]|Any:...执行体支持中断/恢复的方式执行所以在中断时需要将当时的状态以Checkpoint的形式保存下来恢复执行的时候利用它们恢复现场。持久化机制也使时间旅行成为可能我们可以从任一个Checkpoint开始执行。也正是因为此持久化机制的存在我们可以提取某一个Superstep的状态还可以查看整个执行历史这两个功能分别对应PregelProtocol的get_state/aget_state和get_state_history/aget_state_history方法。具体的状态以StateSnapshot对象描述的快照表示。执行体应该具有将执行结果作为新的状态进行保存的能力所以PregelProtocol定义了update_state/aupdate_state和bulk_update_state/abulk_update_state方法前者保存单一状态更新后者对多个状态更新进行批量执行。单一状态更新通过如下这个名为StateUpdate的命名元组表示我们不仅可以利用values字段得到以字典形式表示的状态值还可以通过as_node和task_id字段的得到实施更新的Node和具体任务标识。classStateUpdate(NamedTuple):values:dict[str,Any]|Noneas_node:str|NoneNonetask_id:str|NoneNone5. 嵌套结构我们一直在强调图的嵌套结构这种结构也可以从Pregel、PregelNode和PregelProtocol在三个类型的定义。一个Pregel是PregelProtocol的实现、作为其节点的PregelNode对象可以由一个或者多个PregelProtocol组成表示子图的subgraphs字段返回一个PregelProtocol对象的序列。classPregelNode:subgraphs:Sequence[PregelProtocol]classPregel(PregelProtocol[StateT,ContextT,InputT,OutputT],Generic[StateT,ContextT,InputT,OutputT]):defget_subgraphs(self,*,namespace:str|NoneNone,recurse:boolFalse)-Iterator[tuple[str,PregelProtocol]]asyncdefaget_subgraphs(self,*,namespace:str|NoneNone,recurse:boolFalse)-AsyncIterator[tuple[str,PregelProtocol]]PregelNode的subgraphs字段提供了子图的静态注册其实任何一个Pregel对象都可以在无需注册前提下被另一个Pregel的节点调用而且反映当前执行上下文的一些执行配置会通过上下文变量ContenxtVars流向作为子图的Pregel对象。前面我们演示子图调用涉及的Checkpoint命名空间的例子已经充分体现了这一点。但是这种显式的静态声明对于图的静态图分析与可视化有着积极的作用。