feat(channel): 支持多个 API 密钥和自动重试功能

- 更新 ChannelConfig 接口,添加 api_keys、auto_retry 和 auto_rotate 字段
- 修改 Channels 组件以支持新的配置
- 实现 API 请求的重试机制,增强稳定性
This commit is contained in:
Jeff Gu committed 2026-04-09 11:17:20 +08:00
1 parent 1ef5502083
commit a2c97959ba
19 files changed
+432 -119

No files matched your search

+15 -3
View File
@@ -3,6 +3,7 @@ import { contentJson, OpenAPIRoute } from 'chanfana';
import { z } from 'zod';
import { CommonErrorResponse, CommonSuccessfulResponse } from "../model";
import { sanitizeChannelConfig } from "../channel-config";
// 获取所有 Channel 配置
export class ChannelGetEndpoint extends OpenAPIRoute {
@@ -48,8 +49,10 @@ export class ChannelUpsertEndpoint extends OpenAPIRoute {
name: z.string().describe('Channel name'),
type: z.string().describe('Channel type'),
endpoint: z.string().describe('API endpoint'),
api_key: z.string().describe('API key'),
api_version: z.string().optional().describe('API version'),
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'),
supported_models: z.array(z.string()).describe('Supported request model list'),
deployment_mapper: z.record(z.string()).describe('Model deployment mapping'),
model_pricing: z.record(z.object({
@@ -71,7 +74,16 @@ export class ChannelUpsertEndpoint extends OpenAPIRoute {
async handle(c: Context<HonoCustomType>) {
const { key } = c.req.param();
const config = await c.req.json<ChannelConfig>();
const rawConfig = await c.req.json<ChannelConfig>();
const config = sanitizeChannelConfig(rawConfig);
if (!config.name || !config.endpoint) {
return c.text('Channel name and endpoint are required', 400);
}
if (!config.api_keys || config.api_keys.length === 0) {
return c.text('At least one API key is required', 400);
}
// Upsert channel config directly using SQL
// excluded.value 指的是 INSERT 语句中要插入的新值
+70
View File
@@ -0,0 +1,70 @@
export const DEFAULT_CHANNEL_AUTO_RETRY = true;
export const DEFAULT_CHANNEL_AUTO_ROTATE = true;
export const DEFAULT_CLAUDE_API_VERSION = "2023-06-01";
export const MAX_RETRIES_PER_KEY = 3;
export const MAX_ROTATION_ATTEMPTS = 3;
export type NormalizedChannelConfig = Omit<ChannelConfig, "api_keys" | "auto_retry" | "auto_rotate"> & {
api_keys: string[];
auto_retry: boolean;
auto_rotate: boolean;
};
const normalizeApiKeys = (config: Partial<ChannelConfig>): string[] => {
const rawKeys = Array.isArray(config.api_keys)
? [...config.api_keys]
: [];
if (typeof config.api_key === "string") {
rawKeys.unshift(config.api_key);
}
const normalizedKeys: string[] = [];
const seen = new Set<string>();
for (const key of rawKeys) {
const trimmedKey = typeof key === "string" ? key.trim() : "";
if (!trimmedKey || seen.has(trimmedKey)) {
continue;
}
seen.add(trimmedKey);
normalizedKeys.push(trimmedKey);
}
return normalizedKeys;
};
export const normalizeChannelConfig = (
config: Partial<ChannelConfig>
): NormalizedChannelConfig => {
return {
name: config.name || "",
type: config.type,
endpoint: config.endpoint || "",
api_key: typeof config.api_key === "string" ? config.api_key.trim() : undefined,
api_keys: normalizeApiKeys(config),
auto_retry: config.auto_retry ?? DEFAULT_CHANNEL_AUTO_RETRY,
auto_rotate: config.auto_rotate ?? DEFAULT_CHANNEL_AUTO_ROTATE,
supported_models: Array.isArray(config.supported_models) ? config.supported_models : [],
deployment_mapper: config.deployment_mapper || {},
model_pricing: config.model_pricing,
};
};
export const sanitizeChannelConfig = (
config: Partial<ChannelConfig>
): ChannelConfig => {
const normalized = normalizeChannelConfig(config);
return {
name: normalized.name,
type: normalized.type,
endpoint: normalized.endpoint,
api_keys: normalized.api_keys,
auto_retry: normalized.auto_retry,
auto_rotate: normalized.auto_rotate,
supported_models: normalized.supported_models,
deployment_mapper: normalized.deployment_mapper,
model_pricing: normalized.model_pricing,
};
};
+7 -18
View File
@@ -1,31 +1,20 @@
import { Context } from "hono"
import { checkoutUsageData, handleStreamResponse } from "./shared/openai-stream-utils"
import { buildAzureTargetUrl } from "./shared/azure-target-url"
const buildProxyRequest = (
request: Request,
reqJson: any,
config: ChannelConfig,
deploymentName: string
config: ChannelConfig
): Request => {
const url = new URL(request.url)
const targetUrl = new URL(config.endpoint)
if (!config.endpoint.endsWith('#')) {
const basePath = targetUrl.pathname.endsWith('/')
? targetUrl.pathname.slice(0, -1)
: targetUrl.pathname
targetUrl.pathname = `${basePath}/openai/deployments/${deploymentName}/${url.pathname.replace('/v1/', '')}`
}
if (config.api_version) {
targetUrl.searchParams.set('api-version', config.api_version)
}
const targetUrl = buildAzureTargetUrl(request, config.endpoint)
const apiKey = config.api_key || ""
const targetHeaders = new Headers(request.headers)
targetHeaders.delete("Authorization")
targetHeaders.delete("Host")
targetHeaders.delete("Cookie")
targetHeaders.set("api-key", config.api_key)
targetHeaders.set("api-key", apiKey)
return new Request(targetUrl, {
method: request.method,
@@ -41,7 +30,7 @@ export default {
requestBody: any,
saveUsage: (usage: Usage) => Promise<void>,
): Promise<Response> {
const { model: deploymentName, stream } = requestBody;
const { stream } = requestBody;
if (stream) {
requestBody.stream_options = {
@@ -50,7 +39,7 @@ export default {
}
}
const proxyRequest = buildProxyRequest(c.req.raw, requestBody, config, deploymentName)
const proxyRequest = buildProxyRequest(c.req.raw, requestBody, config)
const response = await fetch(proxyRequest)
if (stream) {
+4 -11
View File
@@ -3,28 +3,21 @@ import {
handleStreamResponse,
checkoutResponsesUsageData,
} from "./shared/responses-stream-utils"
import { buildAzureTargetUrl } from "./shared/azure-target-url"
const buildProxyRequest = (
request: Request,
reqJson: any,
config: ChannelConfig
): Request => {
const url = new URL(request.url)
const targetUrl = new URL(config.endpoint)
if (!config.endpoint.endsWith('#')) {
targetUrl.pathname = `/openai${url.pathname}`
}
if (config.api_version) {
targetUrl.searchParams.set('api-version', config.api_version)
}
const targetUrl = buildAzureTargetUrl(request, config.endpoint)
const apiKey = config.api_key || ""
const targetHeaders = new Headers(request.headers)
targetHeaders.delete("Authorization")
targetHeaders.delete("Host")
targetHeaders.delete("Cookie")
targetHeaders.set("api-key", config.api_key)
targetHeaders.set("api-key", apiKey)
return new Request(targetUrl, {
method: request.method,
+5 -5
View File
@@ -1,4 +1,5 @@
import { Context } from "hono"
import { DEFAULT_CLAUDE_API_VERSION } from "../channel-config"
/**
* Claude API Proxy Provider
@@ -55,6 +56,7 @@ const buildProxyRequest = (
): Request => {
const url = new URL(request.url)
const targetUrl = new URL(config.endpoint)
const apiKey = config.api_key || ""
// Claude API uses /v1/messages endpoint
targetUrl.pathname = url.pathname
@@ -66,11 +68,9 @@ const buildProxyRequest = (
targetHeaders.delete("x-api-key")
// Claude uses x-api-key header instead of Authorization
targetHeaders.set("x-api-key", config.api_key)
targetHeaders.set("x-api-key", apiKey)
// Claude requires anthropic-version header
if (config.api_version) {
targetHeaders.set("anthropic-version", config.api_version)
}
targetHeaders.set("anthropic-version", DEFAULT_CLAUDE_API_VERSION)
return new Request(targetUrl, {
method: request.method,
@@ -91,7 +91,7 @@ const checkoutUsageData = async (
): Promise<void> => {
try {
const res = response instanceof Response
? await response.clone().json<ClaudeMessageResponse>()
? await response.clone().json() as ClaudeMessageResponse
: response
if (!res.usage) return;
+3 -2
View File
@@ -80,10 +80,11 @@ const buildProxyRequest = (
): Request => {
const targetUrl = new URL("v1/chat/completions", config.endpoint);
const targetHeaders = new Headers(request.headers);
const apiKey = config.api_key || "";
targetHeaders.delete("Authorization");
targetHeaders.delete("x-api-key");
targetHeaders.set("Authorization", `Bearer ${config.api_key}`);
targetHeaders.set("Authorization", `Bearer ${apiKey}`);
return new Request(targetUrl, {
method: request.method,
@@ -564,7 +565,7 @@ export default {
if (response.ok) {
try {
const resJson = await response.clone().json<OpenAIResponse & Record<string, any>>();
const resJson = await response.clone().json() as OpenAIResponse & Record<string, any>;
const usage = normalizeResponsesUsage(resJson?.usage);
if (usage) {
await saveUsage(usage);
+2 -1
View File
@@ -6,6 +6,7 @@ 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 { ModelsEndpoint } from "./models"
export const api = fromHono(new Hono<HonoCustomType>())
@@ -50,7 +51,7 @@ class UnifiedProxyEndpoint extends OpenAPIRoute {
return c.text("Channel type not supported", 400)
}
return provider(c, channel.config, requestBody, saveUsage)
return executeWithChannelKeys(c, channel.config, requestBody, saveUsage, provider)
}
}
+2 -1
View File
@@ -3,6 +3,7 @@ import { OpenAPIRoute } from "chanfana";
import { z } from "zod";
import { getApiKeyFromHeaders, fetchTokenData, fetchChannelsForToken } from "./shared/auth";
import { getSupportedModels } from "../utils";
import { normalizeChannelConfig } from "../channel-config";
export class ModelsEndpoint extends OpenAPIRoute {
schema = {
@@ -57,7 +58,7 @@ export class ModelsEndpoint extends OpenAPIRoute {
const modelsSet = new Set<string>();
for (const row of channelsResult.results) {
const config = JSON.parse(row.value) as ChannelConfig;
const config = normalizeChannelConfig(JSON.parse(row.value) as ChannelConfig);
for (const modelName of getSupportedModels(config)) {
modelsSet.add(modelName);
}
+2 -1
View File
@@ -8,13 +8,14 @@ const buildProxyRequest = (
): Request => {
const url = new URL(request.url)
const targetUrl = new URL(config.endpoint)
const apiKey = config.api_key || ""
targetUrl.pathname = url.pathname
const targetHeaders = new Headers(request.headers)
targetHeaders.delete("Host")
targetHeaders.delete("Cookie")
targetHeaders.set("Authorization", `Bearer ${config.api_key}`)
targetHeaders.set("Authorization", `Bearer ${apiKey}`)
return new Request(targetUrl, {
method: request.method,
+2 -1
View File
@@ -11,13 +11,14 @@ const buildProxyRequest = (
): Request => {
const url = new URL(request.url)
const targetUrl = new URL(config.endpoint)
const apiKey = config.api_key || ""
targetUrl.pathname = url.pathname
const targetHeaders = new Headers(request.headers)
targetHeaders.delete("Host")
targetHeaders.delete("Cookie")
targetHeaders.set("Authorization", `Bearer ${config.api_key}`)
targetHeaders.set("Authorization", `Bearer ${apiKey}`)
return new Request(targetUrl, {
method: request.method,
+33
View File
@@ -0,0 +1,33 @@
const trimSlashes = (value: string): string => {
return value.replace(/^\/+|\/+$/g, "");
};
const joinPath = (...segments: string[]): string => {
const cleanedSegments = segments
.map((segment) => trimSlashes(segment))
.filter((segment) => segment.length > 0);
return `/${cleanedSegments.join("/")}`;
};
export const buildAzureTargetUrl = (
request: Request,
endpoint: string
): URL => {
const requestUrl = new URL(request.url);
const targetUrl = new URL(endpoint);
if (endpoint.endsWith("#")) {
return targetUrl;
}
const requestPath = requestUrl.pathname.replace(/^\/v1/, "");
const currentBasePath = trimSlashes(targetUrl.pathname);
const azureBasePath = currentBasePath.endsWith("openai/v1")
? currentBasePath
: joinPath(currentBasePath, "openai/v1");
targetUrl.pathname = joinPath(azureBasePath, requestPath);
return targetUrl;
};
+8 -1
View File
@@ -3,6 +3,7 @@ import { getApiKeyFromHeaders, fetchTokenData, fetchChannelsForToken } from "./a
import { RouteId, getRoutePolicy } from "./route-policy"
import { findDeploymentMapping, findSupportedModel, getSupportedModels } from "../../utils"
import { TokenUtils } from "../../admin/token_utils"
import { normalizeChannelConfig } from "../../channel-config"
export type ChannelResolution = {
channel: { key: string; config: ChannelConfig }
@@ -53,7 +54,13 @@ export const resolveChannel = async (
const availableChannels: Array<{ key: string, config: ChannelConfig, mapping: { pattern: string, deployment: string } }> = [];
for (const row of channelsResult.results) {
const config = (() => { try { return JSON.parse(row.value) as ChannelConfig; } catch { return null; } })();
const config = (() => {
try {
return normalizeChannelConfig(JSON.parse(row.value) as ChannelConfig);
} catch {
return null;
}
})();
if (!config) {
console.error(`Invalid channel config for key: ${row.key}`);
continue;
+1 -1
View File
@@ -66,7 +66,7 @@ export const checkoutUsageData = async (
return
}
const res = await response.clone().json<OpenAIResponse>()
const res = await response.clone().json() as OpenAIResponse
if (!res.usage) return;
await saveUsage(res.usage)
return
+1 -1
View File
@@ -7,7 +7,7 @@ import claudeToOpenaiProxy from "../claude-to-openai-proxy"
import openaiResponsesProxy from "../openai-responses-proxy"
import azureOpenaiResponsesProxy from "../azure-openai-responses-proxy"
type ProviderFetch = (
export type ProviderFetch = (
c: Context<HonoCustomType>,
config: ChannelConfig,
requestBody: any,
@@ -101,7 +101,7 @@ export const checkoutResponsesUsageData = async (
requestBody: any,
): Promise<void> => {
try {
const resJson = await response.clone().json<OpenAIResponsesResponse>()
const resJson = await response.clone().json() as OpenAIResponsesResponse
const usage = extractUsageFromResponse(resJson)
if (usage) {
await saveUsage(usage)
+130
View File
@@ -0,0 +1,130 @@
import { Context } from "hono";
import {
MAX_RETRIES_PER_KEY,
MAX_ROTATION_ATTEMPTS,
normalizeChannelConfig,
} from "../../channel-config";
import { ProviderFetch } from "./provider-registry";
const RETRYABLE_STATUS_CODES = new Set([401, 403, 408, 409, 429, 500, 502, 503, 504, 529]);
const shuffleKeys = (keys: string[]): string[] => {
const clonedKeys = [...keys];
for (let index = clonedKeys.length - 1; index > 0; index -= 1) {
const randomIndex = Math.floor(Math.random() * (index + 1));
[clonedKeys[index], clonedKeys[randomIndex]] = [clonedKeys[randomIndex], clonedKeys[index]];
}
return clonedKeys;
};
const cloneRequestBody = <T>(requestBody: T): T => {
if (requestBody == null) {
return requestBody;
}
return JSON.parse(JSON.stringify(requestBody)) as T;
};
const shouldRetryResponse = (response: Response): boolean => {
return RETRYABLE_STATUS_CODES.has(response.status);
};
const discardResponse = async (response: Response): Promise<void> => {
try {
await response.body?.cancel();
} catch (error) {
console.warn("Failed to cancel upstream response body:", error);
}
};
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));
};
export const executeWithChannelKeys = async (
c: Context<HonoCustomType>,
config: ChannelConfig,
requestBody: any,
saveUsage: (usage: Usage) => Promise<void>,
provider: ProviderFetch,
): Promise<Response> => {
const normalizedConfig = normalizeChannelConfig(config);
const candidateKeys = getCandidateKeys(normalizedConfig);
if (candidateKeys.length === 0) {
return c.text("Channel API keys not configured", 500);
}
const attemptsPerKey = normalizedConfig.auto_retry
? 1 + MAX_RETRIES_PER_KEY
: 1;
for (let keyIndex = 0; keyIndex < candidateKeys.length; keyIndex += 1) {
const currentKey = candidateKeys[keyIndex];
for (let attemptIndex = 0; attemptIndex < attemptsPerKey; attemptIndex += 1) {
const isLastAttemptForKey = attemptIndex === attemptsPerKey - 1;
const isLastKey = keyIndex === candidateKeys.length - 1;
try {
const runtimeConfig: ChannelConfig = {
...normalizedConfig,
api_key: currentKey,
api_keys: [currentKey],
};
const response = await provider(
c,
runtimeConfig,
cloneRequestBody(requestBody),
saveUsage,
);
if (response.ok) {
return response;
}
if (!shouldRetryResponse(response)) {
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) {
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) {
const message = error instanceof Error ? error.message : "Unknown upstream error";
return c.text(`Upstream request failed: ${message}`, 502);
}
}
}
}
return c.text("Upstream request failed after retries", 502);
};