4. Task Decomposition

Overview

In the previous capsule you implemented plan-and-execute: the agent generates a plan (a list of steps) and executes them one by one. But there's a question we didn't answer: how does it generate those steps? If you tell the LLM "decompose this task", you'll get something — but that something might be 3 vague steps or 50 useless micro-steps. Task decomposition is the discipline of turning a complex task into manageable sub-tasks, with structure, dependencies, and clear limits. It isn't just "making a list of steps" — it's understanding a problem's topology.

The difference between an agent that decomposes well and one that decomposes badly is the difference between a project with clear milestones and one with an infinite backlog where nobody knows where to start. When the decomposition is good, every sub-task has a clear objective, a success criterion, and explicit dependencies. When it's bad, you get sub-tasks that overlap, depend on each other circularly, or are so granular that the overhead of coordinating them exceeds the benefit.

This capsule covers the how of planning: decomposition strategies (top-down, bottom-up, analogical), modeling dependencies between sub-tasks with graphs, parallel vs sequential execution decisions, and — crucially — guardrails to stop the decomposition from exploding in complexity. By the end you'll implement a decomposition node in LangGraph that produces structured sub-tasks with execution metadata.

The Art of Decomposing Tasks

Why decomposition matters

Imagine you ask your Research Agent: "Analyze the impact of AI regulation on European startups in 2025, compare it with regulation in the United States, and recommend a strategy for a startup that wants to operate in both markets."

Without decomposition, the agent tries to solve everything at once: it searches for something, synthesizes with the first thing it finds, and delivers a superficial result. With decomposition, the agent identifies that there are three distinct problems:

Original task:
"Analyze AI regulation in the EU vs the US and recommend a strategy for a startup"

Identified sub-tasks:
1. Research AI regulation in Europe (EU AI Act, current state)
2. Research AI regulation in the United States (federal level, state by state)
3. Compare the two regulatory frameworks (key differences)
4. Research precedents of startups operating in both markets
5. Synthesize a strategic recommendation based on the findings

Each sub-task is independently researchable (the first four), and the synthesis depends on all of them. That gives you an executable structure.

The granularity problem

Perfect decomposition doesn't exist. There's always a trade-off between granularity and overhead:

Too coarse (3 steps):                 Too fine (20 steps):
┌─────────────────────┐               ┌──────────────────────────────┐
│ 1. Research the EU  │               │ 1. Search "EU AI Act"        │
│ 2. Research the US  │               │ 2. Filter by 2025            │
│ 3. Compare and      │               │ 3. Extract key articles      │
│    recommend        │               │ 4. Search enforcement cases  │
└─────────────────────┘               │ 5. Search legal opinions     │
                                      │ 6. Categorize by sector      │
Problem: each step                    │ 7. Search US federal rules   │
is too big                            │ 8. Search US state rules     │
to execute well.                      │ 9. Compare enforcement...    │
                                      │ ...                          │
                                      │ 20. Write the recommendation │
                                      └──────────────────────────────┘
                                      Problem: coordination overhead
                                      exceeds the benefit.

The sweet spot depends on the context, but here's a rule of thumb: each sub-task should require 1-3 tool calls. If it needs 0, it's too fine. If it needs 10+, it's too coarse.

Decomposition as structured reasoning

Task decomposition isn't just partitioning — it's a form of reasoning. When you decompose "analyze X", you're implicitly answering:

• What information do I need? (research sub-tasks)
• What processing does it require? (analysis sub-tasks)
• In what order? (dependencies)
• What can I do in parallel? (independencies)
• How do I know when I'm done? (success criteria)

An agent that decomposes well is an agent that reasons well about the structure of problems.

Decomposition Strategies

Top-Down: Divide and Conquer

The most intuitive strategy. You start from the complete task and divide it recursively:

Level 0: "Analyze AI regulation in the EU vs the US"
         │
         ├── Level 1: "Research EU regulation"
         │       ├── Level 2: "Find the EU AI Act text"
         │       └── Level 2: "Find EU enforcement cases"
         │
         ├── Level 1: "Research US regulation"
         │       ├── Level 2: "Find US federal regulation"
         │       └── Level 2: "Find US state regulation"
         │
         └── Level 1: "Compare and recommend"
                 ├── Level 2: "Comparison table"
                 └── Level 2: "Strategic recommendation"

When to use it: When you understand the problem's general structure. It works well for research tasks, comparative analysis, and structured reports.

Implementation with an LLM:

from pydantic import BaseModel, Field

class SubTask(BaseModel):
    id: str = Field(description="Unique identifier (e.g., 'task_1')")
    description: str = Field(description="What this sub-task must do")
    depends_on: list[str] = Field(default_factory=list, description="IDs of prerequisite sub-tasks")
    estimated_calls: int = Field(description="Estimated tool calls (1-3)")

class DecompositionResult(BaseModel):
    subtasks: list[SubTask] = Field(description="Ordered list of sub-tasks")
    reasoning: str = Field(description="Why this decomposition")

TOP_DOWN_PROMPT = """Decompose this task into executable sub-tasks.

Rules:
- Each sub-task must be solvable with 1-3 searches or tool calls
- Maximum 8 sub-tasks
- State explicit dependencies (which sub-task must complete first)
- The last sub-task must be the final synthesis

Task: {task}"""

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")
decomposer = model.with_structured_output(DecompositionResult)

result = decomposer.invoke(
    TOP_DOWN_PROMPT.format(task="Analyze the impact of AI on the tech job market in 2025")
)

