6. Multi-Step Reasoning Pipelines

Overview

For complex problems, a single CoT prompt isn't enough. Multi-step pipelines break the reasoning into explicit stages: Understand → Plan → Execute → Verify. Each stage is a separate call, and the output of each stage feeds the next one.

This capsule covers: the 4-stage pattern, when to use a pipeline vs. single CoT, error handling between stages, cost optimization, and real use cases.

Estimated time: 90-120 minutes

Why Multi-Step Pipelines?

Single CoT asks the model to do everything in one call: understand the problem, plan the solution, execute it, and verify it. For simple problems, that works. For complex ones:

Problems with single CoT on complex problems:

1. The model can "forget" part of the problem by the time it reaches the end (an attention limitation)
2. Errors in the initial understanding propagate through every step
3. You can't inspect or intervene in the intermediate steps
4. Reasoning blended into one blob is hard to audit

Advantages of the pipeline:

1. Each stage has a clear, verifiable purpose
2. You can inspect, log and fix each stage
3. The context can be summarized between stages so nothing gets lost
4. Stages can run with different parameters (temperature, model, etc.)
5. You can cache the expensive stages

The 4-Stage Pipeline: UPEV

U - Understand: What is being asked? What data is there?
P - Plan: What steps need to be followed?
E - Execute: Run each step of the plan
V - Verify: Is the answer correct?

Full Implementation

from openai import OpenAI
from dataclasses import dataclass, field

client = OpenAI()


@dataclass
class PipelineStage:
    name: str
    output: str = ""
    tokens: int = 0
    error: str | None = None


@dataclass
class PipelineResult:
    problem: str
    stages: list[PipelineStage] = field(default_factory=list)
    final_answer: str = ""
    total_tokens: int = 0
    success: bool = False


