diff --git a/CHANGELOG.md b/CHANGELOG.md index cf74e7227..a54bddf09 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- Show output tokens per second (TPS) in the chat usage payload for streamed model responses. + ### Docs site - Added: `docs/evolution`. This holds internal planning docs and visual changelogs related to developing the docs site. It is excluded from the published site with `exclude_docs` - Added: Notes on the Configuration sidebar structure in `docs/evolution/planning/sidebar-navigation.md` @@ -31,7 +33,6 @@ - Changed: "Something not working?" section is a bulleted list with the doc page link style - Changed: Emoji arrows replaced with plain Unicode text arrows in 61 headings: → ← ↩ ↪ - ## 0.166.0 - Fix resuming a chat restoring the default agent instead of the chat's own agent when it was picked before the first message. #657 diff --git a/docs/protocol.md b/docs/protocol.md index 40ce2a229..b2b836255 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -981,6 +981,16 @@ interface ChatUsageContent { * The total input + output tokens of the whole chat session so far. */ sessionTokens: number; + + /** + * Output tokens per second of the last model response, rounded to an + * integer. Measured from the first generated token to the end of the + * response. Includes reasoning tokens when the provider counts them as + * output and server-side tool waits such as web search. Excludes client + * tool execution and approval waits. Absent for non-streaming requests or + * when usage or timing is unknown. + */ + messageOutputTps?: number; /** * The cost of the last sent message summing input + output tokens. diff --git a/src/eca/llm_api.clj b/src/eca/llm_api.clj index cdda2a750..055f3fac4 100644 --- a/src/eca/llm_api.clj +++ b/src/eca/llm_api.clj @@ -539,6 +539,10 @@ on-history-sanitized identity cancelled? (constantly false)}}] (let [first-response-received* (atom false) + ;; First generation event of the current model response. Consumed by + ;; the next usage update, so tool-call loops start a new measurement. + generation-start-ms* (atom nil) + mark-generation-start! #(compare-and-set! generation-start-ms* nil (System/currentTimeMillis)) ;; Fire :on-history-sanitized at most once per sync-or-async-prompt! call: ;; prompt! is invoked multiple times on retries and we don't want to ;; spam the chat with repeated "history sanitized" notices. @@ -558,24 +562,45 @@ ;; tool call. Wrapped here rather than in `prompt!` because the sync path ;; invokes the callback itself, without going through it. on-tools-called-wrapper (fn [tool-calls] + ;; Tool execution can block on user approval. + ;; End the generation window first so a usage + ;; event that arrives after tools (Bedrock) + ;; cannot include that time. + (reset! generation-start-ms* nil) (let [result (on-tools-called tool-calls)] (cond-> result (map? result) (update :tools tools-for-request)))) - on-message-received-wrapper (fn [& args] + on-message-received-wrapper (fn [& [message :as args]] + (when (#{:text :url :refusal} (:type message)) + (mark-generation-start!)) (apply emit-first-message-fn args) (apply on-message-received args)) - on-reason-wrapper (fn [& args] + on-reason-wrapper (fn [& [reason :as args]] + (when (#{:started :thinking} (:status reason)) + (mark-generation-start!)) (apply emit-first-message-fn args) (apply on-reason args)) on-prepare-tool-call-wrapper (fn [& args] + (mark-generation-start!) (apply emit-first-message-fn args) (apply on-prepare-tool-call args)) - on-server-web-search-wrapper (fn [& args] + on-server-web-search-wrapper (fn [& [event :as args]] + (when (= :started (:status event)) + (mark-generation-start!)) (apply emit-first-message-fn args) (apply on-server-web-search args)) - on-server-image-generation-wrapper (fn [& args] + on-server-image-generation-wrapper (fn [& [event :as args]] + (when (= :started (:status event)) + (mark-generation-start!)) (apply emit-first-message-fn args) (apply on-server-image-generation args)) + on-usage-updated-wrapper (fn [usage] + (let [start-ms @generation-start-ms* + elapsed-ms (when start-ms (- (System/currentTimeMillis) start-ms)) + tps (shared/tokens-per-second (:output-tokens usage) elapsed-ms)] + (reset! generation-start-ms* nil) + (on-usage-updated (cond-> usage + tps (assoc :message-output-tps tps))))) ;; Deliver at most one terminal error per prompt: when a shared ;; connection dies, every stacked tool-continuation request unwinds ;; and fires its own on-error for the same failure. #547 @@ -718,7 +743,7 @@ :on-message-received on-message-received-wrapper :on-prepare-tool-call on-prepare-tool-call-wrapper :on-tools-called on-tools-called-wrapper - :on-usage-updated on-usage-updated + :on-usage-updated on-usage-updated-wrapper :on-server-web-search on-server-web-search-wrapper :on-server-image-generation on-server-image-generation-wrapper :on-reason on-reason-wrapper diff --git a/src/eca/shared.clj b/src/eca/shared.clj index f8eccd66c..7e8b303d2 100644 --- a/src/eca/shared.clj +++ b/src/eca/shared.clj @@ -330,11 +330,20 @@ (format "%.2f" (+ input-cost (* output-tokens output-token-cost))))))) +(defn tokens-per-second + "Output tokens per second as an integer. Returns nil when the token count or + elapsed time is not positive." + [tokens elapsed-ms] + (when (and (number? tokens) (number? elapsed-ms) + (pos? (long tokens)) (pos? (long elapsed-ms))) + (Math/round (/ (* 1000.0 tokens) elapsed-ms)))) + (defn ^:private usage-sumary* [chat-id full-model db] (let [last-input-tokens (or (get-in db [:chats chat-id :usage :last-input-tokens]) 0) last-output-tokens (or (get-in db [:chats chat-id :usage :last-output-tokens]) 0) last-input-cache-creation-tokens (or (get-in db [:chats chat-id :usage :last-input-cache-creation-tokens]) 0) last-input-cache-read-tokens (or (get-in db [:chats chat-id :usage :last-input-cache-read-tokens]) 0) + message-output-tps (get-in db [:chats chat-id :usage :last-output-tps]) total-input-tokens (or (get-in db [:chats chat-id :usage :total-input-tokens]) 0) total-input-cache-creation-tokens (or (get-in db [:chats chat-id :usage :total-input-cache-creation-tokens]) 0) total-input-cache-read-tokens (or (get-in db [:chats chat-id :usage :total-input-cache-read-tokens]) 0) @@ -344,6 +353,7 @@ last-input-cache-read-tokens last-input-cache-creation-tokens last-output-tokens)} + :message-output-tps message-output-tps :limit (:limit model-capabilities) :last-message-cost (tokens->cost last-input-tokens last-input-cache-creation-tokens last-input-cache-read-tokens last-output-tokens model-capabilities) :session-cost (tokens->cost total-input-tokens total-input-cache-creation-tokens total-input-cache-read-tokens total-output-tokens model-capabilities)))) @@ -368,12 +378,14 @@ tokens considered, but we save them for cost calculation - cost: we count the tokens in past requests done + current one" [{:keys [input-tokens output-tokens - input-cache-creation-tokens input-cache-read-tokens]} + input-cache-creation-tokens input-cache-read-tokens + message-output-tps]} full-model {:keys [chat-id db*] :as chat-ctx}] (when (and output-tokens input-tokens) (swap! db* assoc-in [:chats chat-id :usage :last-input-tokens] input-tokens) (swap! db* assoc-in [:chats chat-id :usage :last-output-tokens] output-tokens) + (swap! db* assoc-in [:chats chat-id :usage :last-output-tps] message-output-tps) (swap! db* update-in [:chats chat-id :usage :total-input-tokens] (fnil + 0) input-tokens) (swap! db* update-in [:chats chat-id :usage :total-output-tokens] (fnil + 0) output-tokens) (when input-cache-creation-tokens diff --git a/test/eca/llm_api_test.clj b/test/eca/llm_api_test.clj index 5b49d1ff1..8947ca01a 100644 --- a/test/eca/llm_api_test.clj +++ b/test/eca/llm_api_test.clj @@ -11,6 +11,7 @@ [eca.llm-providers.openai :as llm-providers.openai] [eca.llm-providers.openai-chat :as llm-providers.openai-chat] [eca.secrets :as secrets] + [eca.shared :as shared] [eca.test-helper :as h] [matcher-combinators.test :refer [match?]])) @@ -1214,6 +1215,50 @@ (is (= 1 @attempt*)) (is (true? @on-error-called*))))) +(deftest message-output-tps-test + (testing "usage gets TPS from output tokens and measured time" + (let [usage* (atom nil) + tps-args* (atom nil)] + (with-redefs [eca.llm-api/prompt! (fn [{:keys [on-message-received on-usage-updated]}] + (on-message-received {:type :text :text "hi"}) + (Thread/sleep 20) + (on-usage-updated {:input-tokens 1000 :output-tokens 2})) + shared/tokens-per-second (fn [tokens elapsed-ms] + (reset! tps-args* [tokens elapsed-ms]) + 42)] + (llm-api/sync-or-async-prompt! + (make-prompt-opts + {:on-message-received identity + :on-usage-updated #(reset! usage* %)}))) + (let [[tokens elapsed-ms] @tps-args*] + (is (= 2 tokens) "uses output tokens") + (is (and (number? elapsed-ms) (pos? elapsed-ms)) "measures elapsed time")) + (is (= 42 (:message-output-tps @usage*)) "attaches the computed TPS"))) + + (testing "usage has no TPS without a generation event" + (let [usage* (atom nil)] + (with-redefs [eca.llm-api/prompt! (fn [{:keys [on-usage-updated]}] + (on-usage-updated {:input-tokens 1 :output-tokens 2}))] + (llm-api/sync-or-async-prompt! + (make-prompt-opts + {:on-message-received identity + :on-usage-updated #(reset! usage* %)}))) + (is (nil? (:message-output-tps @usage*))))) + + (testing "usage after tools has no TPS because tool time is not generation time" + (let [usage* (atom nil)] + (with-redefs [eca.llm-api/prompt! (fn [{:keys [on-message-received on-tools-called on-usage-updated]}] + (on-message-received {:type :text :text "hi"}) + (Thread/sleep 20) + (on-tools-called []) + (on-usage-updated {:input-tokens 1 :output-tokens 2}))] + (llm-api/sync-or-async-prompt! + (make-prompt-opts + {:on-message-received identity + :on-tools-called (constantly nil) + :on-usage-updated #(reset! usage* %)}))) + (is (nil? (:message-output-tps @usage*)))))) + (deftest async-retry-on-custom-retry-rule-error-pattern-test (testing "retries async when custom retryRules errorPattern matches error message" (let [attempt* (atom 0) diff --git a/test/eca/shared_test.clj b/test/eca/shared_test.clj index d06a2c871..857645756 100644 --- a/test/eca/shared_test.clj +++ b/test/eca/shared_test.clj @@ -14,8 +14,13 @@ :limit {:context 1000 :output 100}}} :provider-quotas {"anthropic" {:updated-at 1 :windows [{:id "5h" :label "5h" :used-percent 42.0}]}}}] (testing "API billing shows costs" - (is (match? {:session-tokens 15 :last-message-cost "0.20" :session-cost "2.00" :quota m/absent} + (is (match? {:session-tokens 15 :last-message-cost "0.20" :session-cost "2.00" + :quota m/absent :message-output-tps m/absent} (shared/usage-sumary "c1" "anthropic/claude" db)))) + (testing "includes the last response TPS when known" + (is (match? {:message-output-tps 42} + (shared/usage-sumary "c1" "anthropic/claude" + (assoc-in db [:chats "c1" :usage :last-output-tps] 42))))) (testing "subscription logins show the provider quota instead of costs" (is (match? {:session-tokens 15 :limit {:context 1000} @@ -174,6 +179,26 @@ (testing "returns nil when mandatory costs are missing" (is (nil? (shared/tokens->cost 30 nil nil 20 {:input-token-cost 0.01})))))) +(deftest tokens-per-second-test + (is (= 50 (shared/tokens-per-second 100 2000))) + (is (= 3 (shared/tokens-per-second 10 3000))) + (is (= 3 (shared/tokens-per-second 10 4000)) "rounds 2.5 up") + (is (nil? (shared/tokens-per-second 0 1000))) + (is (nil? (shared/tokens-per-second 10 0))) + (is (nil? (shared/tokens-per-second nil 1000)))) + +(deftest usage-msg->usage-test + (let [db* (atom {:chats {"c1" {:usage {}}}}) + chat-ctx {:chat-id "c1" :db* db*}] + (shared/usage-msg->usage {:input-tokens 10 :output-tokens 20 :message-output-tps 42} + "anthropic/claude" + chat-ctx) + (is (= 42 (get-in @db* [:chats "c1" :usage :last-output-tps]))) + (shared/usage-msg->usage {:input-tokens 10 :output-tokens 20} + "anthropic/claude" + chat-ctx) + (is (nil? (get-in @db* [:chats "c1" :usage :last-output-tps]))))) + (deftest deep-merge-test (testing "second map as nil returns first map" (is (match?