Building an AI agent automatic task decomposition loop requires more than a single prompt—you need a planner that emits structured subtasks and an executor that tracks state. This tutorial walks through a minimal but production-minded implementation using the OpenAI SDK and a hand-rolled task graph. You’ll end with a runnable Python script that breaks a goal into dependent steps and executes them.
Prerequisites
- Python 3.11 or newer
openaiandpython-dotenvinstalled (pip install openai python-dotenv)- An API key from a provider with an OpenAI-compatible chat endpoint. If you want automatic fallback across 240+ models when a provider is degraded, point the client at n4n.ai’s OpenAI-compatible endpoint instead of OpenAI’s default.
- Comfort with synchronous Python, basic JSON, and environment variables.
Project layout
Create a directory with a .env file and a single agent.py. Keep it flat; this is a tutorial, not a framework.
mkdir task-agent && cd task-agent
touch agent.py .envIn .env, set your base URL and key:
# Default OpenAI
OPENAI_API_KEY=sk-...
OPENAI_BASE_URL=https://api.openai.com/v1
# Or use n4n.ai for fallback across providers
# OPENAI_API_KEY=your-n4n-key
# OPENAI_BASE_URL=https://api.n4n.ai/v1Client initialization
Load the env vars and instantiate the client. The OpenAI SDK accepts a base_url, so swapping providers is a one-line change.
import os
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
client = OpenAI(
api_key=os.environ["OPENAI_API_KEY"],
base_url=os.environ["OPENAI_BASE_URL"],
)The decomposition contract
The planner must return a deterministic shape. I use a flat list of tasks with explicit depends_on arrays. This avoids nested trees and makes topological execution trivial. The AI agent automatic task decomposition approach lives or dies on this schema being strict.
{
"tasks": [
{"id": "1", "depends_on": [], "description": "Research current API rate limits"},
{"id": "2", "depends_on": ["1"], "description": "Write a retry wrapper using those limits"},
{"id": "3", "depends_on": ["2"], "description": "Add unit tests for the wrapper"}
]
}A flat list with IDs lets you resolve dependencies with a simple set membership check. If you let the model emit nested children, you inherit a recursion problem for no real gain at this scale.
Writing the planner
The planner is a single chat completion call with response_format={"type": "json_object"}. The system prompt enforces the contract. Keep the model weak and cheap; gpt-4o-mini is sufficient for most goals.
import json
def plan_tasks(goal: str) -> dict:
resp = client.chat.completions.create(
model="gpt-4o-mini",
response_format={"type": "json_object"},
messages=[
{
"role": "system",
"content": (
"You are a planning module for an AI agent automatic task decomposition "
"system. Given a goal, return a JSON object with a 'tasks' array. "
"Each task has 'id' (string), 'depends_on' (array of ids), and "
"'description' (string). Keep tasks concrete and ordered."
),
},
{"role": "user", "content": goal},
],
)
return json.loads(resp.choices[0].message.content)Call it with a sample goal:
plan = plan_tasks("Build a Python script that emails me a daily weather summary")
print(json.dumps(plan, indent=2))Expected output (truncated for brevity):
{
"tasks": [
{"id": "1", "depends_on": [], "description": "Find a free weather API and get an API key"},
{"id": "2", "depends_on": ["1"], "description": "Write a function to fetch forecast for my city"},
{"id": "3", "depends_on": ["2"], "description": "Write a function to send an email via SMTP"},
{"id": "4", "depends_on": ["2", "3"], "description": "Schedule the script with cron"}
]
}The model will occasionally emit extra keys; ignore them. What matters is id, depends_on, and description.
Executing tasks with dependency resolution
The executor loops until all tasks are done. Each iteration picks tasks whose dependencies are satisfied and runs them. For this tutorial the “execution” is a lightweight LLM call that expands the description into a code snippet or action note; in production you’d swap in real tools.
def execute_task(task: dict, completed: dict) -> str:
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": "Execute the given task. Return a concise result string.",
},
{
"role": "user",
"content": f"Task: {task['description']}\nContext from deps: {completed}",
},
],
)
return resp.choices[0].message.content.strip()
def run_agent(goal: str):
plan = plan_tasks(goal)
tasks = {t["id"]: t for t in plan["tasks"]}
completed: dict[str, str] = {}
pending = set(tasks.keys())
while pending:
progress = False
for tid in list(pending):
task = tasks[tid]
if all(d in completed for d in task["depends_on"]):
result = execute_task(task, completed)
completed[tid] = result
pending.remove(tid)
progress = True
print(f"[done {tid}] {task['description']} -> {result[:60]}...")
if not progress:
raise RuntimeError(f"Cycle detected or unsatisfied deps: {pending}")
return completedThe progress flag catches cycles. If no task became executable in a full pass, the remaining graph is unsatisfiable and we bail instead of spinning.
Running the full agent
Add a guard in __main__:
if __name__ == "__main__":
results = run_agent("Build a Python script that emails me a daily weather summary")
print("\nAll tasks completed:")
for k, v in results.items():
print(f" {k}: {v}")Sample console output at a checkpoint (mid-run):
[done 1] Find a free weather API and get an API key -> Used Open-Meteo, no key required.
[done 2] Write a function to fetch forecast for my city -> Defined fetch_forecast(city) using requests.
[done 3] Write a function to send an email via SMTP -> Implemented send_email(host, user, pwd, msg).
[done 4] Schedule the script with cron -> Added '0 8 * * * python weather_email.py' to crontab.
All tasks completed:
1: Used Open-Meteo, no key required.
2: Defined fetch_forecast(city) using requests.
3: Implemented send_email(host, user, pwd, msg).
4: Added '0 8 * * * python weather_email.py' to crontab.Hardening the loop
The naive loop above blocks on LLM calls and has no timeout. For real deployments:
- Wrap
execute_taskin atenacityretry with exponential backoff. Providers throttle; n4n.ai’s automatic fallback hides some of this, but you still need local retries. - Cap total steps (
max_steps=25) to avoid runaway planning. - Persist
completedto disk so a crash resumes instead of replanning. - Validate planner output against a Pydantic model before execution.
from pydantic import BaseModel, Field
class Task(BaseModel):
id: str
depends_on: list[str] = Field(default_factory=list)
class Plan(BaseModel):
tasks: list[Task]Swap json.loads for Plan.model_validate_json to fail fast on malformed plans. The AI agent automatic task decomposition pattern only pays off if bad plans surface before you burn tokens executing them.
Tuning the planner prompt
The default system prompt is deliberately generic. In practice, domain constraints matter. If you’re generating infrastructure code, add: “Tasks must be runnable in order on a fresh Ubuntu 22.04 box.” If the agent calls external APIs, specify authentication boundaries.
A tighter prompt reduces dependency errors. I’ve seen plan quality jump from ~60% satisfiable on first try to >90% just by adding two sentences about the execution environment. The planner is cheap; iterate on it aggressively.
Observing cost and usage
Every completion returns token counts in resp.usage. If you’re on n4n.ai, per-token usage metering also shows up in response headers, so you can attribute planning vs execution cost per run without extra instrumentation. Log prompt_tokens and completion_tokens per phase; when the executor drifts higher than the planner, your task descriptions are too vague.
Why AI agent automatic task decomposition earns its complexity
A single prompt that tries to both plan and act produces entangled, undebuggable output. Splitting the two gives you an audit trail: the plan is inspectable, dependencies are explicit, and a failed step doesn’t trash the whole run. The AI agent automatic task decomposition pattern also lets you parallelize independent tasks later—just replace the sequential for loop with an asyncio.gather over ready tasks.
The same separation makes testing sane. Mock the planner, assert the executor handles dependency order, and you’ve covered the core logic without burning tokens.
Closing notes
You now have a runnable skeleton. Replace execute_task with real tool calls—file writes, shell commands, API requests—and the agent becomes useful. The planner prompt is the lever; tighten it to your domain and the decomposition quality climbs fast.