beastbullet-experts / code /grok_logic.py

Complete package: 18 experts + full codebase

ae1e90a 11 months ago

13.8 kB

#!/usr/bin/env python3
"""
Grok Logic Layer - ISL-Safe Implementation
Structured reasoning with artifacts, not text-based chain-of-thought
"""

from typing import List, Dict, Any
from lego_moe import LegoMoE
import re
import json

class GrokLogicLayer:
    """
    ISL-compatible reasoning layer
    Uses structured artifacts, not prose reasoning
    """
    
    def __init__(self, lego_moe: LegoMoE):
        self.moe = lego_moe
        self.reasoning_artifacts = {}
        print("✓ Grok Logic Layer initialized (ISL-safe mode)")
    
    def parse_intent(self, query: str) -> Dict[str, Any]:
        """
        Parse query into structured intent vector
        
        Returns structured intent, not text
        """
        query_lower = query.lower()
        
        # Classify type
        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'
        elif any(word in query_lower for word in ['code', 'function', 'debug', 'fix']):
            qtype = 'code_analysis'
        else:
            qtype = 'general_qa'
        
        # Extract entities
        numbers = [float(n) for n in re.findall(r'\d+\.?\d*', query)]
        
        # Extract operators (including word forms)
        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 'plus' in query_lower:
            operators.append('+')
        if 'minus' in query_lower:
            operators.append('-')
        
        # Determine complexity
        complexity_score = len(query.split()) + len(numbers) * 2
        
        # Force math type if we have numbers and operators
        if len(numbers) >= 2 and len(operators) >= 1:
            qtype = 'math_arithmetic'
        
        return {
            "type": qtype,
            "complexity": complexity_score,
            "entities": {
                "numbers": numbers,
                "operators": operators
            },
            "requires_decomposition": complexity_score > 15
        }
    
    def decompose_to_graph(self, query: str, intent: Dict) -> Dict[str, Any]:
        """
        Decompose into structured task graph
        NOT text-based sub-problems
        """
        if not intent['requires_decomposition']:
            return {
                "nodes": [{
                    "id": 0,
                    "task": query,
                    "type": intent['type'],
                    "entities": intent['entities']  # Pass entities through
                }],
                "edges": [],
                "execution_order": [0]
            }
        
        # For math: decompose into operations
        if intent['type'] == 'math_arithmetic':
            return self._decompose_math(query, intent)
        
        # For logic: decompose into premises and conclusion
        elif intent['type'] == 'logic_deduction':
            return self._decompose_logic(query)
        
        # Default: single node
        return {
            "nodes": [{
                "id": 0,
                "task": query,
                "type": intent['type'],
                "entities": intent['entities']
            }],
            "edges": [],
            "execution_order": [0]
        }
    
    def _decompose_math(self, query: str, intent: Dict) -> Dict:
        """Decompose math problem into operation graph"""
        numbers = intent['entities']['numbers']
        operators = intent['entities']['operators']
        
        if not numbers or not operators:
            return {
                "nodes": [{"id": 0, "task": query, "type": "math_arithmetic"}],
                "edges": [],
                "execution_order": [0]
            }
        
        # Build operation sequence
        nodes = []
        for i, (num, op) in enumerate(zip(numbers[:-1], operators)):
            nodes.append({
                "id": i,
                "operation": op,
                "operands": [num, numbers[i+1]],
                "type": "math_operation"
            })
        
        return {
            "nodes": nodes,
            "edges": [(i, i+1) for i in range(len(nodes)-1)],
            "execution_order": list(range(len(nodes)))
        }
    
    def _decompose_logic(self, query: str) -> Dict:
        """Decompose logic problem into premises and conclusion"""
        # Simple heuristic: split on periods and "therefore"
        sentences = [s.strip() for s in re.split(r'[.?!]', query) if s.strip()]
        
        nodes = []
        for i, sentence in enumerate(sentences):
            node_type = "conclusion" if any(word in sentence.lower() for word in ['therefore', 'thus', 'so']) else "premise"
            nodes.append({
                "id": i,
                "statement": sentence,
                "type": node_type
            })
        
        return {
            "nodes": nodes,
            "edges": [(i, len(nodes)-1) for i in range(len(nodes)-1)],
            "execution_order": list(range(len(nodes)))
        }
    
    def plan_execution(self, task_graph: Dict) -> Dict[str, Any]:
        """
        Create execution DAG with expert routing
        """
        execution_plan = {
            "steps": [],
            "expert_routing": {},
            "verification_rules": []
        }
        
        for node in task_graph['nodes']:
            step = {
                "node_id": node['id'],
                "expert": self._select_expert(node),
                "input": node,
                "verification": self._get_verification_rules(node)
            }
            execution_plan['steps'].append(step)
        
        return execution_plan
    
    def _select_expert(self, node: Dict) -> str:
        """Select expert for node"""
        node_type = node.get('type', 'general')
        
        if 'math' in node_type:
            return 'mathematical_reasoning'
        elif 'logic' in node_type:
            return 'logical_inference'
        elif 'code' in node_type:
            return 'code_generation'
        else:
            return 'general'
    
    def _get_verification_rules(self, node: Dict) -> List[str]:
        """Get verification rules for node"""
        rules = ["bounds_check"]
        
        if node.get('type') == 'math_operation':
            rules.extend(["reverse_check", "unit_check"])
        elif node.get('type') == 'premise':
            rules.append("consistency_check")
        
        return rules
    
    def execute(self, execution_plan: Dict) -> Dict[str, Any]:
        """Execute plan using LEGO-MoE"""
        results = {
            "step_results": [],
            "final_state": None
        }
        
        for step in execution_plan['steps']:
            # Execute via LEGO-MoE
            node_input = step['input']
            
            # For simple math with 2 numbers and 1 operator, calculate directly
            if node_input.get('type') == 'math_arithmetic':
                entities = node_input.get('entities', {})
                numbers = entities.get('numbers', [])
                operators = entities.get('operators', [])
                
                if len(numbers) == 2 and len(operators) == 1:
                    # Direct calculation
                    result = self._execute_math_op(operators[0], numbers)
                else:
                    # Fallback to task execution
                    task = node_input.get('task', '')
                    result = self.moe.generate(task, verbose=False)
            elif node_input.get('type') == 'math_operation':
                # Direct calculation
                op = node_input['operation']
                operands = node_input['operands']
                result = self._execute_math_op(op, operands)
            else:
                # Route to LEGO-MoE
                task = node_input.get('task') or node_input.get('statement')
                result = self.moe.generate(task, verbose=False)
            
            results['step_results'].append({
                "step_id": step['node_id'],
                "result": result,
                "verified": False  # Will verify next
            })
        
        return results
    
    def _execute_math_op(self, operator: str, operands: List[float]) -> float:
        """Execute math operation"""
        if operator == '+':
            return operands[0] + operands[1]
        elif operator == '-':
            return operands[0] - operands[1]
        elif operator == '*' or operator == '×':
            return operands[0] * operands[1]
        elif operator == '/':
            return operands[0] / operands[1] if operands[1] != 0 else None
        return None
    
    def verify(self, results: Dict, task_graph: Dict) -> Dict[str, Any]:
        """
        Verify results with constraint checks
        NOT multiple re-solves
        """
        verification = {
            "passed": True,
            "checks": [],
            "confidence": 1.0
        }
        
        for step_result in results['step_results']:
            result_value = step_result['result']
            
            # Bounds check
            if isinstance(result_value, (int, float)):
                bounds_ok = -1e10 < result_value < 1e10
                verification['checks'].append({
                    "type": "bounds_check",
                    "passed": bounds_ok
                })
                if not bounds_ok:
                    verification['passed'] = False
                    verification['confidence'] *= 0.5
        
        return verification
    
    def synthesize(self, verified_results: Dict, verification: Dict) -> Dict[str, Any]:
        """
        Synthesize final answer
        NO reasoning dump
        """
        if not verified_results['step_results']:
            return {
                "answer": "Unable to process query",
                "confidence": 0.0
            }
        
        # Get final result
        final_result = verified_results['step_results'][-1]['result']
        
        return {
            "answer": str(final_result),
            "confidence": verification['confidence'],
            "verified": verification['passed']
        }
    
    def self_correct(self, query: str, result: Dict, max_iterations: int = 2) -> Dict:
        """
        Self-correction with hard cap
        Downgrades confidence if uncertain
        """
        for i in range(max_iterations):
            if result['verified'] and result['confidence'] > 0.8:
                return result
            
            # Re-execute with alternative approach
            # (simplified for now)
            result['confidence'] *= 0.9
        
        # If still uncertain, downgrade
        if result['confidence'] < 0.5:
            result['note'] = "Uncertain, partial verification"
        
        return result
    
    def reason(self, query: str, verbose: bool = True) -> Dict[str, Any]:
        """
        Main reasoning pipeline
        Returns structured result, NOT text trace
        """
        if verbose:
            print(f"\n{'='*70}")
            print(f"🧠 GROK LOGIC LAYER (ISL-Safe)")
            print(f"{'='*70}")
            print(f"Query: {query}\n")
        
        # 1. Understand (structured intent)
        intent = self.parse_intent(query)
        if verbose:
            print(f"Intent: {intent['type']} (complexity: {intent['complexity']})")
        
        # 2. Decompose (task graph)
        task_graph = self.decompose_to_graph(query, intent)
        if verbose:
            print(f"Task graph: {len(task_graph['nodes'])} nodes")
        
        # 3. Plan (execution DAG)
        execution_plan = self.plan_execution(task_graph)
        if verbose:
            print(f"Execution plan: {len(execution_plan['steps'])} steps")
        
        # 4. Execute
        results = self.execute(execution_plan)
        if verbose:
            print(f"Executed: {len(results['step_results'])} steps")
        
        # 5. Verify
        verification = self.verify(results, task_graph)
        if verbose:
            print(f"Verification: {'PASSED' if verification['passed'] else 'FAILED'}")
        
        # 6. Synthesize
        final_answer = self.synthesize(results, verification)
        
        # 7. Self-correct if needed
        final_answer = self.self_correct(query, final_answer)
        
        if verbose:
            print(f"\n{'='*70}")
            print(f"Answer: {final_answer['answer']}")
            print(f"Confidence: {final_answer['confidence']:.2f}")
            print(f"{'='*70}\n")
        
        # Store artifacts (machine-readable)
        self.reasoning_artifacts = {
            "intent": intent,
            "task_graph": task_graph,
            "execution_plan": execution_plan,
            "results": results,
            "verification": verification
        }
        
        return final_answer


def main():
    """Test ISL-safe Grok Logic Layer"""
    print("\n🧠 Grok Logic Layer - ISL-Safe Implementation\n")
    
    # Initialize
    print("Loading LEGO-MoE...")
    moe = LegoMoE()
    
    print("\nInitializing Grok Logic Layer...")
    grok = GrokLogicLayer(moe)
    
    # Test queries
    test_queries = [
        "What is 15 times 23?",
        "What is 100 divided by 4?",
    ]
    
    for query in test_queries:
        result = grok.reason(query)
        
        print(f"\nReasoning Artifacts (structured):")
        print(json.dumps(grok.reasoning_artifacts['intent'], indent=2))


if __name__ == "__main__":
    main()

	#!/usr/bin/env python3
	"""
	Grok Logic Layer - ISL-Safe Implementation
	Structured reasoning with artifacts, not text-based chain-of-thought
	"""

	from typing import List, Dict, Any
	from lego_moe import LegoMoE
	import re
	import json

	class GrokLogicLayer:
	"""
	ISL-compatible reasoning layer
	Uses structured artifacts, not prose reasoning
	"""

	def __init__(self, lego_moe: LegoMoE):
	self.moe = lego_moe
	self.reasoning_artifacts = {}
	print("✓ Grok Logic Layer initialized (ISL-safe mode)")

	def parse_intent(self, query: str) -> Dict[str, Any]:
	"""
	Parse query into structured intent vector

	Returns structured intent, not text
	"""
	query_lower = query.lower()

	# Classify type
	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'
	elif any(word in query_lower for word in ['code', 'function', 'debug', 'fix']):
	qtype = 'code_analysis'
	else:
	qtype = 'general_qa'

	# Extract entities
	numbers = [float(n) for n in re.findall(r'\d+\.?\d*', query)]

	# Extract operators (including word forms)
	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 'plus' in query_lower:
	operators.append('+')
	if 'minus' in query_lower:
	operators.append('-')

	# Determine complexity
	complexity_score = len(query.split()) + len(numbers) * 2

	# Force math type if we have numbers and operators
	if len(numbers) >= 2 and len(operators) >= 1:
	qtype = 'math_arithmetic'

	return {
	"type": qtype,
	"complexity": complexity_score,
	"entities": {
	"numbers": numbers,
	"operators": operators
	},
	"requires_decomposition": complexity_score > 15
	}

	def decompose_to_graph(self, query: str, intent: Dict) -> Dict[str, Any]:
	"""
	Decompose into structured task graph
	NOT text-based sub-problems
	"""
	if not intent['requires_decomposition']:
	return {
	"nodes": [{
	"id": 0,
	"task": query,
	"type": intent['type'],
	"entities": intent['entities'] # Pass entities through
	}],
	"edges": [],
	"execution_order": [0]
	}

	# For math: decompose into operations
	if intent['type'] == 'math_arithmetic':
	return self._decompose_math(query, intent)

	# For logic: decompose into premises and conclusion
	elif intent['type'] == 'logic_deduction':
	return self._decompose_logic(query)

	# Default: single node
	return {
	"nodes": [{
	"id": 0,
	"task": query,
	"type": intent['type'],
	"entities": intent['entities']
	}],
	"edges": [],
	"execution_order": [0]
	}

	def _decompose_math(self, query: str, intent: Dict) -> Dict:
	"""Decompose math problem into operation graph"""
	numbers = intent['entities']['numbers']
	operators = intent['entities']['operators']

	if not numbers or not operators:
	return {
	"nodes": [{"id": 0, "task": query, "type": "math_arithmetic"}],
	"edges": [],
	"execution_order": [0]
	}

	# Build operation sequence
	nodes = []
	for i, (num, op) in enumerate(zip(numbers[:-1], operators)):
	nodes.append({
	"id": i,
	"operation": op,
	"operands": [num, numbers[i+1]],
	"type": "math_operation"
	})

	return {
	"nodes": nodes,
	"edges": [(i, i+1) for i in range(len(nodes)-1)],
	"execution_order": list(range(len(nodes)))
	}

	def _decompose_logic(self, query: str) -> Dict:
	"""Decompose logic problem into premises and conclusion"""
	# Simple heuristic: split on periods and "therefore"
	sentences = [s.strip() for s in re.split(r'[.?!]', query) if s.strip()]

	nodes = []
	for i, sentence in enumerate(sentences):
	node_type = "conclusion" if any(word in sentence.lower() for word in ['therefore', 'thus', 'so']) else "premise"
	nodes.append({
	"id": i,
	"statement": sentence,
	"type": node_type
	})

	return {
	"nodes": nodes,
	"edges": [(i, len(nodes)-1) for i in range(len(nodes)-1)],
	"execution_order": list(range(len(nodes)))
	}

	def plan_execution(self, task_graph: Dict) -> Dict[str, Any]:
	"""
	Create execution DAG with expert routing
	"""
	execution_plan = {
	"steps": [],
	"expert_routing": {},
	"verification_rules": []
	}

	for node in task_graph['nodes']:
	step = {
	"node_id": node['id'],
	"expert": self._select_expert(node),
	"input": node,
	"verification": self._get_verification_rules(node)
	}
	execution_plan['steps'].append(step)

	return execution_plan

	def _select_expert(self, node: Dict) -> str:
	"""Select expert for node"""
	node_type = node.get('type', 'general')

	if 'math' in node_type:
	return 'mathematical_reasoning'
	elif 'logic' in node_type:
	return 'logical_inference'
	elif 'code' in node_type:
	return 'code_generation'
	else:
	return 'general'

	def _get_verification_rules(self, node: Dict) -> List[str]:
	"""Get verification rules for node"""
	rules = ["bounds_check"]

	if node.get('type') == 'math_operation':
	rules.extend(["reverse_check", "unit_check"])
	elif node.get('type') == 'premise':
	rules.append("consistency_check")

	return rules

	def execute(self, execution_plan: Dict) -> Dict[str, Any]:
	"""Execute plan using LEGO-MoE"""
	results = {
	"step_results": [],
	"final_state": None
	}

	for step in execution_plan['steps']:
	# Execute via LEGO-MoE
	node_input = step['input']

	# For simple math with 2 numbers and 1 operator, calculate directly
	if node_input.get('type') == 'math_arithmetic':
	entities = node_input.get('entities', {})
	numbers = entities.get('numbers', [])
	operators = entities.get('operators', [])

	if len(numbers) == 2 and len(operators) == 1:
	# Direct calculation
	result = self._execute_math_op(operators[0], numbers)
	else:
	# Fallback to task execution
	task = node_input.get('task', '')
	result = self.moe.generate(task, verbose=False)
	elif node_input.get('type') == 'math_operation':
	# Direct calculation
	op = node_input['operation']
	operands = node_input['operands']
	result = self._execute_math_op(op, operands)
	else:
	# Route to LEGO-MoE
	task = node_input.get('task') or node_input.get('statement')
	result = self.moe.generate(task, verbose=False)

	results['step_results'].append({
	"step_id": step['node_id'],
	"result": result,
	"verified": False # Will verify next
	})

	return results

	def _execute_math_op(self, operator: str, operands: List[float]) -> float:
	"""Execute math operation"""
	if operator == '+':
	return operands[0] + operands[1]
	elif operator == '-':
	return operands[0] - operands[1]
	elif operator == '*' or operator == '×':
	return operands[0] * operands[1]
	elif operator == '/':
	return operands[0] / operands[1] if operands[1] != 0 else None
	return None

	def verify(self, results: Dict, task_graph: Dict) -> Dict[str, Any]:
	"""
	Verify results with constraint checks
	NOT multiple re-solves
	"""
	verification = {
	"passed": True,
	"checks": [],
	"confidence": 1.0
	}

	for step_result in results['step_results']:
	result_value = step_result['result']

	# Bounds check
	if isinstance(result_value, (int, float)):
	bounds_ok = -1e10 < result_value < 1e10
	verification['checks'].append({
	"type": "bounds_check",
	"passed": bounds_ok
	})
	if not bounds_ok:
	verification['passed'] = False
	verification['confidence'] *= 0.5

	return verification

	def synthesize(self, verified_results: Dict, verification: Dict) -> Dict[str, Any]:
	"""
	Synthesize final answer
	NO reasoning dump
	"""
	if not verified_results['step_results']:
	return {
	"answer": "Unable to process query",
	"confidence": 0.0
	}

	# Get final result
	final_result = verified_results['step_results'][-1]['result']

	return {
	"answer": str(final_result),
	"confidence": verification['confidence'],
	"verified": verification['passed']
	}

	def self_correct(self, query: str, result: Dict, max_iterations: int = 2) -> Dict:
	"""
	Self-correction with hard cap
	Downgrades confidence if uncertain
	"""
	for i in range(max_iterations):
	if result['verified'] and result['confidence'] > 0.8:
	return result

	# Re-execute with alternative approach
	# (simplified for now)
	result['confidence'] *= 0.9

	# If still uncertain, downgrade
	if result['confidence'] < 0.5:
	result['note'] = "Uncertain, partial verification"

	return result

	def reason(self, query: str, verbose: bool = True) -> Dict[str, Any]:
	"""
	Main reasoning pipeline
	Returns structured result, NOT text trace
	"""
	if verbose:
	print(f"\n{'='*70}")
	print(f"🧠 GROK LOGIC LAYER (ISL-Safe)")
	print(f"{'='*70}")
	print(f"Query: {query}\n")

	# 1. Understand (structured intent)
	intent = self.parse_intent(query)
	if verbose:
	print(f"Intent: {intent['type']} (complexity: {intent['complexity']})")

	# 2. Decompose (task graph)
	task_graph = self.decompose_to_graph(query, intent)
	if verbose:
	print(f"Task graph: {len(task_graph['nodes'])} nodes")

	# 3. Plan (execution DAG)
	execution_plan = self.plan_execution(task_graph)
	if verbose:
	print(f"Execution plan: {len(execution_plan['steps'])} steps")

	# 4. Execute
	results = self.execute(execution_plan)
	if verbose:
	print(f"Executed: {len(results['step_results'])} steps")

	# 5. Verify
	verification = self.verify(results, task_graph)
	if verbose:
	print(f"Verification: {'PASSED' if verification['passed'] else 'FAILED'}")

	# 6. Synthesize
	final_answer = self.synthesize(results, verification)

	# 7. Self-correct if needed
	final_answer = self.self_correct(query, final_answer)

	if verbose:
	print(f"\n{'='*70}")
	print(f"Answer: {final_answer['answer']}")
	print(f"Confidence: {final_answer['confidence']:.2f}")
	print(f"{'='*70}\n")

	# Store artifacts (machine-readable)
	self.reasoning_artifacts = {
	"intent": intent,
	"task_graph": task_graph,
	"execution_plan": execution_plan,
	"results": results,
	"verification": verification
	}

	return final_answer


	def main():
	"""Test ISL-safe Grok Logic Layer"""
	print("\n🧠 Grok Logic Layer - ISL-Safe Implementation\n")

	# Initialize
	print("Loading LEGO-MoE...")
	moe = LegoMoE()

	print("\nInitializing Grok Logic Layer...")
	grok = GrokLogicLayer(moe)

	# Test queries
	test_queries = [
	"What is 15 times 23?",
	"What is 100 divided by 4?",
	]

	for query in test_queries:
	result = grok.reason(query)

	print(f"\nReasoning Artifacts (structured):")
	print(json.dumps(grok.reasoning_artifacts['intent'], indent=2))


	if __name__ == "__main__":
	main()