arbitration.md 14 KB
trixy_core.arbitration
Wakeword-Arbitration für Multi-Satellite-Entscheidung.
Wenn mehrere Satellites gleichzeitig ein Wakeword erkennen, entscheidet das Arbitration-System, welcher Satellite die Konversation führen soll.
Übersicht
from trixy_core.arbitration import (
# Candidate
ArbitrationCandidate, CandidateState,
# Window
ArbitrationWindow, WindowConfig, WindowState,
# Strategies
ArbitrationStrategy, LoudestStrategy, ConfidenceStrategy,
CombinedStrategy, ProximityStrategy,
# Arbitrator
WakewordArbitrator, ArbitrationConfig, ArbitrationResult,
)
Ablauf
┌─────────────────────────────────────────────────────────────────┐
│ Wakeword-Arbitration │
├─────────────────────────────────────────────────────────────────┤
│ │
│ t=0ms Küche sendet WakewordDetected(audio_level=0.7) │
│ → Arbitration-Fenster öffnet (1000ms Timer) │
│ │
│ t=50ms Wohnzimmer sendet WakewordDetected(audio_level=0.85) │
│ → Kandidat wird hinzugefügt │
│ │
│ t=1000ms Timer abgelaufen │
│ → Strategy.select(candidates) → Wohnzimmer gewinnt │
│ → WakewordSelected an Wohnzimmer │
│ → WakewordAbort an Küche │
│ → Conversation-Session wird gestartet │
│ │
└─────────────────────────────────────────────────────────────────┘
ArbitrationCandidate
Repräsentiert einen Satellite, der ein Wakeword erkannt hat.
from trixy_core.arbitration import ArbitrationCandidate, CandidateState
candidate = ArbitrationCandidate(
satellite_id="living_room",
room_id="wohnzimmer",
alias="Wohnzimmer",
wakeword_model="trixy",
confidence=0.92,
audio_level=0.85,
buffer_samples=16000,
)
# Eigenschaften
print(candidate.satellite_id) # "living_room"
print(candidate.audio_level) # 0.85 (normalisiert 0.0-1.0)
print(candidate.confidence) # 0.92 (normalisiert 0.0-1.0)
print(candidate.state) # CandidateState.PENDING
print(candidate.age_ms) # Millisekunden seit Erkennung
# Zustand
print(candidate.is_pending) # True
print(candidate.is_selected) # False
print(candidate.is_rejected) # False
# Zustandsänderung
candidate.select() # Markiert als ausgewählt
candidate.reject("Grund") # Markiert als abgelehnt
# Score (von Strategy gesetzt)
print(candidate.score) # 0.0-1.0
print(candidate.score_details) # {"audio_level": 0.85, ...}
# Serialisierung
data = candidate.to_dict()
CandidateState
from trixy_core.arbitration import CandidateState
class CandidateState(Enum):
PENDING = "pending" # Wartet auf Entscheidung
SELECTED = "selected" # Wurde ausgewählt
REJECTED = "rejected" # Wurde abgelehnt
TIMEOUT = "timeout" # Zeitüberschreitung
ArbitrationWindow
Zeitfenster für die Sammlung von Kandidaten.
from trixy_core.arbitration import ArbitrationWindow, WindowConfig
config = WindowConfig(
window_duration_ms=1000, # Gesamtdauer des Fensters
min_wait_ms=100, # Minimale Wartezeit
max_candidates=10, # Maximale Kandidaten
immediate_threshold=0.95, # Sofortige Entscheidung bei diesem Wert
)
window = ArbitrationWindow(
config=config,
on_complete=async_callback, # Wird bei Abschluss aufgerufen
)
# Fenster öffnen
await window.open()
# Kandidaten hinzufügen
candidate = ArbitrationCandidate(satellite_id="test", audio_level=0.8)
await window.add_candidate(candidate)
# Status
print(window.is_open) # True
print(window.candidate_count) # 1
print(window.duration_ms) # Zeit seit Öffnung
print(window.remaining_ms) # Verbleibende Zeit
# Kandidat abrufen
c = window.get_candidate("test")
has = window.has_candidate("test")
# Manuell schließen
candidates = await window.close(reason="manual")
# Zurücksetzen
await window.reset()
WindowState
from trixy_core.arbitration import WindowState
class WindowState(Enum):
CLOSED = "closed" # Kein aktives Fenster
OPEN = "open" # Fenster offen, sammelt Kandidaten
DECIDING = "deciding" # Entscheidung wird getroffen
COMPLETED = "completed" # Entscheidung abgeschlossen
Strategien
ArbitrationStrategy (Basis)
from trixy_core.arbitration import ArbitrationStrategy
class MyStrategy(ArbitrationStrategy):
@property
def name(self) -> str:
return "my_strategy"
def calculate_score(
self,
candidate: ArbitrationCandidate,
context: dict = None,
) -> float:
return candidate.audio_level * 0.7 + candidate.confidence * 0.3
def select(
self,
candidates: list[ArbitrationCandidate],
context: dict = None,
) -> ArbitrationCandidate | None:
valid = self.filter_valid(candidates)
if not valid:
return None
for c in valid:
c.score = self.calculate_score(c, context)
return max(valid, key=lambda c: c.score)
LoudestStrategy
Wählt den Satellite mit dem höchsten Audio-Level.
from trixy_core.arbitration import LoudestStrategy
strategy = LoudestStrategy(min_audio_level=0.1)
candidates = [
ArbitrationCandidate(satellite_id="quiet", audio_level=0.5),
ArbitrationCandidate(satellite_id="loud", audio_level=0.9),
]
winner = strategy.select(candidates)
# winner.satellite_id == "loud"
ConfidenceStrategy
Wählt den Satellite mit der höchsten Wakeword-Confidence.
from trixy_core.arbitration import ConfidenceStrategy
strategy = ConfidenceStrategy(min_confidence=0.5)
candidates = [
ArbitrationCandidate(satellite_id="low", confidence=0.6),
ArbitrationCandidate(satellite_id="high", confidence=0.95),
]
winner = strategy.select(candidates)
# winner.satellite_id == "high"
CombinedStrategy
Kombiniert mehrere Faktoren mit Gewichtungen.
from trixy_core.arbitration import CombinedStrategy
from trixy_core.arbitration.strategies.combined import CombinedWeights
weights = CombinedWeights(
audio_level=0.6, # 60% Audio-Level
confidence=0.3, # 30% Confidence
recency=0.1, # 10% Frühe Erkennung
)
strategy = CombinedStrategy(
weights=weights,
min_audio_level=0.1,
min_confidence=0.5,
)
candidates = [
ArbitrationCandidate(
satellite_id="balanced",
audio_level=0.8,
confidence=0.8,
),
ArbitrationCandidate(
satellite_id="loud_only",
audio_level=0.95,
confidence=0.5,
),
]
winner = strategy.select(candidates)
# "balanced" gewinnt mit höherem Gesamtscore
ProximityStrategy
Berücksichtigt Raum-Relevanz und Kontext.
from trixy_core.arbitration import ProximityStrategy
from trixy_core.arbitration.strategies.proximity import (
ProximityWeights,
RoomRelevance,
)
room_relevance = RoomRelevance(
# Räume die für bestimmte Aktionen bevorzugt werden
room_priorities={
"music": ["living_room", "bedroom"],
"alarm": ["bedroom", "kitchen"],
},
# Letzter aktiver Raum bekommt Bonus
last_active_bonus=0.15,
# Benachbarte Räume
adjacent_rooms={
"living_room": ["hallway", "kitchen"],
"bedroom": ["hallway", "bathroom"],
},
adjacent_bonus=0.1,
)
strategy = ProximityStrategy(room_relevance=room_relevance)
context = {
"last_active_room": "living_room",
"last_active_satellite": "living_room_1",
}
candidates = [
ArbitrationCandidate(
satellite_id="kitchen",
room_id="kitchen",
audio_level=0.8,
),
ArbitrationCandidate(
satellite_id="living",
room_id="living_room",
audio_level=0.75,
),
]
winner = strategy.select(candidates, context)
# "living" könnte gewinnen wegen last_active_bonus
WakewordArbitrator
Hauptservice für Wakeword-Arbitration.
from trixy_core.arbitration import (
WakewordArbitrator,
ArbitrationConfig,
ArbitrationResult,
ArbitrationCandidate,
)
config = ArbitrationConfig(
# Zeitfenster
window_duration_ms=1000,
min_wait_ms=100,
max_candidates=10,
# Schwellenwerte
immediate_threshold=0.95,
min_audio_level=0.1,
min_confidence=0.5,
# Strategie
strategy="combined", # "loudest", "confidence", "combined", "proximity"
# Combined-Strategie Gewichtungen
audio_weight=0.6,
confidence_weight=0.3,
recency_weight=0.1,
# Verhalten
auto_start_conversation=True,
notify_rejected=True,
)
arbitrator = WakewordArbitrator(config)
Kandidaten hinzufügen
candidate = ArbitrationCandidate(
satellite_id="living_room",
room_id="wohnzimmer",
audio_level=0.85,
confidence=0.9,
)
await arbitrator.add_candidate(candidate)
Entscheidungs-Callback
async def on_decision(result: ArbitrationResult) -> None:
if result.success:
print(f"Gewinner: {result.winner.satellite_id}")
print(f"Conversation: {result.conversation_id}")
print(f"Abgelehnt: {len(result.rejected)}")
else:
print("Keine gültigen Kandidaten")
arbitrator.on_decision(on_decision)
# Callback entfernen
arbitrator.remove_decision_callback(on_decision)
Status und Kontrolle
# Status prüfen
print(arbitrator.is_arbitrating) # Arbitration aktiv?
print(arbitrator.current_candidates) # Aktuelle Kandidaten-Liste
# Fenster-Status
status = arbitrator.get_window_status()
# {
# "state": "open",
# "candidates": 2,
# "duration_ms": 500,
# "remaining_ms": 500,
# "arbitration_id": "arb_12345678",
# }
# Strategie ändern
from trixy_core.arbitration import LoudestStrategy
arbitrator.strategy = LoudestStrategy()
# Aktuelle Arbitration abbrechen
await arbitrator.cancel_current()
# Statistiken
stats = arbitrator.stats
# {
# "total_arbitrations": 10,
# "successful": 9,
# "single_candidate": 3,
# "multiple_candidates": 7,
# "average_candidates": 2.5,
# "average_duration_ms": 950,
# }
Als Service starten
# Mit Application
arbitrator = WakewordArbitrator(config, application=app)
await arbitrator.start()
# Event-Handler wird automatisch registriert für "wakeword_detected"
# ...
await arbitrator.stop()
ArbitrationResult
Ergebnis einer Arbitration.
from trixy_core.arbitration import ArbitrationResult
# Felder
result.success # True wenn Gewinner gefunden
result.winner # ArbitrationCandidate oder None
result.rejected # Liste abgelehnter Kandidaten
result.conversation_id # Generierte Conversation-ID
result.arbitration_id # Eindeutige Arbitration-ID
result.started_at # Start-Zeitstempel
result.decided_at # Entscheidungs-Zeitstempel
result.duration_ms # Dauer in Millisekunden
result.candidate_count # Anzahl Kandidaten
result.strategy_used # Name der verwendeten Strategie
result.close_reason # Grund für Fenster-Schließung
# Serialisierung
data = result.to_dict()
Integration
Mit Event-System
from trixy_core.events import TrixyEvent
class MyService:
def __init__(self, arbitrator: WakewordArbitrator):
self.arbitrator = arbitrator
self.arbitrator.on_decision(self.handle_decision)
async def handle_decision(self, result: ArbitrationResult) -> None:
if result.success:
# Konversation starten
await self.start_conversation(
satellite_id=result.winner.satellite_id,
conversation_id=result.conversation_id,
)
In Plugins
from trixy_core.plugins import TrixyPlugin
from trixy_core.arbitration import ArbitrationCandidate
class MyPlugin(TrixyPlugin):
@TrixyEvent(["wakeword_detected"])
async def on_wakeword(self, event_name: str, data: dict) -> None:
# Arbitrator aus Application holen
arbitrator = self.application.get_service("wakeword_arbitrator")
candidate = ArbitrationCandidate(
satellite_id=data["satellite_id"],
room_id=data.get("room_id", ""),
audio_level=data.get("audio_level", 0.0),
confidence=data.get("confidence", 0.0),
)
await arbitrator.add_candidate(candidate)
Konfigurationsbeispiele
Standard (Combined)
{
"arbitration": {
"strategy": "combined",
"window_duration_ms": 1000,
"audio_weight": 0.6,
"confidence_weight": 0.3,
"recency_weight": 0.1
}
}
Nur Audio-Level
{
"arbitration": {
"strategy": "loudest",
"window_duration_ms": 800,
"min_audio_level": 0.2
}
}
Mit Raum-Relevanz
{
"arbitration": {
"strategy": "proximity",
"window_duration_ms": 1200,
"room_priorities": {
"music": ["living_room", "bedroom"],
"alarm": ["bedroom", "kitchen"]
},
"adjacent_rooms": {
"living_room": ["hallway", "kitchen"]
}
}
}