
    """
Pattern 1.4: Waterfall Pattern
Strictly sequential with phase gates and validation checkpoints.
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
import os


class WaterfallState(TypedDict):
    """State for waterfall pattern with validation gates."""
    requirements: str
    design: str
    implementation: str
    testing: str
    deployment: str
    validation_passed: bool


def requirements_phase(state: WaterfallState) -> WaterfallState:
    """Phase 1: Requirements gathering."""
    llm = ChatOpenAI()
    prompt = "Define comprehensive requirements for a task management system"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["requirements"] = str(response.content)
    state["validation_passed"] = True  # Validation gate
    return state


def design_phase(state: WaterfallState) -> WaterfallState:
    """Phase 2: Design based on requirements."""
    if not state.get("validation_passed"):
        return state
    llm = ChatOpenAI()
    prompt = f"Create system design based on: {state['requirements']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["design"] = str(response.content)
    return state


def implementation_phase(state: WaterfallState) -> WaterfallState:
    """Phase 3: Implementation."""
    llm = ChatOpenAI()
    prompt = f"Outline implementation steps for: {state['design']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["implementation"] = str(response.content)
    return state


def testing_phase(state: WaterfallState) -> WaterfallState:
    """Phase 4: Testing."""
    llm = ChatOpenAI()
    prompt = f"Define test plan for: {state['implementation']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["testing"] = str(response.content)
    return state


def deployment_phase(state: WaterfallState) -> WaterfallState:
    """Phase 5: Deployment."""
    llm = ChatOpenAI()
    prompt = f"Create deployment strategy for tested system: {state['testing'][:100]}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["deployment"] = str(response.content)
    return state


def create_waterfall():
    """Create waterfall workflow."""
    workflow = StateGraph(WaterfallState)
    
    workflow.add_node("requirements", requirements_phase)
    workflow.add_node("design", design_phase)
    workflow.add_node("implementation", implementation_phase)
    workflow.add_node("testing", testing_phase)
    workflow.add_node("deployment", deployment_phase)
    
    workflow.set_entry_point("requirements")
    workflow.add_edge("requirements", "design")
    workflow.add_edge("design", "implementation")
    workflow.add_edge("implementation", "testing")
    workflow.add_edge("testing", "deployment")
    workflow.add_edge("deployment", END)
    
    return workflow.compile()


if __name__ == "__main__":
    waterfall = create_waterfall()
    result = waterfall.invoke({
        "requirements": "",
        "design": "",
        "implementation": "",
        "testing": "",
        "deployment": "",
        "validation_passed": False
    })
    print(f"\n=== Waterfall Pattern ===\nDeployment Plan: {result['deployment'][:300]}...")

Agentic Orchestration Pattern: Waterfall Pattern

04 Waterfall Pattern – Detailed Documentation

Overview

File: 01_basic_coordination/04_waterfall.py
Pattern Category: 01 Basic Coordination
Implementation: LangGraph/CrewAI

Pattern Description

Pattern 1.4: Waterfall Pattern Strictly sequential with phase gates and validation checkpoints.

Code Structure

Imports

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
import os

Purpose of imports: – TypedDict: For type-safe state definitions – StateGraph: LangGraph’s graph-based workflow orchestration – ChatOpenAI: OpenAI LLM integration – HumanMessage: LangChain message formatting

State Definition

class WaterfallState(TypedDict):
    """State for waterfall pattern with validation gates."""
    requirements: str
    design: str
    implementation: str
    testing: str
    deployment: str
    validation_passed: bool


def requirements_phase(state: WaterfallState) -> WaterfallState:

WaterfallState: State for waterfall pattern with validation gates.

State Fields: – requirements: Stores requirements – design: Stores design – implementation: Stores implementation – testing: Stores testing – deployment: Stores deployment – validation_passed: Stores validation passed – def requirements_phase(state: Stores def requirements phase(state

Functions

`requirements_phase()`

Line: 22
Arguments: state

Phase 1: Requirements gathering.

def requirements_phase(state: WaterfallState) -> WaterfallState:
    """Phase 1: Requirements gathering."""
    llm = ChatOpenAI()
    prompt = "Define comprehensive requirements for a task management system"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["requirements"] = str(response.content)
    state["validation_passed"] = True  # Validation gate
    return state


def design_phase(state: WaterfallState) -> WaterfallState:

`design_phase()`

Line: 32
Arguments: state

Phase 2: Design based on requirements.

def design_phase(state: WaterfallState) -> WaterfallState:
    """Phase 2: Design based on requirements."""
    if not state.get("validation_passed"):
        return state
    llm = ChatOpenAI()
    prompt = f"Create system design based on: {state['requirements']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["design"] = str(response.content)
    return state


def implementation_phase(state: WaterfallState) -> WaterfallState:

`implementation_phase()`

Line: 43
Arguments: state

Phase 3: Implementation.

def implementation_phase(state: WaterfallState) -> WaterfallState:
    """Phase 3: Implementation."""
    llm = ChatOpenAI()
    prompt = f"Outline implementation steps for: {state['design']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["implementation"] = str(response.content)
    return state


def testing_phase(state: WaterfallState) -> WaterfallState:

`testing_phase()`

Line: 52
Arguments: state

Phase 4: Testing.

def testing_phase(state: WaterfallState) -> WaterfallState:
    """Phase 4: Testing."""
    llm = ChatOpenAI()
    prompt = f"Define test plan for: {state['implementation']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["testing"] = str(response.content)
    return state


def deployment_phase(state: WaterfallState) -> WaterfallState:

`deployment_phase()`

Line: 61
Arguments: state

Phase 5: Deployment.

def deployment_phase(state: WaterfallState) -> WaterfallState:
    """Phase 5: Deployment."""
    llm = ChatOpenAI()
    prompt = f"Create deployment strategy for tested system: {state['testing'][:100]}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["deployment"] = str(response.content)
    return state


def create_waterfall():

`create_waterfall()`

Line: 70
Arguments: “

Create waterfall workflow.

def create_waterfall():
    """Create waterfall workflow."""
    workflow = StateGraph(WaterfallState)

    workflow.add_node("requirements", requirements_phase)
    workflow.add_node("design", design_phase)
    workflow.add_node("implementation", implementation_phase)
    workflow.add_node("testing", testing_phase)
    workflow.add_node("deployment", deployment_phase)

    workflow.set_entry_point("requirements")
    workflow.add_edge("requirements", "design")
    workflow.add_edge("design", "implementation")
    workflow.add_edge("implementation", "testing")
    workflow.add_edge("testing", "deployment")
    workflow.add_edge("deployment", END)

    return workflow.compile()


if __name__ == "__main__":

Line-by-Line Explanation

Line 5: Import statement

from typing import TypedDict

Imports TypedDict for creating type-safe state dictionaries.

Line 6: Import statement

from langgraph.graph import StateGraph, END

Imports StateGraph – the core LangGraph class for building workflows.

Line 7: Import statement

from langchain_core.messages import HumanMessage

Imports HumanMessage for formatting prompts to the LLM.

Line 8: Import statement

from langchain_openai import ChatOpenAI

Imports ChatOpenAI for LLM interactions.

Line 9: Import statement

import os

Required import for the pattern implementation.

Line 12: Class definition

class WaterfallState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 13: Module docstring

    """State for waterfall pattern with validation gates."""