aishastr.live

Agentic Architecture Pattern: Event Driven

Written by

admin

in


    """
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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts