beastbullet-experts / code /shared_blackboard.py

Complete package: 18 experts + full codebase

ae1e90a 11 months ago

10.3 kB

#!/usr/bin/env python3
"""
Shared Blackboard System
ISL-safe expert collaboration via evidence store
NO direct expert-to-expert messaging
"""

from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Evidence:
    """
    Immutable evidence entry from an expert
    Write-only, no modification after creation
    """
    expert: str
    finding: Any
    method: str
    confidence: float
    scope: str
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict:
        """Convert to dictionary"""
        return {
            "expert": self.expert,
            "finding": self.finding,
            "method": self.method,
            "confidence": self.confidence,
            "scope": self.scope,
            "timestamp": self.timestamp,
            "metadata": self.metadata
        }


class SharedBlackboard:
    """
    Shared evidence store for expert collaboration
    
    Rules:
    - Experts can ONLY write (no modification)
    - Entries are immutable
    - No expert-to-expert messaging
    - Orchestration by Grok Layer only
    """
    
    def __init__(self):
        self.evidence: List[Evidence] = []
        self.request_id: Optional[str] = None
        self._locked = False
    
    def start_request(self, request_id: str):
        """Start new request - clears previous evidence"""
        self.evidence = []
        self.request_id = request_id
        self._locked = False  # Unlock for new request
    
    def write_evidence(
        self,
        expert: str,
        finding: Any,
        method: str,
        confidence: float,
        scope: str = "general",
        metadata: Optional[Dict] = None
    ) -> Evidence:
        """
        Expert writes evidence to blackboard
        
        Args:
            expert: Expert identifier
            finding: The finding (structured data)
            method: How it was derived
            confidence: 0.0-1.0
            scope: What domain this applies to
            metadata: Additional context
        
        Returns:
            Evidence entry
        """
        if self._locked:
            raise RuntimeError("Blackboard is locked - no more writes allowed")
        
        # Validate confidence
        if not 0.0 <= confidence <= 1.0:
            raise ValueError(f"Confidence must be 0.0-1.0, got {confidence}")
        
        # Create immutable evidence
        evidence = Evidence(
            expert=expert,
            finding=finding,
            method=method,
            confidence=confidence,
            scope=scope,
            metadata=metadata or {}
        )
        
        self.evidence.append(evidence)
        return evidence
    
    def read_evidence(
        self,
        expert_filter: Optional[str] = None,
        scope_filter: Optional[str] = None
    ) -> List[Evidence]:
        """
        Read evidence from blackboard (read-only)
        
        Args:
            expert_filter: Only return evidence from this expert
            scope_filter: Only return evidence for this scope
        
        Returns:
            List of matching evidence
        """
        results = self.evidence
        
        if expert_filter:
            results = [e for e in results if e.expert == expert_filter]
        
        if scope_filter:
            results = [e for e in results if e.scope == scope_filter]
        
        return results
    
    def lock(self):
        """Lock blackboard - no more writes allowed"""
        self._locked = True
    
    def get_all_evidence(self) -> List[Dict]:
        """Get all evidence as dictionaries"""
        return [e.to_dict() for e in self.evidence]
    
    def get_summary(self) -> Dict[str, Any]:
        """Get summary of blackboard state"""
        return {
            "request_id": self.request_id,
            "total_evidence": len(self.evidence),
            "experts_contributed": len(set(e.expert for e in self.evidence)),
            "locked": self._locked,
            "evidence": self.get_all_evidence()
        }


