
    """
Pattern 1.1: Sequential Chain Pattern
Linear execution of agents in a predefined order using LangGraph.
"""
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from common import get_openai_api_key, get_model_name, log_agent_action


class SequentialState(TypedDict):
    """State for sequential chain pattern."""
    input: str
    step1_output: str
    step2_output: str
    step3_output: str
    final_output: str


def agent_step1(state: SequentialState) -> SequentialState:
    """First agent in the chain - data collection."""
    log_agent_action("agent_step1", "processing", {"input": state["input"]})
    
    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Analyze the following request and extract key requirements: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    
    state["step1_output"] = response.content
    return state


def agent_step2(state: SequentialState) -> SequentialState:
    """Second agent in the chain - processing."""
    log_agent_action("agent_step2", "processing", {"previous": state["step1_output"]})
    
    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Based on these requirements, create a plan: {state['step1_output']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    
    state["step2_output"] = response.content
    return state


def agent_step3(state: SequentialState) -> SequentialState:
    """Third agent in the chain - synthesis."""
    log_agent_action("agent_step3", "finalizing", {"previous": state["step2_output"]})
    
    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Summarize this plan into actionable steps: {state['step2_output']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    
    state["step3_output"] = response.content
    state["final_output"] = response.content
    return state


def create_sequential_chain():
    """Create and return a sequential chain workflow."""
    workflow = StateGraph(SequentialState)
    
    # Add nodes
    workflow.add_node("step1", agent_step1)
    workflow.add_node("step2", agent_step2)
    workflow.add_node("step3", agent_step3)
    
    # Define sequential edges
    workflow.set_entry_point("step1")
    workflow.add_edge("step1", "step2")
    workflow.add_edge("step2", "step3")
    workflow.add_edge("step3", END)
    
    return workflow.compile()


def run_sequential_chain(user_input: str):
    """Run the sequential chain pattern."""
    chain = create_sequential_chain()
    
    initial_state = {
        "input": user_input,
        "step1_output": "",
        "step2_output": "",
        "step3_output": "",
        "final_output": ""
    }
    
    result = chain.invoke(initial_state)
    return result


if __name__ == "__main__":
    # Example usage
    result = run_sequential_chain("Build a web application for task management")
    print("\n=== Sequential Chain Pattern ===")
    print(f"Input: {result['input']}")
    print(f"\nStep 1 Output: {result['step1_output']}")
    print(f"\nStep 2 Output: {result['step2_output']}")
    print(f"\nFinal Output: {result['final_output']}")

Agentic Orchestration Pattern: Sequential Chain

01 Sequential Chain Pattern – Detailed Documentation

Overview

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

Pattern Description

Pattern 1.1: Sequential Chain Pattern Linear execution of agents in a predefined order using LangGraph.

Code Structure

Imports

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from common import get_openai_api_key, get_model_name, log_agent_action

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 SequentialState(TypedDict):
    """State for sequential chain pattern."""
    input: str
    step1_output: str
    step2_output: str
    step3_output: str
    final_output: str


def agent_step1(state: SequentialState) -> SequentialState:

SequentialState: State for sequential chain pattern.

State Fields: – input: Stores input – step1_output: Stores step1 output – step2_output: Stores step2 output – step3_output: Stores step3 output – final_output: Stores final output – def agent_step1(state: Stores def agent step1(state – log_agent_action("agent_step1", "processing", {"input": Stores log agent action(“agent step1”, “processing”, {“input” – prompt = f"Analyze the following request and extract key requirements: Stores prompt = f”Analyze the following request and extract key requirements

Functions

`agent_step1()`

Line: 21
Arguments: state

First agent in the chain – data collection.

def agent_step1(state: SequentialState) -> SequentialState:
    """First agent in the chain - data collection."""
    log_agent_action("agent_step1", "processing", {"input": state["input"]})

    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Analyze the following request and extract key requirements: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])

    state["step1_output"] = response.content
    return state


def agent_step2(state: SequentialState) -> SequentialState:

`agent_step2()`

Line: 33
Arguments: state

Second agent in the chain – processing.

def agent_step2(state: SequentialState) -> SequentialState:
    """Second agent in the chain - processing."""
    log_agent_action("agent_step2", "processing", {"previous": state["step1_output"]})

    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Based on these requirements, create a plan: {state['step1_output']}"
    response = llm.invoke([HumanMessage(content=prompt)])

    state["step2_output"] = response.content
    return state


def agent_step3(state: SequentialState) -> SequentialState:

`agent_step3()`

Line: 45
Arguments: state

Third agent in the chain – synthesis.

def agent_step3(state: SequentialState) -> SequentialState:
    """Third agent in the chain - synthesis."""
    log_agent_action("agent_step3", "finalizing", {"previous": state["step2_output"]})

    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Summarize this plan into actionable steps: {state['step2_output']}"
    response = llm.invoke([HumanMessage(content=prompt)])

    state["step3_output"] = response.content
    state["final_output"] = response.content
    return state


def create_sequential_chain():

`create_sequential_chain()`

Line: 58
Arguments: “

Create and return a sequential chain workflow.

def create_sequential_chain():
    """Create and return a sequential chain workflow."""
    workflow = StateGraph(SequentialState)

    # Add nodes
    workflow.add_node("step1", agent_step1)
    workflow.add_node("step2", agent_step2)
    workflow.add_node("step3", agent_step3)

    # Define sequential edges
    workflow.set_entry_point("step1")
    workflow.add_edge("step1", "step2")
    workflow.add_edge("step2", "step3")
    workflow.add_edge("step3", END)

    return workflow.compile()


def run_sequential_chain(user_input: str):

`run_sequential_chain()`

Line: 76
Arguments: user_input

Run the sequential chain pattern.

def run_sequential_chain(user_input: str):
    """Run the sequential chain pattern."""
    chain = create_sequential_chain()

    initial_state = {
        "input": user_input,
        "step1_output": "",
        "step2_output": "",
        "step3_output": "",
        "final_output": ""
    }

    result = chain.invoke(initial_state)
    return result


if __name__ == "__main__":

Line-by-Line Explanation

Line 5: Import statement

from typing import TypedDict, Annotated

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

from common import get_openai_api_key, get_model_name, log_agent_action

Required import for the pattern implementation.

Line 12: Class definition

class SequentialState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 13: Module docstring

    """State for sequential chain pattern."""