"""
Pattern 121: Event-Driven Architecture

Description:
    Agents react to events in the system using pub-sub patterns and
    event buses for loose coupling and scalability.

Components:
    - Event bus
    - Publishers
    - Subscribers
    - Event handlers

Use Cases:
    - Real-time systems
    - Microservices
    - Reactive applications

LangChain Implementation:
    Implements event-driven pattern with event handlers and async processing.
"""

import os
from typing import List, Dict, Any, Callable
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class EventType(Enum):
    """Types of events in the system."""
    USER_ACTION = "user_action"
    SYSTEM_STATE = "system_state"
    DATA_UPDATE = "data_update"
    ERROR = "error"
    NOTIFICATION = "notification"


@dataclass
class Event:
    """Represents an event in the system."""
    id: str
    type: EventType
    source: str
    data: Dict[str, Any]
    timestamp: str
    priority: int = 5


class EventBus:
    """Central event bus for publish-subscribe pattern."""
    
    def __init__(self):
        self.subscribers: Dict[EventType, List[Callable]] = {}
        self.event_history: List[Event] = []
        
    def subscribe(self, event_type: EventType, handler: Callable):
        """Subscribe a handler to an event type."""
        if event_type not in self.subscribers:
            self.subscribers[event_type] = []
        self.subscribers[event_type].append(handler)
        print(f"✓ Subscribed handler to {event_type.value} events")
    
    def publish(self, event: Event):
        """Publish an event to all subscribers."""
        print(f"\n📤 Publishing event: {event.type.value} from {event.source}")
        print(f"   Data: {event.data}")
        
        self.event_history.append(event)
        
        # Notify subscribers
        if event.type in self.subscribers:
            for handler in self.subscribers[event.type]:
                try:
                    handler(event)
                except Exception as e:
                    print(f"   ⚠ Handler error: {e}")
    
    def get_event_count(self, event_type: EventType = None) -> int:
        """Get count of events."""
        if event_type:
            return sum(1 for e in self.event_history if e.type == event_type)
        return len(self.event_history)


class EventDrivenAgent:
    """Agent that reacts to events."""
    
    def __init__(self, name: str, event_bus: EventBus, model_name: str = "gpt-4"):
        self.name = name
        self.event_bus = event_bus
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.handled_events = []
        
    def handle_event(self, event: Event):
        """Handle an incoming event."""
        print(f"   📥 {self.name} received event")
        
        # Process event with LLM
        response_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an event-driven agent named {agent_name}.
React appropriately to the event based on its type and data."""),
            ("user", """Event Type: {event_type}
Source: {source}
Data: {data}

Provide your response action:""")
        ])
        
        chain = response_prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "agent_name": self.name,
            "event_type": event.type.value,
            "source": event.source,
            "data": str(event.data)
        })
        
        print(f"   🔧 {self.name} action: {response[:100]}...")
        
        self.handled_events.append({
            "event_id": event.id,
            "response": response
        })
        
        # Potentially publish new events
        if "notify" in response.lower():
            self.publish_notification(event)
    
    def publish_notification(self, original_event: Event):
        """Publish a notification event."""
        notification_event = Event(
            id=f"notif_{len(self.event_bus.event_history)}",
            type=EventType.NOTIFICATION,
            source=self.name,
            data={"message": f"Processed event {original_event.id}"},
            timestamp=datetime.now().isoformat()
        )
        self.event_bus.publish(notification_event)


class EventAnalyzer:
    """Analyzes event patterns."""
    
    def __init__(self, event_bus: EventBus, model_name: str = "gpt-4"):
        self.event_bus = event_bus
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
    
    def analyze_patterns(self) -> str:
        """Analyze event patterns using LLM."""
        events_summary = "\n".join([
            f"{e.type.value}: {e.source} -> {e.data}"
            for e in self.event_bus.event_history[-20:]
        ])
        
        analysis_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze these event patterns and identify trends."),
            ("user", """Recent Events:
{events}

