Ваша структура агентів має бути у форматі 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(...) та 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
Жодна інструкція import не пов’язує граф з цією функцією. Завантажувач розв’язує це за допомогою 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 відображає параметри, які оголошує функція, і виконувач формує саме такий виклик. Синхронні та асинхронні випадки обробляються однаково: якщо результат є awaitable, його очікують, інакше використовують безпосередньо. Автори обирають ту форму підпису, яка звучить природно. Допоміжна функція _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. - Діаграма архітектури: намальована вручну та швидко стає застарілою, на відміну від автоматично створеної на основі актуальної графіки.
- Хто може змінювати потік обробки: людина, яка добре розбирається в API графіки, на відміну від будь-кого, хто вміє читати YAML.
Останній пункт є найважливішим. Специфікація робочого процесу, яку можуть читати менеджери продукту, яку може перевірити співробітник, незнайомий з LangGraph, у запиті на зміни, та де різниці представлені у вигляді звичайного тексту — це означає, що сама різниця є зміною. Нікому не потрібно уявно виконувати 40 рядків переорганізованого коду на Python.
Фреймворки з багатьма агентами надають вузли та ребра, механізм їх взаємодії, але не створюють організованого простору для визначення політики, тож структура робочого процесу залишається окремою від того, що робить кожен крок. Якщо граф агентів перетворився на складну схему, яку всі бояться змінювати, спробуйте вивести його топологію у вигляд даних. Завантажувач складається лише з кількох сотень рядків: динамічного об’єкта TypedDict, модуля importlib для плагінів, функції виклику, адаптованої до формату підпису, та циклів, які перетворюють вузли та ребра на відповідні виклики в графі. Через місяць, коли хтось хоче додати крок перед аналізом, відповіддю є трирядковий YAML-диф, без зайвих витрат часу на створення графа.
Пов’язана література
- Тріаж підтримки клієнтів за допомогою кількох агентів з LangGraph та Ollama — Створіть граф тріажу заявок із трьома агентами за допомогою LangGraph: класифікатор, оператор та людський модератор, які працюють на локальному Ollama або через API, з однією зміненою рядком коду.
- Самовідновлювана команда кількох агентів у LangGraph: архітектор, тестувальник, дебагер — Створіть машину станів LangGraph на Python, у якій архітектор, розробник, среда pytest sandbox, дебагер та рецензент співпрацюють та виконують цикли до тих пір, поки тести не пройдуть.