Skip to content

Session 3: From Linear Scripts to Orchestrated Workflows

Synopsis

Introduces workflow patterns such as routing, branching, iterative refinement, and human-in-the-loop checkpoints. Learners begin designing systems as explicit processes instead of opaque model calls.

Session Content

Session 3: From Linear Scripts to Orchestrated Workflows

Session Overview

In the first two sessions, learners typically move from basic prompting and API usage toward structured generation and practical Python integration. This session builds on that foundation by introducing a major shift in how GenAI applications are designed:

  • from single-call scripts
  • to multi-step, orchestrated workflows

By the end of this session, learners will understand how to break a task into stages, coordinate LLM calls, pass structured outputs between steps, and build simple workflow controllers in Python using the OpenAI Responses API and gpt-5.4-mini.

Learning Objectives

By the end of this 45-minute session, learners should be able to:

  1. Explain the difference between a linear script and an orchestrated workflow.
  2. Identify when a task should be split into multiple LLM-assisted steps.
  3. Implement a multi-step workflow in Python using the OpenAI Responses API.
  4. Pass outputs safely between workflow stages.
  5. Add basic validation, retries, and logging to an LLM-driven workflow.
  6. Build a practical mini-pipeline that performs planning, generation, and review.

Agenda

  • 0–10 min: Theory — linear scripts vs orchestrated workflows
  • 10–20 min: Theory — workflow patterns and design principles
  • 20–35 min: Hands-on Exercise 1 — build a 3-step content workflow
  • 35–43 min: Hands-on Exercise 2 — add validation and retry logic
  • 43–45 min: Recap and takeaways

1. Why Move Beyond Linear Scripts?

A linear script usually looks like this:

  1. Accept input
  2. Send one prompt to the model
  3. Print the result

This is a great starting point, but it has limitations:

  • prompts become overly large and complicated
  • outputs are harder to control
  • failures are harder to recover from
  • reasoning and transformation steps are mixed together
  • debugging is difficult because everything happens in one call

Example of a Linear Script Mentality

You might ask a model:

“Read this customer complaint, summarize it, categorize it, draft a response, and identify escalation risk.”

This can work, but it often produces inconsistent or hard-to-parse output.

Why Orchestrated Workflows Help

An orchestrated workflow breaks the problem into stages:

  1. Summarize the complaint
  2. Classify the issue
  3. Draft a response
  4. Review the response for tone and policy compliance

This provides:

  • clearer prompts
  • better modularity
  • easier testing
  • reusable components
  • improved reliability
  • the ability to add validation between steps

2. Core Concepts of Workflow Orchestration

2.1 Step Decomposition

Break a complex task into smaller units where each step has:

  • a clear input
  • a clear purpose
  • a clear output

Good workflow steps are:

  • narrow in scope
  • easy to test
  • easy to retry independently

2.2 State Passing

Workflows usually maintain a shared state object such as a Python dictionary.

Example:

state = {
    "topic": "Benefits of retrieval-augmented generation",
    "outline": None,
    "draft": None,
    "review": None,
}

Each step updates part of the state.

2.3 Deterministic Code + Probabilistic Model

A key pattern in GenAI systems:

  • Python code controls sequence, validation, branching, retries, and logging
  • LLM calls perform language-heavy tasks such as planning, transforming, summarizing, and reviewing

The orchestration layer should remain deterministic even if the model output is probabilistic.

2.4 Structured Outputs

Whenever possible, ask the model for structured JSON-like outputs. This makes downstream automation easier.

For example:

  • planner step returns a list of sections
  • reviewer step returns {"approved": true, "issues": []}

2.5 Validation Gates

Do not blindly trust step outputs. Add checks such as:

  • required keys present
  • list lengths within limits
  • text not empty
  • category is from an allowed set

2.6 Retryable Failures

