LangChain и LangGraph — агенты, графы и вся инфраструктура вокруг них

Привет!

Языковая модель умеет ровно одно: получить текст и вернуть текст. Больше ничего. Она не помнит, о чём вы говорили минуту назад, и не может сама сходить в базу данных или в интернет.

Возьмём простой пример: пользователь спрашивает у ассистента, какая сейчас погода. Модель этого не знает. Чтобы она смогла ответить, нужно проделать вот что:

1. Заранее рассказать модели, что у неё есть функция «узнать погоду», и описать, какие у неё аргументы.
2. Получить ответ, в котором модель просит эту функцию вызвать, и разобрать его.
3. Вызвать функцию своим кодом.
4. Отправить модели результат и попросить ответить ещё раз.

Получился цикл, а не один запрос. Добавьте к нему память диалога, чтобы следующий вопрос понимался в контексте предыдущего; вывод по мере генерации, чтобы пользователь не смотрел сорок секунд в пустой экран; повтор при сбое провайдера; запись того, что модель видела, — иначе неправильный ответ невозможно разобрать. Кода вокруг простого действия набирается прилично.

LangChain и LangGraph — библиотеки, в которых всё это уже написано. Первая даёт готовые части: модели, tools, промпты, агентный цикл. Вторая выполняет процесс: хранит state, ветвит шаги, сохраняет прогресс, отдаёт поток событий.

В статье — как этим пользоваться: что за что отвечает, из чего собирается agent, когда вместо агента нужен graph, как работают память и streaming, зачем нужны LangSmith и готовый сервер запуска и что изменилось при переходе с прошлой мажорной версии.

Примеры на Python. Версии пакетов на момент написания: langchain 1.3.14, langchain-core 1.5.3, langgraph 1.2.10, langchain-classic 1.0.8. Минимальный Python — 3.10, для JavaScript-версии — Node 20.

Словарь терминов

Термины дальше идут в оригинальном написании — так же, как в документации и в коде, чтобы одно было легко сопоставить с другим. Вот короткие определения; каждое подробно разбирается в своём разделе.

Model — языковая модель провайдера: OpenAI, Anthropic, Google, локальная через Ollama. В коде это объект с методами вызова и потокового вызова.

Message — единица диалога. Четыре типа: SystemMessage (инструкции), HumanMessage (ввод пользователя), AIMessage (ответ модели вместе с запросами на вызовы), ToolMessage (результат выполнения функции).

Tool — функция вашего кода, описание которой отдаётся модели. Модель её не выполняет: она возвращает запрос «вызови вот это с такими аргументами», а выполняет ваш код.

Agent — цикл: вызвать model, выполнить tools, которые она запросила, вернуть ей результаты, повторить. Цикл заканчивается, когда модель отвечает без запроса tools.

Graph — описание процесса шагами и переходами. Нужен, когда порядок действий определяете вы, а не модель.

Node — один шаг графа, обычная функция. Edge — переход между шагами.

State — данные, которые nodes читают и дополняют по ходу работы.

Reducer — правило, по которому обновление от node сливается с тем, что уже лежит в state.

Super-step — один такт исполнения графа: выполняется всё запланированное, затем обновления попадают в state, затем планируется следующий такт.

Checkpointer — механизм сохранения state. Благодаря ему второй ход диалога помнит первый, а упавший процесс можно продолжить с места остановки.

Thread — идентификатор диалога, под которым сохраняется state. Один пользователь, один разговор — один thread.

Store — долговременная память поверх всех threads: предпочтения, накопленные факты.

Context — неизменяемые данные запуска: кто пользователь, какой арендатор, какие флаги.

Runtime — объект, через который node или tool получают доступ к context, store и потоку событий.

Middleware — код, выполняемый вокруг шагов агентного цикла: до вызова модели, после, вокруг вызова tool.

Interrupt — пауза запуска, чтобы дождаться человека.

Streaming — потоковая выдача: токены и события приходят по мере появления.

Из чего состоит экосистема

Под одним брендом выпускается несколько продуктов с похожими названиями. Разложим по полкам.

Слои зависят снизу вверх:

Функция create_agent из LangChain возвращает скомпилированный graph LangGraph — то есть всё, что рантайм умеет делать с графом, он умеет и с агентом: сохранять state, стримить, ставить на паузу, вкладывать внутрь другого графа как обычный node. Выбирать между agent и graph не нужно, это одно и то же на разной высоте абстракции.

Пакеты

Библиотека разложена на много пакетов; ставить нужно не всё.

Провайдеры моделей

Сама библиотека ни к одному вендору не привязана: провайдер подключается отдельным пакетом, и переключение между ними — это смена строки инициализации, а не переписывание кода. Пакеты версионируются независимо от ядра.

Строка вида "openai:gpt-5.5" только выбирает уже установленную интеграцию — пакет вендора должен стоять в зависимостях.

Помимо моделей отдельными пакетами подключаются и остальные внешние сервисы: векторные хранилища (langchain-chroma, langchain-postgres для pgvector, langchain-pinecone, langchain-qdrant, langchain-weaviate), веб-поиск (langchain-tavily), а всё, у чего нет своего пакета, живёт в langchain-community.

Версионирование

Про версионирование стоит знать одну вещь. Основные пакеты — langchain, langchain-core, langgraph — следуют семантическому версионированию: ломающие изменения бывают только на смене мажорной версии, а устаревшие возможности продолжают работать с предупреждением на всей линейке. Отсюда рабочая нижняя граница >=1.0,<2.0. А langchain-community семантическому версионированию не следует, и его обычно закрепляют на конкретной минорной серии.

Какой слой брать под задачу

Проверять сверху вниз и останавливаться на первом совпадении:

Последний пункт применим чаще, чем кажется: агентный цикл вокруг одного вызова добавляет и токены, и задержку.

Установка и первый вызов

pip install "langchain>=1.0,<2.0" "langchain-core>=1.0,<2.0" "langgraph>=1.0,<2.0" langchain-openai

langchain-core ставится явно. Он приедет и транзитивно, но тогда его версией управляет не ваш файл зависимостей.

Пакет провайдера обязателен: строка вида "openai:gpt-5.5" разрешается в конкретную интеграцию, но сама её не устанавливает — это поиск среди уже установленного.

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-5.5", temperature=0, timeout=30)

response = await model.ainvoke("Почему у попугаев яркие перья?")
print(response.text)

Формат строки — provider:model. Примеры из документации: openai:gpt-5.5, anthropic:claude-sonnet-4-6, google_genai:gemini-2.5-flash-lite. Вместо строки можно создать объект провайдера напрямую — так открывается доступ к параметрам, которых нет в общем интерфейсе.

response.text — свойство, а не метод. В прошлой версии это был вызов со скобками.

Синхронный и асинхронный вызовы

У каждого метода две формы: invoke и ainvoke, stream и astream. Для скриптов и ноутбуков подходит любая. Для веб-сервиса разница принципиальная: синхронный вызов блокирует цикл событий целиком, и пока он ждёт ответа модели, все параллельные запросы стоят. В сервисе на пути запроса используются только асинхронные формы.

Есть ещё abatch — параллельная обработка нескольких независимых промптов. Она полезна для офлайн-задач вроде прогона датасета, а не для одного пользовательского хода.

Параметры model

Про rate_limiter есть нюанс: встроенная реализация InMemoryRateLimiter работает в пределах одного процесса. При пяти репликах сервиса фактическая частота будет впятеро больше заданной, и для общего на кластер лимита нужен общий ограничитель.

Про reasoning_effort: на рассуждающих моделях глубина размышления задаётся этим параметром. Инструкция «думай шаг за шагом» в промпте работает с собственным механизмом рассуждения модели параллельно, а не дополняет его. Тема промптов большая, у меня про неё есть отдельная статья.

