langgraph-state

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

LangGraph State Management

LangGraph状态管理

Design and manage state schemas for LangGraph workflows.

为LangGraph工作流设计并管理状态架构。

TypedDict Approach (Simple)

TypedDict方案（简易版）

python

from typing import TypedDict, Annotated
from operator import add

class WorkflowState(TypedDict):
    input: str
    output: str
    agent_responses: Annotated[list[dict], add]  # Accumulates
    metadata: dict

python

from typing import TypedDict, Annotated
from operator import add

class WorkflowState(TypedDict):
    input: str
    output: str
    agent_responses: Annotated[list[dict], add]  # Accumulates
    metadata: dict

MessagesState Pattern (2026 Best Practice)

MessagesState模式（2026最佳实践）

python

from langgraph.graph import MessagesState
from langgraph.graph.message import add_messages
from typing import Annotated

python

from langgraph.graph import MessagesState
from langgraph.graph.message import add_messages
from typing import Annotated

Option 1: Use built-in MessagesState (recommended)

class AgentState(MessagesState): """Extends MessagesState with custom fields.""" user_id: str context: dict

Option 2: Define messages manually with add_messages reducer

class CustomState(TypedDict): messages: Annotated[list, add_messages] # Smart append/update by ID metadata: dict


**Why `add_messages` matters:**
- Appends new messages (doesn't overwrite)
- Updates existing messages by ID
- Handles message deduplication automatically

> **Note**: `MessageGraph` is deprecated in LangGraph v1.0.0. Use `StateGraph` with a `messages` key instead.

class CustomState(TypedDict): messages: Annotated[list, add_messages] # Smart append/update by ID metadata: dict


**`add_messages`的重要性：**
- 追加新消息（不会覆盖原有内容）
- 根据ID更新现有消息
- 自动处理消息去重

> **注意**：`MessageGraph`在LangGraph v1.0.0中已被弃用，请改用带有`messages`键的`StateGraph`。

Pydantic Approach (Validation)

Pydantic方案（带验证）

python

from pydantic import BaseModel, Field

class WorkflowState(BaseModel):
    input: str = Field(description="User input")
    output: str = ""
    agent_responses: list[dict] = Field(default_factory=list)

    def add_response(self, agent: str, result: str):
        self.agent_responses.append({"agent": agent, "result": result})

python

from pydantic import BaseModel, Field

class WorkflowState(BaseModel):
    input: str = Field(description="User input")
    output: str = ""
    agent_responses: list[dict] = Field(default_factory=list)

    def add_response(self, agent: str, result: str):
        self.agent_responses.append({"agent": agent, "result": result})

Accumulating State Pattern

状态累加模式

python

from typing import Annotated
from operator import add

class AnalysisState(TypedDict):
    url: str
    raw_content: str

    # Accumulate agent outputs
    findings: Annotated[list[Finding], add]
    embeddings: Annotated[list[Embedding], add]

    # Control flow
    current_agent: str
    agents_completed: list[str]
    quality_passed: bool

Key Pattern:
Annotated[list[T], add]

Without
```
add
```
: Each node replaces the list
With
```
add
```
: Each node appends to the list
Critical for multi-agent workflows

python

from typing import Annotated
from operator import add

class AnalysisState(TypedDict):
    url: str
    raw_content: str

    # Accumulate agent outputs
    findings: Annotated[list[Finding], add]
    embeddings: Annotated[list[Embedding], add]

    # Control flow
    current_agent: str
    agents_completed: list[str]
    quality_passed: bool

核心模式：
Annotated[list[T], add]

不使用
```
add
```
：每个节点会覆盖列表内容
使用
```
add
```
：每个节点会向列表追加内容
这在多Agent工作流中至关重要

Custom Reducers

自定义归约器

python

from typing import Annotated

def merge_dicts(a: dict, b: dict) -> dict:
    """Custom reducer that merges dictionaries."""
    return {**a, **b}

class State(TypedDict):
    config: Annotated[dict, merge_dicts]  # Merges updates

def last_value(a, b):
    """Keep only the latest value."""
    return b

class State(TypedDict):
    status: Annotated[str, last_value]  # Overwrites

python

from typing import Annotated

def merge_dicts(a: dict, b: dict) -> dict:
    """Custom reducer that merges dictionaries."""
    return {**a, **b}

class State(TypedDict):
    config: Annotated[dict, merge_dicts]  # Merges updates

def last_value(a, b):
    """Keep only the latest value."""
    return b

class State(TypedDict):
    status: Annotated[str, last_value]  # Overwrites

State Immutability

状态不可变性

python

def node(state: WorkflowState) -> WorkflowState:
    """Return new state, don't mutate in place."""
    # Wrong: state["output"] = "result"
    # Right:
    return {
        **state,
        "output": "result"
    }

python

def node(state: WorkflowState) -> WorkflowState:
    """Return new state, don't mutate in place."""
    # Wrong: state["output"] = "result"
    # Right:
    return {
        **state,
        "output": "result"
    }

Context Schema (2026 Pattern)

上下文架构（2026模式）

Pass runtime configuration without polluting state:

python

from dataclasses import dataclass
from langgraph.graph import StateGraph

@dataclass
class ContextSchema:
    """Runtime configuration, not persisted in state."""
    llm_provider: str = "anthropic"
    temperature: float = 0.7
    max_retries: int = 3
    debug_mode: bool = False

在不污染状态的前提下传递运行时配置：

python

from dataclasses import dataclass
from langgraph.graph import StateGraph

@dataclass
class ContextSchema:
    """Runtime configuration, not persisted in state."""
    llm_provider: str = "anthropic"
    temperature: float = 0.7
    max_retries: int = 3
    debug_mode: bool = False

Create graph with context schema

graph = StateGraph(WorkflowState, context_schema=ContextSchema)

Access context in nodes

def my_node(state: WorkflowState, context: ContextSchema): if context.llm_provider == "anthropic": response = call_claude(state["input"], context.temperature) else: response = call_openai(state["input"], context.temperature)

if context.debug_mode:
    logger.debug(f"Response: {response}")

return {"output": response}

if context.debug_mode:
    logger.debug(f"Response: {response}")

return {"output": response}

Invoke with context

graph.invoke( {"input": "Hello"}, context={"llm_provider": "openai", "temperature": 0.5} )

undefined

graph.invoke( {"input": "Hello"}, context={"llm_provider": "openai", "temperature": 0.5} )

undefined

Node Caching (2026 Pattern)

节点缓存（2026模式）

Cache expensive node results with TTL:

python

from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy

通过TTL缓存开销较大的节点结果：

python

from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy

Add node with cache policy

builder.add_node( "embed_content", embed_content_node, cache_policy=CachePolicy(ttl=300) # Cache for 5 minutes )

builder.add_node( "llm_call", llm_node, cache_policy=CachePolicy(ttl=60) # Cache for 1 minute )

builder.add_node( "embed_content", embed_content_node, cache_policy=CachePolicy(ttl=300) # Cache for 5 minutes )

builder.add_node( "llm_call", llm_node, cache_policy=CachePolicy(ttl=60) # Cache for 1 minute )

Compile with cache

graph = builder.compile(cache=InMemoryCache())

undefined

graph = builder.compile(cache=InMemoryCache())

undefined

RemainingSteps (Proactive Recursion Handling)

RemainingSteps（主动递归处理）

Check remaining steps to wrap up gracefully:

python

from langgraph.types import RemainingSteps

def agent_node(state: WorkflowState, remaining: RemainingSteps):
    """Proactively handle recursion limit."""
    if remaining.steps < 5:
        # Running low on steps, wrap up
        return {
            "action": "summarize_and_exit",
            "reason": f"Only {remaining.steps} steps remaining"
        }

    # Continue normal processing
    return {"action": "continue"}

检查剩余步骤数，优雅收尾：

python

from langgraph.types import RemainingSteps

def agent_node(state: WorkflowState, remaining: RemainingSteps):
    """Proactively handle recursion limit."""
    if remaining.steps < 5:
        # Running low on steps, wrap up
        return {
            "action": "summarize_and_exit",
            "reason": f"Only {remaining.steps} steps remaining"
        }

    # Continue normal processing
    return {"action": "continue"}

Key Decisions

关键决策

Decision	Recommendation
TypedDict vs Pydantic	TypedDict for internal state, Pydantic at boundaries
Messages state	Use `MessagesState` or `add_messages` reducer
Accumulators	Always use `Annotated[list, add]` for multi-agent
Nesting	Keep state flat (easier debugging)
Immutability	Return new state, don't mutate
Runtime config	Use `context_schema` for non-persistent config
Expensive ops	Use `CachePolicy` to cache node results
Recursion	Use `RemainingSteps` for proactive handling

2026 Guidance: Use TypedDict inside the graph (lightweight, no runtime overhead). Use Pydantic at boundaries (inputs/outputs, user-facing data) for validation.

决策	推荐方案
TypedDict vs Pydantic	内部状态使用TypedDict，边界处使用Pydantic
消息状态	使用 `MessagesState` 或 `add_messages` 归约器
累加器	多Agent场景下始终使用 `Annotated[list, add]`
嵌套	保持状态扁平化（便于调试）
不可变性	返回新状态，不要原地修改
运行时配置	使用 `context_schema` 存储非持久化配置
高开销操作	使用 `CachePolicy` 缓存节点结果
递归	使用 `RemainingSteps` 主动处理

2026年指导方针：在图内部使用TypedDict（轻量、无运行时开销）。在边界处（输入/输出、面向用户的数据）使用Pydantic进行验证。

Common Mistakes

常见错误

Forgetting
```
add
```
reducer (overwrites instead of accumulates)
Mutating state in place (breaks checkpointing)
Deeply nested state (hard to debug)
No type hints (lose IDE support)
Putting runtime config in state (use context_schema instead)
Not caching expensive operations (repeated embedding calls)

忘记使用
```
add
```
归约器（覆盖而非累加）
原地修改状态（破坏检查点功能）
状态嵌套过深（难以调试）
未添加类型提示（失去IDE支持）
将运行时配置放入状态（应使用context_schema）
未缓存高开销操作（重复调用嵌入接口）

Evaluations

评估

See references/evaluations.md for test cases.

查看references/evaluations.md获取测试用例。

Related Skills

Capability Details

能力详情

state-definition

Keywords: StateGraph, TypedDict, state schema, define state Solves:

Define workflow state with TypedDict
Create Pydantic state models
Structure agent state properly

关键词： StateGraph, TypedDict, state schema, define state 解决场景：

使用TypedDict定义工作流状态
创建Pydantic状态模型
合理构建Agent状态

state-channels

Keywords: channel, Annotated, state channel, MessageChannel Solves:

Configure state channels for data flow
Implement message accumulation
Handle channel-based state updates

关键词： channel, Annotated, state channel, MessageChannel 解决场景：

配置状态通道以实现数据流
实现消息累加
处理基于通道的状态更新

state-reducers

Keywords: reducer, add_messages, operator.add, accumulate Solves:

Implement state reducers with Annotated
Accumulate messages across nodes
Handle state merging strategies

关键词： reducer, add_messages, operator.add, accumulate 解决场景：

通过Annotated实现状态归约器
跨节点累加消息
处理状态合并策略

subgraphs

Keywords: subgraph, nested graph, parent state, child graph Solves:

Compose graphs with subgraphs
Pass state between parent and child
Implement modular workflow components

关键词： subgraph, nested graph, parent state, child graph 解决场景：

通过子图组合图结构
在父图与子图之间传递状态
实现模块化工作流组件

state-persistence

Keywords: persist, state persistence, durable state, save state Solves:

Persist state across executions
Implement durable workflows
Handle state serialization

关键词： persist, state persistence, durable state, save state 解决场景：

跨执行会话持久化状态
实现持久化工作流
处理状态序列化

langgraph-state

Original

Translation

LangGraph State Management

LangGraph状态管理

TypedDict Approach (Simple)

TypedDict方案（简易版）

MessagesState Pattern (2026 Best Practice)

MessagesState模式（2026最佳实践）

Option 1: Use built-in MessagesState (recommended)

Option 1: Use built-in MessagesState (recommended)

Option 2: Define messages manually with add_messages reducer

Option 2: Define messages manually with add_messages reducer

Pydantic Approach (Validation)

Pydantic方案（带验证）

Accumulating State Pattern

状态累加模式

Custom Reducers

自定义归约器

State Immutability

状态不可变性

Context Schema (2026 Pattern)

上下文架构（2026模式）

Create graph with context schema

Create graph with context schema

Access context in nodes

Access context in nodes

Invoke with context

Invoke with context

Node Caching (2026 Pattern)

节点缓存（2026模式）

Add node with cache policy

Add node with cache policy

Compile with cache

Compile with cache

RemainingSteps (Proactive Recursion Handling)

RemainingSteps（主动递归处理）

Key Decisions

关键决策

Common Mistakes

常见错误

Evaluations

评估

Related Skills

相关技能

Capability Details

能力详情

state-definition

state-definition

state-channels

state-channels

state-reducers

state-reducers

subgraphs

subgraphs

state-persistence

state-persistence