for st in result.subtasks:
    deps = f" (depends on: {', '.join(st.depends_on)})" if st.depends_on else ""
    print(f"  {st.id}: {st.description}{deps}")
print(f"\nReasoning: {result.reasoning}")

Bottom-Up: Find Components

Instead of starting from the top, you identify the "atoms" of information you need and group them:

Identified atoms:
• AI adoption statistics in companies
• Papers on labor displacement
• Reports on new roles created by AI
• Tech sector salary data 2024-2025
• Opinions from industry leaders
• Labor regulations related to AI

Grouping:
Group A (Quantitative data): statistics + salary data
Group B (Qualitative analysis): papers + reports + opinions
Group C (Legal framework): regulations
Synthesis: Group A + B + C → final report

When to use it: When you don't know the general structure but you do know which pieces of information you need. It works well for exploratory tasks where the result has no predefined shape.

Implementation with an LLM:

BOTTOM_UP_PROMPT = """Identify the pieces of information needed to solve this task.

Step 1: List every atomic piece of data/information you need (maximum 10)
Step 2: Group the information into logical clusters (maximum 4 groups)
Step 3: Define one sub-task per group + one synthesis sub-task
Step 4: Establish dependencies (the synthesis depends on every group)

Task: {task}"""

Analogical: Similar Past Tasks

If the agent has solved similar tasks before, it can reuse the decomposition structure:

ANALOGICAL_PROMPT = """I have to solve this new task.

New task: {new_task}

Here's an example of how I decomposed a similar task:
{example_decomposition}

Use the same decomposition structure, adapted to the new task.
Keep the same granularity and dependency style."""

When to use it: When you have a catalog of successful decompositions. In production, this connects with memory (M6): the agent remembers how it decomposed past tasks and reuses the patterns.

Quick comparison

In practice, the most robust strategy is top-down with constraints: you tell the LLM the general structure you want and ask it to fill in the details.

Dependency Graphs

Sub-tasks as nodes, dependencies as edges

When you decompose a task, the sub-tasks aren't a flat list — they're a directed acyclic graph (DAG). Each sub-task is a node, and each dependency is an edge:

    task_1: Research the EU AI Act
    task_2: Research US regulation
    task_3: Research startup precedents
    task_4: Compare the frameworks (depends on task_1, task_2)
    task_5: Synthesize a recommendation (depends on task_3, task_4)

    Graph:
    task_1 ──┐
             ├──→ task_4 ──┐
    task_2 ──┘              ├──→ task_5
    task_3 ─────────────────┘

