Planner-Executor Architecture
Separating planning from execution enables complex multi-step task completion.
Task
Implement a Planner-Executor system that:
- Planner decomposes a goal into
SubTaskobjects with dependencies. - Executor runs tasks in dependency order (topological sort).
- Results from completed tasks are passed as context to dependent tasks.
- Failed tasks propagate failure to dependents.
Constraints
- Dependencies must be resolved before execution.
- Detect and raise error on circular dependencies.
- Return a full execution report: tasks, statuses, results, total time.
Examples
Example 1:
Input:
Plan: [Task A (no deps), Task B (depends on A)]
Executor runs A first, passes A's result as context to B.Output:
{'tasks': 2, 'completed': 2, 'results': {'A': ..., 'B': ...}}Explanation: Topological order ensures A completes before B.
Starter Code
from typing import List, Dict, Any
from dataclasses import dataclass, field
@dataclass
class SubTask:
id: str
description: str
dependencies: List[str] = field(default_factory=list)
status: str = 'pending' # pending | running | done | failed
result: Any = None
class PlannerAgent:
def __init__(self, llm_fn: callable):
self.llm_fn = llm_fn
def plan(self, goal: str) -> List[SubTask]:
# TODO: Decompose goal into ordered subtasks
pass
class ExecutorAgent:
def __init__(self, tools: Dict[str, callable]):
self.tools = tools
def execute(self, subtask: SubTask, context: Dict[str, Any]) -> SubTask:
# TODO: Execute subtask using tools, update status/result
pass
class PlannerExecutorSystem:
def __init__(self, planner: PlannerAgent, executor: ExecutorAgent):
self.planner = planner
self.executor = executor
def run(self, goal: str) -> Dict:
# TODO: Plan then execute respecting dependencies
pass
Python3
ReadyLines: 1Characters: 0
Ready