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