Padrão Planejar e Executar

Implemente o padrão planejar-e-executar em LangGraph, um nó de planejamento cria um plano passo a passo, um nó de execução executa etapas e um loop de replanejamento se adapta aos resultados.

O padrão plan-and-execute separa o planejamento (pensar o que fazer) da execução (fazer). O agente cria um plano, executa os passos, observa os resultados e replaneja conforme necessário.

Arquitetura

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 ──────┘

Planejar e Executar 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

O planejador decompõe a tarefa em passos discretos e acionáveis. Cada passo deve ser específico o bastante para que o executor saiba exatamente o que fazer.

Nó Executor

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]}..."]
    }

Nó de Verificação e Replanejamento

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"

✅Sucesso

O verificador avalia o progresso após cada passo. Se os resultados indicarem que existe uma abordagem melhor, ele dispara o replanejamento. Caso contrário, a execução continua.

Nó de Replanejamento

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()

⚠️Aviso

Sem um número máximo de replanejamentos, o agente poderia replanejar indefinidamente. Acompanhe a contagem de replanejamentos e force a conclusão após N replanejamentos.

Rastreando Contagem de Replanejamentos

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}

Exemplo: Tarefa de Pesquisa com Planejar-e-Executar

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

Planejar-e-Executar vs ReAct

Aspecto	Plan-and-Execute	ReAct
Planejamento	Plano explícito e antecipado	Sem plano explícito
Execução	Segue o plano passo a passo	Reage a cada observação
Adaptabilidade	Replaneja quando necessário	Naturalmente adaptativo
Ideal para	Tarefas complexas com múltiplos passos	Tarefas interativas com muitas ferramentas
Previsibilidade	Alta (o plano é visível)	Baixa (comportamento emergente)
Consumo de tokens	Maior (plano + execução)	Menor