LangGraph编程基本围绕StateGraph进行，所以我们有必要对这个类型具有一个深刻的认识。这是一个泛型类型，四个泛型参数StateT、ContextT、InputT和OutputT分别表示状态、静态上下文、输入和输出类型，而且它们的类型都是一个StateLike类型。StateLike是四个类型的联合，其中前三个针对类TypedDict和Dataclass的协议。针对StateLike的定义，我们一般会使用TypedDict、DataClass和Pydantic模型。
classStateGraph(Generic[StateT,ContextT,InputT,OutputT])StateT=TypeVar("StateT",bound=StateLike)StateLike:TypeAlias=TypedDictLikeV1|TypedDictLikeV2|DataclassLike|BaseModel ContextT=TypeVar("ContextT",bound=StateLike|None,default=None)InputT=TypeVar("InputT",bound=StateLike,default=StateT)OutputT=TypeVar("OutputT",bound=StateLike,default=StateT)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和Schema
StateGraph使用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]|None=None,*,input_schema:type[InputT]|None=None,output_schema:type[OutputT]|None=None,**kwargs:Unpack[DeprecatedKwargs],)->None:...self.nodes={}self.edges=set()self.branches=defaultdict(dict)self.schemas={}self.channels={}self.managed={}self.compiled=Falseself.waiting_edges=set()self.state_schema=state_schema self.input_schema=cast(type[InputT],input_schemaorstate_schema)self.output_schema=cast(type[OutputT],output_schemaorstate_schema)self.context_schema=context_schema self._add_schema(self.state_schema)self._add_schema(self.input_schema,allow_managed=False)self._add_schema(self.output_schema,allow_managed=False)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_managed=False)self._add_schema(self.output_schema,allow_managed=False)由StateGraph编译生成的Pregel对象会将输入通道统一设置为__start__，所以我们提供的整个输入将统一写入此通道。__start__通道将会触发执行同名的节点，后者会将输入拆解写入对应的通道，然后驱动执行入口节点。对于上面演示程序创建的agent，我们可以通过如下的断言来证明这一点。
assertagent.input_channels=="__start__"assertagent.output_channels==["foo","bar","baz"]input:State={"foo":"abc","bar":123,"baz":["a","b"]}result=agent.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:bool=Trueoutput:bool=TrueOmitFromInput=OmitFromSchema(input=True,output=False)OmitFromOutput=OmitFromSchema(input=False,output=True)PrivateStateAttr=OmitFromSchema(input=True,output=True)但是针对它们的标注对于直接利用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']agent=create_agent(model=ChatOpenAI(model="gpt-5.2-chat"),state_schema=State)assertagent.input_channels=="__start__"assertagent.output_channels==['messages','structured_response','foo','baz']
