advanced⏱40 minutesLição 7 de 10

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.

Padrão Planejar e Executar

The plan-and-execute pattern separates planning (thinking what to do) from execution (doing it). The agent creates a plan, executes steps, observes results, and re-plans as needed.

Architecture

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] The planner decomposes the task into discrete, actionable steps. Each step should be specific enough that the executor knows exactly what to do.

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] The checker evaluates progress after each step. If results indicate a better approach exists, it triggers re-planning. Otherwise, execution continues.

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] Without a maximum number of re-plans, the agent could re-plan indefinitely. Track re-plan count and force completion after N re-plans.

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
Planning	Explicit upfront plan	No explicit plan
Execution	Follows plan step by step	Reacts to each observation
Adaptability	Re-plans when needed	Naturally adaptive
Best for	Complex multi-step tasks	Interactive, tool-heavy tasks
Predictability	High (plan is visible)	Low (emergent behavior)
Token usage	Higher (plan + execution)	Lower

Perguntas Práticas

Practice Question

What is the first node executed in the plan-and-execute pattern?

Practice Question

What triggers a re-plan in the plan-and-execute pattern?

Practice Question

What risk should you guard against in the re-planning loop?

Practice Question

What does the executor node do?

Practice Question

How does the checker node determine if re-planning is needed?

Practice Question

What is the main advantage of plan-and-execute over ReAct?

Practice Question

How does the planner receive context from previous results?

Practice Question

What happens when all plan steps are executed?

Practice Question

What does the finalizer node do?

Practice Question

In what scenarios is plan-and-execute preferred over simpler patterns?

[!SUCESSO]

Principais Conclusões

Plan-and-execute separates strategic planning from tactical execution
Planner creates an explicit step-by-step plan before execution begins
Executor runs one step at a time and records results
Checker evaluates progress and decides to continue, re-plan, or complete
Re-planner creates updated plans based on intermediate results
Track re-plan count to prevent infinite re-planning loops
The explicit plan provides predictability and visibility
Best for complex multi-step tasks that benefit from upfront strategy

Progresso70%

Padrão de Agente Reflexivo

Execução Paralela

Contéudo do Curso

1. Implantação 2. Observabilidade com LangSmith 3. Agentes Hierárquicos 4. RAG Agentic 5. Agentes SQL 6. Padrão de Agente Reflexivo 7. Padrão Planejar e Executar 8. Execução Paralela 9. Otimização 10. Projeto Final: Agente de Suporte ao Cliente de Produção