task_1 and task_2 can run in parallel (they don't depend on each other). task_3 is also independent. task_4 needs task_1 and task_2 to finish. task_5 needs task_3 and task_4 to finish.

Modeling the graph in Python

You don't need a complex graph library. An adjacency-list dictionary is enough:

from typing import TypedDict

class TaskNode(TypedDict):
    id: str
    description: str
    depends_on: list[str]
    status: str  # "pending", "running", "completed", "failed"
    result: str | None

def build_dependency_graph(subtasks: list[SubTask]) -> dict[str, TaskNode]:
    """Turn a list of sub-tasks into a dependency graph."""
    graph = {}
    for st in subtasks:
        graph[st.id] = TaskNode(
            id=st.id,
            description=st.description,
            depends_on=st.depends_on,
            status="pending",
            result=None,
        )
    return graph

def get_ready_tasks(graph: dict[str, TaskNode]) -> list[str]:
    """Return the IDs of tasks whose dependencies are completed."""
    ready = []
    for task_id, task in graph.items():
        if task["status"] != "pending":
            continue
        deps_met = all(
            graph[dep]["status"] == "completed"
            for dep in task["depends_on"]
        )
        if deps_met:
            ready.append(task_id)
    return ready

get_ready_tasks is the key function: on each iteration, it asks "what can I run right now?". If multiple tasks are ready, you can run them in parallel. If there's one, sequential. If there are none and there are still pending tasks, you have a dependency problem.

Validating the graph

Before executing, validate that the graph is a valid DAG — no cycles, no broken dependencies:

def validate_graph(graph: dict[str, TaskNode]) -> list[str]:
    """Validate the dependency graph. Returns a list of errors."""
    errors = []
    all_ids = set(graph.keys())
    
    for task_id, task in graph.items():
        for dep in task["depends_on"]:
            if dep not in all_ids:
                errors.append(f"{task_id} depends on {dep}, which doesn't exist")
            if dep == task_id:
                errors.append(f"{task_id} depends on itself")
    
    if has_cycle(graph):
        errors.append("The graph has cycles — impossible to execute")
    
    return errors

def has_cycle(graph: dict[str, TaskNode]) -> bool:
    """Detect cycles with DFS."""
    visited = set()
    in_stack = set()
    
    def dfs(node_id: str) -> bool:
        visited.add(node_id)
        in_stack.add(node_id)
        for dep in graph[node_id]["depends_on"]:
            if dep in in_stack:
                return True
            if dep not in visited and dfs(dep):
                return True
        in_stack.discard(node_id)
        return False
    
    return any(dfs(nid) for nid in graph if nid not in visited)

Topological sort for the execution order

If you need a linear order that respects every dependency (purely sequential execution), a topological sort gives you that:

def topological_sort(graph: dict[str, TaskNode]) -> list[str]:
    """Return an execution order that respects dependencies."""
    in_degree = {tid: len(t["depends_on"]) for tid, t in graph.items()}
    queue = [tid for tid, deg in in_degree.items() if deg == 0]
    order = []
    
    while queue:
        current = queue.pop(0)
        order.append(current)
        for tid, task in graph.items():
            if current in task["depends_on"]:
                in_degree[tid] -= 1
                if in_degree[tid] == 0:
                    queue.append(tid)
    
    if len(order) != len(graph):
        raise ValueError("Cycle detected — can't be ordered")
    
    return order

Parallel vs Sequential Execution

When to use each strategy

The parallel vs sequential decision isn't binary — it's per pair of sub-tasks:

Rule: If task_A isn't in task_B's depends_on
      and task_B isn't in task_A's depends_on
      → they can run in parallel.

In practice, there are three patterns:

1. Fully Sequential: Each task depends on the previous one.

task_1 → task_2 → task_3 → task_4
Example: an ETL pipeline where each step transforms the previous one's output

2. Fully Parallel (fan-out/fan-in): Independent tasks followed by a synthesis.

task_1 ──┐
task_2 ──┼──→ task_synthesis
task_3 ──┘
Example: researching 3 independent sources and synthesizing

3. Mixed (DAG): Some parallel, some sequential.

task_1 ──┐
         ├──→ task_3 ──┐
task_2 ──┘              ├──→ task_5
task_4 ─────────────────┘
Example: task_3 needs results from task_1 and task_2,
         task_5 needs task_3 and task_4

Parallel execution with asyncio

When get_ready_tasks returns multiple tasks, you run them in parallel:

import asyncio
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage

model = init_chat_model("openai:gpt-4.1-mini")

async def execute_subtask(task: TaskNode, context: str) -> str:
    """Execute a single sub-task."""
    prompt = f"""Previous context:
{context}

Sub-task: {task['description']}

Execute this sub-task and return the findings concisely."""
    
    response = await model.ainvoke([HumanMessage(content=prompt)])
    return response.content

async def execute_parallel(
    graph: dict[str, TaskNode],
    context: str = ""
) -> dict[str, str]:
    """Execute the graph respecting dependencies, parallelizing where possible."""
    results = {}
    
    while True:
        ready = get_ready_tasks(graph)
        if not ready:
            break
        
        tasks_to_run = [
            execute_subtask(graph[tid], context)
            for tid in ready
        ]
        
        task_results = await asyncio.gather(*tasks_to_run)
        
        for tid, result in zip(ready, task_results):
            graph[tid]["status"] = "completed"
            graph[tid]["result"] = result
            results[tid] = result
            context += f"\n[{tid}]: {result[:200]}"
    
    return results

asyncio.gather runs every ready sub-task at the same time. When they finish, it marks them completed, updates the context, and looks for the next ready tasks. The loop ends when there are no more pending tasks with met dependencies.

Trade-offs: parallel vs sequential

Decision rule: Use parallel when you have 3+ independent sub-tasks and latency matters. Use sequential when the sub-tasks build on each other or when the API's rate limits are restrictive.

Rate limiting in parallel execution

If you fire off 8 parallel sub-tasks, you can blow past the API's rate limits. A semaphore controls the concurrency:

MAX_CONCURRENT = 3

async def execute_parallel_with_limit(
    graph: dict[str, TaskNode],
    context: str = ""
) -> dict[str, str]:
    """Parallel execution with a concurrency limit."""
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)
    results = {}
    
    async def run_with_limit(tid: str) -> tuple[str, str]:
        async with semaphore:
            result = await execute_subtask(graph[tid], context)
            return tid, result
    
    while True:
        ready = get_ready_tasks(graph)
        if not ready:
            break
        
        completed = await asyncio.gather(
            *[run_with_limit(tid) for tid in ready]
        )
        
        for tid, result in completed:
            graph[tid]["status"] = "completed"
            graph[tid]["result"] = result
            results[tid] = result
            context += f"\n[{tid}]: {result[:200]}"
    
    return results

MAX_CONCURRENT = 3 means that even if 8 tasks are ready, only 3 run at the same time. The others wait their turn.

Guardrails for Decomposition

The sub-task explosion problem

Without limits, the LLM can generate absurd decompositions. A prompt like "analyze the state of AI in 2025" can produce 50 sub-tasks if you don't contain it. Every extra sub-task adds latency, cost, and coordination complexity.

Guardrail 1: Sub-task limit

MAX_SUBTASKS = 8
MIN_SUBTASKS = 2

def validate_decomposition(result: DecompositionResult) -> DecompositionResult:
    """Validate and adjust the decomposition."""
    if len(result.subtasks) > MAX_SUBTASKS:
        result.subtasks = merge_subtasks(result.subtasks, MAX_SUBTASKS)
    
    if len(result.subtasks) < MIN_SUBTASKS:
        raise ValueError(
            f"Decomposition too coarse: {len(result.subtasks)} sub-tasks. "
            f"Minimum: {MIN_SUBTASKS}"
        )
    
    return result

def merge_subtasks(subtasks: list[SubTask], target: int) -> list[SubTask]:
    """Merge similar sub-tasks until you reach the target."""
    while len(subtasks) > target:
        merged = subtasks[-2]
        to_remove = subtasks[-1]
        merged.description += f" + {to_remove.description}"
        for st in subtasks:
            st.depends_on = [
                merged.id if dep == to_remove.id else dep
                for dep in st.depends_on
            ]
        subtasks.remove(to_remove)
    return subtasks

Why 8? It's a pragmatic number. With 5-8 sub-tasks, the agent can solve complex problems without excessive overhead. Research in planning (HuggingGPT, Chameleon) uses similar limits.

Guardrail 2: Maximum depth

If you use recursive decomposition (sub-tasks that decompose into sub-sub-tasks), limit the depth:

