
    """
Pattern 1.3: Pipeline Pattern
Continuous flow of data through agent stages using LangGraph.
"""
from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
import os


class PipelineState(TypedDict):
    """State for pipeline pattern with streaming capability."""
    input_data: str
    preprocessed: str
    transformed: str
    enriched: str
    output: str


def preprocess_stage(state: PipelineState) -> PipelineState:
    """Stage 1: Preprocess input data."""
    llm = ChatOpenAI(openai_api_key=os.getenv("OPENAI_API_KEY"))
    prompt = f"Clean and normalize this data: {state['input_data']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["preprocessed"] = str(response.content)
    return state


def transform_stage(state: PipelineState) -> PipelineState:
    """Stage 2: Transform preprocessed data."""
    llm = ChatOpenAI(openai_api_key=os.getenv("OPENAI_API_KEY"))
    prompt = f"Extract key insights from: {state['preprocessed']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["transformed"] = str(response.content)
    return state


def enrich_stage(state: PipelineState) -> PipelineState:
    """Stage 3: Enrich transformed data."""
    llm = ChatOpenAI(openai_api_key=os.getenv("OPENAI_API_KEY"))
    prompt = f"Add context and recommendations to: {state['transformed']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["enriched"] = str(response.content)
    state["output"] = str(response.content)
    return state


def create_pipeline():
    """Create pipeline workflow."""
    workflow = StateGraph(PipelineState)
    
    workflow.add_node("preprocess", preprocess_stage)
    workflow.add_node("transform", transform_stage)
    workflow.add_node("enrich", enrich_stage)
    
    workflow.set_entry_point("preprocess")
    workflow.add_edge("preprocess", "transform")
    workflow.add_edge("transform", "enrich")
    workflow.add_edge("enrich", END)
    
    return workflow.compile()


if __name__ == "__main__":
    pipeline = create_pipeline()
    result = pipeline.invoke({
        "input_data": "Customer feedback: Great product but slow delivery",
        "preprocessed": "",
        "transformed": "",
        "enriched": "",
        "output": ""
    })
    print(f"\n=== Pipeline Pattern ===\nOutput: {result['output']}")

Agentic Orchestration Pattern: Pipeline Pattern

03 Pipeline Pattern – Detailed Documentation

Overview

File: 01_basic_coordination/03_pipeline.py
Pattern Category: 01 Basic Coordination
Implementation: LangGraph/CrewAI

Pattern Description

Pattern 1.3: Pipeline Pattern Continuous flow of data through agent stages using LangGraph.

Code Structure

Imports

from typing import TypedDict, List
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 PipelineState(TypedDict):
    """State for pipeline pattern with streaming capability."""
    input_data: str
    preprocessed: str
    transformed: str
    enriched: str
    output: str


def preprocess_stage(state: PipelineState) -> PipelineState:

PipelineState: State for pipeline pattern with streaming capability.

State Fields: – input_data: Stores input data – preprocessed: Stores preprocessed – transformed: Stores transformed – enriched: Stores enriched – output: Stores output – def preprocess_stage(state: Stores def preprocess stage(state – prompt = f"Clean and normalize this data: Stores prompt = f”Clean and normalize this data

Functions

`preprocess_stage()`

Line: 21
Arguments: state

Stage 1: Preprocess input data.

def preprocess_stage(state: PipelineState) -> PipelineState:
    """Stage 1: Preprocess input data."""
    llm = ChatOpenAI(openai_api_key=os.getenv("OPENAI_API_KEY"))
    prompt = f"Clean and normalize this data: {state['input_data']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["preprocessed"] = str(response.content)
    return state


def transform_stage(state: PipelineState) -> PipelineState:

`transform_stage()`

Line: 30
Arguments: state

Stage 2: Transform preprocessed data.

def transform_stage(state: PipelineState) -> PipelineState:
    """Stage 2: Transform preprocessed data."""
    llm = ChatOpenAI(openai_api_key=os.getenv("OPENAI_API_KEY"))
    prompt = f"Extract key insights from: {state['preprocessed']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["transformed"] = str(response.content)
    return state


def enrich_stage(state: PipelineState) -> PipelineState:

`enrich_stage()`

Line: 39
Arguments: state

Stage 3: Enrich transformed data.

def enrich_stage(state: PipelineState) -> PipelineState:
    """Stage 3: Enrich transformed data."""
    llm = ChatOpenAI(openai_api_key=os.getenv("OPENAI_API_KEY"))
    prompt = f"Add context and recommendations to: {state['transformed']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["enriched"] = str(response.content)
    state["output"] = str(response.content)
    return state


def create_pipeline():

`create_pipeline()`

Line: 49
Arguments: “

Create pipeline workflow.

def create_pipeline():
    """Create pipeline workflow."""
    workflow = StateGraph(PipelineState)

    workflow.add_node("preprocess", preprocess_stage)
    workflow.add_node("transform", transform_stage)
    workflow.add_node("enrich", enrich_stage)

    workflow.set_entry_point("preprocess")
    workflow.add_edge("preprocess", "transform")
    workflow.add_edge("transform", "enrich")
    workflow.add_edge("enrich", END)

    return workflow.compile()


if __name__ == "__main__":

Line-by-Line Explanation

Line 5: Import statement

from typing import TypedDict, List

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 PipelineState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 13: Module docstring

    """State for pipeline pattern with streaming capability."""