Plan-and-Execute Agent Architecture
Source: mortalapps.com- Plan-and-execute agent architecture is an agentic design pattern that decouples task planning from task execution by generating an explicit, structured execution graph before running tools.
- It solves the unpredictability, high token consumption, and infinite-loop risks inherent in reactive, step-by-step agent loops like ReAct.
- By separating these concerns, engineering teams can use highly capable, expensive models for planning, and cheaper, specialized models for execution.
- This separation yields highly auditable, deterministic, and parallelizable workflows suitable for complex enterprise pipelines.
- The primary tradeoff is reduced adaptability to unexpected runtime changes compared to fully dynamic reactive architectures.
Why This Matters
In complex enterprise automation, the plan and execute agent architecture has emerged as a critical blueprint for building predictable and auditable AI systems. Traditional reactive architectures, such as the ReAct pattern, combine reasoning and tool execution into a single, tightly coupled loop. While highly flexible, this coupling introduces significant production risks: agents can easily fall into infinite loops, consume excessive tokens on trivial tasks, and produce non-deterministic execution paths that are nearly impossible to audit or debug.
The plan-and-execute pattern was invented to solve these exact failure modes by separating the cognitive load of high-level strategy from the mechanical execution of individual tasks. A highly capable planner model analyzes the user's objective and generates a structured, dependency-aware execution graph of sub-goals. Specialized, lower-cost executor models then process these sub-goals sequentially or in parallel, utilizing targeted tools without needing to understand the overarching business logic.
If ignored in production, complex multi-step tasks deployed via naive reactive loops will suffer from high P99 latencies, unpredictable API costs, and frequent state corruption. This architecture should be selected over alternatives like ReAct when the target workflow requires strict compliance auditing, parallel execution of independent sub-tasks, predictable cost boundaries, or integration with rigid state machines. Conversely, if the operational environment is highly dynamic and requires constant real-time strategy adjustments based on tool outputs, a hybrid or reactive approach remains preferable.
Core Concepts
To build and operate a plan-and-execute system, engineers must understand several foundational components and their interactions:
- Planner: A high-tier LLM optimized for reasoning, instruction-following, and structured output generation. It parses the user's high-level goal and decomposes it into a structured plan.
- Executor: A lightweight LLM or deterministic code block designed to execute a single, narrow sub-task using designated tools. It does not need global context of the overall goal.
- Sub-goal Dependency Graph: A Directed Acyclic Graph (DAG) generated by the planner where nodes represent discrete sub-tasks and directed edges represent execution dependencies.
- Replanner: An auxiliary LLM loop triggered only when an executor encounters an unrecoverable failure. It modifies the remaining unexecuted portion of the DAG without restarting the entire workflow.
- Execution Context: A shared, append-only state store containing the original goal, the active DAG, and the structured outputs of all completed sub-tasks.
Architectural Comparison
| Architectural Dimension | Plan-and-Execute Pattern | ReAct (Reactive) Pattern |
|---|---|---|
| LLM Coupling | Decoupled (Planner vs. Executor) | Coupled (Single model handles both) |
| Execution Path | Deterministic DAG generated upfront | Dynamic step-by-step discovery |
| Concurrency | High (Independent branches run in parallel) | Low (Strictly sequential execution) |
| Token Efficiency | High (Cheaper models used for execution) | Low (Context window grows with every step) |
| Auditability | High (Plan can be inspected before execution) | Low (Path emerges dynamically at runtime) |
How It Works
The plan-and-execute architecture operates through a highly structured, multi-phase lifecycle designed to maximize predictability and parallel execution while minimizing token costs.
Phase 1: Goal Decomposition and DAG Generation
The system receives the user's high-level objective. The Orchestrator routes this objective to the Planner LLM, accompanied by a strict JSON schema. The Planner analyzes the goal, identifies the necessary steps, and outputs a structured JSON payload representing a Directed Acyclic Graph (DAG). Each node in the DAG contains a unique identifier, a natural language instruction, required input fields, and an array of dependency node IDs that must complete before this node can run.
Phase 2: Dependency Resolution and Scheduling
The Orchestrator parses the generated DAG and initializes their states to PENDING. It evaluates the dependency graph to identify "root nodes" - tasks with zero unresolved dependencies. These tasks are transitioned to the READY state and pushed to the Task Queue for immediate execution.
Phase 3: Sub-task Execution
The Executor Pool pulls tasks from the queue. Each task is assigned to an Executor instance. The Orchestrator injects only the specific task instructions and the output data from its parent dependency nodes into the Executor's context window. The Executor runs its assigned tools (e.g., database queries, API calls, document parsing), processes the results, and returns a structured output to the Orchestrator. The Orchestrator writes this output to the State Store and marks the node as COMPLETED.
Phase 4: State Monitoring and Replanning
Upon completion of any node, the Orchestrator re-evaluates the DAG. Any PENDING nodes whose dependencies are now fully satisfied are transitioned to READY and queued. If an Executor encounters an error, the Orchestrator attempts local, deterministic retries. If these fail, the Orchestrator halts the affected branch and invokes the Replanner LLM. The Replanner receives the current state of the DAG, the failed node's error log, and the completed node outputs. It outputs an updated DAG, pruning dead branches or inserting corrective steps, allowing execution to resume safely.
Phase 5: Synthesis and Final Output
Once all leaf nodes in the DAG reach a terminal COMPLETED state, the Orchestrator gathers the final outputs. It passes these outputs, along with the original user goal, to a final synthesizer model (or uses a deterministic template) to construct the final response delivered to the user.
Architecture
The conceptual architecture of a plan-and-execute system is built around five decoupled components. Execution starts when a user request enters the Orchestrator. The Orchestrator sends the request to the Planner (a high-tier LLM), which generates a structured JSON DAG. The Orchestrator writes this DAG to the State Store, which acts as the single source of truth for task states and step outputs.
The Orchestrator continuously monitors the State Store. It resolves dependencies and pushes executable, independent tasks into the Task Queue. The Executor Pool, consisting of multiple worker threads or serverless functions running cheaper LLMs, pulls tasks from the queue. Each worker reads the required parent outputs from the State Store, executes its assigned tools, and writes the output back to the State Store, updating the task status to Completed.
If an executor reports an unrecoverable failure, the Orchestrator routes the State Store history to the Replanner component, which updates the DAG in the State Store. Once all nodes are completed, the Orchestrator pulls the final node's output from the State Store and returns it to the user, ending the execution.
Decoupling Planner and Executor LLMs
In a production-grade plan-and-execute system, decoupling the planning model from the execution models is the primary driver of cost efficiency and reliability. High-level planning requires advanced reasoning, spatial-temporal understanding, and strict adherence to complex structured outputs (such as JSON schemas representing DAGs). This requires frontier models (e.g., GPT-4o, Claude 3.5 Sonnet).
Conversely, executing a single sub-task (e.g., "extract the total revenue from this PDF" or "query the database for user ID 402") requires far less reasoning capability. These tasks can be executed by smaller, faster, and significantly cheaper models (e.g., GPT-4o-mini, Claude 3.5 Haiku, or fine-tuned local Llama-3 models).
By decoupling these roles, you prevent context window bloat. In a ReAct loop, the LLM must carry the entire history of all previous tool calls, thoughts, and observations in its context window for every single step. In a plan-and-execute architecture, the Executor only receives the specific instructions and direct parent outputs for its assigned sub-task. This keeps the execution context minimal, reducing token consumption and mitigating the risk of model attention drift.
Structured DAG Representation and Parsing
To ensure deterministic execution, the plan must be represented as a formal Directed Acyclic Graph (DAG). The Planner must output this graph using a strict JSON schema. Below is the conceptual schema structure represented in Pydantic:
class TaskNode(BaseModel):
id: str
task_type: str
instruction: str
dependencies: List[str]
class ExecutionPlan(BaseModel):
tasks: List[TaskNode]Upon receiving this plan, the Orchestrator must immediately validate that the graph is acyclic before executing any tasks. This is achieved using Kahn's algorithm or a Depth-First Search (DFS) cycle-detection algorithm. If a cycle is detected, the plan is rejected, and the Planner is prompted to regenerate the plan with an explicit error message detailing the cycle.
Handling Runtime Failures and Replanning Strategies
When executing complex workflows, individual tasks will inevitably fail due to rate limits, network timeouts, or unexpected tool outputs. The plan-and-execute architecture handles these failures through a tiered recovery strategy:
- Local Retries with Exponential Backoff: For transient network errors, the Orchestrator retries the specific Executor task without modifying the plan. This is handled at the worker level and does not involve any LLM calls.
- Dynamic Replanning: If a task fails permanently (e.g., an API returns a 404 or invalid data), the Orchestrator pauses execution of all dependent downstream nodes. It packages the current state of the State Store, the failed task ID, and the error message, and dispatches them to the Replanner LLM. The Replanner can modify the remaining unexecuted nodes of the DAG by inserting alternative retrieval steps, bypassing non-critical tasks, or changing downstream inputs.
- Graceful Degradation and Human-in-the-Loop (HITL): If the Replanner cannot find a viable path to the goal, the Orchestrator marks the DAG as blocked, persists the state, and escalates the issue to a human operator via an interrupt node, preventing runaway automated loops.
Tradeoff Analysis: Plan-and-Execute vs. ReAct
Choosing between Plan-and-Execute and ReAct depends on the operational requirements of the workflow:
- Determinism and Auditing: Plan-and-Execute is highly superior for compliance-heavy environments. The plan can be presented to a user or system for approval before a single tool is executed. ReAct's execution path is emergent and cannot be audited beforehand.
- Parallelism: Plan-and-Execute allows independent tasks to run concurrently, reducing P99 latency. ReAct is strictly sequential.
- Adaptability: ReAct excels in highly volatile environments where the outcome of step 1 completely changes the nature of the goal. Plan-and-Execute requires an expensive replanning step to adapt to unexpected findings, which can introduce latency overhead if the environment changes constantly.
Code Example
import os
import logging
import concurrent.futures
from typing import List, Dict, Any
from pydantic import BaseModel, Field
# Configure structured logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
logger = logging.getLogger("PlanExecuteEngine")
# Ensure API keys or configuration are loaded from environment variables
API_KEY = os.environ.get("AGENT_PLATFORM_API_KEY", "default-sandbox-key")
if API_KEY == "default-sandbox-key":
logger.warning("AGENT_PLATFORM_API_KEY not set. Running in local simulation mode.")
class TaskNode(BaseModel):
id: str = Field(..., description="Unique identifier for the task node")
instruction: str = Field(..., description="The specific instruction for the executor")
dependencies: List[str] = Field(default_factory=list, description="List of task IDs that must complete first")
status: str = Field(default="PENDING", description="Current state: PENDING, RUNNING, COMPLETED, FAILED")
output: Any = Field(default=None, description="The output result of the task execution")
class ExecutionPlan(BaseModel):
tasks: List[TaskNode] = Field(..., description="The Directed Acyclic Graph of tasks")
def execute_sub_task(task: TaskNode, parent_outputs: Dict[str, Any]) -> Dict[str, Any]:
"""
Simulates a specialized Executor LLM or tool execution.
In production, this would call a cheaper LLM (e.g., Claude Haiku) or a specific tool.
"""
logger.info(f"[Executor] Starting task '{task.id}' | Instruction: {task.instruction}")
try:
# Simulate tool execution logic based on instruction content
if "fetch" in task.instruction.lower():
result = {"status": "success", "data": f"Raw data from source (using parents: {list(parent_outputs.keys())})"}
elif "analyze" in task.instruction.lower():
parent_data = [val.get("data", "") for val in parent_outputs.values()]
result = {"status": "success", "analysis": f"Analyzed: {', '.join(parent_data)}"}
elif "format" in task.instruction.lower():
result = {"status": "success", "report": f"Final Report containing: {parent_outputs}"}
else:
result = {"status": "success", "output": f"Processed: {task.instruction}"}
logger.info(f"[Executor] Completed task '{task.id}'")
return result
except Exception as e:
logger.error(f"[Executor] Failed task '{task.id}': {str(e)}")
raise e
class PlanExecuteOrchestrator:
def __init__(self, plan: ExecutionPlan):
self.plan = plan
self.state: Dict[str, TaskNode] = {task.id: task for task in plan.tasks}
self._validate_dag()
def _validate_dag(self):
"""
Validates that the execution plan is a valid Directed Acyclic Graph (DAG)
using Kahn's algorithm for topological sorting.
"""
in_degree = {task.id: 0 for task in self.plan.tasks}
adj_list = {task.id: [] for task in self.plan.tasks}
for task in self.plan.tasks:
for dep in task.dependencies:
if dep not in adj_list:
raise ValueError(f"Task '{task.id}' depends on non-existent task '{dep}'")
adj_list[dep].append(task.id)
in_degree[task.id] += 1
queue = [node for node, deg in in_degree.items() if deg == 0]
visited_count = 0
while queue:
curr = queue.pop(0)
visited_count += 1
for neighbor in adj_list[curr]:
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
queue.append(neighbor)
if visited_count != len(self.plan.tasks):
raise ValueError("Invalid execution plan: Circular dependency detected in DAG.")
logger.info("Execution plan DAG validated successfully. No cycles detected.")
def _get_runnable_tasks(self) -> List[TaskNode]:
runnable = []
for task in self.state.values():
if task.status != "PENDING":
continue
# Check if all dependencies are completed
deps_satisfied = True
for dep_id in task.dependencies:
dep_task = self.state.get(dep_id)
if not dep_task or dep_task.status != "COMPLETED":
deps_satisfied = False
break
if deps_satisfied:
runnable.append(task)
return runnable
def execute(self) -> Dict[str, Any]:
logger.info("Starting execution of the plan...")
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
active_futures = {}
while True:
# 1. Identify and submit runnable tasks
runnable_tasks = self._get_runnable_tasks()
for task in runnable_tasks:
task.status = "RUNNING"
# Gather outputs from immediate parent dependencies
parent_outputs = {dep: self.state[dep].output for dep in task.dependencies}
future = executor.submit(execute_sub_task, task, parent_outputs)
active_futures[future] = task
if not active_futures:
# Check if we are finished or blocked
all_done = all(t.status == "COMPLETED" for t in self.state.values())
any_failed = any(t.status == "FAILED" for t in self.state.values())
if all_done:
logger.info("All tasks in the DAG completed successfully.")
break
elif any_failed:
logger.error("Execution halted due to unrecoverable task failures.")
break
else:
logger.error("Execution deadlocked. Check dependency configurations.")
break
# 2. Wait for at least one running task to complete
done, _ = concurrent.futures.wait(
active_futures.keys(),
return_when=concurrent.futures.FIRST_COMPLETED
)
for future in done:
task = active_futures.pop(future)
try:
result = future.result()
task.output = result
task.status = "COMPLETED"
except Exception:
task.status = "FAILED"
# In production, trigger Replanner here if appropriate
return {task_id: task.output for task_id, task in self.state.items()}
# Example usage
if __name__ == "__main__":
# Define a sample plan: Task C depends on A and B; Task D depends on C.
sample_plan = ExecutionPlan(tasks=[
TaskNode(id="fetch_competitor_data", instruction="Fetch competitor pricing from API", dependencies=[]),
TaskNode(id="fetch_internal_costs", instruction="Fetch internal manufacturing costs from DB", dependencies=[]),
TaskNode(id="analyze_margins", instruction="Analyze profit margins comparing internal vs competitor", dependencies=["fetch_competitor_data", "fetch_internal_costs"]),
TaskNode(id="format_report", instruction="Format final executive margin report", dependencies=["analyze_margins"])
])
orchestrator = PlanExecuteOrchestrator(sample_plan)
final_results = orchestrator.execute()
print("
Execution Results:")
for k, v in final_results.items():
print(f"{k}: {v}")2026-03-30 12:00:00,000 [INFO] PlanExecuteEngine: Execution plan DAG validated successfully. No cycles detected.
2026-03-30 12:00:00,001 [INFO] PlanExecuteEngine: Starting execution of the plan...
2026-03-30 12:00:00,002 [INFO] PlanExecuteEngine: [Executor] Starting task 'fetch_competitor_data' | Instruction: Fetch competitor pricing from API
2026-03-30 12:00:00,003 [INFO] PlanExecuteEngine: [Executor] Starting task 'fetch_internal_costs' | Instruction: Fetch internal manufacturing costs from DB
2026-03-30 12:00:00,104 [INFO] PlanExecuteEngine: [Executor] Completed task 'fetch_competitor_data'
2026-03-30 12:00:00,105 [INFO] PlanExecuteEngine: [Executor] Completed task 'fetch_internal_costs'
2026-03-30 12:00:00,106 [INFO] PlanExecuteEngine: [Executor] Starting task 'analyze_margins' | Instruction: Analyze profit margins comparing internal vs competitor
2026-03-30 12:00:00,207 [INFO] PlanExecuteEngine: [Executor] Completed task 'analyze_margins'
2026-03-30 12:00:00,208 [INFO] PlanExecuteEngine: [Executor] Starting task 'format_report' | Instruction: Format final executive margin report
2026-03-30 12:00:00,309 [INFO] PlanExecuteEngine: [Executor] Completed task 'format_report'
2026-03-30 12:00:00,310 [INFO] PlanExecuteEngine: All tasks in the DAG completed successfully.
Execution Results:
fetch_competitor_data: {'status': 'success', 'data': 'Raw data from source (using parents: [])'}
fetch_internal_costs: {'status': 'success', 'data': 'Raw data from source (using parents: [])'}
analyze_margins: {'status': 'success', 'analysis': 'Analyzed: Raw data from source (using parents: []), Raw data from source (using parents: [])'}
format_report: {'status': 'success', 'report': "Final Report containing: {'analyze_margins': {'status': 'success', 'analysis': 'Analyzed: Raw data from source (using parents: []), Raw data from source (using parents: [])'}}"}