Agentic Orchestration Pattern: Router Pattern

Written by

in


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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *