From 002f1e2f36d19693304a29b505f701ff1621d82f Mon Sep 17 00:00:00 2001 From: "mingholy.lmh" Date: Mon, 1 Sep 2025 20:12:50 +0800 Subject: [PATCH] refactor: openaiContentGenerator --- packages/core/src/core/contentGenerator.ts | 10 +- packages/core/src/core/refactor/constants.ts | 2 + packages/core/src/core/refactor/converter.ts | 878 ++++++++++++++++++ .../core/src/core/refactor/errorHandler.ts | 129 +++ packages/core/src/core/refactor/index.ts | 88 ++ .../core/refactor/openaiContentGenerator.ts | 81 ++ packages/core/src/core/refactor/pipeline.ts | 380 ++++++++ .../core/src/core/refactor/provider/README.md | 61 ++ .../src/core/refactor/provider/dashscope.ts | 210 +++++ .../src/core/refactor/provider/default.ts | 58 ++ .../core/src/core/refactor/provider/index.ts | 9 + .../src/core/refactor/provider/openrouter.ts | 64 ++ .../core/src/core/refactor/provider/types.ts | 28 + .../src/core/refactor/streamingManager.ts | 111 +++ .../src/core/refactor/telemetryService.ts | 137 +++ .../src/qwen/qwenContentGenerator.test.ts | 35 +- .../core/src/qwen/qwenContentGenerator.ts | 59 +- 17 files changed, 2283 insertions(+), 57 deletions(-) create mode 100644 packages/core/src/core/refactor/constants.ts create mode 100644 packages/core/src/core/refactor/converter.ts create mode 100644 packages/core/src/core/refactor/errorHandler.ts create mode 100644 packages/core/src/core/refactor/index.ts create mode 100644 packages/core/src/core/refactor/openaiContentGenerator.ts create mode 100644 packages/core/src/core/refactor/pipeline.ts create mode 100644 packages/core/src/core/refactor/provider/README.md create mode 100644 packages/core/src/core/refactor/provider/dashscope.ts create mode 100644 packages/core/src/core/refactor/provider/default.ts create mode 100644 packages/core/src/core/refactor/provider/index.ts create mode 100644 packages/core/src/core/refactor/provider/openrouter.ts create mode 100644 packages/core/src/core/refactor/provider/types.ts create mode 100644 packages/core/src/core/refactor/streamingManager.ts create mode 100644 packages/core/src/core/refactor/telemetryService.ts diff --git a/packages/core/src/core/contentGenerator.ts b/packages/core/src/core/contentGenerator.ts index 476ebfe0..a375cdce 100644 --- a/packages/core/src/core/contentGenerator.ts +++ b/packages/core/src/core/contentGenerator.ts @@ -208,19 +208,13 @@ export async function createContentGenerator( } // Import OpenAIContentGenerator dynamically to avoid circular dependencies - const { OpenAIContentGenerator } = await import( - './openaiContentGenerator.js' - ); + const { createContentGenerator } = await import('./refactor/index.js'); // Always use OpenAIContentGenerator, logging is controlled by enableOpenAILogging flag - return new OpenAIContentGenerator(config, gcConfig); + return createContentGenerator(config, gcConfig); } if (config.authType === AuthType.QWEN_OAUTH) { - if (config.apiKey !== 'QWEN_OAUTH_DYNAMIC_TOKEN') { - throw new Error('Invalid Qwen OAuth configuration'); - } - // Import required classes dynamically const { getQwenOAuthClient: getQwenOauthClient } = await import( '../qwen/qwenOAuth2.js' diff --git a/packages/core/src/core/refactor/constants.ts b/packages/core/src/core/refactor/constants.ts new file mode 100644 index 00000000..d2b5ce81 --- /dev/null +++ b/packages/core/src/core/refactor/constants.ts @@ -0,0 +1,2 @@ +export const DEFAULT_TIMEOUT = 120000; +export const DEFAULT_MAX_RETRIES = 3; diff --git a/packages/core/src/core/refactor/converter.ts b/packages/core/src/core/refactor/converter.ts new file mode 100644 index 00000000..706c56d7 --- /dev/null +++ b/packages/core/src/core/refactor/converter.ts @@ -0,0 +1,878 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import { + GenerateContentResponse, + GenerateContentParameters, + FinishReason, + Part, + Content, + Tool, + ToolListUnion, + CallableTool, + FunctionCall, + FunctionResponse, +} from '@google/genai'; +import OpenAI from 'openai'; +import { safeJsonParse } from '../../utils/safeJsonParse.js'; + +/** + * Converter class for transforming data between Gemini and OpenAI formats + */ +export class Converter { + private model: string; + private streamingToolCalls: Map< + number, + { + id?: string; + name?: string; + arguments: string; + } + > = new Map(); + + constructor(model: string) { + this.model = model; + } + + /** + * Convert Gemini tool parameters to OpenAI JSON Schema format + */ + convertGeminiToolParametersToOpenAI( + parameters: Record, + ): Record | undefined { + if (!parameters || typeof parameters !== 'object') { + return parameters; + } + + const converted = JSON.parse(JSON.stringify(parameters)); + + const convertTypes = (obj: unknown): unknown => { + if (typeof obj !== 'object' || obj === null) { + return obj; + } + + if (Array.isArray(obj)) { + return obj.map(convertTypes); + } + + const result: Record = {}; + for (const [key, value] of Object.entries(obj)) { + if (key === 'type' && typeof value === 'string') { + // Convert Gemini types to OpenAI JSON Schema types + const lowerValue = value.toLowerCase(); + if (lowerValue === 'integer') { + result[key] = 'integer'; + } else if (lowerValue === 'number') { + result[key] = 'number'; + } else { + result[key] = lowerValue; + } + } else if ( + key === 'minimum' || + key === 'maximum' || + key === 'multipleOf' + ) { + // Ensure numeric constraints are actual numbers, not strings + if (typeof value === 'string' && !isNaN(Number(value))) { + result[key] = Number(value); + } else { + result[key] = value; + } + } else if ( + key === 'minLength' || + key === 'maxLength' || + key === 'minItems' || + key === 'maxItems' + ) { + // Ensure length constraints are integers, not strings + if (typeof value === 'string' && !isNaN(Number(value))) { + result[key] = parseInt(value, 10); + } else { + result[key] = value; + } + } else if (typeof value === 'object') { + result[key] = convertTypes(value); + } else { + result[key] = value; + } + } + return result; + }; + + return convertTypes(converted) as Record | undefined; + } + + /** + * Convert Gemini tools to OpenAI format for API compatibility. + * Handles both Gemini tools (using 'parameters' field) and MCP tools (using 'parametersJsonSchema' field). + */ + async convertGeminiToolsToOpenAI( + geminiTools: ToolListUnion, + ): Promise { + const openAITools: OpenAI.Chat.ChatCompletionTool[] = []; + + for (const tool of geminiTools) { + let actualTool: Tool; + + // Handle CallableTool vs Tool + if ('tool' in tool) { + // This is a CallableTool + actualTool = await (tool as CallableTool).tool(); + } else { + // This is already a Tool + actualTool = tool as Tool; + } + + if (actualTool.functionDeclarations) { + for (const func of actualTool.functionDeclarations) { + if (func.name && func.description) { + let parameters: Record | undefined; + + // Handle both Gemini tools (parameters) and MCP tools (parametersJsonSchema) + if (func.parametersJsonSchema) { + // MCP tool format - use parametersJsonSchema directly + if (func.parametersJsonSchema) { + // Create a shallow copy to avoid mutating the original object + const paramsCopy = { + ...(func.parametersJsonSchema as Record), + }; + parameters = paramsCopy; + } + } else if (func.parameters) { + // Gemini tool format - convert parameters to OpenAI format + parameters = this.convertGeminiToolParametersToOpenAI( + func.parameters as Record, + ); + } + + openAITools.push({ + type: 'function', + function: { + name: func.name, + description: func.description, + parameters, + }, + }); + } + } + } + } + + return openAITools; + } + + /** + * Convert Gemini request to OpenAI message format + */ + convertGeminiRequestToOpenAI( + request: GenerateContentParameters, + ): OpenAI.Chat.ChatCompletionMessageParam[] { + const messages: OpenAI.Chat.ChatCompletionMessageParam[] = []; + + // Handle system instruction from config + if (request.config?.systemInstruction) { + const systemInstruction = request.config.systemInstruction; + let systemText = ''; + + if (Array.isArray(systemInstruction)) { + systemText = systemInstruction + .map((content) => { + if (typeof content === 'string') return content; + if ('parts' in content) { + const contentObj = content as Content; + return ( + contentObj.parts + ?.map((p: Part) => + typeof p === 'string' ? p : 'text' in p ? p.text : '', + ) + .join('\n') || '' + ); + } + return ''; + }) + .join('\n'); + } else if (typeof systemInstruction === 'string') { + systemText = systemInstruction; + } else if ( + typeof systemInstruction === 'object' && + 'parts' in systemInstruction + ) { + const systemContent = systemInstruction as Content; + systemText = + systemContent.parts + ?.map((p: Part) => + typeof p === 'string' ? p : 'text' in p ? p.text : '', + ) + .join('\n') || ''; + } + + if (systemText) { + messages.push({ + role: 'system' as const, + content: systemText, + }); + } + } + + // Handle contents + if (Array.isArray(request.contents)) { + for (const content of request.contents) { + if (typeof content === 'string') { + messages.push({ role: 'user' as const, content }); + } else if ('role' in content && 'parts' in content) { + // Check if this content has function calls or responses + const functionCalls: FunctionCall[] = []; + const functionResponses: FunctionResponse[] = []; + const textParts: string[] = []; + + for (const part of content.parts || []) { + if (typeof part === 'string') { + textParts.push(part); + } else if ('text' in part && part.text) { + textParts.push(part.text); + } else if ('functionCall' in part && part.functionCall) { + functionCalls.push(part.functionCall); + } else if ('functionResponse' in part && part.functionResponse) { + functionResponses.push(part.functionResponse); + } + } + + // Handle function responses (tool results) + if (functionResponses.length > 0) { + for (const funcResponse of functionResponses) { + messages.push({ + role: 'tool' as const, + tool_call_id: funcResponse.id || '', + content: + typeof funcResponse.response === 'string' + ? funcResponse.response + : JSON.stringify(funcResponse.response), + }); + } + } + // Handle model messages with function calls + else if (content.role === 'model' && functionCalls.length > 0) { + const toolCalls = functionCalls.map((fc, index) => ({ + id: fc.id || `call_${index}`, + type: 'function' as const, + function: { + name: fc.name || '', + arguments: JSON.stringify(fc.args || {}), + }, + })); + + messages.push({ + role: 'assistant' as const, + content: textParts.join('') || null, + tool_calls: toolCalls, + }); + } + // Handle regular text messages + else { + const role = + content.role === 'model' + ? ('assistant' as const) + : ('user' as const); + const text = textParts.join(''); + if (text) { + messages.push({ role, content: text }); + } + } + } + } + } else if (request.contents) { + if (typeof request.contents === 'string') { + messages.push({ role: 'user' as const, content: request.contents }); + } else if ('role' in request.contents && 'parts' in request.contents) { + const content = request.contents; + const role = + content.role === 'model' ? ('assistant' as const) : ('user' as const); + const text = + content.parts + ?.map((p: Part) => + typeof p === 'string' ? p : 'text' in p ? p.text : '', + ) + .join('\n') || ''; + messages.push({ role, content: text }); + } + } + + // Clean up orphaned tool calls and merge consecutive assistant messages + const cleanedMessages = this.cleanOrphanedToolCalls(messages); + const mergedMessages = + this.mergeConsecutiveAssistantMessages(cleanedMessages); + + return mergedMessages; + } + + /** + * Convert OpenAI response to Gemini format + */ + convertOpenAIResponseToGemini( + openaiResponse: OpenAI.Chat.ChatCompletion, + ): GenerateContentResponse { + const choice = openaiResponse.choices[0]; + const response = new GenerateContentResponse(); + + const parts: Part[] = []; + + // Handle text content + if (choice.message.content) { + parts.push({ text: choice.message.content }); + } + + // Handle tool calls + if (choice.message.tool_calls) { + for (const toolCall of choice.message.tool_calls) { + if (toolCall.function) { + let args: Record = {}; + if (toolCall.function.arguments) { + args = safeJsonParse(toolCall.function.arguments, {}); + } + + parts.push({ + functionCall: { + id: toolCall.id, + name: toolCall.function.name, + args, + }, + }); + } + } + } + + response.responseId = openaiResponse.id; + response.createTime = openaiResponse.created + ? openaiResponse.created.toString() + : new Date().getTime().toString(); + + response.candidates = [ + { + content: { + parts, + role: 'model' as const, + }, + finishReason: this.mapOpenAIFinishReasonToGemini( + choice.finish_reason || 'stop', + ), + index: 0, + safetyRatings: [], + }, + ]; + + response.modelVersion = this.model; + response.promptFeedback = { safetyRatings: [] }; + + // Add usage metadata if available + if (openaiResponse.usage) { + const usage = openaiResponse.usage; + + const promptTokens = usage.prompt_tokens || 0; + const completionTokens = usage.completion_tokens || 0; + const totalTokens = usage.total_tokens || 0; + const cachedTokens = usage.prompt_tokens_details?.cached_tokens || 0; + + // If we only have total tokens but no breakdown, estimate the split + // Typically input is ~70% and output is ~30% for most conversations + let finalPromptTokens = promptTokens; + let finalCompletionTokens = completionTokens; + + if (totalTokens > 0 && promptTokens === 0 && completionTokens === 0) { + // Estimate: assume 70% input, 30% output + finalPromptTokens = Math.round(totalTokens * 0.7); + finalCompletionTokens = Math.round(totalTokens * 0.3); + } + + response.usageMetadata = { + promptTokenCount: finalPromptTokens, + candidatesTokenCount: finalCompletionTokens, + totalTokenCount: totalTokens, + cachedContentTokenCount: cachedTokens, + }; + } + + return response; + } + + /** + * Convert OpenAI stream chunk to Gemini format + */ + convertOpenAIChunkToGemini( + chunk: OpenAI.Chat.ChatCompletionChunk, + ): GenerateContentResponse { + const choice = chunk.choices?.[0]; + const response = new GenerateContentResponse(); + + if (choice) { + const parts: Part[] = []; + + // Handle text content + if (choice.delta?.content) { + if (typeof choice.delta.content === 'string') { + parts.push({ text: choice.delta.content }); + } + } + + // Handle tool calls - only accumulate during streaming, emit when complete + if (choice.delta?.tool_calls) { + for (const toolCall of choice.delta.tool_calls) { + const index = toolCall.index ?? 0; + + // Get or create the tool call accumulator for this index + let accumulatedCall = this.streamingToolCalls.get(index); + if (!accumulatedCall) { + accumulatedCall = { arguments: '' }; + this.streamingToolCalls.set(index, accumulatedCall); + } + + // Update accumulated data + if (toolCall.id) { + accumulatedCall.id = toolCall.id; + } + if (toolCall.function?.name) { + // If this is a new function name, reset the arguments + if (accumulatedCall.name !== toolCall.function.name) { + accumulatedCall.arguments = ''; + } + accumulatedCall.name = toolCall.function.name; + } + if (toolCall.function?.arguments) { + // Check if we already have a complete JSON object + const currentArgs = accumulatedCall.arguments; + const newArgs = toolCall.function.arguments; + + // If current arguments already form a complete JSON and new arguments start a new object, + // this indicates a new tool call with the same name + let shouldReset = false; + if (currentArgs && newArgs.trim().startsWith('{')) { + try { + JSON.parse(currentArgs); + // If we can parse current arguments as complete JSON and new args start with {, + // this is likely a new tool call + shouldReset = true; + } catch { + // Current arguments are not complete JSON, continue accumulating + } + } + + if (shouldReset) { + accumulatedCall.arguments = newArgs; + } else { + accumulatedCall.arguments += newArgs; + } + } + } + } + + // Only emit function calls when streaming is complete (finish_reason is present) + if (choice.finish_reason) { + for (const [, accumulatedCall] of this.streamingToolCalls) { + if (accumulatedCall.name) { + let args: Record = {}; + if (accumulatedCall.arguments) { + args = safeJsonParse(accumulatedCall.arguments, {}); + } + + parts.push({ + functionCall: { + id: + accumulatedCall.id || + `call_${Date.now()}_${Math.random().toString(36).substring(2, 9)}`, + name: accumulatedCall.name, + args, + }, + }); + } + } + // Clear all accumulated tool calls + this.streamingToolCalls.clear(); + } + + response.candidates = [ + { + content: { + parts, + role: 'model' as const, + }, + finishReason: choice.finish_reason + ? this.mapOpenAIFinishReasonToGemini(choice.finish_reason) + : FinishReason.FINISH_REASON_UNSPECIFIED, + index: 0, + safetyRatings: [], + }, + ]; + } else { + response.candidates = []; + } + + response.responseId = chunk.id; + response.createTime = chunk.created + ? chunk.created.toString() + : new Date().getTime().toString(); + + response.modelVersion = this.model; + response.promptFeedback = { safetyRatings: [] }; + + // Add usage metadata if available in the chunk + if (chunk.usage) { + const usage = chunk.usage; + + const promptTokens = usage.prompt_tokens || 0; + const completionTokens = usage.completion_tokens || 0; + const totalTokens = usage.total_tokens || 0; + const cachedTokens = usage.prompt_tokens_details?.cached_tokens || 0; + + // If we only have total tokens but no breakdown, estimate the split + // Typically input is ~70% and output is ~30% for most conversations + let finalPromptTokens = promptTokens; + let finalCompletionTokens = completionTokens; + + if (totalTokens > 0 && promptTokens === 0 && completionTokens === 0) { + // Estimate: assume 70% input, 30% output + finalPromptTokens = Math.round(totalTokens * 0.7); + finalCompletionTokens = Math.round(totalTokens * 0.3); + } + + response.usageMetadata = { + promptTokenCount: finalPromptTokens, + candidatesTokenCount: finalCompletionTokens, + totalTokenCount: totalTokens, + cachedContentTokenCount: cachedTokens, + }; + } + + return response; + } + + /** + * Convert Gemini response format to OpenAI chat completion format for logging + */ + convertGeminiResponseToOpenAI( + response: GenerateContentResponse, + ): OpenAI.Chat.ChatCompletion { + const candidate = response.candidates?.[0]; + const content = candidate?.content; + + let messageContent: string | null = null; + const toolCalls: OpenAI.Chat.ChatCompletionMessageToolCall[] = []; + + if (content?.parts) { + const textParts: string[] = []; + + for (const part of content.parts) { + if ('text' in part && part.text) { + textParts.push(part.text); + } else if ('functionCall' in part && part.functionCall) { + toolCalls.push({ + id: part.functionCall.id || `call_${toolCalls.length}`, + type: 'function' as const, + function: { + name: part.functionCall.name || '', + arguments: JSON.stringify(part.functionCall.args || {}), + }, + }); + } + } + + messageContent = textParts.join('').trimEnd(); + } + + const choice: OpenAI.Chat.ChatCompletion.Choice = { + index: 0, + message: { + role: 'assistant', + content: messageContent, + refusal: null, + }, + finish_reason: this.mapGeminiFinishReasonToOpenAI( + candidate?.finishReason, + ) as OpenAI.Chat.ChatCompletion.Choice['finish_reason'], + logprobs: null, + }; + + if (toolCalls.length > 0) { + choice.message.tool_calls = toolCalls; + } + + const openaiResponse: OpenAI.Chat.ChatCompletion = { + id: response.responseId || `chatcmpl-${Date.now()}`, + object: 'chat.completion', + created: response.createTime + ? Number(response.createTime) + : Math.floor(Date.now() / 1000), + model: this.model, + choices: [choice], + }; + + // Add usage metadata if available + if (response.usageMetadata) { + openaiResponse.usage = { + prompt_tokens: response.usageMetadata.promptTokenCount || 0, + completion_tokens: response.usageMetadata.candidatesTokenCount || 0, + total_tokens: response.usageMetadata.totalTokenCount || 0, + }; + + if (response.usageMetadata.cachedContentTokenCount) { + openaiResponse.usage.prompt_tokens_details = { + cached_tokens: response.usageMetadata.cachedContentTokenCount, + }; + } + } + + return openaiResponse; + } + + /** + * Map OpenAI finish reasons to Gemini finish reasons + */ + private mapOpenAIFinishReasonToGemini( + openaiReason: string | null, + ): FinishReason { + if (!openaiReason) return FinishReason.FINISH_REASON_UNSPECIFIED; + const mapping: Record = { + stop: FinishReason.STOP, + length: FinishReason.MAX_TOKENS, + content_filter: FinishReason.SAFETY, + function_call: FinishReason.STOP, + tool_calls: FinishReason.STOP, + }; + return mapping[openaiReason] || FinishReason.FINISH_REASON_UNSPECIFIED; + } + + /** + * Map Gemini finish reasons to OpenAI finish reasons + */ + private mapGeminiFinishReasonToOpenAI(geminiReason?: unknown): string { + if (!geminiReason) return 'stop'; + + switch (geminiReason) { + case 'STOP': + case 1: // FinishReason.STOP + return 'stop'; + case 'MAX_TOKENS': + case 2: // FinishReason.MAX_TOKENS + return 'length'; + case 'SAFETY': + case 3: // FinishReason.SAFETY + return 'content_filter'; + case 'RECITATION': + case 4: // FinishReason.RECITATION + return 'content_filter'; + case 'OTHER': + case 5: // FinishReason.OTHER + return 'stop'; + default: + return 'stop'; + } + } + + /** + * Clean up orphaned tool calls from message history to prevent OpenAI API errors + */ + private cleanOrphanedToolCalls( + messages: OpenAI.Chat.ChatCompletionMessageParam[], + ): OpenAI.Chat.ChatCompletionMessageParam[] { + const cleaned: OpenAI.Chat.ChatCompletionMessageParam[] = []; + const toolCallIds = new Set(); + const toolResponseIds = new Set(); + + // First pass: collect all tool call IDs and tool response IDs + for (const message of messages) { + if ( + message.role === 'assistant' && + 'tool_calls' in message && + message.tool_calls + ) { + for (const toolCall of message.tool_calls) { + if (toolCall.id) { + toolCallIds.add(toolCall.id); + } + } + } else if ( + message.role === 'tool' && + 'tool_call_id' in message && + message.tool_call_id + ) { + toolResponseIds.add(message.tool_call_id); + } + } + + // Second pass: filter out orphaned messages + for (const message of messages) { + if ( + message.role === 'assistant' && + 'tool_calls' in message && + message.tool_calls + ) { + // Filter out tool calls that don't have corresponding responses + const validToolCalls = message.tool_calls.filter( + (toolCall) => toolCall.id && toolResponseIds.has(toolCall.id), + ); + + if (validToolCalls.length > 0) { + // Keep the message but only with valid tool calls + const cleanedMessage = { ...message }; + ( + cleanedMessage as OpenAI.Chat.ChatCompletionMessageParam & { + tool_calls?: OpenAI.Chat.ChatCompletionMessageToolCall[]; + } + ).tool_calls = validToolCalls; + cleaned.push(cleanedMessage); + } else if ( + typeof message.content === 'string' && + message.content.trim() + ) { + // Keep the message if it has text content, but remove tool calls + const cleanedMessage = { ...message }; + delete ( + cleanedMessage as OpenAI.Chat.ChatCompletionMessageParam & { + tool_calls?: OpenAI.Chat.ChatCompletionMessageToolCall[]; + } + ).tool_calls; + cleaned.push(cleanedMessage); + } + // If no valid tool calls and no content, skip the message entirely + } else if ( + message.role === 'tool' && + 'tool_call_id' in message && + message.tool_call_id + ) { + // Only keep tool responses that have corresponding tool calls + if (toolCallIds.has(message.tool_call_id)) { + cleaned.push(message); + } + } else { + // Keep all other messages as-is + cleaned.push(message); + } + } + + // Final validation: ensure every assistant message with tool_calls has corresponding tool responses + const finalCleaned: OpenAI.Chat.ChatCompletionMessageParam[] = []; + const finalToolCallIds = new Set(); + + // Collect all remaining tool call IDs + for (const message of cleaned) { + if ( + message.role === 'assistant' && + 'tool_calls' in message && + message.tool_calls + ) { + for (const toolCall of message.tool_calls) { + if (toolCall.id) { + finalToolCallIds.add(toolCall.id); + } + } + } + } + + // Verify all tool calls have responses + const finalToolResponseIds = new Set(); + for (const message of cleaned) { + if ( + message.role === 'tool' && + 'tool_call_id' in message && + message.tool_call_id + ) { + finalToolResponseIds.add(message.tool_call_id); + } + } + + // Remove any remaining orphaned tool calls + for (const message of cleaned) { + if ( + message.role === 'assistant' && + 'tool_calls' in message && + message.tool_calls + ) { + const finalValidToolCalls = message.tool_calls.filter( + (toolCall) => toolCall.id && finalToolResponseIds.has(toolCall.id), + ); + + if (finalValidToolCalls.length > 0) { + const cleanedMessage = { ...message }; + ( + cleanedMessage as OpenAI.Chat.ChatCompletionMessageParam & { + tool_calls?: OpenAI.Chat.ChatCompletionMessageToolCall[]; + } + ).tool_calls = finalValidToolCalls; + finalCleaned.push(cleanedMessage); + } else if ( + typeof message.content === 'string' && + message.content.trim() + ) { + const cleanedMessage = { ...message }; + delete ( + cleanedMessage as OpenAI.Chat.ChatCompletionMessageParam & { + tool_calls?: OpenAI.Chat.ChatCompletionMessageToolCall[]; + } + ).tool_calls; + finalCleaned.push(cleanedMessage); + } + } else { + finalCleaned.push(message); + } + } + + return finalCleaned; + } + + /** + * Merge consecutive assistant messages to combine split text and tool calls + */ + private mergeConsecutiveAssistantMessages( + messages: OpenAI.Chat.ChatCompletionMessageParam[], + ): OpenAI.Chat.ChatCompletionMessageParam[] { + const merged: OpenAI.Chat.ChatCompletionMessageParam[] = []; + + for (const message of messages) { + if (message.role === 'assistant' && merged.length > 0) { + const lastMessage = merged[merged.length - 1]; + + // If the last message is also an assistant message, merge them + if (lastMessage.role === 'assistant') { + // Combine content + const combinedContent = [ + typeof lastMessage.content === 'string' ? lastMessage.content : '', + typeof message.content === 'string' ? message.content : '', + ] + .filter(Boolean) + .join(''); + + // Combine tool calls + const lastToolCalls = + 'tool_calls' in lastMessage ? lastMessage.tool_calls || [] : []; + const currentToolCalls = + 'tool_calls' in message ? message.tool_calls || [] : []; + const combinedToolCalls = [...lastToolCalls, ...currentToolCalls]; + + // Update the last message with combined data + ( + lastMessage as OpenAI.Chat.ChatCompletionMessageParam & { + content: string | null; + tool_calls?: OpenAI.Chat.ChatCompletionMessageToolCall[]; + } + ).content = combinedContent || null; + if (combinedToolCalls.length > 0) { + ( + lastMessage as OpenAI.Chat.ChatCompletionMessageParam & { + content: string | null; + tool_calls?: OpenAI.Chat.ChatCompletionMessageToolCall[]; + } + ).tool_calls = combinedToolCalls; + } + + continue; // Skip adding the current message since it's been merged + } + } + + // Add the message as-is if no merging is needed + merged.push(message); + } + + return merged; + } +} diff --git a/packages/core/src/core/refactor/errorHandler.ts b/packages/core/src/core/refactor/errorHandler.ts new file mode 100644 index 00000000..780caa19 --- /dev/null +++ b/packages/core/src/core/refactor/errorHandler.ts @@ -0,0 +1,129 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import { GenerateContentParameters } from '@google/genai'; +import { RequestContext } from './telemetryService.js'; + +export interface ErrorHandler { + handle( + error: unknown, + context: RequestContext, + request: GenerateContentParameters, + ): never; + shouldSuppressErrorLogging( + error: unknown, + request: GenerateContentParameters, + ): boolean; +} + +export class EnhancedErrorHandler implements ErrorHandler { + constructor( + private shouldSuppressLogging: ( + error: unknown, + request: GenerateContentParameters, + ) => boolean = () => false, + ) {} + + handle( + error: unknown, + context: RequestContext, + request: GenerateContentParameters, + ): never { + const isTimeoutError = this.isTimeoutError(error); + const errorMessage = this.buildErrorMessage(error, context, isTimeoutError); + + // Allow subclasses to suppress error logging for specific scenarios + if (!this.shouldSuppressErrorLogging(error, request)) { + const logPrefix = context.isStreaming + ? 'OpenAI API Streaming Error:' + : 'OpenAI API Error:'; + console.error(logPrefix, errorMessage); + } + + // Provide helpful timeout-specific error message + if (isTimeoutError) { + throw new Error( + `${errorMessage}\n\n${this.getTimeoutTroubleshootingTips(context)}`, + ); + } + + throw error; + } + + shouldSuppressErrorLogging( + error: unknown, + request: GenerateContentParameters, + ): boolean { + return this.shouldSuppressLogging(error, request); + } + + private isTimeoutError(error: unknown): boolean { + if (!error) return false; + + const errorMessage = + error instanceof Error + ? error.message.toLowerCase() + : String(error).toLowerCase(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const errorCode = (error as any)?.code; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const errorType = (error as any)?.type; + + // Check for common timeout indicators + return ( + errorMessage.includes('timeout') || + errorMessage.includes('timed out') || + errorMessage.includes('connection timeout') || + errorMessage.includes('request timeout') || + errorMessage.includes('read timeout') || + errorMessage.includes('etimedout') || + errorMessage.includes('esockettimedout') || + errorCode === 'ETIMEDOUT' || + errorCode === 'ESOCKETTIMEDOUT' || + errorType === 'timeout' || + errorMessage.includes('request timed out') || + errorMessage.includes('deadline exceeded') + ); + } + + private buildErrorMessage( + error: unknown, + context: RequestContext, + isTimeoutError: boolean, + ): string { + const durationSeconds = Math.round(context.duration / 1000); + + if (isTimeoutError) { + const prefix = context.isStreaming + ? 'Streaming request timeout' + : 'Request timeout'; + return `${prefix} after ${durationSeconds}s. Try reducing input length or increasing timeout in config.`; + } + + return error instanceof Error ? error.message : String(error); + } + + private getTimeoutTroubleshootingTips(context: RequestContext): string { + const baseTitle = context.isStreaming + ? 'Streaming timeout troubleshooting:' + : 'Troubleshooting tips:'; + + const baseTips = [ + '- Reduce input length or complexity', + '- Increase timeout in config: contentGenerator.timeout', + '- Check network connectivity', + ]; + + const streamingSpecificTips = context.isStreaming + ? [ + '- Check network stability for streaming connections', + '- Consider using non-streaming mode for very long inputs', + ] + : ['- Consider using streaming mode for long responses']; + + return `${baseTitle}\n${[...baseTips, ...streamingSpecificTips].join('\n')}`; + } +} diff --git a/packages/core/src/core/refactor/index.ts b/packages/core/src/core/refactor/index.ts new file mode 100644 index 00000000..5cc4728b --- /dev/null +++ b/packages/core/src/core/refactor/index.ts @@ -0,0 +1,88 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import { + ContentGenerator, + ContentGeneratorConfig, +} from '../contentGenerator.js'; +import { Config } from '../../config/config.js'; +import { OpenAIContentGenerator } from './openaiContentGenerator.js'; +import { + DashScopeOpenAICompatibleProvider, + OpenRouterOpenAICompatibleProvider, + type OpenAICompatibleProvider, + DefaultOpenAICompatibleProvider, +} from './provider/index.js'; + +// Main classes +export { OpenAIContentGenerator } from './openaiContentGenerator.js'; +export { ContentGenerationPipeline, type PipelineConfig } from './pipeline.js'; + +// Providers +export { + type OpenAICompatibleProvider, + DashScopeOpenAICompatibleProvider, + OpenRouterOpenAICompatibleProvider, +} from './provider/index.js'; + +// Utilities +export { Converter } from './converter.js'; +export { StreamingManager } from './streamingManager.js'; + +// Factory utility functions +/** + * Create an OpenAI-compatible content generator with the appropriate provider + */ +export function createContentGenerator( + contentGeneratorConfig: ContentGeneratorConfig, + cliConfig: Config, +): ContentGenerator { + const provider = determineProvider(contentGeneratorConfig, cliConfig); + return new OpenAIContentGenerator( + contentGeneratorConfig, + cliConfig, + provider, + ); +} + +/** + * Determine the appropriate provider based on configuration + */ +export function determineProvider( + contentGeneratorConfig: ContentGeneratorConfig, + cliConfig: Config, +): OpenAICompatibleProvider { + const config = + contentGeneratorConfig || cliConfig.getContentGeneratorConfig(); + + // Check for DashScope provider + if (DashScopeOpenAICompatibleProvider.isDashScopeProvider(config)) { + return new DashScopeOpenAICompatibleProvider( + contentGeneratorConfig, + cliConfig, + ); + } + + // Check for OpenRouter provider + if (OpenRouterOpenAICompatibleProvider.isOpenRouterProvider(config)) { + return new OpenRouterOpenAICompatibleProvider( + contentGeneratorConfig, + cliConfig, + ); + } + + // Default provider for standard OpenAI-compatible APIs + return new DefaultOpenAICompatibleProvider(contentGeneratorConfig, cliConfig); +} + +// Services +export { + type TelemetryService, + type RequestContext, + DefaultTelemetryService, +} from './telemetryService.js'; + +export { type ErrorHandler, EnhancedErrorHandler } from './errorHandler.js'; diff --git a/packages/core/src/core/refactor/openaiContentGenerator.ts b/packages/core/src/core/refactor/openaiContentGenerator.ts new file mode 100644 index 00000000..ed0cdc38 --- /dev/null +++ b/packages/core/src/core/refactor/openaiContentGenerator.ts @@ -0,0 +1,81 @@ +import { ContentGenerator } from '../contentGenerator.js'; +import { Config } from '../../config/config.js'; +import { type OpenAICompatibleProvider } from './provider/index.js'; +import { + CountTokensParameters, + CountTokensResponse, + EmbedContentParameters, + EmbedContentResponse, + GenerateContentParameters, + GenerateContentResponse, +} from '@google/genai'; +import { ContentGenerationPipeline, PipelineConfig } from './pipeline.js'; +import { DefaultTelemetryService } from './telemetryService.js'; +import { EnhancedErrorHandler } from './errorHandler.js'; +import { ContentGeneratorConfig } from '../contentGenerator.js'; + +export class OpenAIContentGenerator implements ContentGenerator { + protected pipeline: ContentGenerationPipeline; + + constructor( + contentGeneratorConfig: ContentGeneratorConfig, + cliConfig: Config, + provider: OpenAICompatibleProvider, + ) { + // Create pipeline configuration + const pipelineConfig: PipelineConfig = { + cliConfig, + provider, + contentGeneratorConfig, + telemetryService: new DefaultTelemetryService( + cliConfig, + contentGeneratorConfig.enableOpenAILogging, + ), + errorHandler: new EnhancedErrorHandler( + (error: unknown, request: GenerateContentParameters) => + this.shouldSuppressErrorLogging(error, request), + ), + }; + + this.pipeline = new ContentGenerationPipeline(pipelineConfig); + } + + /** + * Hook for subclasses to customize error handling behavior + * @param error The error that occurred + * @param request The original request + * @returns true if error logging should be suppressed, false otherwise + */ + protected shouldSuppressErrorLogging( + _error: unknown, + _request: GenerateContentParameters, + ): boolean { + return false; // Default behavior: never suppress error logging + } + + async generateContent( + request: GenerateContentParameters, + userPromptId: string, + ): Promise { + return this.pipeline.execute(request, userPromptId); + } + + async generateContentStream( + request: GenerateContentParameters, + userPromptId: string, + ): Promise> { + return this.pipeline.executeStream(request, userPromptId); + } + + async countTokens( + request: CountTokensParameters, + ): Promise { + return this.pipeline.countTokens(request); + } + + async embedContent( + request: EmbedContentParameters, + ): Promise { + return this.pipeline.embedContent(request); + } +} diff --git a/packages/core/src/core/refactor/pipeline.ts b/packages/core/src/core/refactor/pipeline.ts new file mode 100644 index 00000000..11ffa851 --- /dev/null +++ b/packages/core/src/core/refactor/pipeline.ts @@ -0,0 +1,380 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import OpenAI from 'openai'; +import { + GenerateContentParameters, + GenerateContentResponse, + CountTokensParameters, + CountTokensResponse, + EmbedContentParameters, + EmbedContentResponse, +} from '@google/genai'; +import { Config } from '../../config/config.js'; +import { ContentGeneratorConfig } from '../contentGenerator.js'; +import { type OpenAICompatibleProvider } from './provider/index.js'; +import { Converter } from './converter.js'; +import { TelemetryService, RequestContext } from './telemetryService.js'; +import { ErrorHandler } from './errorHandler.js'; +import { StreamingManager } from './streamingManager.js'; + +export interface PipelineConfig { + cliConfig: Config; + provider: OpenAICompatibleProvider; + contentGeneratorConfig: ContentGeneratorConfig; + telemetryService: TelemetryService; + errorHandler: ErrorHandler; +} + +export class ContentGenerationPipeline { + client: OpenAI; + private converter: Converter; + private streamingManager: StreamingManager; + private contentGeneratorConfig: ContentGeneratorConfig; + + constructor(private config: PipelineConfig) { + this.contentGeneratorConfig = config.contentGeneratorConfig; + this.client = this.config.provider.buildClient(); + this.converter = new Converter(this.contentGeneratorConfig.model); + this.streamingManager = new StreamingManager(this.converter); + } + + async execute( + request: GenerateContentParameters, + userPromptId: string, + ): Promise { + return this.executeWithErrorHandling( + request, + userPromptId, + false, + async (openaiRequest, context) => { + const openaiResponse = (await this.client.chat.completions.create( + openaiRequest, + )) as OpenAI.Chat.ChatCompletion; + + const geminiResponse = + this.converter.convertOpenAIResponseToGemini(openaiResponse); + + // Log success + await this.config.telemetryService.logSuccess( + context, + geminiResponse, + openaiRequest, + openaiResponse, + ); + + return geminiResponse; + }, + ); + } + + async executeStream( + request: GenerateContentParameters, + userPromptId: string, + ): Promise> { + return this.executeWithErrorHandling( + request, + userPromptId, + true, + async (openaiRequest, context) => { + const stream = (await this.client.chat.completions.create( + openaiRequest, + )) as AsyncIterable; + + const originalStream = this.streamingManager.processStream(stream); + + // Create a logging stream decorator that handles collection and logging + return this.createLoggingStream( + originalStream, + context, + openaiRequest, + request, + ); + }, + ); + } + + async countTokens( + request: CountTokensParameters, + ): Promise { + // Use tiktoken for accurate token counting + const content = JSON.stringify(request.contents); + let totalTokens = 0; + + try { + const { get_encoding } = await import('tiktoken'); + const encoding = get_encoding('cl100k_base'); // GPT-4 encoding, but estimate for qwen + totalTokens = encoding.encode(content).length; + encoding.free(); + } catch (error) { + console.warn( + 'Failed to load tiktoken, falling back to character approximation:', + error, + ); + // Fallback: rough approximation using character count + totalTokens = Math.ceil(content.length / 4); // Rough estimate: 1 token ≈ 4 characters + } + + return { + totalTokens, + }; + } + + async embedContent( + request: EmbedContentParameters, + ): Promise { + // Extract text from contents + let text = ''; + if (Array.isArray(request.contents)) { + text = request.contents + .map((content) => { + if (typeof content === 'string') return content; + if ('parts' in content && content.parts) { + return content.parts + .map((part) => + typeof part === 'string' + ? part + : 'text' in part + ? (part as { text?: string }).text || '' + : '', + ) + .join(' '); + } + return ''; + }) + .join(' '); + } else if (request.contents) { + if (typeof request.contents === 'string') { + text = request.contents; + } else if ('parts' in request.contents && request.contents.parts) { + text = request.contents.parts + .map((part) => + typeof part === 'string' ? part : 'text' in part ? part.text : '', + ) + .join(' '); + } + } + + try { + const embedding = await this.client.embeddings.create({ + model: 'text-embedding-ada-002', // Default embedding model + input: text, + }); + + return { + embeddings: [ + { + values: embedding.data[0].embedding, + }, + ], + }; + } catch (error) { + console.error('OpenAI API Embedding Error:', error); + throw new Error( + `OpenAI API error: ${error instanceof Error ? error.message : String(error)}`, + ); + } + } + + private async buildRequest( + request: GenerateContentParameters, + userPromptId: string, + streaming: boolean = false, + ): Promise { + const messages = this.converter.convertGeminiRequestToOpenAI(request); + + // Apply provider-specific enhancements + const baseRequest: OpenAI.Chat.ChatCompletionCreateParams = { + model: this.contentGeneratorConfig.model, + messages, + ...this.buildSamplingParameters(request), + }; + + // Let provider enhance the request (e.g., add metadata, cache control) + const enhancedRequest = this.config.provider.buildRequest( + baseRequest, + userPromptId, + ); + + // Add tools if present + if (request.config?.tools) { + enhancedRequest.tools = await this.converter.convertGeminiToolsToOpenAI( + request.config.tools, + ); + } + + // Add streaming options if needed + if (streaming) { + enhancedRequest.stream = true; + enhancedRequest.stream_options = { include_usage: true }; + } + + return enhancedRequest; + } + + private buildSamplingParameters( + request: GenerateContentParameters, + ): Record { + const configSamplingParams = this.contentGeneratorConfig.samplingParams; + + // Helper function to get parameter value with priority: config > request > default + const getParameterValue = ( + configKey: keyof NonNullable, + requestKey: keyof NonNullable, + defaultValue?: T, + ): T | undefined => { + const configValue = configSamplingParams?.[configKey] as T | undefined; + const requestValue = request.config?.[requestKey] as T | undefined; + + if (configValue !== undefined) return configValue; + if (requestValue !== undefined) return requestValue; + return defaultValue; + }; + + // Helper function to conditionally add parameter if it has a value + const addParameterIfDefined = ( + key: string, + configKey: keyof NonNullable, + requestKey?: keyof NonNullable, + defaultValue?: T, + ): Record | Record => { + const value = requestKey + ? getParameterValue(configKey, requestKey, defaultValue) + : ((configSamplingParams?.[configKey] as T | undefined) ?? + defaultValue); + + return value !== undefined ? { [key]: value } : {}; + }; + + const params = { + // Parameters with request fallback and defaults + temperature: getParameterValue('temperature', 'temperature', 0.0), + top_p: getParameterValue('top_p', 'topP', 1.0), + + // Max tokens (special case: different property names) + ...addParameterIfDefined('max_tokens', 'max_tokens', 'maxOutputTokens'), + + // Config-only parameters (no request fallback) + ...addParameterIfDefined('top_k', 'top_k'), + ...addParameterIfDefined('repetition_penalty', 'repetition_penalty'), + ...addParameterIfDefined('presence_penalty', 'presence_penalty'), + ...addParameterIfDefined('frequency_penalty', 'frequency_penalty'), + }; + + return params; + } + + /** + * Creates a stream decorator that collects responses and handles logging + */ + private async *createLoggingStream( + originalStream: AsyncGenerator, + context: RequestContext, + openaiRequest: OpenAI.Chat.ChatCompletionCreateParams, + request: GenerateContentParameters, + ): AsyncGenerator { + const responses: GenerateContentResponse[] = []; + + try { + // Yield all responses while collecting them + for await (const response of originalStream) { + responses.push(response); + yield response; + } + + // Stream completed successfully - perform logging + context.duration = Date.now() - context.startTime; + + const combinedResponse = + this.streamingManager.combineStreamResponsesForLogging( + responses, + this.contentGeneratorConfig.model, + ); + const openaiResponse = + this.converter.convertGeminiResponseToOpenAI(combinedResponse); + + await this.config.telemetryService.logStreamingSuccess( + context, + responses, + openaiRequest, + openaiResponse, + ); + } catch (error) { + // Stream failed - handle error and logging + context.duration = Date.now() - context.startTime; + + await this.config.telemetryService.logError( + context, + error, + openaiRequest, + ); + + this.config.errorHandler.handle(error, context, request); + } + } + + /** + * Common error handling wrapper for execute methods + */ + private async executeWithErrorHandling( + request: GenerateContentParameters, + userPromptId: string, + isStreaming: boolean, + executor: ( + openaiRequest: OpenAI.Chat.ChatCompletionCreateParams, + context: RequestContext, + ) => Promise, + ): Promise { + const context = this.createRequestContext(userPromptId, isStreaming); + + try { + const openaiRequest = await this.buildRequest( + request, + userPromptId, + isStreaming, + ); + + const result = await executor(openaiRequest, context); + + context.duration = Date.now() - context.startTime; + return result; + } catch (error) { + context.duration = Date.now() - context.startTime; + + // Log error + const openaiRequest = await this.buildRequest( + request, + userPromptId, + isStreaming, + ); + await this.config.telemetryService.logError( + context, + error, + openaiRequest, + ); + + // Handle and throw enhanced error + this.config.errorHandler.handle(error, context, request); + } + } + + /** + * Create request context with common properties + */ + private createRequestContext( + userPromptId: string, + isStreaming: boolean, + ): RequestContext { + return { + userPromptId, + model: this.contentGeneratorConfig.model, + authType: this.contentGeneratorConfig.authType || 'unknown', + startTime: Date.now(), + duration: 0, + isStreaming, + }; + } +} diff --git a/packages/core/src/core/refactor/provider/README.md b/packages/core/src/core/refactor/provider/README.md new file mode 100644 index 00000000..4c81b793 --- /dev/null +++ b/packages/core/src/core/refactor/provider/README.md @@ -0,0 +1,61 @@ +# Provider Structure + +This folder contains the different provider implementations for the Qwen Code refactor system. + +## File Structure + +- `constants.ts` - Common constants used across all providers +- `types.ts` - Type definitions and interfaces for providers +- `default.ts` - Default provider for standard OpenAI-compatible APIs +- `dashscope.ts` - DashScope (Qwen) specific provider implementation +- `openrouter.ts` - OpenRouter specific provider implementation +- `index.ts` - Main export file for all providers + +## Provider Types + +### Default Provider + +The `DefaultOpenAICompatibleProvider` is the fallback provider for standard OpenAI-compatible APIs. It provides basic functionality without special enhancements and passes through all request parameters. + +### DashScope Provider + +The `DashScopeOpenAICompatibleProvider` handles DashScope (Qwen) specific features like cache control and metadata. + +### OpenRouter Provider + +The `OpenRouterOpenAICompatibleProvider` handles OpenRouter specific headers and configurations. + +## Adding a New Provider + +To add a new provider: + +1. Create a new file (e.g., `newprovider.ts`) in this folder +2. Implement the `OpenAICompatibleProvider` interface +3. Add a static method to identify if a config belongs to this provider +4. Export the class from `index.ts` +5. The main `provider.ts` file will automatically re-export it + +## Provider Interface + +All providers must implement: + +- `buildHeaders()` - Build HTTP headers for the provider +- `buildClient()` - Create and configure the OpenAI client +- `buildRequest()` - Transform requests before sending to the provider + +## Example + +```typescript +export class NewProviderOpenAICompatibleProvider + implements OpenAICompatibleProvider +{ + // Implementation... + + static isNewProviderProvider( + contentGeneratorConfig: ContentGeneratorConfig, + ): boolean { + // Logic to identify this provider + return true; + } +} +``` diff --git a/packages/core/src/core/refactor/provider/dashscope.ts b/packages/core/src/core/refactor/provider/dashscope.ts new file mode 100644 index 00000000..acd9d770 --- /dev/null +++ b/packages/core/src/core/refactor/provider/dashscope.ts @@ -0,0 +1,210 @@ +import OpenAI from 'openai'; +import { Config } from '../../../config/config.js'; +import { AuthType, ContentGeneratorConfig } from '../../contentGenerator.js'; +import { DEFAULT_TIMEOUT, DEFAULT_MAX_RETRIES } from '../constants.js'; +import { + OpenAICompatibleProvider, + DashScopeRequestMetadata, + ChatCompletionContentPartTextWithCache, + ChatCompletionContentPartWithCache, +} from './types.js'; + +export class DashScopeOpenAICompatibleProvider + implements OpenAICompatibleProvider +{ + private contentGeneratorConfig: ContentGeneratorConfig; + private cliConfig: Config; + + constructor( + contentGeneratorConfig: ContentGeneratorConfig, + cliConfig: Config, + ) { + this.cliConfig = cliConfig; + this.contentGeneratorConfig = contentGeneratorConfig; + } + + static isDashScopeProvider( + contentGeneratorConfig: ContentGeneratorConfig, + ): boolean { + const authType = contentGeneratorConfig.authType; + const baseUrl = contentGeneratorConfig.baseUrl; + return ( + authType === AuthType.QWEN_OAUTH || + baseUrl === 'https://dashscope.aliyuncs.com/compatible-mode/v1' || + baseUrl === 'https://dashscope-intl.aliyuncs.com/compatible-mode/v1' + ); + } + + buildHeaders(): Record { + const version = this.cliConfig.getCliVersion() || 'unknown'; + const userAgent = `QwenCode/${version} (${process.platform}; ${process.arch})`; + const { authType } = this.contentGeneratorConfig; + return { + 'User-Agent': userAgent, + 'X-DashScope-CacheControl': 'enable', + 'X-DashScope-UserAgent': userAgent, + 'X-DashScope-AuthType': authType, + }; + } + + buildClient(): OpenAI { + const { + apiKey, + baseUrl, + timeout = DEFAULT_TIMEOUT, + maxRetries = DEFAULT_MAX_RETRIES, + } = this.contentGeneratorConfig; + const defaultHeaders = this.buildHeaders(); + return new OpenAI({ + apiKey, + baseURL: baseUrl, + timeout, + maxRetries, + defaultHeaders, + }); + } + + buildRequest( + request: OpenAI.Chat.ChatCompletionCreateParams, + userPromptId: string, + ): OpenAI.Chat.ChatCompletionCreateParams { + let messages = request.messages; + + // Apply DashScope cache control only if not disabled + if (!this.shouldDisableCacheControl()) { + // Add cache control to system and last messages for DashScope providers + // Only add cache control to system message for non-streaming requests + const cacheTarget = request.stream ? 'both' : 'system'; + messages = this.addDashScopeCacheControl(messages, cacheTarget); + } + + return { + ...request, // Preserve all original parameters including sampling params + messages, + ...(this.buildMetadata(userPromptId) || {}), + }; + } + + buildMetadata(userPromptId: string): DashScopeRequestMetadata { + return { + metadata: { + sessionId: this.cliConfig.getSessionId?.(), + promptId: userPromptId, + }, + }; + } + + /** + * Add cache control flag to specified message(s) for DashScope providers + */ + private addDashScopeCacheControl( + messages: OpenAI.Chat.ChatCompletionMessageParam[], + target: 'system' | 'last' | 'both' = 'both', + ): OpenAI.Chat.ChatCompletionMessageParam[] { + if (messages.length === 0) { + return messages; + } + + let updatedMessages = [...messages]; + + // Add cache control to system message if requested + if (target === 'system' || target === 'both') { + updatedMessages = this.addCacheControlToMessage( + updatedMessages, + 'system', + ); + } + + // Add cache control to last message if requested + if (target === 'last' || target === 'both') { + updatedMessages = this.addCacheControlToMessage(updatedMessages, 'last'); + } + + return updatedMessages; + } + + /** + * Helper method to add cache control to a specific message + */ + private addCacheControlToMessage( + messages: OpenAI.Chat.ChatCompletionMessageParam[], + target: 'system' | 'last', + ): OpenAI.Chat.ChatCompletionMessageParam[] { + const updatedMessages = [...messages]; + let messageIndex: number; + + if (target === 'system') { + // Find the first system message + messageIndex = messages.findIndex((msg) => msg.role === 'system'); + if (messageIndex === -1) { + return updatedMessages; + } + } else { + // Get the last message + messageIndex = messages.length - 1; + } + + const message = updatedMessages[messageIndex]; + + // Only process messages that have content + if ('content' in message && message.content !== null) { + if (typeof message.content === 'string') { + // Convert string content to array format with cache control + const messageWithArrayContent = { + ...message, + content: [ + { + type: 'text', + text: message.content, + cache_control: { type: 'ephemeral' }, + } as ChatCompletionContentPartTextWithCache, + ], + }; + updatedMessages[messageIndex] = + messageWithArrayContent as OpenAI.Chat.ChatCompletionMessageParam; + } else if (Array.isArray(message.content)) { + // If content is already an array, add cache_control to the last item + const contentArray = [ + ...message.content, + ] as ChatCompletionContentPartWithCache[]; + if (contentArray.length > 0) { + const lastItem = contentArray[contentArray.length - 1]; + if (lastItem.type === 'text') { + // Add cache_control to the last text item + contentArray[contentArray.length - 1] = { + ...lastItem, + cache_control: { type: 'ephemeral' }, + } as ChatCompletionContentPartTextWithCache; + } else { + // If the last item is not text, add a new text item with cache_control + contentArray.push({ + type: 'text', + text: '', + cache_control: { type: 'ephemeral' }, + } as ChatCompletionContentPartTextWithCache); + } + + const messageWithCache = { + ...message, + content: contentArray, + }; + updatedMessages[messageIndex] = + messageWithCache as OpenAI.Chat.ChatCompletionMessageParam; + } + } + } + + return updatedMessages; + } + + /** + * Check if cache control should be disabled based on configuration. + * + * @returns true if cache control should be disabled, false otherwise + */ + private shouldDisableCacheControl(): boolean { + return ( + this.cliConfig.getContentGeneratorConfig()?.disableCacheControl === true + ); + } +} diff --git a/packages/core/src/core/refactor/provider/default.ts b/packages/core/src/core/refactor/provider/default.ts new file mode 100644 index 00000000..64e64a27 --- /dev/null +++ b/packages/core/src/core/refactor/provider/default.ts @@ -0,0 +1,58 @@ +import OpenAI from 'openai'; +import { Config } from '../../../config/config.js'; +import { ContentGeneratorConfig } from '../../contentGenerator.js'; +import { DEFAULT_TIMEOUT, DEFAULT_MAX_RETRIES } from '../constants.js'; +import { OpenAICompatibleProvider } from './types.js'; + +/** + * Default provider for standard OpenAI-compatible APIs + */ +export class DefaultOpenAICompatibleProvider + implements OpenAICompatibleProvider +{ + private contentGeneratorConfig: ContentGeneratorConfig; + private cliConfig: Config; + + constructor( + contentGeneratorConfig: ContentGeneratorConfig, + cliConfig: Config, + ) { + this.cliConfig = cliConfig; + this.contentGeneratorConfig = contentGeneratorConfig; + } + + buildHeaders(): Record { + const version = this.cliConfig.getCliVersion() || 'unknown'; + const userAgent = `QwenCode/${version} (${process.platform}; ${process.arch})`; + return { + 'User-Agent': userAgent, + }; + } + + buildClient(): OpenAI { + const { + apiKey, + baseUrl, + timeout = DEFAULT_TIMEOUT, + maxRetries = DEFAULT_MAX_RETRIES, + } = this.contentGeneratorConfig; + const defaultHeaders = this.buildHeaders(); + return new OpenAI({ + apiKey, + baseURL: baseUrl, + timeout, + maxRetries, + defaultHeaders, + }); + } + + buildRequest( + request: OpenAI.Chat.ChatCompletionCreateParams, + _userPromptId: string, + ): OpenAI.Chat.ChatCompletionCreateParams { + // Default provider doesn't need special enhancements, just pass through all parameters + return { + ...request, // Preserve all original parameters including sampling params + }; + } +} diff --git a/packages/core/src/core/refactor/provider/index.ts b/packages/core/src/core/refactor/provider/index.ts new file mode 100644 index 00000000..8fdf56e2 --- /dev/null +++ b/packages/core/src/core/refactor/provider/index.ts @@ -0,0 +1,9 @@ +export { DashScopeOpenAICompatibleProvider } from './dashscope.js'; +export { OpenRouterOpenAICompatibleProvider } from './openrouter.js'; +export { DefaultOpenAICompatibleProvider } from './default.js'; +export type { + OpenAICompatibleProvider, + DashScopeRequestMetadata, + ChatCompletionContentPartTextWithCache, + ChatCompletionContentPartWithCache, +} from './types.js'; diff --git a/packages/core/src/core/refactor/provider/openrouter.ts b/packages/core/src/core/refactor/provider/openrouter.ts new file mode 100644 index 00000000..2275b147 --- /dev/null +++ b/packages/core/src/core/refactor/provider/openrouter.ts @@ -0,0 +1,64 @@ +import OpenAI from 'openai'; +import { Config } from '../../../config/config.js'; +import { ContentGeneratorConfig } from '../../contentGenerator.js'; +import { DEFAULT_TIMEOUT, DEFAULT_MAX_RETRIES } from '../constants.js'; +import { OpenAICompatibleProvider } from './types.js'; + +export class OpenRouterOpenAICompatibleProvider + implements OpenAICompatibleProvider +{ + private contentGeneratorConfig: ContentGeneratorConfig; + private cliConfig: Config; + + constructor( + contentGeneratorConfig: ContentGeneratorConfig, + cliConfig: Config, + ) { + this.cliConfig = cliConfig; + this.contentGeneratorConfig = contentGeneratorConfig; + } + + static isOpenRouterProvider( + contentGeneratorConfig: ContentGeneratorConfig, + ): boolean { + const baseURL = contentGeneratorConfig.baseUrl || ''; + return baseURL.includes('openrouter.ai'); + } + + buildHeaders(): Record { + const version = this.cliConfig.getCliVersion() || 'unknown'; + const userAgent = `QwenCode/${version} (${process.platform}; ${process.arch})`; + return { + 'User-Agent': userAgent, + 'HTTP-Referer': 'https://github.com/QwenLM/qwen-code.git', + 'X-Title': 'Qwen Code', + }; + } + + buildClient(): OpenAI { + const { + apiKey, + baseUrl, + timeout = DEFAULT_TIMEOUT, + maxRetries = DEFAULT_MAX_RETRIES, + } = this.contentGeneratorConfig; + const defaultHeaders = this.buildHeaders(); + return new OpenAI({ + apiKey, + baseURL: baseUrl, + timeout, + maxRetries, + defaultHeaders, + }); + } + + buildRequest( + request: OpenAI.Chat.ChatCompletionCreateParams, + _userPromptId: string, + ): OpenAI.Chat.ChatCompletionCreateParams { + // OpenRouter doesn't need special enhancements, just pass through all parameters + return { + ...request, // Preserve all original parameters including sampling params + }; + } +} diff --git a/packages/core/src/core/refactor/provider/types.ts b/packages/core/src/core/refactor/provider/types.ts new file mode 100644 index 00000000..ea24f31d --- /dev/null +++ b/packages/core/src/core/refactor/provider/types.ts @@ -0,0 +1,28 @@ +import OpenAI from 'openai'; + +// Extended types to support cache_control for DashScope +export interface ChatCompletionContentPartTextWithCache + extends OpenAI.Chat.ChatCompletionContentPartText { + cache_control?: { type: 'ephemeral' }; +} + +export type ChatCompletionContentPartWithCache = + | ChatCompletionContentPartTextWithCache + | OpenAI.Chat.ChatCompletionContentPartImage + | OpenAI.Chat.ChatCompletionContentPartRefusal; + +export interface OpenAICompatibleProvider { + buildHeaders(): Record; + buildClient(): OpenAI; + buildRequest( + request: OpenAI.Chat.ChatCompletionCreateParams, + userPromptId: string, + ): OpenAI.Chat.ChatCompletionCreateParams; +} + +export type DashScopeRequestMetadata = { + metadata: { + sessionId?: string; + promptId: string; + }; +}; diff --git a/packages/core/src/core/refactor/streamingManager.ts b/packages/core/src/core/refactor/streamingManager.ts new file mode 100644 index 00000000..e82013ae --- /dev/null +++ b/packages/core/src/core/refactor/streamingManager.ts @@ -0,0 +1,111 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import OpenAI from 'openai'; +import { GenerateContentResponse, Part, FinishReason } from '@google/genai'; +import { Converter } from './converter.js'; + +export interface ToolCallAccumulator { + id?: string; + name?: string; + arguments: string; +} + +export class StreamingManager { + private toolCallAccumulator = new Map(); + + constructor(private converter: Converter) {} + + async *processStream( + stream: AsyncIterable, + ): AsyncGenerator { + // Reset the accumulator for each new stream + this.toolCallAccumulator.clear(); + + for await (const chunk of stream) { + const response = this.converter.convertOpenAIChunkToGemini(chunk); + + // Ignore empty responses, which would cause problems with downstream code + // that expects a valid response. + if ( + response.candidates?.[0]?.content?.parts?.length === 0 && + !response.usageMetadata + ) { + continue; + } + + yield response; + } + } + + /** + * Combine streaming responses for logging purposes + */ + combineStreamResponsesForLogging( + responses: GenerateContentResponse[], + model: string, + ): GenerateContentResponse { + if (responses.length === 0) { + return new GenerateContentResponse(); + } + + const lastResponse = responses[responses.length - 1]; + + // Find the last response with usage metadata + const finalUsageMetadata = responses + .slice() + .reverse() + .find((r) => r.usageMetadata)?.usageMetadata; + + // Combine all text content from the stream + const combinedParts: Part[] = []; + let combinedText = ''; + const functionCalls: Part[] = []; + + for (const response of responses) { + if (response.candidates?.[0]?.content?.parts) { + for (const part of response.candidates[0].content.parts) { + if ('text' in part && part.text) { + combinedText += part.text; + } else if ('functionCall' in part && part.functionCall) { + functionCalls.push(part); + } + } + } + } + + // Add combined text if any + if (combinedText) { + combinedParts.push({ text: combinedText }); + } + + // Add function calls + combinedParts.push(...functionCalls); + + // Create combined response + const combinedResponse = new GenerateContentResponse(); + combinedResponse.candidates = [ + { + content: { + parts: combinedParts, + role: 'model' as const, + }, + finishReason: + responses[responses.length - 1]?.candidates?.[0]?.finishReason || + FinishReason.FINISH_REASON_UNSPECIFIED, + index: 0, + safetyRatings: [], + }, + ]; + combinedResponse.responseId = lastResponse?.responseId; + combinedResponse.createTime = lastResponse?.createTime; + combinedResponse.modelVersion = model; + combinedResponse.promptFeedback = { safetyRatings: [] }; + combinedResponse.usageMetadata = finalUsageMetadata; + + return combinedResponse; + } +} diff --git a/packages/core/src/core/refactor/telemetryService.ts b/packages/core/src/core/refactor/telemetryService.ts new file mode 100644 index 00000000..3a677255 --- /dev/null +++ b/packages/core/src/core/refactor/telemetryService.ts @@ -0,0 +1,137 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import { Config } from '../../config/config.js'; +import { logApiError, logApiResponse } from '../../telemetry/loggers.js'; +import { ApiErrorEvent, ApiResponseEvent } from '../../telemetry/types.js'; +import { openaiLogger } from '../../utils/openaiLogger.js'; +import { GenerateContentResponse } from '@google/genai'; +import OpenAI from 'openai'; + +export interface RequestContext { + userPromptId: string; + model: string; + authType: string; + startTime: number; + duration: number; + isStreaming: boolean; +} + +export interface TelemetryService { + logSuccess( + context: RequestContext, + response: GenerateContentResponse, + openaiRequest?: OpenAI.Chat.ChatCompletionCreateParams, + openaiResponse?: OpenAI.Chat.ChatCompletion, + ): Promise; + + logError( + context: RequestContext, + error: unknown, + openaiRequest?: OpenAI.Chat.ChatCompletionCreateParams, + ): Promise; + + logStreamingSuccess( + context: RequestContext, + responses: GenerateContentResponse[], + openaiRequest?: OpenAI.Chat.ChatCompletionCreateParams, + openaiResponse?: OpenAI.Chat.ChatCompletion, + ): Promise; +} + +export class DefaultTelemetryService implements TelemetryService { + constructor( + private config: Config, + private enableOpenAILogging: boolean = false, + ) {} + + async logSuccess( + context: RequestContext, + response: GenerateContentResponse, + openaiRequest?: OpenAI.Chat.ChatCompletionCreateParams, + openaiResponse?: OpenAI.Chat.ChatCompletion, + ): Promise { + // Log API response event for UI telemetry + const responseEvent = new ApiResponseEvent( + response.responseId || 'unknown', + context.model, + context.duration, + context.userPromptId, + context.authType, + response.usageMetadata, + ); + + logApiResponse(this.config, responseEvent); + + // Log interaction if enabled + if (this.enableOpenAILogging && openaiRequest && openaiResponse) { + await openaiLogger.logInteraction(openaiRequest, openaiResponse); + } + } + + async logError( + context: RequestContext, + error: unknown, + openaiRequest?: OpenAI.Chat.ChatCompletionCreateParams, + ): Promise { + const errorMessage = error instanceof Error ? error.message : String(error); + + // Log API error event for UI telemetry + const errorEvent = new ApiErrorEvent( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (error as any).requestID || 'unknown', + context.model, + errorMessage, + context.duration, + context.userPromptId, + context.authType, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (error as any).type, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (error as any).code, + ); + logApiError(this.config, errorEvent); + + // Log error interaction if enabled + if (this.enableOpenAILogging && openaiRequest) { + await openaiLogger.logInteraction( + openaiRequest, + undefined, + error as Error, + ); + } + } + + async logStreamingSuccess( + context: RequestContext, + responses: GenerateContentResponse[], + openaiRequest?: OpenAI.Chat.ChatCompletionCreateParams, + openaiResponse?: OpenAI.Chat.ChatCompletion, + ): Promise { + // Get final usage metadata from the last response that has it + const finalUsageMetadata = responses + .slice() + .reverse() + .find((r) => r.usageMetadata)?.usageMetadata; + + // Log API response event for UI telemetry + const responseEvent = new ApiResponseEvent( + responses[responses.length - 1]?.responseId || 'unknown', + context.model, + context.duration, + context.userPromptId, + context.authType, + finalUsageMetadata, + ); + + logApiResponse(this.config, responseEvent); + + // Log interaction if enabled + if (this.enableOpenAILogging && openaiRequest && openaiResponse) { + await openaiLogger.logInteraction(openaiRequest, openaiResponse); + } + } +} diff --git a/packages/core/src/qwen/qwenContentGenerator.test.ts b/packages/core/src/qwen/qwenContentGenerator.test.ts index e8dfd3c3..f5f1912c 100644 --- a/packages/core/src/qwen/qwenContentGenerator.test.ts +++ b/packages/core/src/qwen/qwenContentGenerator.test.ts @@ -22,7 +22,7 @@ import { import { QwenContentGenerator } from './qwenContentGenerator.js'; import { SharedTokenManager } from './sharedTokenManager.js'; import { Config } from '../config/config.js'; -import { AuthType, ContentGeneratorConfig } from '../core/contentGenerator.js'; +import { AuthType } from '../core/contentGenerator.js'; // Mock SharedTokenManager vi.mock('./sharedTokenManager.js', () => ({ @@ -132,20 +132,21 @@ vi.mock('./sharedTokenManager.js', () => ({ })); // Mock the OpenAIContentGenerator parent class -vi.mock('../core/openaiContentGenerator.js', () => ({ +vi.mock('../core/refactor/openaiContentGenerator.js', () => ({ OpenAIContentGenerator: class { - client: { - apiKey: string; - baseURL: string; + pipeline: { + client: { + apiKey: string; + baseURL: string; + }; }; - constructor( - contentGeneratorConfig: ContentGeneratorConfig, - _config: Config, - ) { - this.client = { - apiKey: contentGeneratorConfig.apiKey || 'test-key', - baseURL: contentGeneratorConfig.baseUrl || 'https://api.openai.com/v1', + constructor(_config: Config, _provider: unknown) { + this.pipeline = { + client: { + apiKey: 'test-key', + baseURL: 'https://api.openai.com/v1', + }, }; } @@ -220,7 +221,10 @@ describe('QwenContentGenerator', () => { // Mock Config mockConfig = { getContentGeneratorConfig: vi.fn().mockReturnValue({ + model: 'qwen-turbo', + apiKey: 'test-api-key', authType: 'qwen', + baseUrl: 'https://dashscope.aliyuncs.com/compatible-mode/v1', enableOpenAILogging: false, timeout: 120000, maxRetries: 3, @@ -230,6 +234,9 @@ describe('QwenContentGenerator', () => { top_p: 0.9, }, }), + getCliVersion: vi.fn().mockReturnValue('1.0.0'), + getSessionId: vi.fn().mockReturnValue('test-session-id'), + getUsageStatisticsEnabled: vi.fn().mockReturnValue(false), } as unknown as Config; // Mock QwenOAuth2Client @@ -245,7 +252,11 @@ describe('QwenContentGenerator', () => { // Create QwenContentGenerator instance const contentGeneratorConfig = { model: 'qwen-turbo', + apiKey: 'test-api-key', authType: AuthType.QWEN_OAUTH, + baseUrl: 'https://dashscope.aliyuncs.com/compatible-mode/v1', + timeout: 120000, + maxRetries: 3, }; qwenContentGenerator = new QwenContentGenerator( mockQwenClient, diff --git a/packages/core/src/qwen/qwenContentGenerator.ts b/packages/core/src/qwen/qwenContentGenerator.ts index 9ef89497..883b6718 100644 --- a/packages/core/src/qwen/qwenContentGenerator.ts +++ b/packages/core/src/qwen/qwenContentGenerator.ts @@ -4,7 +4,8 @@ * SPDX-License-Identifier: Apache-2.0 */ -import { OpenAIContentGenerator } from '../core/openaiContentGenerator.js'; +import { OpenAIContentGenerator } from '../core/refactor/openaiContentGenerator.js'; +import { DashScopeOpenAICompatibleProvider } from '../core/refactor/provider/dashscope.js'; import { IQwenOAuth2Client } from './qwenOAuth2.js'; import { SharedTokenManager } from './sharedTokenManager.js'; import { Config } from '../config/config.js'; @@ -33,15 +34,24 @@ export class QwenContentGenerator extends OpenAIContentGenerator { constructor( qwenClient: IQwenOAuth2Client, contentGeneratorConfig: ContentGeneratorConfig, - config: Config, + cliConfig: Config, ) { - // Initialize with empty API key, we'll override it dynamically - super(contentGeneratorConfig, config); + // Create DashScope provider for Qwen + const dashscopeProvider = new DashScopeOpenAICompatibleProvider( + contentGeneratorConfig, + cliConfig, + ); + + // Initialize with DashScope provider + super(contentGeneratorConfig, cliConfig, dashscopeProvider); this.qwenClient = qwenClient; this.sharedManager = SharedTokenManager.getInstance(); // Set default base URL, will be updated dynamically - this.client.baseURL = DEFAULT_QWEN_BASE_URL; + if (contentGeneratorConfig?.baseUrl && contentGeneratorConfig?.apiKey) { + this.pipeline.client.baseURL = contentGeneratorConfig?.baseUrl; + this.pipeline.client.apiKey = contentGeneratorConfig?.apiKey; + } } /** @@ -106,46 +116,24 @@ export class QwenContentGenerator extends OpenAIContentGenerator { * Execute an operation with automatic credential management and retry logic. * This method handles: * - Dynamic token and endpoint retrieval - * - Temporary client configuration updates - * - Automatic restoration of original configuration + * - Client configuration updates * - Retry logic on authentication errors with token refresh * * @param operation - The operation to execute with updated client configuration - * @param restoreOnCompletion - Whether to restore original config after operation completes * @returns The result of the operation */ private async executeWithCredentialManagement( operation: () => Promise, - restoreOnCompletion: boolean = true, ): Promise { // Attempt the operation with credential management and retry logic const attemptOperation = async (): Promise => { const { token, endpoint } = await this.getValidToken(); - // Store original configuration - const originalApiKey = this.client.apiKey; - const originalBaseURL = this.client.baseURL; - // Apply dynamic configuration - this.client.apiKey = token; - this.client.baseURL = endpoint; + this.pipeline.client.apiKey = token; + this.pipeline.client.baseURL = endpoint; - try { - const result = await operation(); - - // For streaming operations, we may need to keep the configuration active - if (restoreOnCompletion) { - this.client.apiKey = originalApiKey; - this.client.baseURL = originalBaseURL; - } - - return result; - } catch (error) { - // Always restore on error - this.client.apiKey = originalApiKey; - this.client.baseURL = originalBaseURL; - throw error; - } + return await operation(); }; // Execute with retry logic for auth errors @@ -182,17 +170,14 @@ export class QwenContentGenerator extends OpenAIContentGenerator { } /** - * Override to use dynamic token and endpoint with automatic retry. - * Note: For streaming, the client configuration is not restored immediately - * since the generator may continue to be used after this method returns. + * Override to use dynamic token and endpoint with automatic retry */ override async generateContentStream( request: GenerateContentParameters, userPromptId: string, ): Promise> { - return this.executeWithCredentialManagement( - () => super.generateContentStream(request, userPromptId), - false, // Don't restore immediately for streaming + return this.executeWithCredentialManagement(() => + super.generateContentStream(request, userPromptId), ); }