# architectures.py from enum import Enum from dataclasses import dataclass from typing import Literal class ArchitecturePattern(Enum): SUPERVISOR = "supervisor" # Central coordinator  HIERARCHICAL = "hierarchical" # Tree of supervisors  SWARM = "swarm" # Peer-to-peer collaboration  PIPELINE = "pipeline" # Sequential stages  MARKET = "market" # Auction-based task allocation  BLACKBOARD = "blackboard" # Shared knowledge space  HYBRID = "hybrid" # Combination  @dataclass class AgentSpec: name: str role: str capabilities: list[str] tools: list[str] model: str system_prompt: str # Pattern 1: Supervisor (Most Common) SUPERVISOR_SPEC = AgentSpec( name="supervisor", role="Coordinate specialists, decompose tasks, synthesize results", capabilities=["planning", "routing", "synthesis"], tools=["delegate", "aggregate"], model="gpt-4o", system_prompt="""You are a supervisor managing specialist agents. Break down user requests, delegate to appropriate specialists, and synthesize their outputs into a coherent response.""" ) SPECIALISTS = [ AgentSpec("researcher", "Information gathering", ["search", "read", "summarize"], ["web_search", "doc_lookup"], "gpt-4o-mini", "..."), AgentSpec("coder", "Code generation", ["write", "debug", "test"], ["code_exec", "lint"], "gpt-4o", "..."), AgentSpec("reviewer", "Quality assurance", ["review", "security", "performance"], ["static_analysis"], "gpt-4o", "..."), AgentSpec("writer", "Documentation", ["explain", "document", "summarize"], [], "gpt-4o-mini", "..."), ] # Pattern 2: Hierarchical (For complex orgs) HIERARCHICAL_SPECS = { "ceo": AgentSpec("ceo", "Strategic planning", ["strategy", "delegation"], [], "gpt-4o", "..."), "cto": AgentSpec("cto", "Technical leadership", ["architecture", "code_review"], [], "gpt-4o", "..."), "pm": AgentSpec("pm", "Project management", ["planning", "tracking"], [], "gpt-4o-mini", "..."), "dev_lead": AgentSpec("dev_lead", "Team lead", ["code_review", "mentoring"], [], "gpt-4o", "..."), "developers": [AgentSpec(f"dev_{i}", "Developer", ["coding", "testing"], [], "gpt-4o-mini", "...") for i in range(3)], "qa": AgentSpec("qa", "Quality assurance", ["testing", "validation"], [], "gpt-4o-mini", "..."), } # Pattern 3: Swarm (Peer-to-peer) SWARM_PROMPT = """You are part of a swarm of agents solving: {task} Other agents: {other_agents} Your specialization: {my_specialization} Communicate via shared messages. Propose actions, vote on decisions. Reach consensus before taking final action. Current state: {shared_state} Your proposal:""" 

TASK CHARACTERISTICS → ARCHITECTURE Clear decomposition, known specialists → SUPERVISOR Deep hierarchy, org-like structure → HIERARCHICAL Exploratory, creative, consensus needed → SWARM Sequential stages, clear handoffs → PIPELINE Resource allocation, competing bids → MARKET Complex reasoning, shared knowledge → BLACKBOARD Real-world complexity → HYBRID 

# communication.py from dataclasses import dataclass, field from datetime import datetime from typing import Any, Optional from enum import Enum import uuid class MessageType(Enum): TASK_REQUEST = "task_request" TASK_RESPONSE = "task_response" TASK_DELEGATE = "task_delegate" STATUS_UPDATE = "status_update" QUERY = "query" RESPONSE = "response" BROADCAST = "broadcast" CONSENSUS_PROPOSE = "consensus_propose" CONSENSUS_VOTE = "consensus_vote" CONSENSUS_RESULT = "consensus_result" @dataclass class AgentMessage: id: str = field(default_factory=lambda: str(uuid.uuid4())) type: MessageType = MessageType.TASK_REQUEST sender: str = "" recipient: str | list[str] = "" # Empty = broadcast  content: Any = None metadata: dict = field(default_factory=dict) timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat()) correlation_id: str = "" # For request-response pairing  reply_to: str = "" # Message ID being replied to  class MessageBus: """Central message bus for agent communication.""" def __init__(self): self.subscribers: dict[str, list[callable]] = {} self.history: list[AgentMessage] = [] def subscribe(self, agent_name: str, handler: callable): if agent_name not in self.subscribers: self.subscribers[agent_name] = [] self.subscribers[agent_name].append(handler) def publish(self, message: AgentMessage): self.history.append(message) recipients = message.recipient if isinstance(message.recipient, list) else [message.recipient] if not recipients or recipients == ["*"]: # Broadcast  for agent, handlers in self.subscribers.items(): for handler in handlers: handler(message) else: for recipient in recipients: if recipient in self.subscribers: for handler in self.subscribers[recipient]: handler(message) def request_response(self, sender: str, recipient: str, content: Any, timeout: float = 30) -> Any: """Synchronous request-response pattern.""" import asyncio correlation_id = str(uuid.uuid4()) future = asyncio.Future() def response_handler(msg: AgentMessage): if msg.correlation_id == correlation_id and msg.sender == recipient: future.set_result(msg.content) self.subscribe(f"__response_{correlation_id}", response_handler) msg = AgentMessage( type=MessageType.TASK_REQUEST, sender=sender, recipient=recipient, content=content, correlation_id=correlation_id ) self.publish(msg) try: return asyncio.wait_for(future, timeout=timeout) except asyncio.TimeoutError: raise TimeoutError(f"No response from {recipient}") 

