abstraction_engine.py

class AbstractionEngine:

    def __init__(self, meta_engine):

        self.rules = []
        self.meta_engine = meta_engine

    # =====================================================
    # INGEST EXPERIENCE (ADAPTIVE MODE)
    # =====================================================
    def ingest(self, batch):

        if not hasattr(self, "rules"):
            self.rules = []

        for item in batch:

            rule = self._extract_rule(item)

            if rule:

                rule["strength"] = rule.get("strength", 0.0)

                self.rules.append(rule)
                
            rule = self._extract_rule(item, mode="online")
    # =====================================================
    # ADAPTIVE RULE EXTRACTION
    # =====================================================
    def _extract_rule(self, item, mode="default"):

        rule = {
            "id": item.get("action", "unknown"),
            "pattern": item.get("state", {}),
            "strength": 0.0
        }

        return rule
    
    def get_strategies(self):

        # return learned rules sorted by strength if present
        if not hasattr(self, "rules"):
            return []

        return sorted(
            self.rules,
            key=lambda x: x.get("strength", 0),
            reverse=True
        )
    
    def reinforce(self, strategy_id, reward):

        for rule in self.rules:

            if rule.get("id") == strategy_id:

                rule["strength"] += reward * 0.1

class MetaEngineMock:

    def extract(self, data):

        return {
            "rules": {},
            "entities": [],
            "confidence": 0.5
        }