Some steps can fail transiently or produce malformed outputs. You should design workflows so that:

  • a failed step can be retried
  • earlier successful steps do not need to be rerun
  • logs capture what happened

3. Common Workflow Patterns

Pattern A: Chain

Output of step A feeds into step B, then step C.

Example: summarize → classify → draft

Pattern B: Fan-Out / Fan-In

One input is split into several parallel tasks, whose outputs are later combined.

Example: generate 3 headline options, then choose the best one

Pattern C: Critique-and-Revise

One model call generates content, another reviews it, and optionally a third revises it.

Example: draft email → quality review → improved email

Pattern D: Router

A classifier step decides which downstream path to follow.

Example: customer message routed to billing, technical support, or refund handling

4. Designing a Good Workflow

When designing a workflow, ask:

4.1 What belongs in code?

Use Python for:

  • branching logic
  • allowed-value checks
  • retry loops
  • timestamps and IDs
  • API error handling
  • saving intermediate outputs

4.2 What belongs in the model?

Use the LLM for:

  • summaries
  • rewriting
  • extraction from messy text
  • planning
  • quality critique
  • natural-language decisions

4.3 When should a task be multi-step?

Split a task if:

  • one prompt is doing too many things
  • outputs need machine-readable structure
  • some steps should be reviewed independently
  • different prompts/personas are useful
  • you want more control or observability

5. Workflow Example Architecture

We will build a simple content creation workflow:

  1. Planner
    Input: topic
    Output: short outline in JSON

  2. Writer
    Input: topic + outline
    Output: draft article

  3. Reviewer
    Input: draft
    Output: review JSON with approval and suggestions

Optionally:

  1. Reviser
    Input: draft + review suggestions
    Output: improved draft

6. Hands-On Exercise 1: Build a 3-Step Workflow

Goal

Create a Python workflow that:

  • accepts a topic
  • generates an outline
  • writes a draft
  • reviews the draft

What You Will Practice

  • using the OpenAI Responses API
  • creating reusable helper functions
  • passing state between steps
  • parsing structured outputs
  • organizing multi-step LLM logic

6.1 Setup

Install the OpenAI Python SDK:

pip install openai

Set your API key:

export OPENAI_API_KEY="your_api_key_here"

On Windows PowerShell:

$env:OPENAI_API_KEY="your_api_key_here"

6.2 Full Example Code

"""
Session 3 - Exercise 1
Build a 3-step orchestrated workflow using the OpenAI Responses API.

Workflow:
1. Plan an outline
2. Write a draft
3. Review the draft

Model:
- gpt-5.4-mini

Requirements:
- pip install openai
- export OPENAI_API_KEY=...

This example demonstrates:
- modular workflow steps
- JSON parsing
- shared workflow state
- clean, well-commented helper functions
"""

from __future__ import annotations

import json
import os
from typing import Any, Dict, List

from openai import OpenAI


# Create the OpenAI client once and reuse it.
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))


def call_model(prompt: str) -> str:
    """
    Send a prompt to the OpenAI Responses API and return plain text output.

    We use gpt-5.4-mini as requested.
    """
    response = client.responses.create(
        model="gpt-5.4-mini",
        input=prompt,
    )

    # output_text is the easiest way to retrieve the model's text response.
    return response.output_text.strip()


def plan_outline(topic: str) -> Dict[str, Any]:
    """
    Ask the model to return a small article plan as JSON.

    Expected schema:
    {
      "title": "...",
      "sections": ["...", "...", "..."]
    }
    """
    prompt = f"""
You are a helpful planning assistant.

Create a concise article outline for the topic below.

Topic: {topic}

Return ONLY valid JSON with this schema:
{{
  "title": "short article title",
  "sections": ["section 1", "section 2", "section 3"]
}}

Rules:
- 3 to 5 sections
- section names must be short
- no markdown
- no explanation outside JSON
""".strip()

    raw_text = call_model(prompt)

    try:
        data = json.loads(raw_text)
    except json.JSONDecodeError as exc:
        raise ValueError(f"Planner returned invalid JSON: {raw_text}") from exc

    if "title" not in data or "sections" not in data:
        raise ValueError(f"Planner output missing required keys: {data}")

    if not isinstance(data["sections"], list) or not data["sections"]:
        raise ValueError(f"'sections' must be a non-empty list: {data}")

    return data


