+95
-38
@@ -20,7 +20,10 @@
|
||||
the caller's job.
|
||||
|
||||
Every turn is traced (see tbd.chat.trace): what was asked, how the reply went, and whether the
|
||||
model looks stuck in a loop. The config's `:trace` says where that goes."
|
||||
model looks stuck in a loop. The config's `:trace` says where that goes. A turn that does loop is
|
||||
not just left running (a model that loops keeps generating, so it is never silent long enough for
|
||||
a timeout to catch it) nor simply failed: `pump` cancels it and retries once, nudged to answer, an
|
||||
error only if that loops too — see `pump`."
|
||||
(:require [clojure.string :as str]
|
||||
[tbd.agent.json-schema :as json-schema]
|
||||
[tbd.chat.trace :as trace])
|
||||
@@ -152,6 +155,19 @@
|
||||
(println (cond-> (str "[tbd] chat: " text) detail (str " (" detail ")")))
|
||||
text))
|
||||
|
||||
(defn- loop-abort-text
|
||||
"What the user is told when a retried turn (see `pump`) still loops."
|
||||
[{:keys [kind]}]
|
||||
(case kind
|
||||
:repeated-thinking "the model got stuck repeating itself, and the reply was stopped"
|
||||
:long-reasoning "the model's reasoning went on without end, and the reply was stopped"))
|
||||
|
||||
(def ^:private loop-nudge
|
||||
"Appended to the system prompt for the one retry `pump` gives a turn that loops (see its
|
||||
docstring): blunt, since reasoning that has not converged in thousands of characters is not
|
||||
going to be talked into converging gently."
|
||||
"\n\nYou have been reasoning for a long time without answering. Stop reasoning now: give your final answer immediately, or call the tool you need.")
|
||||
|
||||
(defn- streaming-handler
|
||||
"A StreamingChatResponseHandler that pushes every callback onto `queue` as a map: {:content s},
|
||||
{:thinking s}, {:tool-call {:id :name :arguments}} (once a call is fully assembled — LangChain4j
|
||||
@@ -195,14 +211,40 @@
|
||||
(defn- finish-reason-of [^ChatResponse response]
|
||||
(some-> (.finishReason response) str str/lower-case))
|
||||
|
||||
(defn- send-request!
|
||||
"Sends `system-text`/`messages`/`tools`, and waits (up to `timeout-seconds`) for the first sign of
|
||||
life: {:ok {:queue :handle-atom :first-item}}, or {:error text :detail text} for an LLM that is
|
||||
down, refuses, or stays silent too long — so that is a clean error instead of one inside a stream
|
||||
that has already begun. Used for the turn's first attempt (see `start`) and, unlike it, for the
|
||||
retry `pump` sends after a turn that loops."
|
||||
[config system-text messages tools timeout-seconds]
|
||||
(try
|
||||
(let [queue (LinkedBlockingQueue.)
|
||||
handle-atom (atom nil)
|
||||
handler (streaming-handler queue #(reset! handle-atom %))]
|
||||
(.chat (chat-model config) (chat-request system-text messages tools) handler)
|
||||
(if-let [first-item (.poll queue (long timeout-seconds) TimeUnit/SECONDS)]
|
||||
(if (:error first-item)
|
||||
{:error "the LLM server can't be reached or refused" :detail (exception-detail (:error first-item))}
|
||||
{:ok {:queue queue :handle-atom handle-atom :first-item first-item}})
|
||||
{:error "the LLM server took too long to answer" :detail (str timeout-seconds "s of silence")}))
|
||||
(catch Exception e
|
||||
{:error "the LLM server can't be reached" :detail (exception-detail e)})))
|
||||
|
||||
(defn- pump
|
||||
"Drains `queue`, starting from the already-polled `first-item` (see `start`, which peeks one item
|
||||
before deciding {:ok}/{:error}), calling `emit` with each event (see the ns docstring).
|
||||
`timeout-seconds` of silence between items is an error, closing the stream via `handle-atom` (a
|
||||
StreamingHandle, once one has arrived). An exception from `emit` (the client went away, say) ends
|
||||
it too, closing the stream, and is rethrown. `tracer` (see tbd.chat.trace) sees everything the
|
||||
LLM says, and how the turn ended, before the client is told."
|
||||
[^LinkedBlockingQueue queue handle-atom timeout-seconds emit tracer first-item]
|
||||
"Drains `queue`, starting from the already-polled `first-item` (see `start`/`send-request!`, which
|
||||
peek one item before deciding {:ok}/{:error}), calling `emit` with each event (see the ns
|
||||
docstring). `timeout-seconds` of silence between items is an error, closing the stream via
|
||||
`handle-atom` (a StreamingHandle, once one has arrived).
|
||||
|
||||
A model stuck in a loop is never silent, so that timeout never catches it; `tracer` (see
|
||||
tbd.chat.trace/observe!) watches for it instead. The first time a turn loops (`retried?` false),
|
||||
the stream is cancelled and sent again — same `system-text`, with `loop-nudge` appended, same
|
||||
`messages`/`tools` — as a fresh, separately traced attempt; a caller only sees this as a pause; a
|
||||
turn that loops again on the retry, or that fails to even start it, is an error. An exception from
|
||||
`emit` (the client went away, say) ends the attempt too, closing the stream, and is rethrown."
|
||||
[config system-text messages tools timeout-seconds emit tracer
|
||||
^LinkedBlockingQueue queue handle-atom first-item retried?]
|
||||
(let [timer (atom nil)
|
||||
timed-out? (atom false)
|
||||
cancel-stream! (fn []
|
||||
@@ -212,7 +254,20 @@
|
||||
(some-> ^ScheduledFuture @timer (.cancel false))
|
||||
(reset! timer (.schedule ^ScheduledThreadPoolExecutor @watchdog
|
||||
^Runnable (fn [] (reset! timed-out? true) (cancel-stream!))
|
||||
(long timeout-seconds) TimeUnit/SECONDS)))]
|
||||
(long timeout-seconds) TimeUnit/SECONDS)))
|
||||
retry! (fn []
|
||||
(cancel-stream!)
|
||||
(some-> ^ScheduledFuture @timer (.cancel false))
|
||||
(let [nudged-text (str system-text loop-nudge)
|
||||
tracer' (trace/begin! (:trace config) {:messages messages :tools tools :model (:model config)
|
||||
:system-text nudged-text})
|
||||
{:keys [ok error detail]} (send-request! config nudged-text messages tools timeout-seconds)]
|
||||
(if error
|
||||
(let [text (fail error detail)]
|
||||
(trace/finish! tracer' {:outcome :upstream-error :error text :detail detail})
|
||||
(emit {:error text}))
|
||||
(pump config nudged-text messages tools timeout-seconds emit tracer'
|
||||
(:queue ok) (:handle-atom ok) (:first-item ok) true))))]
|
||||
(arm!)
|
||||
(try
|
||||
(loop [item first-item
|
||||
@@ -254,12 +309,23 @@
|
||||
(emit {:error text}))))
|
||||
|
||||
:else
|
||||
(do (arm!)
|
||||
(trace/observe! tracer (select-keys item [:content :thinking]))
|
||||
(when-let [out (not-empty (select-keys item [:content :thinking]))] (emit out))
|
||||
(recur (.take queue)
|
||||
(or replied? (contains? item :content))
|
||||
(cond-> calls (:tool-call item) (conj (:tool-call item)))))))
|
||||
(let [abort (trace/observe! tracer (select-keys item [:content :thinking]))]
|
||||
(when-let [out (not-empty (select-keys item [:content :thinking]))] (emit out))
|
||||
(cond
|
||||
(and abort (not retried?))
|
||||
(do (trace/finish! tracer {:outcome :looped}) (retry!))
|
||||
|
||||
abort
|
||||
(let [text (fail (loop-abort-text abort))]
|
||||
(trace/finish! tracer {:outcome :error :error text})
|
||||
(cancel-stream!)
|
||||
(emit {:error text}))
|
||||
|
||||
:else
|
||||
(do (arm!)
|
||||
(recur (.take queue)
|
||||
(or replied? (contains? item :content))
|
||||
(cond-> calls (:tool-call item) (conj (:tool-call item)))))))))
|
||||
(finally
|
||||
(trace/finish! tracer {:outcome :aborted})
|
||||
(some-> ^ScheduledFuture @timer (.cancel false))
|
||||
@@ -271,29 +337,20 @@
|
||||
(the agent's own system prompt, see tbd.agent.instructions — composed by the caller, not the
|
||||
browser). Waits for the first sign of life from the model (a piece of its reply, or an error) so
|
||||
that an LLM that is down or refuses is still a clean {:error text :kind :unavailable|:upstream},
|
||||
not an error inside a stream that has already begun — the same contract the old hand-rolled HTTP
|
||||
client gave, adapted to LangChain4j's async callbacks (which have no separate \"got the HTTP
|
||||
response\" moment to block on the way a blocking HTTP client's `.send()` did). 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."
|
||||
not an error inside a stream that has already begun. Otherwise {:ok pump}: `(pump emit)` blocks
|
||||
until the reply is complete (retrying once, on its own, if the model loops — see `pump`), 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."
|
||||
[{:keys [timeout-seconds trace] :as config} system-text messages tools]
|
||||
(if-not (configured? config)
|
||||
{:error "chat is not configured" :kind :unavailable}
|
||||
(let [tracer (trace/begin! trace {:messages messages :tools tools :model (:model config)})
|
||||
upstream-error (fn [text detail]
|
||||
(fail text detail)
|
||||
(trace/finish! tracer {:outcome :upstream-error :error text :detail detail})
|
||||
{:error text :kind :upstream})]
|
||||
(try
|
||||
(let [queue (LinkedBlockingQueue.)
|
||||
handle-atom (atom nil)
|
||||
handler (streaming-handler queue #(reset! handle-atom %))]
|
||||
(.chat (chat-model config) (chat-request system-text messages tools) handler)
|
||||
(if-let [first-item (.poll queue (long timeout-seconds) TimeUnit/SECONDS)]
|
||||
(if (:error first-item)
|
||||
(upstream-error "the LLM server can't be reached or refused" (exception-detail (:error first-item)))
|
||||
{:ok (fn [emit] (pump queue handle-atom timeout-seconds emit tracer first-item))})
|
||||
(upstream-error "the LLM server took too long to answer" (str timeout-seconds "s of silence"))))
|
||||
(catch Exception e
|
||||
(upstream-error "the LLM server can't be reached" (exception-detail e)))))))
|
||||
(let [tracer (trace/begin! trace {:messages messages :tools tools :model (:model config)
|
||||
:system-text system-text})
|
||||
{:keys [ok error detail]} (send-request! config system-text messages tools timeout-seconds)]
|
||||
(if error
|
||||
(do (fail error detail)
|
||||
(trace/finish! tracer {:outcome :upstream-error :error error :detail detail})
|
||||
{:error error :kind :upstream})
|
||||
{:ok (fn [emit]
|
||||
(pump config system-text messages tools timeout-seconds emit tracer
|
||||
(:queue ok) (:handle-atom ok) (:first-item ok) false))}))))
|
||||
|
||||
@@ -4,8 +4,8 @@
|
||||
on purpose: unlike tbd.agent.prompt (a .cljc the browser bundle also needs, for WebMCP's
|
||||
`guide`/`task`/`tic-tac-toe`), this ns is never on ui/shadow-cljs.edn's `:source-paths`, so the
|
||||
chatbot's own instructions can never end up in the compiled UI. It reuses the wording that is
|
||||
genuinely shared (`act`, `work-it-out`, `tic-tac-toe-rules`, `examples`) from tbd.agent.prompt
|
||||
rather than duplicating it."
|
||||
genuinely shared (`act`, `work-it-out`, `examples`) from tbd.agent.prompt rather than
|
||||
duplicating it."
|
||||
(:require [clojure.string :as str]
|
||||
[jsonista.core :as json]
|
||||
[tbd.agent.prompt :as prompt]))
|
||||
@@ -30,7 +30,6 @@
|
||||
"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"
|
||||
prompt/tic-tac-toe-rules
|
||||
(prompt/examples false)))
|
||||
|
||||
(defn chat-system-message
|
||||
@@ -47,9 +46,9 @@
|
||||
" tools: " (str/join ", " (map :name tools))
|
||||
". The tools of other pages are offered once you navigate there.")
|
||||
(seq state) (str "\n\nCurrent state of the page (JSON), refreshed on every message you get, so it is\n"
|
||||
"always current, more current than anything said earlier in this conversation, a tool\n"
|
||||
"result marked out of date, or what you said before — never call a tool just to read it,\n"
|
||||
"only to act:\n" (json/write-value-as-string state))))
|
||||
"always current — more current than anything said earlier in this conversation, any\n"
|
||||
"tool result in it, or what you said before. Go by this, not an earlier result:\n"
|
||||
"never call a tool just to read it, only to act:\n" (json/write-value-as-string state))))
|
||||
|
||||
(defn a2a
|
||||
"The A2A server's system prompt: an outside agent calling this app directly, not through a
|
||||
|
||||
+24
-4
@@ -112,11 +112,26 @@
|
||||
than max-messages, since this is scrollback for the user, not context for the LLM."
|
||||
100)
|
||||
|
||||
(def max-history-content-length
|
||||
"The longest tool result, or reply's reasoning, /api/chat/history stores: far over
|
||||
`max-message-length` (what the LLM is sent), since the stored conversation is the whole log for
|
||||
\"Expand tool use\"/\"Show reasoning\" to show after a reload, not context for the LLM."
|
||||
100000)
|
||||
|
||||
(def max-history-length
|
||||
"How long the stored conversation may be, printed, all told (see tbd.ui.events/storable, which
|
||||
drops its oldest messages to stay within it)."
|
||||
1000000)
|
||||
|
||||
(defn- small-history? [{:keys [messages]}]
|
||||
(<= (count (pr-str messages)) max-history-length))
|
||||
|
||||
(def HistoryMessage
|
||||
"A message as the browser's own :chat :messages keeps it (see tbd.ui.db), not AG-UI's wire
|
||||
Message: /api/chat/history is this app's own addition, not part of the AG-UI run protocol, so it
|
||||
round-trips the client's fields as-is (kebab-case, and a tool result's :name/:error? for
|
||||
display, which AG-UI's Message has no place for) instead of forcing them through Message's shape."
|
||||
round-trips the client's fields as-is (kebab-case, a tool result's :name/:error?, and an
|
||||
assistant reply's :thinking, for display, which AG-UI's Message has no place for) instead of
|
||||
forcing them through Message's shape."
|
||||
[:multi {:dispatch :role}
|
||||
["user" [:map
|
||||
[:role [:= "user"]]
|
||||
@@ -125,6 +140,9 @@
|
||||
[:map
|
||||
[:role [:= "assistant"]]
|
||||
[:content {:optional true} [:string {:max max-message-length}]]
|
||||
[:thinking {:optional true} [:string {:max max-history-content-length}]]
|
||||
;; the user halted the reply before it was over (see tbd.ui.events/::chat-halt)
|
||||
[:halted? {:optional true} :boolean]
|
||||
[:tool-calls {:optional true}
|
||||
[:vector {:max max-tool-calls}
|
||||
[:map
|
||||
@@ -136,12 +154,14 @@
|
||||
[:role [:= "tool"]]
|
||||
[:tool-call-id Id]
|
||||
[:name {:optional true} [:string {:max 200}]]
|
||||
[:content [:string {:min 1 :max max-message-length}]]
|
||||
[:content [:string {:min 1 :max max-history-content-length}]]
|
||||
[:error? {:optional true} :boolean]]]])
|
||||
|
||||
(def History
|
||||
"The body of a PUT to /api/chat/history: the browser's whole conversation, oldest first."
|
||||
[:map [:messages [:vector {:max max-history-messages} HistoryMessage]]])
|
||||
[:and
|
||||
[:map [:messages [:vector {:max max-history-messages} HistoryMessage]]]
|
||||
[:fn small-history?]])
|
||||
|
||||
(defn page-of
|
||||
"The `page` context entry's value, or nil."
|
||||
|
||||
+44
-14
@@ -14,17 +14,21 @@
|
||||
conversation each request carries.
|
||||
|
||||
Every turn is one `tracer`: `begin!` when the request goes out (logs `:turn-start`), `observe!` for
|
||||
each event of the reply (logs a `:loop-suspect` when one of the above trips), `finish!` when it ends
|
||||
(logs `:turn-end`, with the outcome, timings, sizes and flags). A tracer is for one thread, the one
|
||||
that reads the reply.
|
||||
each event of the reply (logs a `:loop-suspect` when one of the above trips, and, past
|
||||
`repeat-abort-count`/`long-reasoning-abort-chars`, tells the caller to stop the turn — a model
|
||||
that loops keeps generating, so it never goes silent long enough for a timeout to catch it),
|
||||
`finish!` when it ends (logs `:turn-end`, with the outcome, timings, sizes and flags). A tracer is
|
||||
for one thread, the one that reads the reply.
|
||||
|
||||
Turns are correlated by `req` (one per turn) and `run` (a hash of the user's last message, the same
|
||||
for every turn of the tool loop it started), with `step` counting the turns of the run from 0.
|
||||
|
||||
Entries are maps that go to a `sink`. `sink` prints each as one `[tbd] chat <event> k=v ...` line and,
|
||||
if given a file, appends it there as JSON. Only the file has the text of the reasoning and the reply
|
||||
(`:reasoning`, `:reply`) and the run's messages (`:input`): they are what you read to see why a
|
||||
model is looping, and they are the user's words, so they are off unless a file is asked for
|
||||
(`:reasoning`, `:reply`), the run's messages (`:input`), and the turn's system prompt (`:system-text`,
|
||||
the \"Current state of the page\" it was told included — see tbd.agent.instructions/chat-system-message):
|
||||
they are what you read to see why a model is looping, or confused about the state it was acting on,
|
||||
and they are the user's words (or the app's data), so they are off unless a file is asked for
|
||||
(CHAT_TRACE_FILE, see tbd.agent.core/env-config)."
|
||||
(:require [clojure.java.io :as io]
|
||||
[clojure.string :as str]
|
||||
@@ -48,6 +52,17 @@
|
||||
"The same tool called with the same arguments this many times in one run is a `:repeated-call`."
|
||||
3)
|
||||
|
||||
(def repeat-abort-count
|
||||
"A sentence said this many times is not a false positive anymore: `observe!` says to stop the
|
||||
turn. A milestone of `repeat-warn-count` (see `repeat-milestone?`), so it is always hit exactly,
|
||||
not stepped over."
|
||||
64)
|
||||
|
||||
(def long-reasoning-abort-chars
|
||||
"Reasoning still going, without an answer or a call, at this many characters: `observe!` says to
|
||||
stop the turn. A milestone of `reasoning-warn-chars` (each doubling), for the same reason."
|
||||
32000)
|
||||
|
||||
(def ^:private max-distinct-sentences 5000)
|
||||
(def ^:private max-pending-chars 1000)
|
||||
(def ^:private max-args-length 500)
|
||||
@@ -162,7 +177,7 @@
|
||||
|
||||
(def ^:private heavy-fields
|
||||
"What only the trace file gets."
|
||||
#{:reasoning :reply :input})
|
||||
#{:reasoning :reply :input :system-text})
|
||||
|
||||
(defn- render-field [k v]
|
||||
(str (name k) "="
|
||||
@@ -220,9 +235,12 @@
|
||||
|
||||
(defn begin!
|
||||
"Starts tracing a turn, and logs `:turn-start`; `messages`, `tools` and `page` are the request's, and
|
||||
`model` the LLM's, if the config names one. `trace` is a config of `config-for`, or nil for
|
||||
`default-config`. Returns the tracer."
|
||||
[trace {:keys [messages tools page model]}]
|
||||
`model` the LLM's, if the config names one. `system-text` is the turn's system prompt (see
|
||||
tbd.agent.instructions/chat-system-message) — with `full-text?` on, it goes to the trace file too,
|
||||
since it is the only place the page's state (whose turn it is, a board, a list) as the model was
|
||||
actually told it, rather than as the model later recalls it in its reasoning, is ever recorded.
|
||||
`trace` is a config of `config-for`, or nil for `default-config`. Returns the tracer."
|
||||
[trace {:keys [messages tools page model system-text]}]
|
||||
(let [{:keys [sink full-text?]} (or trace default-config)
|
||||
{:keys [run step run-calls top-call input]} (run-summary messages)
|
||||
tracer {:sink sink :full-text? full-text?
|
||||
@@ -234,27 +252,39 @@
|
||||
(cond-> {:page page :model model :messages (count messages) :tools (count tools)
|
||||
:prompt-chars (transduce (map #(count (:content %))) + 0 messages)
|
||||
:run-calls run-calls}
|
||||
full-text? (assoc :input input)))
|
||||
full-text? (assoc :input input :system-text system-text)))
|
||||
(when (and top-call (>= (:count top-call) repeated-call-warn-count))
|
||||
(log! tracer :loop-suspect {:kind :repeated-call :tool (:tool top-call) :count (:count top-call)
|
||||
:args (truncate (:args top-call) sample-length)}))
|
||||
tracer))
|
||||
|
||||
(defn- abortable [alert]
|
||||
(case (:kind alert)
|
||||
:repeated-thinking (when (>= (:count alert) repeat-abort-count) alert)
|
||||
:long-reasoning (when (>= (:chars alert) long-reasoning-abort-chars) alert)
|
||||
nil))
|
||||
|
||||
(defn observe!
|
||||
"Takes one event of the LLM's reply (see tbd.agent.core, whose pump calls this: :content, :thinking,
|
||||
or, at the end of a turn, :finish-reason and :usage), logging a `:loop-suspect` if the reasoning has
|
||||
gone on for long, or is repeating itself."
|
||||
gone on for long, or is repeating itself. Returns the alert to stop the turn over (see
|
||||
`repeat-abort-count`/`long-reasoning-abort-chars`), or nil to keep going: a model that repeats
|
||||
itself, or reasons on forever, does so at every following event too, so the caller does not have
|
||||
to wait for the silence-timeout that a model still actively (if uselessly) generating never trips."
|
||||
[{:keys [state full-text?] :as tracer} event]
|
||||
(let [[state' alerts] (advance @state event (now-ms) full-text?)]
|
||||
(reset! state (update state' :flags into (map :kind) alerts))
|
||||
(doseq [alert alerts]
|
||||
(log! tracer :loop-suspect alert))))
|
||||
(log! tracer :loop-suspect alert))
|
||||
(some abortable alerts)))
|
||||
|
||||
(defn finish!
|
||||
"Ends the turn, logging `:turn-end`; only the first call does anything, so it is safe to call
|
||||
again in a `finally`. `outcome` is a map: {:outcome :done | :tool-calls | :error | :upstream-error
|
||||
| :aborted, :calls [{:name :arguments}] for :tool-calls, :error text and :detail text for the
|
||||
failures}. :aborted is a turn that just stopped (the client went away, or something threw)."
|
||||
| :looped | :aborted, :calls [{:name :arguments}] for :tool-calls, :error text and :detail text
|
||||
for the failures}. :aborted is a turn that just stopped (the client went away, or something
|
||||
threw); :looped is one tbd.agent.core/pump cut short and retried because it was stuck (see
|
||||
`observe!`) — the retry is a turn of its own, with its own `:turn-start`/`:turn-end`."
|
||||
[{:keys [state full-text?] :as tracer} {:keys [outcome calls error detail]}]
|
||||
(let [s @state]
|
||||
(when-not (:finished? s)
|
||||
|
||||
Reference in New Issue
Block a user