LangGraph定义状态（State）

未完待续

1.State概述

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

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

对比项	`TypedDict`	`BaseModel`
来源	标准库	pydantic
定位	类型提示，轻量字典	强类型数据类型，含验证逻辑
运行时检查	无运行时校验	自动校验字段类型，默认值
继承自	`dict`	`BaseModel`
性能	快	稍慢，需要解析和验证
序列化/反序列化	手动处理	自动，`.dict()` `.json()`
用途场景	简单的数据结构定义	需要验证，解析和约束的模型
langgraph支持	官方推荐	不推荐，除非自己控制模型转换

定义图的第一件事，就是定义图的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

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)'}