State в LangGraph:
TypedDict как схема состояния агента

Зачем состоянию нужна схема?

Узлы в LangGraph не вызывают друг друга и не передают аргументы напрямую. Единственный канал связи между ними — общее состояние. Узел читает то, что положили предыдущие, и дописывает своё. По сути это «доска объявлений»: каждый шаг подходит, читает доску, что-то на ней меняет и уходит.

Если эту доску оставить бесформенной (просто dict с произвольными ключами), очень быстро начинается хаос: один узел пишет "result", другой ждёт "results", третий кладёт строку туда, где ожидался список. Ошибки всплывают не в момент опечатки, а спустя несколько шагов — и их тяжело искать.

Поэтому LangGraph требует объявить схему состояния заранее: перечислить все поля и их типы. Схема выполняет три задачи:

• Документирует агента — по схеме сразу видно, какими данными он оперирует.
• Задаёт контракт — какие ключи существуют и какого они типа; IDE подсказывает поля, а не угадывает.
• Управляет обновлением — именно к схеме привязываются правила слияния (reducers, тема следующего урока).

TypedDict: схема по умолчанию

Стандартный способ описать состояние — TypedDict из модуля typing. Это «словарь с объявленными полями»: на этапе исполнения он остаётся обычным dict (быстрым и сериализуемым), но для типов и IDE выглядит как структура с известными ключами.

from typing import TypedDict

class AgentState(TypedDict):
    query: str            # исходный вопрос пользователя
    draft: str            # черновик ответа
    steps: int            # счётчик шагов

# Это просто словарь — но с известной формой
initial: AgentState = {"query": "Что такое LangGraph?", "draft": "", "steps": 0}

# Передаём схему в граф — он будет проверять, что узлы пишут известные ключи
from langgraph.graph import StateGraph
builder = StateGraph(AgentState)

Почему именно TypedDict, а не обычный класс? Состояние постоянно копируется между шагами и сериализуется (для checkpointing). TypedDict — это под капотом dict: его легко сохранить в JSON, сравнить и склонировать. Узлы читают поля как state["query"] и возвращают изменения тоже словарём — никаких лишних обёрток.

Частичное обновление: как узел меняет доску

Самое важное и неинтуитивное правило: узел возвращает не всё состояние, а только изменённые ключи. LangGraph берёт этот частичный словарь и вливает его в общее состояние. Ключи, которых в возврате нет, остаются нетронутыми.

Почему так? Чтобы узлы были независимыми. Если бы каждый узел был обязан вернуть состояние целиком, ему пришлось бы знать про все поля и аккуратно копировать чужие данные. А так каждый отвечает только за свой кусочек, и добавить новое поле в схему можно, не трогая существующие узлы.

def draft_node(state: AgentState) -> dict:
    # читаем нужное поле
    query = state["query"]
    # возвращаем ТОЛЬКО изменённые ключи — query не упоминаем
    return {"draft": f"Черновик ответа на: {query}", "steps": state["steps"] + 1}

# До:    {"query": "вопрос", "draft": "",         "steps": 0}
# После: {"query": "вопрос", "draft": "Черновик...", "steps": 1}
#         ↑ query сохранился сам собой

По умолчанию правило слияния простое: возвращённое значение перезаписывает старое для этого ключа («последний записавший побеждает»). Этого достаточно для большинства полей — статусов, счётчиков, текущего текста. Но иногда поле нужно не перезаписывать, а накапливать (например, историю сообщений). Как поменять правило слияния — тема следующего урока про reducers.

Необязательные поля и значения по умолчанию

На старте графа заполнены не все поля: query приходит от пользователя, а draft появится только после первого узла. Если объявить поле в TypedDict, формально оно считается обязательным — и обращение state["draft"] до того, как его кто-то записал, бросит KeyError.

Есть два аккуратных подхода.

from typing import TypedDict

# Способ 1: total=False — все поля необязательны при создании
class AgentState(TypedDict, total=False):
    query: str
    draft: str
    steps: int

# Тогда стартовать можно с неполного словаря:
graph.invoke({"query": "вопрос"})     # draft и steps появятся по ходу

