feat(proxy): 增强 responses 兼容与推理强度记录
- 抽取 reasoning_effort 解析并复用 - /v1/responses 400/404 时回退 /responses - 记录实际请求路径用于 usage 统计 - 更新相关文档与 dev 脚本
This commit is contained in:
1 parent
88aa121190
commit
512a1fa45e
5 files changed
+75
-18
No files matched your search
@@ -9,6 +9,7 @@ import {
|
||||
import { recordUsage } from "../services/usage";
|
||||
import { jsonError } from "../utils/http";
|
||||
import { safeJsonParse } from "../utils/json";
|
||||
import { extractReasoningEffort } from "../utils/reasoning";
|
||||
import { normalizeBaseUrl } from "../utils/url";
|
||||
import {
|
||||
type NormalizedUsage,
|
||||
@@ -84,6 +85,7 @@ async function sleep(ms: number): Promise<void> {
|
||||
proxy.all("/*", tokenAuth, async (c) => {
|
||||
const tokenRecord = c.get("tokenRecord") as TokenRecord;
|
||||
let requestText = await c.req.text();
|
||||
const originalRequestText = requestText;
|
||||
const parsedBody = requestText
|
||||
? safeJsonParse<Record<string, unknown> | null>(requestText, null)
|
||||
: null;
|
||||
@@ -92,17 +94,8 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
? String(parsedBody.model)
|
||||
: null;
|
||||
const isStream = parsedBody?.stream === true;
|
||||
const reasoningEffortRaw =
|
||||
parsedBody && typeof parsedBody === "object"
|
||||
? ((parsedBody as Record<string, unknown>).reasoning_effort ??
|
||||
(parsedBody as Record<string, unknown>).reasoningEffort ??
|
||||
null)
|
||||
: null;
|
||||
const reasoningEffort =
|
||||
typeof reasoningEffortRaw === "string" ||
|
||||
typeof reasoningEffortRaw === "number"
|
||||
? reasoningEffortRaw
|
||||
: null;
|
||||
const reasoningEffort = extractReasoningEffort(parsedBody);
|
||||
let mutatedStreamOptions = false;
|
||||
if (isStream && parsedBody && typeof parsedBody === "object") {
|
||||
const streamOptions = (parsedBody as Record<string, unknown>)
|
||||
.stream_options;
|
||||
@@ -110,10 +103,12 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
(parsedBody as Record<string, unknown>).stream_options = {
|
||||
include_usage: true,
|
||||
};
|
||||
mutatedStreamOptions = true;
|
||||
} else if (
|
||||
(streamOptions as Record<string, unknown>).include_usage !== true
|
||||
) {
|
||||
(streamOptions as Record<string, unknown>).include_usage = true;
|
||||
mutatedStreamOptions = true;
|
||||
}
|
||||
requestText = JSON.stringify(parsedBody);
|
||||
}
|
||||
@@ -136,10 +131,16 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
|
||||
const ordered = createWeightedOrder(candidates);
|
||||
const targetPath = c.req.path;
|
||||
const fallbackPath =
|
||||
targetPath.toLowerCase() === "/v1/responses" ? "/responses" : targetPath;
|
||||
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"));
|
||||
let lastResponse: Response | null = null;
|
||||
let lastChannel: ChannelRecord | null = null;
|
||||
let lastRequestPath = targetPath;
|
||||
const start = Date.now();
|
||||
let selectedChannel: ChannelRecord | null = null;
|
||||
|
||||
@@ -148,7 +149,8 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
let shouldRetry = false;
|
||||
for (const channel of ordered) {
|
||||
lastChannel = channel;
|
||||
const target = `${normalizeBaseUrl(channel.base_url)}${targetPath}${c.req.url.includes("?") ? `?${c.req.url.split("?")[1]}` : ""}`;
|
||||
const baseUrl = normalizeBaseUrl(channel.base_url);
|
||||
const target = `${baseUrl}${targetPath}${querySuffix}`;
|
||||
const headers = new Headers(c.req.header());
|
||||
headers.set("Authorization", `Bearer ${channel.api_key}`);
|
||||
headers.set("x-api-key", String(channel.api_key));
|
||||
@@ -156,13 +158,30 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
headers.delete("content-length");
|
||||
|
||||
try {
|
||||
const response = await fetch(target, {
|
||||
let response = await fetch(target, {
|
||||
method: c.req.method,
|
||||
headers,
|
||||
body: requestText || undefined,
|
||||
});
|
||||
let responsePath = targetPath;
|
||||
if (
|
||||
(response.status === 400 || response.status === 404) &&
|
||||
fallbackPath !== targetPath
|
||||
) {
|
||||
const fallbackTarget = `${baseUrl}${fallbackPath}${querySuffix}`;
|
||||
const fallbackBody = mutatedStreamOptions
|
||||
? originalRequestText
|
||||
: requestText;
|
||||
response = await fetch(fallbackTarget, {
|
||||
method: c.req.method,
|
||||
headers,
|
||||
body: fallbackBody || undefined,
|
||||
});
|
||||
responsePath = fallbackPath;
|
||||
}
|
||||
|
||||
lastResponse = response;
|
||||
lastRequestPath = responsePath;
|
||||
if (response.ok) {
|
||||
selectedChannel = channel;
|
||||
break;
|
||||
@@ -193,7 +212,7 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
await recordUsage(c.env.DB, {
|
||||
tokenId: tokenRecord.id,
|
||||
model,
|
||||
requestPath: targetPath,
|
||||
requestPath: lastRequestPath,
|
||||
totalTokens: 0,
|
||||
latencyMs,
|
||||
firstTokenLatencyMs: isStream ? null : latencyMs,
|
||||
@@ -221,7 +240,7 @@ proxy.all("/*", tokenAuth, async (c) => {
|
||||
tokenId: tokenRecord.id,
|
||||
channelId: channelForUsage.id,
|
||||
model,
|
||||
requestPath: targetPath,
|
||||
requestPath: lastRequestPath,
|
||||
totalTokens: normalized.totalTokens,
|
||||
promptTokens: normalized.promptTokens,
|
||||
completionTokens: normalized.completionTokens,
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
type ReasoningInput = Record<string, unknown>;
|
||||
|
||||
function readEffort(value: unknown): string | number | null {
|
||||
if (typeof value === "string" || typeof value === "number") {
|
||||
return value;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Extracts reasoning effort from request payloads.
|
||||
*
|
||||
* Args:
|
||||
* input: Request body payload.
|
||||
*
|
||||
* Returns:
|
||||
* Reasoning effort value, if present.
|
||||
*/
|
||||
export function extractReasoningEffort(
|
||||
input: unknown,
|
||||
): string | number | null {
|
||||
if (!input || typeof input !== "object") {
|
||||
return null;
|
||||
}
|
||||
const body = input as ReasoningInput;
|
||||
const direct = readEffort(body.reasoning_effort ?? body.reasoningEffort);
|
||||
if (direct !== null) {
|
||||
return direct;
|
||||
}
|
||||
const reasoning = body.reasoning;
|
||||
if (typeof reasoning === "string" || typeof reasoning === "number") {
|
||||
return reasoning;
|
||||
}
|
||||
if (reasoning && typeof reasoning === "object" && !Array.isArray(reasoning)) {
|
||||
return readEffort((reasoning as ReasoningInput).effort);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
@@ -13,6 +13,7 @@
|
||||
- 按渠道权重随机排序
|
||||
- 支持从非流式 JSON、响应头与流式 SSE 解析 usage 字段
|
||||
- 流式请求自动补 `stream_options.include_usage=true` 以便上游返回 usage
|
||||
- 对 `/v1/responses` 且上游返回 400/404 时回退为 `/responses` 重试一次
|
||||
- 可配置失败重试轮询(响应 5xx/429 时触发)
|
||||
- 记录流式请求标记、首 token 延迟与推理强度到 usage_logs
|
||||
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
- 保留天数可配置,查询时触发清理
|
||||
- 查询结果附带渠道与令牌名称
|
||||
- usage_logs 记录输入/输出 tokens、首 token 延迟、流式标记与推理强度
|
||||
- 推理强度来自请求体 `reasoning` / `reasoning_effort` 字段
|
||||
|
||||
## 依赖关系
|
||||
- `usage_logs` 表
|
||||
|
||||
+1
-3
@@ -6,7 +6,7 @@
|
||||
"apps/*"
|
||||
],
|
||||
"scripts": {
|
||||
"dev:worker": "bun --filter api-worker dev",
|
||||
"dev:worker": "bun --cwd apps/worker dev",
|
||||
"dev:ui": "bun --filter api-worker-ui dev -- --port 4173",
|
||||
"test": "vitest",
|
||||
"format": "bunx --bun biome format --write .",
|
||||
@@ -24,5 +24,3 @@
|
||||
"wrangler": "4"
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user