Ihr Agenten-Graph gehört in YAML: Kompilierung eines 17-Agenten-LangGraphs beim Starten
Ihr Agenten-Graph gehört in YAML: Ein kleiner Laufzeit-Compiler wandelt eine Spezifikation in einen 17-Agenten-LangGraph-StateGraph um, inklusive generierter Zustände, Plugins und Label-Routern.
Quellen: Your Agent Graph Doesn’t Belong in Python: Compiling a Multi-Agent Workflow from a Single YAML File
LangGraph beginnt angenehm. Ein paar Knoten, ein oder zwei Kanten – sie werden kompiliert und ausgeführt. Dann wächst der Workflow auf ein Dutzend Agenten an, es taucht bedingte Routenbildung auf, und bald werden weitere Routen auf die bereits vorhandenen aufgebaut. Die Datei, die das Graph erstellt, umfasst 600 Zeilen mit graph.add_node(...) und graph.add_conditional_edges(...), und die Zustandsklasse ist stillschweigend zum wichtigsten und zugleich zerbrechlichsten Objekt im Codebasis geworden. Wenn jemand vor der Analyse um Klarstellungen bittet, muss man Python bearbeiten, Importe nachverfolgen und Verbindungen erneut testen, die nichts mit der geplanten Änderung zu tun haben.
Der Grund ist, dass die Struktur eines Workflows – also wer welche Aufgaben ausführt, in welcher Reihenfolge und wer entscheidet, was als Nächstes kommt – im Grunde eine Spezifikation darstellt. Hier ist sie in imperativen Code eingebettet und mit der Implementierung jedes Agenten vermischt. Die unten beschriebene Lösung trennt sie voneinander. Die gesamte Topologie befindet sich in einer deklarativen config.yaml-Datei; für ein 17-Agenten-System umfasst sie etwa 260 Zeilen. Ein kleiner Laufzeitkomponente kompiliert sie beim Start in einen echten LangGraph StateGraph. Das Hinzufügen eines Agenten oder das Umleiten des Flusses erfolgt durch eine Änderung im YAML, und der Python-Code enthält nur noch das, was jeder Agent ausführt.
Der Workflow als Datenstruktur
In der Spezifikation werden Knoten, Agenten, Router sowie das Zustandschema deklariert und nicht programmiert. Die Eingangs- und Ausgangspunkte bestehen aus zwei einfachen Zeilen:
entry: entry_agent
exit: exit
Und der größere Teil beschreibt Schutzmechanismen, das Zustandschema, Agenten, Router und Knoten:
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
Nehmen Sie einen Router-Block. Ein bedingter Zweig, normalerweise eine Ansammlung von if state["route"] == ...-Prüfungen, reduziert sich auf eine Funktion, die ein Label sowie eine edges-Karte von jedem Label zu seinem nächsten Knoten zurückgibt – insgesamt drei Zeilen. Die Routing-Regeln bleiben als Code erhalten, während die Topologie des Routings zu Daten wird.
Erzeugung der State-Klasse zur Laufzeit
LangGraph benötigt einen typisierten State, normalerweise ein manuell geschriebener TypedDict, der mit der Realität übereinstimmen muss. Hier wird er aus dem state_schema-Block erzeugt:
# 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
Das Ergebnis wird an die Spezifikation übergeben:
spec = Spec(
...
state=TypedDict("State", annotations), # <- dynamic class, born at boot
...
)
Zwei Dinge geschehen gleichzeitig. Die funktionale Form TypedDict("State", annotations) erzeugt einen neuen Typ aus einem Dictionary mit Feldnamen und -typen, sodass der Zustand bei jedem Start vom Loader erstellt wird anstelle dessen manuell gepflegt zu werden. Gleichzeitig integriert sich annotated_with: add_messages in die Reduzierer von LangGraph, wobei ein Feld über verschiedene Knoten hinweg akkumuliert wird (zum Beispiel Chatverlauf) und nicht überschrieben wird. Die Angabe in YAML erspart demjenigen, der das Gespräch konzipiert, das Wissen darüber, was Annotated[list, add_messages] bedeutet.
Eine Warnung bezüglich eval: Der Loader verwendet py_type = eval(type_str), wodurch eine Zeichenkette aus einer Konfigurationsdatei ausgewertet wird. Das ist nur zulässig, weil config.yaml als vertrauenswürdig gilt und von den Entwicklern gesteuert wird. Sollten unzuverlässige Parteien das Schema liefern können, sollte es durch eine explizite Erlaubnisliste wie {"str": str, "list": list, "dict": dict, "int": int} ersetzt werden – das ist eine einzeilige Änderung.
Der Vorteil besteht darin, dass die Definition des Zustands und die Topologie der Agenten dieselbe Datei teilen, sodass sie gemeinsam überprüft und geändert werden können. Der Fehler, bei dem ein Feld dem Graph hinzugefügt wurde, aber nicht dem TypedDict, verschwindet dadurch.
Agenten als punktierte Zeichenketten
In YAML ist jeder Agent lediglich eine Zeichenkette:
impl: your_package.agents.agent_one.agent_one_fn
Keine Import-Anweisung verknüpft das Diagramm mit dieser Funktion. Der Loader löst es mit importlib auf:
# 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}
Das ist das Plugin-Muster. Um einen Agent hinzuzufügen, schreiben Sie die Funktion und fügen drei YAML-Zeilen hinzu, die auf diese verweisen – ohne den Graph-Builder zu berühren. Das Framework findet Ihren Code zur Laufzeit nach seinem Namen, genauso wie ein Plugin-Host es tut. Router werden auf exakt dieselbe Weise über router_impl_map gelöst.
Der Compiler
Der Kern ist GraphRunner.build(), welcher die Spezifikation durchläuft und den eigentlichen StateGraph erstellt:
# 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
Mehrere Punkte verdienen Aufmerksamkeit. Der Schlüssel writes fungiert als Zugriffssteuerungsliste und ist nicht lediglich Dokumentation. Wenn ein Knoten writes: [query, result] deklariert, reduziert der Compiler das von dem Agenten zurückgegebene Dictionary auf genau diese Schlüssel, sodass ein Agent keine Zustandsfelder verfälschen kann, die er nicht deklariert hat. Die Spezifikation wird somit zu einem Vertrag, der tatsächlich durchgesetzt wird. Es gibt drei deklarative Randtypen: explizite edges, die inline-Schreibweise next: sowie terminal: true, welches automatisch mit LangGraphs END verbunden wird. Nach der Kompilierung rendernt der Ausführer außerdem das Diagramm mithilfe von draw_mermaid_png() als PNG, sodass das Diagramm nicht veraltet – es stammt aus derselben Quelle, die in der Produktion genutzt wird.
Router zeigen den Vorteil des Übergangs vom imperativen zum deklarativen Stil am deutlichsten:
# 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)
Eine Router-Funktion gibt eine Beschriftung zurück, einen einfachen String wie "next_agent", und LangGraphs add_conditional_edges sucht diese in dem YAML-Map unter node.router.edges nach, um den nächsten Knoten zu finden. Der Router kennt die Ergebnisse, nicht die Knoten IDs; daher bedeutet das Ändern des Ziels eines Ergebnisses, das edges-Map zu bearbeiten, ohne den Router-Code zu verändern. Ein Router kann auch (updates, label) zurückgeben, wodurch der Zustand angepasst und in einer einzigen Rückgabevorgang eine Entscheidung getroffen wird, ohne dass ein separater Knoten zur Aufzeichnung der Entscheidung benötigt wird.
Adaptive Aufrufe
Wenn 17 Agenten von verschiedenen Personen zu unterschiedlichen Zeiten geschrieben werden, sind die Signaturen ungleichmäßig. Ein Entwickler schreibt async def agent(state):, ein anderer async def agent(query):, ein Dritter async def agent(**kwargs):. Ein normales Diagramm würde Konformität erzwingen oder manuell geschriebene Adapter benötigen. Hier prüft der Ausführer zur Laufzeit die Signatur jeder Funktion:
# 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 gibt die von einer Funktion deklarierten Parameter preis, und der Ausführer stellt genau diesen Aufruf zusammen. Sowohl synchron als auch asynchron wird einheitlich umgegangen: Wenn das Ergebnis awaitbar ist, wird darauf gewartet, ansonsten wird es direkt verwendet. Die Autoren wählen die Signatur aus, die am natürlichsten klingt. Ein Hilfsfunktion _fallback_query bietet einen Vorteil: Wenn ein Zwischenknoten es versäumt, die ursprüngliche query im Zustand weiterzugeben, greift der Ausführer auf den bei Beginn der Ausführung erfassten Ausgangszustand zurück, sodass jeder Agent die Frage des Benutzers dennoch erhalten kann.
Austausch eines Knotens pro Sitzung
Die Behandlung des Graphen als Daten lohnt sich erneut bei Schritten mit menschlicher Beteiligung. Ein einzelner Knoten, human_agent, stoppt den Ablauf, stellt dem Benutzer über ein WebSocket eine klärende Frage und blockiert, bis eine Antwort eintrifft. Das Problem ist, dass jede Sitzung ihren eigenen Socket hat, der nicht in einen einmal bei der Startzeit kompilierten Graphen integriert werden kann. Da die Agenten lediglich Einträge im Dictionary self.agent_fns sind, wechselt der Ausführer für jede aktive Sitzung die Implementierung dieses einen Knotens aus und installiert einen Handler, der an die Verbindung dieses Benutzers gebunden ist:
# 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)
)
Die Topologie bleibt fest und vorkompiliert, während das Verhalten eines Knotens auf Anfrage ersetzt wird. Eine stabile Struktur mit austauschbaren Implementierungen entsteht so nahezu kostenlos im Rahmen des Designs.
Was ändert sich
Vergleichen Sie die beiden Ansätze nach Aspekten:
- Hinzufügen eines Agenten: Bearbeitung des Graphen-Bauers, der Importe und Tests – im Vergleich zu drei Zeilen YAML.
- Umformung des Ablaufs: Neuverfassen der Aufrufe von
add_edgeundadd_conditional_edges– im Vergleich zur Bearbeitung vonnextoderedges. - State-Schema: Ein manuell geschriebener
TypedDict, der diszipliniert synchron gehalten wird – im Vergleich zu einem ausstate_schemagenerierten Schema. - Routing-Logik:
if/elif-Anweisungen im Graphen-Code – im Vergleich zu einer Funktion, die ein Label zurückgibt, zusammen mit einer YAML-Kartei.
writes.Der letzte Punkt ist der wichtigste. Eine Workflow-Spezifikation, die Produktmanager lesen können, die ein Prüfer, der mit LangGraph nicht vertraut ist, in einem Pull Request überprüfen kann, und bei der die Differenzen als reiner Text dargestellt werden – das bedeutet, die Differenz selbst ist die Änderung. Niemand muss 40 Zeilen umorganisierten Python im Kopf ausführen.
Multi-Agent-Frameworks bieten Knoten und Kanten sowie das zugrundeliegende Mechanismus, bieten aber keinen strukturierten Ort für Richtlinien – wodurch die Struktur des Workflows von der Funktionsweise jeder einzelnen Schritt getrennt bleibt. Wenn ein Agentengraph zu einem komplexen Netzwerk wird, das alle scheuen, anzufassen, sollte man versuchen, die Topologie in Daten umzuwandeln. Der Lader benötigt nur ein paar hundert Zeilen: ein dynamischer TypedDict, importlib für Plugins, eine auf die Signatur angepasste Aufrufstruktur sowie Schleifen, die Knoten und Kanten in Graphenaufrufe umwandeln. Nach einem Monat, wenn jemand einen Schritt vor der Analyse hinzufügen möchte, reicht eine dreizeilige YAML-Diff-Datei aus – ohne dass Zeit im Graphenbauer verschwendet wird.
Verwandte Literatur
- Mehr-Agenten-System zur Sortierung von Support-Anfragen mit LangGraph und Ollama — Erstellen Sie ein dreifach-Agenten-System zur Sortierung von Anfragen mithilfe von LangGraph: ein Klassifikator, ein Antwortgeber und ein menschlicher Prüfer, das auf lokalem Ollama oder über eine API mit nur einer geänderten Zeile läuft.
- Selbst-heilendes Mehr-Agenten-Team in LangGraph: Architekt, Tester, Debugger — Erstellen Sie eine Zustandsmaschine in Python mit LangGraph, in der ein Architekt, ein Entwickler, ein pytest-Sandbox-Umfeld, ein Debugger und ein Prüfer zusammenarbeiten und solange Schleifen laufen, bis die Tests bestehen.