Merge pull request dora-rs#451 from dora-rs/llm-example

Llm example

examples/multiple-daemons/run.rs

@@ -55,7 +55,7 @@ async fn main() -> eyre::Result<()> {
         } else if retries > 20 {
             bail!("daemon not connected after {retries} retries");
         } else {
-            std::thread::sleep(Duration::from_millis(100));
+            std::thread::sleep(Duration::from_millis(500));
             retries += 1
         }
     }

examples/python-operator-dataflow/README.md

@@ -13,21 +13,46 @@ The [`dataflow.yml`](./dataflow.yml) defines a simple dataflow graph with the fo
 ## Getting started
 
 ```bash
-cargo run --example python-dataflow 
+pip install -r requirements.txt
+cargo run --example python-operator-dataflow
 ```
 
 ## Installation
 
-To install, you should run the `install.sh` script.
-
 ```bash
-install.sh
+conda create -n example_env python=3.11
+conda activate test_env
+pip install -r requirements.txt
+pip install -r requirements_llm.txt
 ```
 
-## Run the dataflow as a standalone
+## Run the dataflow
+
+- Start the object detection dataflow alone:
+
+```bash
+dora start dataflow.yml
+```
 
-- Start the `dora-coordinator`:
+- Start the llm dataflow (Only works on Windows and Linux):
 
+```bash
+dora start dataflow_llm.yml
 ```
-../../target/release/dora-daemon --run-dataflow dataflow.yml
+
+Within the window you can ask question such as:
+
+```bash
+ask how are you
+change bounding box plot to red
+change confidence value to percentage
+change object detection to only detect person
+send 200 200 200 400 to topic line
+record
 ```
+
+The keyboard, microphone, whisper node, works in a very similar fashion as the object detection dataflow and I'll let you check it out by yourself.
+
+The code modification flow works by first comparing an instruction with a vectordb of operators source code and then feeding the most similar operator to an llm with the instruction for code modification.
+
+The end result is then saved using a file saver.

examples/python-operator-dataflow/dataflow.yml

@@ -9,8 +9,8 @@ nodes:
 
   - id: object_detection
     operator:
-      python: object_detection.py
       send_stdout_as: stdout
+      python: object_detection.py
       inputs:
         image: webcam/image
       outputs:
@@ -23,4 +23,4 @@ nodes:
       inputs:
         image: webcam/image
         bbox: object_detection/bbox
-        object_detection_stdout: object_detection/stdout
+        assistant_message: object_detection/stdout

examples/python-operator-dataflow/dataflow_llm.yml

@@ -0,0 +1,87 @@
+nodes:
+  - id: webcam
+    operator:
+      python: webcam.py
+      inputs:
+        tick: dora/timer/millis/50
+      outputs:
+        - image
+
+  - id: object_detection
+    operator:
+      python: object_detection.py
+      inputs:
+        image: webcam/image
+      outputs:
+        - bbox
+
+  - id: plot
+    operator:
+      python: plot.py
+      inputs:
+        image: webcam/image
+        bbox: object_detection/bbox
+        line: llm/line
+        keyboard_buffer: keyboard/buffer
+        user_message: keyboard/submitted
+        assistant_message: llm/assistant_message
+
+  ## Speech to text
+  - id: keyboard
+    custom:
+      source: keyboard_op.py
+      outputs:
+        - buffer
+        - submitted
+        - record
+        - ask
+        - send
+        - change
+      inputs:
+        recording: whisper/text
+
+  - id: microphone
+    operator:
+      python: microphone_op.py
+      inputs:
+        record: keyboard/record
+      outputs:
+        - audio
+
+  - id: whisper
+    operator:
+      python: whisper_op.py
+      inputs:
+        audio: microphone/audio
+      outputs:
+        - text
+
+  ## Code Modifier
+  - id: vectordb
+    operator:
+      python: sentence_transformers_op.py
+      inputs:
+        query: keyboard/change
+        saved_file: file_saver/saved_file
+      outputs:
+        - raw_file
+
+  - id: llm
+    operator:
+      python: llm_op.py
+      inputs:
+        code_modifier: vectordb/raw_file
+        assistant: keyboard/ask
+        message_sender: keyboard/send
+      outputs:
+        - modified_file
+        - line
+        - assistant_message
+
+  - id: file_saver
+    operator:
+      python: file_saver_op.py
+      inputs:
+        file: llm/modified_file
+      outputs:
+        - saved_file

