Add --input flag

Author: slimslenderslacks

.gitignore

@@ -11,3 +11,5 @@
 /functions/tree-sitter-clj/target/
 /functions/tree-sitter-clj/.cpcache/
 /.cpcache/
+/.nrepl-port
+/target/

dev/conversations_t.clj

@@ -0,0 +1,37 @@
+(ns conversations-t
+  (:require
+   [babashka.fs :as fs]
+   [clojure.core.async :as async]
+   [docker.main :as main]
+   graph
+   jsonrpc))
+
+(def opts {:platform "darwin"
+           :prompts (fs/file "/Users/slim/docker/labs-ai-tools-for-devs/prompts/curl/README.md")
+           :host-dir "/Users/slim/vonwig/altaservice"})
+
+(alter-var-root
+ #'jsonrpc/notify
+ (fn [_] (partial (if (:jsonrpc opts) jsonrpc/-notify jsonrpc/-println) opts)))
+
+(try
+  (main/with-options opts ["run"])
+  (catch Throwable t
+    (println t)))
+
+(def run-command (main/command opts "run"))
+
+(try
+  (def x (run-command))
+  (catch Throwable t
+    (println t)))
+
+(->> (:messages x)
+     (map :role))
+
+(keys x)
+(count (:messages x))
+
+(def y (async/<!! (graph/stream (graph/chat-with-tools opts))))
+
+(count (:messages y))

graphs/prompts/assets/graph_1729537135592_0.png

@@ -0,0 +1 @@
+-

graphs/prompts/journals/2024_10_01.md

@@ -0,0 +1 @@
+-

graphs/prompts/journals/2024_10_17.md

@@ -50,4 +50,10 @@
 
   ```json
   {"params": {"content": "here is the user prompt"}}
+  ```
+- #### exit
+  
+  Exit the engine
+  ```json
+  {"params": {}}
   ```

graphs/prompts/pages/jsonrpc.md

@@ -0,0 +1,6 @@
+- ![graph.png](../assets/graph_1729537135592_0.png)
+- ```
+  nix-store --query --graph ./result > graph.dot
+  dot -Tpng graph.dot -o graph.png
+  ```
+-

graphs/prompts/pages/runtime graph.md

@@ -1,4 +1,4 @@
-(ns docker.main 
+(ns docker.main
   (:require
    [babashka.fs :as fs]
    [cheshire.core :as json]
@@ -12,7 +12,8 @@
    graph
    jsonrpc
    [logging :refer [warn]]
-   prompts)
+   prompts
+   user-loop)
   (:gen-class))
 
 (set! *warn-on-reflection* true)
@@ -68,6 +69,7 @@
                 :id :stream
                 :assoc-fn (fn [m k _] (assoc m k false))]
                [nil "--debug" "add debug logging"]
+               [nil "--input" "read jsonrpc messages from stdin"]
                [nil "--help" "print option summary"]])
 
 (def output-handler (fn [x]
@@ -131,6 +133,12 @@
       (warn "Docker Desktop not logged in" {}))
     (warn "unable to check Docker Desktop for login" {})))
 
