Task Decomposition
Complex goals must be broken into executable subtasks with clear dependencies.
Task
Implement TaskDecomposer that:
- Decomposes goals into tasks via LLM.
- Performs topological sort for execution ordering.
- Identifies parallel execution groups (tasks with same dependency level).
- Validates tasks: checks for cycles, missing deps, token overruns.
Constraints
- Detect circular dependencies (raise
CyclicDependencyError). - Tasks with no unresolved deps form one parallel group.
- Validation errors: missing dependency references, token limit exceeded.
Examples
Example 1:
Input:
tasks = [Task('a','desc',100,[]), Task('b','desc',100,['a'])]
decomp = TaskDecomposer()
order = decomp.topological_sort(tasks)
[t.id for t in order]Output:
['a', 'b']Explanation: Task b depends on a, so a must come first.
Starter Code
from typing import List, Dict, Set, Optional
from dataclasses import dataclass, field
@dataclass
class Task:
id: str
description: str
estimated_tokens: int
dependencies: List[str] = field(default_factory=list)
class TaskDecomposer:
def __init__(self, max_tokens_per_task: int = 500):
self.max_tokens_per_task = max_tokens_per_task
def decompose(self, goal: str, llm_fn: callable) -> List[Task]:
# TODO: Use LLM to break goal into tasks
pass
def topological_sort(self, tasks: List[Task]) -> List[Task]:
# TODO: Return tasks in execution order
pass
def identify_parallel_groups(self, tasks: List[Task]) -> List[List[Task]]:
# TODO: Group tasks that can run in parallel
pass
def validate(self, tasks: List[Task]) -> List[str]:
# TODO: Return list of validation errors
pass
Python3
ReadyLines: 1Characters: 0
Ready