Fallbacks — запасные модели

model = init_chat_model("openai:gpt-5.5").with_fallbacks([init_chat_model("openai:gpt-5-nano")])

Переключение срабатывает при исключении от основной модели. Это способ пережить сбой провайдера, не отдавая пользователю ошибку.

Messages: что приходит в ответе

Типов messages четыре: SystemMessage несёт инструкции, HumanMessage — ввод пользователя, AIMessage — ответ модели вместе с запросами на вызовы tools и метаданными, ToolMessage — результат вызова, уходящий обратно в модель.

У ответа модели есть два представления одних и тех же данных:

• content — сырая нагрузка провайдера. Строка или его собственный список словарей, сохранённый как есть.
• content_blocks — типизированный разбор поверх того же самого, одинаковый для всех провайдеров.

Зачем второе. Провайдеры возвращают одинаковые по смыслу вещи в разной форме: у одного рассуждение приезжает под именем thinking, у другого — reasoning. В content_blocks это в обоих случаях блок типа reasoning.

for block in response.content_blocks:
    if block["type"] == "reasoning":
        log_reasoning(block)
    elif block["type"] == "text":
        show_to_user(block)
    elif block["type"] == "tool_call":
        schedule(block)

Типы блоков покрывают текст и рассуждение; картинки, аудио, видео и файлы; вызовы tools и их фрагменты при streaming; вызовы tools, выполняемых на стороне провайдера; и отдельный тип non_standard для всего, что в стандарт не уложилось.

Практическое следствие: код, читающий content_blocks, переживает смену провайдера. Код, разбирающий content вручную, привязан к конкретному вендору.

Мультимодальный ввод

from langchain.messages import HumanMessage

message = HumanMessage(content=[
    {"type": "text", "text": "Что на этой картинке?"},
    {"type": "image", "url": "https://example.com/photo.png"},
])

Обрезка истории

Диалог растёт, контекстное окно конечно. Простой способ — детерминированная обрезка:

from langchain.messages import trim_messages

trimmed = trim_messages(messages, max_tokens=8000, strategy="last", token_counter=model)

Более умный способ — суммаризация старой части истории, для agents она доступна готовым middleware. Разница в том, что обрезка предсказуема и теряет старое целиком, а суммаризация сохраняет смысл, но переписывает то, что ассистент помнит.

Учёт токенов

from langchain_core.callbacks import UsageMetadataCallbackHandler

callback = UsageMetadataCallbackHandler()
await model.ainvoke("Привет", config={"callbacks": [callback]})
callback.usage_metadata      # {'input_tokens': 8, 'output_tokens': 10, ...}

То же самое доступно у каждого ответа в поле usage_metadata. Это основа для подсчёта расходов и пользовательских квот.

Промпты

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты помогаешь разобраться в логах сборки. Отвечай коротко."),
    ("human", "{question}"),
])

chain = prompt | model

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

Вертикальная черта — композиция: результат левого объекта передаётся правому. Получившаяся цепочка сама умеет ainvoke, astream и abatch и может быть частью чего-то большего.

Structured output — ответ по схеме

Часто от модели нужен не текст, а данные: категория обращения, извлечённые поля, решение о маршруте.

from pydantic import BaseModel, Field

class BuildFailure(BaseModel):
    module: str
    reason: str
    is_flaky: bool = Field(description="Похоже ли на плавающий тест")

structured = model.with_structured_output(BuildFailure)
result = await structured.ainvoke(log_text)   # -> BuildFailure(...)

Принимаются Pydantic-модели, TypedDict и голая JSON-схема. Каким способом это реализуется под капотом — нативный режим провайдера, вызов функции или JSON-режим — выбирается по возможностям конкретной модели.

Подход «попросить JSON в промпте и разобрать прозу» в первой версии убрали как ненадёжный, и обратно его вводить руками не стоит: разобранная схема даёт ошибку сразу, а разбор прозы ломается тихо и проявляется дальше по коду.

Tools

Tool — функция вашего кода, описание которой уходит модели вместе с промптом. Модель сама её не выполняет: она возвращает запрос на вызов с аргументами, вызов выполняет ваш код, результат уходит обратно в модель.

from langchain.tools import tool

@tool
def search_listings(city: str, max_price: int = 1_000_000) -> str:
    """Поиск активных объявлений о продаже недвижимости в городе.

    Args:
        city: Название города, например «Одесса»
        max_price: Верхняя граница цены в долларах
    """
    return render(query(city, max_price))

Три элемента здесь несущие:

Аннотации типов формируют схему, которую увидит модель. Без них у аргумента нет типа, и значение придётся угадывать.

Docstring — это часть промпта. Он единственный объясняет модели, в каких случаях tool применим. От его формулировки напрямую зависит, будет ли tool вызываться вовремя.

Имя в snake_case. Часть провайдеров другие формы отклоняет.

Когда аннотаций недостаточно — нужен перечислимый набор значений, описание на каждое поле или валидация — задаётся явная схема аргументов:

from pydantic import BaseModel, Field
from typing import Literal

class WeatherInput(BaseModel):
    location: str = Field(description="Город или координаты")
    units: Literal["celsius", "fahrenheit"] = "celsius"

@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius") -> str:
    """Текущая погода."""
    ...

ToolRuntime — доступ к окружению

Tool часто нужны данные, которых нет в аргументах: кто сейчас пользователь, какое state у диалога, куда писать долговременную память. Для этого в сигнатуру добавляется параметр ToolRuntime — фреймворк подставит его сам, и в схему для модели он не попадёт.

from langchain.tools import tool, ToolRuntime

@tool
def get_account_info(runtime: ToolRuntime[UserContext]) -> str:
    """Возвращает информацию по счёту текущего пользователя."""
    user_id = runtime.context.user_id
    return describe(lookup(user_id))

Через него доступны: runtime.state — текущее state агента вместе с историей messages; runtime.context — неизменяемый context запуска; runtime.store — долговременная память; runtime.stream_writer — писатель в поток событий; runtime.tool_call_id — идентификатор текущего вызова; runtime.execution_info — идентификаторы thread и запуска; плюс конфигурация запуска.

Момент, важный для безопасности: личность пользователя, идентификатор арендатора и ключи берутся из runtime.context, а не из аргументов. Аргументы заполняет модель, а на модель можно повлиять текстом запроса — идентификатор пользователя в схеме tool означает, что подставить туда чужой можно, просто попросив об этом в чате.

Что tool возвращает

from langchain.messages import ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.types import Command

@tool
def set_language(language: str, runtime: ToolRuntime) -> Command:
    """Задаёт язык, на котором отвечать дальше."""
    return Command(update={
        "preferred_language": language,
        "messages": [ToolMessage(content=f"Язык переключён на {language}.",
                                 tool_call_id=runtime.tool_call_id)],
    })

Когда tool возвращает Command, ToolMessage нужно включить в обновление самостоятельно. Без него провайдер увидит вызов tool без результата и на следующем ходу его отклонит.

Флаг return_direct завершает цикл и отдаёт результат tool как финальный ответ. Он подходит, когда вывод tool и есть ответ, и не подходит, когда модель должна его прокомментировать или объединить с другими данными.

Прогресс из tool

Долгий tool может сообщать о ходе работы — это попадёт в поток событий:

@tool
def index_documents(folder: str, runtime: ToolRuntime) -> str:
    """Индексирует папку с документами."""
    writer = runtime.stream_writer
    writer({"type": "progress", "done": 0, "total": 100})
    ...

Ошибки tools

По умолчанию исключение из tool прекращает весь запуск. Для части случаев это не то, что нужно: неудачный поиск — информация, с которой модель может работать дальше.