# Способ 2: читать через .get() с дефолтом — безопасно, если ключа ещё нет
def node(state: AgentState) -> dict:
    steps = state.get("steps", 0)     # 0, если шаг ещё не писал steps
    return {"steps": steps + 1}

Практическое правило: поля, которые точно есть на входе, держи обязательными и заполняй в стартовом словаре; поля, которые появляются по ходу, либо клади в total=False, либо инициализируй явными дефолтами при запуске ({"query": q, "draft": "", "steps": 0}). Второе — надёжнее: меньше сюрпризов с отсутствующими ключами.

Как проектировать состояние агента

Хорошая схема состояния читается как описание задачи. Удобно мысленно делить поля на три группы:

from typing import TypedDict

class RAGState(TypedDict, total=False):
    # --- входные ---
    question: str            # вопрос пользователя
    top_k: int               # сколько документов доставать

    # --- рабочие ---
    documents: list[str]     # найденный контекст
    attempts: int            # сколько раз переформулировали запрос
    enough_context: bool     # хватает ли данных для ответа

    # --- выходные ---
    answer: str              # финальный ответ
    sources: list[str]       # ссылки на источники

Держи состояние плоским и явным: лучше пять понятных полей, чем один словарь "data" со всем подряд внутри. Плоское состояние проще читать в стриминге, проще сохранять и проще объяснять reducer'у, как сливать каждое поле.

Альтернативы: dataclass и доступ через атрибуты

TypedDict — путь по умолчанию, но не единственный. LangGraph принимает в качестве схемы и dataclass. Разница в основном в синтаксисе: к полям обращаются через точку (state.query), а не по ключу (state["query"]), и легко задать значения по умолчанию.

from dataclasses import dataclass, field

@dataclass
class AgentState:
    query: str
    draft: str = ""             # дефолты прямо в схеме
    steps: int = 0
    documents: list = field(default_factory=list)

def node(state: AgentState) -> dict:
    # читаем через атрибут
    return {"draft": f"Черновик: {state.query}", "steps": state.steps + 1}

Узел по-прежнему возвращает словарь с изменениями — это не меняется. Меняется только то, как ты читаешь состояние внутри. dataclass удобен, когда хочется дефолтов и привычного доступа через точку, но он, как и TypedDict, не проверяет типы значений в рантайме.

Когда нужна валидация: Pydantic

Если важно, чтобы некорректные данные падали сразу при записи, а не отравляли состояние, — берут Pydantic-модель. Она проверяет типы и ограничения каждый раз, когда поле обновляется: вернул узел строку вместо числа — получишь понятную ошибку валидации прямо на этом шаге.

from pydantic import BaseModel, Field

class AgentState(BaseModel):
    query: str
    draft: str = ""
    steps: int = Field(default=0, ge=0)      # steps не может быть отрицательным
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)

# Граф строится так же
builder = StateGraph(AgentState)

# Если узел вернёт {"steps": -1} — Pydantic бросит ValidationError
# прямо на этом шаге, а не «когда-нибудь потом»

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

Шпаргалка

from typing import TypedDict

# 1. Схема по умолчанию — TypedDict
class State(TypedDict):
    query: str          # обязательное поле
    draft: str
    steps: int

# 2. Необязательные поля
class State(TypedDict, total=False):
    draft: str          # можно не передавать на старте

# 3. Узел читает поля и ВОЗВРАЩАЕТ только изменённые ключи
def node(state: State) -> dict:
    return {"steps": state.get("steps", 0) + 1}   # .get() — безопасно

# 4. Правило слияния по умолчанию — перезапись ключа
#    (накопление вместо перезаписи — reducers, следующий урок)

# 5. Альтернативы:
#    dataclass  -> доступ через точку (state.query), дефолты
#    Pydantic   -> + валидация типов/ограничений в рантайме

# 6. НИКОГДА не мутируй state на месте:
#    state["items"].append(x)   # ❌
#    return {"items": [x]}      # ✅ возвращай изменения

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

Спроектируй состояние для конкретного агента:

Что дальше