Что такое state_schema и как определить кастомные редукторы для полей состояния?

Простая схема без редукторов

from typing import TypedDict
from langgraph.graph import StateGraph, END


class SimpleState(TypedDict):
    counter: int
    name: str


def update(state: SimpleState) -> dict:
    return {"counter": state["counter"] + 1}  # полностью заменяет поле counter


builder = StateGraph(SimpleState)
builder.add_node("update", update)
builder.set_entry_point("update")
builder.add_edge("update", END)
graph = builder.compile()
print(graph.invoke({"counter": 0, "name": "test"}))  # {"counter": 1, "name": "test"}

state_schema и редукторы в LangGraph

Каждый граф LangGraph работает с объектом состояния — словарём, структура которого задана схемой. Схема — это TypedDict или Pydantic BaseModel, переданный при создании StateGraph(schema).

Кастомные редукторы через Annotated

По умолчанию каждый возврат из узла заменяет поле. Если нужно накапливать значения (например, добавлять в список), используйте Annotated[тип, функция]:

import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END


class AccState(TypedDict):
    items: Annotated[list[str], operator.add]  # конкатенация списков
    total: Annotated[int, operator.add]         # сложение чисел
    latest: str                                  # простая замена


def node_a(state: AccState) -> dict:
    return {"items": ["a1", "a2"], "total": 10, "latest": "from_a"}


def node_b(state: AccState) -> dict:
    return {"items": ["b1"], "total": 5, "latest": "from_b"}


builder = StateGraph(AccState)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.set_entry_point("a")
builder.add_edge("a", "b")
builder.add_edge("b", END)
graph = builder.compile()

result = graph.invoke({"items": [], "total": 0, "latest": ""})
# items = ["a1", "a2", "b1"], total = 15, latest = "from_b"
print(result)

Кастомная функция-редуктор

def keep_unique(existing: list, new: list) -> list:
    """Добавляем только уникальные элементы."""
    return existing + [x for x in new if x not in existing]


class UniqueState(TypedDict):
    tags: Annotated[list[str], keep_unique]

Pydantic-модель как схема

from pydantic import BaseModel
from typing import Annotated
import operator


class PydanticState(BaseModel):
    messages: Annotated[list[str], operator.add] = []
    score: float = 0.0


# StateGraph(PydanticState) — Pydantic валидирует типы при каждом обновлении

add_messages — встроенный умный редуктор

from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, RemoveMessage
from typing import Annotated, TypedDict


class ChatState(TypedDict):
    messages: Annotated[list, add_messages]


# add_messages:
# 1. Добавляет новые сообщения в список
# 2. Обновляет существующие (по id)
# 3. Удаляет через RemoveMessage(id=...)

Подводные камни

• Мутация вместо возврата нового объекта — если редуктор изменяет existing in-place и возвращает его, возникают трудноотлаживаемые side-эффекты.
• operator.add для int в параллельных ветвях — LangGraph вызовет редуктор дважды; убедитесь, что сложение коммутативно и идемпотентно.
• Pydantic и Annotated — в Pydantic v2 синтаксис Annotated поддержан, но Field(default_factory=...) может конфликтовать с редуктором.
• Отсутствие поля в возврате узла — если узел не возвращает поле, оно не трогается редуктором; это нормальное поведение, но часто путает новичков.
• None-значения — редуктор вызывается даже если узел вернул {"items": None}; защищайте редуктор от None.
• Сериализация checkpointer — кастомные объекты в состоянии должны быть JSON-сериализуемы для SqliteSaver/RedisSaver.

• Объяснять state schema reducers только синтаксисом без shape, dtype, состояния или режима выполнения.
• Игнорировать leakage, воспроизводимость, пустые входы и скрытые копии данных.
• Не проверять production-симптомы: latency, память, ретраи, дрейф качества и несовпадение версий.

• Может ли связать state schema reducers с реальным контрактом входов и выходов.
• Упоминает ли тесты, метрики, reproducibility и диагностику ошибок.
• Видит ли различие между demo-кодом в ноутбуке и production-пайплайном.