from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage

@wrap_tool_call
def handle_tool_errors(request, handler):
    try:
        return handler(request)
    except Exception as exc:
        return ToolMessage(content=f"Инструмент не сработал: {exc}",
                           tool_call_id=request.tool_call["id"])

Есть и готовые middleware для этого — ToolErrorMiddleware превращает исключение в message, ToolRetryMiddleware повторяет с задержкой. Внутри собранного вручную графа ту же роль играет флаг handle_tool_errors у ToolNode.

Случаи стоит разделять, потому что лечатся они по-разному: временный сбой сети или ответ 429 — повтор; ничего не нашлось или неверный аргумент — вернуть message, модель попробует иначе; не хватает данных, которые есть только у человека — interrupt; настоящий дефект — исключение наружу.

Отдельно: текст внутреннего исключения не стоит класть в ToolMessage на пользовательском пути. Он уходит в модель, а оттуда может попасть в ответ пользователю вместе со стектрейсом и параметрами подключения.

Tools на стороне провайдера

Часть tools выполняет сам провайдер — веб-поиск, интерпретатор кода. В ответе они видны блоками server_tool_call и server_tool_result, тарифицируются провайдером и ваш код не выполняют. Из этого следует, что ваши таймауты, повторы и журналирование к ним не применяются.

Есть и обратный вариант — headless tools: схема объявлена, реализации нет. Запуск ставится на паузу с описанием нужного действия, приложение выполняет его там, где положено (в браузере, в другом сервисе) и возобновляет работу с результатом.

Agent

from langchain.agents import create_agent

agent = create_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[search_listings, send_email],
    system_prompt="Ты ассистент риелтора. Отвечай коротко и по делу.",
)

result = agent.invoke({"messages": [{"role": "user", "content": "Что есть в Одессе до 200 тысяч?"}]})
print(result["messages"][-1].content)

Что здесь собралось:

Цикл завершается, когда модель отвечает без вызова tools. Если она запросила несколько tools за один ход, они выполняются параллельно, и каждый возвращает свой ToolMessage.

Сам цикл небольшой и фиксированный. Места, где в него можно вмешаться, вынесены в middleware — то есть настройка агента идёт не через аргументы конструктора, а через хуки вокруг цикла.

Параметры create_agent

В примерах модель задают строкой, в рабочем коде её обычно берут из конфигурации вместе с таймаутом, повторами и цепочкой fallbacks.

Результат — словарь

result["messages"][-1].content    # текст ответа
result["structured_response"]     # разобранная схема, если задан response_format

Возвращается финальное state, а не message, поэтому обращение вида result.content даст AttributeError.

Response format — схема внутри цикла

from pydantic import BaseModel

class Weather(BaseModel):
    temperature: float
    condition: str

agent = create_agent("openai:gpt-5.5", tools=[weather_tool], response_format=Weather)

result = agent.invoke({"messages": [{"role": "user", "content": "Погода в Одессе?"}]})
result["structured_response"]   # Weather(temperature=27.0, condition='ясно')

Схема применяется внутри цикла, а не вторым вызовом модели поверх готового ответа. Стратегий две: ProviderStrategy использует нативную структурированную выдачу провайдера (надёжнее, дешевле по токенам, доступна не везде), ToolStrategy эмулирует её через вызов tool (работает на любой модели с поддержкой tools и позволяет объединять несколько альтернативных схем). Если передать схему без указания стратегии, LangChain посмотрит на профиль модели и выберет сам.

У ToolStrategy настраивается handle_errors — поведение при ошибке валидации: по умолчанию запрос повторяется, а модели передаётся текст ошибки. Можно задать своё сообщение, ограничить повторы конкретными типами исключений, передать функцию-обработчик или отключить повторы совсем.

Profile — возможности модели

model.profile
# {'max_input_tokens': 400000, 'tool_calling': True, 'reasoning_output': True, 'multimodal': True, ...}

Описание возможностей конкретной модели. По нему можно проверить поддержку tools перед их привязкой или свериться с размером контекста — вместо ветвлений по названию модели. Тем же профилем пользуется механизм выбора стратегии для response_format.

Context и state — разные вещи

from dataclasses import dataclass

@dataclass
class Context:
    user_id: str

agent.invoke(
    {"messages": [{"role": "user", "content": "Какой у меня баланс?"}]},
    config={"configurable": {"thread_id": "t-1"}},
    context=Context(user_id="user-123"),
)

Личность пользователя кладут в context, а не в state: state сохраняется и переигрывается, поэтому при возобновлении старого диалога в свежий запуск попал бы вчерашний пользователь.

Память между ходами

from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(model=..., tools=tools, checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "conversation-42"}}

agent.invoke({"messages": [{"role": "user", "content": "Меня зовут Алиса"}]}, config=config)
agent.invoke({"messages": [{"role": "user", "content": "Как меня зовут?"}]}, config=config)

Нужны обе части: checkpointer и thread_id. Checkpointer без thread_id ничего не сохраняет и при этом не выдаёт ошибку — проявляется это тем, что ассистент не помнит предыдущий ход.

InMemorySaver живёт до перезапуска процесса и предназначен для разработки; для продакшена есть Postgres.

Recursion limit

Цикл ограничен количеством super-steps, по умолчанию их 25. При исчерпании поднимается исключение.

result = agent.invoke(payload, config={"recursion_limit": 40})

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

Middleware — шесть точек настройки

Middleware — код, который выполняется вокруг шагов агентного цикла. Через него настраивается всё: подмена модели на лету, ограничения, повторы, маскирование персональных данных, пауза на человека, сборка динамического промпта.

На схеме это выглядит как слои вокруг цикла:

Форм две. Node-style хуки (before_*, after_*) получают state и возвращают его обновление — они наблюдают и дополняют. Wrap-style (wrap_*) стоят прямо в пути вызова, получают запрос и функцию-продолжение: могут повторить, подменить параметры, вернуть свой результат вместо настоящего.

Порядок выполнения

Для списка из трёх middleware порядок такой:

Хуки before_* идут сверху вниз, after_* — снизу вверх, wrap_* вкладываются друг в друга, и первое middleware в списке оказывается самым внешним.

Отсюда два практических следствия. Повтор, который должен обернуть всё остальное, ставят первым. Проверка, которая обязана увидеть финальный ответ, тоже ставится ближе к началу списка — на обратном пути она сработает последней.

Как пишется

Один хук — декоратором:

from langchain.agents.middleware import before_model, wrap_model_call
from langchain.messages import AIMessage

@before_model(can_jump_to=["end"])
def stop_long_conversation(state, runtime):
    if len(state["messages"]) >= 50:
        return {"messages": [AIMessage("Диалог стал слишком длинным.")], "jump_to": "end"}
    return None

@wrap_model_call
def upgrade_for_experts(request, handler):
    if request.runtime.context.tier == "expert":
        return handler(request.override(model=strong_model, tools=advanced_tools))
    return handler(request)

ModelRequest несёт messages, model, набор tools, системное сообщение, state и runtime. Он не мутируется: request.override(...) возвращает изменённую копию, которую и передают дальше.

Node-style хуки могут не возвращаться в цикл, а перейти сразу к концу, к tools или к model. Такой переход объявляется заранее списком возможных целей — параметром can_jump_to, как в примере выше; в классовой форме то же самое задаётся декоратором @hook_config над методом.

Wrap-style хуки обязаны возвращать значение: генератор с yield внутри даст ошибку в рантайме.

Несколько хуков сразу, собственное state или свои tools — это класс:

from langchain.agents.middleware import AgentMiddleware, AgentState
from typing_extensions import NotRequired