Provide:
1. Event frequency patterns
2. Common event chains
3. Potential issues
4. Optimization suggestions""")
        ])
        
        chain = analysis_prompt | self.llm | StrOutputParser()
        return chain.invoke({"events": events_summary})


def demonstrate_event_driven_architecture():
    """Demonstrate event-driven architecture pattern."""
    print("=== Event-Driven Architecture Pattern ===\n")
    
    # Create event bus
    event_bus = EventBus()
    
    # Create agents
    print("1. Creating Event-Driven Agents")
    print("-" * 50)
    
    monitoring_agent = EventDrivenAgent("MonitoringAgent", event_bus)
    analytics_agent = EventDrivenAgent("AnalyticsAgent", event_bus)
    notification_agent = EventDrivenAgent("NotificationAgent", event_bus)
    
    # Subscribe agents to events
    print("\n2. Subscribing to Events")
    print("-" * 50)
    
    event_bus.subscribe(EventType.USER_ACTION, monitoring_agent.handle_event)
    event_bus.subscribe(EventType.USER_ACTION, analytics_agent.handle_event)
    
    event_bus.subscribe(EventType.SYSTEM_STATE, monitoring_agent.handle_event)
    
    event_bus.subscribe(EventType.ERROR, monitoring_agent.handle_event)
    event_bus.subscribe(EventType.ERROR, notification_agent.handle_event)
    
    event_bus.subscribe(EventType.NOTIFICATION, notification_agent.handle_event)
    
    print()
    
    # Publish events
    print("3. Publishing Events")
    print("-" * 50)
    
    # User action event
    event_bus.publish(Event(
        id="evt_001",
        type=EventType.USER_ACTION,
        source="WebApp",
        data={"action": "login", "user_id": "user123"},
        timestamp=datetime.now().isoformat(),
        priority=5
    ))
    
    # System state event
    event_bus.publish(Event(
        id="evt_002",
        type=EventType.SYSTEM_STATE,
        source="SystemMonitor",
        data={"cpu": 75, "memory": 60, "status": "healthy"},
        timestamp=datetime.now().isoformat(),
        priority=3
    ))
    
    # Data update event
    event_bus.publish(Event(
        id="evt_003",
        type=EventType.DATA_UPDATE,
        source="Database",
        data={"table": "users", "operation": "insert", "count": 5},
        timestamp=datetime.now().isoformat(),
        priority=4
    ))
    
    # Error event
    event_bus.publish(Event(
        id="evt_004",
        type=EventType.ERROR,
        source="APIGateway",
        data={"error": "RateLimitExceeded", "endpoint": "/api/data"},
        timestamp=datetime.now().isoformat(),
        priority=9
    ))
    
    # Another user action
    event_bus.publish(Event(
        id="evt_005",
        type=EventType.USER_ACTION,
        source="MobileApp",
        data={"action": "purchase", "amount": 99.99},
        timestamp=datetime.now().isoformat(),
        priority=7
    ))
    
    # Analyze event patterns
    print("\n4. Analyzing Event Patterns")
    print("-" * 50)
    
    analyzer = EventAnalyzer(event_bus)
    analysis = analyzer.analyze_patterns()
    print(analysis)
    
    # Statistics
    print("\n5. Event Statistics")
    print("-" * 50)
    print(f"Total events: {event_bus.get_event_count()}")
    print(f"User actions: {event_bus.get_event_count(EventType.USER_ACTION)}")
    print(f"System states: {event_bus.get_event_count(EventType.SYSTEM_STATE)}")
    print(f"Errors: {event_bus.get_event_count(EventType.ERROR)}")
    print(f"Notifications: {event_bus.get_event_count(EventType.NOTIFICATION)}")
    
    print(f"\nMonitoring agent handled: {len(monitoring_agent.handled_events)} events")
    print(f"Analytics agent handled: {len(analytics_agent.handled_events)} events")
    print(f"Notification agent handled: {len(notification_agent.handled_events)} events")
    
    print("\n=== Summary ===")
    print("Event-driven architecture demonstrated with:")
    print("- Event bus (pub-sub pattern)")
    print("- Multiple event types")
    print("- Event subscribers and handlers")
    print("- Event-driven agents")
    print("- Event chain reactions")
    print("- Pattern analysis")
    print("- Loose coupling between components")


if __name__ == "__main__":
    demonstrate_event_driven_architecture()

    

Agentic Design Pattern: Event Driven Architecture

Pattern 121: Event Driven Architecture – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Event Driven Architecture pattern implementation.

---

Pattern Description

Pattern 121: Event-Driven Architecture

Description: Agents react to events in the system using pub-sub patterns and event buses for loose coupling and scalability.

Components: – Event bus – Publishers – Subscribers – Event handlers

Use Cases: – Real-time systems – Microservices – Reactive applications

LangChain Implementation: Implements event-driven pattern with event handlers and async processing.

---

File Statistics

• Total Lines: 297
• Classes: 5
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 121: Event-Driven Architecture

Description:
    Agents react to events in the system using pub-sub patterns and
    event buses for loose coupling and scalability.

Components:
    - Event bus
    - Publishers
    - Subscribers
    - Event handlers

Use Cases:
    - Real-time systems
    - Microservices
    - Reactive applications

LangChain Implementation:
    Implements event-driven pattern with event handlers and async processing.
"""

