diff --git a/api_webchat/controller.js b/api_webchat/controller.js index 7434816e..dc8a30c2 100644 --- a/api_webchat/controller.js +++ b/api_webchat/controller.js @@ -156,6 +156,10 @@ worker.onmessage = async function(event) { messagesArea.handleNewToken(payload.token); messageInput.setStatusMessage(browser.i18n.getMessage("apiwebchat_receiving_data") + '...'); break; + case 'tokenBatch': + messagesArea.handleTokenBatch(payload.tokens); + messageInput.setStatusMessage(browser.i18n.getMessage("apiwebchat_receiving_data") + '...'); + break; case 'tokensDone': await messagesArea.handleTokensDone(promptData); messageInput.enableInput(); diff --git a/api_webchat/messagesArea.js b/api_webchat/messagesArea.js index eb61eca0..997cb748 100644 --- a/api_webchat/messagesArea.js +++ b/api_webchat/messagesArea.js @@ -324,6 +324,32 @@ class MessagesArea extends HTMLElement { } } + // Optimized method to handle batched tokens + handleTokenBatch(tokens) { + + if (!this.accumulatingMessageEl) { + this.createNewAccumulatingMessage(); + } + + // Create a single text node instead of multiple spans for better performance + const batchElement = document.createElement('span'); + batchElement.classList.add('token'); + batchElement.textContent = tokens; + this.accumulatingMessageEl.appendChild(batchElement); + + // Only scroll if we haven't scrolled recently (reduce scroll frequency) + const now = performance.now(); + if (!this.lastScrollTime || now - this.lastScrollTime > 100) { // Max 10 scrolls per second + this.scrollToBottom(); + this.lastScrollTime = now; + } + + // Check for newlines in the batch + if (tokens.includes('\n')) { + this.flushAccumulatingMessage(); + } + } + scrollToBottom() { this.messages.scrollTop = this.messages.scrollHeight; } diff --git a/js/api/google_gemini.js b/js/api/google_gemini.js index d0d8a055..fcf72767 100644 --- a/js/api/google_gemini.js +++ b/js/api/google_gemini.js @@ -24,6 +24,7 @@ export class GoogleGemini { model = ''; system_instruction = ''; stream = false; + adaptiveStreaming = true; // Smart streaming constructor(apiKey, model, system_instruction, stream) { this.apiKey = apiKey; @@ -70,14 +71,14 @@ export class GoogleGemini { } fetchResponse = async (messages) => { + // Smart streaming: disabilita streaming per risposte piccole + const messageLength = messages.map(m => m.parts?.map(p => p.text).join('') || '').join(''); + const shouldStream = false; ///this.stream && (messageLength > 500 || !this.adaptiveStreaming); + //console.log(">>>>>>>>>> Google Gemini shouldStream: " + shouldStream); try { - let google_gemini_body = { contents:messages }; - - // console.log("[ThunderAI] Google Gemini API system_instruction: " + JSON.stringify(this.system_instruction)); - if(this.system_instruction !== '') { google_gemini_body.system_instruction = { parts:{ @@ -85,10 +86,10 @@ export class GoogleGemini { } } } - + // console.log("[ThunderAI] Google Gemini API request: " + JSON.stringify(google_gemini_body)); - const response = await fetch("https://generativelanguage.googleapis.com/v1beta/models/" + this.model + ":" + (this.stream ? 'streamGenerateContent?alt=sse&' : 'generateContent?') + "key=" + this.apiKey, { + const response = await fetch("https://generativelanguage.googleapis.com/v1beta/models/" + this.model + ":" + (shouldStream ? 'streamGenerateContent?alt=sse&' : 'generateContent?') + "key=" + this.apiKey, { method: "POST", headers: { "Content-Type": "application/json" diff --git a/js/api/openai.js b/js/api/openai.js index 11884e7e..225f7d46 100644 --- a/js/api/openai.js +++ b/js/api/openai.js @@ -33,6 +33,7 @@ export class OpenAI { this.developer_messages = developer_messages; this.stream = stream; this.store = store; + this.adaptiveStreaming = true; // Enable adaptive streaming based on request size } @@ -73,28 +74,34 @@ export class OpenAI { } fetchResponse = async (messages, maxTokens = 0) => { - + if(this.developer_messages !== ''){ messages.push({role: "developer", content: [{"type": "text", "text": this.developer_messages}]}); } + // console.log(">>>>>>>>>>> OpenAI API request: " + JSON.stringify(messages)); + + // Determine if we should use streaming based on request complexity + const messageLength = messages.map(m => m.content).join('').length; + const shouldStream = this.stream && (messageLength > 500 || !this.adaptiveStreaming); + + const requestBody = { + model: this.model, + messages: messages, + stream: shouldStream, + store: this.store, + ...(maxTokens > 0 ? { 'max_tokens': parseInt(maxTokens) } : {}) + }; try { + const response = await fetch("https://api.openai.com/v1/chat/completions", { method: "POST", - headers: { - "Content-Type": "application/json", - Authorization: "Bearer "+ this.apiKey - }, - body: JSON.stringify({ - model: this.model, - messages: messages, - stream: this.stream, - store: this.store, - ...(maxTokens > 0 ? { 'max_tokens': parseInt(maxTokens) } : {}) - }), + headers: headers, + body: bodyString, }); + return response; }catch (error) { console.error("[ThunderAI] OpenAI API request failed: " + error); diff --git a/js/workers/model-worker-google_gemini.js b/js/workers/model-worker-google_gemini.js index e4557f52..4a764c04 100644 --- a/js/workers/model-worker-google_gemini.js +++ b/js/workers/model-worker-google_gemini.js @@ -34,6 +34,58 @@ let taLog = null; let conversationHistory = []; let assistantResponseAccumulator = ''; +// Token batching configuration - optimized for better performance +const TOKEN_BATCH_SIZE = 25; // Send tokens in batches of 25 characters +const TOKEN_BATCH_DELAY = 25; // Max 25ms between batches +const TOKEN_BATCH_TIMEOUT = 100; // Force flush after 100ms regardless of size +let tokenBatch = ''; +let batchTimer = null; +let timeoutTimer = null; +let lastBatchTime = 0; +let batchStartTime = 0; + +// Function to send batched tokens +function sendTokenBatch(force = false, reason = 'unknown') { + if (tokenBatch && (force || tokenBatch.length >= TOKEN_BATCH_SIZE || performance.now() - lastBatchTime >= TOKEN_BATCH_DELAY)) { + postMessage({ type: 'tokenBatch', payload: { tokens: tokenBatch } }); + // Reset batch state + tokenBatch = ''; + lastBatchTime = performance.now(); + batchStartTime = 0; + // Clear all timers + if (batchTimer) { + clearTimeout(batchTimer); + batchTimer = null; + } + if (timeoutTimer) { + clearTimeout(timeoutTimer); + timeoutTimer = null; + } + } +} + +// Function to add token to batch +function addTokenToBatch(token) { + tokenBatch += token; + // Set batch start time for the first token + if (tokenBatch.length === token.length) { + batchStartTime = performance.now(); + } + // Send immediately if batch is full + if (tokenBatch.length >= TOKEN_BATCH_SIZE) { + sendTokenBatch(true, 'size-limit'); + } else { + // Set timer to send batch if it's the first token in a new batch + if (tokenBatch.length === token.length && !batchTimer) { + batchTimer = setTimeout(() => sendTokenBatch(true, 'delay-timeout'), TOKEN_BATCH_DELAY); + } + // Set timeout-based flushing if not already set + if (!timeoutTimer) { + timeoutTimer = setTimeout(() => sendTokenBatch(true, 'timeout-flush'), TOKEN_BATCH_TIMEOUT); + } + } +} + self.onmessage = async function(event) { if (event.data.type === 'init') { google_gemini_api_key = event.data.google_gemini_api_key; @@ -70,11 +122,12 @@ self.onmessage = async function(event) { const reader = response.body.getReader(); const decoder = new TextDecoder("utf-8"); let buffer = ''; - while (true) { if (stopStreaming) { stopStreaming = false; reader.cancel(); + // Send any remaining tokens in the batch + sendTokenBatch(true, 'stream-stop'); conversationHistory.push({ role: 'model', parts: [{"text": assistantResponseAccumulator}] }); assistantResponseAccumulator = ''; postMessage({ type: 'tokensDone' }); @@ -82,6 +135,8 @@ self.onmessage = async function(event) { } const { done, value } = await reader.read(); if (done) { + // Send any remaining tokens in the batch + sendTokenBatch(true, 'stream-stop'); conversationHistory.push({ role: 'model', parts: [{"text": assistantResponseAccumulator}] }); assistantResponseAccumulator = ''; postMessage({ type: 'tokensDone' }); @@ -94,6 +149,7 @@ self.onmessage = async function(event) { const lines = buffer.split("\n"); buffer = lines.pop(); let parsedLines = []; + //console.log(">>>>>>>>>>>>>>> lines: " + JSON.stringify(lines)); try{ parsedLines = lines .map((line) => line.replace(/^data: /, "").trim()) // Remove the "data: " prefix @@ -106,7 +162,6 @@ self.onmessage = async function(event) { }catch(e){ taLog.error("Error parsing lines: " + e); } - for (const parsedLine of parsedLines) { const { candidates } = parsedLine; const { content } = candidates[0]; @@ -115,7 +170,8 @@ self.onmessage = async function(event) { // Update the UI with the new content if (text) { assistantResponseAccumulator += text; - postMessage({ type: 'newToken', payload: { token: text } }); + // Add to batch instead of sending immediately + addTokenToBatch(text); } } } diff --git a/js/workers/model-worker-openai.js b/js/workers/model-worker-openai.js index 61e4bc22..963f6c39 100644 --- a/js/workers/model-worker-openai.js +++ b/js/workers/model-worker-openai.js @@ -34,6 +34,63 @@ let taLog = null; let conversationHistory = []; let assistantResponseAccumulator = ''; +// Token batching configuration - optimized for better performance +const TOKEN_BATCH_SIZE = 25; // Send tokens in batches of 25 characters +const TOKEN_BATCH_DELAY = 25; // Max 25ms between batches +const TOKEN_BATCH_TIMEOUT = 100; // Force flush after 100ms regardless of size +let tokenBatch = ''; +let batchTimer = null; +let timeoutTimer = null; +let lastBatchTime = 0; +let batchStartTime = 0; + +// Function to send batched tokens +function sendTokenBatch(force = false, reason = 'unknown') { + if (tokenBatch && (force || tokenBatch.length >= TOKEN_BATCH_SIZE || performance.now() - lastBatchTime >= TOKEN_BATCH_DELAY)) { + postMessage({ type: 'tokenBatch', payload: { tokens: tokenBatch } }); + + // Reset batch state + tokenBatch = ''; + lastBatchTime = performance.now(); + batchStartTime = 0; + + // Clear all timers + if (batchTimer) { + clearTimeout(batchTimer); + batchTimer = null; + } + if (timeoutTimer) { + clearTimeout(timeoutTimer); + timeoutTimer = null; + } + } +} + +// Function to add token to batch +function addTokenToBatch(token) { + tokenBatch += token; + + // Set batch start time for the first token + if (tokenBatch.length === token.length) { + batchStartTime = performance.now(); + } + + // Send immediately if batch is full + if (tokenBatch.length >= TOKEN_BATCH_SIZE) { + sendTokenBatch(true, 'size-limit'); + } else { + // Set timer to send batch if it's the first token in a new batch + if (tokenBatch.length === token.length && !batchTimer) { + batchTimer = setTimeout(() => sendTokenBatch(true, 'delay-timeout'), TOKEN_BATCH_DELAY); + } + + // Set timeout-based flushing if not already set + if (!timeoutTimer) { + timeoutTimer = setTimeout(() => sendTokenBatch(true, 'timeout-flush'), TOKEN_BATCH_TIMEOUT); + } + } +} + self.onmessage = async function(event) { if (event.data.type === 'init') { chatgpt_api_key = event.data.chatgpt_api_key; @@ -43,9 +100,11 @@ self.onmessage = async function(event) { i18nStrings = event.data.i18nStrings; taLog = new taLogger('model-worker-openai', do_debug); } else if (event.data.type === 'chatMessage') { + conversationHistory.push({ role: 'user', content: event.data.message }); - const response = await openai.fetchResponse(conversationHistory); //4096); + const response = await openai.fetchResponse(conversationHistory); //4096); + postMessage({ type: 'messageSent' }); if (!response.ok) { @@ -67,26 +126,61 @@ self.onmessage = async function(event) { throw new Error("[ThunderAI] OpenAI ChatGPT API request failed: " + response.status + " " + response.statusText + ", Detail: " + error_message + " " + errorDetail); } + // Check if this is a streaming or non-streaming response + const isStreaming = response.headers.get('content-type')?.includes('text/event-stream'); + + if (!isStreaming) { + // Handle non-streaming response + try { + const responseData = await response.json(); + const content = responseData.choices[0].message.content; + + conversationHistory.push({ role: 'assistant', content: content }); + assistantResponseAccumulator = content; + + // Send the complete response as a single batch + postMessage({ type: 'tokenBatch', payload: { tokens: content } }); + postMessage({ type: 'tokensDone' }); + return; + } catch (error) { + console.error(`[ThunderAI Worker Debug] Error processing non-streaming response:`, error); + postMessage({ type: 'error', payload: 'Failed to process non-streaming response: ' + error.message }); + return; + } + } + const reader = response.body.getReader(); const decoder = new TextDecoder("utf-8"); let buffer = ''; + + console.log(`[ThunderAI Worker Debug] Starting to read stream...`); while (true) { if (stopStreaming) { stopStreaming = false; reader.cancel(); + + // Send any remaining tokens in the batch + sendTokenBatch(true, 'stream-stop'); + conversationHistory.push({ role: 'assistant', content: assistantResponseAccumulator }); assistantResponseAccumulator = ''; postMessage({ type: 'tokensDone' }); break; } const { done, value } = await reader.read(); + if (done) { + // Send any remaining tokens in the batch + sendTokenBatch(true, 'stream-stop'); + conversationHistory.push({ role: 'assistant', content: assistantResponseAccumulator }); assistantResponseAccumulator = ''; + postMessage({ type: 'tokensDone' }); break; } + // lots of low-level OpenAI response parsing stuff const chunk = decoder.decode(value); buffer += chunk; @@ -114,7 +208,9 @@ self.onmessage = async function(event) { // Update the UI with the new content if (content) { assistantResponseAccumulator += content; - postMessage({ type: 'newToken', payload: { token: content } }); + + // Add to batch instead of sending immediately + addTokenToBatch(content); } } }