Plan & Execute Agent

The Plan-and-Execute pattern (Wang et al., 2023) separates planning from execution: 1. Plan: LLM generates a list of discrete steps. 2. Execute: Each step is run (tool call or LLM call) independently. 3. Synthesise: LLM merges results into a final answer.

Usage

from brain.patterns import PlanAndExecuteAgent

agent = PlanAndExecuteAgent(
    tools={"web_search": lambda q: f"[{q}]"},
    max_plan_steps=5,
)
result = agent.run("Summarise the latest AI safety research")
print(result.answer)
print(result.metadata["plan"])   # list of step descriptions

API Reference

brain.patterns.plan_execute.PlanAndExecuteAgent

PlanAndExecuteAgent(tools: dict[str, Tool] | None = None, provider: LLMProvider | None = None, max_plan_steps: int = 10, replan_on_failure: bool = False)

Bases: BasePattern

Plan-and-Execute agent: plan first, then execute step by step.

Parameters:

Name	Type	Description	Default
tools	dict[str, Tool] | None	mapping of tool name → callable(str) -> str. If provided, the executor first tries to match the step to a tool name; otherwise it calls the LLM to execute each step.	None
provider	LLMProvider | None	LLMProvider instance; defaults to LocalEchoProvider.	None
max_plan_steps	int	cap on number of plan steps (prevents runaway plans).	10
replan_on_failure	bool	if a step fails, ask the LLM to replan from that point.	False

Source code in brain/patterns/plan_execute.py

def __init__(
    self,
    tools: dict[str, Tool] | None = None,
    provider: LLMProvider | None = None,
    max_plan_steps: int = 10,
    replan_on_failure: bool = False,
) -> None:
    self.tools: dict[str, Tool] = tools or {}
    self.max_plan_steps = max_plan_steps
    self.replan_on_failure = replan_on_failure
    self._provider = provider

run

run(task: str, **kwargs: Any) -> PatternResult

Source code in brain/patterns/plan_execute.py

def run(self, task: str, **kwargs: Any) -> PatternResult:
    provider = self._get_provider()

    # Phase 1: Plan
    try:
        plan = self._plan(task, provider)
    except Exception as exc:  # noqa: BLE001  # silent-ok: fail-soft, surface via metric/log elsewhere
        return PatternResult(answer="", ok=False, error=f"Planning failed: {exc}")

    if not plan:
        return PatternResult(answer="", ok=False, error="Empty plan generated")

    # Phase 2: Execute
    steps: list[Step] = []
    context_parts: list[str] = []

    for i, step_desc in enumerate(plan):
        context = "\n".join(context_parts)
        observation = self._execute_step(step_desc, context, provider)
        step = Step(
            index=i,
            action=step_desc,
            observation=observation,
        )
        steps.append(step)
        context_parts.append(f"Step {i + 1}: {step_desc} → {observation}")

    # Phase 3: Synthesize
    answer = self._synthesize(task, steps, provider)

    return PatternResult(
        answer=answer,
        steps=steps,
        iterations=len(steps),
    )

When to Use

Situation	Recommendation
Long-horizon multi-step tasks	Plan & Execute
Steps are independent / parallelisable	Plan & Execute
Short single-turn Q&A	ReAct
Iterative self-improvement	Reflexion