MAX_DEPTH = 2

def decompose_recursive(
    task: str,
    depth: int = 0,
    max_depth: int = MAX_DEPTH
) -> list[SubTask]:
    """Recursive decomposition with a depth limit."""
    if depth >= max_depth:
        return [SubTask(
            id=f"leaf_{depth}",
            description=task,
            depends_on=[],
            estimated_calls=2,
        )]
    
    result = decomposer.invoke(
        TOP_DOWN_PROMPT.format(task=task)
    )
    
    all_subtasks = []
    for st in result.subtasks:
        if st.estimated_calls > 3:
            children = decompose_recursive(
                st.description, depth + 1, max_depth
            )
            all_subtasks.extend(children)
        else:
            all_subtasks.append(st)
    
    return all_subtasks

MAX_DEPTH = 2 means: the original task gets decomposed (level 1), and if any sub-task is too big, it gets decomposed once more (level 2). After that, it runs as is.

Guardrail 3: Complexity estimation

Before executing, estimate the decomposition's total cost:

from pydantic import BaseModel, Field

class ComplexityEstimate(BaseModel):
    total_tool_calls: int = Field(description="Total estimated tool calls")
    estimated_tokens: int = Field(description="Estimated tokens")
    estimated_latency_seconds: int = Field(description="Estimated latency in seconds")
    parallel_depth: int = Field(description="DAG depth (parallel layers)")

def estimate_complexity(graph: dict[str, TaskNode]) -> ComplexityEstimate:
    """Estimate the total complexity of executing the graph."""
    total_calls = sum(
        t.get("estimated_calls", 2) for t in graph.values()
    )
    
    depth = calculate_dag_depth(graph)
    
    return ComplexityEstimate(
        total_tool_calls=total_calls,
        estimated_tokens=total_calls * 1500,
        estimated_latency_seconds=depth * 5,
        parallel_depth=depth,
    )

def calculate_dag_depth(graph: dict[str, TaskNode]) -> int:
    """Compute the DAG's depth (the longest path)."""
    memo = {}
    
    def depth_of(task_id: str) -> int:
        if task_id in memo:
            return memo[task_id]
        deps = graph[task_id]["depends_on"]
        if not deps:
            memo[task_id] = 1
        else:
            memo[task_id] = 1 + max(depth_of(d) for d in deps)
        return memo[task_id]
    
    return max(depth_of(tid) for tid in graph)

Guardrail 4: Budget gate

If the estimate exceeds a budget, simplify the decomposition or reject it:

MAX_TOOL_CALLS = 20
MAX_LATENCY_SECONDS = 60

def budget_check(estimate: ComplexityEstimate) -> tuple[bool, str]:
    """Check that the decomposition is within budget."""
    if estimate.total_tool_calls > MAX_TOOL_CALLS:
        return False, (
            f"Too many estimated tool calls: {estimate.total_tool_calls} "
            f"(max: {MAX_TOOL_CALLS}). Reduce sub-tasks or granularity."
        )
    
    if estimate.estimated_latency_seconds > MAX_LATENCY_SECONDS:
        return False, (
            f"Excessive estimated latency: {estimate.estimated_latency_seconds}s "
            f"(max: {MAX_LATENCY_SECONDS}s). Consider more parallelism."
        )
    
    return True, "Within budget"

The guardrails together

The complete flow with every guardrail:

Input task
    │
    ▼
Decompose (LLM)
    │
    ▼
Validate count (2-8 sub-tasks)
    │
    ▼
Build dependency graph
    │
    ▼
Validate graph (no cycles, no broken dependencies)
    │
    ▼
Estimate complexity
    │
    ▼
Budget check
    │
    ├── PASS → Execute
    └── FAIL → Re-decompose with stricter constraints

Implementation with LangGraph

The complete decomposition node

Let's integrate everything into a LangGraph node that takes a task and produces structured sub-tasks with an execution strategy:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage
from pydantic import BaseModel, Field

model = init_chat_model("openai:gpt-4.1-mini")

State and schemas

class SubTaskSchema(BaseModel):
    id: str = Field(description="Unique ID, e.g. 'task_1'")
    description: str = Field(description="What this sub-task must do")
    depends_on: list[str] = Field(
        default_factory=list,
        description="IDs of sub-tasks that must complete first"
    )
    estimated_calls: int = Field(
        default=2,
        description="Estimated tool calls (1-3)"
    )

class DecompositionOutput(BaseModel):
    subtasks: list[SubTaskSchema] = Field(
        description="Logically ordered sub-tasks"
    )
    strategy: Literal["sequential", "parallel", "mixed"] = Field(
        description="Recommended execution strategy"
    )
    reasoning: str = Field(description="Justification for the decomposition")

class DecompState(TypedDict):
    messages: Annotated[list, add_messages]
    original_task: str
    subtasks: list[dict]
    execution_strategy: str
    task_graph: dict
    current_ready: list[str]
    results: dict
    iteration_count: int
    max_iterations: int
    decomposition_valid: bool

The decomposition node

DECOMPOSE_PROMPT = """You are an expert planner. Decompose this task into executable sub-tasks.

RULES:
1. Minimum 2, maximum 8 sub-tasks
2. Each sub-task must be solvable with 1-3 searches/tool calls
3. State explicit dependencies with IDs
4. The last sub-task must be the final synthesis
5. If sub-tasks are independent of each other, do NOT put dependencies between them
6. Choose a strategy: "sequential" if each depends on the previous one,
   "parallel" if most are independent, "mixed" if there are both

Task: {task}"""

