Module 6: Prompt Composition and Chaining

3. Decomposition of Complex Tasks

Next →

Overview

Decomposition is the systematic process of splitting a complex task into manageable sub-tasks, each with a clear input, a specialized prompt, and an output with a validatable schema. It's the fundamental skill for designing scalable LLM pipelines.

In this capsule you'll learn the complete decomposition framework: how to identify the natural boundaries of sub-tasks, how to design interfaces between prompts, and how to handle dependencies so you can parallelize whenever possible.

The Decomposition Framework

The 4 Steps

1. IDENTIFY sub-tasks
   └── Which steps are independent? Which ones depend on others?

2. DEFINE interfaces
   └── Input/output of each stage (schema with data types)

3. DESIGN prompts
   └── One specialized prompt per sub-task

4. ORDER by dependencies
   └── What can run in parallel vs. what must be sequential

When to split a task

A task should be split when:

  • It has multiple distinct goals (extract + analyze + report)
  • The output of one part is needed as the input of another
  • Parts of the task can be optimized independently
  • Error handling needs to be granular (retry only the failed part)
  • Independent parts can run in parallel

Complete Example: Research → Analyze → Write

from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Optional
import json

client = OpenAI()

# ============================================================
# INTERFACE SCHEMAS (Pydantic)
# ============================================================

class ResearchFindings(BaseModel):
    """Output of the research sub-task."""
    findings: list[str] = Field(description="List of key findings")
    source_types: list[str] = Field(description="Source type of each finding")
    confidence: float = Field(ge=0, le=1, description="Overall confidence in the findings")

class AnalysisOutput(BaseModel):
    """Output of the analysis sub-task."""
    trends: list[str] = Field(description="Identified trends")
    evidence: dict[str, list[str]] = Field(description="Evidence per trend")
    contradictions: list[str] = Field(description="Contradictions found")
    main_conclusion: str = Field(description="The most important conclusion")

class ReportSection(BaseModel):
    """A section of the final report."""
    title: str
    content: str
    citations: list[str]

class ReportOutput(BaseModel):
    """Final output of the pipeline."""
    title: str
    executive_summary: str
    sections: list[ReportSection]
    conclusions: list[str]
    recommendations: list[str]


# ============================================================
# SPECIALIZED PROMPTS
# ============================================================

PROMPT_RESEARCH = """You are an expert researcher. Gather information about the following topic.

TOPIC: {query}

Provide 8-12 key findings based on your knowledge. For each finding:
- Be specific and grounded in known facts
- State the source type (statistic, academic study, market trend, practical example)

Respond in JSON:
{{
    "findings": [
        "finding 1",
        "finding 2"
    ],
    "source_types": [
        "statistic",
        "study"
    ],
    "confidence": 0.85
}}"""

PROMPT_ANALYZE = """You are a senior analyst. Synthesize the following findings into actionable insights.

FINDINGS:
{findings_str}

ORIGINAL TOPIC: {query}

Identify:
1. Main trends (3-5 clear trends)
2. Evidence supporting each trend
3. Contradictions or tensions between findings
4. The most important conclusion

Respond in JSON:
{{
    "trends": ["trend 1", "trend 2"],
    "evidence": {{
        "trend 1": ["evidence A", "evidence B"],
        "trend 2": ["evidence C"]
    }},
    "contradictions": ["contradiction if there is one"],
    "main_conclusion": "The most important conclusion in 2 sentences"
}}"""

PROMPT_WRITE = """You are a technical writer. Generate a professional report based on this analysis.

TOPIC: {query}
MAIN CONCLUSION: {conclusion}
IDENTIFIED TRENDS:
{trends_str}

The report must:
1. Have a descriptive title
2. An executive summary (2-3 sentences)
3. 2-3 main sections (one per key trend)
4. Conclusions (3 points)
5. Concrete recommendations (3-5 actions)

Respond in JSON following the provided schema."""


# ============================================================
# PIPELINE IMPLEMENTATION
# ============================================================

def subtask_research(query: str) -> ResearchFindings:
    """
    Sub-task 1: Research - Gather information about the topic.
    """
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": PROMPT_RESEARCH.format(query=query)}],
        temperature=0,
        max_tokens=800,
        response_format={"type": "json_object"}
    )
    
    data = json.loads(response.choices[0].message.content)
    return ResearchFindings(**data)