def write_draft(topic: str, outline: Dict[str, Any]) -> str:
    """
    Generate a short draft article based on the topic and outline.
    """
    prompt = f"""
You are a technical writer for Python developers who are learning GenAI.

Write a short educational article.

Topic:
{topic}

Use this outline:
Title: {outline["title"]}
Sections: {json.dumps(outline["sections"])}

Requirements:
- 250 to 400 words
- clear and beginner-friendly
- explain concepts simply
- use plain markdown headings
- include a short conclusion
""".strip()

    return call_model(prompt)


def review_draft(draft: str) -> Dict[str, Any]:
    """
    Ask the model to review the draft and return structured JSON.
    """
    prompt = f"""
You are a strict content reviewer.

Review the article draft below.

Draft:
{draft}

Return ONLY valid JSON with this schema:
{{
  "approved": true,
  "issues": ["issue 1", "issue 2"],
  "suggestions": ["suggestion 1", "suggestion 2"]
}}

Rules:
- approved must be true or false
- issues and suggestions must be arrays
- if the draft is good, issues can be empty
- no explanation outside JSON
""".strip()

    raw_text = call_model(prompt)

    try:
        data = json.loads(raw_text)
    except json.JSONDecodeError as exc:
        raise ValueError(f"Reviewer returned invalid JSON: {raw_text}") from exc

    if not all(key in data for key in ("approved", "issues", "suggestions")):
        raise ValueError(f"Reviewer output missing required keys: {data}")

    return data


def run_workflow(topic: str) -> Dict[str, Any]:
    """
    Orchestrate the full 3-step workflow with shared state.
    """
    state: Dict[str, Any] = {
        "topic": topic,
        "outline": None,
        "draft": None,
        "review": None,
    }

    print("Step 1/3: Planning outline...")
    state["outline"] = plan_outline(state["topic"])

    print("Step 2/3: Writing draft...")
    state["draft"] = write_draft(state["topic"], state["outline"])

    print("Step 3/3: Reviewing draft...")
    state["review"] = review_draft(state["draft"])

    return state


def main() -> None:
    """
    Run the workflow for a sample topic and print results.
    """
    topic = "Why multi-step workflows are more reliable than one-shot prompts"

    state = run_workflow(topic)

    print("\n=== OUTLINE ===")
    print(json.dumps(state["outline"], indent=2))

    print("\n=== DRAFT ===")
    print(state["draft"])

    print("\n=== REVIEW ===")
    print(json.dumps(state["review"], indent=2))


if __name__ == "__main__":
    main()

6.3 Example Output

Step 1/3: Planning outline...
Step 2/3: Writing draft...
Step 3/3: Reviewing draft...

=== OUTLINE ===
{
  "title": "Why Multi-Step Workflows Improve Reliability",
  "sections": [
    "The Limits of One-Shot Prompts",
    "Breaking Tasks into Stages",
    "Validation Between Steps",
    "Easier Debugging and Iteration"
  ]
}

=== DRAFT ===
# Why Multi-Step Workflows Improve Reliability

## The Limits of One-Shot Prompts
A one-shot prompt asks a model to do many things at once...

## Breaking Tasks into Stages
By splitting a task into planning, drafting, and reviewing...

## Validation Between Steps
When each step produces a smaller output, it becomes easier...

## Easier Debugging and Iteration
Developers can inspect the outline before generating the draft...

## Conclusion
Multi-step workflows give developers more control and more reliable outcomes.

=== REVIEW ===
{
  "approved": true,
  "issues": [],
  "suggestions": [
    "Consider adding one concrete real-world example.",
    "Mention that workflow steps can be retried independently."
  ]
}

6.4 Discussion Questions

  1. Which step would be easiest to unit test?
  2. Where would you add retry logic?
  3. What if the reviewer rejects the draft?
  4. Which parts of this system are deterministic, and which are probabilistic?

7. Hands-On Exercise 2: Add Validation and Retry Logic

Goal

Improve the workflow so that:

  • invalid planner/reviewer JSON triggers a retry
  • the review step can reject the draft
  • the draft can be revised once if needed
  • execution logs show what happened

This turns a simple chain into a more realistic orchestrated workflow.

7.1 Design Improvements

We will add:

  • a generic retry wrapper
  • validation for planner output
  • a revise step if review says approved = false
  • lightweight workflow logging

7.2 Full Example Code

"""
Session 3 - Exercise 2
Add validation, retry logic, and conditional revision to an orchestrated workflow.

Workflow:
1. Plan
2. Write
3. Review
4. Optionally revise if not approved

This example demonstrates:
- retries for fragile structured outputs
- explicit validation
- conditional branching
- simple logging
"""

from __future__ import annotations

import json
import os
import time
from typing import Any, Callable, Dict

from openai import OpenAI


client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))


def log(message: str) -> None:
    """
    Print a timestamped log message.
    """
    timestamp = time.strftime("%H:%M:%S")
    print(f"[{timestamp}] {message}")


def call_model(prompt: str) -> str:
    """
    Call the Responses API and return plain text.
    """
    response = client.responses.create(
        model="gpt-5.4-mini",
        input=prompt,
    )
    return response.output_text.strip()


def with_retries(func: Callable[..., Any], *args: Any, retries: int = 2, **kwargs: Any) -> Any:
    """
    Run a function with retries.

    Parameters:
    - func: the function to call
    - retries: number of retry attempts after the first try

    Example:
        result = with_retries(plan_outline, "my topic", retries=2)
    """
    attempts = retries + 1

    for attempt in range(1, attempts + 1):
        try:
            return func(*args, **kwargs)
        except Exception as exc:
            log(f"Attempt {attempt}/{attempts} failed in {func.__name__}: {exc}")
            if attempt == attempts:
                raise
            time.sleep(1)

    raise RuntimeError("Unexpected retry control flow error.")


def validate_outline(data: Dict[str, Any]) -> None:
    """
    Validate planner output.
    """
    if not isinstance(data, dict):
        raise ValueError("Outline must be a dictionary.")

    if "title" not in data or "sections" not in data:
        raise ValueError("Outline missing required keys.")

    if not isinstance(data["title"], str) or not data["title"].strip():
        raise ValueError("Outline title must be a non-empty string.")

    if not isinstance(data["sections"], list):
        raise ValueError("Outline sections must be a list.")

    if not 3 <= len(data["sections"]) <= 5:
        raise ValueError("Outline sections must contain 3 to 5 items.")

    for section in data["sections"]:
        if not isinstance(section, str) or not section.strip():
            raise ValueError("Each section must be a non-empty string.")


def validate_review(data: Dict[str, Any]) -> None:
    """
    Validate reviewer output.
    """
    if not isinstance(data, dict):
        raise ValueError("Review must be a dictionary.")

    required_keys = {"approved", "issues", "suggestions"}
    if not required_keys.issubset(data.keys()):
        raise ValueError("Review missing required keys.")

    if not isinstance(data["approved"], bool):
        raise ValueError("'approved' must be a boolean.")

    if not isinstance(data["issues"], list):
        raise ValueError("'issues' must be a list.")

    if not isinstance(data["suggestions"], list):
        raise ValueError("'suggestions' must be a list.")


