Merge pull request #852 from trueagi-io/event_agent

agent with events

Author: Necr0x0Der

python/hyperon/exts/agents/__init__.py

@@ -1,3 +1,3 @@
-from .agent_base import AgentObject
+from .agent_base import AgentObject, EventAgent
 from .agent_base import agent_atoms
-from .agent_base import BaseListeningAgent
+

python/hyperon/exts/agents/agent_base.py

@@ -1,5 +1,6 @@
 from hyperon import *
 from hyperon.ext import register_atoms
+import queue
 
 '''
 This is very preliminary and incomplete PoC version.
@@ -24,6 +25,7 @@
 
 import threading
 from queue import Queue
+
 class StreamMethod(threading.Thread):
     def __init__(self, method, args):
         super().__init__() #daemon=True
@@ -32,16 +34,33 @@ def __init__(self, method, args):
         self.args = args
 
     def run(self):
-        for r in self.method(*self.args):
-            self._result.put(r)
+        # FIXME? if we raise the exception here, the thread is not stopped
+        # but should we put Error into the result?
+        try:
+            for r in self.method(*self.args):
+                self._result.put(r)
+        except Exception as e:
+            self._result.put(E(S('Error'), ValueAtom(self.args), ValueAtom(e)))
 
     def __iter__(self):
         return self
 
     def __next__(self):
-        if self._result.empty() and not self.is_alive():
-            raise StopIteration
-        return self._result.get()
+        try:
+            while self.is_alive():
+                yield self._result.get_nowait()
+        except queue.Empty:
+            pass
+        raise StopIteration
+
+def _try_atom2str(val):
+    if isinstance(val, GroundedAtom):
+        val = val.get_object().content
+    elif isinstance(val, SymbolAtom):
+        val = val.get_name()
+    if val is None or isinstance(val, str):
+        return val
+    return repr(val)
 
 
 class AgentObject:
@@ -60,16 +79,17 @@ class AgentObject:
     _name = None
 
     @classmethod
-    def get_agent_atom(cls, metta, *args, unwrap=True):
+    def get_agent_atom(cls, metta, *args, unwrap=True, **kwargs):
         # metta and unwrap are not passed to __init__, because
         # they are needed only for __metta_call__, so children
         # classes do not need to pass them to super().__init__
+        # TODO: catch exceptions and turn them into error messages?
         if unwrap:
             # a hacky way to unwrap args
-            agent_atom = OperationObject("_", cls).execute(*args)[0]
+            agent_atom = OperationObject("_", cls).execute(*args, **kwargs)[0]
             agent = agent_atom.get_object().content
         else:
-            agent = cls(*args)
+            agent = cls(*args, **kwargs)
         if metta is not None:
             if hasattr(agent, '_metta') and agent._metta is not None:
                 raise RuntimeError(f"MeTTa is already defined for {agent}")
@@ -88,13 +108,6 @@ def agent_creator_atom(cls, metta=None, unwrap=True):
     def name(cls):
         return cls._name if cls._name is not None else str(cls)
 
-    def _try_unwrap(self, val):
-        if val is None or isinstance(val, str):
-            return val
-        if isinstance(val, GroundedAtom):
-            return str(val.get_object().content)
-        return repr(val)
-
     def __init__(self, path=None, atoms={}, include_paths=None, code=None):
         if path is None and code is None:
             # purely Python agent
@@ -103,12 +116,12 @@ def __init__(self, path=None, atoms={}, include_paths=None, code=None):
         if isinstance(path, ExpressionAtom):# and path != E():
             code = path
         elif path is not None:
-            path = self._try_unwrap(path)
+            path = _try_atom2str(path)
             with open(path, mode='r') as f:
                 code = f.read()
         # _code can remain None if the agent uses parent runner (when called from MeTTa)
         self._code = code.get_children()[1] if isinstance(code, ExpressionAtom) else \
-            self._try_unwrap(code)
+            _try_atom2str(code)
         self._atoms = atoms
         self._include_paths = include_paths
         self._context_space = None
@@ -118,7 +131,11 @@ def _create_metta(self):
         if self._code is None:
             return None
         self._init_metta()
-        self._load_code()  # TODO: check that the result contains only units
+        result = self._load_code()
+        for r in result:
+            if r != [E()]:
+                # TODO: catch only errors? unwrap error messages?
+                raise RuntimeError(r)
 
     def _init_metta(self):
         ### =========== Creating MeTTa runner ===========
@@ -137,8 +154,11 @@ def _init_metta(self):
         self._metta = metta
 
     def _load_code(self):
-        return self._metta.run(self._code) if isinstance(self._code, str) else \
+        if isinstance(self._code, str):
+            return self._metta.run(self._code)
+        else:
             self._metta.space().add_atom(self._code)
+            return []
 
     def __call__(self, atom):
         if self._unwrap or self._metta is None:
@@ -152,80 +172,156 @@ def is_daemon(self):
 
     def __metta_call__(self, *args):
         call = True
+        unwrap = self._unwrap
         method = self.__call__
         if len(args) > 0 and isinstance(args[0], SymbolAtom):
             n = args[0].get_name()
             if n[0] == '.' and hasattr(self, n[1:]):
                 method = getattr(self, n[1:])
                 args = args[1:]
                 call = False
-        if self._unwrap:
+                # FIXME? Python methods called via . are supposed to be purely Python
+                unwrap = True
+        if unwrap:
             method = OperationObject(f"{method}", method).execute
-        st = StreamMethod(method, args)
-        st.start()
-        # We don't return the stream here; otherwise it will be consumed immediately.
-        # If the agent itself would be StreamMethod, its results could be accessbile.
-        # Here, they are lost (TODO?).
         if call and self.is_daemon():
+            st = StreamMethod(method, args)
+            st.start()
+            # We don't return the stream here; otherwise it will be consumed immediately.
+            # If the agent itself would be StreamMethod, its results could be accessbile.
+            # Here, they are lost (TODO?).
             return [E()]
-        return st
+        # NOTE: previously, `StreamMethod` object was created always and returned here
+        # instead of calling `method` directly. The idea was that agents could consume
+        # a part of the stream from other agents they are calling, but this cannot work
+        # without support from MeTTa, because StreamMethod also calls `method` (just in
+        # a separate thread), thus, it should be MeTTa itself, which turns `match` results
+        # into a stream. Thus, we return the result directly now
+        return method(*args)
 
-class BaseListeningAgent(AgentObject):
-    def __init__(self, path=None, atoms={}, include_paths=None, code=None):
+class EventAgent(AgentObject):
+
+    StopEvent = object()
+
+    def __init__(self, path=None, atoms={}, include_paths=None, code=None, event_bus=None):
+        if event_bus is not None:
+            # EventAgent is not a daemon by default: although its `event_processor` runs in a thread
+            # and should be stopped, other non-event methods are more convenient to call directly to
+            # get their results back in a caller
+            # self.daemon = True
+            atoms = {**atoms}
+            atoms['&event_bus'] = event_bus if isinstance(event_bus, Atom) else ValueAtom(event_bus)
+            atoms['queue-subscription'] = OperationAtom('queue-subscription', self.queue_subscription, unwrap=False)
+            atoms['has-event-bus'] = OperationAtom('has-event-bus',
+                lambda: self.event_bus is not None and \
+                        hasattr(self.event_bus, "create_subscription"))
+        self.event_bus = event_bus.get_object().value if isinstance(event_bus, GroundedAtom) else event_bus
         super().__init__(path, atoms, include_paths, code)
-        self.messages = Queue()
+        # Even if there is no event bus, events can be submitted by child class methods
+        self.events = Queue()
         self.running = False
-        self._output = Queue()
-        self.lock = threading.RLock()
-        self.said = False
+        self.outputs = Queue()
 
-    def start(self,  *args):
+    def _init_metta(self):
+        # NOTE: atm, there is no utility for the base agent to import `agents` by default,
+        # but event agents in metta typically need it
+        super()._init_metta()
+        if self._metta is not None:
+            self._metta.run("! (import! &self agents)")
+
+    def start(self, *args):
+        if self.running:
+            raise RuntimeError("Currently, EventAgent is supposed to be running in one thread")
         if not args:
             args = ()
         self.running = True
-        st = StreamMethod(self.messages_processor, args)
+        st = StreamMethod(self.event_processor, args)
         st.start()
 
-    def message_processor(self, message, *args):
-        yield None
+    # TODO? it's similar to `input` in former BaseListeningAgent, but
+    # it uses event_id and func, which might be inconvenient...
+    # either event_id or func can be made excessive, so we could
+    # turn this function into a queued call to an agent method...
+    def recv_queue_event(self, event_id, func, *args):
+        # TODO? we can keep {event_id: func} dict in self
+        # instead of having `func` as an argument here
+        self.events.put((event_id, func, args))
 
-    def handle_event(self):
-        pass
+    def queue_subscription(self, event_id, func):
+        '''
+        Subscribing `recv_queue_event` to the given event channel
+        to put incoming events to queue processed in a separate
+        thread instead of subscribing `func` directly
+        '''
+        self.event_bus.create_subscription(_try_atom2str(event_id),
+            lambda *args: self.recv_queue_event(event_id, func, *args))
+        return [E()]
 
-    def messages_processor(self, *args):
+    def event_processor(self, *args):
+        # `*args` received on `start`
         while self.running:
-            self.handle_event()
-            if not self.messages.empty():
-                self.clear_output()
-                with self.lock:
-                    m = self.messages.get()
-                    self.said = False
-                for resp in self.message_processor(m, *args):
-                    with self.lock:
-                        self._output.put(resp)
-        return []
+            # TODO? func can be a Python function?
+            (event_id, func, args) = self.events.get()
+            if event_id is self.StopEvent:
+                break
+            # Wrapping into ValueAtom if arg is not an atom yet
+            resp = self._metta.evaluate_atom(E(func,
+                *[a if isinstance(a, Atom) else ValueAtom(a) for a in args]))
+            # TODO? do we need `outputs` here? we may want to publish `resp` to a certain channel
+            # or let `func` to do this direclty...
+            # ??? self.clear_outputs()
+            for r in resp:
+                self.outputs.put(r)
 
     def stop(self):
+        self.events.put((self.StopEvent, None, None))
         self.running = False
-        return []
 
-    def input(self, msg):
-        with self.lock:
-            self.messages.put(msg)
-        return []
+    # TODO? choose the model of dealing with outputs... do we need them at all?
+    def clear_outputs(self):
+        try:
+            while True:
+                self.outputs.get_nowait()
+        except queue.Empty:
+            pass
 
     def get_output(self):
-        while not self._output.empty():
-            with self.lock:
-                self.said = True
-            yield self._output.get()
+        try:
+            while True:
+                yield self.outputs.get_nowait()
+        except queue.Empty:
+            pass
+
+def subscribe_metta_func(metta: MeTTa, event_bus: GroundedAtom, event_id: Atom, func: Atom): #metta,
+    event_bus = event_bus.get_object().content
+    if not hasattr(event_bus, "create_subscription"):
+        raise RuntimeError("Event bus should have create_subscription method")
+    event_id = _try_atom2str(event_id)
+    event_bus.create_subscription(event_id,
+        lambda *args: metta.evaluate_atom(E(func, *[ValueAtom(a) for a in args])))
+    return [E()]
 
-    def clear_output(self):
-        with self.lock:
-            self._output = Queue()
+# The function to be called from MeTTa
+def publish_event(event_bus: Atom, event_id: Atom, content: Atom):
+    assert isinstance(event_bus, GroundedAtom), f"{event_bus} is not a grounded object"
+    event_bus = event_bus.get_object().value
+    event_id = _try_atom2str(event_id)
+    # FIXME? We want to be able to pass Atoms as event content, but not
+    # any event bus can support this... or should we always use wrappers,
+    # which provide this support?
+    event_bus.publish(event_id, content)
+    return [E()]
 
 @register_atoms(pass_metta=True)
 def agent_atoms(metta):
     return {
         r"create-agent": AgentObject.agent_creator_atom(unwrap=False),
+        # We have to avoid generic agent_creator_atom here to pass ordered parameters
+        r"event-agent": OperationAtom('event-agent',
+            lambda path=None, event_bus=None: EventAgent.get_agent_atom(None, unwrap=False, path=path, event_bus=event_bus),
+            unwrap=False),
+        r"direct-subscription": OperationAtom('direct-subscription',
+            lambda *args: subscribe_metta_func(metta, *args), unwrap=False),
+        r"publish-event": OperationAtom('publish-event',
+            publish_event, unwrap=False)
     }

python/hyperon/exts/agents/events/__init__.py

@@ -0,0 +1,3 @@
+from .basic_bus import event_atoms
+
+

python/hyperon/exts/agents/events/basic_bus.py

@@ -0,0 +1,24 @@
+from collections import defaultdict
+import logging
+from hyperon import OperationAtom
+from hyperon.ext import register_atoms
+
+class BasicEventBus:
+    def __init__(self):
+        self.subscriptions = defaultdict(list)
+        self.is_running = True
+    def terminate(self):
+        self.is_running = False
+    def create_subscription(self, topic, cb):
+        self.subscriptions[topic].append(cb)
+    def publish(self, topic, msg):
+        for cb in self.subscriptions[topic]:
+            cb(msg)
+    def get_logger(self):
+        return logging.getLogger("MockNode")
+
+@register_atoms
+def event_atoms():
+    return {
+        r"basic-event-bus": OperationAtom('basic-event-bus', BasicEventBus)
+    }

python/hyperon/exts/agents/tests/test_0_passive_agents.metta

@@ -0,0 +1,33 @@
+!(import! &self agents)
+
+; checking that we can create passive agents and call their functions
+; passive agents do nothing unless they are directly called
+
+; basic agent from file
+!(assertEqual
+  ((create-agent agent.metta) (g 3))
+  9)
+
+; basic agent from expression
+!(assertEqual
+  ((create-agent (quote (= (f $x) (+ 1 $x)))) (f 10))
+  11)
+
+; event-agent has the same behavior when event_bus is not provided
+!(add-reduct &self
+  (= (my-agent) (event-agent agent.metta)))
+; multiple consequent calls should work
+!(assertEqual
+   ((my-agent) (h 3 5))
+   15)
+!(assertEqual
+   ((my-agent) (g 4))
+   16)
+
+; atm, we can access Python methods of the agent, although
+; they are all supposed to be unwrap=True
+; also, event-agent without event bus is not a daemon
+!(assertEqual
+   ((my-agent) .is_daemon)
+   False)
+