---
title: "LangChain и LangGraph — агенты, графы и вся инфраструктура вокруг них"
url: "https://romankryvolapov.com/ru/langchain-langgraph/"
description: "Как устроен стек LangChain 1.x и LangGraph 1.x: models и messages, tools, create_agent и middleware, StateGraph с reducers, checkpointers и store, streaming, human-in-the-loop, LangSmith, Agent Server и langgraph CLI, миграция с 0.x."
language: ru
updated: 2026-08-06
---
**Привет!**
Языковая модель умеет ровно одно: получить текст и вернуть текст. Больше ничего. Она не помнит, о чём вы говорили минуту назад, и не может сама сходить в базу данных или в интернет.
Возьмём простой пример: пользователь спрашивает у ассистента, какая сейчас погода. Модель этого не знает. Чтобы она смогла ответить, нужно проделать вот что:
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** — потоковая выдача: токены и события приходят по мере появления.
## Из чего состоит экосистема
Под одним брендом выпускается несколько продуктов с похожими названиями. Разложим по полкам.
| Что | Чем является | Чем занимается |
|---|---|---|
| **LangChain** | Фреймворк | Models, messages, промпты, tools, агентный цикл |
| **LangGraph** | Рантайм под ним | State, nodes и edges, checkpoints, streaming, устойчивое исполнение |
| **Deep Agents** | Обвязка сверху | Планирование, виртуальная файловая система, subagents, сжатие контекста |
| **LangSmith** | Платформа | Трассировка запусков, датасеты, оценка качества, работа с промптами |
| **LangSmith Deployment** | Хостинг | Готовый сервер для запуска графов, с threads, очередью задач и API |
Слои зависят снизу вверх:
Функция `create_agent` из LangChain возвращает **скомпилированный graph LangGraph** — то есть всё, что рантайм умеет делать с графом, он умеет и с агентом: сохранять state, стримить, ставить на паузу, вкладывать внутрь другого графа как обычный node. Выбирать между agent и graph не нужно, это одно и то же на разной высоте абстракции.
### Пакеты
Библиотека разложена на много пакетов; ставить нужно не всё.
| Пакет | Что внутри |
|---|---|
| `langchain-core` | Базовые типы: messages, блоки содержимого, tools, шаблоны промптов, интерфейс `Runnable` |
| `langchain` | Agents, middleware и удобные пространства имён поверх ядра |
| `langgraph` | Рантайм: state, graph, исполнение, streaming |
| `langgraph-checkpoint` | Интерфейсы checkpointers и реализация в памяти; приезжает вместе с рантаймом |
| `langgraph-checkpoint-postgres` | Checkpointer и store на Postgres |
| `langgraph-checkpoint-sqlite` | Checkpointer на SQLite для локальной разработки |
| `langgraph-prebuilt` | Готовые компоненты графа, в том числе `ToolNode` |
| `langgraph-cli` | Командная строка: шаблон проекта, локальный сервер, сборка образа, выкатка |
| `langgraph-sdk` | Клиент к развёрнутому Agent Server |
| `langchain-text-splitters` | Нарезка документов на куски для поиска |
| `langchain-community` | Длинный хвост интеграций, у которых нет отдельного пакета |
| `langchain-classic` | Наследие прошлой версии: старые chains, старые retrievers, indexing API, hub |
| `langsmith` | Трассировка, датасеты, оценка |
| `deepagents` | Готовая обвязка для длинных задач |
### Провайдеры моделей
Сама библиотека ни к одному вендору не привязана: провайдер подключается отдельным пакетом, и переключение между ними — это смена строки инициализации, а не переписывание кода. Пакеты версионируются независимо от ядра.
| Вендор | Пакет | Строка модели |
|---|---|---|
| OpenAI | `langchain-openai` | `openai:gpt-5.5` |
| Anthropic | `langchain-anthropic` | `anthropic:claude-sonnet-4-6` |
| Google Gemini | `langchain-google-genai` | `google_genai:gemini-2.5-flash-lite` |
| AWS Bedrock | `langchain-aws` | `us.anthropic.claude-sonnet-4-6` |
| Azure AI | `langchain-azure-ai` | развёртывание Azure |
| Mistral | `langchain-mistralai` | `mistralai:...` |
| Groq | `langchain-groq` | `groq:...` |
| Cohere | `langchain-cohere` | `cohere:...` |
| Hugging Face | `langchain-huggingface` | идентификатор модели на хабе |
| Ollama, локальные модели | `langchain-ollama` | `ollama:...` |
Строка вида `"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` семантическому версионированию **не следует**, и его обычно закрепляют на конкретной минорной серии.
### Какой слой брать под задачу
Проверять сверху вниз и останавливаться на первом совпадении:
Последний пункт применим чаще, чем кажется: агентный цикл вокруг одного вызова добавляет и токены, и задержку.
## Установка и первый вызов
```bash
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"` разрешается в конкретную интеграцию, но сама её не устанавливает — это поиск среди уже установленного.
```python
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
| Параметр | Что делает |
|---|---|
| `temperature` | Разброс ответов. На рассуждающих моделях часто недоступен — часть из них его отклоняет |
| `max_tokens` | Верхняя граница длины ответа |
| `timeout` | Ограничение времени на запрос. Без него длительность ответа определяет провайдер |
| `max_retries` | Повторы на стороне клиента, по умолчанию шесть |
| `rate_limiter` | Ограничитель частоты запросов на стороне клиента |
| `reasoning_effort` | Глубина рассуждения у моделей, которые умеют рассуждать: низкая, средняя, высокая |
Про `rate_limiter` есть нюанс: встроенная реализация `InMemoryRateLimiter` работает **в пределах одного процесса**. При пяти репликах сервиса фактическая частота будет впятеро больше заданной, и для общего на кластер лимита нужен общий ограничитель.
Про `reasoning_effort`: на рассуждающих моделях глубина размышления задаётся этим параметром. Инструкция «думай шаг за шагом» в промпте работает с собственным механизмом рассуждения модели параллельно, а не дополняет его. Тема промптов большая, у меня про неё есть [отдельная статья](/ru/prompt-engineering/).
### Fallbacks — запасные модели
```python
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`.
```python
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` вручную, привязан к конкретному вендору.
### Мультимодальный ввод
```python
from langchain.messages import HumanMessage
message = HumanMessage(content=[
{"type": "text", "text": "Что на этой картинке?"},
{"type": "image", "url": "https://example.com/photo.png"},
])
```
### Обрезка истории
Диалог растёт, контекстное окно конечно. Простой способ — детерминированная обрезка:
```python
from langchain.messages import trim_messages
trimmed = trim_messages(messages, max_tokens=8000, strategy="last", token_counter=model)
```
Более умный способ — суммаризация старой части истории, для agents она доступна готовым middleware. Разница в том, что обрезка предсказуема и теряет старое целиком, а суммаризация сохраняет смысл, но переписывает то, что ассистент помнит.
### Учёт токенов
```python
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`. Это основа для подсчёта расходов и пользовательских квот.
## Промпты
```python
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "Ты помогаешь разобраться в логах сборки. Отвечай коротко."),
("human", "{question}"),
])
chain = prompt | model
```
`ChatPromptTemplate` собирает список messages с подстановкой переменных. Роли при этом сохраняются раздельно — а это существенно: провайдер обрабатывает системную инструкцию и пользовательский ввод по-разному, у них разный вес и разный уровень доверия. Если склеить их в одну строку, текст пользователя окажется там, где модель ожидает ваших инструкций — это и потеря качества, и открытая дорога для инъекции промпта.
Вертикальная черта — композиция: результат левого объекта передаётся правому. Получившаяся цепочка сама умеет `ainvoke`, `astream` и `abatch` и может быть частью чего-то большего.
## Structured output — ответ по схеме
Часто от модели нужен не текст, а данные: категория обращения, извлечённые поля, решение о маршруте.
```python
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 — функция вашего кода, описание которой уходит модели вместе с промптом. Модель сама её не выполняет: она возвращает запрос на вызов с аргументами, вызов выполняет ваш код, результат уходит обратно в модель.
```python
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.** Часть провайдеров другие формы отклоняет.
Когда аннотаций недостаточно — нужен перечислимый набор значений, описание на каждое поле или валидация — задаётся явная схема аргументов:
```python
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` — фреймворк подставит его сам, **и в схему для модели он не попадёт**.
```python
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 возвращает
| Возврат | Результат |
|---|---|
| Строка | Становится содержимым `ToolMessage` |
| Словарь или объект | Сериализуется в message, модель читает поля |
| Список блоков содержимого | Мультимодальный результат — например, текст вместе с картинкой |
| `Command` | Обновляет state агента заодно с ответом на вызов |
| Что угодно при `return_direct=True` | Уходит пользователю без ещё одного хода модели |
```python
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 может сообщать о ходе работы — это попадёт в поток событий:
```python
@tool
def index_documents(folder: str, runtime: ToolRuntime) -> str:
"""Индексирует папку с документами."""
writer = runtime.stream_writer
writer({"type": "progress", "done": 0, "total": 100})
...
```
### Ошибки tools
По умолчанию исключение из tool прекращает весь запуск. Для части случаев это не то, что нужно: неудачный поиск — информация, с которой модель может работать дальше.
```python
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
```python
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
| Параметр | Что принимает |
|---|---|
| `model` | Строку `provider:model` или готовый объект модели |
| `tools` | Список tools. Пустой список допустим — тогда agent вырождается в один вызов модели с обвязкой |
| `system_prompt` | Строку или `SystemMessage`; значение статично |
| `response_format` | Схему ответа; результат приходит в `structured_response` |
| `middleware` | Список middleware |
| `context_schema` | Описание неизменяемых данных запуска: пользователь, арендатор, флаги |
| `state_schema` | Дополнительные поля state, общие для агента и его tools |
| `checkpointer` | Сохранение state; без него памяти между ходами нет |
| `store` | Долговременная память поверх всех threads |
| `name` | Имя, под которым agent виден, когда становится node или subagent |
В примерах модель задают строкой, в рабочем коде её обычно берут из конфигурации вместе с таймаутом, повторами и цепочкой fallbacks.
### Результат — словарь
```python
result["messages"][-1].content # текст ответа
result["structured_response"] # разобранная схема, если задан response_format
```
Возвращается финальное state, а не message, поэтому обращение вида `result.content` даст `AttributeError`.
### Response format — схема внутри цикла
```python
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 — возможности модели
```python
model.profile
# {'max_input_tokens': 400000, 'tool_calling': True, 'reasoning_output': True, 'multimodal': True, ...}
```
Описание возможностей конкретной модели. По нему можно проверить поддержку tools перед их привязкой или свериться с размером контекста — вместо ветвлений по названию модели. Тем же профилем пользуется механизм выбора стратегии для `response_format`.
### Context и state — разные вещи
| | Context | State |
|---|---|---|
| Живёт | Один запуск, неизменяем | Меняется по ходу, попадает в checkpoint |
| Передаётся | Параметром `context` при вызове | Внутри входного словаря |
| Хранит | Пользователя, арендатора, флаги, дескрипторы | Messages, накопленные данные, счётчики |
```python
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 сохраняется и переигрывается, поэтому при возобновлении старого диалога в свежий запуск попал бы вчерашний пользователь.
### Память между ходами
```python
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. При исчерпании поднимается исключение.
```python
result = agent.invoke(payload, config={"recursion_limit": 40})
```
Это защита от бесконечного цикла. Если нужен продуктовый предел — например, не больше пяти поисков за диалог — для этого есть `ModelCallLimitMiddleware` и `ToolCallLimitMiddleware`: они завершают работу штатно, а не исключением.
## Middleware — шесть точек настройки
Middleware — код, который выполняется вокруг шагов агентного цикла. Через него настраивается всё: подмена модели на лету, ограничения, повторы, маскирование персональных данных, пауза на человека, сборка динамического промпта.
| Хук | Когда выполняется |
|---|---|
| `before_agent` | Один раз, до начала цикла |
| `before_model` | Перед каждым вызовом model |
| `wrap_model_call` | Вокруг каждого вызова model |
| `wrap_tool_call` | Вокруг каждого вызова tool |
| `after_model` | После каждого ответа model |
| `after_agent` | Один раз, после завершения цикла |
На схеме это выглядит как слои вокруг цикла:
Форм две. Node-style хуки (`before_*`, `after_*`) получают state и возвращают его обновление — они наблюдают и дополняют. Wrap-style (`wrap_*`) стоят прямо в пути вызова, получают запрос и функцию-продолжение: могут повторить, подменить параметры, вернуть свой результат вместо настоящего.
### Порядок выполнения
Для списка из трёх middleware порядок такой:
Хуки `before_*` идут сверху вниз, `after_*` — снизу вверх, `wrap_*` вкладываются друг в друга, и первое middleware в списке оказывается самым внешним.
Отсюда два практических следствия. Повтор, который должен обернуть всё остальное, ставят первым. Проверка, которая обязана увидеть финальный ответ, тоже ставится ближе к началу списка — на обратном пути она сработает последней.
### Как пишется
Один хук — декоратором:
```python
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 — это класс:
```python
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 или найденных документов, собирается отдельным хуком:
```python
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, затем планируется следующий такт. Перед запуском граф компилируется.
```python
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.
```python
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 целиком:
```python
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 легко тестировать отдельно.
### Маршрутизация
```python
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`:
```python
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 — параллельный запуск
Когда количество ветвей известно только в рантайме:
```python
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 другого графа:
```python
builder.add_node("research", research_graph)
```
Общие ключи state перетекают автоматически. Если схемы state разные, нужен node-переходник, который преобразует данные на входе и на выходе.
### Управление на уровне node
```python
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())
```
| Настройка | Для чего | Что учесть |
|---|---|---|
| `retry_policy` | Временные сбои: сеть, ответы 429 и 5xx | Повтор перезапускает **весь node с начала**, поэтому node должен быть идемпотентным |
| `cache_policy` | Пропуск пересчёта при одинаковом входе | Кеш передаётся при компиляции |
| `timeout` | Ограничение по общему времени и по простою | Появился в версии 1.2 |
| `error_handler` | Выполняется, когда повторы исчерпаны: компенсация, откат | Появился в версии 1.2 |
Оттуда же — `RunControl.request_drain()`: кооперативная остановка, которая просит запуск остановиться в ближайшей безопасной точке и оставляет checkpoint для продолжения. Это штатный способ пережить выкатку новой версии или остановку контейнера посреди работы.
`recursion_limit` у графа такой же, как у агента, и задаётся в конфигурации запуска. Есть и специальное поле state `RemainingSteps` — node может посмотреть в него и свернуть работу заранее, вместо того чтобы упереться в предел.
## Checkpointer и store: память и устойчивость
Памяти две, и они решают разные задачи.
| | Checkpointer | Store |
|---|---|---|
| Область действия | Один thread, один диалог | Все threads сразу |
| Хранит | State графа на каждом super-step | Произвольные документы вашего формата |
| Даёт | Память между ходами, продолжение после сбоя, interrupts, перемотку | Предпочтения пользователя, накопленные факты, общее знание |
| Ключ | `thread_id` плюс идентификатор checkpoint | Кортеж namespace плюс ключ |
Ни то ни другое по умолчанию не включено.
### Checkpointers
```python
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) # видит первый ход
```
| Реализация | Пакет | Назначение |
|---|---|---|
| `InMemorySaver` | `langgraph-checkpoint` | Тесты и разработка; исчезает при перезапуске |
| `SqliteSaver` | `langgraph-checkpoint-sqlite` | Локальная разработка с сохранением между запусками |
| `PostgresSaver` / `AsyncPostgresSaver` | `langgraph-checkpoint-postgres` | Продакшен; в асинхронном сервисе — асинхронная форма |
Две эксплуатационные детали. Метод `.setup()`, создающий таблицы, — операция уровня выкатки, а не старта приложения: изменение схемы на старте при нескольких репликах выполняется параллельно само с собой. И `thread_id` для Postgres держат короче 255 символов — колонка ограничена.
### Просмотр истории и перемотка
```python
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`:
```python
from langgraph.types import Overwrite
await graph.aupdate_state(config, {"items": Overwrite(["C"])})
```
### Хранение checkpoints
Checkpoints создаются по одному на super-step для каждого thread и сами не удаляются. Для долгоживущего продукта это означает постоянный рост таблицы, поэтому политику хранения — сколько времени thread остаётся возобновляемым и что удаляет остальное — стоит определить заранее.
### Store
```python
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, а не через глобальную переменную:
```python
def recall(state, runtime: Runtime):
item = runtime.store.get(("users", runtime.context.user_id), "preferences")
...
```
### Durability — режимы записи
Настройка того, как часто state записывается на диск. Передаётся при вызове:
| Режим | Поведение | Что теряется при сбое |
|---|---|---|
| `"sync"` | Записано до начала следующего шага | Ничего; самый медленный |
| `"async"` | Записывается параллельно следующему шагу | Небольшое окно — последний checkpoint |
| `"exit"` | Только при завершении: успех, ошибка или interrupt | Всё промежуточное; самый быстрый |
```python
await graph.ainvoke(payload, config=config, durability="async")
```
Выбор зависит от цены потерянного шага: если шаги отправляют письма или проводят платежи, подходит `"sync"`; для аналитического прогона по длинному диалогу достаточно `"exit"`.
### Переигрывание при возобновлении
Общее правило рантайма: возобновление **выполняет node с самого начала**, пропускается только завершённая работа, попавшая в checkpoint. Отсюда два следствия:
- Побочные эффекты, выполненные до точки остановки, произойдут ещё раз.
- Недетерминированные значения — время, случайные идентификаторы — при переигрывании получатся другими. Их вычисляют в отдельном node, чтобы значение попало в checkpoint и подставлялось оттуда.
## Human-in-the-loop: пауза на человека
Иногда запуск должен остановиться и дождаться человека: подтвердить отправку, поправить черновик, ответить на уточняющий вопрос. Для этого есть `interrupt`.
```python
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:
```python
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` пропускает вызов без вопросов, словарь настраивает, какие решения доступны проверяющему.
| Решение | Что происходит |
|---|---|
| `approve` | Вызов выполняется как предложен |
| `edit` | Выполняется с изменёнными аргументами; в `edited_action` передаются и `name`, и `args` |
| `reject` | Не выполняется, а текст отказа уходит модели как результат вызова |
| `respond` | Текст человека возвращается вместо результата tool |
```python
await agent.ainvoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config, version="v2")
```
Когда модель за один ход запрашивает несколько защищённых tools, они предъявляются вместе, и список решений должен идти **в том же порядке**, что и предъявленные действия.
Interrupt можно включать по условию — параметром `when`: функция смотрит на аргументы вызова и решает, нужно ли подтверждение. Так проверяются только рискованные случаи, например запись за пределы рабочего каталога, а остальные проходят без вопросов.
### Побочные эффекты и точка остановки
Здесь то же правило переигрывания, но с наглядными последствиями:
```python
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`. Подходит для большинства задач.
| `stream_mode` | Что отдаёт |
|---|---|
| `values` | Всё state после каждого шага |
| `updates` | Только то, что изменил каждый node |
| `messages` | Токены модели по мере поступления вместе с метаданными |
| `custom` | То, что node написал сам через stream writer |
| `checkpoints` | События сохранения state (нужен checkpointer) |
| `tasks` | Старты и завершения задач с результатами и ошибками (нужен checkpointer) |
| `debug` | Checkpoints, tasks и метаданные вместе |
Режимы можно запрашивать по несколько сразу, списком.
### Версия формата потока
У формата есть версия, и по умолчанию действует `v1`, где форма ответа зависит от вызова: один режим отдаёт сырые данные, несколько режимов — кортежи `(mode, data)`, а включённые subgraphs добавляют в кортеж namespace.
Версия `v2` отдаёт всё единообразно — словарь `StreamPart` с полями `type`, `ns` и `data`:
```python
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`:
```python
classifier = model.with_config({"tags": ["nostream"]})
```
Тогда её токены не эмитятся вовсе.
### Subgraphs в потоке
```python
async for chunk in graph.astream(payload, stream_mode="messages", subgraphs=True, version="v2"):
...
```
Без `subgraphs=True` токены, порождённые внутри вложенного графа — включая agent, поставленный node, — в поток не попадают. Ошибки при этом нет, поток просто оказывается неполным. Поле `ns` в каждом фрагменте показывает источник: пустой кортеж для корневого графа, имя node с идентификатором задачи для вложенного.
### Custom events
```python
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` в графе сообщает о старте, о фрагментах и о завершении.
```python
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.
```python
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 делегирует работу, ничего не зная о внутреннем устройстве.
```python
@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 перетекают в обе стороны |
| Кто решает | Модель — когда делегировать | Граф — детерминированно |
| Стоимость | Полный агентный цикл на каждое делегирование | Работа одного node |
| Видимость | Только в трассировке | На отрисованной схеме |
Определяющее свойство — изоляция контекста. Делегирование subagent уместно, когда подзадача иначе заполнит родительский контекст деталями, которые дальше не нужны: чтение длинного документа, шумный поиск. Subgraph уместен, когда шаг — часть процесса и его state важно для последующих шагов.
Про стоимость: каждый subagent выполняет свой цикл со своим системным промптом и своими описаниями tools, поэтому запуск пяти subagents расходует примерно как пять агентов, а не как один.
## Deep Agents
Готовая сборка middleware поверх обычного агента, рассчитанная на длинные задачи, которые не помещаются в одно контекстное окно.
```python
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 и функциональные вызовы](/ru/rag-and-function-calling/) и [статья про векторные базы](/ru/vector-databases/). Здесь — то, что касается стека.
Конвейер состоит из трёх стадий, и каждая тестируется отдельно:
Индексация выполняется отдельно и заранее; поиск и генерация — на каждый запрос пользователя.
Архитектур две. Классическая двухшаговая: всегда искать, потом отвечать — предсказуемая задержка, меньше вызовов модели, подходит для узкого корпуса, где поиск нужен на каждый вопрос. Agentic RAG: поиск оформлен tool, и модель сама решает, нужен ли он, когда и с каким запросом — переменная задержка, больше ходов, подходит для открытых вопросов и нескольких источников. В первой версии agentic описан как основной вариант.
```python
@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:**
```python
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:**
```python
from langchain.embeddings import init_embeddings
embeddings = init_embeddings("openai:text-embedding-3-small")
```
Одна и та же модель и одна и та же размерность должны использоваться и при индексации, и при запросах. Если смешать, расстояния в хранилище потеряют смысл, и ошибки при этом не будет.
**Vector stores:** `InMemoryVectorStore` для тестов, Chroma для локальной разработки, pgvector рядом с существующим Postgres, а также Pinecone, Qdrant и Weaviate отдельными пакетами.
**Retrievers:**
```python
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 включается переменными окружения и не требует правок в коде:
```bash
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=<ключ>
LANGSMITH_PROJECT=<проект>
```
После этого каждый запуск LangChain и LangGraph попадает в трассировку: nodes, вызовы моделей, tools, токены, задержки. Имена переменных с прежним префиксом `LANGCHAIN_` больше не работают.
К запускам стоит добавлять теги и метаданные — по ним трассы потом ищутся:
```python
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** (оценивающие функции). Результат прогона называется экспериментом, и его можно сравнивать с предыдущими.
```python
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 и пути ошибок:
```python
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 без управляющего слоя.
Конфигурация проекта описывается одним файлом:
```json title="langgraph.json"
{
"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`:
| Команда | Что делает |
|---|---|
| `langgraph new` | Создаёт проект из шаблона |
| `langgraph dev` | Локальный сервер с горячей перезагрузкой, без Docker, порт 2024 |
| `langgraph up` | Поднимает всё локально в Docker вместе с Postgres |
| `langgraph build` | Собирает образ сервера |
| `langgraph deploy` | Собирает образ, отправляет его в реестр и создаёт или обновляет развёртывание |
| `langgraph dockerfile` | Выдаёт Dockerfile без сборки |
```bash
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` и реэкспорты интеграций сообщества. Это не «устарело, но работает» — импорт из основного пакета завершается ошибкой.
```python
# 0.x
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`, ответ по нерелевантным кускам текста. Поэтому трассировка запусков здесь — базовая часть работы, а не дополнительная возможность.