From 80a6fedbace226b8cfe4f82bf5450fc2fd826c3b Mon Sep 17 00:00:00 2001 From: dreamhunter2333 Date: Wed, 31 Dec 2025 11:31:07 +0800 Subject: [PATCH] Add responses proxies and usage fixes --- AGENTS.md | 41 +++ README.md | 82 +++++- frontend/src/api/client.ts | 8 +- frontend/src/pages/ApiTest.tsx | 5 + frontend/src/pages/Channels.tsx | 22 +- frontend/src/types/index.ts | 2 +- src/admin/channel_api.ts | 2 +- src/admin/token_utils.ts | 2 +- src/providers/azure-openai-proxy.ts | 15 +- src/providers/azure-openai-responses-proxy.ts | 256 ++++++++++++++++++ src/providers/claude-proxy.ts | 41 ++- src/providers/index.ts | 214 ++++++++++++--- src/providers/openai-proxy.ts | 8 +- src/providers/openai-responses-proxy.ts | 241 +++++++++++++++++ type.d.ts | 9 +- 15 files changed, 864 insertions(+), 84 deletions(-) create mode 100644 AGENTS.md create mode 100644 src/providers/azure-openai-responses-proxy.ts create mode 100644 src/providers/openai-responses-proxy.ts diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..8e64543 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,41 @@ +# Repository Guidelines + +## Project Structure & Module Organization +- `src/`: Cloudflare Workers backend (Hono + chanfana). Key areas: + - `src/providers/`: provider proxies (OpenAI, Azure OpenAI, Claude, Responses). + - `src/admin/`: admin APIs (channels, tokens, pricing, DB init). + - `src/db/`: D1 schema/init helpers. +- `public/`: built admin UI assets served by Workers. +- `frontend/`: React/Vite admin UI source (build outputs to `public/`). +- `wrangler.toml.template`: Workers config template (bindings, routes, D1, assets). +- `type.d.ts`: shared types for bindings and data models. + +## Build, Test, and Development Commands +- `pnpm -C frontend build`: Type-checks and builds the admin UI into `public/`. +- `pnpm dev`: Runs the Worker locally via `wrangler dev` (expects a local `wrangler.toml`). +- `pnpm run deploy`: Deploys the Worker to Cloudflare. +- `pnpm run cf-typegen`: Generates Cloudflare bindings/types. + +## Coding Style & Naming Conventions +- TypeScript throughout; follow existing patterns and file layout. +- Indentation: 4 spaces in backend files, 2 spaces in frontend TSX (match existing). +- API routes: `src/admin/*_api.ts`, provider files in `src/providers/`. +- Channel types: use string literals (e.g., `openai-responses`, `azure-openai-responses`). + +## Testing Guidelines +- No dedicated test framework is currently configured. +- Use `pnpm -C frontend build` as a basic safety check for UI changes. +- For backend changes, run `pnpm dev` and validate with the API Test page. + +## Commit & Pull Request Guidelines +- No formal commit conventions are documented in-repo; use clear, imperative messages (e.g., "Add responses proxy"). +- PRs should include: + - Summary of changes. + - Notes on config changes (`wrangler.toml`, env vars). + - Screenshots for UI changes (from `frontend/`). + +## Security & Configuration Tips +- Admin endpoints require `x-admin-token` (`ADMIN_TOKEN` binding). +- Tokens and channel configs are stored in D1; validate permissions and quotas. +- For Azure Responses v1, leave `api_version` empty and set `endpoint` to the resource host. +- `wrangler.toml` is git-ignored; copy from `wrangler.toml.template` for local use. diff --git a/README.md b/README.md index 5fc4a00..e5dbd89 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ ## ✨ 特性 - 🚀 **基于 Cloudflare Workers**：无服务器架构，全球边缘部署 -- 🔐 **多渠道支持**：支持 Azure OpenAI 和 OpenAI，后续会继续增加其他 AI 服务提供商 +- 🔐 **多渠道支持**：支持 OpenAI、Azure OpenAI、Claude、OpenAI Responses、Azure OpenAI Responses - 🎫 **Token 管理**：完整的 API Token 生成、管理和配额控制 - 📊 **使用量统计**：实时统计 API 使用量和费用 - 💰 **定价管理**：灵活的模型定价配置 @@ -52,6 +52,9 @@ awsl-one-api/ │ ├── providers/ # AI 服务提供商 │ │ ├── azure-openai-proxy.ts # Azure OpenAI 代理 │ │ ├── openai-proxy.ts # OpenAI 代理 +│ │ ├── claude-proxy.ts # Claude 代理 +│ │ ├── openai-responses-proxy.ts # OpenAI Responses 代理 +│ │ ├── azure-openai-responses-proxy.ts # Azure Responses 代理 │ │ └── index.ts # 提供商路由 │ ├── db/ # 数据库相关 │ ├── model/ # 数据模型 @@ -143,10 +146,8 @@ pnpm run deploy 1. 在 Web 界面切换到 **🔗 频道管理** 标签 2. 点击 **➕ 添加频道** 按钮 -3. 选择频道类型（Azure OpenAI 或 OpenAI） -4. 填写频道标识和配置信息： - - 对于 OpenAI：填写基本信息（名称、端点、API密钥）和模型映射配置 - - 对于 Azure OpenAI：还需要配置 API 版本和模型部署映射 +3. 选择频道类型（OpenAI、Azure OpenAI、Claude、Responses） +4. 填写频道标识和配置信息（名称、端点、API 密钥、模型映射） 5. 点击 **💾 保存频道** 按钮 **提示**：系统会根据选择的频道类型自动显示相应的配置字段。 @@ -179,6 +180,18 @@ curl https://your-domain.com/v1/chat/completions \ }' ``` +#### Responses API（OpenAI / Azure） + +```bash +curl https://your-domain.com/v1/responses \ + -H "Content-Type: application/json" \ + -H "Authorization: Bearer YOUR_API_TOKEN" \ + -d '{ + "model": "gpt-5.1-codex-max", + "input": "Hello, Responses API!" + }' +``` + #### API 测试工具 管理界面内置了强大的 API 测试工具，无需额外工具即可进行 API 调试： @@ -206,7 +219,7 @@ curl https://your-domain.com/v1/chat/completions \ 访问 `https://your-domain.com` 即可使用 Web 管理界面，功能包括： - **📊 数据库管理**：一键初始化数据库表结构 -- **🔗 频道配置管理**：添加、编辑、删除 AI 服务提供商频道（支持 Azure OpenAI 和 OpenAI） +- **🔗 频道配置管理**：添加、编辑、删除 AI 服务提供商频道（支持 OpenAI、Azure OpenAI、Claude、OpenAI Responses、Azure Responses） - **🔑 API Token 管理**：生成、管理和监控 API Token 使用情况 - **💰 定价配置**：灵活配置不同模型的定价策略 - **🧪 API 测试工具**：内置 API 测试界面，支持实时调试和错误排查 @@ -224,6 +237,21 @@ curl https://your-domain.com/v1/chat/completions \ 目前支持以下 AI 服务提供商： +**OpenAI 配置** + +```json +{ + "name": "My OpenAI Channel", + "type": "openai", + "endpoint": "https://api.openai.com/v1/", + "api_key": "sk-your-openai-api-key", + "deployment_mapper": { + "gpt-4": "gpt-4", + "gpt-3.5-turbo": "gpt-3.5-turbo" + } +} +``` + **Azure OpenAI 配置** ```json @@ -240,17 +268,45 @@ curl https://your-domain.com/v1/chat/completions \ } ``` -**OpenAI 配置** +**Claude 配置** ```json { - "name": "My OpenAI Channel", - "type": "openai", + "name": "My Claude Channel", + "type": "claude", + "endpoint": "https://api.anthropic.com/v1/", + "api_key": "sk-your-claude-api-key", + "api_version": "2023-06-01", + "deployment_mapper": { + "claude-3-5-sonnet-20241022": "claude-3-5-sonnet-20241022" + } +} +``` + +**OpenAI Responses 配置** + +```json +{ + "name": "My OpenAI Responses", + "type": "openai-responses", "endpoint": "https://api.openai.com/v1/", "api_key": "sk-your-openai-api-key", "deployment_mapper": { - "gpt-4": "gpt-4", - "gpt-3.5-turbo": "gpt-3.5-turbo" + "gpt-5.1-codex-max": "gpt-5.1-codex-max" + } +} +``` + +**Azure OpenAI Responses 配置（v1）** + +```json +{ + "name": "My Azure Responses", + "type": "azure-openai-responses", + "endpoint": "https://your-resource.openai.azure.com/", + "api_key": "your-azure-api-key", + "deployment_mapper": { + "gpt-5.1-codex-max": "your-deployment-name" } } ``` @@ -258,10 +314,10 @@ curl https://your-domain.com/v1/chat/completions \ **配置字段说明**： - `name`: 频道显示名称 -- `type`: 服务提供商类型（`azure-openai` 或 `openai`） +- `type`: 服务提供商类型（`openai`、`azure-openai`、`claude`、`openai-responses`、`azure-openai-responses`） - `endpoint`: API 端点地址 - `api_key`: API 密钥 -- `api_version`: API 版本（仅 Azure OpenAI 需要） +- `api_version`: API 版本（Azure OpenAI / Claude 可用；Azure Responses v1 请留空） - `deployment_mapper`: 模型名称映射关系（用于自定义模型名称映射） ### Token 配置 diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index ecf0c71..ecf1216 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -54,7 +54,13 @@ async function responseInterceptor(response: Response): Promise { try { // 尝试解析为 JSON const errorData = JSON.parse(responseText) - errorMessage = errorData.message || errorData.error || JSON.stringify(errorData) + if (typeof errorData?.message === 'string') { + errorMessage = errorData.message + } else if (typeof errorData?.error === 'string') { + errorMessage = errorData.error + } else { + errorMessage = JSON.stringify(errorData) + } } catch { // 不是 JSON，直接使用文本 errorMessage = responseText || `请求失败: ${response.status}` diff --git a/frontend/src/pages/ApiTest.tsx b/frontend/src/pages/ApiTest.tsx index 0f88c00..33c3e5b 100644 --- a/frontend/src/pages/ApiTest.tsx +++ b/frontend/src/pages/ApiTest.tsx @@ -32,6 +32,10 @@ const requestTemplates: Record = { }, ], }, + '/v1/responses': { + model: 'gpt-4o-mini', + input: 'Hello, Responses API!', + }, '/v1/completions': { model: 'gpt-3.5-turbo-instruct', prompt: 'Once upon a time', @@ -121,6 +125,7 @@ export function ApiTest() { > + diff --git a/frontend/src/pages/Channels.tsx b/frontend/src/pages/Channels.tsx index b708a09..80a4060 100644 --- a/frontend/src/pages/Channels.tsx +++ b/frontend/src/pages/Channels.tsx @@ -346,6 +346,8 @@ export function Channels() { + + @@ -370,15 +372,17 @@ export function Channels() { /> -
- - setFormData({ ...formData, api_version: e.target.value })} - placeholder="例如: 2024-02-01" - /> -
+ {(formData.type === 'azure-openai' || formData.type === 'claude') && ( +
+ + setFormData({ ...formData, api_version: e.target.value })} + placeholder="Claude: 2023-06-01 / Azure: 2024-02-01" + /> +
+ )}
diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index efb0ad3..c41ef7c 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -6,7 +6,7 @@ export interface Channel { export interface ChannelConfig { name: string - type: 'openai' | 'azure-openai' | 'claude' + type: 'openai' | 'azure-openai' | 'claude' | 'openai-responses' | 'azure-openai-responses' endpoint: string api_key: string api_version?: string diff --git a/src/admin/channel_api.ts b/src/admin/channel_api.ts index 094eb35..814a204 100644 --- a/src/admin/channel_api.ts +++ b/src/admin/channel_api.ts @@ -49,7 +49,7 @@ export class ChannelUpsertEndpoint extends OpenAPIRoute { type: z.string().describe('Channel type'), endpoint: z.string().describe('API endpoint'), api_key: z.string().describe('API key'), - api_version: z.string().describe('API version'), + api_version: z.string().optional().describe('API version'), deployment_mapper: z.record(z.string()).describe('Model deployment mapping'), model_pricing: z.record(z.object({ input: z.number().describe('Input token price per 1000 tokens'), diff --git a/src/admin/token_utils.ts b/src/admin/token_utils.ts index d96aaf4..a6235b4 100644 --- a/src/admin/token_utils.ts +++ b/src/admin/token_utils.ts @@ -33,7 +33,7 @@ export const TokenUtils = { // Get pricing and calculate cost const pricing = await this.getPricing(c, model, targetChannelConfig); - const hasTokens = usage.prompt_tokens && usage.completion_tokens; + const hasTokens = usage.prompt_tokens != null && usage.completion_tokens != null; if (pricing && hasTokens) { const inputCost = usage.prompt_tokens! * pricing.input; diff --git a/src/providers/azure-openai-proxy.ts b/src/providers/azure-openai-proxy.ts index bcb66bf..388fb63 100644 --- a/src/providers/azure-openai-proxy.ts +++ b/src/providers/azure-openai-proxy.ts @@ -11,9 +11,12 @@ const buildProxyRequest = ( const url = new URL(request.url) const targetUrl = new URL(config.endpoint) - // 如果 endpoint 末尾不带 #，则设置标准的 Azure OpenAI pathname + // 如果 endpoint 末尾不带 #，则设置标准的 Azure OpenAI pathname（保留原有 path 前缀） if (!config.endpoint.endsWith('#')) { - targetUrl.pathname = `/openai/deployments/${deploymentName}/${url.pathname.replace('/v1/', '')}` + 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) { @@ -103,14 +106,18 @@ export default { async fetch( c: Context, config: ChannelConfig, + requestBody: any, saveUsage: (usage: Usage) => Promise, ): Promise { // 准备请求数据 - const reqJson = await c.req.json() + const reqJson = requestBody // 强制包含使用数据 const { model: modelName, stream } = reqJson; if (stream) { - reqJson.stream_options = { "include_usage": true } + reqJson.stream_options = { + ...(reqJson.stream_options || {}), + include_usage: true, + } } // 获取部署名称 diff --git a/src/providers/azure-openai-responses-proxy.ts b/src/providers/azure-openai-responses-proxy.ts new file mode 100644 index 0000000..2f6aecb --- /dev/null +++ b/src/providers/azure-openai-responses-proxy.ts @@ -0,0 +1,256 @@ +import { Context } from "hono" + +type OpenAIResponsesUsage = { + input_tokens?: number; + output_tokens?: number; + total_tokens?: number; + prompt_tokens?: number; + completion_tokens?: number; +} + +type OpenAIResponsesResponse = { + usage?: OpenAIResponsesUsage; + input_tokens?: number; + output_tokens?: number; + total_tokens?: number; + prompt_tokens?: number; + completion_tokens?: number; +} + +type OpenAIResponsesStreamEvent = { + type?: string; + response?: OpenAIResponsesResponse; + delta?: string; + text?: string; +} + +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('#')) { + // Azure OpenAI Responses uses /openai/v1/responses + targetUrl.pathname = `/openai${url.pathname}` + } + + if (config.api_version) { + targetUrl.searchParams.set('api-version', config.api_version) + } + + const targetHeaders = new Headers(request.headers) + targetHeaders.delete("Authorization") + targetHeaders.set("api-key", config.api_key) + + return new Request(targetUrl, { + method: request.method, + headers: targetHeaders, + body: JSON.stringify(reqJson), + }) +} + +const normalizeUsage = (raw?: OpenAIResponsesUsage): Usage | null => { + if (!raw) return null + const promptTokens = raw.prompt_tokens ?? raw.input_tokens + const completionTokens = raw.completion_tokens ?? raw.output_tokens + const totalTokens = raw.total_tokens ?? ( + (promptTokens ?? 0) + (completionTokens ?? 0) + ) + if (promptTokens == null && completionTokens == null && totalTokens == null) { + return null + } + return { + prompt_tokens: promptTokens, + completion_tokens: completionTokens, + total_tokens: totalTokens, + } +} + +const extractUsageFromResponse = (res?: OpenAIResponsesResponse): Usage | null => { + if (!res) return null + return normalizeUsage(res.usage || res) +} + +const estimateTokensFromText = (text: string): number => { + if (!text) return 0 + const bytes = new TextEncoder().encode(text).length + return Math.ceil(bytes / 4) +} + +const estimateUsageFromBodies = ( + requestBody: any, + responseBody?: any, + streamedOutputText?: string +): Usage | null => { + const promptText = requestBody ? JSON.stringify(requestBody) : "" + const completionText = streamedOutputText + ? streamedOutputText + : (responseBody ? JSON.stringify(responseBody) : "") + const promptTokens = estimateTokensFromText(promptText) + const completionTokens = estimateTokensFromText(completionText) + if (!promptTokens && !completionTokens) return null + return { + prompt_tokens: promptTokens, + completion_tokens: completionTokens, + total_tokens: promptTokens + completionTokens, + } +} + +const checkoutUsageData = async ( + saveUsage: (usage: Usage) => Promise, + response: Response | OpenAIResponsesResponse +): Promise => { + try { + const res = response instanceof Response + ? await response.clone().json() + : response + const usage = extractUsageFromResponse(res) + if (!usage) return + await saveUsage(usage) + } catch (error) { + console.error("Error logging usage data:", error) + } +} + +const processStreamData = async ( + lines: string[], + usageSaved: { value: boolean }, + outputText: { value: string }, + outputLimit: number, + saveUsage: (usage: Usage) => Promise +): Promise => { + if (usageSaved.value) return + const processedLines = lines + .map(line => line.trim()) + .filter(line => line.length > 0 && line.startsWith('data:')) + .map(line => line.replace('data:', '').trim()) + .filter(line => line !== '[DONE]') + + for (const jsonContent of processedLines) { + try { + const event = JSON.parse(jsonContent) as OpenAIResponsesStreamEvent + if (event.type === "response.output_text.delta") { + const deltaText = typeof event.delta === "string" + ? event.delta + : (typeof event.text === "string" ? event.text : "") + if (deltaText) { + if (outputText.value.length < outputLimit) { + outputText.value += deltaText + } + } + continue + } + if (event.type !== "response.completed") { + continue + } + const usage = extractUsageFromResponse(event.response) + if (usage && !usageSaved.value) { + await saveUsage(usage) + usageSaved.value = true + } + } catch (e) { + console.error("Error parsing stream data:", e) + } + } +} + +const handleStreamResponse = async ( + c: Context, + streamForServer: ReadableStream | undefined, + requestBody: any, + saveUsage: (usage: Usage) => Promise +): Promise => { + const reader = streamForServer?.getReader() + if (!reader) { + throw new Error("No reader found in response body") + } + + const decoder = new TextDecoder('utf-8') + let buffer = "" + const usageSaved = { value: false } + const outputText = { value: "" } + const outputLimit = 200_000 + while (true) { + const { done, value } = await reader.read() + if (done) break + + const chunk = decoder.decode(value, { stream: true }) + buffer += chunk + + if (!chunk.includes('\n')) continue + + const lines = buffer.split('\n') + buffer = lines.pop() || "" + + await processStreamData(lines, usageSaved, outputText, outputLimit, saveUsage) + } + + if (buffer.trim()) { + await processStreamData([buffer], usageSaved, outputText, outputLimit, saveUsage) + } + + if (!usageSaved.value) { + const estimatedUsage = estimateUsageFromBodies( + requestBody, + undefined, + outputText.value + ) + if (estimatedUsage) { + await saveUsage(estimatedUsage) + } + } +} + +export default { + async fetch( + c: Context, + config: ChannelConfig, + requestBody: any, + saveUsage: (usage: Usage) => Promise, + ): Promise { + const reqJson = requestBody + const { model: modelName, stream } = reqJson + + const deploymentName = config.deployment_mapper[modelName] + if (!deploymentName) { + throw new Error(`Model ${modelName} not supported`) + } + + reqJson.model = deploymentName + + const proxyRequest = buildProxyRequest(c.req.raw, reqJson, config) + const response = await fetch(proxyRequest) + + if (stream) { + const [streamForClient, streamForServer] = response.body?.tee() || [] + c.executionCtx.waitUntil(handleStreamResponse(c, streamForServer, reqJson, saveUsage)) + return new Response(streamForClient, { + headers: response.headers, + status: response.status, + statusText: response.statusText, + }) + } + + if (response.ok) { + try { + const resJson = await response.clone().json() + const usage = extractUsageFromResponse(resJson) + if (usage) { + await saveUsage(usage) + } else { + const estimatedUsage = estimateUsageFromBodies(reqJson, resJson) + if (estimatedUsage) { + await saveUsage(estimatedUsage) + } + } + } catch (error) { + console.error("Error parsing response JSON for usage:", error) + } + } + + return response + } +} diff --git a/src/providers/claude-proxy.ts b/src/providers/claude-proxy.ts index 18ba875..5614307 100644 --- a/src/providers/claude-proxy.ts +++ b/src/providers/claude-proxy.ts @@ -67,6 +67,10 @@ const buildProxyRequest = ( // Claude uses x-api-key header instead of Authorization targetHeaders.set("x-api-key", config.api_key) + // Claude requires anthropic-version header + if (config.api_version) { + targetHeaders.set("anthropic-version", config.api_version) + } return new Request(targetUrl, { method: request.method, @@ -125,6 +129,7 @@ const checkoutUsageData = async ( const processStreamData = async ( lines: string[], usageAccumulator: { input_tokens: number; output_tokens: number }, + usageSaved: { value: boolean }, saveUsage: (usage: Usage) => Promise ): Promise => { for (const line of lines) { @@ -148,28 +153,42 @@ const processStreamData = async ( } // Extract output tokens from message_delta event - if (event.type === "message_delta" && event.usage) { + if (event.type === "message_delta" && event.usage?.output_tokens != null) { usageAccumulator.output_tokens = event.usage.output_tokens } // Handle message_complete event (has complete usage info) - if (event.type === "message_complete" && event.message?.usage) { + if (event.type === "message_complete" && event.message?.usage && !usageSaved.value) { const usage: Usage = { prompt_tokens: event.message.usage.input_tokens, completion_tokens: event.message.usage.output_tokens, total_tokens: event.message.usage.input_tokens + event.message.usage.output_tokens, } await saveUsage(usage) + usageSaved.value = true } // When stream ends with message_stop (fallback for old API versions) if (event.type === "message_stop") { - const usage: Usage = { - prompt_tokens: usageAccumulator.input_tokens, - completion_tokens: usageAccumulator.output_tokens, - total_tokens: usageAccumulator.input_tokens + usageAccumulator.output_tokens, + if (event.message?.usage && !usageSaved.value) { + const usage: Usage = { + prompt_tokens: event.message.usage.input_tokens, + completion_tokens: event.message.usage.output_tokens, + total_tokens: event.message.usage.input_tokens + event.message.usage.output_tokens, + } + await saveUsage(usage) + usageSaved.value = true + continue + } + if (!usageSaved.value) { + const usage: Usage = { + prompt_tokens: usageAccumulator.input_tokens, + completion_tokens: usageAccumulator.output_tokens, + total_tokens: usageAccumulator.input_tokens + usageAccumulator.output_tokens, + } + await saveUsage(usage) + usageSaved.value = true } - await saveUsage(usage) } } catch (e) { console.error("Error parsing Claude stream data:", e, "Line:", jsonContent) @@ -202,6 +221,7 @@ const handleStreamResponse = async ( input_tokens: 0, output_tokens: 0, } + const usageSaved = { value: false } while (true) { const { done, value } = await reader.read() @@ -216,12 +236,12 @@ const handleStreamResponse = async ( const lines = buffer.split('\n') buffer = lines.pop() || "" - await processStreamData(lines, usageAccumulator, saveUsage) + await processStreamData(lines, usageAccumulator, usageSaved, saveUsage) } // Process any remaining buffered data if (buffer.trim()) { - await processStreamData([buffer], usageAccumulator, saveUsage) + await processStreamData([buffer], usageAccumulator, usageSaved, saveUsage) } } @@ -241,10 +261,11 @@ export default { async fetch( c: Context, config: ChannelConfig, + requestBody: any, saveUsage: (usage: Usage) => Promise, ): Promise { // Parse request body - const reqJson = await c.req.json() + const reqJson = requestBody const { model: modelName, stream } = reqJson; // Validate model support diff --git a/src/providers/index.ts b/src/providers/index.ts index 20357f8..74d345b 100644 --- a/src/providers/index.ts +++ b/src/providers/index.ts @@ -4,6 +4,8 @@ import { contentJson, fromHono, OpenAPIRoute } from 'chanfana'; import azureOpenaiProxy from "./azure-openai-proxy" import openaiProxy from "./openai-proxy" import claudeProxy from "./claude-proxy" +import openaiResponsesProxy from "./openai-responses-proxy" +import azureOpenaiResponsesProxy from "./azure-openai-responses-proxy" import utils from "../utils" import { TokenUtils } from "../admin/token_utils" import { CONSTANTS } from "../constants" @@ -16,14 +18,63 @@ const providerMap: Record< ( c: Context, config: ChannelConfig, + requestBody: any, saveUsage: (usage: Usage) => Promise ) => Promise > = { "azure-openai": azureOpenaiProxy.fetch, "openai": openaiProxy.fetch, "claude": claudeProxy.fetch, + "openai-responses": openaiResponsesProxy.fetch, + "azure-openai-responses": azureOpenaiResponsesProxy.fetch, }; +const getApiKeyFromHeaders = (c: Context): string | null => { + const authHeader = c.req.raw.headers.get('Authorization'); + const xApiKey = c.req.raw.headers.get('x-api-key'); + + if (authHeader) { + return authHeader.replace("Bearer ", "").trim(); + } + if (xApiKey) { + return xApiKey.trim(); + } + return null; +} + +const fetchTokenData = async (c: Context, apiKey: string) => { + const tokenResult = await c.env.DB.prepare( + `SELECT * FROM api_token WHERE key = ?` + ).bind(apiKey).first(); + + if (!tokenResult || !tokenResult.value) { + return null; + } + + return { + tokenData: JSON.parse(tokenResult.value as string) as ApiTokenData, + usage: tokenResult.usage as number || 0, + }; +} + +const fetchChannelsForToken = async ( + c: Context, + tokenData: ApiTokenData +) => { + const channelKeys = tokenData.channel_keys; + + if (!channelKeys || channelKeys.length === 0) { + return await c.env.DB.prepare( + `SELECT key, value FROM channel_config` + ).all(); + } + + const channelQuery = channelKeys.map(() => '?').join(','); + return await c.env.DB.prepare( + `SELECT key, value FROM channel_config WHERE key IN (${channelQuery})` + ).bind(...channelKeys).all(); +} + class ProxyEndpoint extends OpenAPIRoute { schema = { tags: ['OpenAI Proxy'], @@ -45,62 +96,38 @@ class ProxyEndpoint extends OpenAPIRoute { async handle(c: Context) { // Support both OpenAI format (Authorization: Bearer xxx) and Claude format (x-api-key: xxx) - const authHeader = c.req.raw.headers.get('Authorization'); - const xApiKey = c.req.raw.headers.get('x-api-key'); - - let apiKey: string | undefined; - - if (authHeader) { - // OpenAI format: Authorization: Bearer sk-xxx - apiKey = authHeader.replace("Bearer ", "").trim(); - } else if (xApiKey) { - // Claude format: x-api-key: sk-xxx - apiKey = xApiKey.trim(); - } - + const apiKey = getApiKeyFromHeaders(c); if (!apiKey) { return c.text("Authorization header or x-api-key not found", 401); } - // Get token from database using direct SQL query - const tokenResult = await c.env.DB.prepare( - `SELECT * FROM api_token WHERE key = ?` - ).bind(apiKey).first(); - - if (!tokenResult || !tokenResult.value) { + const tokenInfo = await fetchTokenData(c, apiKey); + if (!tokenInfo) { return c.text("Invalid API key", 401); } - const tokenData = JSON.parse(tokenResult.value as string) as ApiTokenData; - const usage = tokenResult.usage as number || 0; + const { tokenData, usage } = tokenInfo; // Check if token has sufficient quota if (usage >= tokenData.total_quota) { return c.text("Quota exceeded", 402); } // Get available channel configs based on token permissions - const channelKeys = tokenData.channel_keys; - - let channelsResult; - if (!channelKeys || channelKeys.length === 0) { - // If channel_keys is empty, get all channels - channelsResult = await c.env.DB.prepare( - `SELECT key, value FROM channel_config` - ).all(); - } else { - // Query specified channels - const channelQuery = channelKeys.map(() => '?').join(','); - channelsResult = await c.env.DB.prepare( - `SELECT key, value FROM channel_config WHERE key IN (${channelQuery})` - ).bind(...channelKeys).all(); - } + const channelsResult = await fetchChannelsForToken(c, tokenData); if (!channelsResult.results || channelsResult.results.length === 0) { return c.text("No available channels for this token", 401); } - // Parse channel configs and get the model from request - const requestBody = await c.req.json(); - const model = requestBody.model || 'gpt-3.5-turbo'; + let requestBody: any; + try { + requestBody = await c.req.json(); + } catch (error) { + return c.text("Invalid JSON body", 400); + } + const model = requestBody.model; + if (!model) { + return c.text("Model is required", 400); + } // Filter channels that support the requested model const availableChannels: Array<{ key: string, config: ChannelConfig }> = []; @@ -118,7 +145,10 @@ class ProxyEndpoint extends OpenAPIRoute { } if (availableChannels.length === 0) { - return c.text(`No channels available for model: ${model}`, 400); + if (!channelsResult.results || channelsResult.results.length === 0) { + return c.text("No available channels for this token", 404); + } + return c.text(`Model not mapped: ${model}. Please configure deployment_mapper.`, 400); } // Randomly select one available channel @@ -136,7 +166,7 @@ class ProxyEndpoint extends OpenAPIRoute { return c.text("Channel type not supported", 400); } - return proxyFetch(c, targetChannelConfig, + return proxyFetch(c, targetChannelConfig, requestBody, async (usage: Usage) => { try { await TokenUtils.processUsage(c, apiKey, model, targetChannelKey, targetChannelConfig, usage); @@ -150,3 +180,105 @@ class ProxyEndpoint extends OpenAPIRoute { api.post("/v1/chat/completions", ProxyEndpoint) api.post("/v1/messages", ProxyEndpoint) // Claude API endpoint + +class ResponsesProxyEndpoint extends OpenAPIRoute { + schema = { + tags: ['OpenAI Responses Proxy'], + request: { + headers: z.object({ + 'Authorization': z.string().optional().describe("Token for authentication (OpenAI format)"), + 'x-api-key': z.string().optional().describe("API key for authentication (Claude format)"), + }), + body: contentJson({ + schema: z.any(), + }), + }, + responses: { + 200: { + description: 'Successful response', + }, + }, + }; + + async handle(c: Context) { + const apiKey = getApiKeyFromHeaders(c); + if (!apiKey) { + return c.text("Authorization header or x-api-key not found", 401); + } + + const tokenInfo = await fetchTokenData(c, apiKey); + if (!tokenInfo) { + return c.text("Invalid API key", 401); + } + + const { tokenData, usage } = tokenInfo; + + if (usage >= tokenData.total_quota) { + return c.text("Quota exceeded", 402); + } + + const channelsResult = await fetchChannelsForToken(c, tokenData); + + if (!channelsResult.results || channelsResult.results.length === 0) { + return c.text("No available channels for this token", 401); + } + + let requestBody: any; + try { + requestBody = await c.req.json(); + } catch (error) { + return c.text("Invalid JSON body", 400); + } + const model = requestBody.model; + if (!model) { + return c.text("Model is required", 400); + } + + const allowedTypes: ChannelType[] = ["openai-responses", "azure-openai-responses"]; + const availableChannels: Array<{ key: string, config: ChannelConfig }> = []; + + for (const row of channelsResult.results) { + const config = JSON.parse(row.value) as ChannelConfig; + + if (!config.type || !allowedTypes.includes(config.type)) { + continue; + } + + if (config.deployment_mapper && config.deployment_mapper[model]) { + availableChannels.push({ + key: row.key, + config: config + }); + } + } + + if (availableChannels.length === 0) { + if (!channelsResult.results || channelsResult.results.length === 0) { + return c.text("No available channels for this token", 404); + } + return c.text(`Model not mapped: ${model}. Please configure deployment_mapper.`, 400); + } + + const randomIndex = Math.floor(Math.random() * availableChannels.length); + const selectedChannel = availableChannels[randomIndex]; + const targetChannelKey = selectedChannel.key; + const targetChannelConfig = selectedChannel.config; + + const proxyFetch = providerMap[targetChannelConfig.type || ""]; + if (!proxyFetch) { + return c.text("Channel type not supported", 400); + } + + return proxyFetch(c, targetChannelConfig, requestBody, + async (usage: Usage) => { + try { + await TokenUtils.processUsage(c, apiKey, model, targetChannelKey, targetChannelConfig, usage); + } catch (error) { + console.error('Error processing usage:', error); + } + } + ); + } +} + +api.post("/v1/responses", ResponsesProxyEndpoint) diff --git a/src/providers/openai-proxy.ts b/src/providers/openai-proxy.ts index 1d08abd..2db16dd 100644 --- a/src/providers/openai-proxy.ts +++ b/src/providers/openai-proxy.ts @@ -94,14 +94,18 @@ export default { async fetch( c: Context, config: ChannelConfig, + requestBody: any, saveUsage: (usage: Usage) => Promise, ): Promise { // 准备请求数据 - const reqJson = await c.req.json() + const reqJson = requestBody // 强制包含使用数据 const { model: modelName, stream } = reqJson; if (stream) { - reqJson.stream_options = { "include_usage": true } + reqJson.stream_options = { + ...(reqJson.stream_options || {}), + include_usage: true, + } } // 检查模型是否受支持 diff --git a/src/providers/openai-responses-proxy.ts b/src/providers/openai-responses-proxy.ts new file mode 100644 index 0000000..ea96bc5 --- /dev/null +++ b/src/providers/openai-responses-proxy.ts @@ -0,0 +1,241 @@ +import { Context } from "hono" + +type OpenAIResponsesUsage = { + input_tokens?: number; + output_tokens?: number; + total_tokens?: number; + prompt_tokens?: number; + completion_tokens?: number; +} + +type OpenAIResponsesResponse = { + usage?: OpenAIResponsesUsage; + input_tokens?: number; + output_tokens?: number; + total_tokens?: number; + prompt_tokens?: number; + completion_tokens?: number; +} + +type OpenAIResponsesStreamEvent = { + type?: string; + response?: OpenAIResponsesResponse; + delta?: string; + text?: string; +} + +const buildProxyRequest = ( + request: Request, + reqJson: any, + config: ChannelConfig +): Request => { + const url = new URL(request.url) + const targetUrl = new URL(config.endpoint) + + targetUrl.pathname = url.pathname + + const targetHeaders = new Headers(request.headers) + targetHeaders.set("Authorization", `Bearer ${config.api_key}`) + + return new Request(targetUrl, { + method: request.method, + headers: targetHeaders, + body: JSON.stringify(reqJson), + }) +} + +const normalizeUsage = (raw?: OpenAIResponsesUsage): Usage | null => { + if (!raw) return null + const promptTokens = raw.prompt_tokens ?? raw.input_tokens + const completionTokens = raw.completion_tokens ?? raw.output_tokens + const totalTokens = raw.total_tokens ?? ( + (promptTokens ?? 0) + (completionTokens ?? 0) + ) + if (promptTokens == null && completionTokens == null && totalTokens == null) { + return null + } + return { + prompt_tokens: promptTokens, + completion_tokens: completionTokens, + total_tokens: totalTokens, + } +} + +const extractUsageFromResponse = (res?: OpenAIResponsesResponse): Usage | null => { + if (!res) return null + return normalizeUsage(res.usage || res) +} + +const estimateTokensFromText = (text: string): number => { + if (!text) return 0 + const bytes = new TextEncoder().encode(text).length + return Math.ceil(bytes / 4) +} + +const estimateUsageFromBodies = ( + requestBody: any, + responseBody?: any, + streamedOutputText?: string +): Usage | null => { + const promptText = requestBody ? JSON.stringify(requestBody) : "" + const completionText = streamedOutputText + ? streamedOutputText + : (responseBody ? JSON.stringify(responseBody) : "") + const promptTokens = estimateTokensFromText(promptText) + const completionTokens = estimateTokensFromText(completionText) + if (!promptTokens && !completionTokens) return null + return { + prompt_tokens: promptTokens, + completion_tokens: completionTokens, + total_tokens: promptTokens + completionTokens, + } +} + +const checkoutUsageData = async ( + saveUsage: (usage: Usage) => Promise, + response: Response | OpenAIResponsesResponse +): Promise => { + try { + const res = response instanceof Response + ? await response.clone().json() + : response + const usage = extractUsageFromResponse(res) + if (!usage) return + await saveUsage(usage) + } catch (error) { + console.error("Error logging usage data:", error) + } +} + +const processStreamData = async ( + lines: string[], + usageSaved: { value: boolean }, + outputText: { value: string }, + outputLimit: number, + saveUsage: (usage: Usage) => Promise +): Promise => { + if (usageSaved.value) return + const processedLines = lines + .map(line => line.trim()) + .filter(line => line.length > 0 && line.startsWith('data:')) + .map(line => line.replace('data:', '').trim()) + .filter(line => line !== '[DONE]') + + for (const jsonContent of processedLines) { + try { + const event = JSON.parse(jsonContent) as OpenAIResponsesStreamEvent + if (event.type === "response.output_text.delta") { + const deltaText = typeof event.delta === "string" + ? event.delta + : (typeof event.text === "string" ? event.text : "") + if (deltaText) { + if (outputText.value.length < outputLimit) { + outputText.value += deltaText + } + } + continue + } + if (event.type !== "response.completed") { + continue + } + const usage = extractUsageFromResponse(event.response) + if (usage && !usageSaved.value) { + await saveUsage(usage) + usageSaved.value = true + } + } catch (e) { + console.error("Error parsing stream data:", e) + } + } +} + +const handleStreamResponse = async ( + c: Context, + streamForServer: ReadableStream | undefined, + requestBody: any, + saveUsage: (usage: Usage) => Promise +): Promise => { + const reader = streamForServer?.getReader() + if (!reader) { + throw new Error("No reader found in response body") + } + + const decoder = new TextDecoder('utf-8') + let buffer = "" + const usageSaved = { value: false } + const outputText = { value: "" } + const outputLimit = 200_000 + while (true) { + const { done, value } = await reader.read() + if (done) break + + const chunk = decoder.decode(value, { stream: true }) + buffer += chunk + + if (!chunk.includes('\n')) continue + + const lines = buffer.split('\n') + buffer = lines.pop() || "" + + await processStreamData(lines, usageSaved, outputText, outputLimit, saveUsage) + } + + if (buffer.trim()) { + await processStreamData([buffer], usageSaved, outputText, outputLimit, saveUsage) + } + + if (!usageSaved.value) { + const estimatedUsage = estimateUsageFromBodies( + requestBody, + undefined, + outputText.value + ) + if (estimatedUsage) { + await saveUsage(estimatedUsage) + } + } +} + +export default { + async fetch( + c: Context, + config: ChannelConfig, + requestBody: any, + saveUsage: (usage: Usage) => Promise, + ): Promise { + const reqJson = requestBody + const { stream } = reqJson + + const proxyRequest = buildProxyRequest(c.req.raw, reqJson, config) + const response = await fetch(proxyRequest) + + if (stream) { + const [streamForClient, streamForServer] = response.body?.tee() || [] + c.executionCtx.waitUntil(handleStreamResponse(c, streamForServer, reqJson, saveUsage)) + return new Response(streamForClient, { + headers: response.headers, + status: response.status, + statusText: response.statusText, + }) + } + + if (response.ok) { + try { + const resJson = await response.clone().json() + const usage = extractUsageFromResponse(resJson) + if (usage) { + await saveUsage(usage) + } else { + const estimatedUsage = estimateUsageFromBodies(reqJson, resJson) + if (estimatedUsage) { + await saveUsage(estimatedUsage) + } + } + } catch (error) { + console.error("Error parsing response JSON for usage:", error) + } + } + + return response + } +} diff --git a/type.d.ts b/type.d.ts index d99d6f7..8aba9c3 100644 --- a/type.d.ts +++ b/type.d.ts @@ -32,7 +32,14 @@ type ApiTokenRow = BaseDbRow & { usage: number; } -type ChannelType = "azure-openai" | "openai" | "claude" | undefined | null; +type ChannelType = + | "azure-openai" + | "openai" + | "claude" + | "openai-responses" + | "azure-openai-responses" + | undefined + | null; type ChannelConfig = { name: string;