Votre graphe d’agents doit être en YAML : compilation d’un LangGraph à 17 agents au démarrage
Votre graphe d’agent doit être au format YAML : un petit outil de exécution compile une spécification en un LangGraph StateGraph de 17 agents, avec des états générés, des plugins et des routeurs d’étiquettes.
Sources: Your Agent Graph Doesn’t Belong in Python: Compiling a Multi-Agent Workflow from a Single YAML File
LangGraph débute de manière agréable : quelques nœuds, un ou deux arêtes, tout se compile et fonctionne. Puis le flux de travail s’élargit avec une dizaine d’agents, la routage conditionnel apparaît, et bientôt d’autres mécanismes de routage s’ajoutent par-dessus. Le fichier qui construit le graphe atteint 600 lignes de graph.add_node(...) et graph.add_conditional_edges(...), et la classe de statut est devenue silencieusement l’objet le plus important — mais aussi le plus fragile — de l’ensemble du code. Lorsqu’on demande des précisions avant l’analyse, il faut modifier du Python, gérer les imports et rétester des connexions qui n’ont rien à voir avec la modification demandée.
La raison en est que la structure d’un flux de travail, c’est-à-dire qui exécute les tâches, dans quel ordre et qui décide de la suite, constitue en réalité une spécification. Actuellement, elle est intégrée dans du code impératif, mêlée à l’implémentation de chaque agent. La solution décrite ci-dessous les sépare. Toute la topologie est contenue dans un seul fichier déclaratif config.yaml ; pour un système de 17 agents, il compte environ 260 lignes, et un petit programme de runtime le compile en un véritable StateGraph LangGraph au démarrage. Ajouter un agent ou rediriger le flux se fait par une modification du fichier YAML, tandis que le code Python ne contient que ce que chaque agent doit faire.
Le flux de travail en tant que données
Dans la spécification, les nœuds, les agents, les routeurs ainsi que le schéma d’état sont déclarés et non codés. Les points d’entrée et de sortie ne sont que deux lignes simples :
entry: entry_agent
exit: exit
Et la majeure partie décrit les garde-fous, le schéma d’état, les agents, les routeurs et les nœuds :
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
Prenez un bloc de routage. Une branche conditionnelle, généralement constituée d’une série de vérifications du type if state["route"] == ..., se transforme en une fonction qui renvoie une étiquette ainsi qu’un tableau edges reliant chaque étiquette à son nœud suivant, soit au total trois lignes de code. La politique de routage reste sous forme de code, tandis que la topologie du routage devient des données.
Génération de la classe d’état en temps de exécution
LangGraph nécessite une étiquette de type défini, généralement un objet TypedDict écrit manuellement qui doit correspondre à la réalité. Ici, elle est générée à partir du bloc 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
Le résultat est transmis à la spécification :
spec = Spec(
...
state=TypedDict("State", annotations), # <- dynamic class, born at boot
...
)
Deux choses se produisent. La forme fonctionnelle TypedDict("State", annotations) crée un nouveau type à partir d’un dictionnaire de noms de champs et de types, de sorte que l’état est généré par le chargeur à chaque démarrage plutôt que géré manuellement. Parallèlement, annotated_with: add_messages s’intègre aux réducteurs de LangGraph, où un champ s’accumule d’un nœud à l’autre (par exemple, l’historique de chat) sans être écrasé. Déclarer cela en YAML permet à celui qui conçoit la conversation d’éviter de devoir comprendre ce que signifie Annotated[list, add_messages].
Précaution concernant eval : le chargeur utilise py_type = eval(type_str), ce qui évalue une chaîne provenant d’un fichier de configuration. Cela est acceptable uniquement parce que config.yaml est fiable et géré par les développeurs. Si des parties non fiables pouvaient fournir le schéma, il conviendrait de le remplacer par une liste d’autorisation explicite telle que {"str": str, "list": list, "dict": dict, "int": int}, ce qui ne nécessite qu’un changement en une seule ligne.
L’avantage est que la définition de l’état et la topologie des agents partagent un même fichier, ce qui permet de les examiner et de les modifier ensemble. Le bug où un champ était ajouté au graphe mais pas au TypedDict disparaît ainsi.
Agents en tant que chaînes ponctuées
Dans le YAML, chaque agent n’est qu’une chaîne de caractères :
impl: your_package.agents.agent_one.agent_one_fn
Aucune instruction import ne lie le graphe à cette fonction. Le chargeur la résout à l’aide de 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}
C’est le modèle des plugins. Pour ajouter un agent, vous écrivez la fonction correspondante et ajoutez trois lignes YAML qui la désignent, sans jamais toucher au constructeur de graphe ; le framework trouve alors votre code par son nom en temps de exécution, tout comme le hôte de plugins. Les routeurs sont résolus de la même manière via router_impl_map.
Le compilateur
Le cœur du système est GraphRunner.build(), qui parcourt la spécification et crée le véritable 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
Quelques points méritent attention. La clé writes fonctionne comme une liste de contrôle d’accès et n’est pas simplement de la documentation. Lorsqu’un nœud déclare writes: [query, result], le compilateur réduit le dictionnaire retourné par l’agent aux seules clés indiquées, de sorte qu’un agent ne peut pas altérer des champs d’état qu’il n’a pas déclarés. La spécification devient ainsi un contrat réellement appliqué. Il existe trois types d’arêtes déclaratives : les edges explicites, l’abréviation next: intégrée, et terminal: true, qui se connecte automatiquement à END de LangGraph. Après compilation, l’exécuteur génère également un PNG du graphe à l’aide de draw_mermaid_png(), de manière à ce que le diagramme reste à jour : il provient de la même source que celle utilisée en production.
Les routeurs montrent le plus clairement les avantages du passage de l’impératif au déclaratif :
# 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)
Une fonction de routage renvoie une étiquette, une chaîne de caractères simple telle que "next_agent", et add_conditional_edges de LangGraph la recherche dans la carte YAML située sous node.router.edges afin de trouver le nœud suivant. Le routage connaît les résultats, et non les IDs des nœuds ; par conséquent, pour modifier la destination d’un résultat, il suffit d’éditer la carte edges sans toucher au code du routage. Un routage peut également renvoyer (updates, label), ajustant l’état et choisissant une branche en une seule opération, sans avoir besoin d’un nœud distinct pour enregistrer la décision.
Appels adaptatifs
Lorsque 17 agents sont écrits par différentes personnes à des moments différents, les signatures seront incohérentes. Un développeur écrit async def agent(state):, un autre async def agent(query):, et un troisième async def agent(**kwargs):. Un graphe normal obligerait à une conformité ou à l’utilisation d’adaptateurs manuels. Ici, l’exécuteur examine la signature de chaque fonction au moment de l’appel :
# 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 révèle les paramètres déclarés par une fonction, et l’exécuteur assemble précisément cette appelation. Le traitement des opérations synchrones et asynchrones est uniforme : si le résultat est attendable, on l’attend, sinon on l’utilise directement. Les auteurs choisissent la signature qui semble la plus naturelle. Un outil _fallback_query offre une commodité : si un nœud intermédiaire échoue à transmettre le query d’origine dans l’état, l’exécuteur recourt à l’état de départ enregistré au début de l’exécution, afin que tout agent puisse toujours obtenir la question de l’utilisateur.
Changement de nœud par session
Considérer le graphe comme des données s’avère à nouveau avantageux avec des étapes impliquant un intervenant humain. Un seul nœud, human_agent, arrête l’exécution, pose une question de clarification à l’utilisateur via WebSocket et bloque jusqu’à ce qu’une réponse arrive. Le problème est que chaque session dispose de son propre socket, ce qui ne peut pas être intégré dans un graphe compilé une seule fois au démarrage. Comme les agents ne sont que des entrées dans le dictionnaire self.agent_fns, le moteur d’exécution remplace l’implémentation de ce nœud pour chaque session en cours, en installant un gestionnaire lié à la connexion de cet utilisateur :
# 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)
)
La topologie reste fixe et précompilée, tandis que le comportement d’un nœud est remplacé à chaque demande. Une structure stable avec des implémentations remplaçables découle presque gratuitement du design.
Quelles modifications ?
Comparez les deux approches par domaine d’application :
- Ajout d’un agent : modification du générateur de graphes, des imports et des tests, contre trois lignes de YAML.
- Redéfinition du flux : réécriture des appels à
add_edgeetadd_conditional_edges, contre modification denextouedges. - Schéma d’état : un
TypedDictmanuellement écrit, maintenu à jour avec rigueur, contre un schéma généré à partir destate_schema. - Logique de routage : instructions
if/elifintégrées dans le code du graphique, contre une fonction retournant une étiquette accompagnée d’un tableau YAML.
writes.Le dernier point est le plus important. Une spécification de flux de travail que les gestionnaires de produit peuvent lire, que quelqu’un peu familier avec LangGraph peut vérifier dans une demande de fusion, et dont les différences sont présentées en texte brut, ce qui signifie que la différence elle-même constitue le changement. Personne n’a besoin d’exécuter mentalement 40 lignes de code Python réorganisé.
Les frameworks multi-agents fournissent des nœuds et des arêtes, ainsi que le mécanisme de fonctionnement, mais pas d’espace organisé pour définir les politiques, ce qui signifie que la structure du flux de travail est séparée de ce que fait chaque étape. Si le graphe des agents est devenu un réseau complexe que tout le monde craint de modifier, essayez de transférer sa topologie en données. Le chargeur ne nécessite que quelques centaines de lignes : un TypedDict dynamique, importlib pour les plugins, une fonctionnalité d’appel adaptative aux signatures, ainsi que des boucles qui transforment les nœuds et les arêtes en appels graphiques. Un mois plus tard, lorsque quelqu’un souhaite ajouter une étape avant l’analyse, la solution se résume à un diff YAML de trois lignes, sans perte de temps dans l’outil de création de graphes.
Lectures complémentaires
- Tri des demandes de support multijoueur avec LangGraph et Ollama — Créez un graphe de tri de demandes utilisant trois agents avec LangGraph : un classificateur, un répondant et un contrôleur humain, fonctionnant sur Ollama local ou via une API, avec une seule ligne modifiée.
- Équipe multijoueur auto-réparatrice dans LangGraph : Architecte, Testeur, Débogueur — Construisez une machine à états LangGraph en Python où un Architecte, un Développeur, un environnement de test pytest, un Débogueur et un Réviseur collaborent et itèrent jusqu’à ce que les tests réussissent.