Merge pull request #1 from CognitiveComputingLab/dev

basic State and Action nodes

doc/index.rst

@@ -10,23 +10,6 @@ Welcome to Meta-Prompting's documentation!
    auto_examples/index.rst
    api_summary
 
-.. ...add more elements to table of contents
-
-You can include code as part of the documentation
-
-   >>> print("Hello World")
-   Hello World
-
-which can be tested by running ``make doctest``. This is also run by the GitHub action to build the documentation.
-
-You can also include executable example files with code and text, which are shown in the :doc:`auto_examples/index`.
-The ``test_examples.py`` unittest automatically runs these examples to check for errors.
-
-.. autoclass:: metaprompting.myclass.MyClass
-   :members:
-   :noindex:
-
-
 .. Indices and tables
 .. ==================
 

examples/plot_example.py

@@ -1,14 +1,54 @@
 """
-Example
-===========================
+Default Action and State nodes
+==============================
 
-An example Python file.
-"""
+A simple example using the :class:`~metaprompting.base.DefaultAction` and :class:`~metaprompting.base.DefaultState`
+classes."""
 
 # %%
-# Some Python Code
-# ------------------------
-#
-# Example code will be executed and shown in the documentation:
+# Define derived classes to make the call dynamics visible
+
+from metaprompting.base import DefaultAction, DefaultState
+
+
+class VerboseAction(DefaultAction):
+
+    def input_trigger(self, input):
+        print(f"{self} was triggered by {input}")
+        super().input_trigger(input)
+
+    def execute(self, *args, **kwargs):
+        print(f"executing {self}")
+        super().execute(*args, **kwargs)
+
+
+class VerboseState(DefaultState):
+
+    def update(self, text):
+        print(f"updating {self}")
+        super().update(text)
 
-print("Hello World")
+
+# %%
+# Create state nodes
+root_1 = VerboseState()
+root_2 = VerboseState()
+root_3 = VerboseState()
+leaf_1 = VerboseState()
+leaf_2 = VerboseState()
+
+# %%
+# Create action nodes, which auto-connects states
+action1 = VerboseAction(input_states=[root_1, root_2, root_3], output_state=leaf_1)
+action2 = VerboseAction(input_states=[root_3, root_2, root_1], output_state=leaf_2)
+
+# %%
+# Update root state nodes, which triggers a cascade to leaf nodes
+root_1.update("smoke")
+root_2.update(" and ")
+root_3.update("mirrors")
+
+# %%
+# Print output of leaf nodes
+print(leaf_1.value)
+print(leaf_2.value)

metaprompting/base.py

@@ -0,0 +1,121 @@
+from abc import ABC, abstractmethod
+
+
+class LLM(ABC):
+
+    @abstractmethod
+    def __call__(self, *args, **kwargs):
+        """
+        Call the LLM with given arguments and return its output.
+        """
+        raise NotImplemented
+
+
+class Action(ABC):
+
+    def __init__(self, input_states=None, output_state=None):
+        """
+        An executable node that takes zero or more input_states, executes an action, and returns its output to the
+        output_state.
+
+        :param input_states: Iterable over input :class:`~State`\s
+        :param output_state: Output :class:`~State`
+        """
+        self.input_states = input_states
+        self.output_state = output_state
+
+    def input_trigger(self, input):
+        """
+        Trigger the :class:`~Action` from a specific input, typically when the input has been updated.
+
+        :param input: input :class:`~State`
+        """
+        pass
+
+    @abstractmethod
+    def execute(self, *args, **kwargs):
+        """
+        Excecute the :class:`~Action` with given arguments and pass on the output to the output :class:`~State`.
+        """
+        raise NotImplementedError
+
+
+class State(ABC):
+
+    def __init__(self, input_action=None, output_actions=None):
+        """
+        A static node holding information generated by an input_action. It may pass on the information to zero or
+        more output_actions.
+
+        :param input_action: Input :class:`~Action`
+        :param output_actions: Iterable over output :class:`~Action`\s
+        """
+        self.input_action = input_action
+        self.output_actions = output_actions
+
+    def trigger_outputs(self):
+        """
+        Trigger all outputs of this :class:`~State`. Should typically be called at the end of :meth:`~update`.
+        """
+        for output in self.output_actions:
+            output.input_trigger(self)
+
+    @abstractmethod
+    def update(self, *args, **kwargs):
+        raise NotImplementedError
+
+
+class DefaultAction(Action):
+
+    def __init__(self, input_states=None, output_state=None, auto_connect=True):
+        if input_states is None:
+            input_states = []
+        super().__init__(input_states=input_states, output_state=output_state)
+        # remember update status of inputs
+        self.inputs_updated = {i: False for i in self.input_states}
+        # connect inputs and outputs
+        if auto_connect:
+            for i in self.input_states:
+                i.output_actions.append(self)
+            self.output_state.input_action = self
+
+    def input_trigger(self, input):
+        # remember updated inputs
+        try:
+            self.inputs_updated[input] = True
+        except KeyError:
+            raise KeyError("Given input is not an input of this node")
+        # execute if all inputs were updated
+        for val in self.inputs_updated.values():
+            if not val:
+                break
+        else:
+            # reset input flags
+            for key in self.inputs_updated.keys():
+                self.inputs_updated[key] = False
+            # execute this action
+            self.execute()
+
+    def execute(self, *args, **kwargs):
+        # simple action: concatenate inputs with " + " in between
+        out = None
+        for i in self.input_states:
+            if out is None:
+                out = i.value
+            else:
+                out = out + i.value
+        # update output
+        self.output_state.update(out)
+
+
+class DefaultState(State):
+
+    def __init__(self, input_action=None, output_actions=None):
+        if output_actions is None:
+            output_actions = []
+        super().__init__(input_action=input_action, output_actions=output_actions)
+        self.value = None
+
+    def update(self, value):
+        self.value = value
+        self.trigger_outputs()