Месца графа агентаў — у YAML: кампіляцыя LangGraph з 17 агентаў пры старце
Граф агентаў лепш трымаць у YAML: невялікае асяроддзе кампілюе адну спецыфікацыю ў StateGraph LangGraph з 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)
Функцыя рутэра вяртае атрыбут label – звычную строчку, напрыклад "next_agent" – і метод LangGraph’s add_conditional_edges шукае яго ў карце YAML пад ключам node.router.edges каб знайсці наступны вузел. Рутэр знае рэзультаты, а не ID вузлаў, таму калі трэба змяніць напрамак пераходу, трэба вырэшыць пытанне ў карце 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, у запросе прыўлаку, і калі разлікі прадставлены як звычны тэкст, сам разлік являецца змянай. Нікому не трэба ментальна выконваць 40 ліній перакантроліраванага Python.
Фрэймворкі з мнагацэлевымі агентамі задаюць вузлы і рэшткі, механізм, але не ствараюць адчытнае месца для наявных правіл, тое значыць, структура робочага працэсу застаецца аддзеленай ад таго, што робіць кожны крок. Калі граф агента стае такім складным, што всі бояцца яго змяніць, спробуйце вынесты топалогію у варажыні. Код-завантажувач складаецца з калькі соткаў ліній: дынамічны TypedDict, importlib для плагінаў, вызов, прыстосаваны да падпіса, і ціклы, якія перакладаюць вузлы і рэшткі у вызовы графа. Чэрэз месцы, калі хтось хочаць дадаць новы крок перад аналізам, адпаведзь — гэта трыя лініі у фармаце YAML, без вялікіх затрачэнняў часу на стварэнне графа.
Супаўзвязаныя матэрыялы
- Трыяж тыкетаў падтрымкі з удзелам калькольніка агентаў з LangGraph і Ollama — Створыце граф трыяжа тыкетаў з трохіма агентамі за дапамогай LangGraph: класыфікатор, адпаведальны за адпаведзь і людзкі кантролер, якія працуюць на локальным Ollama або праз API, з адной змененай лінійкай коду.
- Автаналізуючыся каманда калькольніка агентаў у LangGraph: архітэктар, тэставальнік, дэбаггер — Створыце машыну станоў LangGraph на Python, у якой архітэктар, распрацоўшчык, сандбокс pytest, дэбаггер і рэвізор саавтарсаваць і будуць работаць па цыклу, пакуль тэсты не пройдуць.