langgraph-state-management

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

LangGraph State Management

LangGraph 状态管理

State Design Workflow

状态设计工作流

Follow this workflow when designing or modifying state for a LangGraph application:

Identify data requirements — What data flows through the graph?
Choose a schema pattern — Match the use case to a template
Define reducers — Decide how concurrent updates merge
Configure persistence — Select and set up a checkpointer
Validate and test — Run schema validation and reducer tests

为LangGraph应用设计或修改状态时，请遵循以下工作流：

明确数据需求 —— 图中需要流转哪些数据？
选择模式模板 —— 根据用例匹配合适的模板
定义Reducer —— 确定并发更新的合并方式
配置持久化 —— 选择并设置检查点工具
验证与测试 —— 运行模式验证和Reducer测试

Quick Start

快速开始

Python — Minimal Chat State

Python —— 极简聊天状态

python

from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_core.messages import AIMessage

class State(MessagesState):
    pass

def chat_node(state: State):
    return {"messages": [AIMessage(content="Hello!")]}

graph = StateGraph(State).add_node("chat", chat_node)
graph.add_edge(START, "chat").add_edge("chat", END)
app = graph.compile()

python

from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_core.messages import AIMessage

class State(MessagesState):
    pass

def chat_node(state: State):
    return {"messages": [AIMessage(content="Hello!")]}

graph = StateGraph(State).add_node("chat", chat_node)
graph.add_edge(START, "chat").add_edge("chat", END)
app = graph.compile()

Python — Subclass MessagesState

Python —— 继承MessagesState

For convenience, subclass the built-in

MessagesState

(includes

messages

with

add_messages

reducer):

python

from langgraph.graph import MessagesState

class State(MessagesState):
    documents: list[str]
    query: str

为方便使用，可继承内置的

MessagesState

（包含带有

add_messages

reducer的

messages

字段）：

python

from langgraph.graph import MessagesState

class State(MessagesState):
    documents: list[str]
    query: str

TypeScript — StateSchema with Zod

TypeScript —— 结合Zod的StateSchema

typescript

import { StateGraph, StateSchema, MessagesValue, ReducedValue, START, END } from "@langchain/langgraph";
import { AIMessage } from "@langchain/core/messages";
import { z } from "zod/v4";

const State = new StateSchema({
  messages: MessagesValue,
  documents: z.array(z.string()).default(() => []),
  count: new ReducedValue(
    z.number().default(0),
    { reducer: (current, update) => current + update }
  ),
});

const graph = new StateGraph(State)
  .addNode("chat", (state) => ({ messages: [new AIMessage("Hello!")] }))
  .addEdge(START, "chat")
  .addEdge("chat", END)
  .compile();

typescript

import { StateGraph, StateSchema, MessagesValue, ReducedValue, START, END } from "@langchain/langgraph";
import { AIMessage } from "@langchain/core/messages";
import { z } from "zod/v4";

const State = new StateSchema({
  messages: MessagesValue,
  documents: z.array(z.string()).default(() => []),
  count: new ReducedValue(
    z.number().default(0),
    { reducer: (current, update) => current + update }
  ),
});

const graph = new StateGraph(State)
  .addNode("chat", (state) => ({ messages: [new AIMessage("Hello!")] }))
  .addEdge(START, "chat")
  .addEdge("chat", END)
  .compile();

Schema Patterns

模式模板

Choose the pattern matching the application type. See references/schema-patterns.md for complete examples with both Python and TypeScript.

Pattern	Use Case	Key Fields
Chat	Conversational agents	Built-in `messages` from `MessagesState`
Research	Information gathering	`query` , `search_results` , `summary`
Workflow	Task orchestration	`task` , `status` (Literal), `steps_completed`
Tool-Calling	Agents with tools	`messages` , `tool_calls_made` , `should_continue`
RAG	Retrieval-augmented generation	`query` , `retrieved_docs` , `response`

Template files are available in

assets/

for each pattern:

```
assets/chat_state.py
```
— Chat application
```
assets/research_state.py
```
— Research agent
```
assets/workflow_state.py
```
— Workflow orchestration
```
assets/tool_calling_state.py
```
— Tool-calling agent

For RAG state patterns, use reference examples in references/schema-patterns.md.

根据应用类型选择对应的模式。完整的Python和TypeScript示例请参考references/schema-patterns.md。