examples/python-operator-dataflow/file_saver_op.py

@@ -0,0 +1,44 @@
+import pyarrow as pa
+
+from dora import DoraStatus
+
+
+class Operator:
+    """
+    Infering object from images
+    """
+
+    def __init__(self):
+        self.last_file = ""
+        self.last_path = ""
+        self.last_netadata = None
+
+    def on_event(
+        self,
+        dora_event,
+        send_output,
+    ) -> DoraStatus:
+        if dora_event["type"] == "INPUT" and dora_event["id"] == "file":
+            input = dora_event["value"][0].as_py()
+
+            with open(input["path"], "r") as file:
+                self.last_file = file.read()
+                self.last_path = input["path"]
+                self.last_metadata = dora_event["metadata"]
+            with open(input["path"], "w") as file:
+                file.write(input["raw"])
+
+            send_output(
+                "saved_file",
+                pa.array(
+                    [
+                        {
+                            "raw": input["raw"],
+                            "path": input["path"],
+                            "origin": dora_event["id"],
+                        }
+                    ]
+                ),
+                dora_event["metadata"],
+            )
+        return DoraStatus.CONTINUE

examples/python-operator-dataflow/keyboard_op.py

@@ -0,0 +1,65 @@
+from pynput import keyboard
+from pynput.keyboard import Key, Events
+import pyarrow as pa
+from dora import Node
+
+
+node = Node()
+buffer_text = ""
+ctrl = False
+submitted_text = []
+cursor = 0
+
+NODE_TOPIC = ["record", "send", "ask", "change"]
+
+with keyboard.Events() as events:
+    while True:
+        dora_event = node.next(0.01)
+        if (
+            dora_event is not None
+            and dora_event["type"] == "INPUT"
+            and dora_event["id"] == "recording"
+        ):
+            buffer_text += dora_event["value"][0].as_py()
+            node.send_output("buffer", pa.array([buffer_text]))
+            continue
+
+        event = events.get(1.0)
+        if event is not None and isinstance(event, Events.Press):
+            if hasattr(event.key, "char"):
+                cursor = 0
+                buffer_text += event.key.char
+                node.send_output("buffer", pa.array([buffer_text]))
+            else:
+                if event.key == Key.backspace:
+                    buffer_text = buffer_text[:-1]
+                    node.send_output("buffer", pa.array([buffer_text]))
+                elif event.key == Key.esc:
+                    buffer_text = ""
+                    node.send_output("buffer", pa.array([buffer_text]))
+                elif event.key == Key.enter:
+                    node.send_output("submitted", pa.array([buffer_text]))
+                    first_word = buffer_text.split(" ")[0]
+                    if first_word in NODE_TOPIC:
+                        node.send_output(first_word, pa.array([buffer_text]))
+                    submitted_text.append(buffer_text)
+                    buffer_text = ""
+                    node.send_output("buffer", pa.array([buffer_text]))
+                elif event.key == Key.ctrl:
+                    ctrl = True
+                elif event.key == Key.space:
+                    buffer_text += " "
+                    node.send_output("buffer", pa.array([buffer_text]))
+                elif event.key == Key.up:
+                    if len(submitted_text) > 0:
+                        cursor = max(cursor - 1, -len(submitted_text))
+                        buffer_text = submitted_text[cursor]
+                        node.send_output("buffer", pa.array([buffer_text]))
+                elif event.key == Key.down:
+                    if len(submitted_text) > 0:
+                        cursor = min(cursor + 1, 0)
+                        buffer_text = submitted_text[cursor]
+                        node.send_output("buffer", pa.array([buffer_text]))
+        elif event is not None and isinstance(event, Events.Release):
+            if event.key == Key.ctrl:
+                ctrl = False