LangGraph reducers: Annotated и operator.add

Проблема: перезапись съедает историю

Представим агента, который ведёт лог своих действий в поле logs. Каждый узел дописывает строчку. С обычной схемой (без аннотаций) выходит вот что:

Без reducer — каждый узел затирает список целиком

python

from typing import TypedDict

class State(TypedDict):
    logs: list[str]

def node_a(state: State) -> dict:
    return {"logs": ["шаг A"]}     # вернули список из одного элемента

def node_b(state: State) -> dict:
    return {"logs": ["шаг B"]}

# Поток: START → node_a → node_b → END
# Ожидали:  logs == ["шаг A", "шаг B"]
# Получили: logs == ["шаг B"]      ← "шаг A" затёрт!

Почему так? Правило слияния по умолчанию — «последний записавший побеждает»: возврат node_b целиком заменил то, что положил node_a. Для скаляров (статус, счётчик) это нормально, но для списков и историй — катастрофа.

«Костыльный» вариант — заставить каждый узел самому читать старый список и возвращать склейку: return {"logs": state["logs"] + ["шаг B"]}. Но тогда каждый узел обязан знать про устройство поля, помнить про склейку и не ошибиться — а при параллельных ветках (увидим ниже) этот приём вообще не работает. Нужен системный механизм. Это и есть reducer.

Что такое reducer

Reducer — это функция двух аргументов: текущее значение поля и обновление, которое вернул узел. Она возвращает новое значение поля. LangGraph вызывает reducer каждый раз, когда какой-то узел пишет в это поле.

100%

колёсико — масштаб · зажать и тянуть — перемещение

Иными словами, reducer по умолчанию — это lambda old, update: update (вернуть обновление, забыв старое). Чтобы накапливать, нам нужен другой reducer — например, «склеить два списка». И тут вступает Annotated.

Annotated: привязываем reducer к полю

Annotated из модуля typing позволяет прикрепить к типу метаданные, не меняя сам тип. LangGraph использует этот механизм так: Annotated[тип_поля, reducer_функция]. Первый элемент — обычный тип (для IDE и mypy), второй — функция слияния, которую граф будет применять к этому полю.

Annotated[тип, reducer] — синтаксис

python

from typing import Annotated, TypedDict
import operator

class State(TypedDict):
    answer: str                                  # reducer по умолчанию — перезапись
    logs: Annotated[list[str], operator.add]     # reducer — склейка списков
    #     └── тип поля ──┘  └── функция слияния ──┘

# Читается так: "logs — это list[str], а при обновлении
#               сливай старое и новое через operator.add"

Важно: Annotated для самого Python — «прозрачен». logs остаётся обычным списком, доступ state["logs"] работает как раньше. Метаданные читает только LangGraph при компиляции графа — и берёт оттуда reducer.

operator.add — самый частый reducer

operator.add — это просто функция-обёртка над оператором +: operator.add(a, b) возвращает a + b. Для списков + означает конкатенацию, поэтому именно его берут для накопления.

operator.add чинит наш пример с логами

python

from typing import Annotated, TypedDict
import operator

class State(TypedDict):
    logs: Annotated[list[str], operator.add]    # ← теперь накапливается

def node_a(state: State) -> dict:
    return {"logs": ["шаг A"]}

def node_b(state: State) -> dict:
    return {"logs": ["шаг B"]}

# Поток: START → node_a → node_b → END
# Теперь logs == ["шаг A", "шаг B"]   ✅ ничего не потеряли

# Под капотом граф делает:
#   logs = operator.add([],         ["шаг A"])  →  ["шаг A"]
#   logs = operator.add(["шаг A"],  ["шаг B"])  →  ["шаг A", "шаг B"]

Заметь: узлы вернули списки из одного элемента, а не всю историю. Их склейкой занимается reducer. Это и есть смысл: узел отвечает только за «что я добавил», а «как это накопить» решает поле.

⚠️ Возвращай список, а не элемент

Раз reducer склеивает через +, узел должен вернуть список ({"logs": ["шаг B"]}), а не голую строку ({"logs": "шаг B"}). Иначе operator.add(["шаг A"], "шаг B") упадёт с TypeError: список и строку не сложить.

