"""
Pattern 121: Peer-to-Peer Pattern

Decentralized peer-to-peer agent network where agents communicate directly.
Implements P2P communication with LangChain for agent logic and message passing.
"""

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Any, List, Optional, Set
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate


class PeerStatus(Enum):
    """Status of peer in network."""
    ONLINE = "online"
    OFFLINE = "offline"
    BUSY = "busy"


@dataclass
class Message:
    """Message between peers."""
    message_id: str
    from_peer: str
    to_peer: str
    content: str
    message_type: str = "direct"  # direct, broadcast, query, response
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class PeerAgent:
    """Peer agent in P2P network."""
    peer_id: str
    name: str
    capabilities: List[str]
    status: PeerStatus = PeerStatus.ONLINE
    known_peers: Set[str] = field(default_factory=set)
    message_queue: List[Message] = field(default_factory=list)
    processed_messages: int = 0


class PeerToPeerNetwork:
    """Decentralized peer-to-peer agent network."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.peers: Dict[str, PeerAgent] = {}
        self.message_history: List[Message] = []
        
    def register_peer(self, peer: PeerAgent):
        """Register a new peer in the network."""
        self.peers[peer.peer_id] = peer
        
        # Announce to other peers
        for existing_peer_id in list(self.peers.keys()):
            if existing_peer_id != peer.peer_id:
                peer.known_peers.add(existing_peer_id)
                self.peers[existing_peer_id].known_peers.add(peer.peer_id)
    
    def send_message(self, from_peer_id: str, to_peer_id: str, content: str, message_type: str = "direct"):
        """Send message from one peer to another."""
        if from_peer_id not in self.peers or to_peer_id not in self.peers:
            return False
        
        message = Message(
            message_id=f"msg-{len(self.message_history)}",
            from_peer=from_peer_id,
            to_peer=to_peer_id,
            content=content,
            message_type=message_type
        )
        
        self.peers[to_peer_id].message_queue.append(message)
        self.message_history.append(message)
        return True
    
    def broadcast(self, from_peer_id: str, content: str):
        """Broadcast message to all peers."""
        if from_peer_id not in self.peers:
            return
        
        for peer_id in self.peers:
            if peer_id != from_peer_id:
                self.send_message(from_peer_id, peer_id, content, "broadcast")
    
    def process_peer_messages(self, peer_id: str) -> List[Dict[str, Any]]:
        """Process messages for a specific peer."""
        if peer_id not in self.peers:
            return []
        
        peer = self.peers[peer_id]
        results = []
        
        while peer.message_queue:
            message = peer.message_queue.pop(0)
            
            # Use LLM to generate response
            response_content = self._generate_peer_response(peer, message)
            
            # Send response if needed
            if message.message_type in ["query", "direct"]:
                self.send_message(peer_id, message.from_peer, response_content, "response")
            
            peer.processed_messages += 1
            
            results.append({
                "message_id": message.message_id,
                "from": message.from_peer,
                "type": message.message_type,
                "content": message.content,
                "response": response_content
            })
        
        return results
    
    def _generate_peer_response(self, peer: PeerAgent, message: Message) -> str:
        """Generate response using LLM."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are {peer.name} (ID: {peer.peer_id}), a peer in a P2P network.
Your capabilities: {', '.join(peer.capabilities)}
Known peers: {len(peer.known_peers)}"""),
            ("human", """Message from {from_peer}:
{content}

Type: {message_type}

Provide a helpful response:""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "from_peer": message.from_peer,
            "content": message.content,
            "message_type": message.message_type
        })
        
        return str(response.content)
    
    def get_network_status(self) -> Dict[str, Any]:
        """Get status of the P2P network."""
        return {
            "total_peers": len(self.peers),
            "online_peers": len([p for p in self.peers.values() if p.status == PeerStatus.ONLINE]),
            "total_messages": len(self.message_history),
            "peers": [
                {
                    "peer_id": p.peer_id,
                    "name": p.name,
                    "status": p.status.value,
                    "known_peers": len(p.known_peers),
                    "processed_messages": p.processed_messages,
                    "queued_messages": len(p.message_queue)
                }
                for p in self.peers.values()
            ]
        }


class PeerToPeerPattern:
    """Complete P2P pattern implementation."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.network = PeerToPeerNetwork(model_name=model_name)
        
    def run(self, scenario: str = "data_sharing") -> Dict[str, Any]:
        """Run P2P scenario."""
        # Create peers
        peers = [
            PeerAgent("peer-1", "DataCollector", ["data_collection", "storage"]),
            PeerAgent("peer-2", "DataProcessor", ["processing", "analysis"]),
            PeerAgent("peer-3", "DataVisualizer", ["visualization", "reporting"]),
            PeerAgent("peer-4", "QualityChecker", ["validation", "quality_check"])
        ]
        
        for peer in peers:
            self.network.register_peer(peer)
        
        # Simulate P2P interactions
        self.network.broadcast("peer-1", "I have collected new data. Who can process it?")
        
        # Process messages
        for peer_id in list(self.network.peers.keys()):
            self.network.process_peer_messages(peer_id)
        
        # Peer 2 requests data
        self.network.send_message("peer-2", "peer-1", "Send me the data for processing", "query")
        self.network.process_peer_messages("peer-1")
        
        # Peer 3 asks for results
        self.network.send_message("peer-3", "peer-2", "Are the processing results ready?", "query")
        self.network.process_peer_messages("peer-2")
        
        return self.network.get_network_status()


def main():
    """Example usage of Peer-to-Peer Pattern."""
    print("=" * 80)
    print("Peer-to-Peer Pattern Demo")
    print("=" * 80)
    
    pattern = PeerToPeerPattern()
    
    print("\n🌐 Starting P2P Network...")
    result = pattern.run("data_sharing")
    
    print(f"\n📊 Network Status:")
    print(f"Total Peers: {result['total_peers']}")
    print(f"Online Peers: {result['online_peers']}")
    print(f"Total Messages: {result['total_messages']}")
    
    print(f"\n👥 Peer Details:")
    for peer in result['peers']:
        print(f"\n  {peer['name']} ({peer['peer_id']})")
        print(f"    Status: {peer['status']}")
        print(f"    Known Peers: {peer['known_peers']}")
        print(f"    Processed: {peer['processed_messages']} messages")
        print(f"    Queued: {peer['queued_messages']} messages")


if __name__ == "__main__":
    main()

    

Agentic Architecture Pattern: Peer To Peer

Pattern 121: 121 Peer To Peer Pattern – Line-by-Line Explanation

Overview

Pattern 121: Peer-to-Peer Pattern

Decentralized network where agents are equals, communicating directly with each other without a central coordinator.

Code Structure Analysis

File Information

• Pattern Number: 121
• Pattern Name: 121 Peer To Peer Pattern
• File Path: patterns\distributed_agent\121_peer_to_peer_pattern.py
• Total Lines: 163

Detailed Code Explanation

Module Docstring

"""
Pattern 121: Peer-to-Peer Pattern

Decentralized network where agents are equals, communicating directly
with each other without a central coordinator.
"""

Explanation: This module docstring describes the purpose and key features of the 121 Peer To Peer Pattern pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
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, Set: 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

Peer (Line 16)

Description: Peer agent in P2P network.

Message (Line 26)

Description: P2P message.

PeerToPeerPattern (Line 34)

Description: Peer-to-Peer Pattern – Decentralized agent communication.

Methods:

__init__ (Line 39)

def __init__(self, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – model_name: LLM model to use (e.g., “gpt-4”)

add_peer (Line 44)

def add_peer(self, peer):
    """Add a peer to the network."""

Purpose: Add a peer to the network.

Parameters: – self: Instance reference – peer: Parameter for peer

connect_peers (Line 48)

def connect_peers(self, peer1_id, peer2_id):
    """Establish bidirectional connection between peers."""

Purpose: Establish bidirectional connection between peers.

Parameters: – self: Instance reference – peer1_id: Parameter for peer1_id – peer2_id: Parameter for peer2_id

send_message (Line 54)

def send_message(self, from_peer, to_peer, content):
    """Send message from one peer to another."""

Purpose: Send message from one peer to another.

Parameters: – self: Instance reference – from_peer: Parameter for from_peer – to_peer: Parameter for to_peer – content: Parameter for content

broadcast (Line 73)

def broadcast(self, from_peer, content):
    """Broadcast message to all connected peers."""

Purpose: Broadcast message to all connected peers.

Parameters: – self: Instance reference – from_peer: Parameter for from_peer – content: Parameter for content

run (Line 87)

def run(self, scenario):
    """Execute P2P scenario."""

Purpose: Execute P2P scenario.

Parameters: – self: Instance reference – scenario: Parameter for scenario

get_statistics (Line 105)

def get_statistics(self):
    """Get network statistics."""

Purpose: Get network statistics.

Parameters: – self: Instance reference

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 = Peer(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