Purpose: This module implements the Event Driven Architecture pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Callable
import typing.Dict
import typing.List

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: EventType

Lines 37-43

Inherits from: Enum

Purpose:

Types of events in the system.

---

Class: Event

Lines 47-54

Purpose:

Represents an event in the system.

---

Class: EventBus

Lines 57-90

Purpose:

Central event bus for publish-subscribe pattern.

Methods:

__init__(self)

Lines 60-62

Purpose: Initialize the class instance with required parameters.

Parameters:

subscribe(self, event_type, handler)

Lines 64-69

Purpose:

Subscribe a handler to an event type.

Parameters: – event_type: Event type – handler: Handler

publish(self, event)

Lines 71-84

Purpose:

Publish an event to all subscribers.

Parameters: – event: Event

get_event_count(self, event_type)

Lines 86-90

Purpose:

Get count of events.

Parameters: – event_type: Event type

Returns: int

---

Class: EventDrivenAgent

Lines 93-145

Purpose:

Agent that reacts to events.

Methods:

__init__(self, name, event_bus, model_name)

Lines 96-100

Purpose: Initialize the class instance with required parameters.

Parameters: – name: Name – event_bus: Event bus – model_name: Language model instance

handle_event(self, event)

Lines 102-134

Purpose:

Handle an incoming event.

Parameters: – event: Event

publish_notification(self, original_event)

Lines 136-145

Purpose:

Publish a notification event.

Parameters: – original_event: Original event

---

Class: EventAnalyzer

Lines 148-175

Purpose:

Analyzes event patterns.

Methods:

__init__(self, event_bus, model_name)

Lines 151-153

Purpose: Initialize the class instance with required parameters.

Parameters: – event_bus: Event bus – model_name: Language model instance

analyze_patterns(self)

Lines 155-175

Purpose:

Analyze event patterns using LLM.

Parameters:

Returns: str

---

Functions

demonstrate_event_driven_architecture()

Lines 178-292

Purpose:

Demonstrate event-driven architecture pattern.

---

Key Concepts Summary

Pattern Overview

The Event Driven Architecture pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 120
• Pattern 122

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Event Driven Architecture pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 121_event_driven_architecture.py
Total Lines: 297
Pattern Category: AI Agent Design Pattern #121

---

Generated from actual implementation code
For educational and reference purposes