Coverage for agentos/comm/layer.py: 34%
122 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 12:29 +0800
1"""
2Communication Layer for NexusAgent.
4Provides multiple communication patterns for multi-agent systems:
5- Blackboard: Shared memory space
6- EventBus: Publish-subscribe events
7- Mailbox: Direct point-to-point messaging
8"""
10from __future__ import annotations
12import asyncio
13import time
14import uuid
15from collections.abc import Callable
16from dataclasses import dataclass, field
17from typing import Any
20@dataclass
21class Message:
22 """
23 Communication message.
25 Attributes:
26 id: Unique identifier
27 sender: Sender name
28 receiver: Receiver name
29 content: Message content
30 metadata: Additional metadata
31 timestamp: Message timestamp
32 """
34 id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
35 sender: str = ""
36 receiver: str | None = None
37 content: Any = None
38 metadata: dict[str, Any] = field(default_factory=dict)
39 timestamp: float = field(default_factory=time.time)
41 def to_dict(self) -> dict[str, Any]:
42 """Convert to dict."""
43 return {
44 "id": self.id,
45 "sender": self.sender,
46 "receiver": self.receiver,
47 "content": self.content,
48 "metadata": self.metadata,
49 "timestamp": self.timestamp,
50 }
53class Blackboard:
54 """
55 Shared memory space for agents.
57 Agents can read/write to a shared blackboard.
58 Useful for collaborative problem solving.
60 Usage:
61 blackboard = Blackboard()
62 blackboard.write("agent1", "status", "working")
63 status = blackboard.read("agent1", "status")
64 """
66 def __init__(self):
67 """Initialize blackboard."""
68 self._data: dict[str, dict[str, Any]] = {}
69 self._history: list[dict[str, Any]] = []
71 def write(
72 self,
73 agent_name: str,
74 key: str,
75 value: Any,
76 ) -> None:
77 """
78 Write to blackboard.
80 Args:
81 agent_name: Agent name
82 key: Data key
83 value: Data value
84 """
85 if agent_name not in self._data:
86 self._data[agent_name] = {}
88 self._data[agent_name][key] = value
90 # Record in history
91 self._history.append(
92 {
93 "agent": agent_name,
94 "key": key,
95 "value": value,
96 "timestamp": time.time(),
97 }
98 )
100 def read(
101 self,
102 agent_name: str,
103 key: str,
104 default: Any = None,
105 ) -> Any:
106 """
107 Read from blackboard.
109 Args:
110 agent_name: Agent name
111 key: Data key
112 default: Default value if not found
114 Returns:
115 Data value
116 """
117 if agent_name not in self._data:
118 return default
120 return self._data[agent_name].get(key, default)
122 def read_all(self, key: str) -> dict[str, Any]:
123 """
124 Read key from all agents.
126 Args:
127 key: Data key
129 Returns:
130 Dict of agent_name -> value
131 """
132 return {agent: data.get(key) for agent, data in self._data.items() if key in data}
134 def get_agent_data(self, agent_name: str) -> dict[str, Any]:
135 """
136 Get all data for an agent.
138 Args:
139 agent_name: Agent name
141 Returns:
142 Dict of key -> value
143 """
144 return self._data.get(agent_name, {}).copy()
146 def get_history(
147 self,
148 agent_name: str | None = None,
149 limit: int = 100,
150 ) -> list[dict[str, Any]]:
151 """
152 Get write history.
154 Args:
155 agent_name: Filter by agent (None = all)
156 limit: Max results
158 Returns:
159 List of history entries
160 """
161 history = self._history
163 if agent_name:
164 history = [h for h in history if h["agent"] == agent_name]
166 return history[-limit:]
168 def clear(self, agent_name: str | None = None) -> None:
169 """
170 Clear blackboard.
172 Args:
173 agent_name: Clear specific agent (None = all)
174 """
175 if agent_name:
176 self._data.pop(agent_name, None)
177 else:
178 self._data.clear()
181class EventBus:
182 """
183 Publish-subscribe event system.
185 Agents can subscribe to events and publish events.
186 Useful for event-driven architectures.
188 Usage:
189 bus = EventBus()
191 # Subscribe
192 bus.subscribe("task_completed", callback)
194 # Publish
195 bus.publish("task_completed", {"task_id": "123"})
196 """
198 def __init__(self):
199 """Initialize event bus."""
200 self._subscribers: dict[str, list[Callable[[Any], None]]] = {}
201 self._history: list[dict[str, Any]] = []
203 def subscribe(
204 self,
205 event_type: str,
206 callback: Callable[[Any], None],
207 ) -> None:
208 """
209 Subscribe to an event.
211 Args:
212 event_type: Event type
213 callback: Callback function
214 """
215 if event_type not in self._subscribers:
216 self._subscribers[event_type] = []
218 self._subscribers[event_type].append(callback)
220 def unsubscribe(
221 self,
222 event_type: str,
223 callback: Callable[[Any], None],
224 ) -> bool:
225 """
226 Unsubscribe from an event.
228 Args:
229 event_type: Event type
230 callback: Callback function
232 Returns:
233 True if unsubscribed, False if not found
234 """
235 if event_type not in self._subscribers:
236 return False
238 if callback in self._subscribers[event_type]:
239 self._subscribers[event_type].remove(callback)
240 return True
242 return False
244 def publish(
245 self,
246 event_type: str,
247 data: Any = None,
248 sender: str = "",
249 ) -> int:
250 """
251 Publish an event.
253 Args:
254 event_type: Event type
255 data: Event data
256 sender: Sender name
258 Returns:
259 Number of subscribers notified
260 """
261 # Record in history
262 self._history.append(
263 {
264 "event_type": event_type,
265 "data": data,
266 "sender": sender,
267 "timestamp": time.time(),
268 }
269 )
271 # Notify subscribers
272 subscribers = self._subscribers.get(event_type, [])
273 for callback in subscribers:
274 try:
275 callback(data)
276 except Exception:
277 pass # Don't let one callback break others
279 return len(subscribers)
281 async def publish_async(
282 self,
283 event_type: str,
284 data: Any = None,
285 sender: str = "",
286 ) -> int:
287 """
288 Publish an event asynchronously.
290 Args:
291 event_type: Event type
292 data: Event data
293 sender: Sender name
295 Returns:
296 Number of subscribers notified
297 """
298 # Record in history
299 self._history.append(
300 {
301 "event_type": event_type,
302 "data": data,
303 "sender": sender,
304 "timestamp": time.time(),
305 }
306 )
308 # Notify subscribers
309 subscribers = self._subscribers.get(event_type, [])
310 tasks = []
311 for callback in subscribers:
312 if asyncio.iscoroutinefunction(callback):
313 tasks.append(callback(data))
314 else:
315 callback(data)
317 if tasks:
318 await asyncio.gather(*tasks, return_exceptions=True)
320 return len(subscribers)
322 def get_history(
323 self,
324 event_type: str | None = None,
325 limit: int = 100,
326 ) -> list[dict[str, Any]]:
327 """
328 Get event history.
330 Args:
331 event_type: Filter by event type (None = all)
332 limit: Max results
334 Returns:
335 List of history entries
336 """
337 history = self._history
339 if event_type:
340 history = [h for h in history if h["event_type"] == event_type]
342 return history[-limit:]
344 def clear(self) -> None:
345 """Clear event bus."""
346 self._subscribers.clear()
347 self._history.clear()
350class Mailbox:
351 """
352 Point-to-point messaging system.
354 Agents have mailboxes and can send/receive messages.
355 Useful for direct communication.
357 Usage:
358 mailbox = Mailbox()
359 mailbox.send("agent1", "agent2", "Hello")
360 messages = mailbox.receive("agent2")
361 """
363 def __init__(self):
364 """Initialize mailbox system."""
365 self._mailboxes: dict[str, list[Message]] = {}
366 self._sent: list[Message] = []
368 def send(self, sender: str, receiver: str, content: Any, **metadata) -> Message:
369 """
370 Send a message.
372 Args:
373 sender: Sender name
374 receiver: Receiver name
375 content: Message content
376 **metadata: Additional metadata
378 Returns:
379 Created Message
380 """
381 message = Message(
382 sender=sender,
383 receiver=receiver,
384 content=content,
385 metadata=metadata,
386 )
388 # Add to receiver's mailbox
389 if receiver not in self._mailboxes:
390 self._mailboxes[receiver] = []
392 self._mailboxes[receiver].append(message)
393 self._sent.append(message)
395 return message
397 def receive(
398 self,
399 receiver: str,
400 limit: int = 100,
401 ) -> list[Message]:
402 """
403 Receive messages.
405 Args:
406 receiver: Receiver name
407 limit: Max messages
409 Returns:
410 List of messages
411 """
412 messages = self._mailboxes.get(receiver, [])
413 return messages[:limit]
415 def receive_and_clear(
416 self,
417 receiver: str,
418 limit: int = 100,
419 ) -> list[Message]:
420 """
421 Receive and clear messages.
423 Args:
424 receiver: Receiver name
425 limit: Max messages
427 Returns:
428 List of messages
429 """
430 messages = self._mailboxes.get(receiver, [])[:limit]
431 self._mailboxes[receiver] = self._mailboxes.get(receiver, [])[limit:]
432 return messages
434 def get_sent(
435 self,
436 sender: str | None = None,
437 limit: int = 100,
438 ) -> list[Message]:
439 """
440 Get sent messages.
442 Args:
443 sender: Filter by sender (None = all)
444 limit: Max results
446 Returns:
447 List of messages
448 """
449 sent = self._sent
451 if sender:
452 sent = [m for m in sent if m.sender == sender]
454 return sent[-limit:]
456 def clear(self, receiver: str | None = None) -> None:
457 """
458 Clear mailboxes.
460 Args:
461 receiver: Clear specific receiver (None = all)
462 """
463 if receiver:
464 self._mailboxes.pop(receiver, None)
465 else:
466 self._mailboxes.clear()
469class CommunicationLayer:
470 """
471 Unified communication layer.
473 Combines Blackboard, EventBus, and Mailbox into
474 a single interface.
476 Usage:
477 comm = CommunicationLayer()
479 # Use blackboard
480 comm.blackboard.write("agent1", "status", "working")
482 # Use event bus
483 comm.event_bus.subscribe("task_completed", callback)
485 # Use mailbox
486 comm.mailbox.send("agent1", "agent2", "Hello")
487 """
489 def __init__(self):
490 """Initialize communication layer."""
491 self.blackboard = Blackboard()
492 self.event_bus = EventBus()
493 self.mailbox = Mailbox()
495 def clear(self) -> None:
496 """Clear all communication channels."""
497 self.blackboard.clear()
498 self.event_bus.clear()
499 self.mailbox.clear()