def reasoning_pipeline(
    problem: str,
    verbose: bool = True
) -> PipelineResult:
    """
    A 4-stage pipeline: Understand → Plan → Execute → Verify
    
    Args:
        problem: The problem to solve
        verbose: If True, prints the output of each stage
    
    Returns:
        PipelineResult with every output and its metadata
    """
    result = PipelineResult(problem=problem)
    
    # ===== STAGE 1: UNDERSTAND =====
    prompt_understand = f"""Analyze the following problem and extract its structure.

Problem: {problem}

Reply in this exact format:
CORE QUESTION: [what exactly is being asked, in one sentence]
GIVEN DATA: [list of data/numbers/facts provided]
CONSTRAINTS: [limits or conditions the solution has to meet]
PROBLEM TYPE: [mathematical/logical/optimization/other]
ESTIMATED DIFFICULTY: [simple/moderate/complex]"""

    try:
        r_understand = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt_understand}],
            temperature=0,
            max_tokens=400
        )
        understand_stage = PipelineStage(
            name="understand",
            output=r_understand.choices[0].message.content,
            tokens=r_understand.usage.total_tokens
        )
        result.stages.append(understand_stage)
        
        if verbose:
            print(f"\n{'='*50}")
            print("STAGE 1: UNDERSTAND")
            print(understand_stage.output)
    
    except Exception as e:
        result.stages.append(PipelineStage(name="understand", error=str(e)))
        result.success = False
        return result
    
    # ===== STAGE 2: PLAN =====
    prompt_plan = f"""Given the analysis of the problem, propose a detailed plan to solve it.

Original problem: {problem}

Analysis:
{understand_stage.output}

Create a numbered plan with concrete, actionable steps.
Each step must be specific (not "calculate something", but "calculate X using Y").
Include which tool or technique to use at each step.

SOLUTION PLAN:"""

    try:
        r_plan = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt_plan}],
            temperature=0,
            max_tokens=500
        )
        plan_stage = PipelineStage(
            name="plan",
            output=r_plan.choices[0].message.content,
            tokens=r_plan.usage.total_tokens
        )
        result.stages.append(plan_stage)
        
        if verbose:
            print(f"\n{'='*50}")
            print("STAGE 2: PLAN")
            print(plan_stage.output)
    
    except Exception as e:
        result.stages.append(PipelineStage(name="plan", error=str(e)))
        result.success = False
        return result
    
    # ===== STAGE 3: EXECUTE =====
    prompt_execute = f"""Execute the plan step by step to solve the problem.

Problem: {problem}

Plan to execute:
{plan_stage.output}

Instructions:
- Execute EVERY step of the plan
- Show the work for each step (operations, reasoning)
- If a step produces an intermediate result, write it down clearly
- Don't skip steps

EXECUTION:"""

    try:
        r_execute = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt_execute}],
            temperature=0,
            max_tokens=900
        )
        execute_stage = PipelineStage(
            name="execute",
            output=r_execute.choices[0].message.content,
            tokens=r_execute.usage.total_tokens
        )
        result.stages.append(execute_stage)
        
        if verbose:
            print(f"\n{'='*50}")
            print("STAGE 3: EXECUTE")
            print(execute_stage.output)
    
    except Exception as e:
        result.stages.append(PipelineStage(name="execute", error=str(e)))
        result.success = False
        return result
    
    # ===== STAGE 4: VERIFY =====
    prompt_verify = f"""Verify the solution you were given.

Original problem: {problem}

Solution obtained:
{execute_stage.output}

Verify:
1. Does the solution answer exactly what was asked?
2. Are the calculations correct? (check the key steps)
3. Does the answer make sense in the context of the problem?
4. If there are errors, fix them.

VERIFICATION:
Status: CORRECT / INCORRECT / PARTIALLY CORRECT
Verification notes: [details]
FINAL ANSWER: [the corrected or confirmed answer, the result only]"""

    try:
        r_verify = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt_verify}],
            temperature=0,
            max_tokens=400
        )
        verify_stage = PipelineStage(
            name="verify",
            output=r_verify.choices[0].message.content,
            tokens=r_verify.usage.total_tokens
        )
        result.stages.append(verify_stage)
        
        if verbose:
            print(f"\n{'='*50}")
            print("STAGE 4: VERIFY")
            print(verify_stage.output)
        
        # Extract the final answer
        import re
        match = re.search(
            r'FINAL ANSWER:\s*(.+?)(?:\n|$)', 
            verify_stage.output, 
            re.IGNORECASE
        )
        result.final_answer = match.group(1).strip() if match else verify_stage.output[-200:]
        result.success = True
    
    except Exception as e:
        result.stages.append(PipelineStage(name="verify", error=str(e)))
        result.success = False
    
    # Compute the total tokens
    result.total_tokens = sum(e.tokens for e in result.stages)
    
    if verbose:
        print(f"\n{'='*50}")
        print(f"FINAL ANSWER: {result.final_answer}")
        print(f"Total tokens: {result.total_tokens}")
    
    return result


# Usage example
if __name__ == "__main__":
    complex_problem = """
    A logistics company has 3 available routes:
    - Route A: 200 km, $0.50/km, estimated time 3h
    - Route B: 150 km, $0.70/km, estimated time 2h
    - Route C: 250 km, $0.40/km, estimated time 4h
    
    The customer wants to minimize cost, but the maximum time is 3.5 hours.
    Routes B and C are available, but A has an extra $20 toll charge.
    Which is the optimal route and how much will it cost?
    """
    
    result = reasoning_pipeline(complex_problem)
    print(f"\n\nSUMMARY: {result.final_answer}")

When to Use a Pipeline vs. Single CoT

Decision Criteria

Decision Tree

def decide_strategy(problem: str) -> str:
    """
    A simple heuristic to decide which strategy to use.
    In production, you could use the LLM for this classification.
    """
    pipeline_indicators = [
        len(problem) > 500,  # Long problem
        problem.count('\n') > 5,  # Multiple lines of data
        any(w in problem.lower() for w in ['first', 'then', 'after that', 'finally', 'steps']),
        any(w in problem.lower() for w in ['optimiz', 'compare', 'analyze and', 'design']),
    ]
    
    if sum(pipeline_indicators) >= 2:
        return "pipeline"
    else:
        return "single_cot"


# Test cases
examples = [
    "What is 17 × 23?",
    "Analyze this 50-page legal contract, identify problematic clauses, propose improvements and generate an executive summary",
    "Is this logical argument valid: If P then Q, P, therefore Q?",
    "Design a recommendation system for an e-commerce site: first analyze the requirements, then propose the technical architecture, after that detail the data model, and finally describe how to measure success",
]

for ex in examples:
    strategy = decide_strategy(ex)
    print(f"{'Pipeline' if strategy == 'pipeline' else 'Single CoT':12} | {ex[:70]}...")

A Pipeline with Error Handling and Recovery

from typing import Callable


def robust_pipeline(
    problem: str,
    max_retries: int = 2
) -> PipelineResult:
    """
    A pipeline with error handling and per-stage retries.
    If a stage fails, it tries to recover before failing globally.
    """
    
    stages_config = [
        {
            "name": "understand",
            "prompt_fn": lambda p, prev: f"Analyze the problem:\n{p}\nExtract: core question, data, constraints.",
            "max_tokens": 400,
            "temperature": 0
        },
        {
            "name": "plan",
            "prompt_fn": lambda p, prev: f"Problem: {p}\nAnalysis: {prev}\nCreate a step-by-step solution plan.",
            "max_tokens": 500,
            "temperature": 0
        },
        {
            "name": "execute",
            "prompt_fn": lambda p, prev: f"Execute this plan to solve: {p}\nPlan:\n{prev}",
            "max_tokens": 900,
            "temperature": 0
        },
        {
            "name": "verify",
            "prompt_fn": lambda p, prev: f"Verify this solution. Problem: {p}\nSolution: {prev}\nFINAL ANSWER:",
            "max_tokens": 400,
            "temperature": 0
        }
    ]
    
    result = PipelineResult(problem=problem)
    previous_output = ""
    
    for config in stages_config:
        name = config["name"]
        stage_success = False
        
        for attempt in range(max_retries + 1):
            try:
                prompt = config["prompt_fn"](problem, previous_output)
                
                # On retries, add context about the failure
                if attempt > 0:
                    prompt += f"\n\n[Attempt {attempt + 1}: the previous attempt didn't produce a complete result. Please be more detailed.]"
                
                response = client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[{"role": "user", "content": prompt}],
                    temperature=config["temperature"],
                    max_tokens=config["max_tokens"]
                )
                
                stage = PipelineStage(
                    name=name,
                    output=response.choices[0].message.content,
                    tokens=response.usage.total_tokens
                )
                result.stages.append(stage)
                previous_output = stage.output
                stage_success = True
                break
                
            except Exception as e:
                if attempt == max_retries:
                    stage = PipelineStage(name=name, error=str(e))
                    result.stages.append(stage)
                    # Carry on with what we have (graceful degradation)
                    previous_output = f"[Error in stage {name}: {str(e)[:100]}]"
        
        if not stage_success and name in ["execute"]:
            # The critical stages have to succeed to continue
            result.success = False
            result.total_tokens = sum(e.tokens for e in result.stages)
            return result
    
    # Extract the final answer
    import re
    if result.stages:
        last_stage = result.stages[-1]
        match = re.search(r'FINAL ANSWER:\s*(.+?)(?:\n|$)', last_stage.output, re.IGNORECASE)
        result.final_answer = match.group(1).strip() if match else last_stage.output[-200:]
    
    result.success = True
    result.total_tokens = sum(e.tokens for e in result.stages)
    return result

A Pipeline with Context Summarization

For very long problems, the context between stages can grow out of hand. This implementation summarizes the output before passing it to the next stage:

def summarize_stage(long_output: str, max_chars: int = 500) -> str:
    """Condenses one stage's output for the next one."""
    if len(long_output) <= max_chars:
        return long_output
    
    summary_prompt = f"""Summarize the following text keeping ONLY the key points needed for the next stage of solving the problem. Maximum {max_chars} characters.

Text to summarize:
{long_output}

Summary:"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": summary_prompt}],
        temperature=0,
        max_tokens=200
    )
    return response.choices[0].message.content[:max_chars]


def pipeline_with_summarized_context(problem: str) -> PipelineResult:
    """
    A pipeline that summarizes the context between stages to avoid token overflow.
    Useful for problems whose intermediate stages produce very long outputs.
    """
    result = reasoning_pipeline(problem, verbose=False)
    return result

Domain-Specialized Pipelines

Pipeline: Complex Code Analysis

def code_review_pipeline(code: str, language: str = "Python") -> dict:
    """
    A pipeline specialized in exhaustive code analysis.
    4 stages: Understand → Identify problems → Suggest improvements → Generate improved code
    """
    
    # Stage 1: Understand the code
    r1 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Analyze the following {language} code:
```{language.lower()}
{code}

What does this code do? Describe its purpose, inputs, outputs and structure in 5-7 sentences. """}], temperature=0, max_tokens=300 ) description = r1.choices[0].message.content

# Stage 2: Identify the problems
r2 = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": f"""

Code:

{code}

Description: {description}

Identify EVERY problem (bugs, edge cases, performance, security, readability). For each problem: type, line, description, severity (High/Medium/Low). """}], temperature=0, max_tokens=600 ) problems = r2.choices[0].message.content

# Stage 3: Suggest improvements
r3 = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": f"""

Code:

{code}

Problems identified: {problems}

For each High or Medium severity problem, propose the specific fix. Format: "Problem X → Fix: [corrected code]" """}], temperature=0, max_tokens=700 ) improvements = r3.choices[0].message.content

# Stage 4: Generate the improved code
r4 = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": f"""

Original code:

{code}

Improvements to apply: {improvements}

Generate the improved code applying ALL of the High and Medium severity fixes. Code only, no extra explanations.

        """}],
        temperature=0, max_tokens=800
    )
    improved_code = r4.choices[0].message.content
    
    return {
        "description": description,
        "problems": problems,
        "suggested_improvements": improvements,
        "improved_code": improved_code,
        "total_tokens": sum([
            r1.usage.total_tokens, r2.usage.total_tokens,
            r3.usage.total_tokens, r4.usage.total_tokens
        ])
    }

Pipeline: Research and Synthesis

def research_pipeline(question: str, available_context: str) -> dict:
    """
    A pipeline for answering complex questions that need deep analysis.
    Useful in RAG systems where there are multiple context documents.
    """
    
    # Stage 1: Break down the question
    r1 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Break this complex question down into simpler sub-questions:
Question: {question}

List 3-5 sub-questions that, once answered, answer the main question.
        """}],
        temperature=0, max_tokens=300
    )
    subquestions = r1.choices[0].message.content
    
    # Stage 2: Find the evidence in the context
    r2 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
For each sub-question, find the relevant evidence in the given context.

Sub-questions:
{subquestions}

Available context:
{available_context}

For each sub-question, quote the relevant part of the context or write "No evidence".
        """}],
        temperature=0, max_tokens=700
    )
    evidence = r2.choices[0].message.content
    
    # Stage 3: Synthesize the answer
    r3 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Synthesize the information to answer the main question.

Original question: {question}

Evidence found: {evidence}

Generate a coherent, well-structured answer that:
1. Answers the question directly
2. Is grounded in the evidence
3. Clearly flags any place where the information is insufficient
        """}],
        temperature=0, max_tokens=600
    )
    synthesis = r3.choices[0].message.content
    
    # Stage 4: Validate and attach a confidence level
    r4 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Evaluate the quality of this answer:

Question: {question}
Answer: {synthesis}
Evidence used: {evidence}

Evaluate:
1. Is the answer fully backed by the evidence?
2. Are there claims without evidence?
3. Overall confidence: High/Medium/Low + the reason

VALIDATED ANSWER: [the answer, modified if needed]
CONFIDENCE: [level]
        """}],
        temperature=0, max_tokens=400
    )
    validation = r4.choices[0].message.content
    
    return {
        "question": question,
        "subquestions": subquestions,
        "evidence": evidence,
        "synthesis": synthesis,
        "validation": validation
    }

Cost Optimization in Pipelines

from functools import lru_cache
import hashlib


def hash_text(text: str) -> str:
    """Generates a hash of the text, for caching."""
    return hashlib.md5(text.encode()).hexdigest()


class CachedPipeline:
    """
    A pipeline that caches intermediate results so it doesn't reprocess
    the same problem or identical parts of it.
    """
    
    def __init__(self):
        self._cache: dict[str, str] = {}
    
    def _call_with_cache(self, prompt: str, **kwargs) -> tuple[str, bool]:
        """
        Makes an LLM call, using the cache if the prompt was already processed.
        
        Returns:
            tuple of (answer, from_cache)
        """
        cache_key = hash_text(prompt)
        
        if cache_key in self._cache:
            return self._cache[cache_key], True
        
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            **kwargs
        )
        answer = response.choices[0].message.content
        self._cache[cache_key] = answer
        return answer, False
    
    def run_cached_pipeline(self, problem: str) -> dict:
        """The pipeline, with stage caching."""
        
        prompt_understand = f"Analyze: {problem}\nExtract the question, data, constraints."
        understanding, from_cache = self._call_with_cache(
            prompt_understand, temperature=0, max_tokens=400
        )
        
        prompt_plan = f"Problem: {problem}\nAnalysis: {understanding}\nCreate a solution plan."
        plan, _ = self._call_with_cache(prompt_plan, temperature=0, max_tokens=500)
        
        prompt_execute = f"Execute this plan:\nProblem: {problem}\nPlan: {plan}"
        execution, _ = self._call_with_cache(prompt_execute, temperature=0, max_tokens=900)
        
        prompt_verify = f"Verify. Problem: {problem}\nSolution: {execution}\nFINAL ANSWER:"
        verification, _ = self._call_with_cache(prompt_verify, temperature=0, max_tokens=400)
        
        return {
            "understanding": understanding,
            "plan": plan,
            "execution": execution,
            "verification": verification,
            "understand_from_cache": from_cache
        }

Measuring Performance: Pipeline vs. Single CoT

import time
from statistics import mean


def benchmark_approaches(
    problems: list[str],
    n_runs: int = 3
) -> dict:
    """
    Compares the performance and accuracy of single CoT vs. the pipeline.
    """
    
    single_times = []
    pipeline_times = []
    single_tokens = []
    pipeline_tokens = []
    
    for problem in problems[:n_runs]:
        # Single CoT
        t_start = time.time()
        r_single = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"{problem}\n\nLet's think step by step."}],
            temperature=0, max_tokens=700
        )
        single_times.append(time.time() - t_start)
        single_tokens.append(r_single.usage.total_tokens)
        
        # Pipeline
        t_start = time.time()
        r_pipeline = reasoning_pipeline(problem, verbose=False)
        pipeline_times.append(time.time() - t_start)
        pipeline_tokens.append(r_pipeline.total_tokens)
    
    return {
        "single_cot": {
            "avg_time_s": mean(single_times),
            "avg_tokens": mean(single_tokens),
            "relative_cost": 1.0
        },
        "pipeline": {
            "avg_time_s": mean(pipeline_times),
            "avg_tokens": mean(pipeline_tokens),
            "relative_cost": mean(pipeline_tokens) / mean(single_tokens)
        }
    }

Table: When to Use Each Approach

Troubleshooting

Problem 1: An error in one intermediate stage contaminates the following ones

# ❌ With no error handling, a bad "plan" output produces an incorrect execution
# ✅ Implement per-stage validation

def validate_plan_stage(plan_output: str) -> bool:
    """Checks that the plan has the expected structure."""
    # A valid plan must have at least 3 numbered steps
    import re
    steps = re.findall(r'^\d+[\.\)]\s', plan_output, re.MULTILINE)
    return len(steps) >= 2


def pipeline_with_validation(problem: str) -> PipelineResult:
    """A pipeline that validates each stage before continuing."""
    result = reasoning_pipeline(problem, verbose=False)
    
    # Check whether the plan was generated correctly
    plan_stage = next((e for e in result.stages if e.name == "plan"), None)
    if plan_stage and not validate_plan_stage(plan_stage.output):
        # Regenerate the plan with more explicit instructions
        print("Invalid plan, regenerating...")
        # ... retry logic
    
    return result

Problem 2: Token buildup between stages

# ❌ Passing the whole previous output makes the prompts longer and longer
# ✅ Summarize the previous outputs

CONTEXT_CHAR_LIMIT = 800  # A reasonable limit per previous stage

def pass_summarized_context(stage_output: str, max_chars: int = CONTEXT_CHAR_LIMIT) -> str:
    """Prepares one stage's output to be passed to the next."""
    if len(stage_output) <= max_chars:
        return stage_output
    
    # Take the first and last parts (usually the most important)
    half = max_chars // 2
    return stage_output[:half] + "\n...[summary]...\n" + stage_output[-half:]

Problem 3: High latency in sequential pipelines

# ❌ A sequential pipeline: 4 calls in series = high latency
# ✅ Parallelize the independent stages whenever possible

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def partially_parallel_pipeline(problem: str) -> dict:
    """
    Some stages can run in parallel.
    For example: "understand" for different sub-parts of the problem.
    """
    # Stage 1: Understand (sequential, a prerequisite)
    r_understand = await async_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Analyze: {problem}"}],
        temperature=0, max_tokens=400
    )
    understanding = r_understand.choices[0].message.content
    
    # Stages 2a and 2b: Run in parallel if the problem has sub-parts
    r_plan, r_check_constraints = await asyncio.gather(
        async_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"Plan for: {problem}\n{understanding}"}],
            temperature=0, max_tokens=400
        ),
        async_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"List the constraints of: {problem}"}],
            temperature=0, max_tokens=200
        )
    )
    
    return {
        "understanding": understanding,
        "plan": r_plan.choices[0].message.content,
        "constraints": r_check_constraints.choices[0].message.content
    }


# Usage
async def main():
    result = await partially_parallel_pipeline("Your problem here")
    print(result)

# asyncio.run(main())

Exercises

Exercise 1: Build a pipeline for review analysis

Design a 3-stage pipeline to analyze product reviews: (1) Extract the aspects mentioned, (2) Classify sentiment per aspect, (3) Generate an executive summary.

from openai import OpenAI

client = OpenAI()

def review_analysis_pipeline(review: str, product: str = "") -> dict:
    """A pipeline for analyzing product reviews."""
    
    # Stage 1: Extract the aspects
    r1 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Analyze this review and extract every aspect of the product it mentions.
Product: {product or "unknown"}
Review: {review}

List the aspects mentioned (e.g. quality, price, shipping, customer service, etc.)
        """}],
        temperature=0, max_tokens=300
    )
    aspects = r1.choices[0].message.content
    
    # Stage 2: Sentiment per aspect
    r2 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
For each aspect identified, classify the sentiment expressed.

Review: {review}
Aspects identified: {aspects}

Format: Aspect: [name] | Sentiment: POSITIVE/NEGATIVE/NEUTRAL | Evidence: "[quote]"
        """}],
        temperature=0, max_tokens=500
    )
    sentiments = r2.choices[0].message.content
    
    # Stage 3: Executive summary
    r3 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Generate a 3-4 line executive summary for a product manager.

Review analyzed: {review}
Sentiment analysis: {sentiments}

The summary should include: overall sentiment, strongest points, weakest points, 
and one actionable recommendation.
        """}],
        temperature=0, max_tokens=300
    )
    
    return {
        "aspects": aspects,
        "sentiments_by_aspect": sentiments,
        "executive_summary": r3.choices[0].message.content
    }


# Test
example_review = """
I bought this vacuum 3 months ago. The suction is incredible, better than the one I had before.
That said, the cord is far too short for large rooms and I had to buy an 
extension. Shipping was lightning fast (it arrived the next day) but the box turned up dented.
Customer service, when I called, was very friendly and answered my questions. The price seems 
fair for the quality it delivers.
"""

result = review_analysis_pipeline(example_review, "Vacuum X200")
for stage, content in result.items():
    print(f"\n=== {stage.upper()} ===")
    print(content)

Exercise 2: A pipeline for research with simplified RAG

Design a pipeline that (1) breaks a question into sub-questions, (2) "searches" a local corpus, (3) synthesizes the answer.

EXAMPLE_CORPUS = {
    "python": "Python is an interpreted, high-level language with dynamic typing. Created by Guido van Rossum in 1991. Heavily used in data science, ML, and backend.",
    "fastapi": "FastAPI is a modern web framework for Python 3.8+. Built on Starlette and Pydantic. Generates OpenAPI documentation automatically. Very high performance.",
    "openai": "OpenAI was founded in 2015. It develops GPT-4, DALL-E, and Whisper. The OpenAI API gives access to language models over HTTP.",
}

def search_corpus(query: str, corpus: dict) -> str:
    """A simple keyword search over the corpus."""
    results = []
    query_lower = query.lower()
    for key, content in corpus.items():
        if key in query_lower or any(w in content.lower() for w in query_lower.split()):
            results.append(f"[{key}]: {content}")
    return "\n".join(results) if results else "No relevant results"

def simple_research_pipeline(question: str, corpus: dict) -> dict:
    # Stage 1: Break down the question
    r1 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Break into 2-3 sub-questions: {question}"}],
        temperature=0, max_tokens=200
    )
    subquestions = r1.choices[0].message.content
    
    # Stage 2: Retrieve the context
    context = search_corpus(question, corpus)
    
    # Stage 3: Synthesize
    r3 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"""
