refactor(worker): upstream-first proxy with dynamic response adapters
- remove retry rounds/backoff toggles and keep deterministic upstream selection - add chat response adapter registry for openai/anthropic/gemini json+sse transforms - reduce non-essential request/proxy logs and keep failure summary logs - keep only PROXY_UPSTREAM_TIMEOUT_MS env var and update changelog
This commit is contained in:
1 parent
1e1f23d464
commit
e1c7394bf0
6 files changed
+1580
-323
No files matched your search
@@ -6,8 +6,7 @@ import type {
|
||||
export type Bindings = {
|
||||
DB: D1Database;
|
||||
CORS_ORIGIN?: string;
|
||||
PROXY_RETRY_ROUNDS?: string;
|
||||
PROXY_RETRY_DELAY_MS?: string;
|
||||
PROXY_UPSTREAM_TIMEOUT_MS?: string;
|
||||
CHECKIN_SCHEDULER: DurableObjectNamespace;
|
||||
};
|
||||
|
||||
@@ -21,3 +20,4 @@ export type AppEnv = {
|
||||
Bindings: Bindings;
|
||||
Variables: Variables;
|
||||
};
|
||||
|
||||
+22
-41
@@ -1,6 +1,5 @@
|
||||
import { Hono } from "hono";
|
||||
import { cors } from "hono/cors";
|
||||
import { logger } from "hono/logger";
|
||||
import type { AppEnv } from "./env";
|
||||
import { adminAuth } from "./middleware/adminAuth";
|
||||
import authRoutes from "./routes/auth";
|
||||
@@ -17,46 +16,6 @@ import tokenRoutes from "./routes/tokens";
|
||||
import usageRoutes from "./routes/usage";
|
||||
|
||||
const app = new Hono<AppEnv>({ strict: false });
|
||||
|
||||
app.use("*", async (c, next) => {
|
||||
const method = c.req.method;
|
||||
const path = c.req.path;
|
||||
const contentType = c.req.header("content-type") ?? "";
|
||||
if (["POST", "PUT", "PATCH", "DELETE"].includes(method)) {
|
||||
const raw = c.req.raw.clone();
|
||||
let bodyType = "unknown";
|
||||
let bodyKeys: string[] = [];
|
||||
let bodySize = 0;
|
||||
try {
|
||||
const payload = await raw.json();
|
||||
bodyType = "json";
|
||||
if (payload && typeof payload === "object" && !Array.isArray(payload)) {
|
||||
bodyKeys = Object.keys(payload);
|
||||
}
|
||||
} catch {
|
||||
try {
|
||||
const text = await raw.text();
|
||||
bodyType = "text";
|
||||
bodySize = text.length;
|
||||
} catch {
|
||||
bodyType = "unreadable";
|
||||
}
|
||||
}
|
||||
console.log("[request]", {
|
||||
method,
|
||||
path,
|
||||
content_type: contentType,
|
||||
body_type: bodyType,
|
||||
body_keys: bodyKeys,
|
||||
body_size: bodySize,
|
||||
});
|
||||
} else {
|
||||
console.log("[request]", { method, path });
|
||||
}
|
||||
await next();
|
||||
});
|
||||
|
||||
app.use("*", logger());
|
||||
app.use(
|
||||
"/api/*",
|
||||
cors({
|
||||
@@ -122,6 +81,28 @@ app.route("/api/group", newapiGroupRoutes);
|
||||
app.route("/v1", proxyRoutes);
|
||||
app.route("/v1beta", proxyRoutes);
|
||||
|
||||
app.onError((err, c) => {
|
||||
console.error("[app:error]", {
|
||||
method: c.req.method,
|
||||
path: c.req.path,
|
||||
message: err.message,
|
||||
stack: err.stack,
|
||||
});
|
||||
|
||||
if (
|
||||
c.req.path === "/api" ||
|
||||
c.req.path.startsWith("/api/") ||
|
||||
c.req.path === "/v1" ||
|
||||
c.req.path.startsWith("/v1/") ||
|
||||
c.req.path === "/v1beta" ||
|
||||
c.req.path.startsWith("/v1beta/")
|
||||
) {
|
||||
return c.json({ error: "Internal Server Error" }, 500);
|
||||
}
|
||||
|
||||
return c.text("Internal Server Error", 500);
|
||||
});
|
||||
|
||||
app.notFound(async (c) => {
|
||||
const path = c.req.path;
|
||||
if (
|
||||
|
||||
+333
-278
@@ -7,6 +7,7 @@ import {
|
||||
} from "../services/call-token-selector";
|
||||
import { listCallTokens } from "../services/channel-call-token-repo";
|
||||
import {
|
||||
type ChannelMetadata,
|
||||
parseChannelMetadata,
|
||||
resolveMappedModel,
|
||||
resolveProvider,
|
||||
@@ -44,6 +45,7 @@ import {
|
||||
parseUsageFromJson,
|
||||
parseUsageFromSse,
|
||||
} from "../utils/usage";
|
||||
import { adaptChatResponse } from "../services/chat-response-adapter";
|
||||
|
||||
const proxy = new Hono<AppEnv>();
|
||||
|
||||
@@ -66,6 +68,54 @@ function channelSupportsModel(
|
||||
return Boolean(metadata.model_mapping[model] ?? metadata.model_mapping["*"]);
|
||||
}
|
||||
|
||||
export function selectCandidateChannels(
|
||||
allowedChannels: ChannelRecord[],
|
||||
downstreamModel: string | null,
|
||||
): ChannelRecord[] {
|
||||
const modelChannels = allowedChannels.filter((channel) =>
|
||||
channelSupportsModel(channel, downstreamModel),
|
||||
);
|
||||
if (modelChannels.length > 0) {
|
||||
return modelChannels;
|
||||
}
|
||||
// Fallback to all allowed channels. Downstream protocol should not restrict
|
||||
// upstream channel selection.
|
||||
return allowedChannels;
|
||||
}
|
||||
|
||||
function hasExplicitModelMapping(
|
||||
metadata: ChannelMetadata,
|
||||
downstreamModel: string | null,
|
||||
): boolean {
|
||||
if (downstreamModel) {
|
||||
return (
|
||||
metadata.model_mapping[downstreamModel] !== undefined ||
|
||||
metadata.model_mapping["*"] !== undefined
|
||||
);
|
||||
}
|
||||
return metadata.model_mapping["*"] !== undefined;
|
||||
}
|
||||
|
||||
export function resolveUpstreamModelForChannel(
|
||||
channel: ChannelRecord,
|
||||
metadata: ChannelMetadata,
|
||||
downstreamModel: string | null,
|
||||
): { model: string | null; autoMapped: boolean } {
|
||||
const mapped = resolveMappedModel(metadata.model_mapping, downstreamModel);
|
||||
if (!downstreamModel || hasExplicitModelMapping(metadata, downstreamModel)) {
|
||||
return { model: mapped, autoMapped: false };
|
||||
}
|
||||
|
||||
const declaredModels = extractModels(channel).map((entry) => entry.id);
|
||||
if (declaredModels.length === 0) {
|
||||
return { model: mapped, autoMapped: false };
|
||||
}
|
||||
if (declaredModels.includes(downstreamModel)) {
|
||||
return { model: downstreamModel, autoMapped: false };
|
||||
}
|
||||
return { model: declaredModels[0] ?? mapped, autoMapped: true };
|
||||
}
|
||||
|
||||
function filterAllowedChannels(
|
||||
channels: ChannelRecord[],
|
||||
tokenRecord: TokenRecord,
|
||||
@@ -81,35 +131,6 @@ function filterAllowedChannels(
|
||||
return channels.filter((channel) => allowedSet.has(channel.id));
|
||||
}
|
||||
|
||||
/**
|
||||
* Determines whether a response status should be retried.
|
||||
*
|
||||
* Args:
|
||||
* status: HTTP response status code.
|
||||
*
|
||||
* Returns:
|
||||
* True if the status is retryable.
|
||||
*/
|
||||
function isRetryableStatus(status: number): boolean {
|
||||
return status === 408 || status === 429 || status >= 500;
|
||||
}
|
||||
|
||||
/**
|
||||
* Waits before the next retry round.
|
||||
*
|
||||
* Args:
|
||||
* ms: Delay in milliseconds.
|
||||
*
|
||||
* Returns:
|
||||
* Promise resolved after delay.
|
||||
*/
|
||||
async function sleep(ms: number): Promise<void> {
|
||||
if (ms <= 0) {
|
||||
return;
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
function resolveChannelBaseUrl(channel: ChannelRecord): string {
|
||||
return normalizeBaseUrl(channel.base_url);
|
||||
}
|
||||
@@ -163,6 +184,26 @@ function buildUpstreamHeaders(
|
||||
return headers;
|
||||
}
|
||||
|
||||
async function fetchWithTimeout(
|
||||
url: string,
|
||||
init: RequestInit,
|
||||
timeoutMs: number,
|
||||
): Promise<Response> {
|
||||
if (timeoutMs <= 0) {
|
||||
return fetch(url, init);
|
||||
}
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), timeoutMs);
|
||||
try {
|
||||
return await fetch(url, {
|
||||
...init,
|
||||
signal: controller.signal,
|
||||
});
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Multi-provider proxy handler.
|
||||
*/
|
||||
@@ -254,11 +295,16 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
callTokenMap.set(row.channel_id, list);
|
||||
}
|
||||
const allowedChannels = filterAllowedChannels(activeChannels, tokenRecord);
|
||||
const modelChannels = allowedChannels.filter((channel) =>
|
||||
channelSupportsModel(channel, downstreamModel),
|
||||
);
|
||||
const candidates = modelChannels.length > 0 ? modelChannels : allowedChannels;
|
||||
const candidates = selectCandidateChannels(allowedChannels, downstreamModel);
|
||||
|
||||
if (candidates.length === 0 && allowedChannels.length > 0) {
|
||||
console.warn("[proxy:no_compatible_channels]", {
|
||||
path: c.req.path,
|
||||
model: downstreamModel,
|
||||
downstream_provider: downstreamProvider,
|
||||
allowed_channels: allowedChannels.length,
|
||||
});
|
||||
}
|
||||
if (candidates.length === 0) {
|
||||
return jsonError(c, 503, "no_available_channels", "no_available_channels");
|
||||
}
|
||||
@@ -268,148 +314,80 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
const querySuffix = c.req.url.includes("?")
|
||||
? `?${c.req.url.split("?")[1]}`
|
||||
: "";
|
||||
const retryRounds = Math.max(1, Number(c.env.PROXY_RETRY_ROUNDS ?? "1"));
|
||||
const retryDelayMs = Math.max(0, Number(c.env.PROXY_RETRY_DELAY_MS ?? "200"));
|
||||
const upstreamTimeoutMs = Math.max(
|
||||
1000,
|
||||
Number(c.env.PROXY_UPSTREAM_TIMEOUT_MS ?? "30000"),
|
||||
);
|
||||
let lastResponse: Response | null = null;
|
||||
let lastChannel: ChannelRecord | null = null;
|
||||
let lastRequestPath = targetPath;
|
||||
const start = Date.now();
|
||||
let selectedChannel: ChannelRecord | null = null;
|
||||
|
||||
let round = 0;
|
||||
while (round < retryRounds && !selectedChannel) {
|
||||
let shouldRetry = false;
|
||||
for (const channel of ordered) {
|
||||
lastChannel = channel;
|
||||
const metadata = parseChannelMetadata(channel.metadata_json);
|
||||
const upstreamProvider = resolveProvider(metadata.site_type);
|
||||
const upstreamModel = resolveMappedModel(
|
||||
metadata.model_mapping,
|
||||
downstreamModel,
|
||||
);
|
||||
if (
|
||||
upstreamProvider === "gemini" &&
|
||||
!upstreamModel &&
|
||||
endpointType !== "passthrough"
|
||||
) {
|
||||
let selectedUpstreamProvider: ProviderType | null = null;
|
||||
let selectedUpstreamModel: string | null = null;
|
||||
for (const channel of ordered) {
|
||||
lastChannel = channel;
|
||||
const metadata = parseChannelMetadata(channel.metadata_json);
|
||||
const upstreamProvider = resolveProvider(metadata.site_type);
|
||||
const resolvedModel = resolveUpstreamModelForChannel(
|
||||
channel,
|
||||
metadata,
|
||||
downstreamModel,
|
||||
);
|
||||
const upstreamModel = resolvedModel.model;
|
||||
if (
|
||||
upstreamProvider === "gemini" &&
|
||||
!upstreamModel &&
|
||||
endpointType !== "passthrough"
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
const baseUrl = resolveChannelBaseUrl(channel);
|
||||
const tokens = callTokenMap.get(channel.id) ?? [];
|
||||
const selectedToken = tokens.length > 0 ? selectCallToken(tokens) : null;
|
||||
const apiKey = selectedToken?.api_key ?? channel.api_key;
|
||||
const headers = buildUpstreamHeaders(
|
||||
new Headers(c.req.header()),
|
||||
upstreamProvider,
|
||||
String(apiKey),
|
||||
metadata.header_overrides,
|
||||
);
|
||||
headers.delete("host");
|
||||
headers.delete("content-length");
|
||||
let upstreamRequestPath = targetPath;
|
||||
let upstreamFallbackPath: string | undefined;
|
||||
let upstreamBodyText = requestText || undefined;
|
||||
let absoluteUrl: string | undefined;
|
||||
const sameProvider = upstreamProvider === downstreamProvider;
|
||||
if (endpointType === "passthrough") {
|
||||
if (!sameProvider) {
|
||||
continue;
|
||||
}
|
||||
const baseUrl = resolveChannelBaseUrl(channel);
|
||||
const tokens = callTokenMap.get(channel.id) ?? [];
|
||||
const selectedToken = tokens.length > 0 ? selectCallToken(tokens) : null;
|
||||
const apiKey = selectedToken?.api_key ?? channel.api_key;
|
||||
const headers = buildUpstreamHeaders(
|
||||
new Headers(c.req.header()),
|
||||
upstreamProvider,
|
||||
String(apiKey),
|
||||
metadata.header_overrides,
|
||||
);
|
||||
headers.delete("host");
|
||||
headers.delete("content-length");
|
||||
let upstreamRequestPath = targetPath;
|
||||
let upstreamFallbackPath: string | undefined;
|
||||
let upstreamBodyText = requestText || undefined;
|
||||
let absoluteUrl: string | undefined;
|
||||
const sameProvider = upstreamProvider === downstreamProvider;
|
||||
if (endpointType === "passthrough") {
|
||||
if (!sameProvider) {
|
||||
continue;
|
||||
}
|
||||
if (upstreamProvider === "gemini") {
|
||||
upstreamRequestPath = applyGeminiModelToPath(
|
||||
upstreamRequestPath,
|
||||
upstreamModel,
|
||||
);
|
||||
} else if (upstreamModel && parsedBody) {
|
||||
upstreamBodyText = JSON.stringify({
|
||||
...parsedBody,
|
||||
model: upstreamModel,
|
||||
});
|
||||
}
|
||||
} else if (sameProvider && parsedBody) {
|
||||
if (upstreamProvider === "gemini") {
|
||||
upstreamRequestPath = applyGeminiModelToPath(
|
||||
upstreamRequestPath,
|
||||
upstreamModel,
|
||||
);
|
||||
} else if (upstreamModel) {
|
||||
upstreamBodyText = JSON.stringify({
|
||||
...parsedBody,
|
||||
model: upstreamModel,
|
||||
});
|
||||
}
|
||||
if (endpointType === "chat" || endpointType === "responses") {
|
||||
if (metadata.endpoint_overrides.chat_url && normalizedChat) {
|
||||
const request = buildUpstreamChatRequest(
|
||||
upstreamProvider,
|
||||
normalizedChat,
|
||||
upstreamModel,
|
||||
endpointType,
|
||||
isStream,
|
||||
metadata.endpoint_overrides,
|
||||
);
|
||||
if (request) {
|
||||
upstreamRequestPath = request.path;
|
||||
absoluteUrl = request.absoluteUrl;
|
||||
upstreamFallbackPath = request.absoluteUrl
|
||||
? undefined
|
||||
: request.fallbackPath;
|
||||
}
|
||||
} else if (
|
||||
endpointType === "responses" &&
|
||||
upstreamProvider === "openai"
|
||||
) {
|
||||
upstreamFallbackPath = "/responses";
|
||||
}
|
||||
}
|
||||
if (
|
||||
endpointType === "embeddings" &&
|
||||
metadata.endpoint_overrides.embedding_url
|
||||
) {
|
||||
if (normalizedEmbedding) {
|
||||
const request = buildUpstreamEmbeddingRequest(
|
||||
upstreamProvider,
|
||||
normalizedEmbedding,
|
||||
upstreamModel,
|
||||
metadata.endpoint_overrides,
|
||||
);
|
||||
if (request) {
|
||||
upstreamRequestPath = request.path;
|
||||
absoluteUrl = request.absoluteUrl;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (
|
||||
endpointType === "images" &&
|
||||
metadata.endpoint_overrides.image_url
|
||||
) {
|
||||
if (normalizedImage) {
|
||||
const request = buildUpstreamImageRequest(
|
||||
upstreamProvider,
|
||||
normalizedImage,
|
||||
upstreamModel,
|
||||
metadata.endpoint_overrides,
|
||||
);
|
||||
if (request) {
|
||||
upstreamRequestPath = request.path;
|
||||
absoluteUrl = request.absoluteUrl;
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
let built: {
|
||||
request: {
|
||||
path: string;
|
||||
fallbackPath?: string;
|
||||
absoluteUrl?: string;
|
||||
body: Record<string, unknown> | null;
|
||||
};
|
||||
bodyText?: string;
|
||||
} | null = null;
|
||||
if (endpointType === "chat" || endpointType === "responses") {
|
||||
if (!normalizedChat) {
|
||||
return jsonError(c, 400, "invalid_body", "invalid_body");
|
||||
}
|
||||
if (upstreamProvider === "gemini") {
|
||||
upstreamRequestPath = applyGeminiModelToPath(
|
||||
upstreamRequestPath,
|
||||
upstreamModel,
|
||||
);
|
||||
} else if (upstreamModel && parsedBody) {
|
||||
upstreamBodyText = JSON.stringify({
|
||||
...parsedBody,
|
||||
model: upstreamModel,
|
||||
});
|
||||
}
|
||||
} else if (sameProvider && parsedBody) {
|
||||
if (upstreamProvider === "gemini") {
|
||||
upstreamRequestPath = applyGeminiModelToPath(
|
||||
upstreamRequestPath,
|
||||
upstreamModel,
|
||||
);
|
||||
} else if (upstreamModel) {
|
||||
upstreamBodyText = JSON.stringify({
|
||||
...parsedBody,
|
||||
model: upstreamModel,
|
||||
});
|
||||
}
|
||||
if (endpointType === "chat" || endpointType === "responses") {
|
||||
if (metadata.endpoint_overrides.chat_url && normalizedChat) {
|
||||
const request = buildUpstreamChatRequest(
|
||||
upstreamProvider,
|
||||
normalizedChat,
|
||||
@@ -418,121 +396,214 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
isStream,
|
||||
metadata.endpoint_overrides,
|
||||
);
|
||||
if (!request) {
|
||||
continue;
|
||||
}
|
||||
built = {
|
||||
request,
|
||||
bodyText: request.body ? JSON.stringify(request.body) : undefined,
|
||||
};
|
||||
} else if (endpointType === "embeddings") {
|
||||
if (!normalizedEmbedding) {
|
||||
return jsonError(c, 400, "invalid_body", "invalid_body");
|
||||
if (request) {
|
||||
upstreamRequestPath = request.path;
|
||||
absoluteUrl = request.absoluteUrl;
|
||||
upstreamFallbackPath = request.absoluteUrl
|
||||
? undefined
|
||||
: request.fallbackPath;
|
||||
}
|
||||
} else if (
|
||||
endpointType === "responses" &&
|
||||
upstreamProvider === "openai"
|
||||
) {
|
||||
upstreamFallbackPath = "/responses";
|
||||
}
|
||||
}
|
||||
if (
|
||||
endpointType === "embeddings" &&
|
||||
metadata.endpoint_overrides.embedding_url
|
||||
) {
|
||||
if (normalizedEmbedding) {
|
||||
const request = buildUpstreamEmbeddingRequest(
|
||||
upstreamProvider,
|
||||
normalizedEmbedding,
|
||||
upstreamModel,
|
||||
metadata.endpoint_overrides,
|
||||
);
|
||||
if (!request) {
|
||||
continue;
|
||||
}
|
||||
built = {
|
||||
request,
|
||||
bodyText: request.body ? JSON.stringify(request.body) : undefined,
|
||||
};
|
||||
} else if (endpointType === "images") {
|
||||
if (!normalizedImage) {
|
||||
return jsonError(c, 400, "invalid_body", "invalid_body");
|
||||
if (request) {
|
||||
upstreamRequestPath = request.path;
|
||||
absoluteUrl = request.absoluteUrl;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (
|
||||
endpointType === "images" &&
|
||||
metadata.endpoint_overrides.image_url
|
||||
) {
|
||||
if (normalizedImage) {
|
||||
const request = buildUpstreamImageRequest(
|
||||
upstreamProvider,
|
||||
normalizedImage,
|
||||
upstreamModel,
|
||||
metadata.endpoint_overrides,
|
||||
);
|
||||
if (!request) {
|
||||
continue;
|
||||
if (request) {
|
||||
upstreamRequestPath = request.path;
|
||||
absoluteUrl = request.absoluteUrl;
|
||||
}
|
||||
built = {
|
||||
request,
|
||||
bodyText: request.body ? JSON.stringify(request.body) : undefined,
|
||||
};
|
||||
}
|
||||
if (!built) {
|
||||
}
|
||||
} else {
|
||||
let built: {
|
||||
request: {
|
||||
path: string;
|
||||
fallbackPath?: string;
|
||||
absoluteUrl?: string;
|
||||
body: Record<string, unknown> | null;
|
||||
};
|
||||
bodyText?: string;
|
||||
} | null = null;
|
||||
if (endpointType === "chat" || endpointType === "responses") {
|
||||
if (!normalizedChat) {
|
||||
return jsonError(c, 400, "invalid_body", "invalid_body");
|
||||
}
|
||||
const request = buildUpstreamChatRequest(
|
||||
upstreamProvider,
|
||||
normalizedChat,
|
||||
upstreamModel,
|
||||
endpointType,
|
||||
isStream,
|
||||
metadata.endpoint_overrides,
|
||||
);
|
||||
if (!request) {
|
||||
continue;
|
||||
}
|
||||
upstreamRequestPath = built.request.path;
|
||||
absoluteUrl = built.request.absoluteUrl;
|
||||
upstreamFallbackPath = built.request.absoluteUrl
|
||||
? undefined
|
||||
: built.request.fallbackPath;
|
||||
upstreamBodyText = built.bodyText;
|
||||
built = {
|
||||
request,
|
||||
bodyText: request.body ? JSON.stringify(request.body) : undefined,
|
||||
};
|
||||
} else if (endpointType === "embeddings") {
|
||||
if (!normalizedEmbedding) {
|
||||
return jsonError(c, 400, "invalid_body", "invalid_body");
|
||||
}
|
||||
const request = buildUpstreamEmbeddingRequest(
|
||||
upstreamProvider,
|
||||
normalizedEmbedding,
|
||||
upstreamModel,
|
||||
metadata.endpoint_overrides,
|
||||
);
|
||||
if (!request) {
|
||||
continue;
|
||||
}
|
||||
built = {
|
||||
request,
|
||||
bodyText: request.body ? JSON.stringify(request.body) : undefined,
|
||||
};
|
||||
} else if (endpointType === "images") {
|
||||
if (!normalizedImage) {
|
||||
return jsonError(c, 400, "invalid_body", "invalid_body");
|
||||
}
|
||||
const request = buildUpstreamImageRequest(
|
||||
upstreamProvider,
|
||||
normalizedImage,
|
||||
upstreamModel,
|
||||
metadata.endpoint_overrides,
|
||||
);
|
||||
if (!request) {
|
||||
continue;
|
||||
}
|
||||
built = {
|
||||
request,
|
||||
bodyText: request.body ? JSON.stringify(request.body) : undefined,
|
||||
};
|
||||
}
|
||||
const targetBase = absoluteUrl ?? `${baseUrl}${upstreamRequestPath}`;
|
||||
const target = mergeQuery(
|
||||
targetBase,
|
||||
querySuffix,
|
||||
metadata.query_overrides,
|
||||
);
|
||||
if (!built) {
|
||||
continue;
|
||||
}
|
||||
upstreamRequestPath = built.request.path;
|
||||
absoluteUrl = built.request.absoluteUrl;
|
||||
upstreamFallbackPath = built.request.absoluteUrl
|
||||
? undefined
|
||||
: built.request.fallbackPath;
|
||||
upstreamBodyText = built.bodyText;
|
||||
}
|
||||
const targetBase = absoluteUrl ?? `${baseUrl}${upstreamRequestPath}`;
|
||||
const target = mergeQuery(
|
||||
targetBase,
|
||||
querySuffix,
|
||||
metadata.query_overrides,
|
||||
);
|
||||
|
||||
try {
|
||||
let response = await fetch(target, {
|
||||
try {
|
||||
let response = await fetchWithTimeout(
|
||||
target,
|
||||
{
|
||||
method: c.req.method,
|
||||
headers,
|
||||
body: upstreamBodyText || undefined,
|
||||
});
|
||||
let responsePath = upstreamRequestPath;
|
||||
if (
|
||||
(response.status === 400 || response.status === 404) &&
|
||||
upstreamFallbackPath
|
||||
) {
|
||||
const fallbackTargetBase = absoluteUrl
|
||||
? absoluteUrl
|
||||
: `${baseUrl}${upstreamFallbackPath}`;
|
||||
const fallbackTarget = mergeQuery(
|
||||
fallbackTargetBase,
|
||||
querySuffix,
|
||||
metadata.query_overrides,
|
||||
);
|
||||
response = await fetch(fallbackTarget, {
|
||||
},
|
||||
upstreamTimeoutMs,
|
||||
);
|
||||
let responsePath = upstreamRequestPath;
|
||||
if (
|
||||
(response.status === 400 || response.status === 404) &&
|
||||
upstreamFallbackPath
|
||||
) {
|
||||
const fallbackTargetBase = absoluteUrl
|
||||
? absoluteUrl
|
||||
: `${baseUrl}${upstreamFallbackPath}`;
|
||||
const fallbackTarget = mergeQuery(
|
||||
fallbackTargetBase,
|
||||
querySuffix,
|
||||
metadata.query_overrides,
|
||||
);
|
||||
response = await fetchWithTimeout(
|
||||
fallbackTarget,
|
||||
{
|
||||
method: c.req.method,
|
||||
headers,
|
||||
body: upstreamBodyText || undefined,
|
||||
});
|
||||
responsePath = upstreamFallbackPath;
|
||||
}
|
||||
|
||||
lastResponse = response;
|
||||
lastRequestPath = responsePath;
|
||||
if (response.ok) {
|
||||
selectedChannel = channel;
|
||||
break;
|
||||
}
|
||||
|
||||
if (isRetryableStatus(response.status)) {
|
||||
shouldRetry = true;
|
||||
}
|
||||
} catch {
|
||||
lastResponse = null;
|
||||
shouldRetry = true;
|
||||
},
|
||||
upstreamTimeoutMs,
|
||||
);
|
||||
responsePath = upstreamFallbackPath;
|
||||
}
|
||||
}
|
||||
|
||||
if (selectedChannel || !shouldRetry) {
|
||||
break;
|
||||
}
|
||||
|
||||
round += 1;
|
||||
if (round < retryRounds) {
|
||||
await sleep(retryDelayMs);
|
||||
lastResponse = response;
|
||||
lastRequestPath = responsePath;
|
||||
if (response.ok) {
|
||||
selectedChannel = channel;
|
||||
selectedUpstreamProvider = upstreamProvider;
|
||||
selectedUpstreamModel = upstreamModel;
|
||||
break;
|
||||
}
|
||||
} catch (error) {
|
||||
const isTimeout =
|
||||
error instanceof Error &&
|
||||
(error.name === "AbortError" || error.message.includes("upstream_timeout"));
|
||||
console.error("[proxy:upstream_exception]", {
|
||||
channel_id: channel.id,
|
||||
upstream_provider: upstreamProvider,
|
||||
path: upstreamRequestPath,
|
||||
model: downstreamModel,
|
||||
upstream_model: upstreamModel,
|
||||
timeout_ms: upstreamTimeoutMs,
|
||||
reason: isTimeout ? "timeout" : "exception",
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
lastResponse = null;
|
||||
}
|
||||
}
|
||||
|
||||
const latencyMs = Date.now() - start;
|
||||
if (!selectedChannel && lastResponse && !lastResponse.ok) {
|
||||
console.warn("[proxy:upstream_exhausted]", {
|
||||
path: targetPath,
|
||||
model: downstreamModel,
|
||||
status: lastResponse.status,
|
||||
last_channel_id: lastChannel?.id ?? null,
|
||||
last_request_path: lastRequestPath,
|
||||
});
|
||||
}
|
||||
|
||||
if (!lastResponse) {
|
||||
console.error("[proxy:unavailable]", {
|
||||
path: targetPath,
|
||||
model: downstreamModel,
|
||||
latency_ms: latencyMs,
|
||||
last_channel_id: lastChannel?.id ?? null,
|
||||
});
|
||||
await recordUsage(c.env.DB, {
|
||||
tokenId: tokenRecord.id,
|
||||
model: downstreamModel,
|
||||
@@ -576,23 +647,6 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
status: lastResponse.ok ? "ok" : "error",
|
||||
});
|
||||
};
|
||||
const logUsage = (
|
||||
label: string,
|
||||
usage: NormalizedUsage | null,
|
||||
source: string,
|
||||
) => {
|
||||
console.log(`[usage] ${label}`, {
|
||||
source,
|
||||
total_tokens: usage?.totalTokens ?? 0,
|
||||
prompt_tokens: usage?.promptTokens ?? 0,
|
||||
completion_tokens: usage?.completionTokens ?? 0,
|
||||
stream: isStream,
|
||||
status: lastResponse.status,
|
||||
model: downstreamModel,
|
||||
path: targetPath,
|
||||
});
|
||||
};
|
||||
|
||||
const headerUsage = parseUsageFromHeaders(lastResponse.headers);
|
||||
let jsonUsage: NormalizedUsage | null = null;
|
||||
if (
|
||||
@@ -607,11 +661,6 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
jsonUsage = parseUsageFromJson(data);
|
||||
}
|
||||
const immediateUsage = jsonUsage ?? headerUsage;
|
||||
const immediateSource = jsonUsage
|
||||
? "json"
|
||||
: headerUsage
|
||||
? "header"
|
||||
: "none";
|
||||
|
||||
if (isStream) {
|
||||
const executionCtx = (c as { executionCtx?: ExecutionContextLike })
|
||||
@@ -619,12 +668,6 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
const task = parseUsageFromSse(lastResponse.clone())
|
||||
.then((streamUsage) => {
|
||||
const usageValue = immediateUsage ?? streamUsage.usage;
|
||||
const source = immediateUsage
|
||||
? immediateSource
|
||||
: streamUsage.usage
|
||||
? "sse"
|
||||
: "sse-none";
|
||||
logUsage("stream", usageValue, source);
|
||||
return record(usageValue, streamUsage.firstTokenLatencyMs);
|
||||
})
|
||||
.catch(() => undefined);
|
||||
@@ -634,11 +677,23 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
task.catch(() => undefined);
|
||||
}
|
||||
} else {
|
||||
logUsage("immediate", immediateUsage, immediateSource);
|
||||
await record(immediateUsage, latencyMs);
|
||||
}
|
||||
}
|
||||
|
||||
if (selectedUpstreamProvider && endpointType === "chat") {
|
||||
const transformed = await adaptChatResponse({
|
||||
response: lastResponse,
|
||||
upstreamProvider: selectedUpstreamProvider,
|
||||
downstreamProvider,
|
||||
model: selectedUpstreamModel ?? downstreamModel,
|
||||
isStream,
|
||||
});
|
||||
if (transformed !== lastResponse) {
|
||||
return transformed;
|
||||
}
|
||||
}
|
||||
|
||||
return lastResponse;
|
||||
});
|
||||
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
@@ -4,8 +4,7 @@ compatibility_date = "2026-02-14"
|
||||
|
||||
[vars]
|
||||
CORS_ORIGIN = "http://localhost:5173"
|
||||
PROXY_RETRY_ROUNDS = "2"
|
||||
PROXY_RETRY_DELAY_MS = "200"
|
||||
PROXY_UPSTREAM_TIMEOUT_MS = "30000"
|
||||
|
||||
[[d1_databases]]
|
||||
binding = "DB"
|
||||
|
||||
@@ -17,6 +17,30 @@
|
||||
|
||||
### 微调
|
||||
|
||||
- **[worker/proxy]**: 请求日志改为安全采样并补充上游失败异常日志,便于定位 Claude 请求故障
|
||||
- 类型: 微调(无方案包)
|
||||
- 文件: apps/worker/src/index.ts, apps/worker/src/routes/proxy.ts
|
||||
- **[worker/proxy]**: 上游请求新增超时中断(默认 30s),避免代理请求卡住“有进没出”
|
||||
- 类型: 微调(无方案包)
|
||||
- 文件: apps/worker/src/routes/proxy.ts, apps/worker/src/env.ts, apps/worker/wrangler.toml
|
||||
- **[worker/proxy]**: 候选通道改为“模型匹配优先、同提供商次优、全通道兜底”并补充筛选日志
|
||||
- 类型: 微调(无方案包)
|
||||
- 文件: apps/worker/src/routes/proxy.ts, tests/worker/channel-metadata.test.ts
|
||||
- **[worker/proxy]**: 跨提供商且无显式模型映射时自动选择通道已声明模型,提升上下游互转可用性
|
||||
- 类型: 微调(无方案包)
|
||||
- 文件: apps/worker/src/routes/proxy.ts, tests/worker/channel-metadata.test.ts
|
||||
- **[worker/proxy]**: 自动模型兜底改为学习型复用(成功后记忆映射),并增加失败通道退避与 `PROXY_RETRY_ON_429` 开关降低长耗时重试
|
||||
- 类型: 微调(无方案包)
|
||||
- 文件: apps/worker/src/routes/proxy.ts, apps/worker/src/env.ts, apps/worker/wrangler.toml, tests/worker/channel-metadata.test.ts
|
||||
- **[worker/proxy]**: 通道选择不再受下游协议约束,并新增 OpenAI→Anthropic 响应转换(含 SSE)修复上游成功但客户端无输出
|
||||
- 类型: 微调(无方案包)
|
||||
- 文件: apps/worker/src/routes/proxy.ts, tests/worker/channel-metadata.test.ts
|
||||
- **[worker/proxy]**: 按“以上游为准”收敛,移除学习映射/失败退避/429重试开关,保留确定性转换与超时保护
|
||||
- 类型: 微调(无方案包)
|
||||
- 文件: apps/worker/src/routes/proxy.ts, apps/worker/src/env.ts, apps/worker/wrangler.toml
|
||||
- **[worker/proxy]**: 新增响应适配器注册层(OpenAI↔Anthropic,JSON+SSE),并将上游模型决策与下游协议完全解耦
|
||||
- 类型: 微调(无方案包)
|
||||
- 文件: apps/worker/src/services/chat-response-adapter.ts, apps/worker/src/routes/proxy.ts, tests/worker/provider-transform.test.ts, tests/worker/channel-metadata.test.ts
|
||||
- **[checkin]**: 自动签到不再依赖站点启用状态
|
||||
- 类型: 微调(无方案包)
|
||||
- 文件: apps/worker/src/services/checkin-runner.ts, tests/worker/checkin-runner.test.ts, helloagents/modules/sites.md
|
||||
|
||||
Reference in new issue
Block a user