Update server.js

server.js

@@ -22,16 +22,8 @@ if (falKeys.length === 0) {
 
 console.log(`Loaded ${falKeys.length} Fal AI Keys.`);
 
-let currentFalKeyIndex = 0;
-
-// 获取下一个 Fal Key 并循环
-function getNextFalKey() {
-    const key = falKeys[currentFalKeyIndex];
-    const usedIndex = currentFalKeyIndex; // 记录本次使用的索引，用于日志
-    currentFalKeyIndex = (currentFalKeyIndex + 1) % falKeys.length; // 移动到下一个，如果到末尾则回到开头
-    console.log(`Using Fal Key at index: ${usedIndex}`);
-    return key;
-}
+let currentFalKeyIndex = 0; // Index of the *next* key to try for a *new* request
+
 // --- End Key Management ---
 
 
@@ -40,11 +32,6 @@ if (!API_KEY) {
     process.exit(1);
 }
 
-// 注意：不再在这里全局配置 fal.config
-// fal.config({
-//     credentials: FAL_KEY, // 移除这行
-// });
-
 const app = express();
 app.use(express.json({ limit: '50mb' }));
 app.use(express.urlencoded({ extended: true, limit: '50mb' }));
@@ -223,23 +210,29 @@ function convertMessagesToFalPrompt(messages) {
 
     let final_system_prompt = "";
 
+    // 检查各部分是否有内容 (使用 trim 后的固定部分)
     const hasFixedSystem = fixed_system_prompt_content.length > 0;
     const hasSystemHistory = system_prompt_history_content.length > 0;
 
     if (hasFixedSystem && hasSystemHistory) {
+        // 两部分都有，用分隔符连接
         final_system_prompt = fixed_system_prompt_content + SEPARATOR + system_prompt_history_content;
         console.log("Combining fixed system prompt and history with separator.");
     } else if (hasFixedSystem) {
+        // 只有固定部分
         final_system_prompt = fixed_system_prompt_content;
         console.log("Using only fixed system prompt.");
     } else if (hasSystemHistory) {
+        // 只有历史部分 (固定部分为空)
         final_system_prompt = system_prompt_history_content;
         console.log("Using only history in system prompt slot.");
     }
+    // 如果两部分都为空，final_system_prompt 保持空字符串 ""
 
+    // 6. 返回结果
     const result = {
-        system_prompt: final_system_prompt,
-        prompt: final_prompt
+        system_prompt: final_system_prompt, // 最终结果不需要再 trim
+        prompt: final_prompt              // final_prompt 在组合前已 trim
     };
 
     console.log(`Final system_prompt length (Sys+Separator+Hist): ${result.system_prompt.length}`);
@@ -250,149 +243,249 @@ function convertMessagesToFalPrompt(messages) {
 // === convertMessagesToFalPrompt 函数结束 ===
 
 
-// POST /v1/chat/completions endpoint (主要修改处)
+// POST /v1/chat/completions endpoint (带 Key 重试逻辑)
 app.post('/v1/chat/completions', async (req, res) => {
     const { model, messages, stream = false, reasoning = false, ...restOpenAIParams } = req.body;
+    const requestId = `req-${Date.now()}`; // Unique ID for this incoming request
 
-    console.log(`Received chat completion request for model: ${model}, stream: ${stream}`);
+    console.log(`[${requestId}] Received chat completion request for model: ${model}, stream: ${stream}`);
 
     if (!FAL_SUPPORTED_MODELS.includes(model)) {
-         console.warn(`Warning: Requested model '${model}' is not in the explicitly supported list.`);
+        console.warn(`[${requestId}] Warning: Requested model '${model}' is not in the explicitly supported list.`);
     }
     if (!model || !messages || !Array.isArray(messages) || messages.length === 0) {
-        console.error("Invalid request parameters:", { model, messages: Array.isArray(messages) ? messages.length : typeof messages });
+        console.error(`[${requestId}] Invalid request parameters:`, { model, messages: Array.isArray(messages) ? messages.length : typeof messages });
         return res.status(400).json({ error: 'Missing or invalid parameters: model and messages array are required.' });
     }
 
-    try {
-        // *** 在处理请求前，获取下一个 Fal Key 并配置 fal 客户端 ***
-        const selectedFalKey = getNextFalKey();
-        fal.config({
-            credentials: selectedFalKey,
-        });
-        // *********************************************************
-
-        const { prompt, system_prompt } = convertMessagesToFalPrompt(messages);
-
-        const falInput = {
-            model: model,
-            prompt: prompt,
-            ...(system_prompt && { system_prompt: system_prompt }),
-            reasoning: !!reasoning,
-        };
-	    console.log("Fal Input:", JSON.stringify(falInput, null, 2));
-        console.log("Forwarding request to fal-ai with system-priority + separator + recency input:");
-        console.log("System Prompt Length:", system_prompt?.length || 0);
-        console.log("Prompt Length:", prompt?.length || 0);
-        console.log("--- System Prompt Start ---");
-        console.log(system_prompt);
-        console.log("--- System Prompt End ---");
-        console.log("--- Prompt Start ---");
-        console.log(prompt);
-        console.log("--- Prompt End ---");
-
-
-        // --- 流式/非流���处理逻辑 (保持不变) ---
-        if (stream) {
-            res.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
-            res.setHeader('Cache-Control', 'no-cache');
-            res.setHeader('Connection', 'keep-alive');
-            res.setHeader('Access-Control-Allow-Origin', '*');
-            res.flushHeaders();
-
-            let previousOutput = '';
-
-            const falStream = await fal.stream("fal-ai/any-llm", { input: falInput });
+    let lastError = null; // Store the last error encountered during key rotation
+    let success = false; // Flag to indicate if any key succeeded
+
+    // *** 重试循环：尝试最多 falKeys.length 次 ***
+    for (let attempt = 0; attempt < falKeys.length; attempt++) {
+        const keyIndexToTry = (currentFalKeyIndex + attempt) % falKeys.length;
+        const selectedFalKey = falKeys[keyIndexToTry];
+        console.log(`[${requestId}] Attempt ${attempt + 1}/${falKeys.length}: Trying Fal Key at index ${keyIndexToTry}`);
+
+        try {
+            // 配置 fal 客户端 (每次尝试都重新配置)
+            fal.config({
+                credentials: selectedFalKey,
+            });
+
+            // 准备 Fal Input (只需要准备一次)
+            // 注意：如果 convertMessagesToFalPrompt 很耗时，可以移到循环外
+            const { prompt, system_prompt } = convertMessagesToFalPrompt(messages);
+            const falInput = {
+                model: model,
+                prompt: prompt,
+                ...(system_prompt && { system_prompt: system_prompt }),
+                reasoning: !!reasoning,
+            };
 
-            try {
-                for await (const event of falStream) {
-                    const currentOutput = (event && typeof event.output === 'string') ? event.output : '';
-                    const isPartial = (event && typeof event.partial === 'boolean') ? event.partial : true;
-                    const errorInfo = (event && event.error) ? event.error : null;
-
-                    if (errorInfo) {
-                        console.error("Error received in fal stream event:", errorInfo);
-                        const errorChunk = { id: `chatcmpl-${Date.now()}-error`, object: "chat.completion.chunk", created: Math.floor(Date.now() / 1000), model: model, choices: [{ index: 0, delta: {}, finish_reason: "error", message: { role: 'assistant', content: `Fal Stream Error: ${JSON.stringify(errorInfo)}` } }] };
-                        res.write(`data: ${JSON.stringify(errorChunk)}\n\n`);
-                        break; // Stop processing on error
-                    }
+            // 打印一次 Fal Input 和 Prompt 信息
+            if (attempt === 0) {
+                console.log(`[${requestId}] Fal Input:`, JSON.stringify(falInput, null, 2));
+                console.log(`[${requestId}] Forwarding request to fal-ai with system-priority + separator + recency input:`);
+                console.log(`[${requestId}] System Prompt Length:`, system_prompt?.length || 0);
+                console.log(`[${requestId}] Prompt Length:`, prompt?.length || 0);
+                // 为了简洁，默认注释掉详细内容的打印
+                // console.log(`[${requestId}] --- System Prompt Start ---`);
+                // console.log(system_prompt);
+                // console.log(`[${requestId}] --- System Prompt End ---`);
+                // console.log(`[${requestId}] --- Prompt Start ---`);
+                // console.log(prompt);
+                // console.log(`[${requestId}] --- Prompt End ---`);
+            }
 
-                    let deltaContent = '';
-                    if (currentOutput.startsWith(previousOutput)) {
-                        deltaContent = currentOutput.substring(previousOutput.length);
-                    } else if (currentOutput.length > 0) {
-                        console.warn("Fal stream output mismatch detected. Sending full current output as delta.", { previousLength: previousOutput.length, currentLength: currentOutput.length });
-                        deltaContent = currentOutput;
-                        previousOutput = ''; // Reset previous if mismatch
-                    }
-                    previousOutput = currentOutput;
+            // --- 执行 Fal AI 调用 ---
+            if (stream) {
+                // --- 流式处理 ---
+                res.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
+                res.setHeader('Cache-Control', 'no-cache');
+                res.setHeader('Connection', 'keep-alive');
+                res.setHeader('Access-Control-Allow-Origin', '*');
+                // 注意：Headers 只能发送一次
 
-                    // Send chunk if there's content or if it's the final chunk (isPartial is false)
-                    if (deltaContent || !isPartial) {
-                        const openAIChunk = { id: `chatcmpl-${Date.now()}`, object: "chat.completion.chunk", created: Math.floor(Date.now() / 1000), model: model, choices: [{ index: 0, delta: { content: deltaContent }, finish_reason: isPartial === false ? "stop" : null }] };
-                        res.write(`data: ${JSON.stringify(openAIChunk)}\n\n`);
-                    }
-                 }
-                // After the loop, ensure the [DONE] signal is sent if the stream finished normally
-                if (!res.writableEnded) {
-                    res.write(`data: [DONE]\n\n`);
-                    res.end();
-                    console.log("Stream finished and [DONE] sent.");
-                }
+                let previousOutput = '';
+                const falStream = await fal.stream("fal-ai/any-llm", { input: falInput });
 
-            } catch (streamError) {
-                console.error('Error during fal stream processing loop:', streamError);
+                // 标记成功，设置下一次请求的起始 Key 索引
+                success = true;
+                currentFalKeyIndex = (keyIndexToTry + 1) % falKeys.length;
+                console.log(`[${requestId}] Key at index ${keyIndexToTry} successful. Next request starts at index ${currentFalKeyIndex}.`);
+
+                // 处理流
                 try {
-                    if (!res.writableEnded) { // Check if we can still write to the response
-                        const errorDetails = (streamError instanceof Error) ? streamError.message : JSON.stringify(streamError);
-                        res.write(`data: ${JSON.stringify({ error: { message: "Stream processing error", type: "proxy_error", details: errorDetails } })}\n\n`);
-                        res.write(`data: [DONE]\n\n`); // Send DONE even after error
+                     if (!res.headersSent) { // 确保 header 只发送一次
+                         res.flushHeaders();
+                         console.log(`[${requestId}] Stream headers flushed.`);
+                     }
+                    for await (const event of falStream) {
+                        // ... (流处理逻辑基本不变，添加 requestId 用于日志) ...
+                        const currentOutput = (event && typeof event.output === 'string') ? event.output : '';
+                        const isPartial = (event && typeof event.partial === 'boolean') ? event.partial : true;
+                        const errorInfo = (event && event.error) ? event.error : null;
+
+                        if (errorInfo) {
+                            console.error(`[${requestId}] Error received in fal stream event:`, errorInfo);
+                            const errorChunk = { id: `chatcmpl-${Date.now()}-error`, object: "chat.completion.chunk", created: Math.floor(Date.now() / 1000), model: model, choices: [{ index: 0, delta: {}, finish_reason: "error", message: { role: 'assistant', content: `Fal Stream Error: ${JSON.stringify(errorInfo)}` } }] };
+                            if (!res.writableEnded) {
+                                res.write(`data: ${JSON.stringify(errorChunk)}\n\n`);
+                            }
+                            break; // Stop processing on error in stream event
+                        }
+
+                        let deltaContent = '';
+                        if (currentOutput.startsWith(previousOutput)) {
+                            deltaContent = currentOutput.substring(previousOutput.length);
+                        } else if (currentOutput.length > 0) {
+                            console.warn(`[${requestId}] Fal stream output mismatch detected. Sending full current output as delta.`, { previousLength: previousOutput.length, currentLength: currentOutput.length });
+                            deltaContent = currentOutput;
+                            previousOutput = ''; // Reset previous if mismatch
+                        }
+                        previousOutput = currentOutput;
+
+                        // Send chunk if there's content or if it's the final chunk (isPartial is false)
+                        if (deltaContent || !isPartial) {
+                            const openAIChunk = { id: `chatcmpl-${Date.now()}`, object: "chat.completion.chunk", created: Math.floor(Date.now() / 1000), model: model, choices: [{ index: 0, delta: { content: deltaContent }, finish_reason: isPartial === false ? "stop" : null }] };
+                             if (!res.writableEnded) {
+                                res.write(`data: ${JSON.stringify(openAIChunk)}\n\n`);
+                             }
+                        }
+                     }
+                     // After the loop, ensure the [DONE] signal is sent if the stream finished normally
+                     if (!res.writableEnded) {
+                        res.write(`data: [DONE]\n\n`);
                         res.end();
-                    } else {
-                         console.error("Stream already ended, cannot send error message.");
+                        console.log(`[${requestId}] Stream finished and [DONE] sent.`);
+                     }
+                } catch (streamError) {
+                    console.error(`[${requestId}] Error during fal stream processing loop:`, streamError);
+                    lastError = streamError; // Store error from stream processing
+                    try {
+                        // Don't mark success=false here, the key worked but the stream itself failed.
+                        // The outer loop should break because the response has likely been ended.
+                        if (!res.writableEnded) { // Check if we can still write to the response
+                            const errorDetails = (streamError instanceof Error) ? streamError.message : JSON.stringify(streamError);
+                            res.write(`data: ${JSON.stringify({ error: { message: "Stream processing error", type: "proxy_error", details: errorDetails } })}\n\n`);
+                            res.write(`data: [DONE]\n\n`); // Send DONE even after error
+                            res.end();
+                        } else {
+                             console.error(`[${requestId}] Stream already ended, cannot send error message.`);
+                        }
+                    } catch (finalError) {
+                         console.error(`[${requestId}] Error sending stream error message to client:`, finalError);
+                        if (!res.writableEnded) { res.end(); }
                     }
-                } catch (finalError) {
-                    console.error('Error sending stream error message to client:', finalError);
-                    if (!res.writableEnded) { res.end(); }
+                     // Break the outer key retry loop as the stream failed mid-way
+                     break;
                 }
-            }
-        } else {
-            // --- 非流式处理 (保持不变) ---
-            console.log("Executing non-stream request...");
-            const result = await fal.subscribe("fal-ai/any-llm", { input: falInput, logs: true });
-            console.log("Received non-stream result from fal-ai:", JSON.stringify(result, null, 2));
-
-            if (result && result.error) {
-                 console.error("Fal-ai returned an error in non-stream mode:", result.error);
-                 return res.status(500).json({ object: "error", message: `Fal-ai error: ${JSON.stringify(result.error)}`, type: "fal_ai_error", param: null, code: null });
+                // 如果流成功处理完，直接跳出重试循环
+                break; // Exit the key retry loop
+
+            } else {
+                // --- 非流式处理 ---
+                console.log(`[${requestId}] Executing non-stream request with key index ${keyIndexToTry}...`);
+                const result = await fal.subscribe("fal-ai/any-llm", { input: falInput, logs: true });
+
+                // 检查 Fal AI 返回的业务错误 (例如输入无效)，这种错误不应该因为换 Key 而解决
+                if (result && result.error) {
+                     console.error(`[${requestId}] Fal-ai returned a business error with key index ${keyIndexToTry}:`, result.error);
+                     // 将此视为最终错误，不重试其他 key
+                     lastError = new Error(`Fal-ai error: ${JSON.stringify(result.error)}`);
+                     lastError.status = 500; // Or map from Fal error if possible, default 500
+                     lastError.type = "fal_ai_error";
+                     break; // Exit retry loop, no point trying other keys for bad input
+                }
+
+                console.log(`[${requestId}] Received non-stream result from fal-ai with key index ${keyIndexToTry}`);
+                // console.log("Full non-stream result:", JSON.stringify(result, null, 2)); // Uncomment for detailed logs
+
+                // 标记成功，设置下一次请求的起始 Key 索引
+                success = true;
+                currentFalKeyIndex = (keyIndexToTry + 1) % falKeys.length;
+                console.log(`[${requestId}] Key at index ${keyIndexToTry} successful. Next request starts at index ${currentFalKeyIndex}.`);
+
+
+                const openAIResponse = {
+                    id: `chatcmpl-${result.requestId || Date.now()}`, object: "chat.completion", created: Math.floor(Date.now() / 1000), model: model,
+                    choices: [{ index: 0, message: { role: "assistant", content: result.output || "" }, finish_reason: "stop" }],
+                    usage: { prompt_tokens: null, completion_tokens: null, total_tokens: null }, // Fal doesn't provide token usage
+                    system_fingerprint: null, // Fal doesn't provide system fingerprint
+                    ...(result.reasoning && { fal_reasoning: result.reasoning }), // Include reasoning if present
+                };
+                res.json(openAIResponse);
+                console.log(`[${requestId}] Returned non-stream response.`);
+                break; // 成功，跳出重试循环
             }
 
-            const openAIResponse = {
-                id: `chatcmpl-${result.requestId || Date.now()}`, object: "chat.completion", created: Math.floor(Date.now() / 1000), model: model,
-                choices: [{ index: 0, message: { role: "assistant", content: result.output || "" }, finish_reason: "stop" }],
-                usage: { prompt_tokens: null, completion_tokens: null, total_tokens: null }, system_fingerprint: null,
-                ...(result.reasoning && { fal_reasoning: result.reasoning }),
-            };
-            res.json(openAIResponse);
-            console.log("Returned non-stream response.");
+        } catch (error) {
+            lastError = error; // Store the error from this attempt
+            const status = error?.status; // Fal client errors should have status
+            const errorMessage = error?.body?.detail || error?.message || 'Unknown error'; // Get detailed message if possible
+
+            console.warn(`[${requestId}] Attempt ${attempt + 1} with key index ${keyIndexToTry} failed. Status: ${status || 'N/A'}, Message: ${errorMessage}`);
+
+            // 检查是否是与 Key 相关的错误 (401 Unauthorized, 403 Forbidden, 429 Rate Limit)
+            if (status === 401 || status === 403 || status === 429) {
+                console.log(`[${requestId}] Key-related error (${status}). Trying next key...`);
+                // 继续循环尝试下一个 Key
+            } else {
+                 // 如果是其他类型的错误 (如网络问题、Fal内部服务器错误5xx、请求参数错误400等)
+                 // 通常重试其他 Key 没有意义，直接中断重试
+                 console.error(`[${requestId}] Unrecoverable error encountered. Status: ${status || 'N/A'}. Stopping key rotation for this request.`);
+                 console.error("Error details:", error); // Log the full error object for debugging
+                 break; // 跳出重试循环
+            }
         }
+    } // --- 结束重试循环 ---
 
-    } catch (error) {
-        console.error('Unhandled error in /v1/chat/completions:', error);
+    // 如果循环结束了还没有成功 (所有 Key 都失败了或遇到不可恢复错误)
+    if (!success) {
+        console.error(`[${requestId}] All Fal Key attempts failed or an unrecoverable error occurred.`);
         if (!res.headersSent) {
-            const errorMessage = (error instanceof Error) ? error.message : JSON.stringify(error);
-            res.status(500).json({ error: 'Internal Server Error in Proxy', details: errorMessage });
+            const statusCode = lastError?.status || 503; // Use status from last error if available, default to 503 Service Unavailable
+            const errorMessage = (lastError instanceof Error) ? lastError.message : JSON.stringify(lastError);
+            // Try to extract a more specific message if available
+            const detailMessage = lastError?.body?.detail || errorMessage;
+            const errorType = lastError?.type || (statusCode === 401 || statusCode === 403 || statusCode === 429 ? "key_error" : "proxy_error");
+
+             console.error(`[${requestId}] Sending final error response. Status: ${statusCode}, Type: ${errorType}, Message: ${detailMessage}`);
+
+             // 返回一个标准的 OpenAI 错误格式
+             res.status(statusCode).json({
+                 object: "error",
+                 message: `All Fal Key attempts failed or an unrecoverable error occurred. Last error: ${detailMessage}`,
+                 type: errorType,
+                 param: null,
+                 code: statusCode === 429 ? "rate_limit_exceeded" : (statusCode === 401 || statusCode === 403 ? "invalid_api_key" : null)
+             });
         } else if (!res.writableEnded) {
-             console.error("Headers already sent, ending response.");
-             res.end();
+            console.error(`[${requestId}] Headers already sent, but request failed after stream started or during processing. Ending response with error chunk.`);
+            // 尝试在流式响应中发送错误（如果可能）
+            try {
+                const errorDetails = (lastError instanceof Error) ? lastError.message : JSON.stringify(lastError);
+                const detailMessage = lastError?.body?.detail || errorDetails;
+                const errorChunk = { id: `chatcmpl-${Date.now()}-final-error`, object: "chat.completion.chunk", created: Math.floor(Date.now() / 1000), model: model, choices: [{ index: 0, delta: {}, finish_reason: "error", message: { role: 'assistant', content: `Proxy Error: All key attempts failed or stream processing error. Last error: ${detailMessage}` } }] };
+                res.write(`data: ${JSON.stringify(errorChunk)}\n\n`);
+                res.write(`data: [DONE]\n\n`);
+                res.end();
+            } catch (e) {
+                console.error(`[${requestId}] Failed to write final error to stream:`, e);
+                if (!res.writableEnded) res.end(); // Force end if possible
+            }
+        } else {
+            console.error(`[${requestId}] Request failed, but response stream was already fully ended. Cannot send error.`);
         }
     }
+
 });
 
 // 启动服务器 (更新启动信息)
 app.listen(PORT, () => {
     console.log(`===================================================`);
-    console.log(` Fal OpenAI Proxy Server (Key Rotation + System Top + Separator + Recency)`);
+    console.log(` Fal OpenAI Proxy Server (Key Rotation with Retry + System Top + Separator + Recency)`); // 更新描述
     console.log(` Listening on port: ${PORT}`);
     console.log(` Loaded ${falKeys.length} Fal AI Keys for rotation.`);
     console.log(` Using Limits: System Prompt=${SYSTEM_PROMPT_LIMIT}, Prompt=${PROMPT_LIMIT}`);
@@ -404,5 +497,5 @@ app.listen(PORT, () => {
 
 // 根路径响应 (更新信息)
 app.get('/', (req, res) => {
-    res.send('Fal OpenAI Proxy (Key Rotation + System Top + Separator + Recency Strategy) is running.');
+    res.send('Fal OpenAI Proxy (Key Rotation with Retry + System Top + Separator + Recency Strategy) is running.'); // 更新描述
 });