"""
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
- 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
Leave a Reply