Blackboard Architecture
The blackboard pattern allows multiple agents to share and react to a common state.
Task
Build a thread-safe Blackboard that:
- Allows agents to write keyed values with optional TTL.
- Provides versioned reads with TTL expiry checks.
- Supports pub/sub callbacks when keys are updated.
- Maintains a history of all writes.
Constraints
- All operations must be thread-safe.
- Expired entries return
Noneon read. - Key patterns support
*wildcard for subscriptions. - Version increments on every write to same key.
Examples
Example 1:
Input:
bb = Blackboard()
bb.write('analysis', {'sentiment': 'positive'}, 'agent_a')
bb.read('analysis')Output:
{'sentiment': 'positive'}Explanation: Value written by agent_a and read back.
Starter Code
from typing import Any, Dict, List, Optional, Callable
from datetime import datetime
import threading
class BlackboardEntry:
def __init__(self, key: str, value: Any, author: str, ttl: Optional[float] = None):
self.key = key
self.value = value
self.author = author
self.timestamp = datetime.utcnow()
self.ttl = ttl # seconds until expiry, None = never
self.version = 1
class Blackboard:
def __init__(self):
self._data: Dict[str, BlackboardEntry] = {}
self._lock = threading.Lock()
self._subscribers: Dict[str, List[Callable]] = {}
self._history: List[Dict] = []
def write(self, key: str, value: Any, author: str, ttl: float = None) -> int:
# TODO: Thread-safe write, return new version number
pass
def read(self, key: str) -> Optional[Any]:
# TODO: Read with TTL check
pass
def subscribe(self, key_pattern: str, callback: Callable) -> None:
# TODO: Register callback triggered on key writes
pass
def get_snapshot(self) -> Dict:
pass
Python3
ReadyLines: 1Characters: 0
Ready