Lesson 16.22: Agent Planning & Task Decomposition

01 The Challenge: Executing Multi-Step Complex Goals

In the previous lesson, we learned how an autonomous AI agent uses the ReAct loop (Observe → Decide → Act → Observe Again) to select tools and react dynamically to real-time observations. However, in enterprise environments, users rarely present simple, single-turn prompts. Imagine instructing Byteion's AI Financial Assistant:

"Verify this invoice, reconcile supplier details against our ERP database, check for duplicate historical payments, and generate a comprehensive audit report highlighting discrepancies."

This is not an isolated question answerable by a single tool call. It encompasses multiple interdependent operations across disparate systems:

If an agent is given the vague instruction "verify the invoice" without an explicit execution blueprint, it risks invoking supplier queries before extracting the supplier ID, making redundant queries, or looping indefinitely.

02 Anatomy of a High-Quality Subtask

Proper decomposition requires clarifying the operational scope of a goal. Stating "Verify Invoice INV-1042" actually entails six distinct verification questions: Is the format valid? Is the supplier active? Does the total match the line items?

An enterprise-grade, executable sub-task schema must capture the following metadata attributes:

{
  "task_id": 2,
  "goal": "Lookup verified supplier record",
  "tool": "get_supplier",
  "inputs": {"supplier_id": "SUP-1001"},
  "dependencies": [1],
  "expected_output": "SupplierRecord",
  "failure_policy": "mark_supplier_missing_and_continue"
}

03 Task Dependencies & Directed Acyclic Graphs (DAG)

A dependency exists when Task B cannot proceed until Task A produces its output. In robust software engineering, task relationships are formalized as a Directed Acyclic Graph (DAG):

Notice that Task 3, Task 4, and Task 5 all depend strictly on the validated fields from Task 2. However, they share zero mutual dependencies among themselves. Consequently, all three can be executed simultaneously in parallel!

04 Sequential, Parallel & Conditional Execution

The latency and throughput profile of an agent is determined by how sub-tasks are scheduled across runtime threads:

05 Should LLMs Always Generate Plans?

Relying indiscriminately on an LLM to formulate every execution step introduces unnecessary nondeterminism. Three planning paradigms exist:

• Deterministic Planning (Code-Based): Fixed business logic hardcoded into linear Python functions (e.g., standard regulatory tax audits). Uncompromisingly predictable with zero API cost.
• Model-Generated Planning (Dynamic LLM): Applied when user intent is heterogeneous and the sequence of steps cannot be known in advance. The LLM formulates a custom candidate plan.
• Hybrid Planning (Production Best-Practice): Guardrail policies and core DAG boundaries are enforced by backend code, while the LLM is leveraged for intent decomposition, query reformulation, and semantic classification.

06 Decoupled Architecture: Planner, Executor & Verifier

In reliable agent design, the responsibilities of planning, executing tools, and verifying outcomes are strictly decoupled into three dedicated components:

07 Structured Execution Plans with Pydantic v2

To prevent models from hallucinating invalid tools or circular dependencies, we enforce structured validation contracts with Pydantic v2:

# app/planning_schema.py
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator

class PlanStep(BaseModel):
    model_config = ConfigDict(extra="forbid")
    step_id: int = Field(ge=1, le=10, description="Unique sequential step identifier")
    tool_name: Literal["get_order_status", "search_knowledge_base"]
    purpose: str = Field(min_length=3, max_length=300)
    order_id: str | None = Field(default=None, pattern=r"^ORD-[0-9]{4,}$")
    query: str | None = Field(default=None, min_length=2, max_length=200)
    limit: int = Field(default=3, ge=1, le=5)
    depends_on: list[int] = Field(default_factory=list)

class AgentPlan(BaseModel):
    model_config = ConfigDict(extra="forbid")
    goal: str = Field(min_length=5, max_length=500)
    steps: list[PlanStep] = Field(min_length=1, max_length=5)

    @model_validator(mode="after")
    def validate_plan_dag(self):
        step_ids = [s.step_id for s in self.steps]
        if len(step_ids) != len(set(step_ids)):
            raise ValueError("Step IDs must be strictly unique.")
        
        seen_ids = set()
        for step in self.steps:
            if step.step_id in step.depends_on:
                raise ValueError(f"Step {step.step_id} cannot depend on itself.")
            for dep in step.depends_on:
                if dep not in seen_ids:
                    raise ValueError(f"Step {step.step_id} depends on step {dep} which has not appeared earlier.")
            seen_ids.add(step.step_id)
        return self

08 LLM Plan Generation with Structured Outputs

Using the OpenAI Responses API responses.parse() method, we request structured outputs mapped directly into our validated Pydantic schema:

# app/planner.py
import os
from openai import OpenAI
from app.planning_schema import AgentPlan

client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"), timeout=30.0)
MODEL_NAME = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")

PLANNER_SYSTEM_PROMPT = """
You are an expert deterministic task planner for an enterprise customer support agent.
Decompose the user's objective into a minimal, strictly executable DAG plan.
Allowed Tools:
1. 'get_order_status': Requires order_id (e.g. 'ORD-1001').
2. 'search_knowledge_base': Requires query and limit (1-5).
Rules:
- Never plan destructive write actions (refunds, cancellations, deletions).
- Only declare a dependency when Step B genuinely requires the output from Step A.
- Keep plans within 1 to 5 steps maximum.
"""

def generate_plan(user_goal: str) -> AgentPlan:
    if not user_goal.strip():
        raise ValueError("User goal cannot be empty.")

    response = client.responses.parse(
        model=MODEL_NAME,
        input=[
            {"role": "system", "content": PLANNER_SYSTEM_PROMPT},
            {"role": "user", "content": user_goal}
        ],
        text_format=AgentPlan
    )
    plan = response.output_parsed
    if not plan:
        raise RuntimeError("Planner failed to return a validated AgentPlan.")
    return plan

09 Why Server-Side Plan Validation is Mandatory

Directly dispatching LLM-proposed plans without server-side verification is a critical security risk. Three security invariants must be validated:

# app/plan_validator.py
from app.planning_schema import AgentPlan, PlanStep

def validate_step_arguments(step: PlanStep) -> None:
    if step.tool_name == "get_order_status":
        if not step.order_id:
            raise ValueError(f"Step {step.step_id}: order_id is required for get_order_status.")
    elif step.tool_name == "search_knowledge_base":
        if not step.query or len(step.query.strip()) < 2:
            raise ValueError(f"Step {step.step_id}: valid query required for search_knowledge_base.")

def validate_plan_security(plan: AgentPlan, max_steps: int = 5) -> None:
    if len(plan.steps) > max_steps:
        raise ValueError(f"Plan exceeds safety ceiling of {max_steps} steps.")
    for step in plan.steps:
        validate_step_arguments(step)

10 Engineering the Task Executor

The Executor dispatches steps in topological order and verifies that upstream dependencies have resolved successfully before invoking tools:

# app/executor.py
from app.planning_schema import AgentPlan, PlanStep
from app.plan_validator import validate_plan_security
from app.tools import get_order_status, search_knowledge_base

def execute_single_step(step: PlanStep) -> dict:
    if step.tool_name == "get_order_status":
        return get_order_status(step.order_id)
    elif step.tool_name == "search_knowledge_base":
        return search_knowledge_base(step.query, step.limit)
    raise ValueError(f"Unsupported tool: {step.tool_name}")

def execute_plan(plan: AgentPlan) -> dict:
    validate_plan_security(plan)
    step_results = {}
    completed_log = []

    for step in plan.steps:
        # Dependency Verification
        unmet_deps = [dep for dep in step.depends_on if dep not in step_results]
        if unmet_deps:
            raise RuntimeError(f"Step {step.step_id} cannot execute; missing dependencies: {unmet_deps}")

        result = execute_single_step(step)
        step_results[step.step_id] = result
        completed_log.append({"step_id": step.step_id, "tool": step.tool_name, "status": "completed"})

    return {"goal": plan.goal, "results": step_results, "completed_steps": completed_log}

11 Task Lifecycle & State Tracking

Tracking sub-task lifecycles protects against cascading failures and ensures partial observations are preserved across turns:

The needs_approval state is vital: whenever an action incurs monetary cost or external mutations (e.g. updating bank routing), execution suspends pending human-in-the-loop sign-off.

12 & 13 & 14 Replanning & Verification Architecture

