What is this for? To show how task decomposition—breaking a big goal into smaller steps—makes agents reliable and debuggable.
Why does it exist? Agents that skip decomposition try to "do the whole thing" in one blob of reasoning. Then they cannot tell which part failed, retry safely, or show partial progress.
"Create the weekly support summary" is a goal, not a step.
When step 3 fails, you retry step 3—not the entire week. Decomposition is how reliability and traceability enter agent design.
| Plain-English idea | What it means |
|---|---|
| Goal | The deliverable the user wants |
| Subtask / step | One named action with a checkable output |
| Planner | Breaks the goal into an ordered step list |
| Executor | Runs one step and returns the result |
For a payment outage, a planner might output:
1. Check code diff for recent deploys
2. Check auth-service logs (last 1 hour)
3. Check db-primary status
4. Compare findings and synthesize answer
The executor runs one line at a time—e.g., github.get_commit_diff(...), then splunk.query_logs(...).
fetch_tickets → TicketList).Goal: "Create weekly support summary."
Each step can fail differently: CRM auth vs empty clusters vs tone policy on recommendations.
| Type | Plain-English idea | Best for |
|---|---|---|
| Static playbook | Predefined steps; model fills parameters | Production default |
| Dynamic planning | Model invents the step list | Flexible but easier to go off-rails |
Constrain dynamic plans with templates and validators.
A tiny decomposer with per-step validation and retry.
from dataclasses import dataclass, field
@dataclass
class Step:
name: str
run: callable
check: callable
retries: int = 2
@dataclass
class RunState:
artifacts: dict = field(default_factory=dict)
log: list = field(default_factory=list)
def fetch(state: RunState):
state.artifacts["tickets"] = [{"id": 1, "type": "billing"}, {"id": 2, "type": "billing"}]
def cluster(state: RunState):
types = [t["type"] for t in state.artifacts["tickets"]]
state.artifacts["clusters"] = {t: types.count(t) for t in set(types)}
def summarize(state: RunState):
c = state.artifacts["clusters"]
state.artifacts["summary"] = f"Top: {max(c, key=c.get)} ({max(c.values())})"
STEPS = [
Step("fetch", fetch, lambda s: len(s.artifacts.get("tickets", [])) > 0),
Step("cluster", cluster, lambda s: bool(s.artifacts.get("clusters"))),
Step("summarize", summarize, lambda s: "Top:" in s.artifacts.get("summary", "")),
]
def run_plan(steps: list[Step]) -> RunState:
state = RunState()
for step in steps:
for attempt in range(step.retries + 1):
step.run(state)
if step.check(state):
state.log.append(f"{step.name}:ok")
break
else:
state.log.append(f"{step.name}:failed")
break
return state
print(run_plan(STEPS).artifacts["summary"])
Take one messy goal from your backlog and force it into a table with columns: step name, input artifact, output artifact, check, retryable?, HITL?. If you cannot fill a row, the step is still a wish.
Keep the first production version under eight steps. For side-effecting steps, write the idempotency story before the prompt.
When two steps share no artifacts—fetching CRM tickets and fetching status-page incidents—run them concurrently, then join before summarize. Decomposition makes that parallelism obvious.
Use stable step IDs in logs (fetch_tickets, not "Step 1"). Dashboards and alerts should key off those IDs so a spike in cluster_themes failures pages the right owner.
Decompose goals into named, dependency-aware steps with programmatic checks and retries so agents make reliable partial progress instead of one opaque attempt.