add integrated chatbot with streaming + tools
CI / Test + coverage (push) Successful in 1m23s

This commit is contained in:
2026-09-21 09:09:55 -09:00
parent 10f4ca654b
commit a53f0c3621
25 changed files with 3409 additions and 81 deletions
+306
View File
@@ -0,0 +1,306 @@
(ns tbd.chat.core
"The chatbot's side of the LLM: a streaming, OpenAI-compatible chat completion request to the LM
Studio server named by the environment. No Ring here; tbd.chat.routes owns that.
`start` sends the request and gives {:ok pump} or {:error text :kind :unavailable|:upstream},
like tbd.todo.service's. `pump` forwards the reply, as it is generated, to a function that takes
events: {:thinking \"text\"} (a piece of the model's reasoning), {:content
\"text\"} (a piece of the reply), {:tool-calls [{:id :name :arguments}]} (the model wants these
tools run, once its reply is complete: `arguments` is the JSON text the model wrote), and
finally {:done true} or {:error \"text\"}.
The chatbot lives on the UI's pages and acts on them, so every request starts with a system
message (`default-system-prompt`, then which page the user is on, `page`, and the tools it has).
The tools are the ones the page offers to the chatbot (see tbd.ui.model-context), described as
MCP does in `tools/list`: {:name :description :inputSchema}. The LLM only sees their metadata: it
asks for a call, the browser makes it and sends the result back with the next request, as a
message with the role \"tool\" (see tbd.chat.schema/Message). This ns translates all of that to
and from the OpenAI-style `tools`, `tool_calls` and `tool_call_id` of the LLM server."
(:require [clojure.string :as str]
[muuntaja.core :as m])
(:import (java.io BufferedReader Closeable IOException InputStream InputStreamReader)
(java.net URI)
(java.net.http HttpClient HttpClient$Version HttpRequest HttpRequest$BodyPublishers HttpResponse$BodyHandlers)
(java.nio.charset StandardCharsets)
(java.time Duration)
(java.util.concurrent ScheduledFuture ScheduledThreadPoolExecutor ThreadFactory TimeUnit)))
(def default-timeout-seconds 120)
;; HTTP/1.1 on purpose: the JDK client's default is to ask a plain-http server to upgrade to HTTP/2
;; (`Upgrade: h2c`), and LM Studio's server never answers that request, so the backend would wait
;; out the whole timeout while the LLM sat idle.
(def ^:private http-client
(delay (-> (HttpClient/newBuilder)
(.version HttpClient$Version/HTTP_1_1)
(.connectTimeout (Duration/ofSeconds 5))
.build)))
(defn java-http
"Default `:http` fn: {:url, :headers, :body, :timeout-seconds} is POSTed, giving {:status, :body}
as soon as the response starts; :body is an InputStream of the rest, which the caller closes.
Configs can swap it out, which is how the tests stub the LLM server. It throws when the server
can't be reached or doesn't start answering in time."
[{:keys [url headers body timeout-seconds]}]
(let [builder (-> (HttpRequest/newBuilder (URI/create url))
(.timeout (Duration/ofSeconds timeout-seconds))
(.POST (HttpRequest$BodyPublishers/ofString body)))]
(doseq [[k v] headers] (.header builder k v))
(let [resp (.send ^HttpClient @http-client (.build builder) (HttpResponse$BodyHandlers/ofInputStream))]
{:status (.statusCode resp) :body (.body resp)})))
(defn- env-val [env k]
(let [v (get env k)]
(when-not (str/blank? v) (str/trim v))))
(defn env-config
"Chat config from environment variables:
LMSTUDIO_URL where the LM Studio server is, e.g. http://192.168.1.20:1234 (chat is off without it)
LMSTUDIO_MODEL the model id to ask for (optional: LM Studio uses the model that is loaded)
LMSTUDIO_API_KEY sent as a bearer token, when the server wants one (optional)
LMSTUDIO_TIMEOUT_SECONDS how long the server may stay silent before giving up (default 120)"
([] (env-config (System/getenv)))
([env]
{:url (some-> (env-val env "LMSTUDIO_URL") (str/replace #"/+$" ""))
:model (env-val env "LMSTUDIO_MODEL")
:api-key (env-val env "LMSTUDIO_API_KEY")
:timeout-seconds (or (some-> (env-val env "LMSTUDIO_TIMEOUT_SECONDS") parse-long)
default-timeout-seconds)
:http java-http}))
(defn configured? [{:keys [url]}]
(some? url))
(defn- fail
"Logs a failure and returns the error text for the client. `text` says nothing about the LLM
server's address or its own message; `detail` (for the log only) does, and is where to look when
chat doesn't work."
([text] (fail text nil))
([text detail]
(println (cond-> (str "[tbd] chat: " text) detail (str " (" detail ")")))
text))
(defn- exception-detail [^Exception e]
(str (class e) ": " (.getMessage e)))
(defn- llm-message
"A message of the conversation (see tbd.chat.schema/Message) as the LLM server has it."
[{:keys [role content tool-calls tool-call-id]}]
(case role
"assistant" (cond-> {:role role :content content}
(seq tool-calls) (assoc :tool_calls
(mapv (fn [{:keys [id name arguments]}]
{:id id :type "function"
:function {:name name :arguments arguments}})
tool-calls)))
"tool" {:role role :tool_call_id tool-call-id :content content}
{:role role :content content}))
(defn- llm-tool
"An MCP tool's metadata as a function the LLM may call."
[{:keys [name description inputSchema]}]
{:type "function"
:function (cond-> {:name name :parameters inputSchema}
description (assoc :description description))})
(def default-system-prompt
"What the LLM is told about its job. The chatbot is an agent embedded in a UI, not a
conversationalist: what it does shows on the user's screen the moment it does it, so the prompt
is about acting (calling the page's tools in the same turn, chaining them, saying little
afterwards) and about the few cases where it should not. A model that is only asked to be
helpful answers a request for action with an offer of help. It is short and concrete on purpose,
with examples: the models behind LM Studio follow that better than a long list of rules."
(str "You are the assistant built into this web app. Your tools type, click and read on the page the\n"
"user is looking at, and every call shows on their screen the moment you make it.\n"
"\n"
"Act, don't offer. When the user asks for something a tool can do, call the tool in this same\n"
"turn. Never say you are about to do something, never ask \"would you like me to\", never ask them\n"
"to confirm what they just asked for.\n"
"\n"
"Work it out yourself:\n"
"- Detail missing: take the obvious default and go. Ask only when a required argument is genuinely\n"
" unknowable (which of several todos they mean) or when you would delete or overwrite something\n"
" and cannot tell what.\n"
"- Chain calls without stopping. The tool you need belongs to another page: navigate there, then\n"
" use it. You need an id: list first, then act. Never invent an id, a title or the state of the\n"
" page: read it with a tool.\n"
"- A call fails: read the error, fix the arguments or read the state, and try once more. No\n"
" apologies, no explanation of what went wrong unless it still fails.\n"
"\n"
"Then say one short plain sentence: \"Added it.\" \"That's 23.\" The call and its result are already\n"
"on screen under your message, so never repeat them, never paste JSON, never use markdown,\n"
"headings or bullet lists. Say something was done only if a tool result says it was.\n"
"\n"
"Some things are not tool work: greetings, questions about you, questions you can simply answer.\n"
"Answer those in a sentence or two, with no tool call. Asked what you can do, say it plainly from\n"
"the tools you have. Think briefly before you act.\n"
"\n"
"Tic-tac-toe: check whose turn it is, move only for your own side, one move, then stop and let the\n"
"person play theirs.\n"
"\n"
"Examples:\n"
"\"add a todo to buy milk\" -> create_todo{\"title\":\"buy milk\"} -> \"Added.\"\n"
"\"tick off the milk one\" -> list_todos -> set_todo_done{\"id\":7,\"done\":true} -> \"Ticked off.\"\n"
"\"what's 19 and 4?\" -> add{\"x\":19,\"y\":4} -> \"23.\"\n"
"\"play tic tac toe, you're O\" -> get_tic_tac_toe -> it is X's turn -> \"You're up, X starts.\""))
(defn- system-message
"Everything the LLM is told before the conversation, as one message: the instructions
(`default-system-prompt`), then, last, where the user is (`page`) and the names of the `tools`
on offer there. The tools' names are in the request's `tools` too, but a model picks them
more readily when the instructions say them as well, and it can say what it can do without a
call. One message, and the first: several chat templates (Gemma's, Mistral's) refuse more than
one system message, or one that isn't first."
[page tools]
{:role "system"
:content (cond-> default-system-prompt
page (str "\n\nThe user is on the \"" page "\" page of the app: what it shows is what they can see.")
(seq tools) (str (if page " Its" "\n\nYour")
" tools: " (str/join ", " (map :name tools))
". The tools of other pages are offered once you navigate there."))})
(defn- request [{:keys [model api-key timeout-seconds] :as config} messages tools page]
{:url (str (:url config) "/v1/chat/completions")
:headers (cond-> {"Content-Type" "application/json"}
api-key (assoc "Authorization" (str "Bearer " api-key)))
:body (slurp (m/encode "application/json"
(cond-> {:messages (into [(system-message page tools)] (map llm-message) messages)
:stream true}
model (assoc :model model)
(seq tools) (assoc :tools (mapv llm-tool tools)))))
:timeout-seconds timeout-seconds})
;; --- Reading the reply ---
(def ^:private watchdog
(delay (doto (ScheduledThreadPoolExecutor.
1 (reify ThreadFactory
(newThread [_ r] (doto (Thread. ^Runnable r "chat-watchdog") (.setDaemon true)))))
(.setRemoveOnCancelPolicy true))))
(defn- call-deltas
"The pieces of tool calls in a delta: the model writes a call bit by bit, over many events. The
first piece of a call has its id and the tool's name, the rest the JSON of its arguments; `index`
says which call (of the several a reply may ask for) a piece belongs to."
[delta]
(when (sequential? (:tool_calls delta))
(for [{:keys [index id function]} (:tool_calls delta)]
{:index (or index 0) :id id :name (:name function) :arguments (:arguments function)})))
(defn- event-of
"What one line of the LLM's event stream says: a map with :content s and/or :thinking s (a piece
of the reasoning: kept as it is, blanks too, since a piece can be just the line break between two
paragraphs), and :call-deltas (see call-deltas) if there are any; {:error text :detail text}; or
:end (`data: [DONE]`). nil for a line that says nothing: blank, a comment, or an event without
text."
[line]
(when-let [data (some-> (re-matches #"data:\s?(.*)" line) second str/trim)]
(if (= "[DONE]" data)
:end
(let [event (try (m/decode "application/json" data) (catch Exception _ nil))
delta (get-in event [:choices 0 :delta])
text? (not (str/blank? (:content delta)))
thinking (:reasoning_content delta)]
(if (:error event)
{:error "the LLM server reported an error" :detail (pr-str (:error event))}
(not-empty
(cond-> {}
text? (assoc :content (:content delta))
(and (string? thinking) (seq thinking)) (assoc :thinking thinking)
(seq (call-deltas delta)) (assoc :call-deltas (vec (call-deltas delta))))))))))
(defn- add-deltas
"`calls` ({index {:id :name :arguments}}, in order) with the pieces of calls in `deltas`."
[calls deltas]
(reduce (fn [calls {:keys [index id arguments] :as delta}]
(update calls index
(fn [call]
(cond-> (or call {:arguments ""})
id (assoc :id id)
(:name delta) (assoc :name (:name delta))
arguments (update :arguments str arguments)))))
calls
deltas))
(defn- finished-calls
"The calls the model asked for, as {:id :name :arguments}. A call that never got a tool's name
is no call."
[calls]
(vec (for [[index {:keys [id arguments] :as call}] calls
:when (not (str/blank? (:name call)))]
{:id (or id (str "call_" index)) :name (:name call) :arguments arguments})))
(defn- pump
"Reads the LLM's event stream `in`, calling `emit` with each event (see the ns docstring), and
closes `in`. More than `timeout-seconds` of silence from the LLM is an error. An exception from
`emit` (the client went away, say) ends it too, and is thrown to the caller."
[^InputStream in timeout-seconds emit]
(let [timer (atom nil)
timed-out? (atom false)
arm! (fn []
(some-> ^ScheduledFuture @timer (.cancel false))
(reset! timer (.schedule ^ScheduledThreadPoolExecutor @watchdog
^Runnable (fn []
(reset! timed-out? true)
(try (.close in) (catch IOException _ nil)))
(long timeout-seconds) TimeUnit/SECONDS)))]
(arm!)
(try
(with-open [reader (BufferedReader. (InputStreamReader. in StandardCharsets/UTF_8))]
(loop [replied? false
calls (sorted-map)]
(let [line (try (.readLine reader)
(catch IOException e
(if @timed-out?
{:error "the LLM server took too long to answer"
:detail (str timeout-seconds "s of silence")}
{:error "the LLM server's reply was cut off" :detail (exception-detail e)})))
event (cond (map? line) line ; the read failed
(nil? line) :end ; the stream ended without [DONE]
:else (event-of line))]
(cond
(nil? event) (do (arm!) (recur replied? calls))
(= :end event) (let [calls (finished-calls calls)]
(cond
(seq calls) (do (emit {:tool-calls calls}) (emit {:done true}))
replied? (emit {:done true})
:else (emit {:error (fail "the LLM server sent no reply")})))
(:error event) (emit {:error (fail (:error event) (:detail event))})
:else (do (arm!)
(when-let [out (not-empty (dissoc event :call-deltas))] (emit out))
(recur (or replied? (contains? event :content))
(add-deltas calls (:call-deltas event))))))))
(finally
(some-> ^ScheduledFuture @timer (.cancel false))
(try (.close in) (catch IOException _ nil))))))
(defn start
"Sends the LLM the conversation `messages` (see tbd.chat.schema/Message), and the `tools` it may
ask the page to call (MCP's {:name :description :inputSchema}; none if nil). `page`, if there is
one, is the page of the UI the user is on: the LLM is told so, and which tools it has, in the
system message that goes ahead of the conversation. Gives {:error text
:kind :unavailable|:upstream} when chat isn't configured, or the LLM server can't be reached or
refuses; nothing is streamed then. Otherwise {:ok pump}: `(pump emit)` blocks until the reply is
complete, calling `emit` with its events (see the ns docstring). If `emit` throws, the request to
the LLM is closed (which stops it generating) and the exception is rethrown."
([config messages] (start config messages nil nil))
([config messages tools] (start config messages tools nil))
([{:keys [timeout-seconds http] :as config} messages tools page]
(if-not (configured? config)
{:error "chat is not configured" :kind :unavailable}
(try
(let [{:keys [status body]} (http (request config messages tools page))]
(if (= 200 status)
{:ok (fn [emit] (pump body timeout-seconds emit))}
(let [snippet (try (String. (.readNBytes ^InputStream body 200) StandardCharsets/UTF_8)
(catch Exception _ nil))]
(some-> ^Closeable body .close)
{:error (fail (str "the LLM server answered HTTP " status) (str "body: " snippet))
:kind :upstream})))
(catch java.net.http.HttpTimeoutException e
{:error (fail "the LLM server took too long to answer" (.getMessage e)) :kind :upstream})
(catch IOException e
{:error (fail "the LLM server can't be reached" (exception-detail e)) :kind :upstream})
(catch Exception e
{:error (fail "the LLM server sent an unreadable reply" (exception-detail e)) :kind :upstream})))))
+53
View File
@@ -0,0 +1,53 @@
(ns tbd.chat.routes
(:require [muuntaja.core :as m]
[ring.core.protocols :as ring-protocols]
[tbd.chat.core :as chat]
[tbd.chat.schema :as schema]
[tbd.http :refer [respond]])
(:import (java.io IOException OutputStream)
(java.nio.charset StandardCharsets)))
(def ^:private error-status {:unavailable 503 :upstream 502})
(defn- sse-event
"One server-sent event: its data is `event` as JSON, on a single line."
^bytes [event]
(.getBytes (str "data: " (slurp (m/encode "application/json" event)) "\n\n") StandardCharsets/UTF_8))
(defn- event-stream
"A response body that writes what `pump` (see tbd.chat.core/start) forwards, one server-sent
event each, flushing after every one so they arrive as they are generated. A browser that goes
away stops it, and the LLM's request with it."
[pump]
(reify ring-protocols/StreamableResponseBody
(write-body-to-stream [_ _ out]
(try
(pump (fn [event]
(.write ^OutputStream out (sse-event event))
(.flush ^OutputStream out)))
(catch IOException _ nil)))))
(defn- chat-handler [config]
(fn [{{{:keys [messages tools page]} :body} :parameters}]
;; The LLM is asked before the response starts, so an LLM that is down is still a 502, not an
;; error inside a stream that has already begun.
(let [{:keys [ok error kind]} (chat/start config messages tools page)]
(if error
(respond (error-status kind) {:error error})
{:status 200
:headers {"Content-Type" "text/event-stream; charset=utf-8"
"Cache-Control" "no-cache"
;; nginx would otherwise hold the events back until the reply is over
"X-Accel-Buffering" "no"}
:body (event-stream ok)}))))
(defn routes
"`config` is the chat config (see tbd.chat.core/env-config). Only logged-in users may chat:
every message costs the LLM server's time."
[config]
["/api/chat"
{:post {:tags ["chat"]
:login? true
:summary "Send a conversation to the LLM (LM Studio); the reply streams back as server-sent events: data: {\"thinking\":\"...\"} for each piece of its reasoning, data: {\"content\":\"...\"} for each piece of the reply, data: {\"tool-calls\":[{\"id\":\"...\",\"name\":\"...\",\"arguments\":\"{...}\"}]} if the model asks for tools of the page (the `tools` in the body, as MCP describes them) to be called, and last data: {\"done\":true} or data: {\"error\":\"...\"}"
:parameters {:body schema/Chat}
:handler (chat-handler config)}}])
+74
View File
@@ -0,0 +1,74 @@
(ns tbd.chat.schema
"The shape of the chat request, as Malli schemas, coerced by tbd.chat.routes. The limits are
what keeps the endpoint from being a free pipe to the LLM: a conversation is bounded in
length, and the UI (tbd.ui.events) sends no more than `max-messages` of it.
The conversation can hold the model's use of the page's tools (see tbd.ui.model-context): an
assistant message that asks for calls (`:tool-calls`), followed by one \"tool\" message with the
result of each. Those are the browser's to say, so they are as bounded as the rest.")
(def max-messages 40)
(def max-message-length 8000)
(def max-tools 32)
(def max-tool-calls 16)
(def ^:private max-tool-schema-length
"How long a tool's metadata may be, printed: room for a tool of a dozen parameters."
5000)
(defn- small? [tool]
(<= (count (pr-str tool)) max-tool-schema-length))
(def Tool
"What the page says about one of its tools, as MCP's `tools/list` does. `inputSchema` is JSON Schema
for the arguments, which is the LLM's to read, so it is any (small) object schema."
[:and
[:map
[:name [:re #"^[a-zA-Z0-9_-]{1,64}$"]]
[:description {:optional true} [:string {:max 2000}]]
;; A map-of, not a :map with a :type: the coercion strips the keys a :map doesn't name, and
;; these are all the LLM's to read (`properties`, `required`, ...).
[:inputSchema [:and [:map-of :keyword :any] [:fn (fn [schema] (= "object" (:type schema)))]]]]
[:fn small?]])
(def ^:private Id [:string {:min 1 :max 200}])
(def ToolCall
"A call the model asked for. Its name and arguments are the model's own words: nothing says they
are a tool of the page, or JSON, so they are only bounded, generously (a conversation that the
server refused would be stuck: the UI sends it again with the next message)."
[:map
[:id Id]
[:name [:string {:min 1 :max 200}]]
[:arguments [:string {:max 100000}]]])
(def Message
[:multi {:dispatch :role}
["user" [:map
[:role [:= "user"]]
[:content [:string {:min 1 :max max-message-length}]]]]
["assistant" [:and
[:map
[:role [:= "assistant"]]
[:content [:string {:max max-message-length}]]
[:tool-calls {:optional true} [:vector {:max max-tool-calls} ToolCall]]]
[:fn (fn [{:keys [content tool-calls]}] (or (seq content) (seq tool-calls)))]]]
["tool" [:map
[:role [:= "tool"]]
[:tool-call-id Id]
[:content [:string {:min 1 :max max-message-length}]]]]])
(def Page
"Which page of the UI the user is on: its id, as the UI has it (\"games\", \"ai-setup\"). The server
doesn't know the pages, but the LLM is told it, so it is a short slug and nothing that reads as
an instruction."
[:re #"^[a-z0-9-]{1,64}$"])
(def Chat
"The conversation so far, oldest first; the LLM's reply follows the last message. `page` is the
page the user is on, and `tools` the ones that page offers to the LLM, for this reply."
[:map
[:messages [:vector {:min 1 :max max-messages} Message]]
[:page {:optional true} Page]
[:tools {:optional true} [:vector {:max max-tools} Tool]]])
+7 -1
View File
@@ -17,6 +17,8 @@
[tbd.auth.core :as auth]
[tbd.auth.middleware :as auth-mw]
[tbd.auth.routes :as auth-routes]
[tbd.chat.core :as chat]
[tbd.chat.routes :as chat-routes]
[tbd.health.routes :as health]
[tbd.http :as http]
[tbd.math.routes :as math]
@@ -68,7 +70,7 @@
;; Each logical group (health, users, orders, ...) owns a `tbd.<group>.routes`
;; namespace that returns its own route subtree; they're concatenated here.
(defn make-router [{:keys [auth nrepl todo oauth webmcp]}]
(defn make-router [{:keys [auth nrepl todo oauth webmcp chat]}]
(let [oauth-ds (delay (oauth-db/open (:db-path oauth)))
;; Opened (and migrated) on first use, not when the app loads.
todo-ds (delay (todo-db/open (:db-path todo)))]
@@ -86,6 +88,7 @@
(mcp-routes/routes auth oauth-ds todo-ds)
(oauth-routes/routes auth oauth-ds)
(auth-routes/routes auth)
(chat-routes/routes chat)
(nrepl-routes/routes nrepl)
(todo-routes/routes todo-ds)]
{:data {:muuntaja http/muuntaja
@@ -120,6 +123,7 @@
(def config
{:auth (auth/env-config)
:nrepl (nrepl/env-config)
:chat (chat/env-config)
:todo (todo-db/env-config)
:oauth (oauth-db/env-config)
:webmcp {:origin-trial-token (some-> (System/getenv "WEBMCP_ORIGIN_TRIAL_TOKEN") str/trim not-empty)}})
@@ -141,6 +145,8 @@
(println "[tbd] WARNING: ADMIN_USERS is empty; nobody can use the admin API"))
(when-not (get-in config [:auth :session-secret])
(println "[tbd] WARNING: SESSION_SECRET is not set; logins won't survive a restart"))
(when-not (chat/configured? (:chat config))
(println "[tbd] LMSTUDIO_URL is not set; the chatbot is unavailable"))
(try
(let [server (jetty/run-jetty (wrap-defaults app api-defaults)
{:port port