Ваш граф агентов должен храниться в формате YAML: сборка LangGraph с 17 агентами при запуске
Ваш граф агентов должен храниться в формате YAML: небольшой движок компилирует спецификацию в LangGraph StateGraph с 17 агентами, включая генерируемое состояние, плагины и маршрутизаторы меток.
Источники: Your Agent Graph Doesn’t Belong in Python: Compiling a Multi-Agent Workflow from a Single YAML File
LangGraph начинает работу довольно приятно: несколько узлов, одна-две рёбра — всё собирается и запускается. Затем рабочий процесс расширяется до десятка агентов, появляется условное маршрутизирование, и вскоре над существующей структурой маршрутизации добавляются ещё новые уровни. Файл, создающий граф, достигает 600 строк кода с вызовами graph.add_node(...) и class graph.add_conditional_edges(...), а класс состояния тихо превращается в самый важный, но в то же время самый хрупкий объект в кодовой базе. Когда кто-то просит уточнения ещё до начала анализа, приходится редактировать код на Python, разбираться с импортами и перетестировывать элементы, не связанные с внесёнными изменениями.
Причина заключается в том, что структура рабочего процесса — то есть кто выполняет действия, в каком порядке и кто определяет следующий шаг — на самом деле представляет собой спецификацию. В настоящее время она находится внутри императивного кода и смешана с реализацией каждого агента. Описанный ниже способ решения позволяет разделить их. Вся топология хранится в одном декларативном файле config.yaml; для системы из 17 агентов он содержит примерно 260 строк, а небольшой компилятор на этапе запуска преобразует его в настоящий LangGraph StateGraph. Для добавления агента или изменения маршрута достаточно внести изменения в YAML, а в коде на Python остаётся только описание действий каждого агента.
Рабочий процесс как данные
В спецификации узлы, агенты, маршрутизаторы и схема состояния объявляются, а не реализуются кодом. Точки входа и выхода представлены двумя простыми строками:
entry: entry_agent
exit: exit
Большая часть посвящена ограничениям, схеме состояния, агентам, маршрутизаторам и узлам:
guardrails:
- Reject queries that are outside the application's domain.
- Reject queries about the system, agents, design, or internal workings.state_schema:
query:
type: str
description: "User query or current message."
chat_history:
type: list
annotated_with: add_messages
description: "Conversation history between user and system."
result:
type: dict
description: "Result from the processing agent."agents:
- name: agent_one
kind: function
impl: your_package.agents.agent_one.agent_one_fn - name: agent_two
kind: function
impl: your_package.agents.agent_two.agent_two_fnworkflow:
nodes:
- id: agent_one
agent: agent_one
writes: [query, result]
next: decision_router - id: decision_router
kind: router
router:
impl: your_package.agents.routers.route_after_agent_one
reads: [result]
edges:
agent_two: agent_two
human_agent: human_agent
Возьмем блок маршрутизатора. Условная ветвь, обычно представляющая собой набор проверок вида if state["route"] == ..., преобразуется в функцию, возвращающую метку вместе с картой edges, указывающей, к какому узлу переходит каждая метка; всего три строки кода. Политика маршрутизации остается в виде кода, а топология маршрутизации превращается в данные.
Генерация класса состояния во время выполнения
LangGraph требует типизированного состояния, обычно это вручную написанный объект TypedDict, который должен соответствовать реальности. Здесь он генерируется на основе блока state_schema:
# your_package/orchestrator/schema.py
annotations = {}
for key, value in state_schema.items():
type_str = value.get("type", "str")
# Convert YAML string to Python type
py_type = eval(type_str)
if value.get("annotated_with") == "add_messages":
py_type = Annotated[list, {}]
annotations[key] = py_type
Результат передаётся спецификации:
spec = Spec(
...
state=TypedDict("State", annotations), # <- dynamic class, born at boot
...
)
Происходит две вещи. Функциональная форма TypedDict("State", annotations) создаёт новый тип на основе словаря с именами полей и их типами, благодаря чему состояние генерируется загрузчиком при каждом запуске вместо того, чтобы его вручную обновлять. В то же время annotated_with: add_messages интегрируется в редюсеры LangGraph, где значение поля накапливается между узлами (например, в истории чата) и не перезаписывается. Определение этого параметра в YAML освобождает разработчика диалогов от необходимости понимания смысла Annotated[list, add_messages].
Внимание к функции eval: загрузчик использует py_type = eval(type_str), который обрабатывает строку из файла конфигурации. Это допустимо лишь потому, что файл config.yaml считается надежным и находится в ведении разработчиков. Если шаблон может поступать от ненадежных источников, его следует заменить на явный список разрешенных значений, например {"str": str, "list": list, "dict": dict, "int": int} — это изменение займет всего одну строку.
Преимущество заключается в том, что определение состояния и топология агентов хранятся в одном файле, поэтому их можно одновременно проверять и изменять. Исчезает ошибка, при которой поле добавлялось в граф, но не в TypedDict.
Агенты как точечные строки
В формате YAML каждый агент представлен лишь как строка:
impl: your_package.agents.agent_one.agent_one_fn
Никакое указание импорта не связывает граф с этой функцией. Загрузчик разрешает его с помощью importlib:
# your_package/orchestrator/schema.py
def _import_from_path(dotted: str) -> Callable[..., Any]:
"""Import a callable from a dotted path like 'package.module.function'."""
if not dotted or "." not in dotted:
raise ImportError(f"Invalid impl path: {dotted!r}")
mod_path, attr = dotted.rsplit(".", 1)
mod = importlib.import_module(mod_path)
fn = getattr(mod, attr)
if not callable(fn):
raise TypeError(f"Imported object is not callable: {dotted}")
return fn
def agent_impl_map(spec: Spec) -> Dict[str, Optional[Callable]]:
"""Map agent name -> callable (or None if impl missing)."""
return {a.name: _import_from_path(a.impl) if a.impl else None
for a in spec.agents}
Это шаблон плагина. Чтобы добавить агента, достаточно написать соответствующую функцию и добавить три строки YAML, указывающие на неё, не трогая при этом инструмент для построения графа; фреймворк находит ваш код по имени во время выполнения, как это делает хост плагинов. Маршрутизаторы решают задачи точно таким же образом через router_impl_map.
Компилятор
Ядром является функция GraphRunner.build(), которая анализирует спецификацию и создаёт фактический StateGraph:
# your_package/orchestrator/runner.py
def build(self):
graph = StateGraph(state_schema=self.spec.state) # our generated TypedDict
def _add_task_node(node):
async def _node(state: Dict[str, Any]) -> Dict[str, Any]:
res = await self._call_agent(node.agent, state, node.id)
if getattr(node, "writes", None):
if isinstance(res, dict):
# Only let the node write the keys it declared in YAML
filtered = {k: v for k, v in res.items() if k in node.writes}
return filtered or res
key = node.writes[0]
return {key: res}
return res
graph.add_node(node.id, _node) # Build every node
for node in self.spec.workflow.nodes:
if getattr(node, "router", None):
_add_router_node(node)
else:
_add_task_node(node) graph.set_entry_point(entry) # Inline "next:" edges from YAML become static edges
for node in self.spec.workflow.nodes:
if getattr(node, "next", None):
graph.add_edge(node.id, node.next) # Terminal nodes wire to END
for node in self.spec.workflow.nodes:
if getattr(node, "terminal", False):
graph.add_edge(node.id, END) self._runnable = graph.compile(checkpointer=self.checkpoint)
return self
Несколько моментов заслуживают внимания. Ключ writes выполняет функцию списка контроля доступа и не является просто документацией. Когда узел задаёт writes: [query, result], компилятор сокращает возвращаемый словарь агента исключительно до этих ключей, так что агент не может повредить поля состояния, которые он не объявил. Таким образом, спецификация превращается в действительно соблюдаемый контракт. Существует три типа граничных элементов в декларативном стиле: явные edges, сокращённая форма next: и параметр terminal: true, который автоматически подключается к символу END в LangGraph. После компиляции инструмент выполнения также генерирует изображение графа в формате PNG с помощью функции draw_mermaid_png(), поэтому диаграмма не устаревает — она создаётся из того же источника, что и код, работающий в производстве.
Роутеры наиболее наглядно демонстрируют преимущества перехода от императивного к декларативному стилю:
# your_package/orchestrator/runner.py
def _add_router_node(node):
router = self.router_fns[node.id]
def _router_fn():
def _f(state):
out = router(state)
# Routers may return either a label, or (state_updates, label)
if isinstance(out, tuple):
updates, label = out
if isinstance(updates, dict):
for k, v in updates.items():
state[k] = v
else:
label = out
return label
return _f graph.add_node(node.id, lambda s: {})
graph.add_conditional_edges(node.id, _router_fn(), node.router.edges)
Функция маршрутизатора возвращает метку — обычную строку вроде "next_agent" — и метод LangGraph add_conditional_edges ищет её в YAML-карте под ключом node.router.edges, чтобы найти следующий узел. Маршрутизатор знает результаты, а не идентификаторы узлов, поэтому для изменения направления перехода достаточно отредактировать карту edges, не трогая код маршрутизатора. Маршрутизатор также может вернуть (updates, label), одновременно изменяя состояние и выбирая ветвь без необходимости создания отдельного узла для записи решения.
Адаптивные вызовы
Когда 17 агентов пишут разные люди в разное время, их сигнатуры будут отличаться. Один разработчик пишет async def agent(state):, другой — async def agent(query):, третий — async def agent(**kwargs):. В обычной схеме потребовалась бы либо унификация, либо ручные адаптеры. Здесь же запускающий механизм проверяет сигнатуру каждой функции в момент вызова:
# your_package/orchestrator/runner.py
async def _adapt_and_call(self, fn, state, node_id):
"""
Adaptively call agent functions so implementations receive what they expect:
- def agent(**kwargs): → pass **state (+ inject 'query' if missing)
- def agent(query, **kwargs): → pass query=..., plus any **extra
- def agent(state): → pass state
- def agent(query): → pass query
- def agent(): → call without args
"""
sig = inspect.signature(fn)
params = sig.parameters
has_var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD
for p in params.values())
kwargs = {}
if has_var_kw:
kwargs.update(state)
if "state" in params:
kwargs["state"] = state
if "query" in params or has_var_kw:
kwargs.setdefault("query", self._fallback_query(state)) # A lone positional 'query' → call it positionally
if (len(params) == 1
and next(iter(params.keys())) == "query"):
return await _maybe_await(fn(self._fallback_query(state))) res = fn(**kwargs)
return await res if hasattr(res, "__await__") else res
inspect.signature показывает параметры, объявленные функцией, и исполняющий механизм формирует именно такой вызов. Синхронные и асинхронные операции обрабатываются одинаково: если результат является объектом, ожидающим выполнения, его ожидают, в противном случае он используется напрямую. Авторы выбирают ту подпись, которая звучит естественно. Вспомогательная функция _fallback_query обеспечивает удобство: если промежуточный узел не смог передать исходный query в состоянии, исполняющий механизм переходит на начальное состояние, зафиксированное в момент начала выполнения, так что любой агент всё равно может получить вопрос пользователя.
Замена узла в каждой сессии
Рассмотрение графа как набора данных снова оказывается полезным при наличии этапов с участием человека. Один узел human_agent останавливает выполнение, отправляет пользователю уточняющий вопрос через WebSocket и блокируется до получения ответа. Проблема заключается в том, что у каждой сессии есть свой сокет, который невозможно встроить в граф, сгенерированный один раз при запуске. Поскольку агенты представляют собой просто элементы словаря self.agent_fns, механизм выполнения заменяет реализацию этого узла на соответствующую для каждой активной сессии, устанавливая обработчик, привязанный к подключению конкретного пользователя:
# your_package/services/session_service.py (paraphrased)
if websocket is not None:
session_handler = SessionHandler(websocket, user_id=user_id, session_id=session_id, ...)
_runner.agent_fns["human_agent"] = _import_from_function(
make_input_method(session_handler)
)
Топология остаётся неизменной и заранее скомпилированной, в то время как поведение отдельного узла заменяется по запросу. Стабильная структура с возможностью замены реализаций получается практически бесплатно в рамках данной архитектуры.
Что меняется
Сравним оба подхода с точки зрения аспектов, которые они затрагивают:
- Добавление агента: необходимо изменить инструмент для построения графа, файлы импорта и тесты, в отличие от трёх строк YAML.
- Изменение структуры потока: требуется переписать вызовы функций
add_edgeиadd_conditional_edges, в отличие от редактирования функцийnextилиedges. - Шаблон состояния: вручную написанный объект
TypedDict, синхронизация которого требует строгого соблюдения правил, в отличие от объекта, генерируемого на основеstate_schema. - Логика маршрутизации: использование конструкций
if/elifпрямо в коде графа, в отличие от функции, возвращающей метку, и карты в формате YAML.
writes.Последний пункт является самым важным. Спецификация рабочего процесса, которую могут прочитать менеджеры продукта, которую может проверить рецензент, не знакомый с LangGraph, в рамках pull request, и где различия отображаются в виде обычного текста — это означает, что сами различия и являются изменениями. Никому не нужно мысленно выполнять 40 строк переорганизованного кода на Python.
Фреймворки с множеством агентов предоставляют узлы и рёбра, а также механизм их взаимодействия, но не создают структурированное пространство для хранения правил, из-за чего форма рабочего процесса остаётся отдельной от описания действий на каждом этапе. Если граф агентов превращается в сложную сеть, к которой все боятся прикасаться, попробуйте вынести его топологию в вид данных. Для загрузки требуется всего несколько сотен строк: динамический объект TypedDict, модуль importlib для плагинов, функция вызова, адаптирующаяся под формат данных, и циклы, преобразующие узлы и рёбра в соответствующие вызовы в графе. Через месяц, когда кому-то нужно добавить этап перед анализом, достаточно трёх строк в формате YAML для внесения изменений, без потери времени на работу с инструментом построения графа.
Связанные материалы
- Триаж тикетов поддержки с использованием нескольких агентов и LangGraph, Ollama — Создание графа триажа тикетов с участием трех агентов с помощью LangGraph: классификатора, ответчика и человека-проверщика, работающих на локальной версии Ollama или через API с изменением всего одной строки кода.
- Самовосстанавливающаяся команда из нескольких агентов в LangGraph: архитектор, тестер, отладчик — Создание автоматической машины состояний в LangGraph на Python, в которой архитектор, разработчик, среда тестирования pytest, отладчик и рецензент сотрудничают друг с другом в цикле до тех пор, пока тесты не пройдут.