operator.add работает не только со списками. Для int он складывает числа (удобно для накапливающего счётчика токенов или стоимости), для строк — конкатенирует. Главное, чтобы у типа был осмысленный оператор +.

add_messages — reducer для диалога

История сообщений — самый частый накапливаемый список в агентах. Для неё у LangGraph есть специальный готовый reducer — add_messages. Он умнее простой конкатенации:

Дедуплицирует по id: если сообщение с тем же id приходит повторно (например, при обновлении), оно не задваивается, а заменяется.
Приводит форматы: понимает и dict-сообщения ({"role": ..., "content": ...}), и объекты LangChain (HumanMessage, AIMessage).
Корректно дописывает ответы модели и результаты инструментов в правильном порядке.

add_messages — стандарт для чат-состояния

python

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]    # история копится сама

def agent_node(state: ChatState) -> dict:
    reply = llm.invoke(state["messages"])
    return {"messages": [reply]}     # вернули ОДНО новое сообщение,
                                     # add_messages допишет его к истории

✅ Готовый ярлык: MessagesState

Состояние с единственным полем messages: Annotated[list, add_messages] нужно так часто, что LangGraph поставляет готовый класс: from langgraph.graph import MessagesState. Наследуйся от него и добавляй свои поля — не придётся каждый раз прописывать аннотацию вручную.

Зачем reducers критичны: параллельные ветки

В линейном графе можно было бы обойтись ручной склейкой в узлах. Но как только из одного узла расходятся параллельные ветки (fan-out), которые потом сходятся (fan-in), reducer становится единственным правильным решением.

Представь: узел dispatch запускает три параллельных поиска, и все три пишут в поле results. Они выполняются «одновременно» и не видят результатов друг друга — каждый возвращает свой кусок относительно одного и того же старого состояния. Кто-то должен собрать три обновления в один список. Это и делает reducer на fan-in.

Fan-out / fan-in: три ветки пишут в одно поле

python

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

class State(TypedDict):
    results: Annotated[list[str], operator.add]   # без reducer ветки затрут друг друга

def search_web(state):   return {"results": ["из веба"]}
def search_docs(state):  return {"results": ["из документов"]}
def search_db(state):    return {"results": ["из базы"]}

builder = StateGraph(State)
for name, fn in [("web", search_web), ("docs", search_docs), ("db", search_db)]:
    builder.add_node(name, fn)
    builder.add_edge(START, name)     # все три стартуют параллельно
    builder.add_edge(name, END)

graph = builder.compile()
print(graph.invoke({"results": []})["results"])
# ['из веба', 'из документов', 'из базы']   ← reducer собрал все три ветки
# Без operator.add осталась бы только одна из них (гонка за поле)

❌ Без reducer параллельные записи конфликтуют

Если поле, в которое пишут несколько параллельных веток, не имеет reducer, LangGraph не знает, как объединить их обновления, и бросит ошибку InvalidUpdateError (конфликт записи). Reducer на таком поле обязателен — это не «улучшение», а условие корректности.

Свой reducer

Готовых operator.add и add_messages хватает не всегда. Reducer — это любая функция (old, update) -> new, так что можно написать своё правило: складывать словари, держать только уникальные значения, оставлять максимум и т. п.

Reducer, накапливающий только уникальные значения

python

from typing import Annotated, TypedDict

def merge_unique(old: list[str], new: list[str]) -> list[str]:
    """Добавляет новые элементы, не допуская дубликатов (порядок сохраняется)."""
    seen = set(old)
    return old + [x for x in new if x not in seen]

class State(TypedDict):
    sources: Annotated[list[str], merge_unique]

# Если ветки вернут ["a", "b"] и ["b", "c"],
# в sources окажется ["a", "b", "c"] — без повторной "b"

⚠️ Reducer должен быть чистым

Не мутируй аргументы внутри reducer — не делай old.append(...). Возвращай новый объект (old + [...]). Reducer вызывается часто и в неочевидном порядке (особенно на параллельных ветках); мутация общего списка приведёт к трудноуловимым багам. Также reducer обязан корректно обрабатывать пустое начальное значение.