+(defn with-options [opts args]
+  ((validate ::run-args)
+   (merge
+    (apply merge-deprecated opts args)
+    (login-info))))
+
 (defn command [opts & [c :as args]]
   (case c
     "prompts" (fn []
@@ -142,31 +150,39 @@
     "register" (fn []
                  (if-let [{:keys [owner repo path]} (git/parse-github-ref (second args))]
                    (registry/update-registry (fn [m]
-                                                   (update-in m [:prompts] (fnil conj [])
-                                                              {:type (second args)
-                                                               :title (format "%s %s %s"
-                                                                              owner repo
-                                                                              (if path (str "-> " path) ""))})))
+                                               (update-in m [:prompts] (fnil conj [])
+                                                          {:type (second args)
+                                                           :title (format "%s %s %s"
+                                                                          owner repo
+                                                                          (if path (str "-> " path) ""))})))
                    (throw (ex-info "Bad GitHub ref" {:ref (second args)}))))
     "unregister" (fn []
                    (registry/update-registry
                     (fn [m]
                       (update-in m [:prompts] (fn [coll] (remove (fn [{:keys [type]}] (= type (second args))) coll))))))
     "run" (fn []
-            (with-volume
-              (fn [thread-id]
-                (async/<!! ((comp graph/stream graph/chat-with-tools (validate ::run-args))
-                            (-> opts
-                                (assoc :thread-id thread-id)
-                                ((fn [opts] (merge
-                                             (apply merge-deprecated opts (rest args))
-                                             (login-info))))))))
-
-              opts))
+            (let [[in send]
+                  (if (:input opts)
+                    [*in* (constantly true)]
+                    (let [[[w c] in] (user-loop/create-pipe)]
+                      [in (fn []
+                            (w (jsonrpc/request "exit" {} (constantly 1)))
+                            (c))]))]
+              (send)
+              (with-volume
+                (fn [thread-id]
+                  (async/<!!
+                   (user-loop/start-jsonrpc-loop
+                    (user-loop/create-step
+                     (graph/chat-with-tools
+                      (-> (with-options opts (rest args))
+                          (assoc :thread-id thread-id))))
+                    user-loop/state-reducer
+                    in
+                    {})))
+                opts)))
     (fn []
-      ((comp prompts/get-prompts (validate ::run-args)) (merge
-                                                 (apply merge-deprecated opts args)
-                                                 (login-info))))))
+      (prompts/get-prompts (with-options opts args)))))
 
 (defn -main [& args]
   (try

src/docker/main.clj

@@ -54,15 +54,23 @@
 
 (defn start
   "create starting messages, metadata, and functions to bootstrap the thread"
-  [{:keys [prompts] :as opts} _]
+  [{:keys [prompts] :as opts} state]
   (let [c (async/promise-chan)]
     (try
-      (let [new-prompts (prompts/get-prompts opts)]
-        (jsonrpc/notify :prompts {:messages new-prompts})
-        (async/put! c {:metadata (prompts/collect-metadata prompts)
-                       :functions (prompts/collect-functions prompts)
-                       :opts (merge opts {:level (or (:level opts) 0)})
-                       :messages new-prompts}))
+      (async/put!
+       c
+       (-> state
+           (merge
+             {:metadata (prompts/collect-metadata prompts)
+              :functions (prompts/collect-functions prompts)
+              :opts (merge opts {:level (or (:level opts) 0)})})
+           (update 
+             :messages 
+             (fnil concat [])
+             (when (not (seq (:messages state)))
+               (let [new-prompts (prompts/get-prompts opts)]
+                 (jsonrpc/notify :prompts {:messages new-prompts})
+                 new-prompts)))))
       (catch Throwable ex
         (jsonrpc/notify :error {:content
                                 (format "failure for prompt configuration:\n %s" (with-out-str (pprint (dissoc opts :pat :jwt))))
@@ -75,7 +83,7 @@
   [state]
   (let [c (async/promise-chan)]
     ;; this is a normal ending and we try to add a :done key to the state for this
-    (async/put! c (assoc state :done (:finish-reason state)))
+    (async/put! c {:done (:finish-reason state)})
     c))
 
 (defn completion

src/graph.clj

@@ -10,30 +10,51 @@
     PipedInputStream
     PipedOutputStream]))
 
-(declare graph)
+(defn create-step 
+  "compile a function that inserts a user message to-array 
+   to the conversation and stream it through the graph
+   compiled function returns a channel that will emit the final state"
+  [graph]
+  (fn [state]
+    (graph/stream graph state)))
 
-(def do-stream (partial graph/stream graph))
+(defn state-reducer [state s]
+  (update state :messages (fnil conj []) {:role "user" :content s}))
 
-(defn start-jsonrpc-loop [f in m]
+(defn create-test-step []
+  (fn [state]
+    (async/go
+      state)))
+
+(defn start-jsonrpc-loop
+  "start a jsonrpc loop that will inject jsonrpc requests into an
+   ongoing set of state transitions
+   params
+     run-graph    async state -> state
+     reduce-state state, message -> state
+     in           input stream
+     m   initial  state
+   returns
+     the final state"
+  [run-graph state-reducer in m]
   (let [c (jsonrpc/input-stream->input-chan in {})]
     (async/go-loop
-     [state m]
-      (let [message (async/<! c)
-            s (-> message :params :content)]
-        (println "message content: " s)
-        (if (some (partial = s) ["exit" "quit" "q"])
-          state
-          (recur (async/<! (f state s))))))))
+     [next-state (async/<! (run-graph m)) n 0]
+      (println "### loop " n)
+      (let [message (async/<! c)]
+        (cond
+          (= "exit" (:method message))
+          (assoc next-state :jsonrpc-loop-finished :exit)
+          :else
+          (recur (async/<! ((comp run-graph state-reducer) next-state (-> message :params :content))) (inc n)))))))
 
 (def counter (atom 0))
 (defn get-id [] (swap! counter inc))
 
 (def ^{:private true} start-test-loop
-  (partial start-jsonrpc-loop (fn [state s]
-                                (async/go
-                                  (update state :messages (fnil conj []) s)))))
+  (partial start-jsonrpc-loop (create-test-step) state-reducer))
 
-(defn -create-pipe []
+(defn create-pipe []
   ;; Create a PipedInputStream and PipedOutputStream
   (let [piped-out (PipedOutputStream.)
         piped-in  (PipedInputStream. piped-out)
@@ -43,12 +64,17 @@
      piped-in]))
 
 (comment
-  (let [[[w c] in] (-create-pipe)]
-    (async/go (println "ending: " (async/<! (start-test-loop in {}))))
-    (w (jsonrpc/request "prompt" {:content "hello"} get-id))
-    (w (jsonrpc/request "prompt" {:content "hello1"} get-id))
-    (w (jsonrpc/request "prompt" {:content "exit"} get-id))
-    (c)))
+  (println "should be true: "
+    (async/<!!
+      (let [[[w c] in] (create-pipe)]
+        (w (jsonrpc/request "prompt" {:content "hello"} get-id))
+        (w (jsonrpc/request "prompt" {:content "hello1"} get-id))
+        (w (jsonrpc/request "prompt" {:content "hello2"} get-id))
+        (w (jsonrpc/request "exit" {} get-id))
+        (c)
+        (async/go 
+          (println "ending: " (async/<! (start-test-loop in {})))
+          true)))))
 
 (comment
   ;; an input stream is something from which we can read bytes