1. reducer
State中的每个键 都有自己独立的reducer，通过指定的reducer函数进行数据的更新
即，对于每个节点的返回值，state有自己的一套更新策略，可以人工通过reducer进行定制
2. 一个使用reducer的例子（Annotated）
from 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和invoke
graph.stream({"messages": [("user", user_input)]})这个是一个dict类型的东西，里面是节点名：节点返回值这样构成的
invoke是直接返回完整的最终state
