"""
Pattern 51: Reinforcement Learning Agent
Agent learns optimal behavior through trial-and-error using rewards and penalties.
Implements Q-learning for state-action value estimation and policy improvement.
"""
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import random
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
class ActionType(Enum):
"""Types of actions in RL environment."""
MOVE_UP = "move_up"
MOVE_DOWN = "move_down"
MOVE_LEFT = "move_left"
MOVE_RIGHT = "move_right"
INTERACT = "interact"
WAIT = "wait"
@dataclass
class State:
"""Environment state."""
position: Tuple[int, int]
context: Dict[str, Any] = field(default_factory=dict)
timestamp: datetime = field(default_factory=datetime.now)
def __hash__(self):
return hash((self.position, tuple(sorted(self.context.items()))))
def __eq__(self, other):
return self.position == other.position and self.context == other.context
@dataclass
class Experience:
"""Single experience (s, a, r, s')."""
state: State
action: ActionType
reward: float
next_state: State
done: bool
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class Episode:
"""Episode of experiences."""
episode_number: int
experiences: List[Experience] = field(default_factory=list)
total_reward: float = 0.0
steps: int = 0
success: bool = False
class QLearningAgent:
"""Q-learning agent with exploration-exploitation."""
def __init__(self, name: str, alpha: float = 0.1, gamma: float = 0.9,
epsilon: float = 0.2, model_name: str = "gpt-4"):
self.name = name
self.alpha = alpha # Learning rate
self.gamma = gamma # Discount factor
self.epsilon = epsilon # Exploration rate
self.llm = ChatOpenAI(model=model_name, temperature=0.7)
# Q-table: state -> action -> value
self.q_table: Dict[State, Dict[ActionType, float]] = defaultdict(lambda: defaultdict(float))
self.experiences: List[Experience] = []
self.episodes: List[Episode] = []
def get_q_value(self, state: State, action: ActionType) -> float:
"""Get Q-value for state-action pair."""
return self.q_table[state][action]
def get_max_q_value(self, state: State, available_actions: List[ActionType]) -> float:
"""Get maximum Q-value for state."""
if not available_actions:
return 0.0
return max(self.get_q_value(state, action) for action in available_actions)
def select_action(self, state: State, available_actions: List[ActionType]) -> ActionType:
"""Select action using epsilon-greedy policy."""
if not available_actions:
return ActionType.WAIT
# Exploration
if random.random() Dict[ActionType, float]:
"""Get action probability distribution for state."""
q_values = {action: self.get_q_value(state, action) for action in available_actions}
if not q_values:
return {}
# Softmax policy
import math
exp_q = {a: math.exp(q / 0.1) for a, q in q_values.items()}
total = sum(exp_q.values())
return {a: exp_q[a] / total for a in available_actions}
def decay_epsilon(self, decay_rate: float = 0.99):
"""Decay exploration rate."""
self.epsilon *= decay_rate
self.epsilon = max(0.01, self.epsilon) # Minimum epsilon
class GridEnvironment:
"""Simple grid world environment for RL."""
def __init__(self, size: int = 5):
self.size = size
self.goal_position = (size - 1, size - 1)
self.obstacles = [(1, 1), (2, 2), (3, 1)]
self.current_position = (0, 0)
def reset(self) -> State:
"""Reset to initial state."""
self.current_position = (0, 0)
return State(position=self.current_position, context={"steps": 0})
def get_available_actions(self, position: Tuple[int, int]) -> List[ActionType]:
"""Get valid actions from position."""
actions = []
x, y = position
if x > 0 and (x-1, y) not in self.obstacles:
actions.append(ActionType.MOVE_LEFT)
if x 0 and (x, y-1) not in self.obstacles:
actions.append(ActionType.MOVE_UP)
if y Tuple[State, float, bool]:
"""Execute action and return (next_state, reward, done)."""
x, y = self.current_position
# Apply action
if action == ActionType.MOVE_UP:
y -= 1
elif action == ActionType.MOVE_DOWN:
y += 1
elif action == ActionType.MOVE_LEFT:
x -= 1
elif action == ActionType.MOVE_RIGHT:
x += 1
# Check bounds and obstacles
if (x = self.size or y = self.size or
(x, y) in self.obstacles):
# Invalid move - stay in place with penalty
reward = -1.0
done = False
elif (x, y) == self.goal_position:
# Reached goal
self.current_position = (x, y)
reward = 10.0
done = True
else:
# Valid move
self.current_position = (x, y)
reward = -0.1 # Small penalty for each step
done = False
next_state = State(position=self.current_position)
return next_state, reward, done
class ReinforcementLearningPattern:
"""
Reinforcement Learning Agent - Q-learning for optimal policy discovery.
Agent learns through interaction with environment, updating Q-values
based on rewards received for state-action pairs.
"""
def __init__(self, alpha: float = 0.1, gamma: float = 0.9, epsilon: float = 0.2):
self.alpha = alpha
self.gamma = gamma
self.epsilon = epsilon
self.agent: Optional[QLearningAgent] = None
self.environment: Optional[GridEnvironment] = None
def create_agent(self, name: str) -> QLearningAgent:
"""Create RL agent."""
agent = QLearningAgent(name, self.alpha, self.gamma, self.epsilon)
self.agent = agent
return agent
def create_environment(self, size: int = 5) -> GridEnvironment:
"""Create environment."""
env = GridEnvironment(size)
self.environment = env
return env
def train_episode(self, max_steps: int = 100) -> Episode:
"""Train for one episode."""
if not self.agent or not self.environment:
raise ValueError("Agent and environment must be created first")
state = self.environment.reset()
episode = Episode(episode_number=len(self.agent.episodes))
for step in range(max_steps):
# Select action
available_actions = self.environment.get_available_actions(state.position)
action = self.agent.select_action(state, available_actions)
# Take action
next_state, reward, done = self.environment.step(action)
# Learn
available_next_actions = self.environment.get_available_actions(next_state.position)
experience = Experience(state, action, reward, next_state, done)
self.agent.learn_from_experience(experience, available_next_actions)
# Record
episode.experiences.append(experience)
episode.total_reward += reward
episode.steps += 1
if done:
episode.success = True
break
state = next_state
self.agent.episodes.append(episode)
return episode
def train(self, num_episodes: int = 100, epsilon_decay: float = 0.99) -> List[Episode]:
"""Train agent for multiple episodes."""
print(f"\nšÆ Training RL agent for {num_episodes} episodes...\n")
episodes = []
for i in range(num_episodes):
episode = self.train_episode()
episodes.append(episode)
# Decay exploration
self.agent.decay_epsilon(epsilon_decay)
if (i + 1) % 20 == 0:
recent_episodes = episodes[-20:]
avg_reward = sum(e.total_reward for e in recent_episodes) / len(recent_episodes)
avg_steps = sum(e.steps for e in recent_episodes) / len(recent_episodes)
success_rate = sum(1 for e in recent_episodes if e.success) / len(recent_episodes)
print(f"Episode {i+1}/{num_episodes}:")
print(f" Avg reward: {avg_reward:.2f}")
print(f" Avg steps: {avg_steps:.1f}")
print(f" Success rate: {success_rate:.1%}")
print(f" Epsilon: {self.agent.epsilon:.3f}\n")
return episodes
def get_training_metrics(self, episodes: List[Episode]) -> Dict[str, Any]:
"""Calculate training metrics."""
total_episodes = len(episodes)
successful = sum(1 for e in episodes if e.success)
avg_reward = sum(e.total_reward for e in episodes) / total_episodes
avg_steps = sum(e.steps for e in episodes) / total_episodes
# Learning curve (reward over time)
window = 10
learning_curve = []
for i in range(window, total_episodes, window):
window_episodes = episodes[i-window:i]
avg = sum(e.total_reward for e in window_episodes) / len(window_episodes)
learning_curve.append(avg)
return {
"total_episodes": total_episodes,
"successful_episodes": successful,
"success_rate": successful / total_episodes,
"avg_reward": avg_reward,
"avg_steps": avg_steps,
"q_table_size": len(self.agent.q_table),
"learning_curve": learning_curve
}
# Example usage
if __name__ == "__main__":
print("=== Reinforcement Learning Agent Demo ===\n")
# Create pattern
print("1. Creating RL pattern with Q-learning...")
pattern = ReinforcementLearningPattern(
alpha=0.1, # Learning rate
gamma=0.9, # Discount factor
epsilon=0.3 # Initial exploration rate
)
# Create agent and environment
print("2. Setting up agent and grid environment (5x5)...")
agent = pattern.create_agent("Q-Learner")
env = pattern.create_environment(size=5)
print(f" Goal: {env.goal_position}")
print(f" Obstacles: {env.obstacles}\n")
# Train agent
print("3. Training agent...")
episodes = pattern.train(num_episodes=100, epsilon_decay=0.98)
# Get metrics
print(f"\n{'='*70}")
print("š Training Results")
print(f"{'='*70}")
metrics = pattern.get_training_metrics(episodes)
print(f" Total episodes: {metrics['total_episodes']}")
print(f" Successful: {metrics['successful_episodes']}")
print(f" Success rate: {metrics['success_rate']:.1%}")
print(f" Avg reward: {metrics['avg_reward']:.2f}")
print(f" Avg steps: {metrics['avg_steps']:.1f}")
print(f" Q-table size: {metrics['q_table_size']} states learned")
# Show learning curve
if metrics['learning_curve']:
print(f"\n Learning curve (avg reward per 10 episodes):")
for i, avg in enumerate(metrics['learning_curve'][:5]):
print(f" Episodes {i*10+1}-{(i+1)*10}: {avg:.2f}")
# Test learned policy
print(f"\n{'='*70}")
print("š® Testing Learned Policy")
print(f"{'='*70}\n")
# Set epsilon to 0 for pure exploitation
agent.epsilon = 0.0
state = env.reset()
print(f"Starting position: {state.position}")
print(f"Goal: {env.goal_position}\n")
path = [state.position]
total_reward = 0.0
for step in range(20):
available_actions = env.get_available_actions(state.position)
action = agent.select_action(state, available_actions)
next_state, reward, done = env.step(action)
path.append(next_state.position)
total_reward += reward
print(f"Step {step + 1}: {action.value} -> {next_state.position} (reward: {reward:+.1f})")
if done:
print(f"\nā
Goal reached in {step + 1} steps!")
print(f" Total reward: {total_reward:.2f}")
print(f" Path: {' -> '.join(str(p) for p in path)}")
break
state = next_state
print("\nā RL agent successfully learned optimal policy!")
Agentic Architecture Pattern: Reinforcement Learning
Pattern 51: 51 Reinforcement Learning – Line-by-Line Explanation
Overview
Pattern 51: Reinforcement Learning Agent
Agent learns optimal behavior through trial-and-error using rewards and penalties. Implements Q-learning for state-action value estimation and policy improvement.
Code Structure Analysis
File Information
- Pattern Number: 51
- Pattern Name: 51 Reinforcement Learning
- File Path:
patterns\learning_adaptation\51_reinforcement_learning.py - Total Lines: 412
Detailed Code Explanation
Module Docstring
"""
Pattern 51: Reinforcement Learning Agent
Agent learns optimal behavior through trial-and-error using rewards and penalties.
Implements Q-learning for state-action value estimation and policy improvement.
"""
Explanation: This module docstring describes the purpose and key features of the 51 Reinforcement Learning pattern.
Imports Section
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import random
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import math # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional, Tuple: 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
Enumerations
ActionType (Line 18)
class ActionType(Enum):
MOVE_UP
MOVE_DOWN
MOVE_LEFT
MOVE_RIGHT
INTERACT
WAIT
Purpose: Types of actions in RL environment.
Values:
– MOVE_UP: Represents move up
– MOVE_DOWN: Represents move down
– MOVE_LEFT: Represents move left
– MOVE_RIGHT: Represents move right
– INTERACT: Represents interact
– WAIT: Represents wait
Classes
ActionType (Line 18)
Description: Types of actions in RL environment.
State (Line 29)
Description: Environment state.
Methods:
__hash__ (Line 35)
def __hash__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
__eq__ (Line 38)
def __eq__(self, other):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– other: Parameter for other
Experience (Line 43)
Description: Single experience (s, a, r, s’).
Episode (Line 54)
Description: Episode of experiences.
QLearningAgent (Line 63)
Description: Q-learning agent with exploration-exploitation.
Methods:
__init__ (Line 66)
def __init__(self, name, alpha, gamma, epsilon, model_name):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– name: Identifier or label
– alpha: Parameter for alpha
– gamma: Parameter for gamma
– epsilon: Parameter for epsilon
– model_name: LLM model to use (e.g., “gpt-4”)
get_q_value (Line 79)
def get_q_value(self, state, action):
"""Get Q-value for state-action pair."""
Purpose: Get Q-value for state-action pair.
Parameters:
– self: Instance reference
– state: Current workflow state
– action: Parameter for action
get_max_q_value (Line 83)
def get_max_q_value(self, state, available_actions):
"""Get maximum Q-value for state."""
Purpose: Get maximum Q-value for state.
Parameters:
– self: Instance reference
– state: Current workflow state
– available_actions: Parameter for available_actions
select_action (Line 89)
def select_action(self, state, available_actions):
"""Select action using epsilon-greedy policy."""
Purpose: Select action using epsilon-greedy policy.
Parameters:
– self: Instance reference
– state: Current workflow state
– available_actions: Parameter for available_actions
update_q_value (Line 106)
def update_q_value(self, state, action, reward, next_state, available_next_actions, done):
"""Update Q-value using Q-learning update rule."""
Purpose: Update Q-value using Q-learning update rule.
Parameters:
– self: Instance reference
– state: Current workflow state
– action: Parameter for action
– reward: Parameter for reward
– next_state: Parameter for next_state
– available_next_actions: Parameter for available_next_actions
– done: Parameter for done
learn_from_experience (Line 122)
def learn_from_experience(self, experience, available_next_actions):
"""Learn from single experience."""
Purpose: Learn from single experience.
Parameters:
– self: Instance reference
– experience: Parameter for experience
– available_next_actions: Parameter for available_next_actions
get_policy (Line 134)
def get_policy(self, state, available_actions):
"""Get action probability distribution for state."""
Purpose: Get action probability distribution for state.
Parameters:
– self: Instance reference
– state: Current workflow state
– available_actions: Parameter for available_actions
decay_epsilon (Line 148)
def decay_epsilon(self, decay_rate):
"""Decay exploration rate."""
Purpose: Decay exploration rate.
Parameters:
– self: Instance reference
– decay_rate: Parameter for decay_rate
GridEnvironment (Line 154)
Description: Simple grid world environment for RL.
Methods:
__init__ (Line 157)
def __init__(self, size):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– size: Parameter for size
reset (Line 163)
def reset(self):
"""Reset to initial state."""
Purpose: Reset to initial state.
Parameters:
– self: Instance reference
get_available_actions (Line 168)
def get_available_actions(self, position):
"""Get valid actions from position."""
Purpose: Get valid actions from position.
Parameters:
– self: Instance reference
– position: Parameter for position
step (Line 184)
def step(self, action):
"""Execute action and return (next_state, reward, done)."""
Purpose: Execute action and return (next_state, reward, done).
Parameters:
– self: Instance reference
– action: Parameter for action
ReinforcementLearningPattern (Line 219)
Description: Reinforcement Learning Agent – Q-learning for optimal policy discovery.
Agent learns through interaction with environment, updating Q-values based on rewards received for state-action pairs.
Methods:
__init__ (Line 227)
def __init__(self, alpha, gamma, epsilon):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– alpha: Parameter for alpha
– gamma: Parameter for gamma
– epsilon: Parameter for epsilon
create_agent (Line 234)
def create_agent(self, name):
"""Create RL agent."""
Purpose: Create RL agent.
Parameters:
– self: Instance reference
– name: Identifier or label
create_environment (Line 240)
def create_environment(self, size):
"""Create environment."""
Purpose: Create environment.
Parameters:
– self: Instance reference
– size: Parameter for size
train_episode (Line 246)
def train_episode(self, max_steps):
"""Train for one episode."""
Purpose: Train for one episode.
Parameters:
– self: Instance reference
– max_steps: Parameter for max_steps
train (Line 281)
def train(self, num_episodes, epsilon_decay):
"""Train agent for multiple episodes."""
Purpose: Train agent for multiple episodes.
Parameters:
– self: Instance reference
– num_episodes: Parameter for num_episodes
– epsilon_decay: Parameter for epsilon_decay
get_training_metrics (Line 307)
def get_training_metrics(self, episodes):
"""Calculate training metrics."""
Purpose: Calculate training metrics.
Parameters:
– self: Instance reference
– episodes: Parameter for episodes
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 = ActionType(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
Leave a Reply