decompose_model = model.with_structured_output(DecompositionOutput)

def decompose_node(state: DecompState) -> dict:
    """Decompose the task into structured sub-tasks."""
    task = state["original_task"]
    
    result = decompose_model.invoke(
        DECOMPOSE_PROMPT.format(task=task)
    )
    
    subtasks = [st.model_dump() for st in result.subtasks]
    
    if len(subtasks) > 8:
        subtasks = subtasks[:8]
    
    graph = {}
    all_ids = {st["id"] for st in subtasks}
    for st in subtasks:
        st["depends_on"] = [d for d in st["depends_on"] if d in all_ids]
        st["status"] = "pending"
        st["result"] = None
        graph[st["id"]] = st
    
    ready = [
        tid for tid, t in graph.items()
        if t["status"] == "pending" and all(
            graph[d]["status"] == "completed" for d in t["depends_on"]
        )
    ]
    
    return {
        "subtasks": subtasks,
        "execution_strategy": result.strategy,
        "task_graph": graph,
        "current_ready": ready,
        "results": {},
        "decomposition_valid": len(subtasks) >= 2,
        "messages": [SystemMessage(
            content=f"Plan created: {len(subtasks)} sub-tasks, "
                    f"strategy: {result.strategy}.\n"
                    f"Reasoning: {result.reasoning}"
        )],
    }

The execution node

def execute_node(state: DecompState) -> dict:
    """Execute the sub-tasks that are ready."""
    graph = state["task_graph"].copy()
    results = state["results"].copy()
    ready = state["current_ready"]
    
    if not ready:
        return {"current_ready": [], "iteration_count": state.get("iteration_count", 0) + 1}
    
    context_parts = [
        f"[{tid}]: {r[:300]}" for tid, r in results.items()
    ]
    context = "\n".join(context_parts) if context_parts else "No previous results."
    
    for tid in ready:
        task = graph[tid]
        prompt = (
            f"Original task: {state['original_task']}\n\n"
            f"Context from completed sub-tasks:\n{context}\n\n"
            f"Your current sub-task: {task['description']}\n\n"
            f"Execute this sub-task. Be concise and specific."
        )
        response = model.invoke([HumanMessage(content=prompt)])
        
        graph[tid]["status"] = "completed"
        graph[tid]["result"] = response.content
        results[tid] = response.content
    
    next_ready = [
        tid for tid, t in graph.items()
        if t["status"] == "pending" and all(
            graph[d]["status"] == "completed" for d in t["depends_on"]
        )
    ]
    
    completed_count = sum(1 for t in graph.values() if t["status"] == "completed")
    
    return {
        "task_graph": graph,
        "results": results,
        "current_ready": next_ready,
        "iteration_count": state.get("iteration_count", 0) + 1,
        "messages": [SystemMessage(
            content=f"Completed {completed_count}/{len(graph)} sub-tasks. "
                    f"Next batch: {len(next_ready)} tasks."
        )],
    }

The synthesis node

def synthesize_node(state: DecompState) -> dict:
    """Synthesize every result into a final answer."""
    results_text = "\n\n".join(
        f"## {tid}\n{result}"
        for tid, result in state["results"].items()
    )
    
    prompt = (
        f"Original task: {state['original_task']}\n\n"
        f"Research results:\n{results_text}\n\n"
        f"Synthesize a complete, coherent answer that integrates "
        f"every finding. Structure it with clear sections."
    )
    
    response = model.invoke([HumanMessage(content=prompt)])
    return {"messages": [response]}

Routing and wiring

def route_execution(state: DecompState) -> str:
    """Decide whether to keep executing or synthesize."""
    if not state.get("decomposition_valid", False):
        return "synthesize"
    
    if state.get("iteration_count", 0) >= state.get("max_iterations", 10):
        return "synthesize"
    
    if state.get("current_ready"):
        return "execute"
    
    return "synthesize"

graph = StateGraph(DecompState)
graph.add_node("decompose", decompose_node)
graph.add_node("execute", execute_node)
graph.add_node("synthesize", synthesize_node)

graph.add_edge(START, "decompose")
graph.add_conditional_edges("decompose", route_execution, {
    "execute": "execute",
    "synthesize": "synthesize",
})
graph.add_conditional_edges("execute", route_execution, {
    "execute": "execute",
    "synthesize": "synthesize",
})
graph.add_edge("synthesize", END)

agent = graph.compile()

Execution

result = agent.invoke({
    "messages": [HumanMessage(content="Research the impact of AI on startups")],
    "original_task": "Analyze the impact of AI regulation on European startups, "
                     "compare it with the United States, and recommend a strategy",
    "subtasks": [],
    "execution_strategy": "",
    "task_graph": {},
    "current_ready": [],
    "results": {},
    "iteration_count": 0,
    "max_iterations": 10,
    "decomposition_valid": False,
})

for msg in result["messages"]:
    role = "System" if isinstance(msg, SystemMessage) else "AI"
    print(f"[{role}]: {msg.content[:200]}")

Visualizing the graph

from IPython.display import Image, display
display(Image(agent.get_graph().draw_mermaid_png()))

The diagram should show: START → decompose → (execute ↔ loop) → synthesize → END. The loop between execute and itself is resolved by the conditional edge that checks whether there are more ready tasks.

Comparison: Decomposition Strategies

Recommendation: Use top-down as the default. Add bottom-up when the task is exploratory and you don't know what shape the result will take. Use analogical when you have a catalog of successful decompositions (production). Don't decompose tasks that get solved in 1-2 tool calls — the overhead isn't justified.