def plan_outline(topic: str) -> Dict[str, Any]:
    """
    Generate an outline as JSON.
    """
    prompt = f"""
You are a planning assistant.

Create an outline for this topic:
{topic}

Return ONLY valid JSON:
{{
  "title": "short title",
  "sections": ["section 1", "section 2", "section 3"]
}}

Rules:
- 3 to 5 sections
- concise section names
- no markdown
- no text outside JSON
""".strip()

    raw_text = call_model(prompt)
    data = json.loads(raw_text)
    validate_outline(data)
    return data


def write_draft(topic: str, outline: Dict[str, Any]) -> str:
    """
    Write a short draft article.
    """
    prompt = f"""
You are a beginner-friendly technical educator.

Write a short article for Python developers learning GenAI.

Topic: {topic}
Outline title: {outline["title"]}
Sections: {json.dumps(outline["sections"])}

Requirements:
- 250 to 400 words
- use markdown headings
- practical and clear
- include a conclusion
""".strip()

    draft = call_model(prompt)

    if not draft:
        raise ValueError("Draft must not be empty.")

    return draft


def review_draft(draft: str) -> Dict[str, Any]:
    """
    Review the draft and return structured JSON.
    """
    prompt = f"""
You are a content reviewer.

Review the draft below for:
- clarity
- structure
- completeness
- beginner-friendliness

Draft:
{draft}

Return ONLY valid JSON:
{{
  "approved": true,
  "issues": ["issue 1"],
  "suggestions": ["suggestion 1"]
}}

Rules:
- no text outside JSON
- approved must be true or false
- if there are serious issues, approved should be false
""".strip()

    raw_text = call_model(prompt)
    data = json.loads(raw_text)
    validate_review(data)
    return data


def revise_draft(draft: str, review: Dict[str, Any]) -> str:
    """
    Revise the draft using reviewer suggestions.
    """
    prompt = f"""
You are an editor.

Revise the draft below using the review feedback.

Draft:
{draft}

Review feedback:
{json.dumps(review, indent=2)}

Requirements:
- preserve the original topic
- improve clarity and completeness
- keep it beginner-friendly
- return only the revised article in markdown
""".strip()

    revised = call_model(prompt)

    if not revised:
        raise ValueError("Revised draft must not be empty.")

    return revised


def run_workflow(topic: str) -> Dict[str, Any]:
    """
    Execute the orchestrated workflow with validation and conditional revision.
    """
    state: Dict[str, Any] = {
        "topic": topic,
        "outline": None,
        "draft": None,
        "review": None,
        "revised_draft": None,
        "final_review": None,
    }

    log("Starting workflow")

    log("Planning outline")
    state["outline"] = with_retries(plan_outline, topic, retries=2)

    log("Writing draft")
    state["draft"] = write_draft(topic, state["outline"])

    log("Reviewing draft")
    state["review"] = with_retries(review_draft, state["draft"], retries=2)

    if state["review"]["approved"]:
        log("Draft approved on first review")
        state["revised_draft"] = state["draft"]
        state["final_review"] = state["review"]
    else:
        log("Draft not approved; revising once")
        state["revised_draft"] = revise_draft(state["draft"], state["review"])

        log("Reviewing revised draft")
        state["final_review"] = with_retries(review_draft, state["revised_draft"], retries=2)

    log("Workflow complete")
    return state


def main() -> None:
    """
    Run the improved workflow and print summary results.
    """
    topic = "How orchestrated LLM workflows improve reliability in Python apps"

    state = run_workflow(topic)

    print("\n=== FINAL OUTLINE ===")
    print(json.dumps(state["outline"], indent=2))

    print("\n=== FINAL DRAFT ===")
    print(state["revised_draft"])

    print("\n=== FINAL REVIEW ===")
    print(json.dumps(state["final_review"], indent=2))


if __name__ == "__main__":
    main()

7.3 Example Output

[14:03:01] Starting workflow
[14:03:01] Planning outline
[14:03:03] Writing draft
[14:03:06] Reviewing draft
[14:03:08] Draft not approved; revising once
[14:03:11] Reviewing revised draft
[14:03:13] Workflow complete

=== FINAL OUTLINE ===
{
  "title": "Reliable LLM Workflows in Python",
  "sections": [
    "Why One-Shot Prompts Struggle",
    "Breaking Work into Stages",
    "Validation and Review",
    "Practical Benefits for Developers"
  ]
}

=== FINAL DRAFT ===
# Reliable LLM Workflows in Python

## Why One-Shot Prompts Struggle
...

## Breaking Work into Stages
...

## Validation and Review
...

## Practical Benefits for Developers
...

## Conclusion
...

=== FINAL REVIEW ===
{
  "approved": true,
  "issues": [],
  "suggestions": [
    "You could add an example involving customer support or content generation."
  ]
}

8. Key Implementation Lessons

8.1 Orchestration Is Mostly Software Engineering

The “agentic” feeling often comes not from a single advanced model call, but from combining:

  • prompt design
  • state management
  • validation
  • branching
  • retries
  • modular design

8.2 Small, Clear Steps Beat Huge Prompts

A giant prompt may feel simpler at first, but smaller steps often win in:

  • debuggability
  • maintainability
  • reliability
  • extensibility

8.3 Structured Outputs Are a Superpower

A model returning JSON enables code to:

  • inspect values
  • route execution
  • trigger retries
  • approve/reject outputs
  • store state for later use

8.4 Review Steps Greatly Improve Quality

A reviewer step can catch:

  • missing sections
  • unclear wording
  • poor tone
  • weak explanations
  • formatting issues

This pattern is one of the most practical ways to improve output quality without changing the base model.

9. Mini Design Exercise

Take 5 minutes to sketch a workflow for one of these tasks:

  1. Support triage
  2. summarize incoming message
  3. classify intent
  4. estimate urgency
  5. draft response

  6. escalate if needed

  7. Study-note generator

  8. extract key concepts from text
  9. generate summary
  10. create quiz questions

  11. review for accuracy and difficulty

  12. Blog post helper

  13. brainstorm ideas
  14. build outline
  15. write draft
  16. critique and revise

For each design, answer:

  • What are the workflow steps?
  • What does each step receive?
  • What does each step return?
  • Where would validation happen?
  • Which step might branch?

10. Recap

In this session, you learned how to move from a single LLM call to a coordinated workflow.

Main Takeaways

  • A linear script is useful for simple tasks.
  • An orchestrated workflow is better for complex, multi-stage tasks.
  • Good workflows use:
  • modular steps
  • shared state
  • structured outputs
  • validation
  • retries
  • optional review/revision loops
  • Python should control the process; the LLM should perform language-heavy subtasks.

11. Suggested Practice After Class

Try extending the exercise in one of these ways:

  • save state to a JSON file after each step
  • add a router that chooses a workflow based on topic type
  • generate three draft variants and pick the best review score
  • add a “human approval” step before final output
  • convert the workflow into a reusable Python class

Useful Resources

  • OpenAI Responses API migration guide: https://developers.openai.com/api/docs/guides/migrate-to-responses
  • OpenAI API docs: https://developers.openai.com/api/
  • OpenAI Python SDK: https://github.com/openai/openai-python
  • JSON in Python: https://docs.python.org/3/library/json.html
  • Python typing: https://docs.python.org/3/library/typing.html

End-of-Session Checklist

By the end of this session, learners should be able to say:

  • [ ] I can explain why multi-step workflows are often better than one-shot prompts.
  • [ ] I can build a simple orchestrated pipeline in Python.
  • [ ] I can pass state between workflow stages.
  • [ ] I can parse and validate structured LLM outputs.
  • [ ] I can add retries and conditional branching.
  • [ ] I understand how orchestration supports more agentic application behavior.

Back to Chapter | Back to Master Plan | Previous Session | Next Session