"""
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
Leave a Reply