"""
Pattern 19: Event-Driven Pattern
Agents react to events in the system asynchronously.
Uses event bus for decoupled event publishing and subscription.
"""
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
@dataclass
class Event:
"""Event in the system."""
event_type: str
source: str
data: Any
timestamp: datetime = field(default_factory=datetime.now)
priority: int = 1
class EventDrivenAgent:
"""Agent that reacts to events."""
def __init__(self, name: str, role: str, model_name: str = "gpt-4"):
self.name = name
self.role = role
self.llm = ChatOpenAI(model=model_name, temperature=0.7)
self.event_handlers: Dict[str, Callable] = {}
self.event_bus: Optional['EventBus'] = None
def subscribe(self, event_type: str, handler: Optional[Callable] = None):
"""Subscribe to event type."""
if handler:
self.event_handlers[event_type] = handler
else:
# Use default LLM-based handler
self.event_handlers[event_type] = self._default_handler
if self.event_bus:
self.event_bus.subscribe(event_type, self)
def _default_handler(self, event: Event) -> str:
"""Default event handler using LLM."""
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}, a {self.role}. "
f"React to this event appropriately."),
("human", f"Event: {event.event_type}\n"
f"Source: {event.source}\n"
f"Data: {event.data}")
])
chain = prompt | self.llm
response = chain.invoke({})
return str(response.content)
def handle_event(self, event: Event) -> str:
"""Handle received event."""
if event.event_type in self.event_handlers:
return self.event_handlers[event.event_type](event)
return f"{self.name}: Unhandled event {event.event_type}"
def publish(self, event_type: str, data: Any, priority: int = 1):
"""Publish event to event bus."""
if self.event_bus:
event = Event(
event_type=event_type,
source=self.name,
data=data,
priority=priority
)
self.event_bus.publish(event)
class EventBus:
"""Central event bus for event-driven communication."""
def __init__(self):
self.subscribers: Dict[str, List[EventDrivenAgent]] = defaultdict(list)
self.event_log: List[Event] = []
def subscribe(self, event_type: str, agent: EventDrivenAgent):
"""Subscribe agent to event type."""
if agent not in self.subscribers[event_type]:
self.subscribers[event_type].append(agent)
agent.event_bus = self
def unsubscribe(self, event_type: str, agent: EventDrivenAgent):
"""Unsubscribe agent from event type."""
if agent in self.subscribers[event_type]:
self.subscribers[event_type].remove(agent)
def publish(self, event: Event):
"""Publish event to all subscribers."""
self.event_log.append(event)
# Notify all subscribers
if event.event_type in self.subscribers:
for agent in self.subscribers[event.event_type]:
response = agent.handle_event(event)
# Optionally create follow-up event from response
if response and "event:" in response.lower():
# Agent can trigger new events in response
pass
def get_statistics(self) -> Dict[str, Any]:
"""Get event bus statistics."""
return {
"total_events": len(self.event_log),
"event_types": len(self.subscribers),
"total_subscriptions": sum(len(agents) for agents in self.subscribers.values()),
"events_by_type": {
event_type: sum(1 for e in self.event_log if e.event_type == event_type)
for event_type in self.subscribers.keys()
}
}
class EventDrivenPattern:
"""
Event-Driven Pattern - Asynchronous event-based communication.
Agents publish events to a central bus and subscribe to events
they're interested in, enabling loose coupling and reactivity.
"""
def __init__(self):
self.event_bus = EventBus()
def register_agent(self, agent: EventDrivenAgent):
"""Register agent with event bus."""
agent.event_bus = self.event_bus
def run(self, scenario: str) -> Dict[str, Any]:
"""Execute event-driven scenario."""
stats = self.event_bus.get_statistics()
return {
"pattern": "EventDriven",
"scenario": scenario,
**stats
}
# Example usage
if __name__ == "__main__":
print("=== Event-Driven Pattern Demo ===\n")
# Create pattern
pattern = EventDrivenPattern()
event_bus = pattern.event_bus
# Create agents
monitor = EventDrivenAgent("Monitor", "system monitor")
logger = EventDrivenAgent("Logger", "event logger")
alerter = EventDrivenAgent("Alerter", "alert manager")
# Register and subscribe
print("1. Setting up agents and subscriptions...")
pattern.register_agent(monitor)
pattern.register_agent(logger)
pattern.register_agent(alerter)
# Custom handler for critical events
def critical_handler(event: Event) -> str:
return f"ALERT: Critical event from {event.source}: {event.data}"
monitor.subscribe("system_event")
monitor.subscribe("user_action")
logger.subscribe("system_event")
logger.subscribe("user_action")
logger.subscribe("error")
alerter.subscribe("error", critical_handler)
# Publish events
print("\n2. Publishing events...")
event_bus.publish(Event("system_event", "System", "Server started"))
event_bus.publish(Event("user_action", "User1", "Login successful"))
event_bus.publish(Event("error", "System", "Database connection failed", priority=5))
# Statistics
stats = event_bus.get_statistics()
print(f"\n3. Event Bus Statistics:")
print(f" Total Events: {stats['total_events']}")
print(f" Event Types: {stats['event_types']}")
print(f" Subscriptions: {stats['total_subscriptions']}")
print(f" By Type: {stats['events_by_type']}")
result = pattern.run("System monitoring")
print(f"\n✓ Event-driven pattern enabled reactive agent communication!")
Agentic Architecture Pattern: Event Driven
Pattern 19: 19 Event Driven – Line-by-Line Explanation
Overview
Pattern 19: Event-Driven Pattern
Agents react to events in the system asynchronously. Uses event bus for decoupled event publishing and subscription.
Code Structure Analysis
File Information
- Pattern Number: 19
- Pattern Name: 19 Event Driven
- File Path:
patterns\communication\19_event_driven.py - Total Lines: 194
Detailed Code Explanation
Module Docstring
"""
Pattern 19: Event-Driven Pattern
Agents react to events in the system asynchronously.
Uses event bus for decoupled event publishing and subscription.
"""
Explanation: This module docstring describes the purpose and key features of the 19 Event Driven pattern.
Imports Section
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional, Callable: 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 langchain_core.prompts import ChatPromptTemplate: LangChain framework integration
– from langchain_openai import ChatOpenAI: LangChain framework integration
Classes
Event (Line 17)
Description: Event in the system.
EventDrivenAgent (Line 26)
Description: Agent that reacts to events.
Methods:
__init__ (Line 29)
def __init__(self, name, role, model_name):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– name: Identifier or label
– role: Parameter for role
– model_name: LLM model to use (e.g., “gpt-4”)
subscribe (Line 36)
def subscribe(self, event_type, handler):
"""Subscribe to event type."""
Purpose: Subscribe to event type.
Parameters:
– self: Instance reference
– event_type: Parameter for event_type
– handler: Parameter for handler
_default_handler (Line 47)
def _default_handler(self, event):
"""Default event handler using LLM."""
Purpose: Default event handler using LLM.
Parameters:
– self: Instance reference
– event: Parameter for event
handle_event (Line 61)
def handle_event(self, event):
"""Handle received event."""
Purpose: Handle received event.
Parameters:
– self: Instance reference
– event: Parameter for event
publish (Line 67)
def publish(self, event_type, data, priority):
"""Publish event to event bus."""
Purpose: Publish event to event bus.
Parameters:
– self: Instance reference
– event_type: Parameter for event_type
– data: Input data to process
– priority: Parameter for priority
EventBus (Line 79)
Description: Central event bus for event-driven communication.
Methods:
__init__ (Line 82)
def __init__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
subscribe (Line 86)
def subscribe(self, event_type, agent):
"""Subscribe agent to event type."""
Purpose: Subscribe agent to event type.
Parameters:
– self: Instance reference
– event_type: Parameter for event_type
– agent: Parameter for agent
unsubscribe (Line 92)
def unsubscribe(self, event_type, agent):
"""Unsubscribe agent from event type."""
Purpose: Unsubscribe agent from event type.
Parameters:
– self: Instance reference
– event_type: Parameter for event_type
– agent: Parameter for agent
publish (Line 97)
def publish(self, event):
"""Publish event to all subscribers."""
Purpose: Publish event to all subscribers.
Parameters:
– self: Instance reference
– event: Parameter for event
get_statistics (Line 110)
def get_statistics(self):
"""Get event bus statistics."""
Purpose: Get event bus statistics.
Parameters:
– self: Instance reference
EventDrivenPattern (Line 123)
Description: Event-Driven Pattern – Asynchronous event-based communication.
Agents publish events to a central bus and subscribe to events they’re interested in, enabling loose coupling and reactivity.
Methods:
__init__ (Line 131)
def __init__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
register_agent (Line 134)
def register_agent(self, agent):
"""Register agent with event bus."""
Purpose: Register agent with event bus.
Parameters:
– self: Instance reference
– agent: Parameter for agent
run (Line 138)
def run(self, scenario):
"""Execute event-driven scenario."""
Purpose: Execute event-driven scenario.
Parameters:
– self: Instance reference
– scenario: Parameter for scenario
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 = Event(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