Connection to the Project

In this module's project (capsule 08, Research Agent with Planning and Reflection):

• The Research Agent's planning step uses top-down task decomposition. The research question gets decomposed into sub-questions, each with explicit dependencies. The agent runs the sub-questions in the order the DAG defines.
• The decompose node produces a task_graph that the execute node consumes iteratively. The routing checks current_ready — if there are ready tasks, it executes; if not, it synthesizes.
• The guardrails (max 8 sub-tasks, budget check) prevent explosive decompositions that would burn tokens without improving results.

In later modules:

• M6 (Memory): Successful decompositions get saved in long-term memory. The next time the agent faces a similar task, it uses analogical decomposition with the saved plan as a template.
• M8 (Multi-Agent): In multi-agent systems, each sub-task can be assigned to a specialized agent. The supervisor uses the dependency graph to coordinate: it launches parallel agents for independent tasks, waits for completions for dependent ones.
• M9 (Testing): Decomposition is unit-testable — given an input, does it generate reasonable sub-tasks? Golden datasets of expected decompositions.

Troubleshooting

Problem 1: The LLM generates sub-tasks with circular dependencies

Symptom: has_cycle returns True. task_A depends on task_B and task_B depends on task_A.

Cause: The LLM doesn't always understand what "dependency" means in a DAG. Sometimes it interprets "related to" as "depends on".

Solution: Validate the graph after decomposition and, if there are cycles, re-invoke with explicit instructions:

if has_cycle(graph):
    retry_prompt = (
        f"Your previous decomposition has circular dependencies. "
        f"Rule: if task_A depends on task_B, task_B can NOT depend on task_A. "
        f"Re-decompose without cycles:\n\nTask: {task}"
    )
    result = decompose_model.invoke(retry_prompt)

Problem 2: All the sub-tasks are independent (no synthesis)

Symptom: 5 sub-tasks with depends_on: [] each. There's no synthesis task.

Cause: The LLM decomposed but didn't include the integration step.

Solution: In the decomposition prompt, the rule "the last sub-task must be the final synthesis" should prevent it. If not, force the synthesis:

if not any(len(st["depends_on"]) > 1 for st in subtasks):
    synthesis = SubTaskSchema(
        id=f"task_{len(subtasks) + 1}",
        description="Synthesize every finding into a coherent answer",
        depends_on=[st["id"] for st in subtasks],
        estimated_calls=1,
    )
    subtasks.append(synthesis.model_dump())

Problem 3: Sub-tasks that are too vague ("research the topic")

Symptom: Sub-tasks like "Search for information about AI" that aren't actionable.

Solution: Add specificity criteria to the prompt:

SPECIFIC_CONSTRAINT = """
Every sub-task must specify:
- WHAT to search for (concrete terms, not generic ones)
- WHERE to search (source type: papers, news, data)
- HOW MUCH (1-3 specific data points/sources)

❌ Bad: "Research AI"
✅ Good: "Find 3 statistics on AI adoption in European companies 2024-2025"
"""

Problem 4: The execution loop doesn't finish

Symptom: The agent runs indefinitely because get_ready_tasks always returns tasks.

Cause: A sub-task fails (its status stays "pending") and blocks the dependent ones, but the loop keeps iterating with no progress.

Solution: Detect the lack of progress and force termination:

def route_execution(state: DecompState) -> str:
    if state.get("iteration_count", 0) >= state.get("max_iterations", 10):
        return "synthesize"
    
    graph = state.get("task_graph", {})
    completed = sum(1 for t in graph.values() if t["status"] == "completed")
    
    if state.get("iteration_count", 0) > 2 and completed == state.get("prev_completed", 0):
        return "synthesize"
    
    if state.get("current_ready"):
        return "execute"
    
    return "synthesize"

Problem 5: The cost of decomposition exceeds the benefit

Symptom: A simple task ("look up the weather in Madrid") goes through decomposition, generating 4 unnecessary sub-tasks. The total cost: 5 LLM calls instead of 1.

Solution: Add a gate up front that decides whether the task deserves decomposition:

class DecompDecision(BaseModel):
    needs_decomposition: bool = Field(
        description="True if the task has multiple distinct steps"
    )
    reason: str

def should_decompose_node(state: DecompState) -> dict:
    """Decide whether the task needs decomposition."""
    decision_model = model.with_structured_output(DecompDecision)
    decision = decision_model.invoke(
        f"Does this task require decomposition into sub-tasks, "
        f"or can it be solved directly?\n\nTask: {state['original_task']}"
    )
    return {"decomposition_valid": decision.needs_decomposition}

Exercises

Exercise 1: Identify dependencies (Easy)

Given this list of sub-tasks, draw the dependency graph and identify which tasks can run in parallel:

task_1: Look up the current price of Bitcoin
task_2: Look up the current price of Ethereum
task_3: Look up crypto regulation in the EU
task_4: Compute the Bitcoin-Ethereum correlation (needs task_1, task_2)
task_5: Analyze the regulatory impact on prices (needs task_3, task_4)
task_6: Generate the final report (needs task_5)

Graph:
task_1 ──┐
         ├──→ task_4 ──┐
task_2 ──┘              ├──→ task_5 ──→ task_6
task_3 ─────────────────┘

Execution by layers:
  Layer 1 (parallel): task_1, task_2, task_3
  Layer 2:            task_4 (waits for layer 1: only task_1 and task_2)
  Layer 3:            task_5 (waits for task_3 + task_4)
  Layer 4:            task_6 (waits for task_5)