Какой reducer выбрать

Что нужно полю	Reducer	Пример поля
Хранить текущее значение	— (по умолчанию)	статус, флаг, последний ответ
Копить список / складывать числа	operator.add	logs, найденные документы, счётчик токенов
Копить историю диалога	add_messages	messages
Особое правило слияния	своя функция	уникальные источники, merge словарей, max
Несколько параллельных веток пишут в поле	reducer обязателен	результаты fan-out/fan-in

Типичные ошибки

❌ Ошибка 1: забыли reducer и потеряли историю

Поле messages объявлено как обычный list без Annotated[..., add_messages] — и каждый узел затирает диалог. Симптом: агент «помнит» только последнюю реплику. Лечение: добавить reducer.

❌ Ошибка 2: вернули элемент вместо списка

С operator.add узел должен вернуть {"logs": ["x"]}, а не {"logs": "x"}. Иначе reducer попытается сложить список со строкой и упадёт. Тип в Annotated (list[str]) подсказывает, что возвращать.

❌ Ошибка 3: ручная склейка вместе с reducer

Если у поля уже есть operator.add, не делай в узле ещё и state["logs"] + ["x"]. Reducer применится к твоей склейке повторно — и элементы задвоятся. Правило: есть reducer → возвращай только добавляемое.

❌ Ошибка 4: мутация старого значения в reducer

old.append(new) внутри reducer меняет общий объект состояния «за спиной» графа и ломает checkpointing/стриминг. Всегда возвращай новый объект: return old + new.

Шпаргалка

Reducers — всё в одном месте

python

from typing import Annotated, TypedDict
import operator
from langgraph.graph.message import add_messages

class State(TypedDict):
    # 1. Перезапись (reducer по умолчанию) — для текущих значений
    status: str

    # 2. Накопление списка / суммы чисел
    logs: Annotated[list[str], operator.add]
    total_tokens: Annotated[int, operator.add]

    # 3. История диалога — спец-reducer
    messages: Annotated[list, add_messages]

    # 4. Своё правило слияния
    sources: Annotated[list[str], merge_unique]

# Reducer — это функция (old, update) -> new:
def merge_unique(old, new):
    seen = set(old)
    return old + [x for x in new if x not in seen]

# Правила:
#  • есть reducer → узел возвращает только ДОБАВЛЯЕМОЕ ({"logs": ["x"]})
#  • operator.add требует список (не строку)
#  • параллельные ветки в одно поле → reducer ОБЯЗАТЕЛЕН
#  • reducer чистый: не мутируй old, возвращай новый объект

Практическое задание

Закрепи механику слияния на практике:

Задание: агент-«сборщик фактов»

Опиши State с полями: facts: Annotated[list[str], operator.add], calls: Annotated[int, operator.add] (счётчик обращений) и status: str (перезаписывается).
Сделай три узла-«источника», каждый возвращает {"facts": ["факт N"], "calls": 1}.
Подключи все три как параллельные ветки от START и сведи в END. Запусти и проверь, что в facts три элемента, а calls == 3.
Убери operator.add у facts и посмотри на ошибку InvalidUpdateError — это наглядно показывает, зачем нужен reducer на fan-in.
Со звёздочкой: напиши reducer merge_unique и сделай так, чтобы повторяющиеся факты из разных веток не дублировались.

Что дальше

🗂️

State: TypedDict как схема

Вернуться к устройству состояния

🔀

Conditional edges

Ветвление на основе состояния и ReAct-цикл

💾

Checkpointing

Сохранение состояния между запусками

🧩

Все уроки модуля 03

LangGraph целиком

Аннотации состояния:Annotated и operator.add

Проблема: перезапись съедает историю

Что такое reducer

Annotated: привязываем reducer к полю

operator.add — самый частый reducer

add_messages — reducer для диалога

Зачем reducers критичны: параллельные ветки

Свой reducer

Какой reducer выбрать

Типичные ошибки

Шпаргалка

Практическое задание

Задание: агент-«сборщик фактов»

Что дальше

State: TypedDict как схема

Conditional edges

Checkpointing

Все уроки модуля 03

Аннотации состояния:
Annotated и operator.add