"""
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
- State Management: Uses TypedDict to maintain workflow state
- Graph-Based Flow: LangGraph orchestrates execution flow
- Agent Functions: Each node processes and updates state
- 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