Свой цикл или фреймворк: карта агентных фреймворков

Зачем это нужно

В уроке 8.3.1 агентный цикл занял 40 строк, в ДЗ s8-hw02 к нему добавились реестр инструментов и защита от инъекций. Теперь поставьте этот цикл в продакшен — ассистентом онлайн-школы, который отвечает на вопросы о курсах и записывает на потоки. Сразу появляются требования:

• записать на поток можно только после «да» от человека, а «да» приходит через десять минут, когда процесс уже перезапущен;
• сервис упал на седьмом шаге из девяти — нужно продолжить с седьмого и не записать ученика дважды;
• интерфейс показывает промежуточные шаги, а не 40 секунд тишины;
• у каждого запроса есть трасса, лимит шагов и бюджет.

Каждое требование — от 50 до 300 строк кода, и через месяц у вас самодельный фреймворк без документации. Или вы берёте готовый и платите зависимостями, абстракциями, которые прячут промпты, обновлениями с поломками и привязкой к вендору. Этот урок — карта для такого решения: что делает фреймворк, какие фреймворки есть в октябре 2026 года, как выбирать и как всё устроено внутри (движок графа состояний вы напишете сами). Сначала руками, потом фреймворк.

Карта модуля 8.5

Урок	О чём	Что останется у вас
8.5.1 (этот)	что делает фреймворк, шаблоны Anthropic, карта фреймворков, граф состояний изнутри	свой мини-движок графа
8.5.2 LangGraph	граф состояний, checkpointer и потоки, подтверждение человеком, стриминг, подграфы	ассистент школы как граф с маршрутизатором, RAG-узлом и действием с подтверждением
8.5.3 Agent SDK	OpenAI Agents SDK и Claude Agent SDK	тот же сценарий в двух SDK и сравнение с LangGraph
8.5.4 Мультиагентные системы в продакшене	надёжность, наблюдаемость, оценка траекторий, безопасность	чек-лист надёжности
8.5.5 Практикум	от спецификации до сервиса	мультиагентный сервис и ДЗ s8-hw03-orchestration

В 8.5.5 вы соберёте сервис целиком: спецификация → граф агентов (маршрутизатор, исследователь с RAG из 8.2, исполнитель действий с подтверждением, проверяющий) → инварианты домена в коде инструментов → регрессионные сценарии с фейковой моделью → FastAPI со стримингом → трассы и стоимость → нагрузка. На 4×RTX 4080 — две модели через vLLM: маленькая для маршрутизации, большая для ответов. Эталонный граф и тесты — в ДЗ s8-hw03-orchestration.

Сквозной пример модуля — онлайн-школа из программ DDD и SDD: потоки (Cohort), записи (Enrollment) со статусами PENDING_PAYMENT, CONFIRMED, CANCELLED, лист ожидания, оплата. Действия ассистента меняют данные и требуют подтверждения человеком. Правила домена — инвариант вместимости потока, возврат при отмене за 7 суток и более до старта — проверяет код инструментов, а не модель. Это идея урока DDD.4.4: правило в промпте — просьба, правило в агрегате — гарантия.

Что фреймворк берёт на себя

Вспомните цикл из 8.3.1: for step in range(max_steps), внутри — вызов модели, разбор ответа, вызов инструмента. Состояние агента — список messages в локальной переменной. Он живёт, пока живёт процесс. Вот что фреймворки добавляют поверх такого цикла.

Задача	В цикле из 8.3.1	Что даёт фреймворк	Где в модуле
Состояние	список messages в переменной	общее состояние с правилами слияния	этот урок, 8.5.2
Ветвления и циклы	if внутри while	явный граф или передачи управления	этот урок, 8.5.2–8.5.3
Сохранение и возобновление	нет: процесс упал — всё пропало	чекпоинты после каждого шага, потоки	этот урок, 8.5.2
Человек в контуре	input() блокирует процесс	пауза, которая переживает перезапуск	8.5.2–8.5.3
Стриминг	ответ целиком в конце	поток событий: токены, шаги, вызовы инструментов	8.5.2, 8.5.5
Трассировка	print	спаны вызовов модели и инструментов, OpenTelemetry	8.5.3–8.5.4
Ретраи и лимиты	try/except и max_steps	повторы на узел, таймауты, бюджеты	8.5.4
Параллельность	всё по очереди	параллельные ветви и слияние их результатов	8.5.4

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

Пять шаблонов Anthropic: workflow и агент

Самый цитируемый текст о проектировании агентных систем — статья Anthropic «Building effective agents» (декабрь 2024). Её вывод из работы с десятками команд: самые успешные реализации строились «на простых компонуемых шаблонах, а не на сложных фреймворках» (цитата в нашем переводе). Сверху статьи теперь стоит пометка, что инструменты с тех пор сильно изменились, и список фреймворков в ней обновлён. Сами шаблоны не устарели.

Статья делит агентные системы на два вида (различие вы знаете по 8.3.1, здесь — точные формулировки):

• workflowworkflow (конвейер с заданными шагами)EN workflowПо определению Anthropic («Building effective agents»): система, где LLM и инструменты связаны заранее заданными путями в коде. В агенте путь выбирает сама модель. Обе разновидности Anthropic называет агентными системами. — LLM и инструменты связаны заранее заданными путями в коде;
• агент — LLM сама направляет процесс и выбирает инструменты.

Строительный блок обоих — дополненная LLMдополненная LLMEN augmented LLMБазовый строительный блок агентных систем: модель, у которой есть поиск (RAG), инструменты и память. Она сама формулирует запросы, выбирает инструменты и решает, что запомнить.: модель с поиском (8.2), инструментами (8.3.2) и памятью (8.3.3). Из таких блоков собирают пять шаблонов workflow; для каждого — пример из нашей школы.

1. цепочка промптовцепочка промптовEN prompt chainingШаблон workflow: задача разбита на фиксированные шаги, каждый вызов LLM обрабатывает выход предыдущего; между шагами — программные проверки («ворота»). Больше задержка, выше точность.. Фиксированные шаги, каждый вызов обрабатывает выход предыдущего, между шагами — программные проверки («ворота», gate). Платим задержкой за точность: каждый вызов решает задачу попроще. Школа: письмо о переносе потока — черновик → проверка кодом (есть дата и ссылка?) → перевод для иностранцев.
2. Маршрутизациямаршрутизация (шаблон workflow)EN routingКлассификатор (LLM или обычная модель) определяет тип входа и отправляет его в специализированную ветку со своим промптом, инструментами или моделью.. Вход классифицируется и уходит в специализированную ветку. Так оптимизация промпта под один тип запросов не портит другие; простые вопросы можно отдать маленькой модели. Школа: вопрос о курсе → RAG, «запишите меня» → ветка действий, «верните деньги» → живой оператор.
3. Параллелизацияпараллелизация (шаблон workflow)EN parallelizationНесколько вызовов LLM работают одновременно, а код сводит результаты. Два варианта: секционирование (независимые подзадачи) и голосование (одна задача несколько раз).. Секционирование — независимые подзадачи параллельно (пример из статьи: один вызов отвечает, другой проверяет вход на недопустимое, и это работает лучше одного вызова, который делает всё). Голосование — одна задача несколько раз. Школа: три проверяющих решают, публиковать ли отзыв.
4. оркестратор и исполнителиоркестратор и исполнителиEN orchestrator-workersЦентральная LLM сама разбивает задачу на подзадачи, раздаёт их исполнителям и сводит ответы. В отличие от параллелизации, подзадачи не известны заранее.. Центральная LLM сама разбивает задачу, раздаёт части исполнителям и сводит результат. В отличие от параллелизации, подзадачи не известны заранее. Школа: «подберите программу из трёх курсов под мою цель». Стоимость этой схемы — в уроке 8.3.3.
5. оценщик и оптимизатороценщик и оптимизаторEN evaluator-optimizerОдин вызов LLM генерирует ответ, другой оценивает его и даёт обратную связь; цикл повторяется, пока оценщик не одобрит или не кончится лимит шагов.. Один вызов генерирует, другой оценивает и даёт обратную связь — в цикле. Нужны ясные критерии и измеримая польза от итераций. Школа: описание курса переписывается, пока проверяющий не подтвердит требования. Этот цикл «исполнитель — проверяющий» вы реализуете в задании.

flowchart LR
subgraph C["1. Цепочка"]
  c1["LLM"] --> cg{"ворота"} --> c2["LLM"]
end
subgraph R["2. Маршрутизация"]
  r0["классификатор"] --> r1["ветка A"]
  r0 --> r2["ветка B"]
end
subgraph P["3. Параллелизация"]
  p1["LLM"] --> pa["агрегатор"]
  p2["LLM"] --> pa
end
subgraph O["4. Оркестратор"]
  o0["оркестратор"] -.-> o1["исполнитель"]
  o0 -.-> o2["исполнитель"]
  o1 --> os["синтез"]
  o2 --> os
end
subgraph E["5. Оценщик"]
  e1["генератор"] --> e2{"оценщик"}
  e2 -->|"доработать"| e1
end

Пять шаблонов workflow: пунктир — подзадачи, которые оркестратор выбирает сам во время работы

Агент в статье — модель, которая в цикле пользуется инструментами и на каждом шаге получает «правду» от среды (результат вызова, вывод тестов); условия остановки обязательны. Агенты нужны, где число шагов нельзя предсказать. Цена — стоимость и накопление ошибок ($p^n$ из 8.3.3), поэтому — тесты в песочнице и guardrails.

Совет «начинайте с простого»

• Усложняйте, только когда это измеримо улучшает результат; часто хватает одного вызова LLM с поиском и примерами.
• За слоями абстракции фреймворков не видно промптов и ответов: отлаживать труднее, и тянет усложнять.
• Начинайте с прямых вызовов API. Берёте фреймворк — разберитесь, что внутри: «неверные представления о том, что под капотом, — частая причина ошибок» (цитата в нашем переводе). На пути к продакшену не бойтесь убирать слои абстракции.

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

Карта фреймворков на октябрь 2026