DAG depth: 4
Latency: 4 cycles (vs 6 if it were sequential)
task_3 is independent of task_4 but task_5 needs both.

Exercise 2: Implement get_ready_tasks (Medium)

Implement the get_ready_tasks function that, given a dependency graph, returns the tasks that can run right now (status "pending" and every dependency "completed"):

graph = {
    "t1": {"status": "completed", "depends_on": []},
    "t2": {"status": "pending", "depends_on": []},
    "t3": {"status": "pending", "depends_on": ["t1", "t2"]},
    "t4": {"status": "pending", "depends_on": ["t1"]},
}
# What does get_ready_tasks(graph) return?

def get_ready_tasks(graph: dict) -> list[str]:
    ready = []
    for task_id, task in graph.items():
        if task["status"] != "pending":
            continue
        deps_met = all(
            graph[dep]["status"] == "completed"
            for dep in task["depends_on"]
        )
        if deps_met:
            ready.append(task_id)
    return ready

result = get_ready_tasks(graph)
print(result)  # ["t2", "t4"]

# t2: pending, depends_on=[] → ready (no dependencies)
# t3: pending, depends_on=["t1", "t2"] → NOT ready (t2 isn't completed)
# t4: pending, depends_on=["t1"] → ready (t1 is completed)

Exercise 3: Decomposition with guardrails (Medium)

Implement a function that decomposes a task with validations: maximum 8 sub-tasks, no cycles, and a mandatory synthesis sub-task. If validation fails, retry once.

from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

class SubTask(BaseModel):
    id: str
    description: str
    depends_on: list[str] = []

class Decomposition(BaseModel):
    subtasks: list[SubTask]

decomposer = model.with_structured_output(Decomposition)

def decompose_with_guardrails(task: str, max_retries: int = 1) -> list[dict]:
    prompt = (
        f"Decompose this into sub-tasks (2-8). "
        f"The last one must be a synthesis that depends on the others.\n\n"
        f"Task: {task}"
    )
    
    for attempt in range(max_retries + 1):
        result = decomposer.invoke(prompt)
        subtasks = [st.model_dump() for st in result.subtasks]
        
        if len(subtasks) > 8:
            subtasks = subtasks[:8]
        
        all_ids = {st["id"] for st in subtasks}
        for st in subtasks:
            st["depends_on"] = [d for d in st["depends_on"] if d in all_ids]
            st["status"] = "pending"
        
        has_synthesis = any(len(st["depends_on"]) > 1 for st in subtasks)
        if not has_synthesis:
            subtasks.append({
                "id": "synthesis",
                "description": "Synthesize the findings",
                "depends_on": [st["id"] for st in subtasks],
                "status": "pending",
            })
        
        graph = {st["id"]: st for st in subtasks}
        if not has_cycle_check(graph):
            return subtasks
        
        prompt = (
            f"Re-decompose WITHOUT circular dependencies. "
            f"If A depends on B, B can NOT depend on A.\n\n"
            f"Task: {task}"
        )
    
    raise ValueError("Couldn't generate a valid decomposition")

def has_cycle_check(graph):
    visited, in_stack = set(), set()
    def dfs(nid):
        visited.add(nid)
        in_stack.add(nid)
        for dep in graph[nid].get("depends_on", []):
            if dep in in_stack:
                return True
            if dep not in visited and dep in graph and dfs(dep):
                return True
        in_stack.discard(nid)
        return False
    return any(dfs(n) for n in graph if n not in visited)

tasks = decompose_with_guardrails(
    "Compare AI agent frameworks: LangGraph, CrewAI, AutoGen"
)
for t in tasks:
    deps = f" → depends on {t['depends_on']}" if t["depends_on"] else ""
    print(f"  {t['id']}: {t['description']}{deps}")

Exercise 4: Parallel execution with asyncio (Hard)

Implement an executor that takes a dependency graph and runs the sub-tasks respecting the dependencies, parallelizing the independent ones. Use asyncio.gather for simultaneous execution and a semaphore to limit concurrency to 3.

import asyncio
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage

model = init_chat_model("openai:gpt-4.1-mini")

async def run_subtask(task: dict, context: str, semaphore: asyncio.Semaphore) -> tuple[str, str]:
    """Run a sub-task with a concurrency limit."""
    async with semaphore:
        prompt = f"Context:\n{context}\n\nSub-task: {task['description']}\n\nExecute it concisely."
        response = await model.ainvoke([HumanMessage(content=prompt)])
        return task["id"], response.content

async def execute_graph(graph: dict, max_concurrent: int = 3) -> dict:
    """Execute the whole graph respecting dependencies."""
    semaphore = asyncio.Semaphore(max_concurrent)
    results = {}
    
    iteration = 0
    max_iters = 20
    
    while iteration < max_iters:
        ready = [
            tid for tid, t in graph.items()
            if t["status"] == "pending"
            and all(graph[d]["status"] == "completed" for d in t["depends_on"])
        ]
        
        if not ready:
            break
        
        context = "\n".join(f"[{k}]: {v[:200]}" for k, v in results.items())
        
        completed = await asyncio.gather(*[
            run_subtask(graph[tid], context, semaphore) for tid in ready
        ])
        
        for tid, result in completed:
            graph[tid]["status"] = "completed"
            graph[tid]["result"] = result
            results[tid] = result
            print(f"  ✓ {tid} completed")
        
        iteration += 1
    
    pending = [tid for tid, t in graph.items() if t["status"] == "pending"]
    if pending:
        print(f"  ⚠ Tasks not completed: {pending}")
    
    return results

# Usage:
# results = asyncio.run(execute_graph(my_graph))

