Más allá del agente único: infraestructura para varios agentes
Crear un agente que llame a un modelo de lenguaje y use un par de herramientas es relativamente sencillo. Lo difícil empieza cuando ese prototipo debe sobrevivir a un reinicio sin perder el trabajo, pedir la aprobación de una persona antes de continuar, acceder a herramientas de forma estándar y colaborar con agentes escritos en otros frameworks. En ese punto, el problema deja de ser de prompts y pasa a ser de infraestructura.
Tres piezas se han consolidado para resolverlo. LangGraph orquesta flujos con estado como grafos, con persistencia y puntos de interrupción integrados. El Model Context Protocol (MCP) estandariza cómo un agente descubre y usa herramientas y datos. El protocolo Agent2Agent (A2A) define cómo se comunican agentes independientes, aunque estén construidos con tecnologías distintas.
Esta guía construye paso a paso un sistema de cuatro agentes que funciona por completo en local, sin claves de API ni servicios en la nube, usando Ollama como motor de inferencia. El caso práctico es un asistente de aprendizaje, pero los patrones se trasladan sin cambios a soporte al cliente, incorporación de empleados o formación en cumplimiento normativo. Las versiones y APIs se han actualizado a las publicadas al momento de escribir, incluidos cambios importantes en MCP y A2A que rompen la compatibilidad con ejemplos anteriores.
Cuándo merece la pena dividir el trabajo entre agentes
Antes de diseñar un sistema multiagente conviene preguntarse si de verdad hace falta. Un único agente con buenas herramientas resuelve muchísimos problemas y es más fácil de depurar, más barato y más rápido. Añadir agentes multiplica la latencia, complica el estado compartido y crea nuevos puntos de fallo.
La división tiene sentido cuando las subtareas difieren de forma significativa. Algunas señales claras son las siguientes: las tareas necesitan herramientas diferentes, por ejemplo una lee archivos y otra solo conversa; requieren configuraciones de modelo distintas, como una temperatura baja para planificar con precisión y otra más alta para redactar con variedad; conviene aislar fallos para que un error en una parte no derribe el resto; o una parte la mantiene otro equipo con otro framework y se despliega por separado.
En el caso práctico, cada agente cumple una de esas condiciones. El planificador necesita respuestas deterministas en JSON, el explicador usa herramientas de archivos y memoria, el generador de cuestionarios hace dos llamadas con temperaturas distintas y el entrenador de progreso decide el rumbo del flujo. Si todas esas responsabilidades vivieran en un solo prompt, el resultado sería frágil y difícil de probar.
El caso práctico: un asistente de aprendizaje
El sistema recibe un objetivo, como aprender los fundamentos de Kubernetes, y lo convierte en una ruta de estudio personalizada a partir de las notas del usuario. El flujo completo es el siguiente:
- El planificador de currículo genera una ruta con varios temas ordenados.
- Una persona revisa la ruta y la aprueba o la rechaza para que se regenere.
- El explicador busca en las notas locales y explica el tema actual.
- El generador de cuestionarios formula preguntas y califica las respuestas.
- El entrenador de progreso registra el resultado, detecta áreas débiles y decide si se pasa al siguiente tema o se termina.
Las herramientas de archivos y memoria se exponen mediante servidores MCP, el generador de cuestionarios puede publicarse como servicio A2A para que otros agentes lo consuman, y todo el estado se guarda en SQLite para poder reanudar tras una caída.
Entorno de trabajo y versiones actuales
El proyecto usa Python 3.11 o superior y la siguiente estructura:
asistente-aprendizaje/
├── requirements.txt
├── .env
├── notas/
│ ├── kubernetes-pods.md
│ └── kubernetes-servicios.md
└── src/
├── estado.py
├── agentes/
│ ├── planificador.py
│ ├── explicador.py
│ ├── cuestionario.py
│ └── entrenador.py
├── grafo/
│ └── flujo.py
├── mcp_servers/
│ ├── notas.py
│ └── memoria.py
├── a2a_services/
│ └── cuestionario_a2a.py
└── main.py
Las versiones han cambiado de forma notable en los últimos meses. LangGraph está en la serie 1.2 y su paquete de persistencia en SQLite en la 3.1. El SDK oficial de MCP para Python publicó su versión 2.0, que renombra la clase FastMCP a MCPServer. El SDK de A2A para Python alcanzó la versión 1.0 y después la serie 1.1, con una reorganización importante de su API de servidor. Un requirements.txt coherente con ese estado sería:
langgraph>=1.2,<1.3
langgraph-checkpoint-sqlite>=3.1,<3.2
langchain-core>=1.0
langchain-ollama>=1.0
mcp>=2.1,<3
a2a-sdk>=1.1,<2
uvicorn
python-dotenv
pytest
Fijar rangos de versiones es especialmente importante en este ecosistema, porque las librerías de agentes introducen cambios incompatibles con frecuencia. Si un proyecto existente depende del SDK de MCP 1.x, la propia documentación recomienda fijar mcp>=1.28,<2 hasta completar la migración, ya que la rama 1.x solo recibe correcciones de seguridad.
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
Inferencia local con Ollama
Ollama permite ejecutar modelos abiertos en la propia máquina y expone una API en el puerto 11434. Tras instalarlo, basta con descargar un modelo con buen soporte para llamadas a herramientas:
ollama pull qwen2.5:7b
ollama list
NAME SIZE MODIFIED
qwen2.5:7b 4.7 GB 5 seconds ago
Un modelo de unos 7 mil millones de parámetros es el mínimo razonable para este tipo de sistemas. Con modelos más pequeños, las llamadas a herramientas suelen generar JSON mal formado, lo que rompe el bucle del agente. Como referencia, un equipo con 16 GB de RAM y 8 GB de memoria de vídeo ejecuta bien un modelo de 7B; sin GPU funciona, pero varias veces más lento.
Desde Python, langchain-ollama proporciona el cliente. Conviene centralizar su creación para controlar temperatura y formato por agente:
import os
from langchain_ollama import ChatOllama
MODELO = os.getenv("OLLAMA_MODEL", "qwen2.5:7b")
def crear_llm(temperatura: float, json: bool = False) -> ChatOllama:
return ChatOllama(
model=MODELO,
temperature=temperatura,
format="json" if json else None,
)
El estado compartido del grafo
En LangGraph, todos los nodos leen y escriben un estado común. Definirlo con precisión es la decisión más importante del diseño, porque actúa como contrato entre agentes. Se declara con un TypedDict:
from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages
class EstadoAprendizaje(TypedDict, total=False):
mensajes: Annotated[list[AnyMessage], add_messages]
sesion_id: str
objetivo: str
ruta: dict
ruta_aprobada: bool
indice_tema: int
resultados: list[dict]
areas_debiles: list[str]
ruta_notas: str
error: str | None
La anotación add_messages es un reductor: indica que, cuando un nodo devuelve mensajes, estos se añaden a la lista existente en lugar de reemplazarla. El resto de claves se sobrescriben con el nuevo valor.
Para los datos estructurados internos es preferible usar clases de datos, que evitan errores tipográficos en las claves y documentan la forma esperada:
from dataclasses import dataclass, field, asdict
@dataclass
class Tema:
titulo: str
descripcion: str
estado: str = "pendiente"
@dataclass
class RutaEstudio:
objetivo: str
temas: list[Tema] = field(default_factory=list)
def a_dict(self) -> dict:
return asdict(self)
Nodos como funciones de estado
Cada nodo es una función que recibe el estado y devuelve únicamente las claves que modifica. LangGraph fusiona esa actualización parcial con el estado global. Separar la llamada al modelo, el análisis de la respuesta y la actualización del estado permite probar cada parte por separado.
El planificador hace una sola llamada con temperatura baja y formato JSON obligatorio:
import json
from langchain_core.messages import SystemMessage, HumanMessage
PROMPT_PLANIFICADOR = (
"Eres un planificador de estudio. Devuelve SOLO JSON con la forma "
'{"temas": [{"titulo": str, "descripcion": str}]} y entre 3 y 5 temas.'
)
def analizar_ruta(objetivo: str, texto: str) -> RutaEstudio:
datos = json.loads(texto)
temas = [Tema(t["titulo"], t["descripcion"]) for t in datos.get("temas", [])]
if not temas:
raise ValueError("El modelo no devolvió temas")
return RutaEstudio(objetivo=objetivo, temas=temas)
def nodo_planificador(estado: EstadoAprendizaje) -> dict:
llm = crear_llm(temperatura=0.1, json=True)
respuesta = llm.invoke([
SystemMessage(PROMPT_PLANIFICADOR),
HumanMessage(f"Objetivo: {estado['objetivo']}"),
])
ruta = analizar_ruta(estado["objetivo"], respuesta.content)
return {"ruta": ruta.a_dict(), "indice_tema": 0, "ruta_aprobada": False}
La función analizar_ruta no depende del modelo, de modo que puede probarse con entradas fijas, incluidas respuestas mal formadas, sin ejecutar Ollama:
import pytest
def test_ruta_vacia_falla():
with pytest.raises(ValueError):
analizar_ruta("Kubernetes", '{"temas": []}')
Construir el grafo con aristas fijas y condicionales
Con los nodos definidos, el grafo los conecta. Las aristas fijas indican una transición siempre igual; las condicionales llaman a una función que decide el siguiente nodo según el estado. La orquestación de agentes con LangGraph se apoya por completo en esta definición:
from langgraph.graph import StateGraph, START, END
def tras_aprobacion(estado: EstadoAprendizaje) -> str:
return "explicador" if estado.get("ruta_aprobada") else "planificador"
def tras_entrenador(estado: EstadoAprendizaje) -> str:
total = len(estado["ruta"]["temas"])
return "explicador" if estado["indice_tema"] < total else END
def construir_grafo(checkpointer):
grafo = StateGraph(EstadoAprendizaje)
grafo.add_node("planificador", nodo_planificador)
grafo.add_node("aprobacion", nodo_aprobacion)
grafo.add_node("explicador", nodo_explicador)
grafo.add_node("cuestionario", nodo_cuestionario)
grafo.add_node("entrenador", nodo_entrenador)
grafo.add_edge(START, "planificador")
grafo.add_edge("planificador", "aprobacion")
grafo.add_conditional_edges("aprobacion", tras_aprobacion)
grafo.add_edge("explicador", "cuestionario")
grafo.add_edge("cuestionario", "entrenador")
grafo.add_conditional_edges("entrenador", tras_entrenador)
return grafo.compile(checkpointer=checkpointer)
Un detalle de diseño importante es que el enrutamiento lo decide código Python, no el modelo. Delegar el control del flujo a un LLM introduce imprevisibilidad; aquí el modelo genera contenido y el grafo decide el camino de forma determinista. El bucle de temas tampoco necesita un while externo: el entrenador vuelve al explicador hasta agotar la ruta, y la topología del grafo expresa toda la lógica.
Persistencia con checkpoints en SQLite
Un flujo de aprendizaje puede durar horas y atravesar varias sesiones. Si el proceso se detiene a mitad, no debería perderse el progreso. LangGraph resuelve esto con checkpoints: tras cada nodo, guarda el estado completo asociado a un identificador de hilo.
Los checkpoints persistentes en SQLite se configuran con SqliteSaver. Hay un matiz práctico: la forma con gestor de contexto cierra la conexión al salir del bloque, lo que provoca errores si el grafo se usa después. Para una aplicación de larga duración es más seguro crear la conexión explícitamente y mantenerla abierta:
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
conexion = sqlite3.connect("checkpoints.db", check_same_thread=False)
checkpointer = SqliteSaver(conexion)
app = construir_grafo(checkpointer)
config = {"configurable": {"thread_id": "sesion-kubernetes-01"}}
Invocar de nuevo el grafo con el mismo thread_id reanuda desde el último checkpoint guardado. Además, es posible inspeccionar el estado de cualquier sesión, algo muy útil para depurar:
instantanea = app.get_state(config)
print(instantanea.values.get("indice_tema"), instantanea.next)
2 ('explicador',)
La salida indica que la sesión completó dos temas y que el siguiente paso pendiente es el explicador.
Aprobación humana con interrupt y Command
Muchos flujos de agentes necesitan supervisión humana en puntos críticos: aprobar un plan, confirmar una acción irreversible o corregir un resultado. La aprobación humana en el flujo se implementa en LangGraph con la función interrupt(), que pausa la ejecución, guarda el estado y devuelve una carga útil al llamador:
from langgraph.types import interrupt
def nodo_aprobacion(estado: EstadoAprendizaje) -> dict:
decision = interrupt({
"ruta": estado["ruta"],
"pregunta": "¿Apruebas esta ruta de estudio? (si/no)",
})
aprobada = str(decision).strip().lower() in {"si", "sí", "s", "yes"}
return {
"ruta_aprobada": aprobada,
"ruta": estado["ruta"],
"indice_tema": estado.get("indice_tema", 0),
"objetivo": estado["objetivo"],
}
El nodo devuelve explícitamente todas las claves que necesitan los nodos posteriores. Es una precaución deliberada: tras reanudar una interrupción, conviene que el nodo interrumpido entregue un estado completo para que los agentes siguientes no reciban datos parciales.
En el programa principal, un bucle detecta la interrupción, muestra la ruta, recoge la respuesta y reanuda con Command:
from langgraph.types import Command
resultado = app.invoke(
{"objetivo": "Fundamentos de Kubernetes", "sesion_id": "sesion-kubernetes-01"},
config,
)
while "__interrupt__" in resultado:
carga = resultado["__interrupt__"][0].value
for i, tema in enumerate(carga["ruta"]["temas"], start=1):
print(f"{i}. {tema['titulo']}: {tema['descripcion']}")
respuesta = input(carga["pregunta"] + " ")
resultado = app.invoke(Command(resume=respuesta), config)
Si la persona rechaza la ruta, el enrutamiento condicional vuelve al planificador, que genera una nueva propuesta y provoca otra interrupción. Una alternativa es interrupt_before, que se define al compilar y pausa antes de un nodo concreto sin carga útil; resulta práctica en interfaces web donde no se quiere ejecutar input() dentro del hilo del grafo.
MCP: un estándar para herramientas y datos
Sin un estándar, cada agente define sus herramientas a su manera y reutilizarlas entre proyectos exige reescribirlas. El protocolo de contexto de modelo propone un contrato común: un servidor expone capacidades y cualquier cliente compatible puede descubrirlas y usarlas.
MCP define tres primitivas. Las herramientas son acciones que el modelo puede invocar, como buscar en notas o guardar un dato. Los recursos son datos de solo lectura identificados por URI, como un índice de archivos. Los prompts son plantillas definidas por el servidor que el cliente puede solicitar.
La versión 2.0 del SDK de Python trajo cambios a tener en cuenta. La clase de alto nivel FastMCP pasó a llamarse MCPServer y se importa desde mcp.server.mcpserver, aunque la API de decoradores se mantiene. Las variables de entorno con prefijo MCP_ ya no configuran el servidor, y desde la versión 2.1 las excepciones inesperadas de una herramienta se registran en el servidor y el cliente solo recibe un mensaje genérico. Para errores que el modelo deba ver, hay que lanzar ToolError de forma explícita.
Servidor MCP de notas con MCPServer
El servidor de notas expone el directorio de apuntes del usuario con protección frente a accesos fuera de la carpeta permitida:
import os
from pathlib import Path
from mcp.server.mcpserver import MCPServer
DIR_NOTAS = Path(os.getenv("NOTES_PATH", "notas")).resolve()
mcp = MCPServer("notas-estudio")
def ruta_segura(nombre: str) -> Path:
ruta = (DIR_NOTAS / nombre).resolve()
if not ruta.is_relative_to(DIR_NOTAS) or ruta.suffix != ".md":
raise ValueError("Acceso no permitido")
return ruta
@mcp.tool()
def listar_notas() -> list[str]:
"""Devuelve los nombres de todos los archivos de notas disponibles."""
return sorted(p.name for p in DIR_NOTAS.glob("*.md"))
@mcp.tool()
def leer_nota(nombre: str) -> str:
"""Devuelve el contenido completo de un archivo de notas por su nombre."""
return ruta_segura(nombre).read_text(encoding="utf-8")
@mcp.tool()
def buscar_en_notas(termino: str) -> list[str]:
"""Busca un término en todas las notas y devuelve hasta 20 líneas coincidentes."""
coincidencias = []
for archivo in DIR_NOTAS.glob("*.md"):
for linea in archivo.read_text(encoding="utf-8").splitlines():
if termino.lower() in linea.lower():
coincidencias.append(f"{archivo.name}: {linea.strip()}")
if len(coincidencias) >= 20:
return coincidencias
return coincidencias
@mcp.resource("notas://indice")
def indice_notas() -> str:
"""Índice de archivos de notas en formato de texto."""
return "\n".join(listar_notas())
if __name__ == "__main__":
mcp.run()
Las anotaciones de tipo definen el esquema de argumentos y los docstrings se convierten en la descripción que lee el modelo para decidir qué herramienta usar, por lo que merecen la misma atención que un prompt. Limitar los resultados de búsqueda evita saturar el contexto del modelo con cientos de líneas.
El servidor de memoria sigue el mismo patrón con un diccionario en memoria, y su interfaz permite sustituirlo más adelante por Redis o PostgreSQL sin tocar a los agentes:
from mcp.server.mcpserver import MCPServer
memoria = MCPServer("memoria-sesion")
_almacen: dict[str, str] = {}
@memoria.tool()
def guardar_dato(clave: str, valor: str) -> str:
"""Guarda un valor de texto asociado a una clave."""
_almacen[clave] = valor
return "ok"
@memoria.tool()
def obtener_dato(clave: str) -> str:
"""Devuelve el valor de una clave o 'null' si no existe."""
return _almacen.get(clave, "null")
Bucle de herramientas en el agente explicador
El explicador recibe las herramientas, decide cuáles invocar y construye la explicación. Para mantener el ejemplo autocontenido, las funciones de los servidores se importan y se envuelven como herramientas de LangChain; en un despliegue distribuido se conectarían a través del transporte de MCP sin cambiar la lógica del agente.
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from mcp_servers import notas, memoria
herramientas = [
tool(notas.listar_notas),
tool(notas.leer_nota),
tool(notas.buscar_en_notas),
tool(memoria.guardar_dato),
tool(memoria.obtener_dato),
]
por_nombre = {h.name: h for h in herramientas}
MAX_ITERACIONES = 8
def nodo_explicador(estado: EstadoAprendizaje) -> dict:
tema = estado["ruta"]["temas"][estado["indice_tema"]]
llm = crear_llm(temperatura=0.3).bind_tools(herramientas)
mensajes = [
SystemMessage(
"Explica el tema usando las notas del usuario. Lista las notas, busca "
"términos clave, lee los archivos relevantes y guarda un resumen en memoria."
),
HumanMessage(f"Tema: {tema['titulo']}. {tema['descripcion']}"),
]
for _ in range(MAX_ITERACIONES):
respuesta = llm.invoke(mensajes)
mensajes.append(respuesta)
if not respuesta.tool_calls:
return {"mensajes": [respuesta]}
for llamada in respuesta.tool_calls:
resultado = por_nombre[llamada["name"]].invoke(llamada["args"])
mensajes.append(ToolMessage(content=str(resultado), tool_call_id=llamada["id"]))
return {"error": "El explicador superó el límite de iteraciones"}
Dos detalles son esenciales. Cada resultado se devuelve como ToolMessage con el mismo tool_call_id de la llamada, para que el modelo asocie respuesta y petición. Y el límite de ocho iteraciones actúa como cortacircuitos: si el modelo entra en un bucle de llamadas, el nodo termina con un error controlado en lugar de consumir recursos indefinidamente.
A2A: agentes que colaboran entre frameworks
MCP conecta agentes con herramientas, pero no resuelve cómo se comunican dos agentes independientes. La comunicación entre agentes A2A cubre ese hueco: cada agente publica una tarjeta que describe quién es, qué habilidades ofrece y en qué dirección escucha, y los demás le envían mensajes y reciben tareas con un formato común.
La ventaja práctica es la interoperabilidad. El generador de cuestionarios construido con LangGraph puede ser consumido por un agente de CrewAI, o por cualquier otro cliente compatible, sin que ninguno conozca los detalles internos del otro. Cada agente se despliega, escala y versiona por separado.
El protocolo alcanzó su versión 1.0 y el SDK de Python siguió el mismo camino. La migración desde la serie 0.3 implica cambios relevantes: la tarjeta del agente declara sus puntos de acceso en supported_interfaces mediante objetos AgentInterface, los ejemplos pasan a definirse en cada habilidad, la capacidad de tarjeta extendida se llama ahora extended_agent_card, y el servidor ya no se construye con clases envoltorio como A2AStarletteApplication, sino con funciones que generan rutas de Starlette.
Publicar un agente A2A con el SDK 1.x
El primer paso es describir el agente con su tarjeta:
from a2a.types import AgentCard, AgentCapabilities, AgentInterface, AgentSkill
habilidad = AgentSkill(
id="generar_cuestionario",
name="Generar cuestionario",
description="Crea preguntas de opción múltiple sobre un tema técnico.",
tags=["educacion", "evaluacion"],
input_modes=["text/plain"],
output_modes=["text/plain"],
examples=["Cuestionario de 3 preguntas sobre Pods de Kubernetes"],
)
tarjeta = AgentCard(
name="Generador de cuestionarios",
description="Agente que genera cuestionarios de repaso a partir de un tema.",
supported_interfaces=[
AgentInterface(protocol_binding="JSONRPC", url="http://localhost:9001/a2a/jsonrpc/"),
],
version="1.0.0",
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
capabilities=AgentCapabilities(streaming=False),
skills=[habilidad],
)
El comportamiento se implementa con un AgentExecutor, que recibe el contexto de la petición y publica eventos en una cola. Para una respuesta simple basta con encolar un mensaje de texto:
from a2a.helpers import new_text_message
from a2a.server.agent_execution import AgentExecutor
from a2a.types import Role
class EjecutorCuestionario(AgentExecutor):
async def execute(self, context, event_queue):
tema = context.get_user_input()
llm = crear_llm(temperatura=0.4)
respuesta = await llm.ainvoke(
f"Genera 3 preguntas de opción múltiple con su respuesta correcta sobre: {tema}"
)
await event_queue.enqueue_event(
new_text_message(respuesta.content, role=Role.ROLE_AGENT)
)
async def cancel(self, context, event_queue):
raise NotImplementedError("Este agente no admite cancelación")
Por último, las rutas se generan con las funciones del SDK y se sirven con Uvicorn:
import uvicorn
from starlette.applications import Starlette
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
manejador = DefaultRequestHandler(
agent_executor=EjecutorCuestionario(),
task_store=InMemoryTaskStore(),
agent_card=tarjeta,
)
rutas = [
*create_agent_card_routes(tarjeta),
*create_jsonrpc_routes(manejador, rpc_url="/a2a/jsonrpc/"),
]
app = Starlette(routes=rutas)
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=9001)
Con el servicio en marcha, la tarjeta pública puede consultarse en la ruta conocida del agente:
curl -s http://localhost:9001/.well-known/agent-card.json
Desde otro agente, el cliente del SDK se crea a partir de la URL base con create_client, que descubre la tarjeta y la interfaz adecuada. Como el SDK 1.x reorganizó la API cliente, conviene revisar su guía de migración antes de adaptar código escrito para la serie 0.3.
Observabilidad y evaluación
Un sistema multiagente que funciona en desarrollo puede degradarse en producción sin lanzar un solo error: el modelo empieza a responder peor, una herramienta devuelve datos incompletos o un agente entra en bucles largos. La observabilidad permite detectar estos problemas.
Herramientas como Langfuse registran cada ejecución como una traza con sus llamadas al modelo, herramientas invocadas, latencias y tokens consumidos. En LangGraph se integran pasando un manejador de callbacks en la configuración de la ejecución, lo que deja rastro de cada nodo sin modificar el código de los agentes.
La evaluación cubre otro ángulo: medir la calidad de las salidas. Los marcos de evaluación con un modelo como juez, como DeepEval, comprueban si una explicación es fiel a las notas de origen o si una pregunta es pertinente. Conviene separar estas pruebas, más lentas y no deterministas, de las pruebas unitarias que simulan el modelo y verifican el contrato del estado:
from unittest.mock import patch
from langchain_core.messages import AIMessage
def test_planificador_devuelve_ruta():
falso = AIMessage(content='{"temas": [{"titulo": "Pods", "descripcion": "Unidad básica"}]}')
with patch("agentes.planificador.crear_llm") as fabrica:
fabrica.return_value.invoke.return_value = falso
cambios = nodo_planificador({"objetivo": "Kubernetes"})
assert cambios["indice_tema"] == 0
assert cambios["ruta"]["temas"][0]["titulo"] == "Pods"
pytest -q -m "not eval"
LangGraph frente a otros enfoques de orquestación
LangGraph no es la única opción para coordinar agentes, y elegir bien depende del problema. Frameworks como CrewAI parten de una metáfora de equipos con roles y tareas, lo que acelera los prototipos cuando el flujo encaja en ese molde, aunque ofrece menos control explícito sobre las transiciones. Otras librerías se centran en conversaciones entre agentes que se pasan mensajes de forma libre, un enfoque flexible pero más difícil de hacer predecible.
La propuesta de LangGraph es más cercana a la ingeniería tradicional: el flujo es un grafo que se puede dibujar, revisar y probar, el estado tiene un esquema definido y la persistencia y las interrupciones forman parte del núcleo. Esa explicitud exige algo más de código inicial, pero compensa en cuanto el sistema necesita reanudarse tras fallos, pasar auditorías o incorporar aprobaciones humanas.
Los protocolos reducen además el coste de estas decisiones. Como MCP y A2A son independientes del framework, un servidor de herramientas escrito hoy sirve para cualquier orquestador futuro, y un agente publicado mediante A2A puede reemplazarse por otro construido con una tecnología distinta sin que sus consumidores lo noten. En la práctica, muchas organizaciones combinan frameworks: LangGraph para los flujos críticos con estado, otro framework para agentes más sencillos, y protocolos estándar como pegamento entre ellos. La regla útil es escoger el orquestador por el flujo que hay que controlar y apostar por los protocolos para todo lo que deba compartirse.
Errores comunes y soluciones
El primer problema habitual es el JSON mal formado en llamadas a herramientas. Suele aparecer con modelos demasiado pequeños o temperaturas altas. La solución combina un modelo de al menos 7B, temperatura baja en los nodos que generan estructuras y un análisis tolerante con errores que devuelva un fallo controlado en lugar de detener el grafo.
El segundo es el error de importación al actualizar el SDK de MCP. Un mensaje que indica que no existe mcp.server.fastmcp, o que redirige a la guía de migración, revela código escrito para la serie 1.x. Hay que importar MCPServer desde mcp.server.mcpserver o fijar temporalmente la versión anterior.
El tercero afecta a A2A: código que usa A2AStarletteApplication o el campo url directamente en la tarjeta corresponde a la serie 0.3. En la serie 1.x las rutas se construyen con create_agent_card_routes y create_jsonrpc_routes, y las direcciones van en supported_interfaces.
El cuarto es la conexión de SQLite cerrada antes de tiempo. Si aparecen errores sobre una base de datos cerrada al reanudar un flujo, probablemente el checkpointer se creó con un gestor de contexto que ya terminó.
El quinto son los estados incompletos tras una interrupción. Si un agente posterior a la aprobación falla por claves ausentes, la solución es que el nodo interrumpido devuelva explícitamente todas las claves que necesitan los siguientes.
El sexto es marcar el tema equivocado como completado. En el entrenador, el estado del tema actual debe actualizarse antes de incrementar el índice; invertir el orden registra el resultado en el tema siguiente.
Buenas prácticas para producción
Fijar versiones con rangos acotados es la primera defensa en un ecosistema que cambia tan rápido. Actualizar debe ser una decisión consciente acompañada de pruebas, no un efecto secundario de reinstalar dependencias.
Los servidores MCP deben validar siempre sus entradas, restringir el acceso a rutas permitidas y devolver errores explícitos con ToolError cuando el modelo deba conocer el motivo. Las herramientas con efectos irreversibles merecen confirmación humana mediante interrupciones del grafo.
Los servicios A2A expuestos fuera de la máquina local necesitan autenticación, cifrado TLS y límites de peticiones. La versión 1.0 del protocolo incorpora mecanismos como la firma de tarjetas para que un cliente verifique la identidad del agente antes de confiar en él.
Para el almacenamiento, SQLite es perfecto en local, pero un despliegue con varias réplicas requiere un checkpointer compartido, como PostgreSQL, y una memoria externa en lugar de diccionarios en proceso. Por último, todo bucle de agente debe tener límites de iteraciones, tiempos de espera y trazas que permitan reconstruir qué ocurrió en cada ejecución.
Conclusiones
Construir sistemas multiagente fiables tiene menos que ver con prompts ingeniosos y más con infraestructura sólida. LangGraph aporta un modelo de grafo con estado explícito, enrutamiento determinista, persistencia mediante checkpoints e interrupciones para la supervisión humana. MCP estandariza el acceso a herramientas y datos, y A2A permite que agentes construidos con tecnologías distintas colaboren mediante un contrato común.
El caso práctico demuestra que todo puede ejecutarse en local con Ollama, sin depender de servicios externos, y que los mismos patrones sirven para dominios muy diferentes. Separar responsabilidades por agente, mantener la lógica de control en código, devolver actualizaciones parciales de estado y limitar los bucles de herramientas son decisiones que hacen el sistema más predecible y fácil de probar.
El ecosistema evoluciona deprisa, como muestran los cambios incompatibles del SDK de MCP 2.0 y del SDK de A2A 1.x. Fijar versiones, cubrir el contrato del estado con pruebas y añadir observabilidad desde el principio permite aprovechar esa evolución sin que cada actualización se convierta en una reescritura.