"""
Pattern 04.04: Tree Orchestration
Multi-level hierarchy with parent-child relationships
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage


class TreeOrchestrationState(TypedDict):
    """State for Tree Orchestration pattern."""
    input: str
    output: str


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


def create_tree_orchestration():
    """Create Tree Orchestration workflow."""
    workflow = StateGraph(TreeOrchestrationState)
    workflow.add_node("agent", agent_function)
    workflow.set_entry_point("agent")
    workflow.add_edge("agent", END)
    return workflow.compile()


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

    

Agentic Orchestration Pattern: Tree Orchestration

04 Tree Orchestration Pattern – Detailed Documentation

Overview

File: 04_hierarchical/04_tree_orchestration.py
Pattern Category: 04 Hierarchical
Implementation: LangGraph/CrewAI

Pattern Description

Pattern 04.04: Tree Orchestration Multi-level hierarchy with parent-child relationships

---

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


def agent_function(state: TreeOrchestrationState) -> TreeOrchestrationState:

TreeOrchestrationState: State for Tree Orchestration pattern.

State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Tree Orchestration pattern: Stores prompt = f”Process this using Tree Orchestration pattern – def create_tree_orchestration(): Stores def create tree orchestration()

---

Functions

agent_function()

Line: 17
Arguments: state

Tree Orchestration agent implementation.

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


def create_tree_orchestration():

create_tree_orchestration()

Line: 26
Arguments: “

Create Tree Orchestration workflow.

def create_tree_orchestration():
    """Create Tree Orchestration workflow."""
    workflow = StateGraph(TreeOrchestrationState)
    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 TreeOrchestrationState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 12: Module docstring

    """State for Tree Orchestration pattern."""

Describes the pattern and its purpose.

Line 17: Function agent_function definition

def agent_function(state: TreeOrchestrationState) -> TreeOrchestrationState:

Defines the agent_function function that processes the workflow state.

Line 18: Module docstring

    """Tree Orchestration agent implementation."""

Describes the pattern and its purpose.

Line 26: Function create_tree_orchestration definition

def create_tree_orchestration():

Defines the create_tree_orchestration function that processes the workflow state.

Line 27: Module docstring

    """Create Tree Orchestration workflow."""

Describes the pattern and its purpose.

---

Usage Example

if __name__ == "__main__":
    workflow = create_tree_orchestration()
    result = workflow.invoke({"input": "Test Tree Orchestration", "output": ""})
    print(f"\n=== Tree Orchestration 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 04.03: Delegation
High-level agent delegates to specialized agents
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage


class DelegationState(TypedDict):
    """State for Delegation pattern."""
    input: str
    output: str


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


def create_delegation():
    """Create Delegation workflow."""
    workflow = StateGraph(DelegationState)
    workflow.add_node("agent", agent_function)
    workflow.set_entry_point("agent")
    workflow.add_edge("agent", END)
    return workflow.compile()


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

    

Agentic Orchestration Pattern: Delegation Pattern

03 Delegation Pattern – Detailed Documentation

Overview

File: 04_hierarchical/03_delegation.py
Pattern Category: 04 Hierarchical
Implementation: LangGraph/CrewAI

Pattern Description

Pattern 04.03: Delegation High-level agent delegates to specialized agents

---

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


def agent_function(state: DelegationState) -> DelegationState:

DelegationState: State for Delegation pattern.

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

---

Functions

agent_function()

Line: 17
Arguments: state

Delegation agent implementation.

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


def create_delegation():

create_delegation()

Line: 26
Arguments: “

Create Delegation workflow.

def create_delegation():
    """Create Delegation workflow."""
    workflow = StateGraph(DelegationState)
    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 DelegationState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 12: Module docstring

    """State for Delegation pattern."""

Describes the pattern and its purpose.

Line 17: Function agent_function definition

def agent_function(state: DelegationState) -> DelegationState:

Defines the agent_function function that processes the workflow state.

Line 18: Module docstring

    """Delegation agent implementation."""

Describes the pattern and its purpose.

Line 26: Function create_delegation definition

def create_delegation():

Defines the create_delegation function that processes the workflow state.

Line 27: Module docstring

    """Create Delegation workflow."""

Describes the pattern and its purpose.

---

Usage Example

if __name__ == "__main__":
    workflow = create_delegation()
    result = workflow.invoke({"input": "Test Delegation", "output": ""})
    print(f"\n=== Delegation 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 04.02: Supervisor-Subordinate
Hierarchical command structure with escalation
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage


class SupervisorSubordinateState(TypedDict):
    """State for Supervisor-Subordinate pattern."""
    input: str
    output: str


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


def create_supervisor_subordinate():
    """Create Supervisor-Subordinate workflow."""
    workflow = StateGraph(SupervisorSubordinateState)
    workflow.add_node("agent", agent_function)
    workflow.set_entry_point("agent")
    workflow.add_edge("agent", END)
    return workflow.compile()


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

    

Agentic Orchestration Pattern: Supervisor-Subordinate

02 Supervisor Subordinate Pattern – Detailed Documentation

Overview

File: 04_hierarchical/02_supervisor_subordinate.py
Pattern Category: 04 Hierarchical
Implementation: LangGraph/CrewAI

Pattern Description

Pattern 04.02: Supervisor-Subordinate Hierarchical command structure with escalation

---

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


def agent_function(state: SupervisorSubordinateState) -> SupervisorSubordinateState:

SupervisorSubordinateState: State for Supervisor-Subordinate pattern.

State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Supervisor-Subordinate pattern: Stores prompt = f”Process this using Supervisor-Subordinate pattern – def create_supervisor_subordinate(): Stores def create supervisor subordinate()

---

Functions

agent_function()

Line: 17
Arguments: state

Supervisor-Subordinate agent implementation.

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


def create_supervisor_subordinate():

create_supervisor_subordinate()

Line: 26
Arguments: “

Create Supervisor-Subordinate workflow.

def create_supervisor_subordinate():
    """Create Supervisor-Subordinate workflow."""
    workflow = StateGraph(SupervisorSubordinateState)
    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 SupervisorSubordinateState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 12: Module docstring

    """State for Supervisor-Subordinate pattern."""

Describes the pattern and its purpose.

Line 17: Function agent_function definition

def agent_function(state: SupervisorSubordinateState) -> SupervisorSubordinateState:

Defines the agent_function function that processes the workflow state.

Line 18: Module docstring

    """Supervisor-Subordinate agent implementation."""

Describes the pattern and its purpose.

Line 26: Function create_supervisor_subordinate definition

def create_supervisor_subordinate():

Defines the create_supervisor_subordinate function that processes the workflow state.

Line 27: Module docstring

    """Create Supervisor-Subordinate workflow."""

Describes the pattern and its purpose.

---

Usage Example

if __name__ == "__main__":
    workflow = create_supervisor_subordinate()
    result = workflow.invoke({"input": "Test Supervisor-Subordinate", "output": ""})
    print(f"\n=== Supervisor-Subordinate 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 04.01: Master-Worker
Central coordinator delegates to workers
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage


class MasterWorkerState(TypedDict):
    """State for Master-Worker pattern."""
    input: str
    output: str


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


def create_master_worker():
    """Create Master-Worker workflow."""
    workflow = StateGraph(MasterWorkerState)
    workflow.add_node("agent", agent_function)
    workflow.set_entry_point("agent")
    workflow.add_edge("agent", END)
    return workflow.compile()


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

    

Agentic Orchestration Pattern: Master-Worker Pattern

01 Master Worker Pattern – Detailed Documentation

Overview

File: 04_hierarchical/01_master_worker.py
Pattern Category: 04 Hierarchical
Implementation: LangGraph/CrewAI

Pattern Description

Pattern 04.01: Master-Worker Central coordinator delegates to workers

---

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


def agent_function(state: MasterWorkerState) -> MasterWorkerState:

MasterWorkerState: State for Master-Worker pattern.

State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Master-Worker pattern: Stores prompt = f”Process this using Master-Worker pattern – def create_master_worker(): Stores def create master worker()

---

Functions

agent_function()

Line: 17
Arguments: state

Master-Worker agent implementation.

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


def create_master_worker():

create_master_worker()

Line: 26
Arguments: “

Create Master-Worker workflow.

def create_master_worker():
    """Create Master-Worker workflow."""
    workflow = StateGraph(MasterWorkerState)
    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 MasterWorkerState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 12: Module docstring

    """State for Master-Worker pattern."""

Describes the pattern and its purpose.

Line 17: Function agent_function definition

def agent_function(state: MasterWorkerState) -> MasterWorkerState:

Defines the agent_function function that processes the workflow state.

Line 18: Module docstring

    """Master-Worker agent implementation."""

Describes the pattern and its purpose.

Line 26: Function create_master_worker definition

def create_master_worker():

Defines the create_master_worker function that processes the workflow state.

Line 27: Module docstring

    """Create Master-Worker workflow."""

Describes the pattern and its purpose.

---

Usage Example

if __name__ == "__main__":
    workflow = create_master_worker()
    result = workflow.invoke({"input": "Test Master-Worker", "output": ""})
    print(f"\n=== Master-Worker 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.04: Refinement Loop
Iterative improvement of results with quality assessment
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage


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


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


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


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

    

Agentic Orchestration Pattern: Refinement Loop

04 Refinement Loop Pattern – Detailed Documentation

Overview

File: 03_iterative/04_refinement_loop.py
Pattern Category: 03 Iterative
Implementation: LangGraph/CrewAI

Pattern Description

Pattern 03.04: Refinement Loop Iterative improvement of results with quality assessment

---

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


def agent_function(state: RefinementLoopState) -> RefinementLoopState:

RefinementLoopState: State for Refinement Loop pattern.

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

---

Functions

agent_function()

Line: 17
Arguments: state

Refinement Loop agent implementation.

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


def create_refinement_loop():

create_refinement_loop()

Line: 26
Arguments: “

Create Refinement Loop workflow.

def create_refinement_loop():
    """Create Refinement Loop workflow."""
    workflow = StateGraph(RefinementLoopState)
    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 RefinementLoopState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 12: Module docstring

    """State for Refinement Loop pattern."""

Describes the pattern and its purpose.

Line 17: Function agent_function definition

def agent_function(state: RefinementLoopState) -> RefinementLoopState:

Defines the agent_function function that processes the workflow state.

Line 18: Module docstring

    """Refinement Loop agent implementation."""

Describes the pattern and its purpose.

Line 26: Function create_refinement_loop definition

def create_refinement_loop():

Defines the create_refinement_loop function that processes the workflow state.

Line 27: Module docstring

    """Create Refinement Loop workflow."""

Describes the pattern and its purpose.

---

Usage Example

if __name__ == "__main__":
    workflow = create_refinement_loop()
    result = workflow.invoke({"input": "Test Refinement Loop", "output": ""})
    print(f"\n=== Refinement 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 03.03: Retry
Automatic retry on failure with exponential backoff
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage


class RetryState(TypedDict):
    """State for Retry pattern."""
    input: str
    output: str


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


def create_retry():
    """Create Retry workflow."""
    workflow = StateGraph(RetryState)
    workflow.add_node("agent", agent_function)
    workflow.set_entry_point("agent")
    workflow.add_edge("agent", END)
    return workflow.compile()


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

    

Agentic Orchestration Pattern: Retry Pattern

03 Retry Pattern – Detailed Documentation

Overview

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

Pattern Description

Pattern 03.03: Retry Automatic retry on failure with exponential backoff

---

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


def agent_function(state: RetryState) -> RetryState:

RetryState: State for Retry pattern.

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

---

Functions

agent_function()

Line: 17
Arguments: state

Retry agent implementation.

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


def create_retry():

create_retry()

Line: 26
Arguments: “

Create Retry workflow.

def create_retry():
    """Create Retry workflow."""
    workflow = StateGraph(RetryState)
    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 RetryState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 12: Module docstring

    """State for Retry pattern."""

Describes the pattern and its purpose.

Line 17: Function agent_function definition

def agent_function(state: RetryState) -> RetryState:

Defines the agent_function function that processes the workflow state.

Line 18: Module docstring

    """Retry agent implementation."""

Describes the pattern and its purpose.

Line 26: Function create_retry definition

def create_retry():

Defines the create_retry function that processes the workflow state.

Line 27: Module docstring

    """Create Retry workflow."""

Describes the pattern and its purpose.

---

Usage Example

if __name__ == "__main__":
    workflow = create_retry()
    result = workflow.invoke({"input": "Test Retry", "output": ""})
    print(f"\n=== Retry 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.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