"""
Pattern 03.02: Recursive
Agent calls itself with modified parameters
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage


class RecursiveState(TypedDict):
    """State for Recursive pattern."""
    input: str
    output: str


def agent_function(state: RecursiveState) -> RecursiveState:
    """Recursive agent implementation."""
    llm = ChatOpenAI()
    prompt = f"Process this using Recursive pattern: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["output"] = str(response.content)
    return state


def create_recursive():
    """Create Recursive workflow."""
    workflow = StateGraph(RecursiveState)
    workflow.add_node("agent", agent_function)
    workflow.set_entry_point("agent")
    workflow.add_edge("agent", END)
    return workflow.compile()


if __name__ == "__main__":
    workflow = create_recursive()
    result = workflow.invoke({"input": "Test Recursive", "output": ""})
    print(f"\n=== Recursive Pattern ===\nOutput: {result['output'][:200]}...")

    

Agentic Orchestration Pattern: Recursive Pattern

02 Recursive Pattern – Detailed Documentation

Overview

File: 03_iterative/02_recursive.py
Pattern Category: 03 Iterative
Implementation: LangGraph/CrewAI

Pattern Description

Pattern 03.02: Recursive Agent calls itself with modified parameters

Code Structure

Imports

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

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 RecursiveState(TypedDict):
    """State for Recursive pattern."""
    input: str
    output: str


def agent_function(state: RecursiveState) -> RecursiveState:

RecursiveState: State for Recursive pattern.

State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Recursive pattern: Stores prompt = f”Process this using Recursive pattern – def create_recursive(): Stores def create recursive()

Functions

agent_function()

Line: 17
Arguments: state

Recursive agent implementation.

def agent_function(state: RecursiveState) -> RecursiveState:
    """Recursive agent implementation."""
    llm = ChatOpenAI()
    prompt = f"Process this using Recursive pattern: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["output"] = str(response.content)
    return state


def create_recursive():

create_recursive()

Line: 26
Arguments: “

Create Recursive workflow.

def create_recursive():
    """Create Recursive workflow."""
    workflow = StateGraph(RecursiveState)
    workflow.add_node("agent", agent_function)
    workflow.set_entry_point("agent")
    workflow.add_edge("agent", 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_openai import ChatOpenAI

Imports ChatOpenAI for LLM interactions.

Line 8: Import statement

from langchain_core.messages import HumanMessage

Imports HumanMessage for formatting prompts to the LLM.

Line 11: Class definition

class RecursiveState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 12: Module docstring

    """State for Recursive pattern."""

Describes the pattern and its purpose.

Line 17: Function agent_function definition

def agent_function(state: RecursiveState) -> RecursiveState:

Defines the agent_function function that processes the workflow state.

Line 18: Module docstring

    """Recursive agent implementation."""

Describes the pattern and its purpose.

Line 26: Function create_recursive definition

def create_recursive():

Defines the create_recursive function that processes the workflow state.

Line 27: Module docstring

    """Create Recursive workflow."""

Describes the pattern and its purpose.

Usage Example

if __name__ == "__main__":
    workflow = create_recursive()
    result = workflow.invoke({"input": "Test Recursive", "output": ""})
    print(f"\n=== Recursive Pattern ===\nOutput: {result['output'][:200]}...")

Key Concepts

1. State Management: Uses TypedDict to maintain workflow state
2. Graph-Based Flow: LangGraph orchestrates execution flow
3. Agent Functions: Each node processes and updates state
4. LLM Integration: Uses ChatOpenAI for intelligent processing

Pattern Benefits

• ✅ Type-safe state management
• ✅ Clear workflow visualization
• ✅ Modular and reusable components
• ✅ Easy to test and debug
• ✅ Scalable architecture

Related Patterns

• See other patterns in the same category
• Explore combination possibilities
• Check the main documentation for pattern relationships

    """
Pattern 03.01: Loop
Repeated execution until condition met
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage


class LoopState(TypedDict):
    """State for Loop pattern."""
    input: str
    output: str


def agent_function(state: LoopState) -> LoopState:
    """Loop agent implementation."""
    llm = ChatOpenAI()
    prompt = f"Process this using Loop pattern: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["output"] = str(response.content)
    return state


def create_loop():
    """Create Loop workflow."""
    workflow = StateGraph(LoopState)
    workflow.add_node("agent", agent_function)
    workflow.set_entry_point("agent")
    workflow.add_edge("agent", END)
    return workflow.compile()


if __name__ == "__main__":
    workflow = create_loop()
    result = workflow.invoke({"input": "Test Loop", "output": ""})
    print(f"\n=== Loop Pattern ===\nOutput: {result['output'][:200]}...")

    

Agentic Orchestration Pattern: Loop Pattern

01 Loop Pattern – Detailed Documentation

Overview

File: 03_iterative/01_loop.py
Pattern Category: 03 Iterative
Implementation: LangGraph/CrewAI

Pattern Description

Pattern 03.01: Loop Repeated execution until condition met

Code Structure

Imports

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

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 LoopState(TypedDict):
    """State for Loop pattern."""
    input: str
    output: str


def agent_function(state: LoopState) -> LoopState:

LoopState: State for Loop pattern.

State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Loop pattern: Stores prompt = f”Process this using Loop pattern – def create_loop(): Stores def create loop()