模式	适用场景	核心字段
聊天	对话式Agent	来自 `MessagesState` 的内置 `messages`
研究	信息收集	`query` 、 `search_results` 、 `summary`
工作流	任务编排	`task` 、 `status` （Literal类型）、 `steps_completed`
工具调用	带工具的Agent	`messages` 、 `tool_calls_made` 、 `should_continue`
RAG	检索增强生成	`query` 、 `retrieved_docs` 、 `response`

模板文件可在

assets/

目录下获取，对应每种模式：

```
assets/chat_state.py
```
—— 聊天应用
```
assets/research_state.py
```
—— 研究Agent
```
assets/workflow_state.py
```
—— 工作流编排
```
assets/tool_calling_state.py
```
—— 工具调用Agent

RAG状态模式请参考references/schema-patterns.md中的示例。

Reducers

Reducer

Reducers control how state updates merge when nodes write to the same field.

Reducer用于控制当多个节点写入同一字段时，状态更新的合并方式。

Key Concepts

核心概念

No reducer → value is overwritten (last-write-wins)
With reducer → values are merged using the reducer function
A reducer takes
```
(existing_value, new_value)
```
and returns the merged result

无reducer → 值被覆盖（最后写入者获胜）
有reducer → 使用reducer函数合并值
Reducer接收
```
(existing_value, new_value)
```
参数并返回合并后的结果

Python: Annotated Type with Reducer

Python：带Reducer的Annotated类型

python

from typing import Annotated
import operator
from langgraph.graph import MessagesState

class State(MessagesState):
    # Overwrite (no reducer)
    query: str

    # Sum integers
    count: Annotated[int, operator.add]

    # Custom reducer
    results: Annotated[list[str], lambda left, right: left + right]

python

from typing import Annotated
import operator
from langgraph.graph import MessagesState

class State(MessagesState):
    # 覆盖（无reducer）
    query: str

    # 整数求和
    count: Annotated[int, operator.add]

    # 自定义reducer
    results: Annotated[list[str], lambda left, right: left + right]

TypeScript: ReducedValue and MessagesValue

TypeScript：ReducedValue与MessagesValue

typescript

const State = new StateSchema({
  query: z.string(),                    // Last-write-wins
  messages: MessagesValue,              // Built-in message reducer
  count: new ReducedValue(              // Custom reducer
    z.number().default(0),
    { reducer: (current, update) => current + update }
  ),
});

typescript

const State = new StateSchema({
  query: z.string(),                    // 最后写入者获胜
  messages: MessagesValue,              // 内置消息reducer
  count: new ReducedValue(              // 自定义reducer
    z.number().default(0),
    { reducer: (current, update) => current + update }
  ),
});

Built-in Reducers

内置Reducer

Reducer	Import	Behavior
`add_messages`	`langgraph.graph.message`	Append, update by ID, delete
`operator.add`	`operator`	Numeric addition or list concatenation
`MessagesValue`	`@langchain/langgraph`	JS equivalent of `add_messages`

Reducer	导入路径	行为
`add_messages`	`langgraph.graph.message`	追加、按ID更新、删除
`operator.add`	`operator`	数值相加或列表拼接
`MessagesValue`	`@langchain/langgraph`	JS版本的 `add_messages`

Bypass Reducers with Overwrite

使用Overwrite绕过Reducer

Replace accumulated state instead of merging:

python

from langgraph.types import Overwrite

def reset_messages(state: State):
    return {"messages": Overwrite(["fresh start"])}

替换累积状态而非合并：

python

from langgraph.types import Overwrite

def reset_messages(state: State):
    return {"messages": Overwrite(["fresh start"])}

Delete Messages

删除消息

python

from langchain_core.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES

python

from langchain_core.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES

Delete specific message

删除指定消息

{"messages": [RemoveMessage(id="msg_123")]}

Delete all messages

删除所有消息

{"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}


For advanced reducer patterns (deduplication, deep merge, conditional update, size-limited accumulators), see [references/reducers.md](references/reducers.md).

{"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}


高级Reducer模式（去重、深度合并、条件更新、限长累加器）请参考[references/reducers.md](references/reducers.md)。

Persistence

持久化

Persistence enables multi-turn conversations, human-in-the-loop, time travel, and crash recovery.

持久化支持多轮对话、人在环中、时间回溯和崩溃恢复。

Choosing a Backend

选择后端

Backend	Package	Use Case
InMemorySaver	`langgraph-checkpoint` (included)	Development, testing
SqliteSaver	`langgraph-checkpoint-sqlite`	Local workflows, single-instance
PostgresSaver	`langgraph-checkpoint-postgres`	Production, multi-instance
CosmosDBSaver	`langgraph-checkpoint-cosmosdb`	Azure production

Agent Server note: When using LangGraph Agent Server, checkpointers are configured automatically — no manual setup needed.

后端	包	适用场景
InMemorySaver	`langgraph-checkpoint` （内置）	开发、测试
SqliteSaver	`langgraph-checkpoint-sqlite`	本地工作流、单实例
PostgresSaver	`langgraph-checkpoint-postgres`	生产环境、多实例
CosmosDBSaver	`langgraph-checkpoint-cosmosdb`	Azure生产环境

Agent Server注意事项：使用LangGraph Agent Server时，检查点工具会自动配置，无需手动设置。

Python Setup

Python设置

python

undefined

python

undefined

Development

开发环境

from langgraph.checkpoint.memory import InMemorySaver graph = builder.compile(checkpointer=InMemorySaver())

Production (PostgreSQL)

生产环境（PostgreSQL）

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@host:5432/db" with PostgresSaver.from_conn_string(DB_URI) as checkpointer: # checkpointer.setup() # Run once for initial schema graph = builder.compile(checkpointer=checkpointer)

result = graph.invoke(
    {"messages": [{"role": "user", "content": "Hi"}]},
    {"configurable": {"thread_id": "session-1"}}
)

undefined

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@host:5432/db" with PostgresSaver.from_conn_string(DB_URI) as checkpointer: # checkpointer.setup() # 首次运行初始化模式 graph = builder.compile(checkpointer=checkpointer)

result = graph.invoke(
    {"messages": [{"role": "user", "content": "Hi"}]},
    {"configurable": {"thread_id": "session-1"}}
)

undefined

TypeScript Setup

TypeScript设置

typescript

// Development
import { MemorySaver } from "@langchain/langgraph";
const graph = builder.compile({ checkpointer: new MemorySaver() });

// Production (PostgreSQL)
import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres";
const checkpointer = PostgresSaver.fromConnString(DB_URI);
// await checkpointer.setup();  // Run once
const graph = builder.compile({ checkpointer });

typescript

// 开发环境
import { MemorySaver } from "@langchain/langgraph";
const graph = builder.compile({ checkpointer: new MemorySaver() });

// 生产环境（PostgreSQL）
import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres";
const checkpointer = PostgresSaver.fromConnString(DB_URI);
// await checkpointer.setup();  // 首次运行初始化
const graph = builder.compile({ checkpointer });

Thread Management

线程管理

Every invocation requires a

thread_id

to identify the conversation:

python

config = {"configurable": {"thread_id": "user-123-session-1"}}
result = graph.invoke({"messages": [...]}, config)

每次调用都需要

thread_id

来标识对话：

python

config = {"configurable": {"thread_id": "user-123-session-1"}}
result = graph.invoke({"messages": [...]}, config)

Subgraph Persistence

子图持久化

Provide the checkpointer only on the parent graph — LangGraph propagates it to subgraphs automatically:

python

parent_graph = parent_builder.compile(checkpointer=checkpointer)

仅需在父图上提供检查点工具——LangGraph会自动将其传播到子图：

python

parent_graph = parent_builder.compile(checkpointer=checkpointer)

Subgraphs inherit the checkpointer

子图会继承检查点工具


To give a subgraph its own separate memory:

```python
subgraph = sub_builder.compile(checkpointer=True)

For backend-specific configuration, migration between backends, and TTL settings, see references/persistence-backends.md.


要为子图设置独立的内存：

```python
subgraph = sub_builder.compile(checkpointer=True)

后端特定配置、后端间迁移和TTL设置请参考references/persistence-backends.md。

State Typing

状态类型

Python: TypedDict (Recommended)

Python：TypedDict（推荐）

python

from typing import TypedDict, Annotated, Literal

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next: Literal["agent1", "agent2", "FINISH"]
    context: dict

Note:
create_agent
state schemas support
TypedDict
for custom agent state. Prefer TypedDict for agent state extensions.

python

from typing import TypedDict, Annotated, Literal

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next: Literal["agent1", "agent2", "FINISH"]
    context: dict

注意：
create_agent
状态模式支持
TypedDict
用于自定义Agent状态。扩展Agent状态时优先使用TypedDict。

TypeScript: StateSchema with Zod

TypeScript：结合Zod的StateSchema

typescript

import { StateSchema, MessagesValue, ReducedValue, UntrackedValue } from "@langchain/langgraph";
import { z } from "zod/v4";

const AgentState = new StateSchema({
  messages: MessagesValue,
  currentStep: z.string(),
  retryCount: z.number().default(0),

  // Custom reducer
  allSteps: new ReducedValue(
    z.array(z.string()).default(() => []),
    { inputSchema: z.string(), reducer: (current, newStep) => [...current, newStep] }
  ),

  // Transient state (not checkpointed)
  tempCache: new UntrackedValue(z.record(z.string(), z.unknown())),
});

// Extract types for use outside the graph builder
type State = typeof AgentState.State;
type Update = typeof AgentState.Update;

For Pydantic validation, advanced type patterns, and migration from untyped state, see references/state-typing.md.

typescript

import { StateSchema, MessagesValue, ReducedValue, UntrackedValue } from "@langchain/langgraph";
import { z } from "zod/v4";

const AgentState = new StateSchema({
  messages: MessagesValue,
  currentStep: z.string(),
  retryCount: z.number().default(0),

  // 自定义reducer
  allSteps: new ReducedValue(
    z.array(z.string()).default(() => []),
    { inputSchema: z.string(), reducer: (current, newStep) => [...current, newStep] }
  ),

  // 临时状态（不存入检查点）
  tempCache: new UntrackedValue(z.record(z.string(), z.unknown())),
});

// 提取类型以供图构建器外部使用
type State = typeof AgentState.State;
type Update = typeof AgentState.Update;

Pydantic验证、高级类型模式和无类型状态迁移请参考references/state-typing.md。

Validation and Debugging

验证与调试

Validate State Schema

验证状态模式

Run the validation script to check schema structure:

bash

uv run scripts/validate_state_schema.py my_agent/state.py:MyState --verbose

Checks for: schema parsing issues, empty schemas, reducer annotation problems, message fields without reducers, routing fields without Literal types, and unsupported/unclear schema class patterns.

运行验证脚本检查模式结构：

bash

uv run scripts/validate_state_schema.py my_agent/state.py:MyState --verbose

检查内容包括：模式解析问题、空模式、Reducer注解问题、无Reducer的消息字段、无Literal类型的路由字段，以及不支持/不清晰的模式类模式。

Test Reducers

测试Reducer

Test reducer functions for correctness and edge cases:

bash

uv run scripts/test_reducers.py my_agent/reducers.py:extend_list --verbose

Tests: basic merge, empty inputs, None handling, type consistency, nested structures, large inputs.

测试Reducer函数的正确性和边缘情况：

bash

uv run scripts/test_reducers.py my_agent/reducers.py:extend_list --verbose

测试内容：基础合并、空输入、None处理、类型一致性、嵌套结构、大输入。

Inspect Checkpoints

检查检查点

Debug state evolution by inspecting saved checkpoints:

bash

undefined

通过检查已保存的检查点调试状态演变：

bash

undefined

List recent checkpoints

列出最近的检查点

uv run scripts/inspect_checkpoints.py ./checkpoints.db

Inspect specific checkpoint

检查特定检查点

uv run scripts/inspect_checkpoints.py ./checkpoints.db --checkpoint-id abc123 --thread-id thread-1

View full history for a thread

查看线程的完整历史

uv run scripts/inspect_checkpoints.py ./checkpoints.db --thread-id thread-1 --history


`inspect_checkpoints.py` accepts either a direct SQLite DB path or a directory containing `checkpoints.db`.

uv run scripts/inspect_checkpoints.py ./checkpoints.db --thread-id thread-1 --history


`inspect_checkpoints.py`支持直接传入SQLite数据库路径或包含`checkpoints.db`的目录。

Migrate Persisted State

迁移持久化状态

When state shape changes require updating persisted checkpoint values:

bash

undefined

当状态结构变化需要更新已持久化的检查点值时：

bash

undefined

Dry run first

先执行试运行

uv run scripts/migrate_state.py ./checkpoints.db migrations/add_field.py --dry-run

Apply migration

应用迁移

uv run scripts/migrate_state.py ./checkpoints.db migrations/add_field.py


Migration script format:

```python
def migrate(old_state: dict) -> dict:
    new_state = old_state.copy()
    new_state["new_field"] = "default_value"    # Add field
    new_state.pop("deprecated_field", None)      # Remove field
    return new_state

uv run scripts/migrate_state.py ./checkpoints.db migrations/add_field.py


迁移脚本格式：

```python
def migrate(old_state: dict) -> dict:
    new_state = old_state.copy()
    new_state["new_field"] = "default_value"    # 添加字段
    new_state.pop("deprecated_field", None)      # 删除字段
    return new_state

Common State Issues

常见状态问题

Symptom	Likely Cause	Fix
State not updating	Missing reducer	Add `Annotated[type, reducer]`
Messages overwritten	No `add_messages` reducer	Use `MessagesState` (or `Annotated[list[BaseMessage], add_messages]` )
Duplicate entries	Reducer appends without dedup	Use dedup reducer from references/reducers.md
State grows unbounded	No cleanup	Use `RemoveMessage` or trim strategy
Agent state schema rejected	Non-TypedDict `state_schema` in `create_agent`	Use a `TypedDict` agent state schema
Parallel update conflict	Multiple `Overwrite` on same key	Only one node per super-step can use `Overwrite`

For detailed debugging techniques, LangSmith tracing, and checkpoint inspection patterns, see references/state-debugging.md.

症状	可能原因	修复方案
状态未更新	缺少Reducer	添加 `Annotated[type, reducer]`
消息被覆盖	无 `add_messages` Reducer	使用 `MessagesState` （或 `Annotated[list[BaseMessage], add_messages]` ）
重复条目	Reducer追加时未去重	使用references/reducers.md中的去重Reducer
状态无限增长	无清理机制	使用 `RemoveMessage` 或截断策略
Agent状态模式被拒绝	`create_agent` 中的 `state_schema` 非TypedDict	使用 `TypedDict` 作为Agent状态模式
并行更新冲突	同一超级步骤中有多个 `Overwrite` 操作同一键	每个超级步骤仅允许一个节点使用 `Overwrite`

详细调试技巧、LangSmith追踪和检查点检查模式请参考references/state-debugging.md。

Resources

资源

Scripts

脚本

Script	Purpose
`scripts/validate_state_schema.py`	Validate schema structure and typing
`scripts/test_reducers.py`	Test reducer functions
`scripts/inspect_checkpoints.py`	Inspect checkpoint data
`scripts/migrate_state.py`	Migrate checkpoint state values

脚本	用途
`scripts/validate_state_schema.py`	验证模式结构和类型
`scripts/test_reducers.py`	测试Reducer函数
`scripts/inspect_checkpoints.py`	检查检查点数据
`scripts/migrate_state.py`	迁移检查点状态值

References

参考文档

File	Content
references/schema-patterns.md	Schema examples for chat, research, workflow, RAG, tool-calling
references/reducers.md	Reducer patterns, Overwrite, custom reducers, testing
references/persistence-backends.md	Backend setup, thread management, migration
references/state-typing.md	TypedDict, Pydantic, Zod, validation strategies
references/state-debugging.md	Debugging techniques, LangSmith tracing, common issues

文件	内容
references/schema-patterns.md	聊天、研究、工作流、RAG、工具调用的模式示例
references/reducers.md	Reducer模式、Overwrite、自定义Reducer、测试
references/persistence-backends.md	后端设置、线程管理、迁移
references/state-typing.md	TypedDict、Pydantic、Zod、验证策略
references/state-debugging.md	调试技巧、LangSmith追踪、常见问题

State Templates

状态模板

File	Pattern
`assets/chat_state.py`	Chat with `MessagesState`
`assets/research_state.py`	Research with custom reducers
`assets/workflow_state.py`	Workflow with Literal status
`assets/tool_calling_state.py`	Tool-calling agent with `MessagesState`

文件	模式
`assets/chat_state.py`	基于 `MessagesState` 的聊天应用
`assets/research_state.py`	带自定义Reducer的研究Agent
`assets/workflow_state.py`	带Literal状态的工作流
`assets/tool_calling_state.py`	基于 `MessagesState` 的工具调用Agent