class UsageState(AgentState):
    model_calls: NotRequired[int]

class UsageMiddleware(AgentMiddleware):
    state_schema = UsageState
    tools = [reset_usage]

    async def aafter_model(self, state, runtime):
        return {"model_calls": state.get("model_calls", 0) + 1}

У node-style хуков есть асинхронные варианты с префиксом a — на пути запроса реализуют именно их.

Dynamic prompt

system_prompt задаётся статично. Всё, что зависит от state, context или найденных документов, собирается отдельным хуком:

from langchain.agents.middleware import dynamic_prompt

@dynamic_prompt
def prompt_for_tier(request):
    tier = request.runtime.context.tier
    return f"Ты агент поддержки. У клиента тариф «{tier}»."

agent = create_agent(model=..., tools=tools, middleware=[prompt_for_tier], context_schema=Context)

Каталог готовых middleware

Большая часть типовых требований закрыта поставляемыми реализациями.

Контекст и память. SummarizationMiddleware — суммаризация истории при приближении к лимиту токенов, с настройками порога, сохраняемого хвоста и способа подсчёта. ContextEditingMiddleware — очистка старых результатов tools ради освобождения контекста.

Безопасность и контроль. HumanInTheLoopMiddleware — пауза на одобрение человеком. PIIMiddleware — обнаружение и маскирование персональных данных по типу, с выбором стратегии и с указанием, к чему применять: ко вводу, к выводу, к результатам tools. ModelCallLimitMiddleware и ToolCallLimitMiddleware — ограничение числа вызовов в пределах thread или одного запуска, с выбором поведения при достижении предела.

Надёжность. ModelRetryMiddleware и ToolRetryMiddleware — повтор с экспоненциальной задержкой. ToolErrorMiddleware — превращение исключений в message, с которым модель может работать. ModelFallbackMiddleware — переключение на запасные модели при отказе основной.

Возможности. TodoListMiddleware — список задач для планирования. LLMToolSelectorMiddleware — предварительный отбор релевантных tools дешёвой моделью, когда их стало много. SubAgentMiddleware — делегирование подзадач изолированным subagents. FilesystemMiddleware и FilesystemFileSearchMiddleware — файловая система поверх подключаемого бэкенда и поиск по файлам. ShellToolMiddleware — постоянная сессия командной оболочки. RubricMiddleware — самопроверка по заданным критериям с итерациями. LLMToolEmulator — эмулятор выполнения tools на модели, для тестов.

Специфичное для провайдеров. Кеширование промпта и встроенные tools у Anthropic, кеширование у Bedrock, модерация содержимого у OpenAI.

Два предупреждения по сочетаемости. Механизм fallbacks есть и у самой модели, и в middleware — работать должен один из них, иначе получится два слоя переключения друг поверх друга. И SummarizationMiddleware переписывает историю, то есть меняет то, что ассистент помнит о разговоре: включать её стоит осознанно.

Middleware или node

Обе конструкции выполняют код вокруг вызова модели. Граница такая: middleware — когда поведение сквозное и не связано с содержанием разговора (повторы, маскирование, лимиты, трассировка, сборка промпта). Node — когда шаг является частью процесса: решение о маршруте, отдельная фаза работы, то, что вы нарисовали бы на схеме.

LangGraph: когда порядок шагов задаёте вы

Agent подходит, пока последовательность действий выбирает модель. Другой класс задач выглядит иначе: сначала всегда загрузить профиль пользователя, потом классифицировать запрос, дальше по результату — либо уточняющий вопрос, либо ответ, а после ответа — запись в журнал. Здесь порядок известен заранее, и описывать его словами в промпте необязательно: он описывается кодом.

Graph — способ описать процесс явно. Детерминированные шаги остаются кодом, решения остаются за моделью, и всё вместе образует схему, которую можно нарисовать и обсудить.

Вот как выглядит типичный чат-конвейер, собранный графом. Слева помечено, что делает каждый шаг: обычный код или вызов модели.

Здесь видно главное различие с агентом: порядок шагов задан явно. Модель решает, что за вопрос ей задали, а не то, в каком порядке дальше выполнять работу.

Три сущности

State — общие данные, с которыми работают все шаги. Nodes — сами шаги, обычные функции, каждая возвращает частичное обновление state. Edges — переходы: что выполняется следующим.

Исполнение идёт super-steps. Всё, что запланировано на текущий такт, выполняется — параллельно, если граф это допускает; затем обновления сливаются в state, затем планируется следующий такт. Перед запуском граф компилируется.

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    question: str
    answer: str

def answer(state: State) -> dict:
    return {"answer": f"Ответ на {state['question']}"}

graph = (
    StateGraph(State)
    .add_node("answer", answer)
    .add_edge(START, "answer")
    .add_edge("answer", END)
    .compile()
)

START и END — специальные метки: первая обозначает точку входа, куда попадает пользовательский ввод, вторая — выход. Edge обратно в START недопустим; если нужен цикл, он делается через именованный node.

Схемой state может быть TypedDict, Pydantic-модель или dataclass. Чаще берут первое как самое лёгкое.

Reducers — как сливаются обновления

У каждого ключа state есть reducer: правило, по которому обновление от node соединяется с тем, что уже лежит в state.

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

class State(TypedDict):
    name: str                                     # по умолчанию: перезапись
    findings: Annotated[list[str], operator.add]  # добавление в конец
    messages: Annotated[list, add_messages]       # добавление с дедупликацией по id

Наглядно, что происходит на одном super-step, когда два nodes отработали параллельно:

Reducer по умолчанию перезаписывает значение. Поэтому список без явного reducer, в который пишут два nodes, сохранит только значение одного из них — без ошибки и без предупреждения. Любое накапливающее поле должно иметь reducer.

Для messages есть готовый add_messages: он добавляет новые, отбрасывает дубликаты по идентификатору и превращает словари в объекты messages. Есть и готовый класс MessagesState, который включает его сразу.

Второй способ обойти reducers — вернуть из node всё state целиком:

def good(state: State) -> dict:
    return {"answer": "..."}          # только то, что изменилось

def bad(state: State) -> State:
    state["answer"] = "..."           # мутация проходит мимо reducers
    return state

Ещё у графа можно задать отдельные схемы входа и выхода — тогда внутренние рабочие поля не будут частью публичного интерфейса.

Nodes

Node принимает state, а при необходимости ещё RunnableConfig (в нём лежат thread_id, теги, настраиваемые значения) или Runtime (context, store, писатель в поток). Асинхронные nodes — обычные async def с теми же сигнатурами.

Удобный приём для сервиса: node создаётся фабричной функцией, которая замыкает в себе зависимости — клиент базы, HTTP-сессию, конфигурацию — и возвращает функцию node. Так зависимости не превращаются в глобальные переменные, а node легко тестировать отдельно.

Маршрутизация

def route(state: State) -> Literal["clarify", "generate"]:
    return "clarify" if state["needs_input"] else "generate"

builder.add_conditional_edges("assess", route, {"clarify": "clarify", "generate": "generate"})

Третий аргумент — карта возвращаемых значений в имена nodes. Он необязателен, но с ним граф корректно рисуется, а множество достижимых nodes становится явным.

Когда нужно одновременно обновить state и выбрать следующий шаг, node возвращает Command:

from langgraph.types import Command
from typing import Literal

def triage(state: State) -> Command[Literal["escalate", "resolve"]]:
    if state["severity"] > 3:
        return Command(update={"assigned": "oncall"}, goto="escalate")
    return Command(update={"assigned": "bot"}, goto="resolve")

Аннотация Literal с перечислением достижимых nodes нужна для отрисовки и проверки целей.

