| |
| """ |
| Grok Logic Layer with Shared Blackboard Integration |
| Multi-expert collaboration via evidence store |
| """ |
|
|
| from typing import List, Dict, Any |
| from lego_moe import LegoMoE |
| from shared_blackboard import SharedBlackboard, Arbitrator |
| import re |
| import json |
|
|
| class GrokWithBlackboard: |
| """ |
| Grok logic layer with shared blackboard for expert collaboration |
| Orchestrates multiple experts via evidence store |
| """ |
| |
| def __init__(self, lego_moe: LegoMoE): |
| self.moe = lego_moe |
| self.blackboard = SharedBlackboard() |
| self.arbitrator = Arbitrator(self.blackboard) |
| print("✓ Grok with Blackboard initialized") |
| |
| def parse_intent(self, query: str) -> Dict[str, Any]: |
| """Parse query intent (same as before)""" |
| query_lower = query.lower() |
| |
| if any(word in query_lower for word in ['calculate', 'compute', 'multiply', 'add', 'divide', 'subtract', 'times']): |
| qtype = 'math_arithmetic' |
| elif any(word in query_lower for word in ['if', 'then', 'all', 'some', 'implies']): |
| qtype = 'logic_deduction' |
| else: |
| qtype = 'general_qa' |
| |
| numbers = [float(n) for n in re.findall(r'\d+\.?\d*', query)] |
| operators = re.findall(r'[\+\-\*/×÷]', query) |
| |
| if 'times' in query_lower or 'multiply' in query_lower: |
| operators.append('*') |
| if 'divided' in query_lower or 'divide' in query_lower: |
| operators.append('/') |
| |
| if len(numbers) >= 2 and len(operators) >= 1: |
| qtype = 'math_arithmetic' |
| |
| return { |
| "type": qtype, |
| "entities": {"numbers": numbers, "operators": operators} |
| } |
| |
| def execute_math(self, numbers: List[float], operator: str) -> float: |
| """Direct math calculation""" |
| if operator == '*': |
| return numbers[0] * numbers[1] |
| elif operator == '/': |
| return numbers[0] / numbers[1] if numbers[1] != 0 else None |
| elif operator == '+': |
| return numbers[0] + numbers[1] |
| elif operator == '-': |
| return numbers[0] - numbers[1] |
| return None |
| |
| def run_math_expert(self, query: str, numbers: List[float], operator: str): |
| """Math expert writes to blackboard""" |
| result = self.execute_math(numbers, operator) |
| |
| self.blackboard.write_evidence( |
| expert="math", |
| finding=result, |
| method="direct_calculation", |
| confidence=1.0, |
| scope="calculation", |
| metadata={"operation": f"{numbers[0]} {operator} {numbers[1]}"} |
| ) |
| |
| def run_logic_expert(self, query: str, numbers: List[float], operator: str): |
| """Logic expert validates units/consistency""" |
| |
| result_evidence = self.blackboard.read_evidence(expert_filter="math") |
| |
| if result_evidence: |
| result = result_evidence[0].finding |
| |
| |
| if operator == '*' and result < max(numbers): |
| finding = "Result seems incorrect (multiplication should increase)" |
| confidence = 0.3 |
| elif operator == '/' and result > numbers[0]: |
| finding = "Result seems incorrect (division should decrease)" |
| confidence = 0.3 |
| else: |
| finding = "Result passes sanity checks" |
| confidence = 0.95 |
| |
| self.blackboard.write_evidence( |
| expert="logic", |
| finding=finding, |
| method="sanity_check", |
| confidence=confidence, |
| scope="validation" |
| ) |
| |
| def reason_with_blackboard(self, query: str, verbose: bool = True) -> Dict[str, Any]: |
| """ |
| Multi-expert reasoning via blackboard |
| |
| Returns: |
| Arbitrated result with confidence |
| """ |
| if verbose: |
| print(f"\n{'='*70}") |
| print(f"🧠 GROK WITH BLACKBOARD") |
| print(f"{'='*70}") |
| print(f"Query: {query}\n") |
| |
| |
| request_id = f"req_{hash(query) % 10000}" |
| self.blackboard.start_request(request_id) |
| |
| |
| intent = self.parse_intent(query) |
| if verbose: |
| print(f"Intent: {intent['type']}") |
| |
| |
| if intent['type'] == 'math_arithmetic': |
| numbers = intent['entities']['numbers'] |
| operators = intent['entities']['operators'] |
| |
| if len(numbers) == 2 and len(operators) == 1: |
| if verbose: |
| print(f"\nOrchestrating experts:") |
| |
| |
| if verbose: |
| print(f" 1. Math expert calculating...") |
| self.run_math_expert(query, numbers, operators[0]) |
| |
| |
| if verbose: |
| print(f" 2. Logic expert validating...") |
| self.run_logic_expert(query, numbers, operators[0]) |
| |
| |
| self.blackboard.lock() |
| |
| |
| if verbose: |
| print(f"\nArbitration:") |
| |
| confidence = self.arbitrator.aggregate_confidence() |
| conflicts = self.arbitrator.detect_conflicts() |
| result = self.arbitrator.synthesize_answer() |
| |
| if verbose: |
| print(f" Confidence: {confidence:.3f}") |
| print(f" Conflicts: {len(conflicts)}") |
| print(f"\n{'='*70}") |
| print(f"Answer: {result['answer']}") |
| print(f"Confidence: {result['confidence']:.3f}") |
| print(f"Primary Expert: {result['primary_expert']}") |
| if result['supporting_experts']: |
| print(f"Supporting: {', '.join(result['supporting_experts'])}") |
| print(f"{'='*70}\n") |
| |
| return result |
|
|
|
|
| def main(): |
| """Test Grok with Blackboard""" |
| print("\n🧠 Grok Logic Layer with Shared Blackboard\n") |
| |
| |
| print("Loading LEGO-MoE...") |
| moe = LegoMoE() |
| |
| print("\nInitializing Grok with Blackboard...") |
| grok = GrokWithBlackboard(moe) |
| |
| |
| test_queries = [ |
| "What is 15 times 23?", |
| "What is 100 divided by 4?", |
| ] |
| |
| for query in test_queries: |
| result = grok.reason_with_blackboard(query) |
| |
| print(f"\nBlackboard Evidence:") |
| for evidence in grok.blackboard.get_all_evidence(): |
| print(f" {evidence['expert']}: {evidence['finding']} (conf: {evidence['confidence']:.2f})") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