Question: {question}
Sub-questions: {subquestions}
Available context: {context}

Answer using ONLY the information in the context.
        """}],
        temperature=0, max_tokens=400
    )
    
    return {
        "question": question,
        "subquestions": subquestions,
        "context_used": context,
        "answer": r3.choices[0].message.content
    }

result = simple_research_pipeline("What advantages does FastAPI have over other Python frameworks?", EXAMPLE_CORPUS)
print(result["answer"])

Exercise 3: Measure the pipeline's overhead vs. single CoT

Design an experiment to measure how many extra tokens and how much extra time the pipeline needs compared to single CoT, across 5 identical problems.

import time
from openai import OpenAI

client = OpenAI()

BENCHMARK_PROBLEMS = [
    "If I have 3 pizzas of 8 slices each and I invite 10 people, how many slices does each person get?",
    "A bank offers 5% interest a year. If I invest $1,000 for 3 years with compound interest, how much do I have at the end?",
    "In an 8-team single-elimination tournament, how many matches are there in total?",
    "If a train leaves at 9:00 going 80 km/h, and another leaves at 9:30 on the same route at 120 km/h, when does it catch the first?",
    "How many integers from 1 to 100 are divisible by 3 or by 5?",
]

results = []

for i, problem in enumerate(BENCHMARK_PROBLEMS, 1):
    print(f"\nProblem {i}: {problem[:50]}...")
    
    # Single CoT
    t0 = time.time()
    r_single = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"{problem}\n\nLet's think step by step."}],
        temperature=0, max_tokens=600
    )
    single_time = time.time() - t0
    single_tokens = r_single.usage.total_tokens
    
    # Pipeline
    t0 = time.time()
    r_pipeline = reasoning_pipeline(problem, verbose=False)
    pipeline_time = time.time() - t0
    pipeline_tokens = r_pipeline.total_tokens
    
    results.append({
        "problem": i,
        "single_tokens": single_tokens,
        "pipeline_tokens": pipeline_tokens,
        "tokens_multiplier": pipeline_tokens / single_tokens,
        "single_time_s": single_time,
        "pipeline_time_s": pipeline_time,
        "time_multiplier": pipeline_time / single_time
    })

print("\n=== BENCHMARK SUMMARY ===")
avg_tokens = sum(r["tokens_multiplier"] for r in results) / len(results)
avg_time = sum(r["time_multiplier"] for r in results) / len(results)
print(f"Average tokens (Pipeline vs Single): {avg_tokens:.1f}x more")
print(f"Average time (Pipeline vs Single): {avg_time:.1f}x more")

Summary

• The UPEV pipeline: Understand → Plan → Execute → Verify; each stage is a separate call
• When to use it: Complex problems with multiple sub-tasks, a need for inspection/auditing
• Error handling: Validate each stage, have retry strategies and graceful degradation
• Context: Summarize long outputs before passing them to the next stage
• Cost: 3-5x more tokens than single CoT; justified for critical problems
• Parallelism: Independent stages can run in parallel with asyncio
• Caching: Cache the expensive stages that repeat across calls

Additional resources

1. ReAct: Synergizing Reasoning and Acting (Yao et al., 2022)
2. Least-to-Most Prompting (Zhou et al., 2022)
3. Decomposed Prompting (Khot et al., 2022)
4. LangChain Chains Documentation
5. OpenAI Assistants API - Multi-step tasks
6. Async OpenAI Python SDK