When a task encounters failure, blind repetitive retries waste tokens. We govern dynamic recovery via an Error Classification Matrix:

Task-Level vs Goal-Level Verification:
• Task-Level: Verifies tool output conforms to expected schema (e.g. valid order record retrieved).
• Goal-Level: Evaluates whether cumulative outputs answer the user's overarching objective (e.g. audit report synthesized).

15 & 16 DAG Cycle Detection & Parallel Task Scheduling

If an execution plan contains cyclic dependencies (Step A requires Step B, which requires Step A), the runtime deadlocks. Our Pydantic validator guarantees that dependencies must reference strictly prior steps, mathematically preventing cycles:

# Invalid Cyclic / Forward Dependency Example (Blocked at Schema Level)
bad_plan = {
    "goal": "Verify Order",
    "steps": [
        {"step_id": 1, "tool_name": "get_order_status", "order_id": "ORD-1001", "depends_on": [2]},
        {"step_id": 2, "tool_name": "search_knowledge_base", "query": "policy", "depends_on": []}
    ]
}
# Raises ValueError: Step 1 depends on step 2 which has not appeared earlier.

17 & 18 Key Metrics for Planning Quality

Delivering a correct final answer is insufficient; engineers must evaluate plan efficiency using quantifiable telemetry metrics:

19 Seven Critical Planning Anti-Patterns

20 Custom Orchestrators vs OpenAI Agents SDK & LangGraph

Our custom Pydantic DAG engine provides maximum determinism, auditability, and debuggability. For enterprise fleets with hundreds of collaborating agents, existing frameworks offer higher-level abstractions:

21 Practical Project: AI Invoice Verification Agent

Below is the production FastAPI invoice agent router and its automated Pytest test suite:

# app/invoice_agent_service.py
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel, Field
from app.planner import generate_plan
from app.executor import execute_plan

router = APIRouter(prefix="/v1/invoice-agent", tags=["Invoice Agent"])

class VerificationRequest(BaseModel):
    user_goal: str = Field(min_length=5, max_length=500)

@router.post("/verify")
def process_invoice_verification(request: VerificationRequest):
    try:
        # Step 1: LLM Generates Structured DAG Plan
        plan = generate_plan(request.user_goal)
        # Step 2: Validate & Execute via Hardened Dispatcher
        execution_report = execute_plan(plan)
        return {"status": "success", "plan": plan.model_dump(), "execution": execution_report}
    except ValueError as ve:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Plan validation failed: {str(ve)}")
    except RuntimeError as re:
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Execution error: {str(re)}")

# tests/test_agent_planning.py
import pytest
from pydantic import ValidationError
from app.planning_schema import AgentPlan, PlanStep
from app.executor import execute_plan

def test_prevent_cyclic_and_forward_dependencies():
    # Step 1 depends on future Step 2 -> Must fail validation
    with pytest.raises(ValidationError) as exc:
        AgentPlan(
            goal="Test invalid dependency",
            steps=[
                PlanStep(step_id=1, tool_name="get_order_status", order_id="ORD-1001", purpose="Look up", depends_on=[2]),
                PlanStep(step_id=2, tool_name="search_knowledge_base", query="late delivery", purpose="Search", depends_on=[])
            ]
        )
    assert "depends on step 2 which has not appeared earlier" in str(exc.value)

def test_prevent_self_dependency():
    with pytest.raises(ValidationError) as exc:
        AgentPlan(
            goal="Test self dependency",
            steps=[
                PlanStep(step_id=1, tool_name="get_order_status", order_id="ORD-1001", purpose="Self dep", depends_on=[1])
            ]
        )
    assert "cannot depend on itself" in str(exc.value)

def test_valid_plan_execution():
    plan = AgentPlan(
        goal="Valid two step support lookup",
        steps=[
            PlanStep(step_id=1, tool_name="get_order_status", order_id="ORD-1001", purpose="Check order", depends_on=[]),
            PlanStep(step_id=2, tool_name="search_knowledge_base", query="delay policy", purpose="Check policy", depends_on=[1])
        ]
    )
    res = execute_plan(plan)
    assert len(res["completed_steps"]) == 2
    assert res["results"][1]["found"] is True

22 Knowledge Check: Multiple-Choice Assessment

Test your technical mastery across 8 real-world agent planning scenarios: