Working demo without interruptions

Author: ovistoica

TODO.org

@@ -106,6 +106,7 @@ CLOSED: [2025-01-20 Lun 07:43]
 
 * TODO add core.async.flow support
 :LOGBOOK:
+CLOCK: [2025-01-25 Sat 16:35]
 CLOCK: [2025-01-25 Sat 15:18]--[2025-01-25 Sat 15:43] =>  0:25
 CLOCK: [2025-01-25 Sat 11:14]--[2025-01-25 Sat 11:39] =>  0:25
 CLOCK: [2025-01-25 Sat 09:50]--[2025-01-25 Sat 10:15] =>  0:25

core/src/voice_fn/experiments/flow.clj

@@ -327,8 +327,8 @@
   "Processor that streams audio out in real time so we can account for
   interruptions."
   (flow/process
-    {:describe (fn [] {:ins {:in "Channel for audio output frames "}
-                       :outs {:out "Channel on which serialized buffered output is put"}})
+    {:describe (fn [] {:ins {:in "Channel for audio output frames "
+                             :sys-in "Channel for system messages"}})
      :params {:transport/out-chan "Channel on which to put buffered serialized audio"
               :audio.out/duration-ms "Duration of each audio chunk. Defaults to 20ms"
               :transport/supports-interrupt? "Whether the processor supports interrupt or not"}
@@ -369,9 +369,11 @@
 
                     :else [state]))}))
 
-(def gdef
+(defn make-twilio-flow
+  [in out]
   {:procs
-   {:transport-in {:proc twilio-transport-in}
+   {:transport-in {:proc twilio-transport-in
+                   :args {:transport/in-ch in}}
     :deepgram-transcriptor {:proc deepgram-processor
                             :args {:transcription/api-key (secret [:deepgram :api-key])
                                    :transcription/interim-results? true
@@ -418,13 +420,7 @@
                             :audio.out/channels 1
                             :audio.out/duration-ms 20}}
     :realtime-out {:proc realtime-transport-out-processor
-                   :args {:transport/out-chan (a/chan 1024)}}
-
-    :print-sink {:proc (flow/process
-                         {:describe (fn [] {:ins {:in "Channel for receiving transcriptions"}})
-                          :transform (fn [_ _ frame]
-                                       (when (frame/audio-output-raw? frame)
-                                         (t/log! {:id :print-sink :level :info} ["RESULT: " (:frame/data frame)])))})}}
+                   :args {:transport/out-chan out}}}
 
    :conns [[[:transport-in :sys-out] [:deepgram-transcriptor :sys-in]]
            [[:transport-in :out] [:deepgram-transcriptor :in]]
@@ -440,6 +436,7 @@
            [[:llm-sentence-assembler :out] [:tts :in]]
 
            [[:tts :out] [:audio-splitter :in]]
+           [[:transport-in :sys-out] [:realtime-out :sys-in]]
            [[:audio-splitter :out] [:realtime-out :in]]]})
 
 (comment

core/src/voice_fn/processors/elevenlabs.clj

@@ -39,22 +39,13 @@
                              :output_format (encoding->elevenlabs encoding sample-rate)})))
 
 (comment
-  (make-elevenlabs-url {:audio-in/sample-rate 8000
-                        :audio-in/encoding :ulaw
-                        :audio-in/channels 1
-                        :audio-in/sample-size-bits 8
-                        :audio-out/sample-rate 8000
-                        :audio-out/encoding :ulaw
-                        :audio-out/bitrate 64000
-                        :audio-out/sample-size-bits 8
-                        :audio-out/channels 1
-                        :pipeline/language :ro}
-                       {:elevenlabs/api-key (secrets/secret [:elevenlabs :api-key])
-                        :elevenlabs/model-id "eleven_flash_v2_5"
-                        :elevenlabs/voice-id "7sJPxFeMXAVWZloGIqg2"
-                        :voice/stability 0.5
-                        :voice/similarity-boost 0.8
-                        :voice/use-speaker-boost? true}))
+  (make-elevenlabs-ws-url
+    {:elevenlabs/api-key (secrets/secret [:elevenlabs :api-key])
+     :elevenlabs/model-id "eleven_flash_v2_5"
+     :elevenlabs/voice-id "7sJPxFeMXAVWZloGIqg2"
+     :voice/stability 0.5
+     :voice/similarity-boost 0.8
+     :voice/use-speaker-boost? true}))
 
 (defn begin-stream-message
   [{:voice/keys [stability similarity-boost use-speaker-boost?]
@@ -123,7 +114,7 @@
                             type
                             pipeline
                             processor-config)
-              new-conn @(ws/websocket (make-elevenlabs-url (:pipeline/config @pipeline) processor-config)
+              new-conn @(ws/websocket (make-elevenlabs-ws-url processor-config)
                                       conn-config)]
           (swap! pipeline assoc-in [type :websocket/conn] new-conn)
           (t/log! :debug "Elevenlabs connection ready"))))))

examples/src/voice_fn_examples/twilio_websocket.clj

@@ -19,7 +19,7 @@
    [ring.websocket :as ws]
    [taoensso.telemere :as t]
    [voice-fn.core]
-   [voice-fn.experiments.flow :refer [gdef]]
+   [voice-fn.experiments.flow :refer [make-twilio-flow]]
    [voice-fn.pipeline :as pipeline]
    [voice-fn.processors.elevenlabs]
    [voice-fn.processors.groq]
@@ -187,7 +187,9 @@
 (defn twilio-ws-handler-flow
   [req]
   (assert (ws/upgrade-request? req) "Must be a websocket request")
-  (let [fl (flow/create-flow gdef)]
+  (let [in (a/chan 1024)
+        out (a/chan 1024)
+        fl (flow/create-flow (make-twilio-flow in out))]
     (reset! dbg-flow fl)
     {::ws/listener
      {:on-open (fn on-open [socket]
@@ -196,14 +198,20 @@
                      (when-let [[msg c] (a/alts! [report-chan error-chan])]
                        (when (map? msg)
                          (t/log! {:level :debug :id (if (= c error-chan) :error :report)} msg))
+                       (recur)))
+                   (a/go-loop []
+                     (when-let [output (a/<! out)]
+                       (ws/send socket output)
                        (recur))))
                  (flow/resume fl)
                  nil)
       :on-message (fn on-text [_ws payload]
-                    (flow/inject fl [:transport-in :in] [payload]))
+                    (a/put! in payload))
       :on-close (fn on-close [_ws _status-code _reason]
                   (t/log! "Call closed")
-                  (flow/stop fl))
+                  (flow/stop fl)
+                  (a/close! in)
+                  (a/close! out))
       :on-error (fn on-error [ws error]
                   (prn error)
                   (t/log! :debug error))