LangGraph編程基本圍繞StateGraph進(jìn)行所以我們有必要對(duì)這個(gè)類(lèi)型具有一個(gè)深刻的認(rèn)識(shí)。這是一個(gè)泛型類(lèi)型四個(gè)泛型參數(shù)StateT、ContextT、InputT和OutputT分別表示狀態(tài)、靜態(tài)上下文、輸入和輸出類(lèi)型而且它們的類(lèi)型都是一個(gè)StateLike類(lèi)型。StateLike是四個(gè)類(lèi)型的聯(lián)合其中前三個(gè)針對(duì)類(lèi)TypedDict和Dataclass的協(xié)議。針對(duì)StateLike的定義我們一般會(huì)使用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. 節(jié)點(diǎn)、邊、通道、ManagedValue和SchemaStateGraph使用StateLike類(lèi)型的主要目的是利用它們對(duì)Schema的描述能力對(duì)應(yīng)的字段分別命名為state_schema、context_schema、input_schema和output_schema。創(chuàng)建StateGraph對(duì)象時(shí)state_schema類(lèi)型是必需要指定的。如果作為輸入輸出Schema類(lèi)型的input_schema和output_schema沒(méi)有顯式指定它們默認(rèn)與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這三個(gè)Schema類(lèi)型針對(duì)數(shù)據(jù)成員的描述最終會(huì)轉(zhuǎn)換成Pregel的通道或者M(jìn)anagedValue以下統(tǒng)稱(chēng)為通道針對(duì)通道的創(chuàng)建實(shí)現(xiàn)在私有方法_add_schema方法中該方法會(huì)將它們添加到schemas字段對(duì)應(yīng)的兩層字典中第一層和第二層字典的Key分別為Schema類(lèi)型和通道的名稱(chēng)。所有的通道和ManagedValue統(tǒng)一存儲(chǔ)在channels和managed字段對(duì)應(yīng)的字典中。圖由節(jié)點(diǎn)和邊構(gòu)建而成添加的節(jié)點(diǎn)和確定的邊被保存在nodes和edges字段中運(yùn)行時(shí)動(dòng)態(tài)決定的條件邊則被作為分支存儲(chǔ)在branches字段中。waiting_edges字典用于存儲(chǔ)多對(duì)一的邊即某個(gè)后續(xù)節(jié)點(diǎn)依賴(lài)多個(gè)前序節(jié)點(diǎn)需要等所有前序節(jié)點(diǎn)都完成執(zhí)行后才能執(zhí)行。系統(tǒng)會(huì)為狀態(tài)類(lèi)型的每個(gè)成員創(chuàng)建一個(gè)通道。2. 通道的類(lèi)型如何決定我們可以通過(guò)標(biāo)注Annotated方式來(lái)決定通道類(lèi)型,具體規(guī)則如下默認(rèn)采用的通道類(lèi)型為L(zhǎng)astValue可以利用Annotated的第二個(gè)參數(shù)指定通道類(lèi)型比如Annotated[int, EphemeralValue];如果希望的類(lèi)型為BinaryOperatorAggregate,只需要利用Annotated的第二個(gè)參數(shù)指定對(duì)應(yīng)的reducer函數(shù)就可以了。如下的程序演示了上述這個(gè)規(guī)則。我們?yōu)闋顟B(tài)類(lèi)型State定義了四個(gè)成員foo沒(méi)有任何標(biāo)注bar提供了針對(duì)EphemeralValue類(lèi)型的標(biāo)注baz提供了針對(duì)operator.add這個(gè)reducer函數(shù)的標(biāo)注qux 標(biāo)注的RemainingStepsManager是一個(gè)用于計(jì)算當(dāng)前剩余步數(shù)的ManagedValue類(lèi)型。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__})我們創(chuàng)建了一個(gè)包含單節(jié)點(diǎn)test_node的StateGraph對(duì)象在將其編譯成Pregel對(duì)象后我們將它的channels字典存儲(chǔ)的通道和ManagedValue的類(lèi)型輸出來(lái)channels字典存儲(chǔ)的是通道實(shí)例和ManagedValue類(lèi)型因?yàn)槲覀兛偸鞘褂肕anagedValue的靜態(tài)方法get讀取對(duì)應(yīng)的值。從如下的輸出結(jié)果可以看出四個(gè)狀態(tài)成員對(duì)應(yīng)的通道和ManagedValue的類(lèi)型是吻合的。foo: LastValue bar: EphemeralValue baz: BinaryOperatorAggregate qux: RemainingStepsManager __start__: EphemeralValue __pregel_tasks: Topic branch:to:test_node: EphemeralValue如果狀態(tài)Schema類(lèi)型沒(méi)有__annotations__屬性比如基礎(chǔ)的內(nèi)置類(lèi)型和容器類(lèi)型此時(shí)整個(gè)狀態(tài)將視為一個(gè)整體并為其創(chuàng)建一個(gè)名為_(kāi)_root__的單一通道通道類(lèi)型為L(zhǎng)astValue。如下這個(gè)將字符串作為狀態(tài)Schema類(lèi)型的演示程序體現(xiàn)了這一點(diǎn)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. 輸入和輸出通道如何決定如果沒(méi)有顯式指定輸入和輸出Schema類(lèi)型它們將與狀態(tài)共享相同的Schema而且狀態(tài)成員對(duì)應(yīng)的通道將默認(rèn)作為輸入和輸出通道。由于ManagedValue是只讀的所以不能作為輸入。而且它一般用于返回運(yùn)行時(shí)的狀態(tài)所以也不作為輸出。其實(shí)__init__方法針對(duì)Schema的構(gòu)建已經(jīng)體現(xiàn)了這一點(diǎn)針對(duì)輸入和輸出Schema調(diào)用_add_schema方法時(shí)將allow_managed參數(shù)設(shè)置為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對(duì)象會(huì)將輸入通道統(tǒng)一設(shè)置為_(kāi)_start__所以我們提供的整個(gè)輸入將統(tǒng)一寫(xiě)入此通道。__start__通道將會(huì)觸發(fā)執(zhí)行同名的節(jié)點(diǎn)后者會(huì)將輸入拆解寫(xiě)入對(duì)應(yīng)的通道然后驅(qū)動(dòng)執(zhí)行入口節(jié)點(diǎn)。對(duì)于上面演示程序創(chuàng)建的agent我們可以通過(guò)如下的斷言來(lái)證明這一點(diǎn)。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函數(shù)創(chuàng)建Agent可以進(jìn)一步標(biāo)注如下幾個(gè)類(lèi)型來(lái)顯式控制某個(gè)狀態(tài)成員的輸入輸出特性O(shè)mitFromInput和OmitFromOutput分別表示將指定的狀態(tài)成員從輸入或者輸出Schema中剔除PrivateStateAttr則表示同時(shí)從輸入和輸出Schema中剔除成為一個(gè)內(nèi)部使用的私有狀態(tài)成員。dataclassclassOmitFromSchema:input:boolTrueoutput:boolTrueOmitFromInputOmitFromSchema(inputTrue,outputFalse)OmitFromOutputOmitFromSchema(inputFalse,outputTrue)PrivateStateAttrOmitFromSchema(inputTrue,outputTrue)但是針對(duì)它們的標(biāo)注對(duì)于直接利用StateGraph構(gòu)建的Agent是無(wú)效的。如下的演示程序成分說(shuō)明了這一點(diǎn)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]