# acl.py from dataclasses import dataclass from typing import Literal @dataclass class ACLMessage: """FIPA-ACL inspired message structure.""" performative: Literal[ "inform", "request", "query", "propose", "accept", "reject", "subscribe", "cancel", "confirm", "disconfirm", "failure" ] sender: str receiver: str | list[str] content: str language: str = "json" ontology: str = "default" protocol: str = "request-response" conversation_id: str = "" reply_with: str = "" in_reply_to: str = "" # Example usage: # request = ACLMessage( # performative="request", # sender="supervisor", # receiver="researcher", # content=json.dumps({"task": "Find papers on transformer architectures", "max_results": 10}), # conversation_id="conv-123", # reply_with="req-456" # ) 

# delegation.py from typing import Literal from pydantic import BaseModel class TaskDecomposition(BaseModel): subtasks: list[dict] # {"id": "", "description": "", "assigned_to": "", "dependencies": []}  execution_order: list[str] # Task IDs in order  DECOMPOSITION_PROMPT = """Decompose this task into subtasks for specialist agents. Task: {task} Available agents: {agents} Consider: - Dependencies between subtasks - Which agent is best for each - Parallel vs sequential execution - Context passing requirements Return JSON: { "subtasks": [ {"id": "1", "description": "...", "assigned_to": "researcher", "dependencies": []}, {"id": "2", "description": "...", "assigned_to": "coder", "dependencies": ["1"]} ], "execution_order": ["1", "2"] }""" 

# consensus.py import asyncio from dataclasses import dataclass from typing import Literal @dataclass class ConsensusProposal: id: str proposer: str proposal: dict votes: dict[str, Literal["yes", "no", "abstain"]] = field(default_factory=dict) threshold: float = 0.66 # 2/3 majority  deadline: float = 30.0 # seconds  class ConsensusManager: def __init__(self, message_bus: MessageBus): self.bus = message_bus self.active_proposals: dict[str, ConsensusProposal] = {} async def propose(self, proposer: str, proposal: dict, participants: list[str], threshold: float = 0.66) -> bool: proposal_id = str(uuid.uuid4()) cp = ConsensusProposal( id=proposal_id, proposer=proposer, proposal=proposal, threshold=threshold ) self.active_proposals[proposal_id] = cp # Broadcast proposal  for participant in participants: msg = AgentMessage( type=MessageType.CONSENSUS_PROPOSE, sender=proposer, recipient=participant, content={"proposal_id": proposal_id, "proposal": proposal}, correlation_id=proposal_id ) self.bus.publish(msg) # Wait for votes  await asyncio.sleep(cp.deadline) # Tally  yes_votes = sum(1 for v in cp.votes.values() if v == "yes") total = len(participants) result = yes_votes / total >= threshold # Broadcast result  result_msg = AgentMessage( type=MessageType.CONSENSUS_RESULT, sender="consensus", recipient="*", content={"proposal_id": proposal_id, "accepted": result, "votes": cp.votes} ) self.bus.publish(result_msg) del self.active_proposals[proposal_id] return result def vote(self, proposal_id: str, voter: str, vote: Literal["yes", "no", "abstain"]): if proposal_id in self.active_proposals: self.active_proposals[proposal_id].votes[voter] = vote 

# blackboard.py from typing import Any import asyncio from collections import defaultdict class Blackboard: """Shared knowledge space for multi-agent collaboration.""" def __init__(self): self.data: dict[str, Any] = {} self.subscribers: dict[str, list[callable]] = defaultdict(list) self.locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock) self.version: dict[str, int] = defaultdict(int) async def write(self, ...