LangGraph State 状态

1.概述

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

State既可以是TypedDict类型，也可以是pydantic.BaseModel类型，关于TypedDict和Pydantic可见:TypedDict和Pydantic，langgraph官方推荐TypedDict。

定义图的第一件事，就是定义图的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')]
**************************************************
你好啊！

与langchain结合使用：

from typing import Annotated, TypedDict, List

from langchain_core.messages import HumanMessage
from langgraph.constants import START, END
from langgraph.graph import add_messages, StateGraph
import os

from langchain.chat_models import init_chat_model

from dotenv import load_dotenv

load_dotenv(encoding='utf-8')

class MsgState(TypedDict):
    messages: Annotated[List, add_messages]

llm = init_chat_model(
    model = 'deepseek-chat',
    model_provider = 'openai',
    api_key = os.getenv('DEEPSEEK_API_KEY'),
    base_url = 'https://api.deepseek.com'
)


def chat(state: MsgState):
    reply = llm.invoke(state['messages'])
    return {'messages': [reply]}

if __name__ == '__main__':
    graph = StateGraph(MsgState)
    graph.add_node('chat', chat)

    graph.add_edge(START, 'chat')
    graph.add_edge('chat', END)

    app = graph.compile()

    #res = app.invoke({'messages': '1加1等于几'})
    #res = app.invoke({'messages': [('user','1加1等于几')]})
    res = app.invoke({
        'messages': [HumanMessage(content='1加1等于几')]
    })
    print(res)
{'messages': [HumanMessage(content='1加1等于几', additional_kwargs={}, response_metadata={}, id='d3856bc4-5c1f-4f9b-921e-35b212c32080'), AIMessage(content='1加1等于**2**。  \n\n这是最基本的数学加法运算。但在不同情境下（如二进制中），1+1 可能等于 **10**（二进制）。如果从幽默角度回答，也可以说等于“田”字（把两个1放在一起）。  \n\n不过按照通常的数学定义，答案是 **2**。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 69, 'prompt_tokens': 9, 'total_tokens': 78, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cache_write_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 9}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '0ef8240f-4f93-4758-a771-3bad5dbdb32c', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019f5929-798e-7931-83ca-70831a70685f-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 9, 'output_tokens': 69, 'total_tokens': 78, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})]}

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]
数值累乘：msg: Annotated[float, operator.mul]

但是需要注意，langgraph的设计，operator.mul累乘和operator.add累加从0开始规约，对于加法恒等元是0，没有问题，但是对于乘法，无论怎么乘结果都是0！

对于不满足需求的内置规约函数，可以自己实现自定义的规约函数来替代，比如自定义规约函数来实现正确的累乘

例：自定义规约custom_mul令累乘恒等元为1

import operator
from typing import Annotated, TypedDict, List

from langgraph.constants import START, END
from langgraph.graph import add_messages, StateGraph

def custom_mul(current: float, update: float) -> float:
    # 如果是第一次调用，current会是默认值0.0
    if current == 0.0:
        return 1.0 * update
    return current * update


class MsgState(TypedDict):
    msg: Annotated[float, custom_mul]



def node1(state: MsgState):
    return {'msg': 3.0}

def node2(state: MsgState):
    return {'msg': 4.0}

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': 2.0})
    print(res['msg'])
24.0

"如果文章对您有帮助，可以请作者喝杯咖啡吗？"

微信二维码

微信支付

支付宝二维码

支付宝


LangGraph
#原创 #AI #大模型 #框架 #Agent #智能体 #LangGraph
LangGraph State 状态
https://blog.liuzijian.com/post/2026/06/15/langgraph-state/
作者
Liu Zijian
发布于
2026年6月15日
许可协议