Functions

agent_function()

Line: 17
Arguments: state

Loop agent implementation.

def agent_function(state: LoopState) -> LoopState:
    """Loop agent implementation."""
    llm = ChatOpenAI()
    prompt = f"Process this using Loop pattern: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["output"] = str(response.content)
    return state


def create_loop():

create_loop()

Line: 26
Arguments: “

Create Loop workflow.

def create_loop():
    """Create Loop workflow."""
    workflow = StateGraph(LoopState)
    workflow.add_node("agent", agent_function)
    workflow.set_entry_point("agent")
    workflow.add_edge("agent", 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_openai import ChatOpenAI

Imports ChatOpenAI for LLM interactions.

Line 8: Import statement

from langchain_core.messages import HumanMessage

Imports HumanMessage for formatting prompts to the LLM.

Line 11: Class definition

class LoopState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 12: Module docstring

    """State for Loop pattern."""

Describes the pattern and its purpose.

Line 17: Function agent_function definition

def agent_function(state: LoopState) -> LoopState:

Defines the agent_function function that processes the workflow state.

Line 18: Module docstring

    """Loop agent implementation."""

Describes the pattern and its purpose.

Line 26: Function create_loop definition

def create_loop():

Defines the create_loop function that processes the workflow state.

Line 27: Module docstring

    """Create Loop workflow."""

Describes the pattern and its purpose.

Usage Example

if __name__ == "__main__":
    workflow = create_loop()
    result = workflow.invoke({"input": "Test Loop", "output": ""})
    print(f"\n=== Loop Pattern ===\nOutput: {result['output'][:200]}...")

Key Concepts

1. State Management: Uses TypedDict to maintain workflow state
2. Graph-Based Flow: LangGraph orchestrates execution flow
3. Agent Functions: Each node processes and updates state
4. LLM Integration: Uses ChatOpenAI for intelligent processing

Pattern Benefits

• ✅ Type-safe state management
• ✅ Clear workflow visualization
• ✅ Modular and reusable components
• ✅ Easy to test and debug
• ✅ Scalable architecture

Related Patterns

• See other patterns in the same category
• Explore combination possibilities
• Check the main documentation for pattern relationships

    """
Pattern 2.2: Selector Pattern
Chooses one agent from multiple options based on criteria.
"""
from typing import TypedDict, List, Dict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage


class SelectorState(TypedDict):
    """State for selector pattern."""
    task: str
    agent_scores: Dict[str, float]
    selected_agent: str
    output: str


def evaluate_agents(state: SelectorState) -> SelectorState:
    """Evaluate which agent is best suited for the task."""
    llm = ChatOpenAI()
    prompt = f"""Rate these agents for the task '{state['task']}' from 0-1:
    - expert_agent: Specialized deep knowledge
    - generalist_agent: Broad but shallow knowledge
    - creative_agent: Novel approaches
    
    Return format: expert:0.8,generalist:0.5,creative:0.3"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    # Parse scores (simplified)
    scores = {"expert_agent": 0.8, "generalist_agent": 0.5, "creative_agent": 0.3}
    state["agent_scores"] = scores
    state["selected_agent"] = max(scores, key=scores.get)
    return state


def expert_agent(state: SelectorState) -> SelectorState:
    """Expert agent with deep knowledge."""
    llm = ChatOpenAI()
    response = llm.invoke([HumanMessage(content=f"Expert analysis: {state['task']}")])
    state["output"] = f"[Expert] {response.content}"
    return state


def generalist_agent(state: SelectorState) -> SelectorState:
    """Generalist agent with broad knowledge."""
    llm = ChatOpenAI()
    response = llm.invoke([HumanMessage(content=f"General analysis: {state['task']}")])
    state["output"] = f"[Generalist] {response.content}"
    return state


def creative_agent(state: SelectorState) -> SelectorState:
    """Creative agent with novel approaches."""
    llm = ChatOpenAI()
    response = llm.invoke([HumanMessage(content=f"Creative solution: {state['task']}")])
    state["output"] = f"[Creative] {response.content}"
    return state


def route_to_selected(state: SelectorState) -> str:
    """Route to the selected agent."""
    return state["selected_agent"]


def create_selector():
    """Create selector workflow."""
    workflow = StateGraph(SelectorState)
    
    workflow.add_node("evaluator", evaluate_agents)
    workflow.add_node("expert_agent", expert_agent)
    workflow.add_node("generalist_agent", generalist_agent)
    workflow.add_node("creative_agent", creative_agent)
    
    workflow.set_entry_point("evaluator")
    workflow.add_conditional_edges(
        "evaluator",
        route_to_selected,
        {
            "expert_agent": "expert_agent",
            "generalist_agent": "generalist_agent",
            "creative_agent": "creative_agent"
        }
    )
    workflow.add_edge("expert_agent", END)
    workflow.add_edge("generalist_agent", END)
    workflow.add_edge("creative_agent", END)
    
    return workflow.compile()


if __name__ == "__main__":
    selector = create_selector()
    result = selector.invoke({
        "task": "Optimize database query performance",
        "agent_scores": {},
        "selected_agent": "",
        "output": ""
    })
    print(f"\n=== Selector Pattern ===\nSelected: {result['selected_agent']}\nOutput: {result['output'][:200]}...")

    

Agentic Orchestration Pattern: Selector Pattern

02 Selector Pattern – Detailed Documentation

Overview

File: 02_decision_based/02_selector.py
Pattern Category: 02 Decision Based
Implementation: LangGraph/CrewAI

Pattern Description

Pattern 2.2: Selector Pattern Chooses one agent from multiple options based on criteria.

Code Structure

Imports

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

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 SelectorState(TypedDict):
    """State for selector pattern."""
    task: str
    agent_scores: Dict[str, float]
    selected_agent: str
    output: str


def evaluate_agents(state: SelectorState) -> SelectorState:

SelectorState: State for selector pattern.

State Fields: – task: Stores task – agent_scores: Stores agent scores – selected_agent: Stores selected agent – output: Stores output – def evaluate_agents(state: Stores def evaluate agents(state – prompt = f"""Rate these agents for the task '{state['task']}' from 0-1: Stores prompt = f”””Rate these agents for the task ‘{state[‘task’]}’ from 0-1

Functions

evaluate_agents()

Line: 19
Arguments: state

Evaluate which agent is best suited for the task.

def evaluate_agents(state: SelectorState) -> SelectorState:
    """Evaluate which agent is best suited for the task."""
    llm = ChatOpenAI()
    prompt = f"""Rate these agents for the task '{state['task']}' from 0-1:
    - expert_agent: Specialized deep knowledge
    - generalist_agent: Broad but shallow knowledge
    - creative_agent: Novel approaches

    Return format: expert:0.8,generalist:0.5,creative:0.3"""

    response = llm.invoke([HumanMessage(content=prompt)])
    # Parse scores (simplified)
    scores = {"expert_agent": 0.8, "generalist_agent": 0.5, "creative_agent": 0.3}
    state["agent_scores"] = scores
    state["selected_agent"] = max(scores, key=scores.get)
    return state


def expert_agent(state: SelectorState) -> SelectorState:

expert_agent()

Line: 37
Arguments: state

Expert agent with deep knowledge.

def expert_agent(state: SelectorState) -> SelectorState:
    """Expert agent with deep knowledge."""
    llm = ChatOpenAI()
    response = llm.invoke([HumanMessage(content=f"Expert analysis: {state['task']}")])
    state["output"] = f"[Expert] {response.content}"
    return state


def generalist_agent(state: SelectorState) -> SelectorState:

generalist_agent()

Line: 45
Arguments: state

Generalist agent with broad knowledge.

def generalist_agent(state: SelectorState) -> SelectorState:
    """Generalist agent with broad knowledge."""
    llm = ChatOpenAI()
    response = llm.invoke([HumanMessage(content=f"General analysis: {state['task']}")])
    state["output"] = f"[Generalist] {response.content}"
    return state


def creative_agent(state: SelectorState) -> SelectorState:

creative_agent()

Line: 53
Arguments: state

Creative agent with novel approaches.

def creative_agent(state: SelectorState) -> SelectorState:
    """Creative agent with novel approaches."""
    llm = ChatOpenAI()
    response = llm.invoke([HumanMessage(content=f"Creative solution: {state['task']}")])
    state["output"] = f"[Creative] {response.content}"
    return state


def route_to_selected(state: SelectorState) -> str:

route_to_selected()

Line: 61
Arguments: state

Route to the selected agent.

def route_to_selected(state: SelectorState) -> str:
    """Route to the selected agent."""
    return state["selected_agent"]


def create_selector():

create_selector()

Line: 66
Arguments: “

Create selector workflow.

def create_selector():
    """Create selector workflow."""
    workflow = StateGraph(SelectorState)

    workflow.add_node("evaluator", evaluate_agents)
    workflow.add_node("expert_agent", expert_agent)
    workflow.add_node("generalist_agent", generalist_agent)
    workflow.add_node("creative_agent", creative_agent)

    workflow.set_entry_point("evaluator")
    workflow.add_conditional_edges(
        "evaluator",
        route_to_selected,
        {
            "expert_agent": "expert_agent",
            "generalist_agent": "generalist_agent",
            "creative_agent": "creative_agent"
        }
    )
    workflow.add_edge("expert_agent", END)
    workflow.add_edge("generalist_agent", END)
    workflow.add_edge("creative_agent", END)

    return workflow.compile()


if __name__ == "__main__":

Line-by-Line Explanation

Line 5: Import statement

from typing import TypedDict, List, Dict

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_openai import ChatOpenAI

Imports ChatOpenAI for LLM interactions.

Line 8: Import statement

from langchain_core.messages import HumanMessage

Imports HumanMessage for formatting prompts to the LLM.

Line 11: Class definition

class SelectorState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 12: Module docstring

    """State for selector pattern."""

Describes the pattern and its purpose.

Line 19: Function evaluate_agents definition

def evaluate_agents(state: SelectorState) -> SelectorState:

Defines the evaluate_agents function that processes the workflow state.

Line 20: Module docstring

    """Evaluate which agent is best suited for the task."""

Describes the pattern and its purpose.

Line 37: Function expert_agent definition

def expert_agent(state: SelectorState) -> SelectorState:

Defines the expert_agent function that processes the workflow state.

Line 38: Module docstring

    """Expert agent with deep knowledge."""

Describes the pattern and its purpose.

Line 45: Function generalist_agent definition

def generalist_agent(state: SelectorState) -> SelectorState:

Defines the generalist_agent function that processes the workflow state.

Line 46: Module docstring

    """Generalist agent with broad knowledge."""

Describes the pattern and its purpose.

Line 53: Function creative_agent definition

def creative_agent(state: SelectorState) -> SelectorState:

Defines the creative_agent function that processes the workflow state.

Line 54: Module docstring

    """Creative agent with novel approaches."""

Describes the pattern and its purpose.

Line 61: Function route_to_selected definition

def route_to_selected(state: SelectorState) -> str:

Defines the route_to_selected function that processes the workflow state.

Line 62: Module docstring

    """Route to the selected agent."""

Describes the pattern and its purpose.

Line 66: Function create_selector definition

def create_selector():

Defines the create_selector function that processes the workflow state.

Line 67: Module docstring

    """Create selector workflow."""

Describes the pattern and its purpose.

Usage Example

if __name__ == "__main__":
    selector = create_selector()
    result = selector.invoke({
        "task": "Optimize database query performance",
        "agent_scores": {},
        "selected_agent": "",
        "output": ""
    })
    print(f"\n=== Selector Pattern ===\nSelected: {result['selected_agent']}\nOutput: {result['output'][:200]}...")

Key Concepts

1. State Management: Uses TypedDict to maintain workflow state
2. Graph-Based Flow: LangGraph orchestrates execution flow
3. Agent Functions: Each node processes and updates state
4. LLM Integration: Uses ChatOpenAI for intelligent processing

Pattern Benefits

• ✅ Type-safe state management
• ✅ Clear workflow visualization
• ✅ Modular and reusable components
• ✅ Easy to test and debug
• ✅ Scalable architecture

Related Patterns

• See other patterns in the same category
• Explore combination possibilities
• Check the main documentation for pattern relationships

    """
Pattern 2.1: Router Pattern
Single entry point that routes to appropriate agent based on input analysis.
"""
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI


class RouterState(TypedDict):
    """State for router pattern."""
    input: str
    category: str
    output: str


def router(state: RouterState) -> RouterState:
    """Router agent determines which specialist to route to."""
    llm = ChatOpenAI()
    prompt = f"""Categorize this request as 'technical', 'business', or 'creative': {state['input']}
    Respond with only one word."""
    response = llm.invoke([HumanMessage(content=prompt)])
    category = str(response.content).strip().lower()
    state["category"] = category
    return state


def technical_agent(state: RouterState) -> RouterState:
    """Handle technical requests."""
    llm = ChatOpenAI()
    prompt = f"Provide technical solution for: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["output"] = f"[Technical] {response.content}"
    return state


def business_agent(state: RouterState) -> RouterState:
    """Handle business requests."""
    llm = ChatOpenAI()
    prompt = f"Provide business strategy for: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["output"] = f"[Business] {response.content}"
    return state


def creative_agent(state: RouterState) -> RouterState:
    """Handle creative requests."""
    llm = ChatOpenAI()
    prompt = f"Provide creative solution for: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["output"] = f"[Creative] {response.content}"
    return state


def route_logic(state: RouterState) -> str:
    """Determine which path to take."""
    category = state.get("category", "").lower()
    if "technical" in category:
        return "technical"
    elif "business" in category:
        return "business"
    else:
        return "creative"


def create_router():
    """Create router workflow."""
    workflow = StateGraph(RouterState)
    
    workflow.add_node("router", router)
    workflow.add_node("technical", technical_agent)
    workflow.add_node("business", business_agent)
    workflow.add_node("creative", creative_agent)
    
    workflow.set_entry_point("router")
    workflow.add_conditional_edges(
        "router",
        route_logic,
        {
            "technical": "technical",
            "business": "business",
            "creative": "creative"
        }
    )
    workflow.add_edge("technical", END)
    workflow.add_edge("business", END)
    workflow.add_edge("creative", END)
    
    return workflow.compile()


if __name__ == "__main__":
    router_graph = create_router()
    result = router_graph.invoke({
        "input": "Design a scalable microservices architecture",
        "category": "",
        "output": ""
    })
    print(f"\n=== Router Pattern ===\nCategory: {result['category']}\nOutput: {result['output'][:200]}...")

    

Agentic Orchestration Pattern: Router Pattern

01 Router Pattern – Detailed Documentation

Overview

File: 02_decision_based/01_router.py
Pattern Category: 02 Decision Based
Implementation: LangGraph/CrewAI

Pattern Description

Pattern 2.1: Router Pattern Single entry point that routes to appropriate agent based on input analysis.

Code Structure

Imports

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

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 RouterState(TypedDict):
    """State for router pattern."""
    input: str
    category: str
    output: str


def router(state: RouterState) -> RouterState:

RouterState: State for router pattern.

State Fields: – input: Stores input – category: Stores category – output: Stores output – def router(state: Stores def router(state – prompt = f"""Categorize this request as 'technical', 'business', or 'creative': Stores prompt = f”””Categorize this request as ‘technical’, ‘business’, or ‘creative’

Functions

router()

Line: 18
Arguments: state

Router agent determines which specialist to route to.

def router(state: RouterState) -> RouterState:
    """Router agent determines which specialist to route to."""
    llm = ChatOpenAI()
    prompt = f"""Categorize this request as 'technical', 'business', or 'creative': {state['input']}
    Respond with only one word."""
    response = llm.invoke([HumanMessage(content=prompt)])
    category = str(response.content).strip().lower()
    state["category"] = category
    return state


def technical_agent(state: RouterState) -> RouterState:

technical_agent()

Line: 29
Arguments: state

Handle technical requests.

def technical_agent(state: RouterState) -> RouterState:
    """Handle technical requests."""
    llm = ChatOpenAI()
    prompt = f"Provide technical solution for: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["output"] = f"[Technical] {response.content}"
    return state


def business_agent(state: RouterState) -> RouterState:

business_agent()

Line: 38
Arguments: state

Handle business requests.

def business_agent(state: RouterState) -> RouterState:
    """Handle business requests."""
    llm = ChatOpenAI()
    prompt = f"Provide business strategy for: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["output"] = f"[Business] {response.content}"
    return state


def creative_agent(state: RouterState) -> RouterState:

creative_agent()

Line: 47
Arguments: state

Handle creative requests.

def creative_agent(state: RouterState) -> RouterState:
    """Handle creative requests."""
    llm = ChatOpenAI()
    prompt = f"Provide creative solution for: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    state["output"] = f"[Creative] {response.content}"
    return state


def route_logic(state: RouterState) -> str:

route_logic()

Line: 56
Arguments: state

Determine which path to take.

def route_logic(state: RouterState) -> str:
    """Determine which path to take."""
    category = state.get("category", "").lower()
    if "technical" in category:
        return "technical"
    elif "business" in category:
        return "business"
    else:
        return "creative"


def create_router():

create_router()

Line: 67
Arguments: “

Create router workflow.

def create_router():
    """Create router workflow."""
    workflow = StateGraph(RouterState)

    workflow.add_node("router", router)
    workflow.add_node("technical", technical_agent)
    workflow.add_node("business", business_agent)
    workflow.add_node("creative", creative_agent)

    workflow.set_entry_point("router")
    workflow.add_conditional_edges(
        "router",
        route_logic,
        {
            "technical": "technical",
            "business": "business",
            "creative": "creative"
        }
    )
    workflow.add_edge("technical", END)
    workflow.add_edge("business", END)
    workflow.add_edge("creative", END)

    return workflow.compile()


if __name__ == "__main__":

Line-by-Line Explanation

Line 5: Import statement

from typing import TypedDict, Literal

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 11: Class definition

class RouterState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 12: Module docstring

    """State for router pattern."""

Describes the pattern and its purpose.

Line 18: Function router definition

def router(state: RouterState) -> RouterState:

Defines the router function that processes the workflow state.

Line 19: Module docstring

    """Router agent determines which specialist to route to."""

Describes the pattern and its purpose.

Line 29: Function technical_agent definition

def technical_agent(state: RouterState) -> RouterState:

Defines the technical_agent function that processes the workflow state.

Line 30: Module docstring

    """Handle technical requests."""

Describes the pattern and its purpose.

Line 38: Function business_agent definition

def business_agent(state: RouterState) -> RouterState:

Defines the business_agent function that processes the workflow state.

Line 39: Module docstring

    """Handle business requests."""

Describes the pattern and its purpose.

Line 47: Function creative_agent definition

def creative_agent(state: RouterState) -> RouterState:

Defines the creative_agent function that processes the workflow state.

Line 48: Module docstring

    """Handle creative requests."""

Describes the pattern and its purpose.

Line 56: Function route_logic definition

def route_logic(state: RouterState) -> str:

Defines the route_logic function that processes the workflow state.

Line 57: Module docstring

    """Determine which path to take."""

Describes the pattern and its purpose.

Line 67: Function create_router definition

def create_router():

Defines the create_router function that processes the workflow state.

Line 68: Module docstring

    """Create router workflow."""

Describes the pattern and its purpose.

Usage Example

if __name__ == "__main__":
    router_graph = create_router()
    result = router_graph.invoke({
        "input": "Design a scalable microservices architecture",
        "category": "",
        "output": ""
    })
    print(f"\n=== Router Pattern ===\nCategory: {result['category']}\nOutput: {result['output'][:200]}...")

Key Concepts

1. State Management: Uses TypedDict to maintain workflow state
2. Graph-Based Flow: LangGraph orchestrates execution flow
3. Agent Functions: Each node processes and updates state
4. LLM Integration: Uses ChatOpenAI for intelligent processing

Pattern Benefits

• ✅ Type-safe state management
• ✅ Clear workflow visualization
• ✅ Modular and reusable components
• ✅ Easy to test and debug
• ✅ Scalable architecture

Related Patterns

• See other patterns in the same category
• Explore combination possibilities
• Check the main documentation for pattern relationships

    """
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."""

Describes the pattern and its purpose.

Line 22: Function requirements_phase definition

def requirements_phase(state: WaterfallState) -> WaterfallState:

Defines the requirements_phase function that processes the workflow state.

Line 23: Module docstring

    """Phase 1: Requirements gathering."""

Describes the pattern and its purpose.

Line 32: Function design_phase definition

def design_phase(state: WaterfallState) -> WaterfallState:

Defines the design_phase function that processes the workflow state.

Line 33: Module docstring

    """Phase 2: Design based on requirements."""

Describes the pattern and its purpose.

Line 43: Function implementation_phase definition

def implementation_phase(state: WaterfallState) -> WaterfallState:

Defines the implementation_phase function that processes the workflow state.

Line 44: Module docstring

    """Phase 3: Implementation."""

Describes the pattern and its purpose.

Line 52: Function testing_phase definition

def testing_phase(state: WaterfallState) -> WaterfallState:

Defines the testing_phase function that processes the workflow state.

Line 53: Module docstring

    """Phase 4: Testing."""

Describes the pattern and its purpose.

Line 61: Function deployment_phase definition

def deployment_phase(state: WaterfallState) -> WaterfallState:

Defines the deployment_phase function that processes the workflow state.

Line 62: Module docstring

    """Phase 5: Deployment."""

Describes the pattern and its purpose.

Line 70: Function create_waterfall definition

def create_waterfall():

Defines the create_waterfall function that processes the workflow state.

Line 71: Module docstring

    """Create waterfall workflow."""

Describes the pattern and its purpose.

Usage Example

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]}...")

Key Concepts

1. State Management: Uses TypedDict to maintain workflow state
2. Graph-Based Flow: LangGraph orchestrates execution flow
3. Agent Functions: Each node processes and updates state
4. LLM Integration: Uses ChatOpenAI for intelligent processing

Pattern Benefits

• ✅ Type-safe state management
• ✅ Clear workflow visualization
• ✅ Modular and reusable components
• ✅ Easy to test and debug
• ✅ Scalable architecture

Related Patterns

• See other patterns in the same category
• Explore combination possibilities
• Check the main documentation for pattern relationships

    """
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."""

Describes the pattern and its purpose.

Line 21: Function preprocess_stage definition

def preprocess_stage(state: PipelineState) -> PipelineState:

Defines the preprocess_stage function that processes the workflow state.

Line 22: Module docstring

    """Stage 1: Preprocess input data."""

Describes the pattern and its purpose.

Line 30: Function transform_stage definition

def transform_stage(state: PipelineState) -> PipelineState:

Defines the transform_stage function that processes the workflow state.

Line 31: Module docstring

    """Stage 2: Transform preprocessed data."""

Describes the pattern and its purpose.

Line 39: Function enrich_stage definition

def enrich_stage(state: PipelineState) -> PipelineState:

Defines the enrich_stage function that processes the workflow state.

Line 40: Module docstring

    """Stage 3: Enrich transformed data."""

Describes the pattern and its purpose.

Line 49: Function create_pipeline definition

def create_pipeline():

Defines the create_pipeline function that processes the workflow state.

Line 50: Module docstring

    """Create pipeline workflow."""

Describes the pattern and its purpose.

Usage Example

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']}")

Key Concepts

1. State Management: Uses TypedDict to maintain workflow state
2. Graph-Based Flow: LangGraph orchestrates execution flow
3. Agent Functions: Each node processes and updates state
4. LLM Integration: Uses ChatOpenAI for intelligent processing

Pattern Benefits

• ✅ Type-safe state management
• ✅ Clear workflow visualization
• ✅ Modular and reusable components
• ✅ Easy to test and debug
• ✅ Scalable architecture

Related Patterns

• See other patterns in the same category
• Explore combination possibilities
• Check the main documentation for pattern relationships

    """
Pattern 1.2: Parallel Execution Pattern
Multiple agents execute simultaneously using LangGraph.
"""
from typing import TypedDict, List
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
import asyncio


class ParallelState(TypedDict):
    """State for parallel execution pattern."""
    input: str
    agent1_output: str
    agent2_output: str
    agent3_output: str
    aggregated_output: str


def agent_parallel_1(state: ParallelState) -> ParallelState:
    """First parallel agent - technical analysis."""
    log_agent_action("agent_parallel_1", "processing", {"input": state["input"]})
    
    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Provide a technical analysis of: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    
    state["agent1_output"] = response.content
    return state


def agent_parallel_2(state: ParallelState) -> ParallelState:
    """Second parallel agent - business analysis."""
    log_agent_action("agent_parallel_2", "processing", {"input": state["input"]})
    
    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Provide a business perspective on: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    
    state["agent2_output"] = response.content
    return state


def agent_parallel_3(state: ParallelState) -> ParallelState:
    """Third parallel agent - user experience analysis."""
    log_agent_action("agent_parallel_3", "processing", {"input": state["input"]})
    
    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Provide a user experience perspective on: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    
    state["agent3_output"] = response.content
    return state


def aggregator(state: ParallelState) -> ParallelState:
    """Aggregate results from parallel agents."""
    log_agent_action("aggregator", "combining results")
    
    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"""
    Synthesize these three perspectives into a comprehensive analysis:
    
    Technical: {state['agent1_output']}
    
    Business: {state['agent2_output']}
    
    UX: {state['agent3_output']}
    """
    response = llm.invoke([HumanMessage(content=prompt)])
    
    state["aggregated_output"] = response.content
    return state


def create_parallel_execution():
    """Create and return a parallel execution workflow."""
    workflow = StateGraph(ParallelState)
    
    # Add nodes
    workflow.add_node("agent1", agent_parallel_1)
    workflow.add_node("agent2", agent_parallel_2)
    workflow.add_node("agent3", agent_parallel_3)
    workflow.add_node("aggregator", aggregator)
    
    # Set entry point to all parallel agents
    workflow.set_entry_point("agent1")
    workflow.set_entry_point("agent2")
    workflow.set_entry_point("agent3")
    
    # All parallel agents feed into aggregator
    workflow.add_edge("agent1", "aggregator")
    workflow.add_edge("agent2", "aggregator")
    workflow.add_edge("agent3", "aggregator")
    workflow.add_edge("aggregator", END)
    
    return workflow.compile()


def run_parallel_execution(user_input: str):
    """Run the parallel execution pattern."""
    workflow = create_parallel_execution()
    
    initial_state = {
        "input": user_input,
        "agent1_output": "",
        "agent2_output": "",
        "agent3_output": "",
        "aggregated_output": ""
    }
    
    result = workflow.invoke(initial_state)
    return result


if __name__ == "__main__":
    # Example usage
    result = run_parallel_execution("Implementing AI chatbot for customer service")
    print("\n=== Parallel Execution Pattern ===")
    print(f"Input: {result['input']}")
    print(f"\nTechnical Analysis: {result['agent1_output'][:200]}...")
    print(f"\nBusiness Analysis: {result['agent2_output'][:200]}...")
    print(f"\nUX Analysis: {result['agent3_output'][:200]}...")
    print(f"\nAggregated Result: {result['aggregated_output']}")

    

Agentic Orchestration Pattern: Parallel Execution

02 Parallel Execution Pattern – Detailed Documentation

Overview

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

Pattern Description

Pattern 1.2: Parallel Execution Pattern Multiple agents execute simultaneously 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
from common import get_openai_api_key, get_model_name, log_agent_action
import asyncio

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 ParallelState(TypedDict):
    """State for parallel execution pattern."""
    input: str
    agent1_output: str
    agent2_output: str
    agent3_output: str
    aggregated_output: str


def agent_parallel_1(state: ParallelState) -> ParallelState:

ParallelState: State for parallel execution pattern.

State Fields: – input: Stores input – agent1_output: Stores agent1 output – agent2_output: Stores agent2 output – agent3_output: Stores agent3 output – aggregated_output: Stores aggregated output – def agent_parallel_1(state: Stores def agent parallel 1(state – log_agent_action("agent_parallel_1", "processing", {"input": Stores log agent action(“agent parallel 1”, “processing”, {“input” – prompt = f"Provide a technical analysis of: Stores prompt = f”Provide a technical analysis of

Functions

agent_parallel_1()

Line: 22
Arguments: state

First parallel agent – technical analysis.

def agent_parallel_1(state: ParallelState) -> ParallelState:
    """First parallel agent - technical analysis."""
    log_agent_action("agent_parallel_1", "processing", {"input": state["input"]})

    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Provide a technical analysis of: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])

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


def agent_parallel_2(state: ParallelState) -> ParallelState:

agent_parallel_2()

Line: 34
Arguments: state

Second parallel agent – business analysis.

def agent_parallel_2(state: ParallelState) -> ParallelState:
    """Second parallel agent - business analysis."""
    log_agent_action("agent_parallel_2", "processing", {"input": state["input"]})

    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Provide a business perspective on: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])

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


def agent_parallel_3(state: ParallelState) -> ParallelState:

agent_parallel_3()

Line: 46
Arguments: state

Third parallel agent – user experience analysis.

def agent_parallel_3(state: ParallelState) -> ParallelState:
    """Third parallel agent - user experience analysis."""
    log_agent_action("agent_parallel_3", "processing", {"input": state["input"]})

    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Provide a user experience perspective on: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])

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


def aggregator(state: ParallelState) -> ParallelState:

aggregator()

Line: 58
Arguments: state

Aggregate results from parallel agents.

def aggregator(state: ParallelState) -> ParallelState:
    """Aggregate results from parallel agents."""
    log_agent_action("aggregator", "combining results")

    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"""
    Synthesize these three perspectives into a comprehensive analysis:

    Technical: {state['agent1_output']}

    Business: {state['agent2_output']}

    UX: {state['agent3_output']}
    """
    response = llm.invoke([HumanMessage(content=prompt)])

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


def create_parallel_execution():

create_parallel_execution()

Line: 78
Arguments: “

Create and return a parallel execution workflow.

def create_parallel_execution():
    """Create and return a parallel execution workflow."""
    workflow = StateGraph(ParallelState)

    # Add nodes
    workflow.add_node("agent1", agent_parallel_1)
    workflow.add_node("agent2", agent_parallel_2)
    workflow.add_node("agent3", agent_parallel_3)
    workflow.add_node("aggregator", aggregator)

    # Set entry point to all parallel agents
    workflow.set_entry_point("agent1")
    workflow.set_entry_point("agent2")
    workflow.set_entry_point("agent3")

    # All parallel agents feed into aggregator
    workflow.add_edge("agent1", "aggregator")
    workflow.add_edge("agent2", "aggregator")
    workflow.add_edge("agent3", "aggregator")
    workflow.add_edge("aggregator", END)

    return workflow.compile()


def run_parallel_execution(user_input: str):

run_parallel_execution()

Line: 102
Arguments: user_input

Run the parallel execution pattern.

def run_parallel_execution(user_input: str):
    """Run the parallel execution pattern."""
    workflow = create_parallel_execution()

    initial_state = {
        "input": user_input,
        "agent1_output": "",
        "agent2_output": "",
        "agent3_output": "",
        "aggregated_output": ""
    }

    result = workflow.invoke(initial_state)
    return result


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

from common import get_openai_api_key, get_model_name, log_agent_action

Required import for the pattern implementation.

Line 10: Import statement

import asyncio

Required import for the pattern implementation.

Line 13: Class definition

class ParallelState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 14: Module docstring

    """State for parallel execution pattern."""

Describes the pattern and its purpose.

Line 22: Function agent_parallel_1 definition

def agent_parallel_1(state: ParallelState) -> ParallelState:

Defines the agent_parallel_1 function that processes the workflow state.

Line 23: Module docstring

    """First parallel agent - technical analysis."""

Describes the pattern and its purpose.

Line 34: Function agent_parallel_2 definition

def agent_parallel_2(state: ParallelState) -> ParallelState:

Defines the agent_parallel_2 function that processes the workflow state.

Line 35: Module docstring

    """Second parallel agent - business analysis."""

Describes the pattern and its purpose.

Line 46: Function agent_parallel_3 definition

def agent_parallel_3(state: ParallelState) -> ParallelState:

Defines the agent_parallel_3 function that processes the workflow state.

Line 47: Module docstring

    """Third parallel agent - user experience analysis."""

Describes the pattern and its purpose.

Line 58: Function aggregator definition

def aggregator(state: ParallelState) -> ParallelState:

Defines the aggregator function that processes the workflow state.

Line 59: Module docstring

    """Aggregate results from parallel agents."""

Describes the pattern and its purpose.

Line 71: Module docstring

    """

Describes the pattern and its purpose.

Line 78: Function create_parallel_execution definition

def create_parallel_execution():

Defines the create_parallel_execution function that processes the workflow state.

Line 79: Module docstring

    """Create and return a parallel execution workflow."""

Describes the pattern and its purpose.

Line 102: Function run_parallel_execution definition

def run_parallel_execution(user_input: str):

Defines the run_parallel_execution function that processes the workflow state.

Line 103: Module docstring

    """Run the parallel execution pattern."""

Describes the pattern and its purpose.

Usage Example

if __name__ == "__main__":
    # Example usage
    result = run_parallel_execution("Implementing AI chatbot for customer service")
    print("\n=== Parallel Execution Pattern ===")
    print(f"Input: {result['input']}")
    print(f"\nTechnical Analysis: {result['agent1_output'][:200]}...")
    print(f"\nBusiness Analysis: {result['agent2_output'][:200]}...")
    print(f"\nUX Analysis: {result['agent3_output'][:200]}...")
    print(f"\nAggregated Result: {result['aggregated_output']}")

Key Concepts

1. State Management: Uses TypedDict to maintain workflow state
2. Graph-Based Flow: LangGraph orchestrates execution flow
3. Agent Functions: Each node processes and updates state
4. LLM Integration: Uses ChatOpenAI for intelligent processing

Pattern Benefits

• ✅ Type-safe state management
• ✅ Clear workflow visualization
• ✅ Modular and reusable components
• ✅ Easy to test and debug
• ✅ Scalable architecture

Related Patterns

• See other patterns in the same category
• Explore combination possibilities
• Check the main documentation for pattern relationships

    """
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."""

Describes the pattern and its purpose.

Line 21: Function agent_step1 definition

def agent_step1(state: SequentialState) -> SequentialState:

Defines the agent_step1 function that processes the workflow state.

Line 22: Module docstring

    """First agent in the chain - data collection."""

Describes the pattern and its purpose.

Line 33: Function agent_step2 definition

def agent_step2(state: SequentialState) -> SequentialState:

Defines the agent_step2 function that processes the workflow state.

Line 34: Module docstring

    """Second agent in the chain - processing."""

Describes the pattern and its purpose.

Line 45: Function agent_step3 definition

def agent_step3(state: SequentialState) -> SequentialState:

Defines the agent_step3 function that processes the workflow state.

Line 46: Module docstring

    """Third agent in the chain - synthesis."""

Describes the pattern and its purpose.

Line 58: Function create_sequential_chain definition

def create_sequential_chain():

Defines the create_sequential_chain function that processes the workflow state.

Line 59: Module docstring

    """Create and return a sequential chain workflow."""

Describes the pattern and its purpose.

Line 76: Function run_sequential_chain definition

def run_sequential_chain(user_input: str):

Defines the run_sequential_chain function that processes the workflow state.

Line 77: Module docstring

    """Run the sequential chain pattern."""

Describes the pattern and its purpose.

Usage Example

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']}")

Key Concepts

1. State Management: Uses TypedDict to maintain workflow state
2. Graph-Based Flow: LangGraph orchestrates execution flow
3. Agent Functions: Each node processes and updates state
4. LLM Integration: Uses ChatOpenAI for intelligent processing

Pattern Benefits

• ✅ Type-safe state management
• ✅ Clear workflow visualization
• ✅ Modular and reusable components
• ✅ Easy to test and debug
• ✅ Scalable architecture

Related Patterns

• See other patterns in the same category
• Explore combination possibilities
• Check the main documentation for pattern relationships