From 954e1b539f87fef5ba1d0dc16e5503d7cdd08c2d Mon Sep 17 00:00:00 2001 From: Jeff Gu Date: Sun, 12 Apr 2026 20:15:53 +0800 Subject: [PATCH] =?UTF-8?q?feat(channels):=20=E5=A2=9E=E5=BC=BA=E8=B4=9F?= =?UTF-8?q?=E8=BD=BD=E5=9D=87=E8=A1=A1=E5=92=8C=E9=87=8D=E8=AF=95=E9=80=BB?= =?UTF-8?q?=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 +- frontend/src/pages/Channels.tsx | 4 +- frontend/src/pages/Dashboard.tsx | 2 +- src/admin/channel_api.ts | 4 +- src/channel-config.ts | 4 +- src/providers/index.ts | 17 +- src/providers/shared/channel-resolver.ts | 60 ++-- src/providers/shared/upstream-retry.ts | 354 +++++++++++++++++------ 8 files changed, 317 insertions(+), 130 deletions(-) diff --git a/README.md b/README.md index 36f54ca..b2fe254 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ - 🚀 **基于 Cloudflare Workers**：全球分布式网络、低延迟、高性能 - 🔐 **多渠道**：支持 OpenAI、Gemini、Azure OpenAI、Claude 等多种 AI 服务商 -- 🎫 **负载均衡**：支持按权重多渠道/多密钥随机分配请求，自动重试/轮换，确保高可用性 +- 🎫 **负载均衡**：支持按权重多渠道路由、单渠道多密钥重试/轮换，以及跨渠道 fallback - 📊 **用量统计**：Analytics Engine 实时统计，支持日志回溯便于成本分析 - 💰 **定价管理**：灵活的模型定价配置 - 🎨 **Web 管理界面**：直观的 Web 界面进行配置管理 diff --git a/frontend/src/pages/Channels.tsx b/frontend/src/pages/Channels.tsx index 3bb371c..1cdf6a8 100644 --- a/frontend/src/pages/Channels.tsx +++ b/frontend/src/pages/Channels.tsx @@ -1319,7 +1319,7 @@ export function Channels({ createMode = false, editRoute = false }: { createMode />
自动重试
-

单个密钥遇可重试错误时，自动重试 x3

+

渠道遇可重试错误时，最多自动重试 3 次

diff --git a/frontend/src/pages/Dashboard.tsx b/frontend/src/pages/Dashboard.tsx index 7dee022..69f6018 100644 --- a/frontend/src/pages/Dashboard.tsx +++ b/frontend/src/pages/Dashboard.tsx @@ -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", diff --git a/src/admin/channel_api.ts b/src/admin/channel_api.ts index dc29e7b..627198b 100644 --- a/src/admin/channel_api.ts +++ b/src/admin/channel_api.ts @@ -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'), diff --git a/src/channel-config.ts b/src/channel-config.ts index a5c9b82..5ad8040 100644 --- a/src/channel-config.ts +++ b/src/channel-config.ts @@ -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 & { enabled: boolean; diff --git a/src/providers/index.ts b/src/providers/index.ts index 270d19d..240c3d8 100644 --- a/src/providers/index.ts +++ b/src/providers/index.ts @@ -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()) @@ -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, ) } } diff --git a/src/providers/shared/channel-resolver.ts b/src/providers/shared/channel-resolver.ts index dd17e74..4ee0f50 100644 --- a/src/providers/shared/channel-resolver.ts +++ b/src/providers/shared/channel-resolver.ts @@ -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 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, }; } diff --git a/src/providers/shared/upstream-retry.ts b/src/providers/shared/upstream-retry.ts index 2b172ab..d912568 100644 --- a/src/providers/shared/upstream-retry.ts +++ b/src/providers/shared/upstream-retry.ts @@ -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 = (items: T[]): T => { + return items[Math.floor(Math.random() * items.length)]; +}; + const cloneRequestBody = (requestBody: T): T => { if (requestBody == null) { return requestBody; @@ -32,6 +50,17 @@ const cloneRequestBody = (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 => { } }; -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, +const pickRetryKey = ( config: ChannelConfig, + currentKey: string, + usedKeys: Set +): 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 +): 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, + channel: ResolvedChannelCandidate, + requestBody: any, + saveUsage: (usage: Usage) => Promise, + trackingState: RequestTrackingState, + provider: ProviderFetch, +): Promise => { + 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([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, + channels: ResolvedChannelCandidate[], + initialChannel: ResolvedChannelCandidate, requestBody: any, saveUsage: (usage: Usage) => Promise, logFailure: (errorCode: string, errorSummary?: string) => void, trackingState: RequestTrackingState, - provider: ProviderFetch, + setActiveChannel: (channel: ResolvedChannelCandidate) => void, ): Promise => { - const normalizedConfig = normalizeChannelConfig(config); - const candidateKeys = getCandidateKeys(normalizedConfig); + const attemptedChannelKeys = new Set(); + 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; + } };