From 6eb7b75296092fe831bcf85e7df2debdd5ce8a3c Mon Sep 17 00:00:00 2001 From: Michael Kret Date: Mon, 22 Sep 2025 12:10:34 +0300 Subject: [PATCH] fix, wip --- .../actions/text/message.operation.ts | 10 +- .../vendors/GoogleGemini/transport/index.ts | 105 ++++++++++++++++++ 2 files changed, 108 insertions(+), 7 deletions(-) diff --git a/packages/@n8n/nodes-langchain/nodes/vendors/GoogleGemini/actions/text/message.operation.ts b/packages/@n8n/nodes-langchain/nodes/vendors/GoogleGemini/actions/text/message.operation.ts index ae5d14e89ee..e4f12411923 100644 --- a/packages/@n8n/nodes-langchain/nodes/vendors/GoogleGemini/actions/text/message.operation.ts +++ b/packages/@n8n/nodes-langchain/nodes/vendors/GoogleGemini/actions/text/message.operation.ts @@ -17,7 +17,7 @@ import type { Tool, GenerateContentGenerationConfig, } from '../../helpers/interfaces'; -import { apiRequest } from '../../transport'; +import { streamingRequestWithFallback } from '../../transport'; import { modelRLC } from '../descriptions'; const properties: INodeProperties[] = [ @@ -319,9 +319,7 @@ export async function execute(this: IExecuteFunctions, i: number): Promise { + const { body, qs, option, headers } = parameters ?? {}; + + const credentials = await this.getCredentials('googlePalmApi'); + + let url = `https://generativelanguage.googleapis.com${endpoint}`; + + if (credentials.host) { + url = `${credentials.host}${endpoint}`; + } + + const streamingQs = { ...qs, alt: 'sse' }; + + const options = { + headers: { + ...headers, + Accept: 'text/event-stream', + }, + method, + body, + qs: streamingQs, + url, + json: false, + }; + + if (option && Object.keys(option).length !== 0) { + Object.assign(options, option); + } + + const response = await this.helpers.httpRequestWithAuthentication.call( + this, + 'googlePalmApi', + options, + ); + + return parseStreamingResponse(response); +} + +function parseStreamingResponse(streamData: string): any { + const lines = streamData.split('\n'); + const chunks = []; + for (const line of lines) { + if (line.startsWith('data: ')) { + const data = line.slice(6); + if (data === '[DONE]') { + break; + } + try { + const parsed = JSON.parse(data); + chunks.push(parsed); + } catch (e) { + continue; + } + } + } + if (chunks.length === 0) { + throw new NodeOperationError( + null as any, + 'No valid response chunks received from streaming API', + ); + } + + const combinedResponse = { + candidates: [ + { + content: { + parts: [ + { + //TODO: combine other fields + text: chunks + .filter((chunk) => chunk.candidates?.[0]?.content?.parts?.[0]?.text) + .map((chunk) => chunk.candidates[0].content.parts[0].text) + .join(''), + }, + ], + role: 'model', + }, + }, + ], + }; + return combinedResponse; +} + +export async function streamingRequestWithFallback( + this: IExecuteFunctions, + model: string, + body: GenerateContentRequest, +) { + try { + return (await apiRequestStream.call(this, 'POST', `/v1beta/${model}:streamGenerateContent`, { + body, + })) as GenerateContentResponse; + } catch (streamError: any) { + return (await apiRequest.call(this, 'POST', `/v1beta/${model}:generateContent`, { + body, + })) as GenerateContentResponse; + } +}