Коллективное решение сложных задач
Blackboard — архитектурный паттерн, где несколько независимых «знаний» (knowledge sources) совместно решают задачу, взаимодействуя через общее хранилище — «доску» (blackboard), а «контроллер» управляет итерациями.
┌──────────────┐ triggers ┌──────────────────┐
│ Knowledge A │ ───────────────▶ │ │
└──────────────┘ │ │
┌──────────────┐ reads/writes │ BLACKBOARD │
│ Knowledge B │ ◀──────────────▶ │ (Shared State) │
└──────────────┘ │ │
┌──────────────┐ triggers │ │
│ Knowledge C │ ───────────────▶ └──────────────────┘
└──────────────┘ ▲
│ control
┌──────────────┐
│ Controller │
└──────────────┘
from dataclasses import dataclass, field
from typing import Any, Dict, List, Protocol
@dataclass
class Blackboard:
state: Dict[str, Any] = field(default_factory=dict)
log: List[str] = field(default_factory=list)
class KnowledgeSource(Protocol):
def applicable(self, bb: Blackboard) -> bool: ...
def apply(self, bb: Blackboard) -> None: ...
class Tokenize(KnowledgeSource):
def applicable(self, bb):
return 'text' in bb.state and 'tokens' not in bb.state
def apply(self, bb):
bb.state['tokens'] = bb.state['text'].split()
bb.log.append('Tokenize')
class Lower(KnowledgeSource):
def applicable(self, bb):
return 'tokens' in bb.state and 'lower' not in bb.state
def apply(self, bb):
bb.state['lower'] = [t.lower() for t in bb.state['tokens']]
bb.log.append('Lower')
class Unique(KnowledgeSource):
def applicable(self, bb):
return 'lower' in bb.state and 'unique' not in bb.state
def apply(self, bb):
bb.state['unique'] = sorted(set(bb.state['lower']))
bb.log.append('Unique')
class Controller:
def __init__(self, knowledge: List[KnowledgeSource]):
self.knowledge = knowledge
def run(self, bb: Blackboard, max_iters: int = 100):
for _ in range(max_iters):
fired = False
for ks in self.knowledge:
if ks.applicable(bb):
ks.apply(bb)
fired = True
if not fired:
break
bb = Blackboard(state={'text': 'Hello World Hello'})
Controller([Tokenize(), Lower(), Unique()]).run(bb)
print(bb.state['unique']) # ['hello', 'world']
print(bb.log) # ['Tokenize','Lower','Unique']