class Arbitrator:
    """
    Arbitrates evidence from blackboard
    Implements consensus rules (NOT voting)
    """
    
    def __init__(self, blackboard: SharedBlackboard):
        self.blackboard = blackboard
    
    def aggregate_confidence(self) -> float:
        """
        Aggregate confidence from all evidence
        
        Rule: Confidence bounded by WEAKEST critical check
        NOT average, NOT voting
        """
        evidence = self.blackboard.evidence
        
        if not evidence:
            return 0.0
        
        # Find minimum confidence (weakest link)
        min_confidence = min(e.confidence for e in evidence)
        
        # If any expert has very low confidence, that dominates
        if min_confidence < 0.5:
            return min_confidence
        
        # Otherwise, use weighted average biased toward lower values
        confidences = [e.confidence for e in evidence]
        
        # Geometric mean (more conservative than arithmetic)
        product = 1.0
        for c in confidences:
            product *= c
        
        geometric_mean = product ** (1.0 / len(confidences))
        
        return geometric_mean
    
    def detect_conflicts(self) -> List[Dict[str, Any]]:
        """
        Detect conflicting evidence
        
        Returns:
            List of conflicts found
        """
        conflicts = []
        evidence = self.blackboard.evidence
        
        # Group by scope
        by_scope: Dict[str, List[Evidence]] = {}
        for e in evidence:
            if e.scope not in by_scope:
                by_scope[e.scope] = []
            by_scope[e.scope].append(e)
        
        # Check for contradictions within each scope
        for scope, scope_evidence in by_scope.items():
            if len(scope_evidence) < 2:
                continue
            
            # Check if findings contradict
            findings = [e.finding for e in scope_evidence]
            
            # Simple contradiction check (can be enhanced)
            if len(set(str(f) for f in findings)) > 1:
                conflicts.append({
                    "scope": scope,
                    "experts": [e.expert for e in scope_evidence],
                    "findings": findings,
                    "confidences": [e.confidence for e in scope_evidence]
                })
        
        return conflicts
    
    def synthesize_answer(self) -> Dict[str, Any]:
        """
        Synthesize final answer from all evidence
        
        Returns:
            Synthesized result with confidence
        """
        evidence = self.blackboard.evidence
        
        if not evidence:
            return {
                "answer": None,
                "confidence": 0.0,
                "method": "no_evidence"
            }
        
        # Aggregate confidence
        confidence = self.aggregate_confidence()
        
        # Detect conflicts
        conflicts = self.detect_conflicts()
        
        # If conflicts exist, reduce confidence
        if conflicts:
            confidence *= 0.7  # Penalty for conflicts
        
        # Get primary finding (highest confidence)
        primary = max(evidence, key=lambda e: e.confidence)
        
        return {
            "answer": primary.finding,
            "confidence": confidence,
            "method": "arbitrated",
            "primary_expert": primary.expert,
            "supporting_experts": [e.expert for e in evidence if e.expert != primary.expert],
            "conflicts": conflicts
        }


def test_blackboard():
    """Test shared blackboard system"""
    print("="*70)
    print("Testing Shared Blackboard System")
    print("="*70)
    
    # Create blackboard
    blackboard = SharedBlackboard()
    blackboard.start_request("test_001")
    
    # Math expert writes evidence
    print("\n1. Math Expert writes evidence:")
    math_evidence = blackboard.write_evidence(
        expert="math",
        finding=150.0,
        method="direct_multiplication",
        confidence=1.0,
        scope="calculation",
        metadata={"operation": "60 * 2.5"}
    )
    print(f"   Written: {math_evidence.finding} (confidence: {math_evidence.confidence})")
    
    # Logic expert writes evidence
    print("\n2. Logic Expert writes evidence:")
    logic_evidence = blackboard.write_evidence(
        expert="logic",
        finding="Units consistent: mph × hours → miles",
        method="unit_analysis",
        confidence=0.95,
        scope="validation"
    )
    print(f"   Written: {logic_evidence.finding} (confidence: {logic_evidence.confidence})")
    
    # Security expert writes evidence (lower confidence)
    print("\n3. Security Expert writes evidence:")
    security_evidence = blackboard.write_evidence(
        expert="security",
        finding="No security concerns",
        method="static_analysis",
        confidence=0.6,
        scope="validation"
    )
    print(f"   Written: {security_evidence.finding} (confidence: {security_evidence.confidence})")
    
    # Lock blackboard
    blackboard.lock()
    print("\n4. Blackboard locked")
    
    # Arbitrate
    print("\n5. Arbitration:")
    arbitrator = Arbitrator(blackboard)
    
    confidence = arbitrator.aggregate_confidence()
    print(f"   Aggregated confidence: {confidence:.3f}")
    
    conflicts = arbitrator.detect_conflicts()
    print(f"   Conflicts detected: {len(conflicts)}")
    
    result = arbitrator.synthesize_answer()
    print(f"   Final answer: {result['answer']}")
    print(f"   Final confidence: {result['confidence']:.3f}")
    print(f"   Primary expert: {result['primary_expert']}")
    
    # Summary
    print("\n6. Blackboard Summary:")
    summary = blackboard.get_summary()
    print(f"   Total evidence: {summary['total_evidence']}")
    print(f"   Experts contributed: {summary['experts_contributed']}")
    print(f"   Locked: {summary['locked']}")
    
    print("\n" + "="*70)
    print("✓ Blackboard system working!")
    print("="*70)