Важная деталь поведения: Command добавляет динамический edge, а не заменяет статический. Если из того же node есть обычный edge, выполнятся оба направления. Внешне это выглядит как дублирование работы.

Из subgraph команда может перевести управление в родительский граф — для этого есть Command(goto=..., graph=Command.PARENT).

Send — параллельный запуск

Когда количество ветвей известно только в рантайме:

from langgraph.types import Send

class State(TypedDict):
    topics: list[str]
    drafts: Annotated[list[str], operator.add]     # накопитель обязателен

def fan_out(state: State):
    return [Send("write_draft", {"topic": topic}) for topic in state["topics"]]

builder.add_conditional_edges(START, fan_out, ["write_draft"])
builder.add_edge("write_draft", "synthesise")

Каждый Send несёт свой приватный вход одному вызову обработчика — так делается схема «разослать и собрать»:

Результаты собираются через reducer; если reducer на собирающем поле нет, сохранится результат последнего завершившегося обработчика.

Subgraphs

Скомпилированный граф — исполняемый объект, поэтому его можно поставить node другого графа:

builder.add_node("research", research_graph)

Общие ключи state перетекают автоматически. Если схемы state разные, нужен node-переходник, который преобразует данные на входе и на выходе.

Управление на уровне node

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

builder.add_node("fetch", fetch, retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0))
builder.add_node("embed", embed, cache_policy=CachePolicy(ttl=300))
graph = builder.compile(cache=InMemoryCache())

Оттуда же — RunControl.request_drain(): кооперативная остановка, которая просит запуск остановиться в ближайшей безопасной точке и оставляет checkpoint для продолжения. Это штатный способ пережить выкатку новой версии или остановку контейнера посреди работы.

recursion_limit у графа такой же, как у агента, и задаётся в конфигурации запуска. Есть и специальное поле state RemainingSteps — node может посмотреть в него и свернуть работу заранее, вместо того чтобы упереться в предел.

Checkpointer и store: память и устойчивость

Памяти две, и они решают разные задачи.

Ни то ни другое по умолчанию не включено.

Checkpointers

from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "conversation-1"}}

await graph.ainvoke({"messages": ["Привет"]}, config)
await graph.ainvoke({"messages": ["И ещё раз"]}, config)   # видит первый ход

Две эксплуатационные детали. Метод .setup(), создающий таблицы, — операция уровня выкатки, а не старта приложения: изменение схемы на старте при нескольких репликах выполняется параллельно само с собой. И thread_id для Postgres держат короче 255 символов — колонка ограничена.

Просмотр истории и перемотка

snapshot = await graph.aget_state(config)                        # текущее state и что дальше
history = [s async for s in graph.aget_state_history(config)]    # свежие записи первыми

past = history[-2]
await graph.ainvoke(None, past.config)                           # переиграть с той точки

fork = await graph.aupdate_state(past.config, {"messages": ["исправлено"]})
await graph.ainvoke(None, fork.config)                           # продолжить по новой ветке

Вызов с None вместо входных данных означает «продолжи с checkpoint». На этом же построена отладка: можно отмотать к нужному шагу, поправить state и посмотреть, как пойдёт дальше.

Одна особенность: update_state проходит через reducers. На поле с добавлением оно добавит там, где предполагалась замена. Для замены есть обёртка Overwrite:

from langgraph.types import Overwrite

await graph.aupdate_state(config, {"items": Overwrite(["C"])})

Хранение checkpoints

Checkpoints создаются по одному на super-step для каждого thread и сами не удаляются. Для долгоживущего продукта это означает постоянный рост таблицы, поэтому политику хранения — сколько времени thread остаётся возобновляемым и что удаляет остальное — стоит определить заранее.

Store

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
graph = builder.compile(checkpointer=checkpointer, store=store)

Операции: put, get, search, delete. Namespace — иерархический кортеж вида ("users", user_id, "preferences"), и он же служит границей изоляции. Собирают его из доверенного context, а не из текста, пришедшего от модели: namespace без идентификатора пользователя или арендатора означает, что данные видны всем.

Если задать store функцию эмбеддингов и размерность через IndexConfig, поиск станет семантическим вместо точного. Продакшен-реализация — PostgresStore.

В графе к store обращаются через runtime, а не через глобальную переменную:

def recall(state, runtime: Runtime):
    item = runtime.store.get(("users", runtime.context.user_id), "preferences")
    ...

Durability — режимы записи

Настройка того, как часто state записывается на диск. Передаётся при вызове:

await graph.ainvoke(payload, config=config, durability="async")

Выбор зависит от цены потерянного шага: если шаги отправляют письма или проводят платежи, подходит "sync"; для аналитического прогона по длинному диалогу достаточно "exit".

Переигрывание при возобновлении

Общее правило рантайма: возобновление выполняет node с самого начала, пропускается только завершённая работа, попавшая в checkpoint. Отсюда два следствия:

• Побочные эффекты, выполненные до точки остановки, произойдут ещё раз.
• Недетерминированные значения — время, случайные идентификаторы — при переигрывании получатся другими. Их вычисляют в отдельном node, чтобы значение попало в checkpoint и подставлялось оттуда.

Human-in-the-loop: пауза на человека

Иногда запуск должен остановиться и дождаться человека: подтвердить отправку, поправить черновик, ответить на уточняющий вопрос. Для этого есть interrupt.

from langgraph.types import interrupt, Command

def review(state):
    decision = interrupt({"draft": state["draft"], "question": "Публикуем?"})
    return {"approved": decision == "yes"}

Вызов останавливает запуск, сохраняет всё в checkpoint и отдаёт наружу переданные данные. Запуск остаётся приостановленным — секунду или неделю — пока его не продолжат через Command(resume=...); тогда interrupt вернёт переданное значение.

Обратите внимание на второе появление review в схеме: node с паузой при возобновлении выполняется с самого начала. Это отдельная тема, к ней вернёмся ниже.

Checkpointer и thread_id для этого обязательны: без них паузу некуда сохранить.

Есть и статический вариант — interrupt_before и interrupt_after при компиляции, остановка на границе node. Он удобен для отладки и пошагового прохода, но не несёт полезной нагрузки, поэтому для продуктовых сценариев одобрения используют обычный interrupt.

Одобрение вызовов tools

Для самого частого случая — подтвердить опасный вызов до выполнения — есть готовое middleware:

from langchain.agents.middleware import HumanInTheLoopMiddleware

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[write_file, execute_sql, read_data],
    checkpointer=checkpointer,
    middleware=[HumanInTheLoopMiddleware(
        interrupt_on={
            "write_file": {"allowed_decisions": ["approve", "edit", "reject"]},
            "execute_sql": {"allowed_decisions": ["approve", "reject"]},
            "read_data": False,
        },
        description_prefix="Ожидает одобрения",
    )],
)

True включает interrupt с решениями по умолчанию, False пропускает вызов без вопросов, словарь настраивает, какие решения доступны проверяющему.

await agent.ainvoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config, version="v2")

Когда модель за один ход запрашивает несколько защищённых tools, они предъявляются вместе, и список решений должен идти в том же порядке, что и предъявленные действия.

Interrupt можно включать по условию — параметром when: функция смотрит на аргументы вызова и решает, нужно ли подтверждение. Так проверяются только рискованные случаи, например запись за пределы рабочего каталога, а остальные проходят без вопросов.

Побочные эффекты и точка остановки

Здесь то же правило переигрывания, но с наглядными последствиями:

def send_and_confirm(state):
    send_email(state["draft"])              # отправится ещё раз при возобновлении
    interrupt({"sent": True})

def confirm_then_send(state):
    decision = interrupt({"draft": state["draft"]})
    if decision == "approve":
        send_email(state["draft"])          # после паузы — один раз
    return {"sent": decision == "approve"}