Семь вариантов из программы модуля. Факты — из README репозиториев, версии и даты — из PyPI (https://pypi.org/pypi/<пакет>/json) на 6 октября 2026. Последний столбец — число версий на PyPI за 12 месяцев: это сигнал и активности проекта, и того, как часто придётся обновлять свой код.

Фреймворк	Ключевая абстракция	Сильные стороны	Цена и риски	Лицензия	Версия (дата)	Релизов за год
LangGraph	граф состояний	контроль потока, чекпоинты, пауза для человека, стриминг	многословность, экосистема LangChain	MIT	1.2.14 (06.10.2026)	51
OpenAI Agents SDK	Agent + Runner, handoffs	мало кода, guardrails, сессии, трассы из коробки	версия 0.x, трассы по умолчанию уходят в OpenAI	MIT	0.23.1 (02.10.2026)	87
Claude Agent SDK	цикл Claude Code как библиотека	файлы и терминал, хуки, права, субагенты, MCP	официально только Claude, статус Alpha	MIT + условия Anthropic	0.2.163 (30.09.2026)	146
AutoGen	диалог агентов	—	режим поддержки	MIT	0.7.5 (30.09.2025)	0
AG2	Agent + Network	открытое сообщество	1.0 несовместима со старым кодом	Apache-2.0	1.1.2 (03.10.2026)	33
Microsoft Agent Framework	агенты + графовые workflow	чекпоинты, OpenTelemetry, Python и .NET	тяготение к Azure, много пакетов	MIT	1.20.0 (02.10.2026)	60
CrewAI	команда ролей (Crews) + Flows	быстрый прототип	тяжёлые зависимости, телеметрия по умолчанию	MIT	1.15.23 (28.09.2026)	293
smolagents	агент пишет действия кодом	минимализм, любые модели	нужна песочница, нет чекпоинтов	Apache-2.0	1.26.0 (29.05.2026)	4
PydanticAI	типизированный агент	типы, внедрение зависимостей, тестовая модель	частые релизы, мажорная 2.0 в июне	MIT	2.54.0 (03.10.2026)	199

Что не помещается в таблицу:

• LangGraph — «низкоуровневый фреймворк оркестрации для долгоживущих агентов с состоянием»; вдохновлён Pregel (система Google для обработки графов) и Apache Beam, работает без LangChain. Отладка и деплой — в коммерческом, но необязательном LangSmith. Версия 1.0 — 17.10.2025.
• OpenAI Agents SDK: агент — инструкции, инструменты, guardrails и передачи управленияпередача управления (handoff)EN handoffОдин агент передаёт диалог другому, более подходящему агенту вместе с историей. Ключевая абстракция OpenAI Agents SDK; шаблон есть и в Microsoft Agent Framework.; заявлена работа с сотней моделей и OpenAI-совместимыми адресами, включая ваш vLLM. Но трассировка включена по умолчанию и отправляет трассы на серверы OpenAI — с другим провайдером её выключают (set_tracing_disabled(True) или OPENAI_AGENTS_DISABLE_TRACING=1) или ставят свой обработчик.
• Claude Agent SDK: query() и ClaudeSDKClient; сразу доступны инструменты Claude Code (Read, Write, Edit, Bash…), свои — функции Python, оформленные как MCP-сервер в процессе. Хуки (PreToolUse и др.) могут запретить вызов; есть права, субагенты, max_turns, max_budget_usd. В пакет вшит CLI Claude Code. Файл лицензии — MIT, но README оговаривает: использование регулируется коммерческими условиями Anthropic.
• AutoGen → AG2 и Agent Framework — пример того, как за год меняется почва под ногами. README AutoGen отправляет новых пользователей в Microsoft Agent Framework: агенты плюс графовые workflow (sequential, concurrent, handoff, group collaboration), чекпоинты, OpenTelemetry. AG2 — ответвление AutoGen, его ведут добровольцы, администраторы проекта — соавторы исходной статьи AutoGen; AG2 1.0 (27.07.2026) — по сути новый фреймворк (import ag2, сеть агентов с хабом и каналами), а классические ConversableAgent и GroupChat уехали в пакет ag2-classic.
• CrewAI: Crews — агенты с ролью, целью и предысторией; Flows — событийные workflow с состоянием. Анонимная телеметрия включена по умолчанию (выключает OTEL_SDK_DISABLED=true).
• smolagents (Hugging Face): CodeAgent пишет действия кодом на Python; логика агентов — около 1000 строк. Исполнять код модели — только в песочнице (README называет Blaxel, E2B, Modal и Docker).
• PydanticAI: output_type гарантирует проверенный результат, зависимости приходят через RunContext, модель меняется строкой 'провайдер:модель'; долговечное выполнение — через Temporal, DBOS, Prefect и другие движки.

Таблица стареет быстрее урока

За год AutoGen ушёл в режим поддержки, AG2 переписал API, у PydanticAI вышла мажорная версия. Перед выбором проверьте версию сами: curl -s https://pypi.org/pypi/langgraph/json | python -c "import json,sys; print(json.load(sys.stdin)['info']['version'])", и закрепляйте точные версии (langgraph==1.2.14), иначе обновление сломает агента в пятницу вечером.

Как выбрать

Пять критериев и способ проверить каждый за полчаса:

Критерий	Вопрос	Как проверить
Контроль над потоком	Можно ли задать шаги явно и протестировать узел отдельно?	пример с ветвлением и юнит-тестом
Привязка к провайдеру	Работает ли с vLLM или Ollama из 7.2.2? Куда уходят трассы?	пример с base_url на localhost без интернета
Наблюдаемость	Есть ли OpenTelemetry без платного облака?	поиск по README и docs
Зрелость	Версия ≥ 1.0? Темп релизов, руководства по миграции?	PyPI, GitHub Releases, CHANGELOG
Лицензия и условия	MIT или Apache? Телеметрия? Коммерческие условия?	LICENSE, разделы Telemetry и License

И два вопроса, которые в продакшене важнее остальных: можно ли подменить модель в тестах (фейковая модель со сценарием, как ScriptedLLM из s8-hw02) и где хранятся контрольные точки — в вашей базе или в чужом облаке.

flowchart TD
A["Задача"] --> B{"Хватит одного вызова LLM<br/>с RAG и примерами?"}
B -->|"да"| B1["Без агентов: API напрямую<br/>(уроки 8.1–8.2)"]
B -->|"нет"| C{"Нужны паузы на ответ человека,<br/>возобновление после сбоя,<br/>параллельные ветви?"}
C -->|"нет: один агент, до 5–10 инструментов,<br/>всё в одном запросе"| D["Свой цикл на 100 строк<br/>(8.3.1, s8-hw02)"]
C -->|"да"| E{"Что важнее всего?"}
E -->|"явный поток шагов,<br/>ветвления, циклы"| F["LangGraph<br/>(в .NET и Azure — Agent Framework)"]
E -->|"агенты передают<br/>друг другу диалог"| G["OpenAI Agents SDK"]
E -->|"агент сам работает<br/>с файлами и терминалом"| H["Claude Agent SDK<br/>(если подходит Claude)"]
E -->|"типизированный результат,<br/>свой движок долговечности"| I["PydanticAI<br/>(+ Temporal или DBOS)"]
E -->|"роли и задачи,<br/>быстрый прототип"| J["CrewAI"]
E -->|"действия кодом,<br/>исследования"| K["smolagents<br/>(+ песочница)"]

Схема выбора: сначала решите, нужен ли фреймворк вообще, потом — какой

Когда хватит своего цикла на 100 строк

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

Граф состояний изнутри

граф состоянийграф состоянийEN state graphСпособ описать агентную систему как конечный автомат над общим состоянием: узлы — функции «состояние → частичное обновление», рёбра — переходы (обычные или условные). Основная абстракция LangGraph. — ключевая абстракция LangGraph; те же идеи (узлы, рёбра, общее состояние, контрольные точки) есть в Agent Framework, во Flows у CrewAI и в Pydantic Graph. Разберём их на устройстве LangGraph.

Состояние, узлы, рёбра

• Состояние — словарь с заранее объявленной схемой (TypedDict). Все узлы читают одно общее состояние.
• Узел — обычная функция: получает состояние, возвращает частичное обновление (только изменённые ключи). Внутри — вызов модели, инструмента или просто код.
• Ребро — куда идти после узла. Обычное ребро ведёт всегда в одно место. Условное вызывает функцию-маршрутизатор, которая смотрит на состояние и возвращает имя следующего узла. Именно здесь «решает модель», если маршрутизатор читает её ответ.
• START и END — служебные точки входа и выхода.

Конечный автомат над общим состоянием

Положение выполнения описывается парой: текущий узел $v_t$ (их конечное число — как у конечного автомата) и данные $S_t$ (словарь любого размера). Один шаг:

$$S_{t+1} = S_t \oplus f_{v_t}(S_t), \qquad v_{t+1} = \delta(v_t, S_{t+1}).$$

Узел $v_t$ считает обновление $f_{v_t}(S_t)$, оно сливается с состоянием по правилам слияния ($\oplus$), а функция переходов $\delta$ выбирает следующий узел. Пример: $S =$ {"messages": ["задание"], "attempts": 0}, узел writer возвращает {"attempts": 1, "messages": ["черновик v1"]}. После слияния $S =$ {"messages": ["задание", "черновик v1"], "attempts": 1}, а обычное ребро ведёт в reviewer.

Чем эта формализация полезна:

1. Всё положение выполнения — пара $(v, S)$. Сохранили её — можно продолжить с того же места. Это и есть контрольная точка.
2. Тестируемость. Узел и маршрутизатор — обычные функции от состояния, их проверяют юнит-тестами без модели.
3. Workflow и агент живут в одном графе. Где рёбра обычные — путь задан кодом, где маршрутизатор читает ответ модели — решает модель. Степень «агентности» вы выбираете ребро за ребром.

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

Редьюсеры

По умолчанию обновление перезаписывает ключ. Но новое сообщение должно дописаться к истории, а не заменить её. редьюсерредьюсер (правило слияния состояния)EN reducerФункция (старое значение, обновление) → новое значение для одного ключа состояния графа. Без редьюсера обновление перезаписывает ключ; с operator.add списки сообщений дописываются. задаёт правило слияния для ключа — функцию (старое значение, обновление) → новое:

import operator
from typing import Annotated, TypedDict

class School(TypedDict):
    messages: Annotated[list, operator.add]   # редьюсер: ["a"] + ["b"] = ["a", "b"]
    intent: str                               # без редьюсера: новое значение заменяет старое

Так объявляют редьюсеры и в LangGraph, и в вашем движке. Редьюсер может быть любым: max для лучшей оценки, объединение множеств для документов; в LangGraph есть готовый add_messages. Особенно важны редьюсеры для параллельных ветвей: без правила слияния результат зависел бы от того, какая ветвь закончила последней.

Лимит шагов

Проверяющий может никогда не одобрить, агент — ходить по кругу. Поэтому в вашем движке не больше recursion_limit выполнений узлов за один вызов (по умолчанию 25), дальше GraphRecursionError. В LangGraph параметр называется так же и передаётся в config. Два нюанса, проверенные на langgraph 1.2.14: значение по умолчанию там теперь 10007 (константа DEFAULT_RECURSION_LIMIT, в langchain_core и в ранних версиях LangGraph было 25), а считаются «супершаги», поэтому линейному графу из двух узлов нужен recursion_limit=3. Вывод: ставьте лимит явно и с запасом, а настоящую защиту дают бюджеты токенов и денег (8.5.4).

Контрольные точки и возобновление

Контрольная точкаконтрольная точка графа (checkpoint)EN checkpointСнимок выполнения графа после шага: состояние, следующий узел, номер шага — в потоке с идентификатором thread_id. Позволяет продолжить после сбоя, поставить граф на паузу до ответа человека и вернуться к любому прошлому шагу. — снимок $(S, v_{\text{след}}, \text{шаг})$, который движок сохраняет под идентификатором потока thread_id: перед первым узлом и после каждого узла. Возобновление — взять последний снимок и выполнить узел, на который он указывает.

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

1. Побочные эффекты узлов должны быть идемпотентными. Узел записал ученика на поток и упал до чекпоинта — повтор запишет ещё раз. Защита та же, что у идемпотентного получателя из DDD.4.2: ключ идемпотентности, по которому инструмент узнаёт повтор. В практике вы увидите это вживую.
2. Хранилище чекпоинтов должно пережить процесс: память — для тестов, в продакшене — SQLite или Postgres (у LangGraph это пакеты langgraph-checkpoint-sqlite и langgraph-checkpoint-postgres).
3. Человек в контуре — это пауза. Граф остановился перед действием, чекпоинт хранит всё нужное; человек ответил через час, процесс уже другой — граф продолжает с того же места.
4. История чекпоинтов — «путешествие во времени»: вернуться к любому шагу, поправить состояние, пойти по другой ветке (урок 8.5.2).

Поток (thread_id) — ещё и память диалога: новый ввод в тот же поток сливается с его состоянием через редьюсеры.

Ваш движок выполняет узлы по одному. LangGraph, как Pregel, работает супершагами: все готовые узлы выполняются параллельно, обновления сливаются редьюсерами, затем пишется чекпоинт. Если одна параллельная ветка упала, результаты успешных сохраняются как «отложенные записи» (pending writes) и не пересчитываются — мы это проверили.

flowchart LR
S(["START"]) --> CL["classify"]
CL -->|"question"| AN["answer:<br/>RAG по каталогу"]
CL -->|"enroll"| PR["propose:<br/>записать? ответьте «да»"]
CL -->|"confirm"| EX["execute:<br/>инструмент enroll"]
CL -->|"decline"| DR["drop"]
AN --> E(["END"])
PR --> E
EX --> E
DR --> E
PR -. "чекпоинт: pending = действие;<br/>ждём следующего ввода в поток" .-> CL

Граф ассистента школы из практики: подтверждение человеком — это состояние потока между двумя вызовами

Практика

Задание: мини-движок графа состояний

Движок графа состояний на чистом Python: узлы, обычные и условные рёбра, редьюсеры, лимит шагов, чекпоинты после каждого узла и возобновление. Имена API специально сделаны как в LangGraph: StateGraph(схема), add_node, add_edge, add_conditional_edges(источник, маршрутизатор, path_map), compile(checkpointer=...), invoke(ввод, config), get_state(config), START, END и config={"configurable": {"thread_id": ...}}. В 8.5.2 вы узнаете каждое имя и будете знать, что за ним стоит. Тесты проверяют маршрутизацию, цикл «исполнитель — проверяющий», лимит шагов, чекпоинты, восстановление после сбоя и потоки.

Мини-движок графа состояний8 тестов

"""Мини-движок графа состояний: то, что LangGraph делает внутри, на чистом Python (урок 8.5.1).

Граф состояний — конечный автомат над общим состоянием:
  * состояние — словарь; его схема — TypedDict, а у ключа может быть редьюсер:
        messages: Annotated[list, operator.add]   # обновления дописываются в список, а не заменяют его;
  * узел — функция state -> dict с ЧАСТИЧНЫМ обновлением (только изменённые ключи);
  * ребро — куда идти после узла: обычное (всегда в одно место) или условное (решает функция-маршрутизатор);
  * START и END — служебные точки входа и выхода.
Движок выполняет узлы по одному, после каждого сохраняет контрольную точку и умеет продолжить с места остановки.
Имена API — как в LangGraph (урок 8.5.2): add_node, add_edge, add_conditional_edges, compile, invoke, get_state.
"""
import copy
import operator
import typing

START = "__start__"
END = "__end__"
DEFAULT_RECURSION_LIMIT = 25


class GraphRecursionError(RuntimeError):
    """Граф сделал больше шагов, чем разрешает recursion_limit, — скорее всего, зациклился."""


def reducers_from_schema(schema):
    """По TypedDict возвращает (множество ключей, {ключ: редьюсер}). Готово, менять не нужно.

    Редьюсер — последняя функция в Annotated[тип, функция]: так его объявляют и в LangGraph.
    """
    hints = typing.get_type_hints(schema, include_extras=True)
    reducers = {}
    for key, hint in hints.items():
        if typing.get_origin(hint) is typing.Annotated:
            funcs = [m for m in hint.__metadata__ if callable(m)]
            if funcs:
                reducers[key] = funcs[-1]
    return set(hints), reducers


class InMemorySaver:
    """Хранилище контрольных точек в памяти. Готово, менять не нужно.

    Для каждого потока (thread_id) — список чекпоинтов по порядку. Чекпоинт — словарь
    {"values": состояние, "next": имя следующего узла (или END), "step": сколько узлов выполнено в потоке}.
    put(thread_id, cp) — дописать; get(thread_id) — последний (или None); list(thread_id) — все.
    Хранит глубокие копии: если код потом поменяет словарь состояния, сохранённое прошлое не изменится.
    """

    def __init__(self):
        self.storage = {}

    def put(self, thread_id, checkpoint):
        self.storage.setdefault(thread_id, []).append(copy.deepcopy(checkpoint))

    def get(self, thread_id):
        history = self.storage.get(thread_id)
        return copy.deepcopy(history[-1]) if history else None

    def list(self, thread_id):
        return copy.deepcopy(self.storage.get(thread_id, []))


def apply_update(state, update, reducers, keys=None):
    """Новое состояние = старое + частичное обновление узла. Исходный словарь state НЕ меняется.

    Для каждой пары key, value из update (update может быть None — тогда ничего не меняется):
      * если keys задан и key в него не входит — ValueError (узел опечатался в имени ключа);
      * если у key есть редьюсер и key уже есть в состоянии — new[key] = reducers[key](старое, value);
      * иначе — new[key] = value (перезапись).

    >>> apply_update({"messages": ["a"], "intent": "q"}, {"messages": ["b"], "intent": "x"}, {"messages": operator.add})
    {'messages': ['a', 'b'], 'intent': 'x'}
    """
    # TODO: начните с new = dict(state)
    raise NotImplementedError


class StateGraph:
    """Описание графа: узлы и рёбра. Сам ничего не выполняет — для этого compile()."""

    def __init__(self, schema):
        self.schema = schema
        self.keys, self.reducers = reducers_from_schema(schema)
        self.nodes = {}      # имя -> функция
        self.edges = {}      # откуда -> куда (обычные рёбра)
        self.branches = {}   # откуда -> (маршрутизатор, path_map или None)

    def add_node(self, name, fn):
        """Добавить узел. Имена START и END зарезервированы, повтор имени — тоже ValueError. Вернуть self."""
        # TODO
        raise NotImplementedError

    def _check_source(self, source):
        """Готово: из END рёбер нет; у узла не больше одного исходящего ребра (обычного или условного)."""
        if source == END:
            raise ValueError("из END рёбер не бывает")
        if source in self.edges or source in self.branches:
            raise ValueError(f"у «{source}» уже есть исходящее ребро: в этом движке узлы выполняются по одному")

    def add_edge(self, source, target):
        """Обычное ребро source → target (source может быть START, target — END). В START входить нельзя.
        Сначала вызовите self._check_source(source). Запишите в self.edges. Вернуть self."""
        # TODO
        raise NotImplementedError

    def add_conditional_edges(self, source, path, path_map=None):
        """Условное ребро: после source вызывается path(state) и возвращает метку.
        Если path_map задан (словарь метка -> узел), идём в path_map[метка], иначе метка — это и есть имя узла (или END).
        Сначала self._check_source(source). Запишите в self.branches[source] = (path, path_map). Вернуть self."""
        # TODO
        raise NotImplementedError

    def compile(self, checkpointer=None):
        """Проверить граф и вернуть CompiledGraph(self, checkpointer). ValueError, если:
          * нет ребра из START (ни обычного, ни условного);
          * ребро выходит из неизвестного узла;
          * обычное ребро или значение path_map ведёт в узел, которого нет (END — можно);
          * у какого-то узла нет исходящего ребра (иначе граф «застрянет»)."""
        # TODO
        raise NotImplementedError


class CompiledGraph:
    """Исполнитель графа: invoke() проходит по узлам от START до END."""

    def __init__(self, graph, checkpointer=None):
        self.graph = graph
        self.checkpointer = checkpointer

    def _save(self, thread_id, state, next_node, step):
        """Готово: записать контрольную точку, если подключено хранилище."""
        if self.checkpointer is not None:
            self.checkpointer.put(thread_id, {"values": state, "next": next_node, "step": step})

    def get_state(self, config):
        """Готово: последняя контрольная точка потока (в LangGraph — объект StateSnapshot с .values и .next)."""
        if self.checkpointer is None:
            raise ValueError("get_state работает только с checkpointer")
        return self.checkpointer.get(config["configurable"]["thread_id"])

    def _next(self, source, state):
        """Куда идти после source. Обычное ребро — self.graph.edges[source].
        Условное — label = path(state); с path_map: label обязан быть в path_map (иначе ValueError);
        без path_map: label должен быть узлом или END (иначе ValueError)."""
        # TODO
        raise NotImplementedError

    def invoke(self, input, config=None):
        """Выполнить граф и вернуть итоговое состояние.

        config — как в LangGraph: {"configurable": {"thread_id": "..."}, "recursion_limit": 25}.
          * Есть checkpointer, но нет thread_id — ValueError.
          * input is None — ВОЗОБНОВЛЕНИЕ: взять последний чекпоинт потока (нет его — ValueError)
            и продолжить с узла cp["next"] с состоянием cp["values"] и счётчиком шагов cp["step"].
          * иначе — НОВЫЙ ВВОД: состояние = apply_update(прошлое состояние потока или {}, input, ...),
            шаг = прошлый step или 0, первый узел = self._next(START, state); сохранить чекпоинт ДО первого узла.
        Цикл, пока узел не END:
          * если в ЭТОМ вызове уже выполнено recursion_limit узлов — GraphRecursionError;
          * update = узел(копия состояния) — dict или None; state = apply_update(...с self.graph.keys);
          * node = self._next(node, state); step += 1; сохранить чекпоинт (state, node, step).
        Исключение узла не ловите: пусть летит наружу — последний чекпоинт как раз указывает на упавший узел.
        """
        # TODO
        raise NotImplementedError


def demo():
    """Ассистент онлайн-школы: классификатор отправляет вопрос в RAG, а просьбу записаться — в узел действий."""

    class School(typing.TypedDict):
        messages: typing.Annotated[list, operator.add]
        intent: str

    def classify(state):                       # вместо LLM — ключевые слова
        text = state["messages"][-1].lower()
        return {"intent": "action" if "запиш" in text or "отмени" in text else "question"}

    g = StateGraph(School)
    g.add_node("classify", classify)
    g.add_node("rag", lambda s: {"messages": ["бот: ответ по каталогу курсов"]})
    g.add_node("act", lambda s: {"messages": ["бот: записать вас на поток? Ответьте «да»"]})
    g.add_edge(START, "classify")
    g.add_conditional_edges("classify", lambda s: s["intent"], {"question": "rag", "action": "act"})
    g.add_edge("rag", END)
    g.add_edge("act", END)
    saver = InMemorySaver()
    app = g.compile(checkpointer=saver)
    cfg = {"configurable": {"thread_id": "ученик-1"}}
    print(app.invoke({"messages": ["user: когда старт курса по RAG?"]}, cfg)["messages"])
    print(app.invoke({"messages": ["user: запишите меня на поток"]}, cfg)["messages"][-1])
    print("чекпоинтов в потоке:", len(saver.list("ученик-1")), "| следующий узел:", app.get_state(cfg)["next"])


try:
    demo()
except Exception as e:  # пока движок не дописан
    print("Допишите движок:", type(e).__name__, e)

ЗапуститьПроверитьПодсказкаСбросить

Ассистент школы на своём движке

Когда тесты пройдены, сохраните решение как state_graph.py (без последних строк с вызовом demo()) и соберите граф со схемы выше. Модель пока заменяют правила в classify; в 8.5.2 на их место встанет LLM. В коде три идеи урока сразу: подтверждение человеком как состояние потока, сбой с возобновлением и инструмент, который сам проверяет инвариант и узнаёт повтор по ключу идемпотентности.

import operator
from typing import Annotated, TypedDict

from state_graph import END, START, InMemorySaver, StateGraph   # ваш движок из задания

# «База» школы. Инвариант «не больше мест, чем вместимость» проверяет КОД инструмента, а не модель.
SEATS = {"rag-oct": 1}
ENROLLMENTS = {}                     # ключ идемпотентности -> запись


def enroll(student, cohort, key):
    """Инструмент «записать на поток». Повтор с тем же ключом не создаёт вторую запись."""
    if key in ENROLLMENTS:
        return ENROLLMENTS[key]
    if SEATS.get(cohort, 0) <= 0:
        return f"мест в потоке {cohort} нет — могу поставить в лист ожидания"
    SEATS[cohort] -= 1
    ENROLLMENTS[key] = f"{student} записан(а) на {cohort}, статус PENDING_PAYMENT"
    return ENROLLMENTS[key]


class School(TypedDict):
    messages: Annotated[list, operator.add]   # история диалога: только дописываем
    student: str
    intent: str
    pending: dict                             # действие, которое ждёт «да» от человека


def classify(s):                              # маршрутизатор; в 8.5.2 вместо правил будет модель
    text = s["messages"][-1].removeprefix("user: ").lower()
    if s.get("pending"):
        return {"intent": "confirm" if text in ("да", "да!") else "decline"}
    return {"intent": "enroll" if "запиш" in text else "question"}


def answer(s):
    return {"messages": ["бот: курс «RAG на практике» стартует 1 ноября, 6 недель"]}


def propose(s):                               # человек в контуре: граф останавливается и ждёт ответа
    action = {"cohort": "rag-oct", "key": f"enroll:{s['student']}:rag-oct"}
    return {"pending": action, "messages": [f"бот: записать вас на поток {action['cohort']}? Ответьте «да»"]}


flaky = {"armed": True}


def execute(s):
    result = enroll(s["student"], s["pending"]["cohort"], s["pending"]["key"])
    if flaky["armed"]:                        # имитируем сбой ПОСЛЕ записи, но ДО чекпоинта
        flaky["armed"] = False
        raise TimeoutError("сервис уведомлений не ответил")
    return {"pending": {}, "messages": [f"бот: {result}"]}


def drop(s):
    return {"pending": {}, "messages": ["бот: хорошо, ничего не делаю"]}


g = StateGraph(School)
for name, fn in [("classify", classify), ("answer", answer), ("propose", propose),
                 ("execute", execute), ("drop", drop)]:
    g.add_node(name, fn)
g.add_edge(START, "classify")
g.add_conditional_edges("classify", lambda s: s["intent"],
                        {"question": "answer", "enroll": "propose", "confirm": "execute", "decline": "drop"})
for name in ("answer", "propose", "execute", "drop"):
    g.add_edge(name, END)
app = g.compile(checkpointer=InMemorySaver())


def say(thread, text):
    cfg = {"configurable": {"thread_id": thread}}
    try:
        out = app.invoke({"messages": [f"user: {text}"], "student": thread}, cfg)
    except TimeoutError as e:
        print(f"[{thread}] сбой: {e}; следующий узел по чекпоинту: {app.get_state(cfg)['next']}")
        out = app.invoke(None, cfg)          # повтор с места остановки
    print(f"[{thread}] {text!r} -> {out['messages'][-1]}")


say("anna", "когда старт курса по RAG?")
say("anna", "запишите меня на поток")
say("anna", "да")
say("boris", "запишите меня тоже")
say("boris", "да")
print("записей:", len(ENROLLMENTS), "| свободных мест:", SEATS)

[anna] 'когда старт курса по RAG?' -> бот: курс «RAG на практике» стартует 1 ноября, 6 недель
[anna] 'запишите меня на поток' -> бот: записать вас на поток rag-oct? Ответьте «да»
[anna] сбой: сервис уведомлений не ответил; следующий узел по чекпоинту: execute
[anna] 'да' -> бот: anna записан(а) на rag-oct, статус PENDING_PAYMENT
[boris] 'запишите меня тоже' -> бот: записать вас на поток rag-oct? Ответьте «да»
[boris] 'да' -> бот: мест в потоке rag-oct нет — могу поставить в лист ожидания
записей: 1 | свободных мест: {'rag-oct': 0}

Что здесь произошло:

• Подтверждение человеком — состояние потока. Граф дошёл до END, но в потоке осталось pending; следующее сообщение пришло в тот же thread_id, и маршрутизатор понял, что ждём ответа. Процесс между ними мог перезапуститься.
• Сбой и возобновление. execute записала Анну и упала до чекпоинта, поэтому invoke(None, cfg) выполнил её ещё раз.
• Идемпотентность. Повторный enroll с тем же ключом вернул уже созданную запись. Уберите проверку if key in ENROLLMENTS — и повтор ответит Анне «мест нет»: место заняла она же. Мы проверили, так и происходит.
• Инвариант в коде. Борису отказал инструмент, а не модель: никакой текст не уговорит enroll нарушить вместимость.

Сверка с LangGraph. Поставьте pip install langgraph и замените импорт движка на два: from langgraph.checkpoint.memory import InMemorySaver и from langgraph.graph import END, START, StateGraph, а app.get_state(cfg)['next'] — на app.get_state(cfg).next. Больше ничего менять не нужно: на langgraph 1.2.14 (октябрь 2026) вывод совпадает построчно. Единственное отличие — следующий узел печатается кортежем ('execute',): в LangGraph в одном супершаге может ждать запуска несколько узлов. Что ещё умеет настоящий фреймворк — interrupt, стриминг, подграфы, хранилища чекпоинтов в базе, — разберём в уроке 8.5.2.

🖥 На вашей машине (4×GPU)

Замените правила в classify моделью и проверьте, годится ли маленькая модель в маршрутизаторы (решение из 8.5.5):

CUDA_VISIBLE_DEVICES=0 vllm serve Qwen/Qwen3-0.6B --port 8001 --reasoning-parser qwen3
CUDA_VISIBLE_DEVICES=1 vllm serve Qwen/Qwen3-8B --port 8002 --reasoning-parser qwen3

Ответ — JSON по схеме, как в уроке 8.1.2:

import json
from openai import OpenAI

ROUTE = {"type": "json_schema", "json_schema": {"name": "route", "schema": {
    "type": "object", "required": ["intent"],
    "properties": {"intent": {"type": "string", "enum": ["question", "enroll", "confirm", "decline"]}}}}}
SYSTEM = ("Ты маршрутизатор онлайн-школы. question — вопрос о курсах, enroll — просьба записать, "
          "confirm — согласие на предложенное действие, decline — отказ.")

def make_classify(port, model):
    client = OpenAI(base_url=f"http://localhost:{port}/v1", api_key="EMPTY")
    def classify(s):
        r = client.chat.completions.create(
            model=model, temperature=0, max_tokens=30, response_format=ROUTE,
            extra_body={"chat_template_kwargs": {"enable_thinking": False}},
            messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": s["messages"][-1]}])
        intent = json.loads(r.choices[0].message.content)["intent"]
        if intent in ("confirm", "decline") and not s.get("pending"):   # подтверждать можно только предложенное
            intent = "question"
        return {"intent": intent}
    return classify

Разметьте 60 реплик (по 15 на класс, с опечатками и смешанными просьбами) и сравните точность и задержку двух моделей; вывод зависит от набора, поэтому здесь его не приводим. Допустимость перехода (подтверждать можно только предложенное) проверяет код, а не модель. На процессоре 8B отвечала бы секунды на реплику, здесь оба прогона занимают минуты. Подробнее о машине — «Ваша лаборатория».

Китайский угол: AgentScope 2.0 и «делать колесо»

AgentScope — открытый мультиагентный фреймворк Alibaba (по метаданным PyPI — команда SysML лаборатории Тунъи; agentscope-ai/agentscope, Apache-2.0, 2.0.9 от 28.09.2026). Авторы делают ставку на рассуждение и инструменты самой модели, а не на навязанную оркестрацию, но в августе — сентябре 2026 добавили и конвейеры с фиксированной логикой, TeamPipeline (лидер раздаёт задачи участникам) и SOP в духе MetaGPT из 8.3.3.

Claude Agent SDK официально работает только с Claude, но китайские лаборатории дают Anthropic-совместимые адреса: такой API есть у Moonshot (README Kimi K2), а поваренная книга Zhipu подключает GLM-4.5 к Claude Code (наш разбор). Путь неофициальный, совместимость не гарантирована.

Учебный маршрут — hello-agents (Datawhale): глава 6 — AutoGen, AgentScope и LangGraph на практике, глава 7 — свой фреймворк с нуля. Цель авторов — уметь и «пользоваться колесом», и «делать колесо»; ваш движок — второе.

Проверь себя

Проверь себя6 вопросов·

1 Дешёвая модель определяет тип обращения: вопросы уходят в RAG, просьбы о записи — в ветку с инструментами, жалобы — оператору. Какой это шаблон из «Building effective agents»?

Цепочка промптовМаршрутизацияОркестратор и исполнителиОценщик и оптимизатор

2 Что дают контрольные точки, которые граф пишет после каждого узла?Можно выбрать несколько ответов

Возобновление после сбоя без повтора уже выполненных узловПаузу до ответа человека, которая переживает перезапуск процессаВозврат к любому прошлому шагуГарантию, что повтор узла не повторит его побочный эффект

3 Граф: START → writer → reviewer, после reviewer «одобрено → END, иначе → writer». Одобряется только 4-я попытка. Какой минимальный recursion_limit нужен в вашем движке?

4 Узел execute списал оплату за курс, и процесс упал до записи чекпоинта. Что будет после возобновления?

Фреймворк поймёт, что оплата прошла, и пропустит узелУзел выполнится заново; без ключа идемпотентности оплата спишется второй разФреймворк сам откатит первую оплатуПоток начнётся с START

5 В состоянии messages = ["a"]. Два узла по очереди возвращают {"messages": ["b"]} и {"messages": ["c"]}, ключ объявлен без редьюсера. Что в messages в конце?

["a", "b", "c"]["c"]["b", "c"]["a", "c"]

6 Команда запускает OpenAI Agents SDK с локальной моделью через vLLM, ключа OpenAI нет. О чём позаботиться в первую очередь?

Ни о чём: локальную модель к SDK подключить нельзяВыключить трассировку или поставить свой обработчик: по умолчанию трассы уходят на серверы OpenAIДождаться версии 1.0: в 0.x нет других провайдеров

Проверить ответы

Итоги

• Фреймворк поверх агентного цикла даёт состояние с правилами слияния, ветвления, чекпоинты и возобновление, паузы для человека, стриминг, трассы, ретраи и параллельность. По отдельности всё пишется руками; продают сочетание.
• Пять шаблонов Anthropic — цепочка, маршрутизация, параллелизация, оркестратор и исполнители, оценщик и оптимизатор — это workflow: путь задан кодом. Агент выбирает путь сам. Начинайте с простого и знайте, что под капотом.
• Граф состояний — конечный автомат над общим состоянием: узлы возвращают частичные обновления, редьюсеры их сливают, рёбра выбирают следующий узел, лимит шагов обрывает циклы.
• Чекпоинт пишется после узла, поэтому упавший узел повторяется: побочные эффекты — идемпотентные, инварианты — в коде инструментов.
• Карта на октябрь 2026: LangGraph, OpenAI Agents SDK, Claude Agent SDK, AG2 и Microsoft Agent Framework (AutoGen — в режиме поддержки), CrewAI, smolagents, PydanticAI. Критерии выбора — контроль над потоком, привязка к провайдеру, наблюдаемость, зрелость, лицензия; версии закрепляйте.

Дальше: LangGraph на практике — StateGraph, checkpointer и потоки, interrupt для подтверждения человеком, стриминг и подграфы на сквозном примере ассистента онлайн-школы.