Merge pull request #28 from sonnhfit/feature/add-immune-system

immune system

docs/source/dev.md

@@ -0,0 +1,5 @@
+Dev 
+- fix coding convention
+```
+ruff check --fix .
+```

sonagent/agent.py

@@ -43,7 +43,6 @@ def __init__(self, memory_path, skills, config: dict) -> None:
         self.skills_dict = {}
         logger.info("--------- Start Done.---------")
 
-
         openai = self.config.get('openai')
         if openai.get('api_type', None) == 'openai':
             self.chat_service = OpenAIChatCompletion(

sonagent/immune/immune.py

@@ -0,0 +1,60 @@
+import logging
+from datetime import datetime
+from typing import Any, Dict, Optional
+
+from sonagent.loggers import bufferHandler
+from sonagent.utils.datetime_helpers import format_date
+
+logger = logging.getLogger(__name__)
+
+
+class ImmuneSystem:
+    """
+    Class for Self Immune System that allows self-monitoring and self-recovery of the system.
+    """
+    error_logs = []
+
+    @staticmethod
+    def get_logs(limit: Optional[int]) -> Dict[str, Any]:
+        """Returns the last X logs"""
+        if limit:
+            buffer = bufferHandler.buffer[-limit:]
+        else:
+            buffer = bufferHandler.buffer
+        records = [[format_date(datetime.fromtimestamp(r.created)),
+                   r.created * 1000, r.name, r.levelname,
+                   r.message + ('\n' + r.exc_text if r.exc_text else '')]
+                   for r in buffer]
+        return {'log_count': len(records), 'logs': records}
+
+    def immune_scan(self) -> None:
+        """
+        Scan the system for any anomalies.
+        """
+        logs = ImmuneSystem.get_logs(5)
+
+        # Check for any errors in the logs
+        errors = [log for log in logs['logs'] if log[3] == 'ERROR']
+
+        errors_detected = []
+        for error in errors:
+            if error not in ImmuneSystem.error_logs:
+                ImmuneSystem.error_logs.append(error)
+                errors_detected.append(error)
+
+        if len(errors_detected) > 0:
+            logger.info(f"Errors detected: {errors_detected}")
+        else:
+            logger.debug("No errors detected.")
+
+    def immune_recover(self) -> None:
+        """
+        Recover the system from any anomalies.
+        """
+        if ImmuneSystem.error_logs:
+            print(f"Recovering from errors: {ImmuneSystem.error_logs}")
+            logger.info(f"Recovering from errors: {ImmuneSystem.error_logs}")
+            ImmuneSystem.error_logs = []
+        else:
+            print("No errors detected.")
+            logger.info("No errors detected.")

sonagent/sonbot.py

@@ -213,4 +213,4 @@ def notify_status(self, msg: str, msg_type=RPCMessageType.STATUS) -> None:
         self.rpc.send_msg({
             'type': msg_type,
             'status': msg
-        })
+        })

sonagent/worker.py

@@ -11,6 +11,7 @@
 from sonagent.enums.enums import State
 from sonagent.enums.rpcmessagetype import RPCMessageType
 from sonagent.exceptions import OperationalException, TemporaryError
+from sonagent.immune.immune import ImmuneSystem
 from sonagent.sonbot import SonBot
 
 logger = logging.getLogger(__name__)
@@ -28,10 +29,12 @@ def __init__(self, args: Dict[str, Any], config: Optional[dict] = None) -> None:
         self._init(False)
 
         self._heartbeat_msg: float = 0
+        self._immune = ImmuneSystem()
 
         # Tell systemd that we completed initialization phase
         self._notify("READY=1")
 
+
     def _init(self, reconfig: bool) -> None:
         # Init the instance of the bot
         self.sonbot = SonBot(self._config, args=self._args)
@@ -106,6 +109,9 @@ def _worker(self, old_state: Optional[State]) -> State:
                 )
                 self._heartbeat_msg = now
 
+                # Ping Immune System for scan
+                self._immune.immune_scan()
+
         return state
 
     def _throttle(