[LangGraph编译原理-02]面向通道定义Agent的状态

LangGraph编程基本围绕StateGraph进行所以我们有必要对这个类型具有一个深刻的认识。这是一个泛型类型四个泛型参数StateT、ContextT、InputT和OutputT分别表示状态、静态上下文、输入和输出类型而且它们的类型都是一个StateLike类型。StateLike是四个类型的联合其中前三个针对类TypedDict和Dataclass的协议。针对StateLike的定义我们一般会使用TypedDict、DataClass和Pydantic模型。classStateGraph(Generic[StateT,ContextT,InputT,OutputT])StateTTypeVar(StateT,boundStateLike)StateLike:TypeAliasTypedDictLikeV1|TypedDictLikeV2|DataclassLike|BaseModel ContextTTypeVar(ContextT,boundStateLike|None,defaultNone)InputTTypeVar(InputT,boundStateLike,defaultStateT)OutputTTypeVar(OutputT,boundStateLike,defaultStateT)classTypedDictLikeV1(Protocol):__required_keys__:ClassVar[frozenset[str]]__optional_keys__:ClassVar[frozenset[str]]classTypedDictLikeV2(Protocol):__required_keys__:frozenset[str]__optional_keys__:frozenset[str]classDataclassLike(Protocol):__dataclass_fields__:ClassVar[dict[str,Field[Any]]]1. 节点、边、通道、ManagedValue和SchemaStateGraph使用StateLike类型的主要目的是利用它们对Schema的描述能力对应的字段分别命名为state_schema、context_schema、input_schema和output_schema。创建StateGraph对象时state_schema类型是必需要指定的。如果作为输入输出Schema类型的input_schema和output_schema没有显式指定它们默认与state_schema相同。classStateGraph(Generic[StateT,ContextT,InputT,OutputT]):edges:set[tuple[str,str]]nodes:dict[str,StateNodeSpec[Any,ContextT]]branches:defaultdict[str,dict[str,BranchSpec]]channels:dict[str,BaseChannel]managed:dict[str,ManagedValueSpec]schemas:dict[type[Any],dict[str,BaseChannel|ManagedValueSpec]]waiting_edges:set[tuple[tuple[str,...],str]]compiled:boolstate_schema:type[StateT]context_schema:type[ContextT]|Noneinput_schema:type[InputT]output_schema:type[OutputT]def__init__(self,state_schema:type[StateT],context_schema:type[ContextT]|NoneNone,*,input_schema:type[InputT]|NoneNone,output_schema:type[OutputT]|NoneNone,**kwargs:Unpack[DeprecatedKwargs],)-None:...self.nodes{}self.edgesset()self.branchesdefaultdict(dict)self.schemas{}self.channels{}self.managed{}self.compiledFalseself.waiting_edgesset()self.state_schemastate_schema self.input_schemacast(type[InputT],input_schemaorstate_schema)self.output_schemacast(type[OutputT],output_schemaorstate_schema)self.context_schemacontext_schema self._add_schema(self.state_schema)self._add_schema(self.input_schema,allow_managedFalse)self._add_schema(self.output_schema,allow_managedFalse)state_schema、input_schema和output_schema这三个Schema类型针对数据成员的描述最终会转换成Pregel的通道或者ManagedValue以下统称为通道针对通道的创建实现在私有方法_add_schema方法中该方法会将它们添加到schemas字段对应的两层字典中第一层和第二层字典的Key分别为Schema类型和通道的名称。所有的通道和ManagedValue统一存储在channels和managed字段对应的字典中。图由节点和边构建而成添加的节点和确定的边被保存在nodes和edges字段中运行时动态决定的条件边则被作为分支存储在branches字段中。waiting_edges字典用于存储多对一的边即某个后续节点依赖多个前序节点需要等所有前序节点都完成执行后才能执行。系统会为状态类型的每个成员创建一个通道。2. 通道的类型如何决定我们可以通过标注Annotated方式来决定通道类型,具体规则如下默认采用的通道类型为LastValue可以利用Annotated的第二个参数指定通道类型比如Annotated[int, EphemeralValue];如果希望的类型为BinaryOperatorAggregate,只需要利用Annotated的第二个参数指定对应的reducer函数就可以了。如下的程序演示了上述这个规则。我们为状态类型State定义了四个成员foo没有任何标注bar提供了针对EphemeralValue类型的标注baz提供了针对operator.add这个reducer函数的标注qux 标注的RemainingStepsManager是一个用于计算当前剩余步数的ManagedValue类型。fromtypingimportAnnotated,TypedDict,Required,NotRequiredfromlanggraph.graphimportStateGraphfromlanggraph.channelsimportEphemeralValuefromlanggraph.managed.is_last_stepimportRemainingStepsManagerimportinspectimportoperatorclassState(TypedDict):foo:strbar:Required[Annotated[int,EphemeralValue]]baz:Annotated[list[str],operator.add]qux:NotRequired[Annotated[int,RemainingStepsManager]]deftest_node(state:State):return{k:v*2for(k,v)instate.items()}# type: ignoreagent(StateGraph(State).add_node(test_node).add_edge(__start__,test_node).add_edge(test_node,__end__).compile())for(name,chan)inagent.channels.items():ifinspect.isclass(chan):print(f{name}:{chan.__name__})else:print(f{name}:{chan.__class__.__name__})我们创建了一个包含单节点test_node的StateGraph对象在将其编译成Pregel对象后我们将它的channels字典存储的通道和ManagedValue的类型输出来channels字典存储的是通道实例和ManagedValue类型因为我们总是使用ManagedValue的静态方法get读取对应的值。从如下的输出结果可以看出四个状态成员对应的通道和ManagedValue的类型是吻合的。foo: LastValue bar: EphemeralValue baz: BinaryOperatorAggregate qux: RemainingStepsManager __start__: EphemeralValue __pregel_tasks: Topic branch:to:test_node: EphemeralValue如果状态Schema类型没有__annotations__属性比如基础的内置类型和容器类型此时整个状态将视为一个整体并为其创建一个名为__root__的单一通道通道类型为LastValue。如下这个将字符串作为状态Schema类型的演示程序体现了这一点fromlanggraph.graphimportStateGraphfromlanggraph.channelsimportLastValuedeftest_node(state:list):returnstate*2agent(StateGraph(list)# type: ignore.add_node(test_node)# type: ignore.add_edge(__start__,test_node).add_edge(test_node,__end__).compile())assertagent.invoke(input[foobar])[foobar,foobar]assertisinstance(agent.channels[__root__],LastValue)3. 输入和输出通道如何决定如果没有显式指定输入和输出Schema类型它们将与状态共享相同的Schema而且状态成员对应的通道将默认作为输入和输出通道。由于ManagedValue是只读的所以不能作为输入。而且它一般用于返回运行时的状态所以也不作为输出。其实__init__方法针对Schema的构建已经体现了这一点针对输入和输出Schema调用_add_schema方法时将allow_managed参数设置为False。self._add_schema(self.state_schema)self._add_schema(self.input_schema,allow_managedFalse)self._add_schema(self.output_schema,allow_managedFalse)由StateGraph编译生成的Pregel对象会将输入通道统一设置为__start__所以我们提供的整个输入将统一写入此通道。__start__通道将会触发执行同名的节点后者会将输入拆解写入对应的通道然后驱动执行入口节点。对于上面演示程序创建的agent我们可以通过如下的断言来证明这一点。assertagent.input_channels__start__assertagent.output_channels[foo,bar,baz]input:State{foo:abc,bar:123,baz:[a,b]}resultagent.invoke(input)print(result)assertresult{foo:abcabc,bar:246,baz:[a,b,a,b,a,b]}如果使用crate_agent函数创建Agent可以进一步标注如下几个类型来显式控制某个状态成员的输入输出特性OmitFromInput和OmitFromOutput分别表示将指定的状态成员从输入或者输出Schema中剔除PrivateStateAttr则表示同时从输入和输出Schema中剔除成为一个内部使用的私有状态成员。dataclassclassOmitFromSchema:input:boolTrueoutput:boolTrueOmitFromInputOmitFromSchema(inputTrue,outputFalse)OmitFromOutputOmitFromSchema(inputFalse,outputTrue)PrivateStateAttrOmitFromSchema(inputTrue,outputTrue)但是针对它们的标注对于直接利用StateGraph构建的Agent是无效的。如下的演示程序成分说明了这一点fromtypingimportAnnotatedfromlanggraph.graphimportStateGraphfromlangchain.agentsimportcreate_agent,AgentStatefromlangchain_openaiimportChatOpenAIfromlangchain.agents.middleware.typesimportOmitFromOutput,OmitFromInput,PrivateStateAttrfromdotenvimportload_dotenv load_dotenv()classState(AgentState):foo:strbar:Annotated[str,OmitFromOutput]baz:Annotated[str,OmitFromInput]qux:Annotated[str,PrivateStateAttr]deftest_node(state:State):returnstate agent(StateGraph(State).add_node(test_node).add_edge(__start__,test_node).add_edge(test_node,__end__).compile())assertagent.input_channels__start__assertagent.output_channels[messages,jump_to,structured_response,foo,bar,baz,qux]agentcreate_agent(modelChatOpenAI(modelgpt-5.2-chat),state_schemaState)assertagent.input_channels__start__assertagent.output_channels[messages,structured_response,foo,baz]