LangGraph定义状态（State）

1.State概述

State，是langgraph的核心数据结构，是一个贯穿整个工作流执行过程的共享数据的结构，存储了工作流从开始到结束的所有必要信息（历史对话，检索到的文档，工具执行结果等），在节点之间传递，并且被持久化保存。

State既可以是TypedDict类型，也可以是BaseModel类型

定义图的第一件事，就是定义图的State，State在各个节点中共享，而且每个节点都能进行修改，包含两部分：

• 模式（Schema）

  State的模式将作为图中所有边和节点的输入模式，可以是一个TypedDict或Pydanic模型

• 规约函数（Reducer function）

  指明如何把更新应用到状态上，所有Node都可对State的更新，然后用指定的reducer function函数应用这些更新

2.模式（Schema）

Schema主要使用三种

• state_schema

  完整内部状态，包含所有的节点可能读写的字段，必须指定，不能为空

• input_schema

  定义state接受什么，调用时只允许传入固定的键值对，多余的key会被直接过滤，可实现参数校验和隔离，是state_schema的一个子集，可选指定，不指定默认等于state_schema

• output_schema

  出参规范，指定输出结果里面只包含哪些key，大量中间和私有的状态属性不会暴露，实现输出收紧，隐私隔离，可选指定，不指定默认等于state_schema

三者搭配可以实现：外部只传必要参数，内部自由拓展中间状态，外部只拿目标结果。

例：以往的StateGraph(MsgState)写法，默认指定state_schema是MsgState

graph = StateGraph(MsgState)

现在的graph = StateGraph()明确的指定schema

多传入的'phone': '13099987654'被忽略，流转到最后只输出resp

langgraph建议使用字段覆盖级更新，只把更新了的字段返回，没有指定合并策略默认会覆盖合并，只更新返回的key，未返回的key保留原值，return {'resp': 'sin(a)'}只会覆盖resp对应的值，未返回的key保留原值。

from typing import TypedDict from langgraph.graph import StateGraph from langgraph.constants import START from langgraph.constants import END class DemoState(TypedDict): user_input: str resp: str count: int process_data: dict phone: str class InputState(TypedDict): user_input: str class OutputState(TypedDict): resp: str def node1(state: DemoState) -> dict: print('node1 ') print(state) return {'resp': 'sin(a)'} def node2(state: DemoState) -> dict: print('node2 ') print(state) return state if __name__ == "__main__": graph = StateGraph( input_schema=InputState, output_schema=OutputState, state_schema=DemoState ) graph.add_node('node1', node1) graph.add_node('node2', node2) graph.add_edge(START, 'node1') graph.add_edge('node1', 'node2') graph.add_edge('node2', END) app = graph.compile() app.get_graph().print_ascii() print('*' * 30) res = app.invoke({ 'user_input': '什么是正弦函数', 'phone': '13099987654' }) print('res') print(res)

+-----------+ | __start__ | +-----------+ * * * +-------+ | node1 | +-------+ * * * +-------+ | node2 | +-------+ * * * +---------+ | __end__ | +---------+ ****************************** node1 {'user_input': '什么是正弦函数'} node2 {'user_input': '什么是正弦函数', 'resp': 'sin(a)'} res {'resp': 'sin(a)'}

3.规约函数（Reducer function）

Reducer是理解节点更新如何应用于State的关键，节点更新的方式可能有很多种，不仅仅是覆盖，还有追加和合并。

State中每个键都有自己独立的reducer函数，如果未显式指定reducer函数，则默认的更新行为是覆盖。

例：name，age会被新的值覆盖掉

from typing import TypedDict from langgraph.constants import START from langgraph.constants import END from langgraph.graph import StateGraph # 状态类 class DemoState(TypedDict): name: str age: int # 节点 def name(state: DemoState) -> dict: new_name = f"qiangqiang, { state['name'] }" print(new_name) return {'name': new_name} # 节点 def age(state: DemoState) -> dict: new_age = state['age'] + 10 print(new_age) return {'age': new_age} if __name__ == "__main__": graph = StateGraph(DemoState) graph.add_node('name', name) graph.add_node('age', age) graph.add_edge(START, 'name') graph.add_edge('name', 'age') graph.add_edge('age', END) app = graph.compile() res = app.invoke({'name':'lzj', 'age':15}) print(res)

qiangqiang, lzj 25 {'name': 'qiangqiang, lzj', 'age': 25}

除了默认规约函数外，langgraph还提供了几个常见规约函数：

1.add_messages

消息追加，专用于和大模型对话，声明messages: Annotated[List, add_messages]表明messages是一个消息追加规约变化的状态变量

from typing import Annotated, TypedDict, List from langgraph.constants import START, END from langgraph.graph import add_messages, StateGraph class MsgState(TypedDict): messages: Annotated[List, add_messages] messages2: str def node1(state: MsgState): return {'messages': '在吗？', 'messages2': '在吗？'} def node2(state: MsgState): return {'messages': '你好啊！', 'messages2': '你好啊！'} if __name__ == '__main__': graph = StateGraph(MsgState) graph.add_node('node1', node1) graph.add_node('node2', node2) graph.add_edge(START, 'node1') graph.add_edge('node1', 'node2') graph.add_edge('node2', END) app = graph.compile() res = app.invoke({'messages': 'hi', 'messages2': 'hi'}) print(res['messages']) print('*'*50) print(res['messages2'])

[HumanMessage(content='hi', additional_kwargs={}, response_metadata={}, id='5e8e65ed-d333-43db-b36b-898c3a996686'), HumanMessage(content='在吗？', additional_kwargs={}, response_metadata={}, id='0c8cc7b2-e7fd-4efe-9288-0ed770ba267b'), HumanMessage(content='你好啊！', additional_kwargs={}, response_metadata={}, id='a38a474a-361e-4edf-828f-530e58b6991d')] ************************************************** 你好啊！

2.add追加

add可以实现列表追加

import operator from typing import Annotated, TypedDict, List from langgraph.constants import START, END from langgraph.graph import add_messages, StateGraph class MsgState(TypedDict): msg: Annotated[List[int], operator.add] def node1(state: MsgState): return {'msg': [1, 2]} def node2(state: MsgState): return {'msg': [3, 4]} if __name__ == '__main__': graph = StateGraph(MsgState) graph.add_node('node1', node1) graph.add_node('node2', node2) graph.add_edge(START, 'node1') graph.add_edge('node1', 'node2') graph.add_edge('node2', END) app = graph.compile() res = app.invoke({'msg': [0]}) print(res['msg'])

[0, 1, 2, 3, 4]

除此之外，还能实现字符串拼接和数值累加，比较简单，原理相同，不再赘述

字符串拼接：msg: Annotated[str, operator.add]
数值累加：msg: Annotated[int, operator.add]