Improve LLM streaming response handler.

Author: ericdallo

CHANGELOG.md

@@ -8,6 +8,8 @@
 - Support custom API urls for OpenAI and Anthropic
 - Add `--log-level <level>` option for better debugging.
 - Add support for global config file.
+- Improve MCP response handling.
+- Improve LLM streaming response handler.
 
 ## 0.0.3
 

src/eca/features/mcp.clj

@@ -74,14 +74,14 @@
 (defn call-tool! [^String name ^Map arguments db]
   (let [result (.callTool ^McpSyncClient (get-in db [:mcp-tools name :mcp-client])
                           (McpSchema$CallToolRequest. name arguments))]
-    (if (.isError result)
-      {:error (.content result)}
-      {:contents (map (fn [content]
-                        (case (.type ^McpSchema$Content content)
-                          "text" {:type :text
-                                  :content (.text ^McpSchema$TextContent content)}
-                          nil))
-                      (.content result))})))
+    (logger/debug logger-tag "ToolCall result: " result)
+    {:contents (map (fn [content]
+                      (case (.type ^McpSchema$Content content)
+                        "text" {:type :text
+                                :error (.isError result)
+                                :content (.text ^McpSchema$TextContent content)}
+                        nil))
+                    (.content result))}))
 
 (defn shutdown! [db*]
   (doseq [[_name {:keys [_client]}] (:mcp-clients @db*)]

src/eca/llm_providers/openai.clj

@@ -10,7 +10,7 @@
 
 (def ^:private logger-tag "[OPENAI]")
 
-(def ^:private openai-url "https://api.openai.com/")
+(def ^:private openai-url "https://api.openai.com")
 (def ^:private responses-path "/v1/responses")
 
 (defn ^:private url [path]
@@ -21,13 +21,15 @@
 
 (defn ^:private base-completion-request! [{:keys [body api-key on-error on-response]}]
   (let [api-key (or api-key
-                    (System/getenv "OPENAI_API_KEY"))]
-    (logger/debug logger-tag (format "Sending input: '%s' instructions: '%s' tools: '%s'"
+                    (System/getenv "OPENAI_API_KEY"))
+        url (url responses-path)]
+    (logger/debug logger-tag (format "Sending input: '%s' instructions: '%s' tools: '%s' url: '%s'"
                                      (:input body)
                                      (:instructions body)
-                                     (:tools body)))
+                                     (:tools body)
+                                     url))
     (http/post
-     (url responses-path)
+     url
      {:headers {"Authorization" (str "Bearer " api-key)
                 "Content-Type" "application/json"}
       :body (json/generate-string body)

src/eca/llm_util.clj

@@ -7,14 +7,25 @@
    [java.io BufferedReader]))
 
 (defn event-data-seq [^BufferedReader rdr]
-  (when-let [event (.readLine rdr)]
-    (when (string/starts-with? event "event:")
-      (when-let [data (.readLine rdr)]
-        (.readLine rdr) ;; blank line
-        (when (string/starts-with? data "data:")
-          (cons [(subs event 7)
-                 (json/parse-string (subs data 6) true)]
-                (lazy-seq (event-data-seq rdr))))))))
+  (letfn [(next-group []
+            (loop [event-line nil]
+              (let [line (.readLine rdr)]
+                (cond
+                  (nil? line) nil ; EOF
+                  (string/blank? line) (recur event-line) ; skip blank lines
+                  (string/starts-with? line "event:") (recur line)
+                  (string/starts-with? line "data:")
+                  (let [data-str (subs line 6)]
+                    (if (= data-str "[DONE]")
+                      (recur event-line) ; skip [DONE]
+                      (let [event-type (if event-line
+                                         (subs event-line 7)
+                                         (-> (json/parse-string data-str true)
+                                             :type))]
+                        (cons [event-type (json/parse-string data-str true)]
+                              (lazy-seq (next-group))))))
+                  :else (recur event-line)))))]
+    (next-group)))
 
 (defn log-response [tag event data]
   (logger/debug tag event data))

test/eca/llm_util_test.clj

@@ -0,0 +1,38 @@
+(ns eca.llm-util-test
+  (:require
+   [clojure.java.io :as io]
+   [clojure.test :refer [deftest is testing]]
+   [eca.llm-util :as llm-util]
+   [matcher-combinators.test :refer [match?]])
+  (:import
+   [java.io ByteArrayInputStream]))
+
+(deftest event-data-seq-test
+  (testing "when there is a event line and another data line"
+    (with-open [r (io/reader (ByteArrayInputStream. (.getBytes (str "event: foo.bar\n"
+                                                                    "data: {\"type\": \"foo.bar\"}\n"
+                                                                    "\n"
+                                                                    "event: foo.baz\n"
+                                                                    "data: {\"type\": \"foo.baz\"}"))))]
+      (is (match?
+           [["foo.bar" {:type "foo.bar"}]
+            ["foo.baz" {:type "foo.baz"}]]
+           (llm-util/event-data-seq r)))))
+  (testing "when there is no event line, only a data line"
+    (with-open [r (io/reader (ByteArrayInputStream. (.getBytes (str "data: {\"type\": \"foo.bar\"}\n"
+                                                                    "\n"
+                                                                    "data: {\"type\": \"foo.baz\"}"))))]
+      (is (match?
+           [["foo.bar" {:type "foo.bar"}]
+            ["foo.baz" {:type "foo.baz"}]]
+           (llm-util/event-data-seq r)))))
+  (testing "Ignore [DONE] when exists"
+    (with-open [r (io/reader (ByteArrayInputStream. (.getBytes (str "data: {\"type\": \"foo.bar\"}\n"
+                                                                    "\n"
+                                                                    "data: {\"type\": \"foo.baz\"}\n"
+                                                                    "\n"
+                                                                    "data: [DONE]\n"))))]
+      (is (match?
+           [["foo.bar" {:type "foo.bar"}]
+            ["foo.baz" {:type "foo.baz"}]]
+           (llm-util/event-data-seq r))))))