Su grafo de agentes debe estar en YAML: Compilación de un LangGraph con 17 agentes al iniciar
Tu grafo de agentes debe estar en YAML: un pequeño motor en tiempo de ejecución compila una especificación en un LangGraph StateGraph con 17 agentes, incluyendo el estado generado, plugins y enrutadores de etiquetas.
Fuentes: Your Agent Graph Doesn’t Belong in Python: Compiling a Multi-Agent Workflow from a Single YAML File
LangGraph comienza de manera agradable: unos pocos nodos, uno o dos enlaces, se compilan y se ejecutan. Luego el flujo de trabajo crece hasta incluir una docena de agentes, aparece el enrutamiento condicional, y pronto se añaden más capas de enrutamiento. El archivo que construye el grafo llega a tener 600 líneas de graph.add_node(...) y graph.add_conditional_edges(...), y la clase de estado se ha convertido silenciosamente en el objeto más importante y frágil del código. Cuando alguien solicita una aclaración antes del análisis, uno se ve obligado a editar el código en Python, buscar las importaciones adecuadas y volver a probar conexiones que no tienen nada que ver con el cambio solicitado.
La razón es que la estructura de un flujo de trabajo, es decir, quién lo ejecuta, en qué orden y quién decide qué viene a continuación, es en realidad una especificación. Actualmente está incrustada dentro del código imperativo, mezclada con la implementación de cada agente. La solución descrita a continuación las separa. Toda la topología se encuentra en un único archivo declarativo config.yaml; un sistema con 17 agentes tiene unas 260 líneas, y un pequeño programa en tiempo de ejecución lo compila en un verdadero StateGraph de LangGraph al iniciar. Agregar un agente o redirigir el flujo consiste en editar el archivo YAML, mientras que el código en Python contiene únicamente lo que hace cada agente.
El flujo de trabajo como datos
En la especificación, los nodos, los agentes, los enrutadores y el esquema de estado se declaran y no se codifican. Los puntos de entrada y salida son simplemente dos líneas:
entry: entry_agent
exit: exit
Y la mayor parte describe las barreras de seguridad, el esquema del estado, los agentes, los enrutadores y los nodos:
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
Tome un bloque de enrutamiento. Un ramal condicional, que normalmente consiste en una serie de comprobaciones del tipo if state["route"] == ..., se reduce a una función que devuelve una etiqueta junto con un mapa edges que relaciona cada etiqueta con su nodo siguiente; en total, tres líneas. La política de enrutamiento permanece como código, mientras que la topología de enrutamiento se convierte en datos.
Generación de la clase de estado en tiempo de ejecución
LangGraph requiere un estado tipado, normalmente un TypedDict escrito a mano que debe mantenerse alineado con la realidad. En este caso, se genera a partir del bloque 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
El resultado se entrega a la especificación:
spec = Spec(
...
state=TypedDict("State", annotations), # <- dynamic class, born at boot
...
)
Están ocurriendo dos cosas. La forma funcional TypedDict("State", annotations) crea un nuevo tipo a partir de un diccionario con nombres y tipos de campos, de modo que el estado se construye mediante el cargador en cada inicio de sesión en lugar de ser mantenido manualmente. Mientras tanto, annotated_with: add_messages se integra en los redactores de LangGraph, donde un campo se acumula a lo largo de los nodos (por ejemplo, el historial de chat) y no se sobrescribe. Declarar esto en YAML evita que quien diseña la conversación tenga que saber qué significa Annotated[list, add_messages].
Advertencia sobre eval: el cargador utiliza py_type = eval(type_str), lo cual evalúa una cadena proveniente de un archivo de configuración. Esto es aceptable únicamente porque config.yaml es de confianza y está bajo control de los desarrolladores. Si partes no confiables pudieran proporcionar el esquema, hay que reemplazarlo por una lista de permisos explícita como {"str": str, "list": list, "dict": dict, "int": int}, lo cual es un cambio de una sola línea.
La ventaja es que la definición del estado y la topología de los agentes comparten un mismo archivo, por lo que se revisan y modifican juntos. El error en el que se añadía un campo al grafo pero no al TypedDict desaparece.
Agentes como cadenas con puntos
En YAML, cada agente es simplemente una cadena:
impl: your_package.agents.agent_one.agent_one_fn
Ninguna instrucción import vincula el grafo a esa función. El cargador la resuelve con 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}
Este es el patrón de plugin. Para agregar un agente, se escribe la función y se añaden tres líneas YAML que apuntan a ella, sin tocar nunca el generador de gráficos; el framework encuentra tu código por su nombre en tiempo de ejecución, al igual que lo haría un anfitrión de plugins. Los enrutadores se resuelven de exactamente la misma manera a través de router_impl_map.
El compilador
El núcleo es GraphRunner.build(), que recorre la especificación y construye el StateGraph real:
# 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
Algunos puntos merecen atención. La clave writes funciona como una lista de control de acceso y no es simplemente documentación. Cuando un nodo declara writes: [query, result], el compilador reduce el diccionario devuelto por el agente a esos campos específicos, de modo que el agente no pueda corromper campos de estado que no haya declarado. La especificación se convierte así en un contrato que realmente se aplica. Existen tres tipos de bordes declarativos: edges explícitos, la abreviatura next: inline, y terminal: true, que se conecta automáticamente al END de LangGraph. Después de compilar, el ejecutor también genera el gráfico en un archivo PNG mediante draw_mermaid_png(), por lo que el diagrama no se vuelve obsoleto: proviene de la misma fuente que se utiliza en producción.
Los enrutadores muestran de la manera más clara la ventaja de pasar de un enfoque imperativo a uno declarativo:
# 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)
La función del enrutador devuelve una etiqueta, una cadena simple como "next_agent", y add_conditional_edges de LangGraph la busca en el mapa YAML bajo node.router.edges para encontrar el nodo siguiente. El enrutador conoce los resultados, no los IDs de los nodos, por lo que cambiar a dónde conduce un resultado implica editar el mapa edges sin tocar el código del enrutador. Un enrutador también puede devolver (updates, label), ajustando el estado y eligiendo una rama en una sola llamada, sin necesidad de un nodo separado para registrar la decisión.
Llamadas adaptativas
Cuando 17 agentes son escritos por personas diferentes en momentos distintos, las firmas serán inconsistentes. Un desarrollador escribe async def agent(state):, otro async def agent(query):, y un tercero async def agent(**kwargs):. Un grafo normal obligaría a la conformidad o requeriría adaptadores manuales. Aquí, el ejecutor inspecciona la firma de cada función en el momento de la llamada:
# 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 muestra los parámetros que declara una función, y el ejecutor forma exactamente esa llamada. Tanto los casos síncronos como asíncronos se manejan de manera uniforme: si el resultado es esperable se espera, de lo contrario se utiliza directamente. Los autores escriben la firma que les resulte más natural. Una función auxiliar _fallback_query brinda comodidad: si un nodo intermedio no logra transmitir el query original en el estado, el ejecutor recurre al estado inicial capturado al comenzar la ejecución, de modo que cualquier agente pueda seguir obteniendo la pregunta del usuario.
Intercambio de nodo por sesión
Tratar el grafo como datos resulta útil nuevamente con pasos que involucran a un humano. Un único nodo, human_agent, detiene la ejecución, envía una pregunta de aclaración al usuario a través de un WebSocket y se bloquea hasta que llegue una respuesta. El problema es que cada sesión tiene su propio socket, lo cual no puede integrarse en un grafo compilado una sola vez al iniciar el programa. Dado que los agentes son simplemente entradas en el diccionario self.agent_fns, el ejecutor reemplaza la implementación de ese nodo por una distinta para cada sesión en curso, instalando un manejador vinculado a la conexión de ese usuario:
# 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 topología permanece fija y precompilada, mientras que el comportamiento de un nodo se reemplaza según cada solicitud. Una estructura estable con implementaciones reemplazables se obtiene prácticamente sin costo alguno en el diseño.
Qué cambia
Compare los dos enfoques por área de consideración:
- Agregar un agente: editar el generador de grafos, las importaciones y las pruebas, frente a tres líneas de YAML.
- Reformar el flujo: reescribir las llamadas a
add_edgeyadd_conditional_edges, frente a editarnextoedges. - Estructura de estado: un
TypedDictescrito a mano mantenido sincronizado mediante disciplina, frente a uno generado a partir destate_schema. - Lógica de enrutamiento:
if/elifincrustados en el código del grafo, frente a una función que devuelve una etiqueta más un mapa en YAML. - Seguridad del estado: una convención que establece que los agentes no deben escribir ciertas claves, frente a su aplicación mediante el filtrado de
writes. - Diagrama de arquitectura: dibujado a mano y que pronto se vuelve obsoleto, frente a su generación automática a partir del grafo en tiempo real.
- Quién puede cambiar el flujo: alguien versado en la API del grafo, frente a cualquiera que pueda leer YAML.
El último punto es el más importante. Una especificación de flujo de trabajo que los gerentes de producto puedan leer, que un revisor inexperto en LangGraph pueda verificar en una solicitud de integración y cuyas diferencias se muestren en texto plano, lo que significa que la propia diferencia representa el cambio. Nadie necesita ejecutar mentalmente 40 líneas de código Python reorganizado.
Los marcos multiagente proporcionan nodos y aristas, el mecanismo, pero no un lugar organizado para las políticas, lo que significa que la estructura del flujo de trabajo está separada de lo que hace cada paso. Si el grafo de agentes se ha convertido en un conjunto de conexiones que todos temen tocar, intente extraer la topología en forma de datos. El cargador consta de solo unas pocascientas líneas: un TypedDict dinámico, importlib para los complementos, una llamada adaptable a las firmas, y bucles que convierten nodos y aristas en llamadas al grafo. Un mes después, cuando alguien desea agregar un paso antes del análisis, la solución es un dif de YAML de tres líneas, sin perder tiempo por la tarde en el generador de grafos.
Leer más
- Triaje de tiquetes de soporte multiagente con LangGraph y Ollama — Crea un grafo de triaje de tiquetes con tres agentes utilizando LangGraph: un clasificador, un respondiente y un control humano, que funcionan con Ollama local o a través de una API, con solo una línea modificada.
- Equipo multiagente autoreparable en LangGraph: Arquitecto, Probador, Depurador — Diseña una máquina de estados en LangGraph con Python en la que un Arquitecto, un Desarrollador, un entorno de pruebas pytest, un Depurador y un Revisor colaboran en un ciclo hasta que las pruebas se aprueben.