Twój graf agentów powinien być w formacie YAML: kompilacja LangGraph z 17 agentami podczas uruchamiania
Twój graf agentów powinien być zapisany w formacie YAML: mały silnik w czasie wykonywania kompiluje jedną specyfikację do LangGraph StateGraph zawierającego 17 agentów, wraz z wygenerowanym stanem, pluginami i routerami etykiet.
Źródła: Your Agent Graph Doesn’t Belong in Python: Compiling a Multi-Agent Workflow from a Single YAML File
LangGraph zaczyna się przyjemnie – kilka węzłów, jeden lub dwa krawędzie, kompilacja i uruchomienie. Następnie proces przechodzi do dziesiątek agentów, pojawia się warunkowe routowanie, a wkrótce dodatkowe warstwy routowania są nakładane na istniejące. Plik tworzący graf zawiera już 600 linii kodu typu graph.add_node(...) i graph.add_conditional_edges(...), a klasa stanu w ciszy stała się najważniejszym i najbardziej kruchym obiektem w całym kodzie. Gdy ktoś prosi o wyjaśnienia przed rozpoczęciem analizy, trzeba edytować kod w Pythonie, sprawdzać importy oraz ponownie testować elementy, które w ogóle nie mają związku z dokonaną zmianą.
Powodem jest to, że struktura procesu pracy, czyli to, kto go uruchamia, w jakiej kolejności i kto decyduje o dalszych krokach, stanowi w rzeczywistości specyfikację. Obecnie jest ona zamknięta w kodzie imperatywnym, zmieszana z implementacją każdego agenta. Opisany poniżej sposób rozwiązuje ten problem, oddzielając te elementy. Cała topologia znajduje się w jednym pliku deklaratywnym config.yaml – około 260 linii dla systemu składającego się z 17 agentów – a mały program w czasie uruchamiania kompiluje go do rzeczywistego LangGraph StateGraph. Dodanie agenta lub zmiana ścieżki przepływu polega na edycji pliku YAML, a kod w Pythonie zawiera jedynie informacje o tym, co robi każdy agent.
Proces pracy jako dane
W specyfikacji węzły, agenci, routery oraz schemat stanu są deklarowane, a nie kodowane. Punkty wejścia i wyjścia to zwykłe dwie linie:
entry: entry_agent
exit: exit
A większa część opisuje zasady bezpieczeństwa, schemat stanu, agenty, routery oraz węzły:
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
Weź blok routera. Warunkowa gałąź, zwykle zbiór sprawdzeń typu if state["route"] == ..., sprowadza się do funkcji, która zwraca etykietę wraz z mapą edges łączącą każdą etykietę z jej następnym węzłem – łącznie trzy linie kodu. Zasady routingu pozostają w formie kodu, natomiast topologia routingu staje się danymi.
Generowanie klasy stanu w czasie wykonywania
LangGraph wymaga typizowanego stanu, zwykle ręcznie napisanego obiektu TypedDict, który musi odpowiadać rzeczywistości. W tym przypadku jest on generowany z bloku 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
Wynik jest przekazywany do specyfikacji:
spec = Spec(
...
state=TypedDict("State", annotations), # <- dynamic class, born at boot
...
)
Dzieją się dwie rzeczy. Funkcjonalna forma TypedDict("State", annotations) tworzy nowy typ na podstawie słownika nazw i typów pól, dzięki czemu stan jest budowany przez narzędzie ładowania przy każdym uruchomieniu, zamiast być ręcznie aktualizowany. Tymczasem annotated_with: add_messages łączy się z reduktorami LangGraph, gdzie dane z pola są gromadzone między węzłami (np. w historii rozmowy) i nie są nadpisywane. Deklarowanie tego w formacie YAML pozwala osobie projektującej rozmowę uniknąć konieczności zrozumienia, co oznacza Annotated[list, add_messages].
Uwaga dotycząca eval: loader używa py_type = eval(type_str), który ewaluuje ciąg znaków z pliku konfiguracyjnego. Jest to do przyjęcia tylko dlatego, że config.yaml jest wiarygodny i znajduje się w posiadaniu programistów. Jeśli osoby niezaufane mogłyby dostarczyć tę strukturę, należy zastąpić ją wyraźną listą dozwolonych wartości, np. {"str": str, "list": list, "dict": dict, "int": int} – to zmiana wymagająca tylko jednej linijki kodu.
Zaletą jest to, że definicja stanu i topologia agentów dzielą ten sam plik, dzięki czemu są one jednocześnie sprawdzane i modyfikowane. Problem polegający na dodawaniu pola do grafu, ale nie do TypedDict, znika.
Agenci jako ciągi znaków z kropkami
W formacie YAML każdy agent jest po prostu ciągiem znaków:
impl: your_package.agents.agent_one.agent_one_fn
Żadne polecenie import nie łączy tego grafu z daną funkcją. Ładowarka rozwiązuje to za pomocą 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}
To jest wzorzec pluginu. Aby dodać agenta, piszesz funkcję i dodajesz trzy linie w formacie YAML wskazujące na nią, nie dotykając przy tym narzędzia do budowania grafu; framework znajduje twój kod podczas wykonywania według nazwy, tak jak to robi host pluginu. Routerzy są rozwiązywane w dokładnie ten sam sposób za pomocą router_impl_map.
Kompilator
Jądro stanowi GraphRunner.build(), który przegląda specyfikację i buduje rzeczywisty 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
Kilka kwestii wymaga uwagi. Klucz writes pełni funkcję listy kontrolowania dostępu i nie jest jedynie dokumentacją. Gdy węzeł deklaruje writes: [query, result], kompilator ogranicza zwracany przez agenta słownik wyłącznie do tych kluczy, dzięki czemu agent nie może uszkodzić pól stanu, których nie deklarował. Specyfikacja staje się umową, która jest faktycznie egzekwowana. Istnieją trzy typy krawędzi deklaratywnych: wyraźne edges, skrót next: umieszczony bezpośrednio, oraz terminal: true, który automatycznie łączy się z END w LangGraph. Po skompilowaniu narzędzie renderuje również grafikę jako plik PNG za pomocą draw_mermaid_png(), dzięki czemu diagram nie ulega przestarzeniu – pochodzi z tej samej bazy, która jest używana w produkcji.
Routerzy najwyraźniej pokazują korzyści płynące ze przejścia od stylu imperatywnego do deklaratywnego:
# 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)
Funkcja routera zwraca etykietę, zwykłą ciąg znaków, np. "next_agent", a funkcja LangGrapha add_conditional_edges szuka jej w mapie YAML pod kluczem node.router.edges, aby znaleźć następny węzeł. Router zna wyniki, a nie ID węzłów, więc zmiana kierunku, do którego prowadzi dany wynik, oznacza edycję mapy edges bez konieczności modyfikowania kodu routera. Router może również zwrócić (updates, label), dostosowując stan i wybierając odpowiednią gałąź w jednym zwróceniu, bez konieczności tworzenia osobnego węzła do zapisania decyzji.
Wywołania adaptacyjne
Gdy 17 agentów jest pisanych przez różnych osób w różnym czasie, ich sygnatury będą się różnić. Jeden programista pisze async def agent(state):, inny async def agent(query):, a jeszcze inny async def agent(**kwargs):. Zwykły graf zmusiłby do zgodności lub wymagałby ręcznie napisanych adapterów. Tutaj narzędzie uruchamiające sprawdza sygnaturę każdej funkcji w momencie wywołania:
# 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 pokazuje parametry deklarowane przez funkcję, a narzędzie wykonywania buduje dokładnie tę próbę wywołania. Zarządzanie operacjami synchronicznymi i asynchronicznymi odbywa się w sposób jednolity: jeśli wynik jest wart oczekiwania, jest on oczekiwany, w przeciwnym razie jest używany bezpośrednio. Autorzy wpisują taką sygnaturę, która brzmi naturalnie. Pomocnik _fallback_query zapewnia wygodę: jeśli węzeł pośredni nie zdołał przekazać oryginalnego query w stanie, narzędzie wykonywania powraca do stanu początkowego zapisanego na początku wykonywania, dzięki czemu każdy agent nadal może otrzymać pytanie użytkownika.
Wymiana węzła na sesję
Traktowanie grafu jako danych przynosi ponownie korzyści przy krokach z udziałem człowieka. Jedyny węzeł, human_agent, zatrzymuje działanie, wysyła do użytkownika pytanie wyjaśniające przez WebSocket i czeka na odpowiedź. Problem polega na tym, że każda sesja ma swój własny sokiet, co nie może zostać rozwiązane poprzez skompilowanie grafu raz na starcie. Ponieważ agenci to po prostu elementy słownika self.agent_fns, narzędzie to wymienia implementację tego jednego węzła na każdą aktywną sesję, instalując obsługę powiązaną z połączeniem danego użytkownika:
# 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)
)
Topologia pozostaje niezmieniona i z góry skompilowana, podczas gdy zachowanie jednego węzła jest zastępowane na żądanie. Stabilna struktura z wymiennymi implementacjami powstaje w ramach projektu niemal bez żadnych kosztów.
Co się zmienia
Porównaj obie metody pod kątem różnych aspektów:
- Dodawanie agenta: edycja narzędzia do budowania grafu, plików importowych i testów, w porównaniu z trzema linijkami kodu YAML.
- Przekształcanie struktury przepływu: przepisanie wywołań
add_edgeiadd_conditional_edges, w porównaniu z edycją funkcjinextlubedges. - Schemat stanu: ręcznie napisany
TypedDictutrzymywany w synchronizacji dzięki dyscyplinie, w porównaniu z tym generowanym na podstawiestate_schema. - Logika routingu: instrukcje
if/elifukryte w kodzie grafu, w porównaniu z funkcją zwracającą etykietę oraz mapą w formacie YAML. - Bezpieczeństwo stanu: konwencja mówiąca, że agenci nie powinni edytować określonych kluczy, w przeciwieństwie do egzekwowania tego poprzez filtrowanie operacji
writes. - Diagram architektury: rysowany ręcznie i szybko stający się przestarzały, w przeciwieństwie do tego, który jest generowany automatycznie na podstawie żywego grafu.
- Kto może zmieniać przepływ: osoba biegle znająca API grafu, w przeciwieństwie do kogokolwiek, kto potrafi czytać pliki YAML.
Ostatni punkt jest najważniejszy. Specyfikacja przepływu pracy, którą mogą czytać menedżerowie produktu, którą osoba nieznająca LangGraph może sprawdzić w prośbie o połączenie, a której różnice są przedstawiane w formie zwykłego tekstu – to oznacza, że sama różnica stanowi zmianę. Nikt nie musi w myślach wykonywać 40 linii przearanżowanego kodu Pythona.
Frameworki wieloagentowe dostarczają węzły i krawędzie oraz mechanizm działania, ale nie oferują uporządkowanego miejsca na zapis polityki, co oznacza, że struktura procesu jest oddzielona od tego, co robi każdy krok. Jeśli graf agentów stał się tak skomplikowany, że wszyscy boją się go edytować, spróbuj przenieść topologię do postaci danych. Plik ładowania składa się z zaledwie kilkuset linijek: dynamicznego obiektu TypedDict, modułu importlib do obsługi wtyczek, wywołania dostosowanego do specyfiki sygnału oraz pętli, które przekształcają węzły i krawędzie na odpowiednie wywołania w grafie. Po miesiącu, gdy ktoś chce dodać nowy krok przed analizą, rozwiązaniem jest trzylinijowy plik YAML z różnicami, bez straty czasu na edycję w narzędziu do budowania grafów.
Literatura pokrewna
- Rozdzielanie zapytań klientów za pomocą wielu agentów z LangGraph i Ollama — Stwórz grafikę do rozdzielania zapytań z trzema agentami przy użyciu LangGraph: klasyfikatora, osoby odpowiadającej oraz człowieka kontrolującego, działających na lokalnym Ollama lub przez API przy użyciu zaledwie jednej zmienionej linijki kodu.
- Zespół wielu agentów z funkcją self-healing w LangGraph: architekt, tester, debugger — Stwórz maszynę stanową w LangGraph w języku Python, w której architekt, programista, środowisko testowe pytest, debugger oraz osoba sprawdzająca współpracują ze sobą w pętli, aż testy się udają.