if __name__ == "__main__":
    test_blackboard()

	#!/usr/bin/env python3
	"""
	Shared Blackboard System
	ISL-safe expert collaboration via evidence store
	NO direct expert-to-expert messaging
	"""

	from typing import Dict, List, Any, Optional
	from dataclasses import dataclass, field
	from datetime import datetime
	import json

	@dataclass
	class Evidence:
	"""
	Immutable evidence entry from an expert
	Write-only, no modification after creation
	"""
	expert: str
	finding: Any
	method: str
	confidence: float
	scope: str
	timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
	metadata: Dict[str, Any] = field(default_factory=dict)

	def to_dict(self) -> Dict:
	"""Convert to dictionary"""
	return {
	"expert": self.expert,
	"finding": self.finding,
	"method": self.method,
	"confidence": self.confidence,
	"scope": self.scope,
	"timestamp": self.timestamp,
	"metadata": self.metadata
	}


	class SharedBlackboard:
	"""
	Shared evidence store for expert collaboration

	Rules:
	- Experts can ONLY write (no modification)
	- Entries are immutable
	- No expert-to-expert messaging
	- Orchestration by Grok Layer only
	"""

	def __init__(self):
	self.evidence: List[Evidence] = []
	self.request_id: Optional[str] = None
	self._locked = False

	def start_request(self, request_id: str):
	"""Start new request - clears previous evidence"""
	self.evidence = []
	self.request_id = request_id
	self._locked = False # Unlock for new request

	def write_evidence(
	self,
	expert: str,
	finding: Any,
	method: str,
	confidence: float,
	scope: str = "general",
	metadata: Optional[Dict] = None
	) -> Evidence:
	"""
	Expert writes evidence to blackboard

	Args:
	expert: Expert identifier
	finding: The finding (structured data)
	method: How it was derived
	confidence: 0.0-1.0
	scope: What domain this applies to
	metadata: Additional context

	Returns:
	Evidence entry
	"""
	if self._locked:
	raise RuntimeError("Blackboard is locked - no more writes allowed")

	# Validate confidence
	if not 0.0 <= confidence <= 1.0:
	raise ValueError(f"Confidence must be 0.0-1.0, got {confidence}")

	# Create immutable evidence
	evidence = Evidence(
	expert=expert,
	finding=finding,
	method=method,
	confidence=confidence,
	scope=scope,
	metadata=metadata or {}
	)

	self.evidence.append(evidence)
	return evidence

	def read_evidence(
	self,
	expert_filter: Optional[str] = None,
	scope_filter: Optional[str] = None
	) -> List[Evidence]:
	"""
	Read evidence from blackboard (read-only)

	Args:
	expert_filter: Only return evidence from this expert
	scope_filter: Only return evidence for this scope

	Returns:
	List of matching evidence
	"""
	results = self.evidence

	if expert_filter:
	results = [e for e in results if e.expert == expert_filter]

	if scope_filter:
	results = [e for e in results if e.scope == scope_filter]

	return results

	def lock(self):
	"""Lock blackboard - no more writes allowed"""
	self._locked = True

	def get_all_evidence(self) -> List[Dict]:
	"""Get all evidence as dictionaries"""
	return [e.to_dict() for e in self.evidence]

	def get_summary(self) -> Dict[str, Any]:
	"""Get summary of blackboard state"""
	return {
	"request_id": self.request_id,
	"total_evidence": len(self.evidence),
	"experts_contributed": len(set(e.expert for e in self.evidence)),
	"locked": self._locked,
	"evidence": self.get_all_evidence()
	}


	class Arbitrator:
	"""
	Arbitrates evidence from blackboard
	Implements consensus rules (NOT voting)
	"""

	def __init__(self, blackboard: SharedBlackboard):
	self.blackboard = blackboard

	def aggregate_confidence(self) -> float:
	"""
	Aggregate confidence from all evidence

	Rule: Confidence bounded by WEAKEST critical check
	NOT average, NOT voting
	"""
	evidence = self.blackboard.evidence

	if not evidence:
	return 0.0

	# Find minimum confidence (weakest link)
	min_confidence = min(e.confidence for e in evidence)

	# If any expert has very low confidence, that dominates
	if min_confidence < 0.5:
	return min_confidence

	# Otherwise, use weighted average biased toward lower values
	confidences = [e.confidence for e in evidence]

	# Geometric mean (more conservative than arithmetic)
	product = 1.0
	for c in confidences:
	product *= c

	geometric_mean = product ** (1.0 / len(confidences))

	return geometric_mean

	def detect_conflicts(self) -> List[Dict[str, Any]]:
	"""
	Detect conflicting evidence

	Returns:
	List of conflicts found
	"""
	conflicts = []
	evidence = self.blackboard.evidence

	# Group by scope
	by_scope: Dict[str, List[Evidence]] = {}
	for e in evidence:
	if e.scope not in by_scope:
	by_scope[e.scope] = []
	by_scope[e.scope].append(e)

	# Check for contradictions within each scope
	for scope, scope_evidence in by_scope.items():
	if len(scope_evidence) < 2:
	continue

	# Check if findings contradict
	findings = [e.finding for e in scope_evidence]

	# Simple contradiction check (can be enhanced)
	if len(set(str(f) for f in findings)) > 1:
	conflicts.append({
	"scope": scope,
	"experts": [e.expert for e in scope_evidence],
	"findings": findings,
	"confidences": [e.confidence for e in scope_evidence]
	})

	return conflicts

	def synthesize_answer(self) -> Dict[str, Any]:
	"""
	Synthesize final answer from all evidence

	Returns:
	Synthesized result with confidence
	"""
	evidence = self.blackboard.evidence

	if not evidence:
	return {
	"answer": None,
	"confidence": 0.0,
	"method": "no_evidence"
	}

	# Aggregate confidence
	confidence = self.aggregate_confidence()

	# Detect conflicts
	conflicts = self.detect_conflicts()

	# If conflicts exist, reduce confidence
	if conflicts:
	confidence *= 0.7 # Penalty for conflicts

	# Get primary finding (highest confidence)
	primary = max(evidence, key=lambda e: e.confidence)

	return {
	"answer": primary.finding,
	"confidence": confidence,
	"method": "arbitrated",
	"primary_expert": primary.expert,
	"supporting_experts": [e.expert for e in evidence if e.expert != primary.expert],
	"conflicts": conflicts
	}


	def test_blackboard():
	"""Test shared blackboard system"""
	print("="*70)
	print("Testing Shared Blackboard System")
	print("="*70)

	# Create blackboard
	blackboard = SharedBlackboard()
	blackboard.start_request("test_001")

	# Math expert writes evidence
	print("\n1. Math Expert writes evidence:")
	math_evidence = blackboard.write_evidence(
	expert="math",
	finding=150.0,
	method="direct_multiplication",
	confidence=1.0,
	scope="calculation",
	metadata={"operation": "60 * 2.5"}
	)
	print(f" Written: {math_evidence.finding} (confidence: {math_evidence.confidence})")

	# Logic expert writes evidence
	print("\n2. Logic Expert writes evidence:")
	logic_evidence = blackboard.write_evidence(
	expert="logic",
	finding="Units consistent: mph × hours → miles",
	method="unit_analysis",
	confidence=0.95,
	scope="validation"
	)
	print(f" Written: {logic_evidence.finding} (confidence: {logic_evidence.confidence})")

	# Security expert writes evidence (lower confidence)
	print("\n3. Security Expert writes evidence:")
	security_evidence = blackboard.write_evidence(
	expert="security",
	finding="No security concerns",
	method="static_analysis",
	confidence=0.6,
	scope="validation"
	)
	print(f" Written: {security_evidence.finding} (confidence: {security_evidence.confidence})")

	# Lock blackboard
	blackboard.lock()
	print("\n4. Blackboard locked")

	# Arbitrate
	print("\n5. Arbitration:")
	arbitrator = Arbitrator(blackboard)

	confidence = arbitrator.aggregate_confidence()
	print(f" Aggregated confidence: {confidence:.3f}")

	conflicts = arbitrator.detect_conflicts()
	print(f" Conflicts detected: {len(conflicts)}")

	result = arbitrator.synthesize_answer()
	print(f" Final answer: {result['answer']}")
	print(f" Final confidence: {result['confidence']:.3f}")
	print(f" Primary expert: {result['primary_expert']}")

	# Summary
	print("\n6. Blackboard Summary:")
	summary = blackboard.get_summary()
	print(f" Total evidence: {summary['total_evidence']}")
	print(f" Experts contributed: {summary['experts_contributed']}")
	print(f" Locked: {summary['locked']}")

	print("\n" + "="*70)
	print("✓ Blackboard system working!")
	print("="*70)


	if __name__ == "__main__":
	test_blackboard()