def subtask_analyze(query: str, research: ResearchFindings) -> AnalysisOutput:
    """
    Sub-task 2: Analysis - Synthesize findings into insights.
    Depends on: subtask_research
    """
    findings_str = "\n".join([f"- {h}" for h in research.findings])
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": PROMPT_ANALYZE.format(
                findings_str=findings_str,
                query=query
            )
        }],
        temperature=0,
        max_tokens=800,
        response_format={"type": "json_object"}
    )
    
    data = json.loads(response.choices[0].message.content)
    return AnalysisOutput(**data)

def subtask_write(query: str, analysis: AnalysisOutput) -> ReportOutput:
    """
    Sub-task 3: Write - Generate a structured report.
    Depends on: subtask_analyze
    """
    trends_str = "\n".join([
        f"- {t}: {', '.join(analysis.evidence.get(t, [])[:2])}"
        for t in analysis.trends
    ])
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": PROMPT_WRITE.format(
                query=query,
                conclusion=analysis.main_conclusion,
                trends_str=trends_str
            )
        }],
        temperature=0.2,
        max_tokens=1000,
        response_format={"type": "json_object"}
    )
    
    data = json.loads(response.choices[0].message.content)
    # Build the ReportOutput (flexible schema handling)
    sections = [ReportSection(**s) for s in data.get("sections", [])]
    return ReportOutput(
        title=data.get("title", "Report"),
        executive_summary=data.get("executive_summary", ""),
        sections=sections,
        conclusions=data.get("conclusions", []),
        recommendations=data.get("recommendations", [])
    )

def pipeline_research_analyze_write(query: str, verbose: bool = True) -> dict:
    """
    Complete pipeline: Research → Analyze → Write
    
    Args:
        query: The topic to research and write about
        verbose: If True, prints the progress
    
    Returns:
        dict with the outputs of the 3 stages
    """
    import time
    t_total = time.time()
    
    if verbose:
        print(f"📚 Researching: {query}\n")
    
    # Stage 1: Research
    t0 = time.time()
    research = subtask_research(query)
    if verbose:
        print(f"✓ Research complete ({time.time()-t0:.1f}s)")
        print(f"  {len(research.findings)} findings, confidence: {research.confidence:.0%}")
    
    # Stage 2: Analysis
    t0 = time.time()
    analysis = subtask_analyze(query, research)
    if verbose:
        print(f"✓ Analysis complete ({time.time()-t0:.1f}s)")
        print(f"  {len(analysis.trends)} trends identified")
        print(f"  Conclusion: {analysis.main_conclusion[:80]}...")
    
    # Stage 3: Write
    t0 = time.time()
    report = subtask_write(query, analysis)
    if verbose:
        print(f"✓ Report generated ({time.time()-t0:.1f}s)")
        print(f"  Title: {report.title}")
        print(f"  {len(report.sections)} sections, {len(report.recommendations)} recommendations")
        print(f"\nTotal time: {time.time()-t_total:.1f}s")
    
    return {
        "research": research.model_dump(),
        "analysis": analysis.model_dump(),
        "report": report.model_dump()
    }


# Example:
if __name__ == "__main__":
    result = pipeline_research_analyze_write(
        "The impact of LLMs on the programming job market in 2025-2026"
    )
    print("\n=== FINAL REPORT ===")
    report = result["report"]
    print(f"# {report['title']}\n")
    print(f"## Executive Summary\n{report['executive_summary']}\n")
    print("## Recommendations")
    for r in report["recommendations"]:
        print(f"- {r}")

Decomposition Framework: Common Patterns

Pattern 1: Extract → Transform → Load (ETL for LLMs)

