Patrón Planificar y Ejecutar

Implementa el patrón planificar-y-ejecutar en LangGraph, un nodo de planificación crea un plan paso a paso, un nodo de ejecución ejecuta pasos y un bucle de replanificación se adapta a los resultados.

El patrón plan-and-execute separa la planificación (pensar qué hacer) de la ejecución (hacerlo). El agente crea un plan, ejecuta los pasos, observa los resultados y vuelve a planificar según sea necesario.

Arquitectura

Task Input
    ↓
Planning Node ──→ Creates step-by-step plan
    ↓
Execution Node ──→ Executes next step, observes result
    ↓
Check Node ──→ Is plan complete? ──→ Yes → Return result
    ↓ No                            ↑
Re-Plan ───────── Modify plan ──────┘

Planificar y Ejecutar Básico

python

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict
from typing import List, Annotated
from operator import add

llm = ChatOpenAI(model="gpt-4o")

class PlanState(TypedDict):
    task: str
    plan: List[str]             # Ordered list of steps
    completed_steps: Annotated[List[str], add]
    results: Annotated[List[str], add]
    current_step_index: int
    observations: List[str]
    final_answer: str

def planner(state: PlanState) -> dict:
    prompt = f"""Create a step-by-step plan to accomplish this task:
{state['task']}

Break it down into 3-5 clear, actionable steps.
Return as a numbered list. Each step should be specific and concrete."""

    response = llm.invoke(prompt).content
    steps = [s.strip() for s in response.split("\n")
             if s.strip() and any(s.strip().startswith(str(i)) for i in range(1, 10))]

    print(f"Plan created: {len(steps)} steps")
    return {"plan": steps, "current_step_index": 0, "observations": []}

ℹ️Nota

El planificador descompone la tarea en pasos discretos y accionables. Cada paso debe ser lo suficientemente específico para que el ejecutor sepa exactamente qué hacer.

Nodo Ejecutor

python

def executor(state: PlanState) -> dict:
    step_index = state["current_step_index"]

    if step_index >= len(state["plan"]):
        return {}  # All steps done

    step = state["plan"][step_index]
    context = "\n".join(state.get("results", [])) if state.get("results") else "No prior results."

    prompt = f"""Task: {state['task']}
Current step: {step}
Previous results: {context}

Execute this step. Provide the result."""

    result = llm.invoke(prompt).content

    return {
        "completed_steps": [step],
        "results": [result],
        "current_step_index": step_index + 1,
        "observations": state.get("observations", []) + [f"Step {step_index + 1}: {result[:100]}..."]
    }

Nodo de Verificación y Replanificación

python

def checker(state: PlanState) -> dict:
    """Check if the plan is complete or needs modification."""
    steps_done = len(state.get("completed_steps", []))
    total_steps = len(state.get("plan", []))

    if steps_done >= total_steps:
        return {"observations": state.get("observations", []) + ["All steps completed"]}

    # Check if we need to re-plan based on results
    last_result = state["results"][-1] if state["results"] else ""

    prompt = f"""Task: {state['task']}
Plan: {state['plan']}
Steps completed: {steps_done}/{total_steps}
Last result: {last_result[:200]}

Do we need to re-plan? Options:
1. 'continue' — The plan is working, proceed to next step
2. 'replan' — We need to modify the remaining steps based on new information
3. 'complete' — The task is already done

Respond with one word."""

    decision = llm.invoke(prompt).content.strip().lower()

    return {"observations": state.get("observations", []) + [f"Decision: {decision}"]}

def router(state: PlanState) -> str:
    last_obs = state["observations"][-1] if state["observations"] else ""

    if "complete" in last_obs:
        return "complete"
    elif "replan" in last_obs:
        return "replan"
    else:
        step_index = state["current_step_index"]
        if step_index >= len(state["plan"]):
            return "complete"
        return "continue"

✅Éxito

El verificador evalúa el progreso después de cada paso. Si los resultados indican que existe un enfoque mejor, se dispara la replanificación. En caso contrario, la ejecución continúa.

Nodo de Replanificación

python

def replanner(state: PlanState) -> dict:
    completed = "\n".join(state.get("completed_steps", []))
    results = "\n".join(state.get("results", []))

    prompt = f"""Task: {state['task']}

Original plan: {state['plan']}
Completed steps: {completed}
Results so far: {results[:500]}

Based on the results, create an updated plan for the remaining work.
Return as a numbered list. Focus on what still needs to be done."""

    response = llm.invoke(prompt).content
    new_steps = [s.strip() for s in response.split("\n")
                 if s.strip() and any(s.strip().startswith(str(i)) for i in range(1, 10))]

    return {"plan": new_steps, "current_step_index": 0}

Grafo Completo

python

def finalizer(state: PlanState) -> dict:
    results = "\n".join(state.get("results", []))
    prompt = f"""Task: {state['task']}
Results: {results}

Provide the final answer synthesizing all results."""
    return {"final_answer": llm.invoke(prompt).content}

# Build the graph
builder = StateGraph(PlanState)
builder.add_node("planner", planner)
builder.add_node("executor", executor)
builder.add_node("checker", checker)
builder.add_node("replanner", replanner)
builder.add_node("finalizer", finalizer)

# Flow
builder.add_edge(START, "planner")
builder.add_edge("planner", "executor")
builder.add_edge("executor", "checker")

# Conditional routing based on check
builder.add_conditional_edges("checker", router, {
    "continue": "executor",
    "replan": "replanner",
    "complete": "finalizer"
})
builder.add_edge("replanner", "executor")
builder.add_edge("finalizer", END)

app = builder.compile()

⚠️Advertencia

Sin un número máximo de replanificaciones, el agente podría replanificar indefinidamente. Lleva la cuenta de las replanificaciones y fuerza la finalización tras N replanificaciones.

Seguimiento de Conteo de Replanificaciones

python

class PlanState(TypedDict):
    task: str
    plan: List[str]
    completed_steps: Annotated[List[str], add]
    results: Annotated[List[str], add]
    current_step_index: int
    observations: List[str]
    replan_count: int          # Track re-plans
    max_replans: int           # Maximum allowed re-plans
    final_answer: str

def replanner(state: PlanState) -> dict:
    if state["replan_count"] >= state["max_replans"]:
        return {"observations": ["Max re-plans reached, completing"]}
    # ... rest of replanning logic
    return {"plan": new_steps, "current_step_index": 0, "replan_count": state["replan_count"] + 1}

Ejemplo: Tarea de Investigación con Planificar-y-Ejecutar

python

# Run the agent
result = app.invoke({
    "task": "Research the environmental impact of electric vehicles "
            "compared to gasoline cars. Provide a balanced analysis.",
    "plan": [],
    "completed_steps": [],
    "results": [],
    "current_step_index": 0,
    "observations": [],
    "replan_count": 0,
    "max_replans": 2,
    "final_answer": ""
})

print(result["final_answer"])
# Output: A comprehensive, multi-step research report that was
# planned, executed, checked, and potentially re-planned

Planificar-y-Ejecutar vs ReAct

Aspecto	Plan-and-Execute	ReAct
Planificación	Plan explícito y anticipado	Sin plan explícito
Ejecución	Sigue el plan paso a paso	Reacciona a cada observación
Adaptabilidad	Replanifica cuando es necesario	Naturalmente adaptativo
Ideal para	Tareas complejas de varios pasos	Tareas interactivas con muchas herramientas
Previsibilidad	Alta (el plan es visible)	Baja (comportamiento emergente)
Consumo de tokens	Mayor (plan + ejecución)	Menor