"""
Pattern 298: Rate Limiting
Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict
@dataclass
class RateLimitingConfig:
"""Configuration for Rate Limiting."""
model_name: str = "gpt-4"
temperature: float = 0.7
max_iterations: int = 10
enable_logging: bool = True
@dataclass
class RateLimitingState:
"""State information for Rate Limiting."""
current_step: int = 0
data: Dict[str, Any] = field(default_factory=dict)
results: List[Any] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
timestamp: datetime = field(default_factory=datetime.now)
class RateLimitingAgent:
"""Agent implementing Rate Limiting."""
def __init__(self, name: str, config: RateLimitingConfig = None):
self.name = name
self.config = config or RateLimitingConfig()
self.llm = ChatOpenAI(
model=self.config.model_name,
temperature=self.config.temperature
)
self.state = RateLimitingState()
self.execution_history: List[Dict[str, Any]] = []
def process(self, input_data: Any) -> Dict[str, Any]:
"""Process input according to Rate Limiting."""
self.state.current_step += 1
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}, implementing Rate Limiting. "
f"Provide clear, structured responses."),
("human", "Process this input: {input_data}")
])
chain = prompt | self.llm
response = chain.invoke({"input_data": str(input_data)})
result = {
"agent": self.name,
"step": self.state.current_step,
"input": input_data,
"output": str(response.content),
"timestamp": datetime.now()
}
self.execution_history.append(result)
self.state.results.append(result)
return result
def get_statistics(self) -> Dict[str, Any]:
"""Get agent statistics."""
return {
"name": self.name,
"total_executions": len(self.execution_history),
"current_step": self.state.current_step,
"results_count": len(self.state.results)
}
class RateLimitingPattern:
"""
Rate Limiting - Enhanced implementation
This pattern provides comprehensive functionality for rate limiting
with full LangChain/LangGraph integration.
Features:
- Multi-agent coordination
- State management
- Execution tracking
- LangGraph workflow integration
"""
def __init__(self, config: RateLimitingConfig = None):
self.config = config or RateLimitingConfig()
self.agents: Dict[str, RateLimitingAgent] = {}
self.workflow = self._build_workflow()
self.execution_log: List[Dict[str, Any]] = []
def _build_workflow(self) -> StateGraph:
"""Build LangGraph workflow for pattern execution."""
class WorkflowState(TypedDict):
input: str
processed: str
output: str
metadata: Dict[str, Any]
def process_input(state: WorkflowState) -> WorkflowState:
"""Process input node."""
return {
**state,
"processed": f"Processed: {state['input']}",
"metadata": {"step": "process_input"}
}
def generate_output(state: WorkflowState) -> WorkflowState:
"""Generate output node."""
llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
prompt = ChatPromptTemplate.from_messages([
("system", "You are implementing Rate Limiting. Generate comprehensive output."),
("human", "{input}")
])
chain = prompt | llm
response = chain.invoke({"input": state["processed"]})
return {
**state,
"output": str(response.content),
"metadata": {**state.get("metadata", {}), "step": "generate_output"}
}
# Build workflow graph
workflow = StateGraph(WorkflowState)
workflow.add_node("process", process_input)
workflow.add_node("generate", generate_output)
workflow.set_entry_point("process")
workflow.add_edge("process", "generate")
workflow.add_edge("generate", END)
return workflow.compile()
def register_agent(self, agent: RateLimitingAgent):
"""Register an agent with the pattern."""
self.agents[agent.name] = agent
if self.config.enable_logging:
print(f"Registered agent: {agent.name}")
def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
"""Execute the pattern with given input."""
# Use workflow if no specific agent requested
if agent_name is None:
result = self.workflow.invoke({
"input": str(input_data),
"processed": "",
"output": "",
"metadata": {}
})
execution_record = {
"pattern": "Rate Limiting",
"input": input_data,
"output": result.get("output", ""),
"metadata": result.get("metadata", {}),
"timestamp": datetime.now()
}
self.execution_log.append(execution_record)
return execution_record
# Execute with specific agent
agent = self.agents.get(agent_name)
if not agent:
raise ValueError(f"Agent {agent_name} not found")
result = agent.process(input_data)
self.execution_log.append(result)
return result
def run(self, input_data: Any) -> Dict[str, Any]:
"""Main execution method."""
return self.execute(input_data)
def get_statistics(self) -> Dict[str, Any]:
"""Get pattern statistics."""
return {
"pattern": "Rate Limiting",
"total_executions": len(self.execution_log),
"registered_agents": len(self.agents),
"agents": [agent.get_statistics() for agent in self.agents.values()]
}
# Example usage
if __name__ == "__main__":
print("=== Rate Limiting Demo ===\n")
# Create configuration
config = RateLimitingConfig(
model_name="gpt-4",
temperature=0.7,
enable_logging=True
)
# Create pattern
pattern = RateLimitingPattern(config)
# Create and register agents
agent1 = RateLimitingAgent("Agent1", config)
agent2 = RateLimitingAgent("Agent2", config)
pattern.register_agent(agent1)
pattern.register_agent(agent2)
print("\nExecuting pattern workflow...")
# Execute with workflow
result1 = pattern.run("Demonstrate rate limiting functionality")
print(f"\nWorkflow Result:")
print(f" Output: {result1['output'][:150]}...")
# Execute with specific agents
result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
print(f"\nAgent1 Result:")
print(f" Output: {result2['output'][:150]}...")
result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
print(f"\nAgent2 Result:")
print(f" Output: {result3['output'][:150]}...")
# Show statistics
print("\n=== Statistics ===")
stats = pattern.get_statistics()
print(f"Pattern: {stats['pattern']}")
print(f"Total Executions: {stats['total_executions']}")
print(f"Registered Agents: {stats['registered_agents']}")
print(f"\nAgent Statistics:")
for agent_stats in stats['agents']:
print(f" {agent_stats['name']}: {agent_stats['total_executions']} executions")
Agentic Architecture Pattern: Rate Limiting
Pattern 298: 298 Rate Limiting – Line-by-Line Explanation
Overview
Pattern 298: Rate Limiting
Purpose: Enhanced implementation with full LangChain/LangGraph integration
Code Structure Analysis
File Information
- Pattern Number: 298
- Pattern Name: 298 Rate Limiting
- File Path:
patterns\additional_advanced\298_rate_limiting.py
- Total Lines: 247
Detailed Code Explanation
Module Docstring
"""
Pattern 298: Rate Limiting
Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""
Explanation: This module docstring describes the purpose and key features of the 298 Rate Limiting pattern.
Imports Section
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional: Type hints for better code documentation
– from dataclasses import dataclass, field: Data class decorators for clean data structures
– from datetime import datetime: Date and time utilities
– from enum import Enum: Enumeration support for fixed sets of values
– from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration
– from langchain_openai import ChatOpenAI: LangChain framework integration
– from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration
– from typing_extensions import TypedDict: Type hints for better code documentation
Classes
RateLimitingConfig (Line 18)
Description: Configuration for Rate Limiting.
RateLimitingState (Line 27)
Description: State information for Rate Limiting.
RateLimitingAgent (Line 36)
Description: Agent implementing Rate Limiting.
Methods:
__init__ (Line 39)
def __init__(self, name, config):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– name: Identifier or label
– config: Configuration settings
process (Line 49)
def process(self, input_data):
"""Process input according to Rate Limiting."""
Purpose: Process input according to Rate Limiting.
Parameters:
– self: Instance reference
– input_data: Parameter for input_data
get_statistics (Line 75)
def get_statistics(self):
"""Get agent statistics."""
Purpose: Get agent statistics.
Parameters:
– self: Instance reference
RateLimitingPattern (Line 85)
Description: Rate Limiting – Enhanced implementation
This pattern provides comprehensive functionality for rate limiting
with full LangChain/LangGraph integration.
Features:
– Multi-agent coordination
– State management
– Execution tracking
– LangGraph workflow integration
Methods:
__init__ (Line 99)
def __init__(self, config):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– config: Configuration settings
_build_workflow (Line 105)
def _build_workflow(self):
"""Build LangGraph workflow for pattern execution."""
Purpose: Build LangGraph workflow for pattern execution.
Parameters:
– self: Instance reference
register_agent (Line 150)
def register_agent(self, agent):
"""Register an agent with the pattern."""
Purpose: Register an agent with the pattern.
Parameters:
– self: Instance reference
– agent: Parameter for agent
execute (Line 156)
def execute(self, input_data, agent_name):
"""Execute the pattern with given input."""
Purpose: Execute the pattern with given input.
Parameters:
– self: Instance reference
– input_data: Parameter for input_data
– agent_name: Parameter for agent_name
run (Line 188)
def run(self, input_data):
"""Main execution method."""
Purpose: Main execution method.
Parameters:
– self: Instance reference
– input_data: Parameter for input_data
get_statistics (Line 192)
def get_statistics(self):
"""Get pattern statistics."""
Purpose: Get pattern statistics.
Parameters:
– self: Instance reference
WorkflowState (Line 108)
Description: No class docstring.
Design Patterns Used
- Strategy Pattern: Different execution strategies for various scenarios
- State Pattern: Managing different states in the workflow
- Chain of Responsibility: Sequential processing through workflow nodes
- Observer Pattern: Tracking and monitoring pattern execution
- Factory Pattern: Creating instances with proper configuration
Usage Example
# Initialize the pattern
pattern = RateLimitingConfig(name="example", model_name="gpt-4")
# Configure and execute
result = pattern.execute(data={"key": "value"})
# Check results
print(result)
Key Concepts
- LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
- Workflow Orchestration: LangGraph manages execution flow through nodes
- State Management: TypedDict defines workflow state structure
- Error Handling: Comprehensive error handling and fallback mechanisms
- Extensibility: Designed for easy customization and extension
Extension Points
This pattern can be extended by:
1. Adding custom workflow nodes
2. Implementing additional validation logic
3. Integrating with external services
4. Customizing LLM prompts and responses
5. Adding monitoring and logging capabilities