class ETLPipeline:
    """
    The ETL pattern adapted for LLM pipelines:
    - Extract: Pull information out of the input (entities, data, structure)
    - Transform: Transform/analyze the extracted data
    - Load: Format the output into the desired schema
    """
    
    def __init__(self, domain: str):
        self.domain = domain
    
    def extract(self, document: str) -> dict:
        """Extract structured information from the document."""
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"""Extract the key information from this {self.domain}:
            
{document}

Respond in JSON with the fields relevant to this type of document."""}],
            temperature=0,
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)
    
    def transform(self, extracted_data: dict, goal: str) -> dict:
        """Transform/analyze the data according to the goal."""
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"""Analyze this data for the goal: {goal}

Data: {json.dumps(extracted_data, ensure_ascii=False)}

Generate insights, compute relevant metrics, identify patterns.
Respond in JSON."""}],
            temperature=0,
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)
    
    def load(self, transformed_data: dict, output_format: str) -> str:
        """Format the output into the desired format."""
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"""Format this data as {output_format}:

{json.dumps(transformed_data, ensure_ascii=False)}"""}],
            temperature=0.2
        )
        return response.choices[0].message.content
    
    def run(self, document: str, goal: str, output_format: str) -> dict:
        extracted = self.extract(document)
        transformed = self.transform(extracted, goal)
        output = self.load(transformed, output_format)
        return {"extracted": extracted, "transformed": transformed, "output": output}


# Example: Employment contract analysis
etl = ETLPipeline("employment contract")
result = etl.run(
    document="Employment contract. Employee: Juan García. Salary: €45,000/year. Working hours: 40h/week...",
    goal="identify non-standard conditions or risk clauses",
    output_format="an executive summary in bullet points"
)
print(result["output"])

Pattern 2: Map-Reduce for Long Documents

from typing import Any

def map_reduce_documents(
    documents: list[str],
    prompt_map: str,
    prompt_reduce: str,
    batch_size: int = 5
) -> str:
    """
    Processes multiple documents with Map-Reduce:
    - Map: Applies prompt_map to each document individually
    - Reduce: Synthesizes every result into a single one
    
    Args:
        documents: List of texts to process
        prompt_map: Template to process each document. Uses {document}
        prompt_reduce: Template to synthesize. Uses {results}
        batch_size: Process in batches when there are many documents
    
    Returns:
        Final synthesis of all the documents
    """
    # MAP: Process each document
    map_results = []
    for i, doc in enumerate(documents):
        print(f"Map: processing document {i+1}/{len(documents)}...")
        result = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt_map.format(document=doc)}],
            temperature=0,
            max_tokens=400
        ).choices[0].message.content
        map_results.append(result)
    
    # If there are many results, do a hierarchical reduce
    while len(map_results) > batch_size:
        new_results = []
        for i in range(0, len(map_results), batch_size):
            batch = map_results[i:i+batch_size]
            batch_str = "\n---\n".join([f"Result {i+j+1}:\n{r}" for j, r in enumerate(batch)])
            
            partial_reduce = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt_reduce.format(results=batch_str)}],
                temperature=0,
                max_tokens=400
            ).choices[0].message.content
            new_results.append(partial_reduce)
        
        map_results = new_results
    
    # REDUCE: Final synthesis
    all_str = "\n---\n".join([f"Result {i+1}:\n{r}" for i, r in enumerate(map_results)])
    
    final_synthesis = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt_reduce.format(results=all_str)}],
        temperature=0.1,
        max_tokens=600
    ).choices[0].message.content
    
    return final_synthesis


# Example: Analyze multiple customer reviews
reviews = [
    "The product is excellent but the technical support is terrible.",
    "It arrived late but the quality exceeded my expectations.",
    "I wouldn't recommend it. The manual is in English and there's no Spanish version.",
    "Perfect in every respect. Worth every euro.",
    "It works well but it overheats after 2 hours of use.",
]

prompt_map_reviews = "Extract in 1-2 sentences the main problem or praise in this review: {document}"
prompt_reduce_reviews = "Synthesize these review analyses into an executive summary with: positive trends, negative trends, and 3 priority actions.\n\n{results}"

synthesis = map_reduce_documents(reviews, prompt_map_reviews, prompt_reduce_reviews)
print("Review synthesis:", synthesis)

Pattern 3: Decomposition with Smart Parallelization

import asyncio
from openai import AsyncOpenAI
from dataclasses import dataclass
from typing import set

@dataclass
class SubTaskConfig:
    name: str
    prompt_template: str
    dependencies: list[str]  # Names of the sub-tasks it depends on
    can_parallelize: bool = True

def build_dag(sub_tasks: list[SubTaskConfig]) -> dict[str, set]:
    """Builds a DAG (Directed Acyclic Graph) of dependencies."""
    dag = {t.name: set(t.dependencies) for t in sub_tasks}
    return dag

def topological_order(dag: dict[str, set]) -> list[list[str]]:
    """
    Computes the topological order of the DAG.
    Returns groups of tasks that can run in parallel.
    """
    group_order = []
    dag_copy = {k: set(v) for k, v in dag.items()}
    
    while dag_copy:
        # Tasks with no pending dependencies (they can run in this step)
        no_deps = [name for name, deps in dag_copy.items() if not deps]
        
        if not no_deps:
            raise ValueError("Cycle detected in the dependency DAG")
        
        group_order.append(no_deps)
        
        # Remove these tasks from the graph
        for name in no_deps:
            del dag_copy[name]
        
        # Remove these tasks from the dependencies of the remaining ones
        for name in list(dag_copy.keys()):
            dag_copy[name] -= set(no_deps)
    
    return group_order

async def run_with_parallelization(
    sub_tasks: list[SubTaskConfig],
    initial_input: str,
    verbose: bool = True
) -> dict[str, str]:
    """
    Runs the sub-tasks respecting dependencies and parallelizing whenever possible.
    """
    client_async = AsyncOpenAI()
    dag = build_dag(sub_tasks)
    groups = topological_order(dag)
    
    configs_map = {t.name: t for t in sub_tasks}
    results = {"_initial_input": initial_input}
    
    if verbose:
        print(f"Execution plan: {len(groups)} groups")
        for i, group in enumerate(groups, 1):
            print(f"  Group {i}: {group} ({'parallel' if len(group) > 1 else 'sequential'})")
    
    for i, group in enumerate(groups, 1):
        if verbose:
            print(f"\n[Group {i}] Running: {group}")
        
        async def run_subtask(name: str) -> tuple[str, str]:
            config = configs_map[name]
            # Build the context with the available results
            context = {**results}
            context["input"] = results.get(config.dependencies[-1], initial_input) if config.dependencies else initial_input
            
            prompt = config.prompt_template.format(**context)
            response = await client_async.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                temperature=0,
                max_tokens=400
            )
            return name, response.choices[0].message.content
        
        # Run the group in parallel
        async_tasks = [run_subtask(name) for name in group]
        new_results = await asyncio.gather(*async_tasks)
        results.update(dict(new_results))
        
        if verbose:
            for name, result in new_results:
                print(f"  ✓ {name}: {result[:80]}...")
    
    return {k: v for k, v in results.items() if not k.startswith("_")}


# Usage example with a DAG:
analysis_subtasks = [
    SubTaskConfig(
        name="entity_extraction",
        prompt_template="Extract people, organizations and dates from: {input}",
        dependencies=[]  # Depends on nothing
    ),
    SubTaskConfig(
        name="summary",
        prompt_template="Summarize in 3 sentences: {input}",
        dependencies=[]  # Depends on nothing (parallel to entity_extraction)
    ),
    SubTaskConfig(
        name="bias_analysis",
        prompt_template="Analyze the bias of this text taking the entities into account: {entity_extraction}\n\nText: {input}",
        dependencies=["entity_extraction"]  # Needs the entities
    ),
    SubTaskConfig(
        name="final_report",
        prompt_template="Generate a report based on: Summary: {summary} | Bias: {bias_analysis} | Entities: {entity_extraction}",
        dependencies=["summary", "bias_analysis", "entity_extraction"]  # Needs everything
    )
]

# asyncio.run(run_with_parallelization(analysis_subtasks, "The text to analyze..."))

Designing Interfaces Between Prompts

The Interface Contracts Principle

from pydantic import BaseModel, field_validator
from typing import Literal

class ExtractionInterface(BaseModel):
    """
    Interface contract between the extraction prompt and the analysis prompt.
    Every field has an explicit type and validations.
    """
    entities: list[str]
    dates: list[str]
    key_numbers: list[float]
    sentiment: Literal["positive", "negative", "neutral", "mixed"]
    
    @field_validator("key_numbers", mode="before")
    @classmethod
    def clean_numbers(cls, v):
        """Normalize numbers: strip commas, percent signs, etc."""
        result = []
        for item in v:
            if isinstance(item, (int, float)):
                result.append(float(item))
            elif isinstance(item, str):
                cleaned = item.replace(",", "").replace("%", "").strip()
                try:
                    result.append(float(cleaned))
                except ValueError:
                    pass
        return result

class AnalysisInterface(BaseModel):
    """
    Interface contract between analysis and report generation.
    """
    conclusions: list[str]
    risk_level: Literal["low", "medium", "high", "critical"]
    required_actions: list[str]
    key_data: dict[str, float]

def prompt_with_interface_contract(
    text: str,
    schema: type[BaseModel]
) -> BaseModel:
    """
    Runs a prompt and validates the output against an interface schema.
    Retries if the format is wrong.
    """
    schema_json = schema.model_json_schema()
    
    prompt = f"""Analyze the following text and respond in JSON following this schema:
{json.dumps(schema_json, indent=2)}

Text: {text}

IMPORTANT: The JSON must be valid and contain every required field."""
    
    for attempt in range(3):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
            response_format={"type": "json_object"}
        ).choices[0].message.content
        
        try:
            data = json.loads(response)
            return schema(**data)
        except Exception as e:
            if attempt < 2:
                prompt += f"\n\nError on the previous attempt: {e}. Please fix it."
            else:
                raise ValueError(f"Could not parse the output after 3 attempts: {e}")

# Example:
news_text = "Apple reported revenue of $124.3 million on January 15, 2026. Tim Cook highlighted 23% growth in services."
interface_result = prompt_with_interface_contract(news_text, ExtractionInterface)
print(f"Entities: {interface_result.entities}")
print(f"Numbers: {interface_result.key_numbers}")
print(f"Sentiment: {interface_result.sentiment}")

Decomposition Troubleshooting

Problem 1: Sub-tasks that are too big

Symptom: A sub-task has multiple different outputs and the prompt is far too complex.

Warning sign: A prompt that uses more than 5 "extract/generate/analyze" verbs in the same instruction.

Solution:

# ❌ SUB-TASK THAT IS TOO BIG
huge_prompt = """Analyze this contract.
Extract every party involved.
Identify the risk clauses.
Compute the penalties.
Compare against industry standards.
Generate recommendations.
Translate the summary into English."""

# ✓ SPLIT INTO 5 SPECIALIZED SUB-TASKS
specialized_prompts = {
    "parties": "Identify every party (people and organizations) in this contract: {input}",
    "risk_clauses": "List the clauses that represent a legal or financial risk: {input}",
    "penalties": "Extract and compute the penalties mentioned in the contract: {input}",
    "comparison": "Compare these clauses {risk_clauses} against standard industry practice",
    "recommendations": "Generate 5 recommendations based on the risk analysis: {risk_clauses}"
}

Problem 2: Ambiguous interfaces

Symptom: Stage 2 doesn't know exactly what format to expect from stage 1.

Solution:

# ❌ AMBIGUOUS INTERFACE
vague_prompt = "Extract the important data from the text."

# ✓ INTERFACE WITH AN EXPLICIT SCHEMA
specific_prompt = """Extract from the text:
1. Date (format: DD/MM/YYYY)
2. Total amount (number only, no symbols)
3. Parties (list of proper names)
4. Document type (contract/invoice/letter/other)

Respond in JSON:
{{"date": "DD/MM/YYYY or null", "amount": number or null, "parties": [str], "type": str}}"""

Problem 3: Context loss between sub-tasks

Symptom: Stage 3 doesn't have enough context about the original problem because it only receives the output of stage 2.

Solution: Pass the original input plus the relevant intermediate outputs:

def prompt_with_full_context(
    original_input: str,
    previous_outputs: dict,
    current_template: str
) -> str:
    """
    Builds a prompt that includes context from the original problem
    plus the relevant outputs of the previous stages.
    """
    context = f"ORIGINAL CONTEXT: {original_input[:500]}\n\n"
    
    for name, output in previous_outputs.items():
        context += f"RESULT OF {name.upper()}:\n{output[:200]}\n\n"
    
    return context + current_template.format(input=previous_outputs.get("last", original_input))

Exercises

Exercise 1: Decomposition of a financial data analysis

You have to analyze a company's annual report. Decompose this task into sub-tasks, define the interfaces, and specify the dependencies.

