Construir un agente de planificación en LangGraph: Planificar, ejecutar, replanificar, informar
Un agente de planificación LangGraph de cuatro nodos para la investigación financiera: planes estructurados, un registro de herramientas consciente del espacio de nombres, límites de replaneación y una ejecución de ejemplo con seguimiento.
Fuentes: Your AI Agent Isn’t Smart. Here’s How to Build One That Actually Thinks
La mayoría de los agentes son cadenas de instrucciones disfrazadas
He aquí un hecho incómodo sobre los agentes de IA: muchos son, en realidad, cadenas de instrucciones mejoradas y envueltas en un “abrigo” sofisticado. La fórmula es conocida: una instrucción inicial del sistema, unos pocos herramientas y quizás un bucle ReAct; el resultado se denomina agente. Funciona en las demostraciones y en los escenarios ideales. Pero cuando una herramienta devuelve un error, los datos parecen extraños o la consulta requiere más de tres pasos, todo se desmorona.
La alternativa es un agente de planificación. Este elabora un plan de investigación completo antes de utilizar cualquier herramienta, lo sigue paso a paso y, cuando los resultados intermedios contradicen las expectativas, reescribe lo que queda. La demostración es un sistema de investigación financiera. Si se le hace una solicitud en lenguaje sencillo como “Compare las valoraciones de NVDA y AAPL y dígame cuál parece mejor en este momento”, este crea un plan en varios pasos, ejecuta cada paso con herramientas de datos, se corrige a sí mismo cuando algo falla y redacta un informe al estilo de un analista. Funciona sobre LangGraph, trabaja con modelos en la nube a través de Groq o con modelos locales a través de Ollama, y ocupa aproximadamente cuatro archivos.
Por qué ReAct puro no es suficiente
ReAct (Reason-Act-Observe) es el patrón desde el cual comienzan la mayoría de los agentes. El LLM razona sobre su próximo movimiento, llama a una herramienta, lee lo que recibe, vuelve a razonar y llama a la siguiente herramienta. Es popular, sencillo y adecuado para tareas pequeñas.
La debilidad principal es la codicia, ya que un agente ReAct solo planifica un paso por delante. Si se le pide información sobre cinco acciones según tres métricas además de noticias recientes, simplemente comienza, por ejemplo, con la relación precio/beneficio de NVDA, y el primer resultado desvía su atención. Un prototipo gastó siete llamadas a herramientas en cada métrica de Tesla antes de recordar que la tarea era comparar varias acciones, y su informe final consistió en un resumen de los resultados de Tesla más una sola oración sobre las demás.
La capacidad del modelo no es la causa aquí. La arquitectura sí lo es. Si una sola pasada de inferencia debe decidir qué investigar, cómo investigarlo y qué hacer con los resultados, ninguna de esas decisiones recibe la atención adecuada. Un agente de planificación separa el pensamiento de la acción: primero planifica, luego ejecuta y finalmente verifica si el plan sigue siendo válido. Cada nodo se encarga de una sola tarea, y estas se ejecutan una tras otra.
Cuatro nodos y un bucle
El agente de planificación es un gráfico dirigido con cuatro nodos, similar a un diagrama de flujo donde cada caja se ocupa de una tarea y transmite el estado siguiente.
START --> Planner --> Executor <--> Replanner --> Reporter --> END
- El Planificador toma la consulta del usuario y genera un plan de investigación estructurado, paso a paso. Cada paso indica la herramienta a utilizar, los argumentos y la razón.
- El Ejecutor toma el siguiente paso, llama a la herramienta especificada y registra el resultado en una libreta de apuntes.
- El Replanificador revisa todo lo recopilado hasta el momento después de cada ejecución. Si una herramienta comete un error o los datos muestran algo inesperado, reescribe los pasos restantes. De lo contrario, deja el plan sin cambios.
- El Informante convierte todas las anotaciones de la libreta de apuntes en el informe final del analista, completo con cifras concretas y recomendaciones.
Lo que impulsa el diseño es el bucle Ejecutor-Planificador. Los dos nodos se alternan hasta que se completa el plan, y esa alternancia genera un comportamiento autocorregible, ya que después de cada acción el agente puede reconsiderar su decisión.
Estado: la memoria compartida
En LangGraph, el estado lo constituye todo: la memoria compartida que se transmite entre los nodos. La definición completa es:
import operator
from typing import Annotated, TypedDict
from pydantic import BaseModel, Field
class StrategyState(TypedDict, total=False):
"""Global state that flows through the LangGraph nodes."""
query: str
plan: list[dict]
scratchpad: Annotated[list[dict], operator.add]
current_step: int
final_report: str
replan_count: int
En este estado, query mantiene sin cambios la solicitud original del usuario, plan es la lista ordenada de pasos, scratchpad recopila todos los resultados de las herramientas, current_step hace referencia al plan, y replan_count respalda el límite de seguridad MAX_REPLANS.
La anotación Annotated[list[dict], operator.add] en el scratchpad es importante. Su reductor, operator.add, indica a LangGraph que añada nuevas entradas sin modificar las existentes. Cuando el Ejecutor devuelve {"scratchpad": [new_entry]}, LangGraph lo agrega a la lista actual, de modo que los nodos nunca manejan las listas manualmente y no surgen errores de mutación.
Planes y decisiones estructurados
El Planificador nunca devuelve texto libre. Su salida es un modelo Pydantic, por lo que el LLM debe generar planes en un formato que el código pueda analizar:
AVAILABLE_TOOLS_TEXT = """
- get_metrics(ticker, metric?): Return stock metrics. 'metric' is optional
(P/E, EPS, Revenue, Market Cap, Sector).
- search_news(ticker): Return recent news headlines for a ticker.
- compare_metrics(tickers: list, metric): Compare one metric across
multiple tickers.
"""
class PlanStep(BaseModel):
"""A single executable step inside an analysis plan."""
step_id: int = Field(description="Sequential step number")
tool: str = Field(
description=f"Tool to use. Must be one of:\n{AVAILABLE_TOOLS_TEXT}"
)
args: dict = Field(description="Arguments for the tool call")
purpose: str = Field(description="Why this step is needed")
class Plan(BaseModel):
"""The full plan generated by the planner node."""
goal: str = Field(description="The overall analysis goal")
steps: list[PlanStep] = Field(
description="Ordered list of steps to execute"
)
El campo purpose resulta ser más útil de lo esperado. Indica al Replanner por qué existe un paso y no solo qué hace. Al leer “Comparar relaciones P/E para identificar la valoración relativa” en lugar de simplemente “call compare_metrics,” el Replanner puede decidir con mayor precisión si mantener, modificar o eliminar ese paso.
El Replanner cuenta con su propio esquema:
class ReplanDecision(BaseModel):
"""Output of the replanner node."""
reasoning: str = Field(
description="Analysis of current progress and findings"
)
should_replan: bool = Field(
description="Whether the plan needs modification"
)
updated_steps: list[PlanStep] = Field(
default_factory=list,
description="Remaining steps if replan is needed. Empty if no changes.",
)
Al convertir should_replan en un valor booleano, se obliga al LLM a tomar una decisión clara: ¿el plan va a cambiar o no? Si se deja algo de margen de maniobra al LLM en la salida estructurada, este cambiará de comportamiento cada vez.
El sistema de herramientas
Tres herramientas de investigación financiera funcionan con datos simulados realistas. En entornos de producción se utilizarían llamadas reales a APIs, pero los datos simulados ofrecen una gran ventaja durante la fase de desarrollo: depuración determinística. Cuando el agente se comporta de forma extraña, es inmediatamente evidente si el problema radica en su lógica o en una API.
Un patrón de diseño que merece más atención es el registro de herramientas consciente de los nombres de espacio, que se utiliza en lugar de proporcionarle a cada nodo todas las herramientas disponibles:
from langchain_core.tools import BaseTool
from typing import Iterable, Mapping, Any
class ToolRegistry:
"""Namespace-aware container for LangChain tools."""
def __init__(self) -> None:
self._tools_by_toolset: dict[str, dict[str, BaseTool]] = {}
def add_tools(self, toolset: str, tools: Iterable[BaseTool]) -> None:
bucket = self._tools_by_toolset.setdefault(toolset, {})
bucket.update({t.name: t for t in tools})
def get_tools(self, toolset: str) -> tuple[BaseTool, ...]:
return tuple(self._tools_by_toolset.get(toolset, {}).values())
def invoke(
self, toolset: str, tool_name: str, tool_args: Mapping[str, Any]
) -> Any:
t = self._tools_by_toolset.get(toolset, {}).get(tool_name)
if t is None:
raise ValueError(
f"Unknown tool '{tool_name}' in toolset '{toolset}'"
)
return t.invoke(dict(tool_args))
El registro se realiza en los espacios de nombres "planner", "executor" y "replanner". Todas las herramientas son visibles para el Planner para que pueda planificar con ellas, pero solo el Executor puede llamarlas. Incluso si un error en una solicitud del Replanner emitiera de alguna manera una llamada a una herramienta, el registro la rechazaría porque ese espacio de nombres no cuenta con derechos de ejecución. Eso es el control de acceso basado en roles para agentes, y la separación de responsabilidades funciona además como un mecanismo de seguridad que impide cambios de estado no deseados durante el proceso de planificación.
El nodo Planner
def planner_node(state: StrategyState) -> dict:
"""Create a step-by-step research plan using structured output."""
planner = model.with_structured_output(Plan)
prompt = PLAN_PROMPT.format(
available_tools=AVAILABLE_TOOLS_TEXT,
ticker_choices=ticker_choices_text(),
metric_choices=metric_choices_text(),
query=state["query"],
)
plan: Plan = planner.invoke(prompt)
steps = [s.model_dump() for s in plan.steps]
return {"plan": steps, "current_step": 0}
model.with_structured_output(Plan) hace que el LLM responda con un objeto JSON compatible con Pydantic, y una salida defectuosa simplemente provoca un nuevo intento. Esto supera con creces el análisis de texto sin formato mediante expresiones regulares.
Límites dentro del prompt del plan
PLAN_PROMPT = """\
You are a financial research planner. Given a user's analysis request,
create a step-by-step research plan using the available tools.
Available tools:
{available_tools}
Rules:
- Use only the tools listed above.
- Every plan step must be executable with one of those tools.
- When a tool accepts 'ticker' or 'tickers', use only these exact values:
{ticker_choices}
- When a tool accepts 'metric', use one of these exact values:
{metric_choices}
- There are no other tools available. Final synthesis is handled separately.
Create an efficient plan. Group related lookups. Aim for 4-8 steps.
User request: {query}"""
Hay dos detalles que destacan. La instrucción enumera los valores válidos de los parámetros, por lo que el LLM no necesita adivinar qué valores de acciones o métricas existen. Este podría ser el truco más efectivo contra las alucinaciones para los agentes que utilizan herramientas. Sin él, el modelo planifica con confianza pasos como get_metrics(ticker="BITCOIN", metric="vibes") y luego parece sorprendido cuando fallan.
El otro detalle es la línea “No hay otras herramientas disponibles”. Cuando falta, los LLMs inventan pasos como analyze_sentiment(ticker="NVDA") o generate_report(findings=...), herramientas que en realidad no existen.
La instrucción “apuntar a 4 a 8 pasos” es deliberadamente flexible. Los límites estrictos como “exactamente 5 pasos” tuvieron efectos negativos, ya que obligaban a las consultas simples a incluir pasos redundantes y a las complejas a ajustarse a pasos sobrecargados. Un rango permite al modelo adaptarse, y al contar con pautas en lugar de mandatos, generalmente toma la decisión correcta.
El nodo Ejecutor
El Ejecutor está diseñado intencionalmente como el nodo más sencillo: toma el siguiente paso, ejecuta la herramienta a través del registro y almacena el resultado.
MAX_STEPS = 12 # Safety limit on total steps
def executor_node(state: StrategyState) -> dict:
"""Execute the next pending step from the plan."""
plan = state.get("plan", [])
current_step = state.get("current_step", 0)
if current_step >= len(plan):
return {}
if current_step >= MAX_STEPS:
return {"current_step": len(plan)}
step = plan[current_step]
tool_name = step["tool"]
tool_args = step["args"]
try:
result = str(
TOOL_REGISTRY.invoke(
AgentName.EXECUTOR.value, tool_name, tool_args
)
)
status = "Error" if result.startswith("Error:") else "Success"
except Exception as exc:
result = f"Error: {exc}"
status = "Error"
entry = {
"step": current_step + 1,
"tool": tool_name,
"args": tool_args,
"result": result,
"status": status,
}
return {
"scratchpad": [entry],
"current_step": current_step + 1,
}
MAX_STEPS = 12 funciona como un límite máximo estricto. Si el Replanner sigue añadiendo pasos o se atasca, el agente no puede seguir gastando créditos de API indefinidamente. Una vez alcanzado el límite, el Ejecutor establece current_step en len(plan), y el flujo pasa al Reportador con los datos recopilados hasta ese momento.
Los fallos se reconocen por convención: status = "Error" si result.startswith("Error:") else "Success". Cada una de las tres herramientas añade al principio de sus cadenas de error "Error:", de modo que el Ejecutor puede marcar un fallo sin necesidad de analizar JSON ni manejar excepciones específicas. El Replanner lee este estado al decidir entre intentar de nuevo o ajustar el plan.
El nodo Replanner
Este nodo marca la diferencia entre un agente de planificación y un simple orquestador. Después de ejecutar cada herramienta, evalúa todas las pruebas y se pregunta si el plan restante sigue teniendo sentido:
MAX_REPLANS = 2 # Prevent infinite replanning
def replanner_node(state: StrategyState) -> dict:
"""Review progress and optionally modify the remaining plan."""
plan = state.get("plan", [])
current_step = state.get("current_step", 0)
replan_count = state.get("replan_count", 0)
scratchpad = state.get("scratchpad", [])
remaining = plan[current_step:]
if len(remaining) = MAX_REPLANS:
return {}
scratchpad_text = "\n".join(
f"Step {e['step']}: {format_tool_call(e['tool'], e['args'])} "
f"-> [{e['status']}] {e['result'][:150]}..."
for e in scratchpad
)
remaining_text = "\n".join(
f"Step {s['step_id']}: {format_tool_call(s['tool'], s['args'])} "
f"- {s['purpose']}"
for s in remaining
)
replanner = model.with_structured_output(ReplanDecision)
prompt = REPLAN_PROMPT.format(
goal=state["query"],
scratchpad=scratchpad_text,
remaining_steps=remaining_text,
)
decision: ReplanDecision = replanner.invoke(prompt)
if decision.should_replan and decision.updated_steps:
new_steps = plan[:current_step] + [
s.model_dump() for s in decision.updated_steps
]
return {"plan": new_steps, "replan_count": replan_count + 1}
return {"replan_count": replan_count + 1}
Mira e['result'][:150]. El Replanner muestra cada resultado de la herramienta truncado a 150 caracteres, ya que necesita la esencia y no todo el contenido JSON, mientras que el Reporter recibe posteriormente los datos sin truncar. Los tokens se guardan donde el detalle no es importante y se utilizan donde sí lo es.
Cuando se realiza un replan, plan[:current_step] + [s.model_dump() for s in decision.updated_steps] mantiene los pasos completados y sustituye únicamente los incompletos, dejando la historia intacta.
La solicitud de replan
REPLAN_PROMPT = """\
You are a financial research planner reviewing progress on a research task.
Original goal: {goal}
Completed steps and findings so far:
{scratchpad}
Remaining steps in the plan:
{remaining_steps}
Based on the findings so far, should the remaining plan change?
If an expected tool failed or revealed something unexpected, add a step
to investigate.
If a step is now redundant, remove it.
Use only the available executable tools already shown in the plan.
Do not add recommendation, summary, or report-writing steps."""
La última línea cumple una función crucial. Al eliminar la frase “No se deben añadir pasos de recomendación, resumen o redacción de informes”, Replanner agrega un paso de síntesis a cada plan, aunque esa tarea corresponde al Reporter. Los mensajes de ayuda deben servir para delimitar los límites entre los nodos.
El valor de MAX_REPLANS = 2 resultó ser el óptimo. En una versión inicial se realizaron seis rondas de replaneamiento para una tarea de tres pasos, y el LLM encontró “mejoras” que en realidad no mejoraban nada. Dos rondas son suficientes para una autocorrección real, como manejar una herramienta que falló o eliminar un paso redundante, sin dejar espacio para la parálisis analítica. En algún momento es necesario indicarle al agente que deje de deliberar y termine el trabajo.
El reportero recibe todo el contenido del bloc de notas sin truncar y elabora un informe estructurado en Markdown. Se llama a model.invoke(prompt).content, lo que proporciona texto plano y no un resultado estructurado, lo que permite al LLM presentar el análisis como considere oportuno. Una sola instrucción realiza la mayor parte del trabajo: “Sea específico, cite números de los datos”. Sin ella, los informes resultan vagos, diciendo algo como “NVDA muestra fundamentos sólidos” y omitiendo las cifras reales de P/E e ingresos.
Conectándolo todo en LangGraph
Aquí es donde se encuentran los cuatro nodos:
from langgraph.graph import StateGraph, END
from enum import Enum
class AgentName(Enum):
PLANNER = "planner"
EXECUTOR = "executor"
REPLANNER = "replanner"
REPORT = "report"
def build_graph():
"""Build and compile the LangGraph planning-agent workflow."""
workflow = StateGraph(StrategyState)
workflow.add_node(AgentName.PLANNER.value, planner_node)
workflow.add_node(AgentName.EXECUTOR.value, executor_node)
workflow.add_node(AgentName.REPLANNER.value, replanner_node)
workflow.add_node(AgentName.REPORT.value, report_node)
workflow.set_entry_point(AgentName.PLANNER.value)
workflow.add_edge(AgentName.PLANNER.value, AgentName.EXECUTOR.value)
workflow.add_edge(AgentName.EXECUTOR.value, AgentName.REPLANNER.value)
workflow.add_conditional_edges(
AgentName.REPLANNER.value,
should_continue_execution,
{
AgentName.EXECUTOR.value: AgentName.EXECUTOR.value,
AgentName.REPORT.value: AgentName.REPORT.value,
},
)
workflow.add_edge(AgentName.REPORT.value, END)
return workflow.compile()
def should_continue_execution(state: StrategyState) -> str:
"""Return the next node name after re-planning."""
if state.get("current_step", 0) >= len(state.get("plan", [])):
return AgentName.REPORT.value
return AgentName.EXECUTOR.value
El gráfico se lee casi como en inglés. El Planificador va al Ejecutor, el Ejecutor al Replanificador, y el Replanificador se dirige al Ejecutor si quedan pasos por realizar o al Reportador cuando todo está terminado. El Reportador va a FIN.
La enumeración AgentName parece sencilla pero es útil. Los literales de cadena simples como "planner" dispersos por el código invitan a errores de escritura, y "plannr" crearía silenciosamente un nodo desconectado. Con la enumeración, el error se detecta al momento de la importación.
El enrutamiento verifica una sola condición: ¿están completos todos los pasos? Cuando current_step >= len(plan), el control va al Reportador; de lo contrario, vuelve al Ejecutor. El bucle termina porque current_step aumenta con cada ejecución, y MAX_STEPS = 12 limita todo incluso cuando el Replanificador agrega pasos.
Una consulta, rastreada
La solicitud de ejemplo pide al agente que analice a NVDA y AMD, que presente sus ratios P/E y las noticias recientes uno al lado del otro, y luego elija uno.
Planificación. El planificador genera cinco pasos: métricas completas para NVDA, métricas completas para AMD, una comparación directa de P/E, búsqueda de noticias para NVDA y búsqueda de noticias para AMD. Primero se presentan los perfiles individuales para tener una visión completa, luego la comparación directa, y por último las noticias, ya que constituyen un contexto complementario y no datos esenciales de valoración.
Ejecución. Los pasos 1 y 2 devuelven datos en formato JSON con métricas claras, y cada vez el replanificador confirma que los datos están correctos y que el plan no ha cambiado. El paso 3 devuelve una comparación de los valores de P/E:
{
"metric": "P/E",
"values": {
"NVDA": 58.3,
"AMD": 102.5
}
}
Ambos valores están presentes y son válidos, por lo que no se vuelve a planificar nada. Los pasos 4 y 5 devuelven tres artículos de noticias cada uno. Una vez finalizado el paso 5, current_step es igual a len(plan), y la gestión pasa al Reportero.
Informe. Al recibir cinco entradas completas en el bloc de notas, el Reportero anota números concretos; por ejemplo, que la relación precio/beneficio de NVDA, de 58.3, frente a los 102.5 de AMD, indica que NVDA tiene un mejor valor relativo a pesar de un capitalización de mercado superior de unos 2.8 billones de dólares. Una ejecución completa tarda alrededor de 8 a 10 segundos en Groq y de 30 a 40 segundos en Ollama local con un modelo de 4 mil millones de parámetros.
Mejoras más importantes
La versión actual utiliza datos ficticios, pero la arquitectura está lista para su uso en producción. Tres mejoras son las más relevantes:
- Datos financieros reales. Se sustituyen las búsquedas en el diccionario por llamadas a Yahoo Finance o Alpha Vantage. Las firmas permanecen igual y solo cambian los cuerpos de las funciones, gracias al patrón de decorador
@tool; así, el Planificador y el Ejecutor nunca saben si los números provienen de un archivo JSON de simulación o de una API REST. - Memoria persistente. Actualmente, la pizarra de apuntes solo existe durante una sola ejecución. Con un almacén vectorial como ChromaDB o Pinecone, el agente podría consultar análisis anteriores, respondiendo “¿Cómo se compara NVDA con lo que vimos el mes pasado?”, convirtiéndose así en un analista evolutivo y no en una herramienta sin estado.
Lo que la estructura no puede ofrecer
El ciclo de planificar-ejecutar-replanificar cumple su función. La salida estructurada mantiene al agente en el camino correcto, los registros de herramientas evitan el caos y los límites de seguridad impiden gastos descontrolados. El diseño es sólido y el código permanece ordenado.
Aun así, la inteligencia del sistema en su conjunto depende del LLM que se encuentra bajo su estructura. Un modelo más potente elabora planes mejores, detecta más errores durante la replanificación y genera informes más precisos, mientras que uno más débil sigue la estructura de forma mecánica y ofrece un análisis mediocre. El grafo no puede hacer que el modelo sea más inteligente; su única función es proporcionarle un marco más adecuado para utilizar la inteligencia que ya posee.
La capa de gestión de proyectos está en gran medida resuelta: el estado, la orquestación de herramientas, el flujo de ejecución y la autocorrección siguen patrones claros. El juicio, en cambio, sigue siendo un tema abierto. ¿Cuándo debería el Replanner volver a planificar y cuándo debería permanecer inactivo? ¿Cuándo un número es anómalo y cuándo simplemente sorprende? Actualmente, esas decisiones se incluyen en los prompts, con la esperanza de que el LLM las resuelva, lo cual funciona con más frecuencia de la esperada y con menos frecuencia de la deseada.
Leer más
- DS-STAR explicado: el agente de ciencia de datos de Google y por qué su enfoque funciona — Una explicación detallada del agente de ciencia de datos DS-STAR de Google: sus siete módulos, el ciclo de informes DS-STAR+, las instrucciones publicadas, los resultados de las pruebas de eliminación y las limitaciones actuales.
- Un agente AI para DevOps en Python: LangChain, Claude y herramientas Bash protegidas — Crea un agente que ejecute comandos Bash, lea registros y realice consultas en Kubernetes: primero las reglas de seguridad, luego las herramientas, un ciclo ReAct, memoria, un envoltorio FastAPI y soluciones para fallos comunes.