Как LangGraph представляет состояние с помощью TypedDict и как оно передаётся между узлами?

TypedDict как схема состояния

LangGraph использует TypedDict из модуля typing для определения структуры состояния графа. Это даёт статическую типизацию и явный контракт между узлами — каждый разработчик видит, какие поля существуют в состоянии.

from typing import TypedDict, Optional

class MyState(TypedDict):
    question: str           # Обязательное поле
    context: list[str]      # Список строк
    answer: Optional[str]   # Может быть None
    step_count: int         # Счётчик итераций

Как узлы получают и обновляют состояние

Узел — это обычная функция (или async-функция), которая принимает state: MyState и возвращает dict с полями для обновления. LangGraph делает мёрж: поля из возвращённого словаря перезаписывают соответствующие поля в текущем состоянии, остальные поля остаются неизменными.

from langgraph.graph import StateGraph, END

def step_retrieve(state: MyState) -> dict:
    # Получаем question из состояния
    question = state["question"]
    # Делаем поиск
    results = ["Document 1 about " + question, "Document 2 about " + question]
    # Возвращаем ТОЛЬКО изменённые поля
    return {"context": results}

def step_generate(state: MyState) -> dict:
    context = "\n".join(state["context"])
    question = state["question"]
    # Генерируем ответ
    answer = f"Based on context: {context[:100]}... Answer to: {question}"
    return {
        "answer": answer,
        "step_count": state["step_count"] + 1,
    }

# Узел не обязан возвращать все поля — только изменённые
def step_log(state: MyState) -> dict:
    print(f"Step {state['step_count']}: answer ready = {state['answer'] is not None}")
    return {}  # Пустой dict — состояние не меняется

Сборка графа

graph = StateGraph(MyState)

graph.add_node("retrieve", step_retrieve)
graph.add_node("generate", step_generate)
graph.add_node("log", step_log)

graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", "log")
graph.add_edge("log", END)

app = graph.compile()

# Запуск: передаём начальное состояние
initial_state: MyState = {
    "question": "What is LangGraph?",
    "context": [],
    "answer": None,
    "step_count": 0,
}
result = app.invoke(initial_state)
print(result["answer"])

Аккумулирующие поля через Annotated

По умолчанию каждый return перезаписывает поле. Чтобы поле накапливалось (например, список сообщений), используют Annotated с reducer-функцией:

from typing import Annotated
import operator

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]  # Каждый return += к списку
    final_answer: str

def node_a(state: ChatState) -> dict:
    return {"messages": ["Node A processed"]}  # Добавит в конец списка

def node_b(state: ChatState) -> dict:
    return {"messages": ["Node B processed"]}  # Тоже добавит, не перезапишет

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

• Если узел возвращает поле с типом, несовместимым с TypedDict (например, str вместо list[str]), ошибка возникает в runtime, а не при компиляции.
• Без Annotated[list, operator.add] каждый return полностью перезаписывает список — история сообщений будет содержать только последний элемент.
• TypedDict не поддерживает валидацию значений — если нужна проверка типов в runtime, используйте Pydantic BaseModel вместо TypedDict.
• Изменение схемы TypedDict (добавление/удаление поля) несовместимо с существующими checkpoints — при обновлении нужна миграция или сброс сохранённых состояний.
• Передача изменяемых объектов (списков, словарей) без копирования между узлами может привести к неожиданным мутациям состояния.
• Optional-поля не инициализируются автоматически — начальное состояние должно явно задавать все поля, иначе KeyError в первом узле.

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

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