See solution 
financial_subtasks = [
    SubTaskConfig(
        name="metrics_extraction",
        prompt_template="""Extract the following numbers from the financial report:
- Total revenue
- EBITDA
- Net income
- Net debt
- Employees
Respond in JSON: {{"revenue": float, "ebitda": float, "net_income": float, "net_debt": float, "employees": int}}
Report: {input}""",
        dependencies=[]
    ),
    SubTaskConfig(
        name="narrative_extraction",
        prompt_template="Extract the 5 key points of the CEO's narrative and the strategic message: {input}",
        dependencies=[]  # Parallel to metrics_extraction
    ),
    SubTaskConfig(
        name="ratio_calculation",
        prompt_template="""Compute these financial ratios with the data:
{metrics_extraction}
- EBITDA margin (EBITDA/Revenue)
- ROE if equity is available
- Leverage (Debt/EBITDA)
Respond in JSON with values and interpretation.""",
        dependencies=["metrics_extraction"]
    ),
    SubTaskConfig(
        name="strategic_synthesis",
        prompt_template="""Generate a strategic analysis using:
Metrics: {metrics_extraction}
Ratios: {ratio_calculation}
Narrative: {narrative_extraction}
Include: financial health, risks, opportunities.""",
        dependencies=["metrics_extraction", "ratio_calculation", "narrative_extraction"]
    )
]

Exercise 2: Implement Map-Reduce for an FAQ

You have 20 frequently asked questions from customers. Use Map-Reduce to:

  1. (Map) Classify each question by category
  2. (Reduce) Generate a consolidated answer per category
See solution 
faqs = [
    "How do I change my password?",
    "Why can't I log in?",
    "Where is my order?",
    "How long does shipping take?",
    "How do I return a product?",
    "Do you have a mobile app?",
    "How do I cancel my subscription?",
]

prompt_map = """Classify this question and extract the intent:
Question: {document}
JSON: {{"category": "account|orders|returns|app|subscription|other", "intent": str}}"""

prompt_reduce = """You have these FAQ classifications:
{results}

Generate:
1. Summary per category (N questions per category)
2. The 3 most frequent categories
3. Information gaps that could become additional FAQs"""

result = map_reduce_documents(faqs, prompt_map, prompt_reduce)
print(result)

Exercise 3: Design interfaces with Pydantic

For a Python code analysis pipeline, design the Pydantic schemas for the interfaces between:

  1. Complexity analysis
  2. Code smell detection
  3. Refactoring suggestions
See solution 
from pydantic import BaseModel, Field
from typing import Literal

class ComplexityAnalysis(BaseModel):
    cyclomatic_complexity: int = Field(ge=1, description="Cyclomatic complexity of the function")
    lines_of_code: int
    functions: list[str]
    level: Literal["simple", "moderate", "complex", "very_complex"]

class CodeSmell(BaseModel):
    type: str
    description: str
    line: Optional[int] = None
    severity: Literal["low", "medium", "high"]

class CodeSmellsAnalysis(BaseModel):
    smells: list[CodeSmell]
    estimated_tech_debt_hours: float
    clean_code_score: float = Field(ge=0, le=10)

class RefactoringSuggestion(BaseModel):
    title: str
    description: str
    impact: Literal["low", "medium", "high"]
    effort: Literal["low", "medium", "high"]
    suggested_code: Optional[str] = None

class RefactoringPlan(BaseModel):
    suggestions: list[RefactoringSuggestion]
    recommended_priority: list[str]  # Suggestion names, in order
    expected_benefit: str

Summary

  • Decomposition framework: Identify sub-tasks → define interfaces → design prompts → order by dependencies
  • Patterns: ETL for processing, Map-Reduce for collections, DAG for complex dependencies
  • Interfaces: Use Pydantic for interface schemas. It makes the pipeline robust and validatable.
  • Parallelization: Compute the topological order of the DAG and run the groups in parallel with asyncio
  • Well-designed sub-tasks: One clear responsibility, an explicit input/output schema, < 300 tokens of prompt

Additional resources

  1. Least-to-Most Prompting (Zhou et al., 2022)
  2. Pydantic v2 - Field validation
  3. Python asyncio - gather for parallel tasks
  4. LangChain Chains documentation
  5. OpenAI Cookbook - Structured outputs
  6. Directed Acyclic Graphs (topological sort)
← Previous2. Prompt Chaining: Output → InputNext →4. Multi-Stage Pipelines