Отсюда три правила: побочные эффекты размещают после interrupt либо делают идемпотентными; недетерминированные значения выносят в отдельный node; node с interrupt держат небольшим, потому что повторится всё, что в нём есть.

Тот же механизм используется не только для одобрений: им редактируют state, запрашивают недостающие данные, которые есть только у человека, и выполняют headless tools на стороне клиента.

Streaming

Потоковая выдача нужна, чтобы ответ появлялся по мере генерации, а на длинных операциях было видно, что происходит.

Откуда что берётся:

Здесь два разных API.

Первый — stream / astream с указанием stream_mode. Подходит для большинства задач.

Режимы можно запрашивать по несколько сразу, списком.

Версия формата потока

У формата есть версия, и по умолчанию действует v1, где форма ответа зависит от вызова: один режим отдаёт сырые данные, несколько режимов — кортежи (mode, data), а включённые subgraphs добавляют в кортеж namespace.

Версия v2 отдаёт всё единообразно — словарь StreamPart с полями type, ns и data:

async for chunk in graph.astream(payload, stream_mode="messages", version="v2"):
    if chunk["type"] == "messages":
        message, metadata = chunk["data"]
        if metadata["langgraph_node"] in {"generate", "answer_user"}:
            print(message.content, end="")

В новом коде версию указывают явно: умолчание осталось прежним ради совместимости.

Метаданные несут langgraph_node, теги и идентификаторы запуска. Фильтрация по имени node — способ не пустить в пользовательский поток токены внутренних вызовов, например классификации. Можно поступить и иначе — пометить такую модель тегом nostream:

classifier = model.with_config({"tags": ["nostream"]})

Тогда её токены не эмитятся вовсе.

Subgraphs в потоке

async for chunk in graph.astream(payload, stream_mode="messages", subgraphs=True, version="v2"):
    ...

Без subgraphs=True токены, порождённые внутри вложенного графа — включая agent, поставленный node, — в поток не попадают. Ошибки при этом нет, поток просто оказывается неполным. Поле ns в каждом фрагменте показывает источник: пустой кортеж для корневого графа, имя node с идентификатором задачи для вложенного.

Custom events

from langgraph.config import get_stream_writer

def index_documents(state):
    writer = get_stream_writer()
    writer({"type": "progress", "done": 0, "total": 100})

Читаются в режиме custom. Тот же механизм — способ отдать в поток модель, у которой нет интеграции с LangChain: вызвать её самостоятельно и записывать фрагменты вручную.

astream_events

Второй API подробнее: каждый Runnable в графе сообщает о старте, о фрагментах и о завершении.

async for event in graph.astream_events(state, version="v2"):
    kind = event["event"]          # on_chain_start, on_chat_model_stream, on_chain_end, on_tool_start…
    name = event["name"]
    data = event["data"]
    metadata = event.get("metadata", {})   # здесь же langgraph_node

Это нужно, когда важно знать, какой именно node породил токен: так границы nodes превращаются в стадии для пользователя («ищу», «сверяю», «отвечаю»), а токены разных nodes уходят в разные части интерфейса. Плата — объём: поток событий большой, и потребителю нужна точная фильтрация, чтобы внутреннее содержимое не попало в интерфейс.

Есть и более новый stream_events(version="v3"), построенный вокруг блоков содержимого, с типизированными проекциями по каналам сообщений, значений и жизненного цикла. Переход на него — переписывание потребителя, а не смена значения флага.

Functional API

Второй фасад к тому же рантайму. Процесс описывается обычным Python — с if, for и await, — но сохраняются checkpoints, возобновление, interrupts и streaming.

from langgraph.func import entrypoint, task

@task
def write_essay(topic: str) -> str:
    return generate(topic)

@entrypoint(checkpointer=checkpointer)
def workflow(topic: str) -> dict:
    essay = write_essay(topic).result()
    approved = interrupt({"essay": essay, "action": "одобряем?"})
    return {"essay": essay, "approved": approved}

@entrypoint принимает один позиционный аргумент — несколько значений передают словарём. Вход и выход должны сериализоваться, потому что попадают в checkpoint. Схемы state здесь нет: вместо неё в следующий запуск подставляется результат предыдущего через параметр previous. Есть и entrypoint.final, когда наружу возвращается одно значение, а для следующего запуска сохраняется другое.

@task — единица работы, результат которой записывается в checkpoint. Вызов возвращает future сразу; несколько задач, запущенных до того, как разрешена первая, выполняются параллельно — здесь это заменяет Send.

Правило переигрывания то же, но сформулировано для процедурного кода: при возобновлении тело entrypoint проигрывается сверху, а результаты завершённых задач подставляются из checkpoint. Поэтому каждый побочный эффект и каждое недетерминированное значение размещают внутри @task — обычный код между задачами выполнится заново.

Выбор между двумя API. Functional API короче, когда поток в основном линейный и общего state, которое стоило бы называть, нет. Graph API выигрывает, когда у процесса есть структура, несколько шагов работают с одним state и форму процесса нужно обсуждать: graph можно отрисовать, функцию с задачами — нет.

Multi-agent: несколько агентов

Слово «мультиагент» обычно означает одну из трёх конструкций:

Graph как tool. Скомпилированный граф исполняем, поэтому его оборачивают в tool — и вызывающий agent делегирует работу, ничего не зная о внутреннем устройстве.

@tool
def research(topic: str) -> str:
    """Запускает исследовательский конвейер по теме и возвращает отчёт."""
    return research_graph.invoke({"topic": topic})["report"]

Supervisor. Координатор распределяет работу между специалистами и собирает результаты. В виде графа это node-маршрутизатор плюс по одному node на специалиста, маршрут выбирается через Command. Такая схема рисуется и инспектируется.

Swarm. Агенты передают управление друг другу напрямую, без центрального маршрутизатора. Гибче, но каждый edge передачи — потенциальный цикл, поэтому нужен recursion_limit.

Ещё вариант — subagents через SubAgentMiddleware: тогда у обычного агента появляется tool делегирования, без остальной обвязки Deep Agents.

Subagent или subgraph

Определяющее свойство — изоляция контекста. Делегирование subagent уместно, когда подзадача иначе заполнит родительский контекст деталями, которые дальше не нужны: чтение длинного документа, шумный поиск. Subgraph уместен, когда шаг — часть процесса и его state важно для последующих шагов.

Про стоимость: каждый subagent выполняет свой цикл со своим системным промптом и своими описаниями tools, поэтому запуск пяти subagents расходует примерно как пять агентов, а не как один.

Deep Agents

Готовая сборка middleware поверх обычного агента, рассчитанная на длинные задачи, которые не помещаются в одно контекстное окно.

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[get_weather],
    system_prompt="Ты исследовательский ассистент.",
)

Что включено сразу:

• Файловая система с полным набором tools: просмотр, чтение, запись, правка, удаление, поиск по маске и по содержимому. Отдельные tools можно исключить, сам слой — нет.
• Суммаризация и выгрузка контекста, чтобы длинная задача не упёрлась в размер окна.
• Кеширование промпта для статичных частей у провайдеров, которые это поддерживают.
• Subagents через tool, который порождает временного агента со свежим контекстом.

Планирование задач с версии 0.7 стало опциональным — в материалах постарше оно описано как включённое по умолчанию.

Файловая система подключаемая: память, локальный диск, store графа, композиция из нескольких источников или своя реализация. У бэкендов есть декларативные правила доступа по маскам путей — они определяют, что именно агенту разрешено читать и писать. Для выполнения кода существуют песочницы с командной оболочкой и интерпретаторы, выполняющие JavaScript.