Exercise 5: LangGraph graph with decompose → execute → synthesize (Hard)

Implement this capsule's complete StateGraph: a decompose node that generates sub-tasks, an execute node that processes the ready ones, conditional routing (more tasks → execute, none → synthesize), and a synthesize node that integrates the results. Include max_iterations as a safety net.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage
from pydantic import BaseModel, Field

model = init_chat_model("openai:gpt-4.1-mini")

class SubTaskOut(BaseModel):
    id: str
    description: str
    depends_on: list[str] = []

class DecompOut(BaseModel):
    subtasks: list[SubTaskOut]

decomp_model = model.with_structured_output(DecompOut)

class GraphState(TypedDict):
    messages: Annotated[list, add_messages]
    task: str
    graph: dict
    ready: list[str]
    results: dict
    iter: int
    max_iter: int

def decompose(state: GraphState) -> dict:
    result = decomp_model.invoke(
        f"Decompose this (3-6 sub-tasks, the last is a synthesis): {state['task']}"
    )
    g = {}
    all_ids = {s.id for s in result.subtasks}
    for s in result.subtasks:
        g[s.id] = {
            "id": s.id,
            "description": s.description,
            "depends_on": [d for d in s.depends_on if d in all_ids],
            "status": "pending",
            "result": None,
        }
    ready = [tid for tid, t in g.items()
             if not t["depends_on"]]
    return {
        "graph": g, "ready": ready, "results": {},
        "messages": [SystemMessage(content=f"Decomposed into {len(g)} sub-tasks.")],
    }

def execute(state: GraphState) -> dict:
    g = {k: dict(v) for k, v in state["graph"].items()}
    results = dict(state["results"])
    ctx = "\n".join(f"[{k}]: {v[:200]}" for k, v in results.items())
    
    for tid in state["ready"]:
        resp = model.invoke([HumanMessage(
            content=f"Context:\n{ctx}\n\nSub-task: {g[tid]['description']}"
        )])
        g[tid]["status"] = "completed"
        g[tid]["result"] = resp.content
        results[tid] = resp.content
    
    ready = [
        tid for tid, t in g.items()
        if t["status"] == "pending"
        and all(g[d]["status"] == "completed" for d in t["depends_on"])
    ]
    done = sum(1 for t in g.values() if t["status"] == "completed")
    return {
        "graph": g, "ready": ready, "results": results,
        "iter": state.get("iter", 0) + 1,
        "messages": [SystemMessage(content=f"{done}/{len(g)} completed.")],
    }

def synthesize(state: GraphState) -> dict:
    all_results = "\n\n".join(
        f"**{k}**: {v}" for k, v in state["results"].items()
    )
    resp = model.invoke([HumanMessage(
        content=f"Task: {state['task']}\n\nResults:\n{all_results}\n\n"
                f"Synthesize a complete final answer."
    )])
    return {"messages": [resp]}

def route(state: GraphState) -> str:
    if state.get("iter", 0) >= state.get("max_iter", 10):
        return "synthesize"
    if state.get("ready"):
        return "execute"
    return "synthesize"

g = StateGraph(GraphState)
g.add_node("decompose", decompose)
g.add_node("execute", execute)
g.add_node("synthesize", synthesize)

g.add_edge(START, "decompose")
g.add_conditional_edges("decompose", route, {
    "execute": "execute", "synthesize": "synthesize",
})
g.add_conditional_edges("execute", route, {
    "execute": "execute", "synthesize": "synthesize",
})
g.add_edge("synthesize", END)

agent = g.compile()

result = agent.invoke({
    "messages": [],
    "task": "Compare Python vs Rust for backend development in 2025",
    "graph": {}, "ready": [], "results": {},
    "iter": 0, "max_iter": 8,
})

print("=== Final Answer ===")
print(result["messages"][-1].content)

Summary

In this capsule you learned:

• Task decomposition is the HOW of planning. It isn't "making a list" — it's analyzing a problem's structure, identifying dependencies, and defining an execution strategy. The right granularity: each sub-task requires 1-3 tool calls.
• Three decomposition strategies. Top-down (divide and conquer, the most common), bottom-up (identify atoms and group them), analogical (reuse past decompositions). Each works better in different contexts.
• Dependency graphs model the relationships. Sub-tasks as nodes, dependencies as edges. A DAG tells you what can run in parallel and what must be sequential. get_ready_tasks is the key function.
• Parallel vs sequential is a per-pair decision. Use parallel when there are 3+ independent tasks and latency matters. Use semaphores to control concurrency and respect rate limits.
• Guardrails prevent the explosion. Max sub-tasks (8), max depth (2), cycle validation, complexity estimation, and budget gates. Without guardrails, a decomposition can generate 50 useless sub-tasks.
• Implementation in LangGraph. Three nodes (decompose, execute, synthesize) with conditional routing. The execute loop checks current_ready on every iteration until it runs out of sub-tasks or hits max_iterations.

Next capsule: Reflection and Self-Correction — the agent evaluates the quality of its own results and corrects itself before delivering. If task decomposition is "planning what to do", reflection is "evaluating how well I did it."

Additional Resources

1. HuggingGPT Paper — Task planning with LLMs: decomposition, assignment to specialized models, execution
2. LangGraph DAG Patterns — Branching and fan-out/fan-in in LangGraph
3. Plan-and-Solve Prompting — Improving reasoning via explicit decomposition
4. Chameleon Paper — Plug-and-play compositional reasoning with task decomposition
5. LangGraph Async Guide — Async execution for parallelism in graphs