文章目录GraphAPI之Graph(图)定义核心构建流程GraphAPI之State(状态)定义图的schema图的Reducer最佳实践建议GraphAPI之Graph(图)定义图是一种由节点和边组成的用于描述节点之间关系的数据结构分为无向图和有向图。有向图是带有方向的图LangGraph通过有向图定义AI工作流中的执行步骤和执行顺序从而实现复杂、有状态、可循环的应用程序逻辑。核心State状态图的全局上下文。你可以把它理解为图的“共享内存”。所有的节点都可以读取和修改这个状态。Nodes节点代表具体的执行步骤通常是一个 Python 函数或工具。每个节点接收当前的State执行某些逻辑如调用大模型、查询数据库然后返回更新后的State。Edges边定义了节点之间的流转方向和控制流。构建流程定义状态可选但推荐定义各节点节点就是方法初始化一个StateGraph实例添加节点添加边(将所有的节点连接起来)编译图执行工作流from typingimportTypedDictfrom langgraph.constantsimportSTART,ENDfrom langgraph.graphimportStateGraph#1.定义State可选但推荐classGraphState(TypedDict):process_data:dict #2.定义节点Nodedefinput_node(graph_state:GraphState)-GraphState:print(finput_node:{graph_state[process_data]})return{process_data:{input:111}}defprocess_node(graph_state:GraphState)-GraphState:print(fprocess_node:{graph_state[process_data]})return{process_data:{input:222}}defoutput_node(graph_state:GraphState)-GraphState:print(foutput_node:{graph_state[process_data]})return{process_data:{output:333}}#4.构建图GraphgraphStateGraph(GraphState)#4.1添加节点 graph.add_node(input_node,input_node)graph.add_node(process_node,process_node)graph.add_node(output_node,output_node)graph.add_edge(START,input_node)graph.add_edge(input_node,process_node)graph.add_edge(process_node,output_node)graph.add_edge(output_node,END)#5.编译 appgraph.compile()#6.运行 resultapp.invoke({process_data:{input:xxx}})print(result)GraphAPI之State(状态)定义State状态是整个图的基石图的“共享内存”类似于全局上下文所有的节点都能读写这个状态。在 LangGraph 中通常使用 Python 的 TypedDict 或 Pydantic 的 BaseModel来定义状态。方式A使用TypedDict最常用、最轻量 from typingimportTypedDictclassMyGraphState(TypedDict):input_query:str generation:str steps_count:int方式B使用Pydantic适合需要数据校验、类型转换的复杂场景 from pydanticimportBaseModel,FieldclassMyPydanticState(BaseModel):input_query:str generation:str# 提供默认值和校验 steps_count:intField(default0,ge0)在LangGraph中State状态是一个贯穿整个工作流执行过程中的共享数据的结构代表当前快照它存储了从工作流开始到结束的所有必要的信息(历史对话、检索到的文档、工具执行结果等)。状态在各个节点中共享且每个节点都可以修改状态包含两部分:- 图的模式(schema) - 规约函数(reducer functions)指明如何把更新应用到状态上。图的schema包含state_schema定义图的完整内部状态包含了所有节点可能读写的字段必须指定不能为空特点1. 是图的全局状态空间 2. 所有节点都可以访问和写入这个schema中的任何字段input_schema定义定义图接受什么输入是state_schema的子集特点1. 可选参数如果不指定默认等于state_schema 2. 限制图的输入接口只能传入这些字段output_schema定义定义图返回什么输出是state_schema的子集特点1. 可选参数如果不指定默认等于state_schema 2. 限制图的输出接口只返回这些字段学术上指定3种schema但实践过程中只使用state_schema即可图的Reducer定义**规约函数决定了节点产生的重新如何作用到StateState中的每个字段都拥有自己的独立规约函数。规约函数有多种类型如果未显式指定则默认所有对该字段的更新都会直接覆盖**旧值。一句话规约函数就是字段级合并策略它让节点只需吐出增量框架负责按规则把增量写入全局状态State常见合并策略default默认覆盖更新add_messages消息列表追加operator.add将元素追加到现有元素中支持列表、字符串、数值类型的追operator.mul用于数值相乘自定义Reducer 支持用户自定义合并逻辑案例from typingimportTypedDict,Listfrom langgraph.constantsimportSTART,ENDfrom langgraph.graphimportStateGraph# 需求如果未指定reducer函数默认对该字段进行覆盖行为 #1.定义State# 未指定合并策略reducer默认覆盖classDefaultReducerState(TypedDict):name:str hobby:List[str]#2.定义节点Nodedefnode_1(state:DefaultReducerState)-dict:print(fnode_1:{state[name]})print(fnode_1:{state[hobby]})return{name:Alice,hobby:[篮球]}defnode_2(state:DefaultReducerState)-dict:print(fnode_2:{state[name]})print(fnode_2:{state[hobby]})return{name:Bob,hobby:[足球,乒乓球]}#4.构建图GraphgraphStateGraph(DefaultReducerState)#4.1添加节点 graph.add_node(node_1,node_1)graph.add_node(node_2,node_2)graph.add_edge(START,node_1)graph.add_edge(node_1,node_2)graph.add_edge(node_2,END)#5.编译 appgraph.compile()#6.运行 resultapp.invoke({name:Alice,hobby:[nothing]})print(result)BN最佳实践建议按需使用 Annotated只有需要历史留痕如 messages、logs或需要累加的字段才加 Reducer其余字段如状态开关、临时变量保持默认的覆盖模式。结构清晰即使使用TypedDict也尽量为字段写好类型注解配合 IDE如 VSCode/PyCharm的类型推导编写节点时能极大减少因拼错 key 导致的 Bug。不要在节点内直接修改输入始终通过return {“key”: “value” 的方式让 LangGraph 去更新状态不要在节点内部做类似state[“list”].append(x) 的原位修改In-place mutation这会导致时间旅行Time Travel和调试功能失效。