Обвязка подходит задачам, которым действительно нужны планирование на много шагов, файлы как рабочая память, делегирование и память между сессиями. Для ограниченного цикла с несколькими tools она принесёт возможности, которые задача не использует, но которые занимают часть бюджета промпта.

Retrieval: поиск по документам

Тема отдельная и большая, у меня есть статья про RAG и функциональные вызовы и статья про векторные базы. Здесь — то, что касается стека.

Конвейер состоит из трёх стадий, и каждая тестируется отдельно:

Индексация выполняется отдельно и заранее; поиск и генерация — на каждый запрос пользователя.

Архитектур две. Классическая двухшаговая: всегда искать, потом отвечать — предсказуемая задержка, меньше вызовов модели, подходит для узкого корпуса, где поиск нужен на каждый вопрос. Agentic RAG: поиск оформлен tool, и модель сама решает, нужен ли он, когда и с каким запросом — переменная задержка, больше ходов, подходит для открытых вопросов и нескольких источников. В первой версии agentic описан как основной вариант.

@tool
def search_docs(query: str) -> str:
    """Поиск по документации продукта. Используй для вопросов о том, как продукт работает."""
    return "\n\n".join(d.page_content for d in retriever.invoke(query))

agent = create_agent(model="openai:gpt-5.5", tools=[search_docs])

Docstring здесь работает как политика поиска: он объясняет модели, для каких вопросов этот корпус подходит.

Составные части

Loaders возвращают Document — текст плюс метаданные. Метаданные проставляются при загрузке: источник, раздел, время, арендатор. Добавить их позже — значит переиндексировать корпус.

Splitters:

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""],
)
chunks = splitter.split_documents(docs)

Размер куска — параметр качества поиска. Слишком мелкий теряет контекст, из-за которого кусок вообще отвечал на вопрос; слишком крупный размывает эмбеддинг, и он начинает слабо совпадать со всем подряд. Обычный стартовый диапазон — 500–1500 символов с перекрытием 10–20%, дальше подбирается измерением.

Embeddings:

from langchain.embeddings import init_embeddings

embeddings = init_embeddings("openai:text-embedding-3-small")

Одна и та же модель и одна и та же размерность должны использоваться и при индексации, и при запросах. Если смешать, расстояния в хранилище потеряют смысл, и ошибки при этом не будет.

Vector stores: InMemoryVectorStore для тестов, Chroma для локальной разработки, pgvector рядом с существующим Postgres, а также Pinecone, Qdrant и Weaviate отдельными пакетами.

Retrievers:

retriever = store.as_retriever(search_kwargs={"k": 4})

# Разнообразие вместо почти одинаковых кусков
retriever = store.as_retriever(search_type="mmr",
                               search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.5})

# Ограничение по метаданным — здесь же фильтр по арендатору
docs = store.similarity_search(query, k=5, filter={"tenant_id": tenant})

Фильтр по арендатору и правам ставится именно в retriever: просьба в промпте не показывать чужое механизмом изоляции не является.

Заземление ответа

Поиск помогает, если промпт заставляет модель предпочесть найденный текст собственной памяти. Практика такая: сказать явно, что отвечать нужно только по предоставленному контексту, а если ответа там нет — так и сказать; пронумеровать куски и требовать ссылки на них, тогда неподтверждённые утверждения становятся видны; держать найденный текст в явно ограниченном блоке.

Последнее важно и с точки зрения безопасности: найденный текст — недоверенный ввод. Документ, содержащий фразу «игнорируй предыдущие инструкции», представляет собой инъекцию промпта, поэтому его не вклеивают в системное сообщение.

LangSmith: трассировка, датасеты и оценка

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

LangSmith включается переменными окружения и не требует правок в коде:

LANGSMITH_TRACING=true
LANGSMITH_API_KEY=<ключ>
LANGSMITH_PROJECT=<проект>

После этого каждый запуск LangChain и LangGraph попадает в трассировку: nodes, вызовы моделей, tools, токены, задержки. Имена переменных с прежним префиксом LANGCHAIN_ больше не работают.

К запускам стоит добавлять теги и метаданные — по ним трассы потом ищутся:

import langsmith as ls

with ls.tracing_context(enabled=True, project_name="quality-runs"):
    await agent.ainvoke(payload)

await agent.ainvoke(payload, config={"tags": ["production"], "metadata": {"user_id": user_id}})

Что ещё есть на платформе: панели и оповещения по метрикам качества; правила, вебхуки и онлайн-оценка прямо на потоке продакшена; очереди разметки, где ответы аннотируют вручную или собирают обратную связь пользователей; автоматический анализ повторяющихся проблем в трассах.

Отдельный блок — работа с промптами: playground для экспериментов с промптами и конфигурациями моделей, версионирование с фиксацией изменений, prompt hub с тегами и публичной частью, доступ к промптам из кода. Смысл в том, что промпт меняется чаще кода и редактировать его могут люди, которые не работают с репозиторием.

Важное соображение: трассы содержат промпты и ответы, то есть содержимое пользователей, покидающее ваш периметр. Передача его во внешний сервис — вопрос договорённостей и требований к обработке данных.

Альтернатива, если такое решение не принято: собственная трассировка через OpenTelemetry плюс одна структурированная запись на запуск. В любом случае для каждого вызова модели имеет смысл записывать модель, node, задержку, число токенов, статус и идентификатор корреляции. Полные промпты и ответы в обычные логи не пишут — это те же пользовательские данные, только без контроля доступа.

Точка расширения для своей записи — callbacks: BaseCallbackHandler с событиями on_llm_start, on_llm_end, on_tool_start и так далее. Готовый обработчик для подсчёта токенов уже есть.

Оценка качества

Оценка состоит из трёх частей: dataset (входы и представление о правильном выходе), target (то, что запускается) и evaluators (оценивающие функции). Результат прогона называется экспериментом, и его можно сравнивать с предыдущими.

from langsmith import Client

client = Client()
dataset = client.create_dataset(dataset_name="Sample dataset")

examples = [
    {"inputs": {"question": "В какой стране находится Килиманджаро?"},
     "outputs": {"answer": "Килиманджаро находится в Танзании."}},
]
client.create_examples(dataset_id=dataset.id, examples=examples)

experiment_results = client.evaluate(
    target,
    data="Sample dataset",
    evaluators=[correctness_evaluator],
    experiment_prefix="first-eval",
    max_concurrency=2,
)

Evaluators бывают трёх видов. Детерминированные проверки — схема валидна, есть обязательный идентификатор, длина в границах — применимы везде, где проверка механическая. LLM-as-judge оценивает то, что механически не проверяется: достоверность, тон, соблюдение заданного голоса; у судьи есть смещения по позиции и по многословности, и он сам по себе промпт, который может измениться. Человеческая разметка — самая дорогая, её обычно оставляют для выборки, которую отметили автоматические проверки.

Прогонять оценку имеет смысл при смене промпта, при смене модели и перед релизом — это три момента, когда качество меняется, не роняя ни одного теста. Dataset от прогона к прогону должен быть один и тот же.

Есть и оценка траектории — проверка того, какой путь agent прошёл, а не только каким получился ответ.

Тестирование

Слоями, от дешёвых к дорогим.

Без модели. Разводка графа, маршрутизаторы, reducers и логика nodes — обычные функции. Маршрутизатор является чистой функцией от state, и тестировать его удобнее напрямую.

С подставной моделью. Настоящий граф, фальшивая модель с заранее заданными ответами по порядку. Этого хватает, чтобы детерминированно прогнать маршрутизацию, streaming и пути ошибок:

from langchain_core.language_models.fake_chat_models import FakeListChatModel

model = FakeListChatModel(responses=["CLARIFY", "финальный ответ"])

Для tools ту же роль играет LLMToolEmulator.

