feat(channels): 增强负载均衡和重试逻辑

2026-04-12 20:15:53 +08:00
1 parent 8deda988e4
commit 954e1b539f
@@ -11,7 +11,7 @@

 - 🚀 **基于 Cloudflare Workers**：全球分布式网络、低延迟、高性能
 - 🔐 **多渠道**：支持 OpenAI、Gemini、Azure OpenAI、Claude 等多种 AI 服务商
- 🎫 **负载均衡**：支持按权重多渠道/多密钥随机分配请求，自动重试/轮换，确保高可用性
+- 🎫 **负载均衡**：支持按权重多渠道路由、单渠道多密钥重试/轮换，以及跨渠道 fallback
 - 📊 **用量统计**：Analytics Engine 实时统计，支持日志回溯便于成本分析
 - 💰 **定价管理**：灵活的模型定价配置
 - 🎨 **Web 管理界面**：直观的 Web 界面进行配置管理
@@ -1319,7 +1319,7 @@ export function Channels({ createMode = false, editRoute = false }: { createMode
                        />
                        <div>
                          <div className="text-sm font-medium">自动重试</div>
-                          <p className="mt-1 text-xs text-muted-foreground">单个密钥遇可重试错误时，自动重试 x3</p>
+                          <p className="mt-1 text-xs text-muted-foreground">渠道遇可重试错误时，最多自动重试 3 次</p>
                        </div>
                      </label>
                      <label className="flex items-start gap-3 p-4 rounded-lg border bg-muted/30 cursor-pointer">
@@ -1330,7 +1330,7 @@ export function Channels({ createMode = false, editRoute = false }: { createMode
                        />
                        <div>
                          <div className="text-sm font-medium">自动轮换</div>
-                          <p className="mt-1 text-xs text-muted-foreground">单个密钥多次失败后，自动轮换密钥 x3</p>
+                          <p className="mt-1 text-xs text-muted-foreground">重试时随机切换同渠道其他密钥；关闭后始终重试当前密钥</p>
                        </div>
                      </label>
                    </div>
@@ -16,7 +16,7 @@ const features = [
  {
    icon: Gauge,
    title: "负载均衡",
-    description: "支持按权重多渠道/多密钥随机分配请求，自动重试/轮换，确保高可用性。",
+    description: "支持按权重多渠道路由、单渠道多密钥重试/轮换，以及跨渠道 fallback。",
    gradient: "from-emerald-500/10 to-teal-500/10",
    iconBg: "from-emerald-500 to-teal-500",
    delay: "50ms",
@@ -190,8 +190,8 @@ export class ChannelUpsertEndpoint extends OpenAPIRoute {
                            weight: z.number().int().min(0).max(5).optional().describe('Priority weight for channel routing'),
                            api_key: z.string().optional().describe('Deprecated single API key'),
                            api_keys: z.array(z.string()).optional().describe('API keys, one request will pick one randomly'),
-                            auto_retry: z.boolean().optional().describe('Automatically retry the same API key'),
-                            auto_rotate: z.boolean().optional().describe('Automatically rotate to other API keys on retryable failures'),
+                            auto_retry: z.boolean().optional().describe('Automatically retry the channel up to 3 times on retryable failures'),
+                            auto_rotate: z.boolean().optional().describe('When retrying, randomly rotate to other API keys in the same channel'),
                            models: z.array(ChannelModelSchema).optional().describe('External model name to upstream model ID mappings'),
                            supported_models: z.array(z.string()).optional().describe('Deprecated supported request model list'),
                            deployment_mapper: z.record(z.string(), z.string()).optional().describe('Deprecated model deployment mapping'),
@@ -5,8 +5,8 @@ export const DEFAULT_CHANNEL_WEIGHT = 0;
 export const MAX_CHANNEL_WEIGHT = 5;
 export const DEFAULT_CHANNEL_MODEL_ENABLED = true;
 export const DEFAULT_CLAUDE_API_VERSION = "2023-06-01";
-export const MAX_RETRIES_PER_KEY = 3;
-export const MAX_ROTATION_ATTEMPTS = 3;
+export const MAX_CHANNEL_RETRIES = 3;
+export const MAX_CHANNEL_FALLBACKS = 3;

 export type NormalizedChannelConfig = Omit<ChannelConfig, "enabled" | "weight" | "api_keys" | "auto_retry" | "auto_rotate" | "models" | "supported_models" | "deployment_mapper"> & {
    enabled: boolean;
@@ -5,8 +5,7 @@ import { z } from "zod";
 import db from "../db"
 import { resolveRouteId } from "./shared/route-policy"
 import { resolveChannel } from "./shared/channel-resolver"
-import { getProvider } from "./shared/provider-registry"
-import { executeWithChannelKeys } from "./shared/upstream-retry"
+import { executeWithFallbackChannels } from "./shared/upstream-retry"
 import { ModelsEndpoint } from "./models"

 export const api = fromHono(new Hono<HonoCustomType>())
@@ -42,21 +41,17 @@ class UnifiedProxyEndpoint extends OpenAPIRoute {
        const result = await resolveChannel(c, routeId)
        if (result instanceof Response) return result

-        const { channel, requestBody, saveUsage, logFailure, trackingState } = result
+        const { channels, initialChannel, requestBody, saveUsage, logFailure, trackingState, setActiveChannel } = result

-        const provider = getProvider(channel.config.type || "")
-        if (!provider) {
-            return c.text("Channel type not supported", 400)
-        }
-
-        return executeWithChannelKeys(
+        return executeWithFallbackChannels(
            c,
-            channel.config,
+            channels,
+            initialChannel,
            requestBody,
            saveUsage,
            logFailure,
            trackingState,
-            provider
+            setActiveChannel,
        )
    }
 }
@@ -11,12 +11,33 @@ import {
    writeUsageSuccessEvent,
 } from "../../analytics/usage-logger"

+export type ResolvedChannelCandidate = {
+    key: string
+    config: ChannelConfig
+    mapping: ChannelModelMapping
+}
+
 export type ChannelResolution = {
-    channel: { key: string; config: ChannelConfig }
+    channels: ResolvedChannelCandidate[]
+    initialChannel: ResolvedChannelCandidate
    requestBody: any
    saveUsage: (usage: Usage) => Promise<void>
    logFailure: (errorCode: string, errorSummary?: string) => void
    trackingState: RequestTrackingState
+    setActiveChannel: (channel: ResolvedChannelCandidate) => void
+}
+
+export const pickHighestPriorityChannel = (
+    channels: ResolvedChannelCandidate[]
+): ResolvedChannelCandidate => {
+    const maxWeight = channels.reduce((highest, current) => {
+        return Math.max(highest, current.config.weight ?? 0);
+    }, 0);
+    const highestPriorityChannels = channels.filter((channel) => {
+        return (channel.config.weight ?? 0) === maxWeight;
+    });
+    const randomIndex = Math.floor(Math.random() * highestPriorityChannels.length);
+    return highestPriorityChannels[randomIndex];
 }

 export const resolveChannel = async (
@@ -59,7 +80,7 @@ export const resolveChannel = async (
    const policy = getRoutePolicy(routeId);
    const allowedTypes = policy.allowedTypes;

-    const availableChannels: Array<{ key: string, config: ChannelConfig, mapping: ChannelModelMapping }> = [];
+    const availableChannels: ResolvedChannelCandidate[] = [];
    let hasDisabledMatchingChannel = false;

    for (const row of channelsResult.results) {
@@ -103,16 +124,7 @@ export const resolveChannel = async (
        return c.text(`Model not supported: ${requestedModel}. Please configure models.`, 400);
    }

-    const maxWeight = availableChannels.reduce((highest, current) => {
-        return Math.max(highest, current.config.weight ?? 0);
-    }, 0);
-    const highestPriorityChannels = availableChannels.filter((channel) => {
-        return (channel.config.weight ?? 0) === maxWeight;
-    });
-    const randomIndex = Math.floor(Math.random() * highestPriorityChannels.length);
-    const selectedChannel = highestPriorityChannels[randomIndex];
-    const targetChannelKey = selectedChannel.key;
-    const targetChannelConfig = selectedChannel.config;
+    const selectedChannel = pickHighestPriorityChannel(availableChannels);
    const trackingState: RequestTrackingState = {
        retryCount: 0,
    };
@@ -124,8 +136,8 @@ export const resolveChannel = async (
        routeId,
        tokenHash,
        tokenName,
-        channelKey: targetChannelKey,
-        providerType: targetChannelConfig.type || "unknown",
+        channelKey: selectedChannel.key,
+        providerType: selectedChannel.config.type || "unknown",
        requestedModel,
        upstreamModel: selectedChannel.mapping.id,
        streamMode: requestBody.stream ? "stream" as const : "sync" as const,
@@ -142,12 +154,14 @@ export const resolveChannel = async (
        trackingState,
    };
    let usageLogState: "idle" | "processing" | "done" = "idle";
+    let activeChannel = selectedChannel;

-    requestBody.model = selectedChannel.mapping.id;
-
-    if (!targetChannelConfig.type) {
-        return c.text("Channel type invalid", 400);
-    }
+    const setActiveChannel = (channel: ResolvedChannelCandidate) => {
+        activeChannel = channel;
+        usageContext.channelKey = channel.key;
+        usageContext.providerType = channel.config.type || "unknown";
+        usageContext.upstreamModel = channel.mapping.id;
+    };

    const saveUsage = async (usage: Usage) => {
        if (usageLogState !== "idle") {
@@ -160,8 +174,8 @@ export const resolveChannel = async (
                c,
                apiKey,
                requestedModel,
-                targetChannelKey,
-                targetChannelConfig,
+                activeChannel.key,
+                activeChannel.config,
                usage
            );
            writeUsageSuccessEvent(c, usageContext, usage, costResult);
@@ -181,10 +195,12 @@ export const resolveChannel = async (
    };

    return {
-        channel: { key: targetChannelKey, config: targetChannelConfig },
+        channels: availableChannels,
+        initialChannel: selectedChannel,
        requestBody,
        saveUsage,
        logFailure,
        trackingState,
+        setActiveChannel,
    };
 }
@@ -1,18 +1,32 @@
 import { Context } from "hono";

 import {
-    MAX_RETRIES_PER_KEY,
-    MAX_ROTATION_ATTEMPTS,
+    MAX_CHANNEL_FALLBACKS,
+    MAX_CHANNEL_RETRIES,
    normalizeChannelConfig,
 } from "../../channel-config";
 import {
    summarizeErrorFromResponse,
    summarizeErrorFromUnknown,
 } from "../../analytics/usage-logger";
-import { ProviderFetch } from "./provider-registry";
+import {
+    pickHighestPriorityChannel,
+    ResolvedChannelCandidate,
+} from "./channel-resolver";
+import {
+    getProvider,
+    ProviderFetch,
+} from "./provider-registry";

 const RETRYABLE_STATUS_CODES = new Set([401, 403, 408, 409, 429, 500, 502, 503, 504, 529]);

+type ChannelExecutionResult = {
+    response: Response
+    shouldFallback: boolean
+    errorCode: string
+    errorSummary?: string
+}
+
 const shuffleKeys = (keys: string[]): string[] => {
    const clonedKeys = [...keys];

@@ -24,6 +38,10 @@ const shuffleKeys = (keys: string[]): string[] => {
    return clonedKeys;
 };

+const pickRandomItem = <T>(items: T[]): T => {
+    return items[Math.floor(Math.random() * items.length)];
+};
+
 const cloneRequestBody = <T>(requestBody: T): T => {
    if (requestBody == null) {
        return requestBody;
@@ -32,6 +50,17 @@ const cloneRequestBody = <T>(requestBody: T): T => {
    return JSON.parse(JSON.stringify(requestBody)) as T;
 };

+const buildChannelRequestBody = (
+    requestBody: any,
+    channel: ResolvedChannelCandidate
+) => {
+    const runtimeRequestBody = cloneRequestBody(requestBody);
+    if (runtimeRequestBody && typeof runtimeRequestBody === "object") {
+        runtimeRequestBody.model = channel.mapping.id;
+    }
+    return runtimeRequestBody;
+};
+
 const shouldRetryResponse = (response: Response): boolean => {
    return RETRYABLE_STATUS_CODES.has(response.status);
 };
@@ -44,104 +73,251 @@ const discardResponse = async (response: Response): Promise<void> => {
    }
 };

-const getCandidateKeys = (config: ChannelConfig): string[] => {
-    const normalized = normalizeChannelConfig(config);
-    const shuffledKeys = shuffleKeys(normalized.api_keys);
-
-    if (!normalized.auto_rotate) {
-        return shuffledKeys.slice(0, 1);
-    }
-
-    return shuffledKeys.slice(0, Math.min(shuffledKeys.length, 1 + MAX_ROTATION_ATTEMPTS));
+const pickInitialKey = (config: ChannelConfig): string | null => {
+    const normalizedConfig = normalizeChannelConfig(config);
+    const shuffledKeys = shuffleKeys(normalizedConfig.api_keys);
+    return shuffledKeys[0] || null;
 };

-export const executeWithChannelKeys = async (
-    c: Context<HonoCustomType>,
+const pickRetryKey = (
    config: ChannelConfig,
+    currentKey: string,
+    usedKeys: Set<string>
+): string => {
+    const normalizedConfig = normalizeChannelConfig(config);
+
+    if (!normalizedConfig.auto_rotate) {
+        return currentKey;
+    }
+
+    const unusedOtherKeys = normalizedConfig.api_keys.filter((key) => {
+        return key !== currentKey && !usedKeys.has(key);
+    });
+    const otherKeys = normalizedConfig.api_keys.filter((key) => key !== currentKey);
+    const nextPool = unusedOtherKeys.length > 0
+        ? unusedOtherKeys
+        : (otherKeys.length > 0 ? otherKeys : [currentKey]);
+    const nextKey = pickRandomItem(nextPool);
+    usedKeys.add(nextKey);
+    return nextKey;
+};
+
+const pickNextFallbackChannel = (
+    channels: ResolvedChannelCandidate[],
+    attemptedChannelKeys: Set<string>
+): ResolvedChannelCandidate | null => {
+    const remainingChannels = channels.filter((channel) => {
+        return !attemptedChannelKeys.has(channel.key);
+    });
+
+    if (remainingChannels.length === 0) {
+        return null;
+    }
+
+    return pickHighestPriorityChannel(remainingChannels);
+};
+
+const executeChannelWithRetries = async (
+    c: Context<HonoCustomType>,
+    channel: ResolvedChannelCandidate,
+    requestBody: any,
+    saveUsage: (usage: Usage) => Promise<void>,
+    trackingState: RequestTrackingState,
+    provider: ProviderFetch,
+): Promise<ChannelExecutionResult> => {
+    const normalizedConfig = normalizeChannelConfig(channel.config);
+    const initialKey = pickInitialKey(normalizedConfig);
+
+    if (!initialKey) {
+        const errorSummary = "Channel API keys not configured";
+        trackingState.upstreamStatus = 0;
+        trackingState.errorSummary = errorSummary;
+
+        return {
+            response: c.text(errorSummary, 500),
+            shouldFallback: true,
+            errorCode: "channel_keys_missing",
+            errorSummary,
+        };
+    }
+
+    const maxAttempts = 1 + (normalizedConfig.auto_retry ? MAX_CHANNEL_RETRIES : 0);
+    const usedKeys = new Set<string>([initialKey]);
+    let currentKey = initialKey;
+
+    for (let attemptIndex = 0; attemptIndex < maxAttempts; attemptIndex += 1) {
+        try {
+            const runtimeConfig: ChannelConfig = {
+                ...normalizedConfig,
+                api_key: currentKey,
+                api_keys: [currentKey],
+            };
+
+            const response = await provider(
+                c,
+                runtimeConfig,
+                buildChannelRequestBody(requestBody, channel),
+                saveUsage,
+                trackingState,
+            );
+
+            if (response.ok) {
+                return {
+                    response,
+                    shouldFallback: false,
+                    errorCode: "",
+                };
+            }
+
+            if (!shouldRetryResponse(response)) {
+                const errorSummary = await summarizeErrorFromResponse(response);
+                trackingState.errorSummary = errorSummary;
+
+                return {
+                    response,
+                    shouldFallback: false,
+                    errorCode: `http_${response.status}`,
+                    errorSummary,
+                };
+            }
+
+            const hasNextAttempt = attemptIndex < maxAttempts - 1;
+
+            console.warn(
+                `Retryable upstream failure for channel "${normalizedConfig.name || channel.key}", `
+                + `attempt ${attemptIndex + 1}/${maxAttempts}, `
+                + `status ${response.status}`
+            );
+
+            if (!hasNextAttempt) {
+                const errorSummary = await summarizeErrorFromResponse(response);
+                trackingState.errorSummary = errorSummary;
+
+                return {
+                    response,
+                    shouldFallback: true,
+                    errorCode: `http_${response.status}`,
+                    errorSummary,
+                };
+            }
+
+            await discardResponse(response);
+            currentKey = pickRetryKey(normalizedConfig, currentKey, usedKeys);
+            trackingState.retryCount += 1;
+        } catch (error) {
+            const hasNextAttempt = attemptIndex < maxAttempts - 1;
+
+            console.error(
+                `Upstream request error for channel "${normalizedConfig.name || channel.key}", `
+                + `attempt ${attemptIndex + 1}/${maxAttempts}`,
+                error
+            );
+
+            if (!hasNextAttempt) {
+                trackingState.upstreamStatus = 0;
+                trackingState.errorSummary = summarizeErrorFromUnknown(error);
+                const message = error instanceof Error ? error.message : "Unknown upstream error";
+
+                return {
+                    response: c.text(`Upstream request failed: ${message}`, 502),
+                    shouldFallback: true,
+                    errorCode: "upstream_exception",
+                    errorSummary: trackingState.errorSummary,
+                };
+            }
+
+            trackingState.upstreamStatus = 0;
+            currentKey = pickRetryKey(normalizedConfig, currentKey, usedKeys);
+            trackingState.retryCount += 1;
+        }
+    }
+
+    const errorSummary = trackingState.errorSummary || "Upstream request failed after retries";
+    return {
+        response: c.text("Upstream request failed after retries", 502),
+        shouldFallback: true,
+        errorCode: "upstream_retries_exhausted",
+        errorSummary,
+    };
+};
+
+export const executeWithFallbackChannels = async (
+    c: Context<HonoCustomType>,
+    channels: ResolvedChannelCandidate[],
+    initialChannel: ResolvedChannelCandidate,
    requestBody: any,
    saveUsage: (usage: Usage) => Promise<void>,
    logFailure: (errorCode: string, errorSummary?: string) => void,
    trackingState: RequestTrackingState,
-    provider: ProviderFetch,
+    setActiveChannel: (channel: ResolvedChannelCandidate) => void,
 ): Promise<Response> => {
-    const normalizedConfig = normalizeChannelConfig(config);
-    const candidateKeys = getCandidateKeys(normalizedConfig);
+    const attemptedChannelKeys = new Set<string>();
+    let currentChannel = initialChannel;
+    let fallbackCount = 0;

-    if (candidateKeys.length === 0) {
-        return c.text("Channel API keys not configured", 500);
-    }
+    while (true) {
+        attemptedChannelKeys.add(currentChannel.key);
+        setActiveChannel(currentChannel);
+        trackingState.errorSummary = undefined;

-    const attemptsPerKey = normalizedConfig.auto_retry
-        ? 1 + MAX_RETRIES_PER_KEY
-        : 1;
+        const provider = getProvider(currentChannel.config.type || "");
+        if (!provider) {
+            const errorSummary = "Channel type not supported";
+            const response = c.text(errorSummary, 400);
+            const nextChannel = fallbackCount < MAX_CHANNEL_FALLBACKS
+                ? pickNextFallbackChannel(channels, attemptedChannelKeys)
+                : null;

-    for (let keyIndex = 0; keyIndex < candidateKeys.length; keyIndex += 1) {
-        const currentKey = candidateKeys[keyIndex];
+            trackingState.upstreamStatus = 0;
+            trackingState.errorSummary = errorSummary;

-        for (let attemptIndex = 0; attemptIndex < attemptsPerKey; attemptIndex += 1) {
-            const isLastAttemptForKey = attemptIndex === attemptsPerKey - 1;
-            const isLastKey = keyIndex === candidateKeys.length - 1;
-
-            try {
-                trackingState.retryCount = keyIndex * attemptsPerKey + attemptIndex;
-                const runtimeConfig: ChannelConfig = {
-                    ...normalizedConfig,
-                    api_key: currentKey,
-                    api_keys: [currentKey],
-                };
-
-                const response = await provider(
-                    c,
-                    runtimeConfig,
-                    cloneRequestBody(requestBody),
-                    saveUsage,
-                    trackingState,
-                );
-
-                if (response.ok) {
-                    return response;
-                }
-
-                if (!shouldRetryResponse(response)) {
-                    const errorSummary = await summarizeErrorFromResponse(response);
-                    trackingState.errorSummary = errorSummary;
-                    logFailure(`http_${response.status}`, errorSummary);
-                    return response;
-                }
-
-                console.warn(
-                    `Retryable upstream failure for channel "${normalizedConfig.name || "unknown"}", `
-                    + `key ${keyIndex + 1}/${candidateKeys.length}, `
-                    + `attempt ${attemptIndex + 1}/${attemptsPerKey}, `
-                    + `status ${response.status}`
-                );
-
-                if (isLastAttemptForKey && isLastKey) {
-                    const errorSummary = await summarizeErrorFromResponse(response);
-                    trackingState.errorSummary = errorSummary;
-                    logFailure(`http_${response.status}`, errorSummary);
-                    return response;
-                }
-
-                await discardResponse(response);
-            } catch (error) {
-                console.error(
-                    `Upstream request error for channel "${normalizedConfig.name || "unknown"}", `
-                    + `key ${keyIndex + 1}/${candidateKeys.length}, `
-                    + `attempt ${attemptIndex + 1}/${attemptsPerKey}`,
-                    error
-                );
-
-                if (isLastAttemptForKey && isLastKey) {
-                    trackingState.upstreamStatus = 0;
-                    trackingState.errorSummary = summarizeErrorFromUnknown(error);
-                    logFailure("upstream_exception", trackingState.errorSummary);
-                    const message = error instanceof Error ? error.message : "Unknown upstream error";
-                    return c.text(`Upstream request failed: ${message}`, 502);
-                }
+            if (!nextChannel) {
+                logFailure("channel_type_invalid", errorSummary);
+                return response;
            }
-        }
-    }

-    return c.text("Upstream request failed after retries", 502);
+            console.warn(
+                `Fallback to channel "${nextChannel.key}" after unsupported provider `
+                + `on channel "${currentChannel.key}"`
+            );
+
+            fallbackCount += 1;
+            trackingState.retryCount += 1;
+            currentChannel = nextChannel;
+            continue;
+        }
+
+        const execution = await executeChannelWithRetries(
+            c,
+            currentChannel,
+            requestBody,
+            saveUsage,
+            trackingState,
+            provider,
+        );
+
+        if (execution.response.ok) {
+            return execution.response;
+        }
+
+        const nextChannel = execution.shouldFallback && fallbackCount < MAX_CHANNEL_FALLBACKS
+            ? pickNextFallbackChannel(channels, attemptedChannelKeys)
+            : null;
+
+        if (!nextChannel) {
+            logFailure(execution.errorCode, execution.errorSummary);
+            return execution.response;
+        }
+
+        await discardResponse(execution.response);
+
+        console.warn(
+            `Fallback to channel "${nextChannel.key}" after channel "${currentChannel.key}" `
+            + `failed for model "${currentChannel.mapping.name}"`
+        );
+
+        fallbackCount += 1;
+        trackingState.retryCount += 1;
+        currentChannel = nextChannel;
+    }
 };