1. reducerState中的每个键 都有自己独立的reducer通过指定的reducer函数进行数据的更新即对于每个节点的返回值state有自己的一套更新策略可以人工通过reducer进行定制2. 一个使用reducer的例子Annotatedfrom langgraph.graph import StateGraph import operator from typing import Annotated, TypedDict, List from langgraph.graph import START, END class State(TypedDict): messages:Annotated[list[dict],operator.add] builder StateGraph(State) def addition(state): print(state) msg state[messages][-1] response {x: msg[x] 1} return {messages: [response]} def subtraction(state): print(state) msg state[messages][-1] response {x: msg[x] - 2} return {messages: [response]} # 向图中添加两个节点 builder.add_node(addition, addition) builder.add_node(subtraction, subtraction) # 构建节点之间的边 builder.add_edge(START, addition) builder.add_edge(addition, subtraction) builder.add_edge(subtraction, END) graph builder.compile() input_state {messages: [{x: 10}]} print(graph.invoke(input_state))这里主要看State类的定义这里用了一个东西Annotated 在python中用来增加特殊注解即本来是指定类型list[dict] 而 这里用Annotated 可以增加一个operator.add 即使用加法作为reducer函数所以该demo运行后的结果是多个字典合并 而非直接替换3. stream和invokegraph.stream({messages: [(user, user_input)]})这个是一个dict类型的东西里面是节点名节点返回值这样构成的invoke是直接返回完整的最终state
1. reducerState中的每个键 都有自己独立的reducer通过指定的reducer函数进行数据的更新即对于每个节点的返回值state有自己的一套更新策略可以人工通过reducer进行定制2. 一个使用reducer的例子Annotatedfrom langgraph.graph import StateGraph import operator from typing import Annotated, TypedDict, List from langgraph.graph import START, END class State(TypedDict): messages:Annotated[list[dict],operator.add] builder StateGraph(State) def addition(state): print(state) msg state[messages][-1] response {x: msg[x] 1} return {messages: [response]} def subtraction(state): print(state) msg state[messages][-1] response {x: msg[x] - 2} return {messages: [response]} # 向图中添加两个节点 builder.add_node(addition, addition) builder.add_node(subtraction, subtraction) # 构建节点之间的边 builder.add_edge(START, addition) builder.add_edge(addition, subtraction) builder.add_edge(subtraction, END) graph builder.compile() input_state {messages: [{x: 10}]} print(graph.invoke(input_state))这里主要看State类的定义这里用了一个东西Annotated 在python中用来增加特殊注解即本来是指定类型list[dict] 而 这里用Annotated 可以增加一个operator.add 即使用加法作为reducer函数所以该demo运行后的结果是多个字典合并 而非直接替换3. stream和invokegraph.stream({messages: [(user, user_input)]})这个是一个dict类型的东西里面是节点名节点返回值这样构成的invoke是直接返回完整的最终state