С настоящей моделью, по датасету. Слой, который ловит регрессии промптов — ситуации, когда все тесты зелёные, а качество ответов изменилось. Такие прогоны обычно выносят из общего конвейера сборки, потому что они медленные, платные и слегка плавающие.

Чего подставные модели не покажут: действительно ли промпт вызывает нужное поведение и разбирается ли structured output по грамматике конкретного провайдера.

Запуск и выкатка

Два способа довести graph до пользователей.

Готовый сервер

LangSmith Deployment — среда выполнения для агентов: она берёт на себя очередь задач, устойчивое выполнение, streaming и горизонтальное масштабирование. Работающая часть называется Agent Server и построена вокруг трёх понятий: assistant — граф с конкретной конфигурацией, thread — контекст состояния, run — одно выполнение. Есть cron jobs для регулярных запусков, слой хранения с checkpoints и store, API для создания runs, чтения state и подключения к потоку, а также клиентская библиотека langgraph-sdk.

Разместить это можно четырьмя способами: полностью управляемое облако; собственный кластер Kubernetes вместе с самостоятельно развёрнутым LangSmith; гибрид, где управляющий слой у поставщика, а серверы и данные у вас; отдельно стоящий сервер в Docker или Kubernetes без управляющего слоя.

Конфигурация проекта описывается одним файлом:

{
  "dependencies": ["."],
  "graphs": { "agent": "./my_agent/agent.py:graph" },
  "env": "./.env",
  "python_version": "3.12"
}

Обязательных ключей два: dependencies и graphs — карта «идентификатор графа → файл и переменная». Из необязательных есть env, настройки store (включая семантический поиск и время жизни записей), настройки checkpointer, параметры http вроде CORS и отключения маршрутов, webhooks, обработчик auth, базовый образ и версия Python, а также закрепление версии серверного API.

Команды langgraph-cli:

pip install -U "langgraph-cli[inmem]"
langgraph new path/to/app --template new-langgraph-project-python
cd path/to/app && pip install -e .
langgraph dev

После запуска доступны API на порту 2024, его документация и Studio — визуальный интерфейс, который подключается к локально работающему агенту. В нём видно каждый шаг: какие промпты ушли модели, какие вызовы tools произошли и с какими аргументами, что вернулось, сколько заняло времени и токенов. Оттуда же можно перезапустить thread с любого шага и посмотреть, как изменится поведение.

Разница между langgraph dev и langgraph up: первый быстрый, работает без Docker и держит состояние в памяти — он для итераций; второй поднимает окружение, близкое к продакшену, с настоящим Postgres.

Встроенный в свой сервис

Второй способ: graph остаётся библиотечным вызовом внутри вашего приложения, API и хранилище ваши, выкатка идёт вашим существующим конвейером. Такой вариант подходит, когда agent — часть большой системы, в которой аутентификация, квоты и внешний контракт уже реализованы в другом месте.

Тогда четыре вещи, которые Agent Server даёт сам, нужно обеспечить самостоятельно:

• Компилировать графы один раз при старте приложения. Компиляция на каждый запрос добавляет задержку каждому ходу.
• Не блокировать цикл событий. Синхронный вызов внутри node останавливает все параллельные запросы этого рабочего процесса.
• Готовить схему checkpointer на выкатке, а не при старте, и использовать общий с приложением пул соединений.
• Останавливаться аккуратно. При выкатке контейнер гасится, и запуск на середине обрывается; request_drain() оставляет checkpoint, с которого можно продолжить.

Переход с версии 0.x

Раздел полезен даже тем, кто начинает с нуля: обучающие данные моделей и большая часть статей в сети описывают прошлую версию, поэтому правдоподобно выглядящий код часто оказывается удалённым API.

Удалено полностью: поддержка Python 3.9; привязка tools к модели перед передачей её агенту (теперь tools передаются параметром create_agent); structured output через просьбу вернуть JSON и разбор прозы; .text() в виде метода; AgentExecutor, initialize_agent и набор готовых типов агентов.

Переехало в langchain-classic: старые chains, весь прежний модуль retrievers, indexing API, hub, CacheBackedEmbeddings и реэкспорты интеграций сообщества. Это не «устарело, но работает» — импорт из основного пакета завершается ошибкой.

from langchain.chains import LLMChain
from langchain.retrievers import MultiQueryRetriever

# 1.x
from langchain_classic.chains import LLMChain
from langchain_classic.retrievers import MultiQueryRetriever

Объявлено устаревшим, но работает: create_react_agent из langgraph.prebuilt; переменные окружения трассировки с префиксом LANGCHAIN_; часть интеграций сообщества, у которых появились отдельные пакеты. Устаревшее продолжает работать с предупреждением на всей линейке первой версии.

Соответствие старых аргументов новым сводится к одному правилу: что было аргументом конструктора, стало хуком middleware. prompt как функция стал @dynamic_prompt, pre- и post-model хуки — before_model и after_model, выбор модели на лету — wrap_model_call, обработчик ошибок tool — wrap_tool_call, объект ToolNode в списке — просто списком tools.

Со стороны LangGraph изменений почти нет: state, nodes и edges, модель исполнения, checkpoints, streaming и interrupts перешли без правок.

Порядок миграции: сначала Python 3.10, всё остальное блокируется этим; затем поставить langchain-classic и перенаправить импорты — после этого код снова запускается, и в изменениях становится видно, где «просто переехало», а где нужна переработка; дальше схемы state, агенты по одному, хуки, промпты, работа с messages и structured output вместе с его путём ошибки. Последним шагом — прогон датасета: механическая миграция начинает компилироваться раньше, чем поведение возвращается к прежнему, а меняется оно прежде всего в обработке промптов и structured output.

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

Поведение, которое легко пропустить

Список вещей, которые не сообщают о себе ошибкой.

• Поле-список без reducer. Пишут два nodes — сохранится значение второго.
• Node вернул state целиком вместо частичного обновления. Мутация проходит мимо reducers.
• Command с goto плюс статический edge из того же node. Выполнятся оба направления.
• stream_mode="messages" без subgraphs=True. Токены вложенного агента в поток не попадут.
• Checkpointer без thread_id. Не сохраняется ничего, ошибки нет.
• Побочный эффект до interrupt. Node переигрывается с начала, действие повторяется.
• update_state прошёл через reducer. Добавил там, где предполагалась замена.
• InMemorySaver на продакшене. State исчезает при перезапуске.
• Синхронный вызов модели в асинхронном сервисе. Блокирует все параллельные запросы процесса.
• Идентификатор пользователя аргументом tool. Его значение выбирает модель.
• Tool вернул Command без ToolMessage. Провайдер видит вызов без результата.
• Разные модели эмбеддингов при индексации и при поиске. Расстояния перестают что-либо значить.

Коротко

Что стоит держать в голове по итогам.

Слои различаются областью ответственности: LangChain — models, tools и агентный цикл; LangGraph — state, исполнение, память и streaming; Deep Agents — готовая обвязка для длинных задач; LangSmith — трассировка и оценка; Agent Server — готовая среда запуска. Agent из LangChain является graph LangGraph, поэтому возможности рантайма доступны на любом уровне.

Настройка агента живёт в middleware: шесть хуков вокруг цикла и каталог готовых реализаций.

В графе всё держится на reducers: накапливающее поле нуждается в reducer, node возвращает только то, что изменил.

Память включается явно: checkpointer и thread_id — обе части сразу.

И общее свойство стека: значительная часть его отказов тихая. Не появившиеся токены, несохранённое state, потерянные результаты Send, ответ по нерелевантным кускам текста. Поэтому трассировка запусков здесь — базовая часть работы, а не дополнительная возможность.