// apps/api/src/ai/client.ts // Server-side AI client. Ported from apps/web-legacy/scripts/ai-client.js with // these adjustments: // - No browser CORS proxy. The backend talks directly to model endpoints. // - Optional outbound proxy via UPSTREAM_PROXY (socks5:// or http://) using // undici's ProxyAgent. Replaces the legacy proxy.mjs CONNECT tunnel. // - Streaming support is preserved but defaults to non-streaming since the // analyzer pipeline always wants the full payload before parsing. // - All API keys arrive via constructor / per-call config, never logged. import { ProxyAgent, fetch as undiciFetch, type Dispatcher } from "undici"; // ── Public types ─────────────────────────────────────────────────────────── export interface AiClientConfig { apiKey: string; baseURL?: string; apiMode?: "chat" | "responses"; model?: string; modelPreset?: string; customModel?: string; temperature?: number; topP?: number; maxTokens?: number; enableThinking?: boolean; clearThinking?: boolean; } export interface ChatMessage { role: "system" | "user" | "assistant" | "tool" | string; content: string; } export interface RequestOptions { stream?: boolean; jsonMode?: boolean; jsonModeFallback?: boolean; pluginCompat?: boolean; disableBetaParameterFallback?: boolean; schema?: unknown; schemaName?: string; retries?: number; temperature?: number; topP?: number; maxTokens?: number; endpoint?: string; // computed internally; provider hint for special headers onRetry?: (err: AiError, retryDelayMs: number, attempt: number) => void; onTextDelta?: (collected: string, delta: string) => void; onReasoningDelta?: (collected: string, delta: string) => void; } export interface AiResponse { text: string; endpoint: string; fetchURL: string; apiMode: "chat" | "responses"; } export class AiError extends Error { status: number; isAuthError = false; retryable = false; isTruncated = false; isJsonParseError = false; rawPreview?: string; endpoint?: string; apiMode?: string; original?: unknown; constructor(message: string, status = 0) { super(message); this.status = status; } } // ── Helpers ──────────────────────────────────────────────────────────────── function clamp(value: number, min: number, max: number): number { return Math.min(max, Math.max(min, value)); } function numberOrDefault(value: unknown, fallback: number): number { const parsed = Number(value); return Number.isFinite(parsed) ? parsed : fallback; } function optionOrConfig( options: Record | undefined, config: Record | undefined, key: string, fallback: T ): T { if (options && options[key] !== undefined) return options[key] as T; if (config && config[key] !== undefined) return config[key] as T; return fallback; } export function resolveApiMode( baseURL: string | undefined, fallback: "chat" | "responses" | undefined ): "chat" | "responses" { const value = String(baseURL || ""); if (/\/chat\/completions(\/?|$)/i.test(value)) return "chat"; if (/\/responses(\/?|$)/i.test(value)) return "responses"; return fallback === "chat" ? "chat" : "responses"; } export function resolveEndpoint( baseURL: string | undefined, apiMode: "chat" | "responses" ): string { const targetSuffix = apiMode === "chat" ? "/chat/completions" : "/responses"; const value = String(baseURL || "").trim(); if (!value) return "https://api.openai.com/v1" + targetSuffix; let trimmed = value.replace(/\/+$/g, ""); trimmed = trimmed.replace(/\/(responses|chat\/completions)$/i, ""); if (/\/v\d+$/i.test(trimmed) || /\/openai\/v\d+$/i.test(trimmed)) { return trimmed + targetSuffix; } return trimmed + "/v1" + targetSuffix; } function normalizeModel(config: AiClientConfig): string { if (config.model) return config.model; if (config.modelPreset === "custom") return config.customModel || ""; return config.modelPreset || ""; } function normalizeMessagesForResponses(messages: ChatMessage[]) { return (messages || []).map((m) => ({ role: m.role, content: [{ type: "input_text", text: String(m.content || "") }], })); } function buildHeaders( config: AiClientConfig, endpoint: string ): Record { if (!config.apiKey) { throw new AiError("缺少 API Key(请在后端 .env 中配置 DEFAULT_API_KEY)。", 0); } const headers: Record = { "Content-Type": "application/json" }; if (/azure\.com/i.test(endpoint)) headers["api-key"] = config.apiKey; else headers.Authorization = "Bearer " + config.apiKey; return headers; } function buildRequestBody( config: AiClientConfig, messages: ChatMessage[], options: RequestOptions ): Record { const apiMode: "chat" | "responses" = config.apiMode === "chat" ? "chat" : "responses"; const model = normalizeModel(config); if (!model) throw new AiError("缺少模型名(model)。", 0); if (apiMode === "chat") { if (options.pluginCompat) { return { model, messages: messages || [], temperature: clamp( numberOrDefault( optionOrConfig( options as Record, config as unknown as Record, "temperature", 0.7 ), 0.7 ), 0, 2 ), }; } const body: Record = { model, messages: messages || [], temperature: clamp( numberOrDefault( optionOrConfig(options as Record, config as unknown as Record, "temperature", 0.7), 0.7 ), 0, 2 ), top_p: clamp( numberOrDefault( optionOrConfig(options as Record, config as unknown as Record, "topP", 1), 1 ), 0, 1 ), max_tokens: clamp( numberOrDefault( optionOrConfig(options as Record, config as unknown as Record, "maxTokens", 16384), 16384 ), 1, 131072 ), stream: Boolean(options.stream), }; if (options.jsonMode !== false) { body.response_format = { type: "json_object" }; } const ep = String(options.endpoint || ""); if (/bigmodel\.cn|zhipuai/i.test(ep)) { body.thinking = { type: config.enableThinking === false ? "disabled" : "enabled", }; } else if ( (config.enableThinking || config.clearThinking) && /dashscope|aliyun|qwen|modelscope/i.test(ep) ) { body.chat_template_kwargs = { enable_thinking: Boolean(config.enableThinking), clear_thinking: Boolean(config.clearThinking), }; } return body; } // Responses API const responsesBody: Record = { model, store: false, temperature: clamp( numberOrDefault( optionOrConfig(options as Record, config as unknown as Record, "temperature", 0.7), 0.7 ), 0, 2 ), input: normalizeMessagesForResponses(messages), }; if (options.schema) { responsesBody.text = { format: { type: "json_schema", name: options.schemaName || "structured_result", strict: true, schema: options.schema, }, }; } if (options.stream) responsesBody.stream = true; return responsesBody; } // ── Response parsing ─────────────────────────────────────────────────────── function safePreview(text: string): string { return String(text || "") .replace(/sk-[A-Za-z0-9_-]{8,}/g, "sk-***") .replace(/nvapi-[A-Za-z0-9_-]{8,}/g, "nvapi-***") .replace(/Bearer\s+[A-Za-z0-9._-]+/gi, "Bearer ***") .slice(0, 1200); } function normalizeText(value: unknown): string { if (typeof value === "string") return value; if (Array.isArray(value)) { return value .map((part) => { if (typeof part === "string") return part; if (part && typeof (part as Record).text === "string") return (part as Record).text as string; if (part && typeof (part as Record).content === "string") return (part as Record).content as string; return ""; }) .join(""); } return ""; } function extractChatText(payload: unknown): string { const p = payload as | { choices?: Array<{ finish_reason?: string; message?: { content?: unknown } }> } | undefined; const choice = p?.choices?.[0]; if (!choice) return ""; const finish = choice.finish_reason || ""; if (finish === "length" || finish === "max_tokens") { const err = new AiError( "模型输出被截断(token 超限)。请减少录制文件大小,或在设置中提高 Max Tokens。", 0 ); err.isTruncated = true; err.rawPreview = safePreview(normalizeText(choice.message?.content)); throw err; } return choice.message ? normalizeText(choice.message.content) : ""; } function extractResponsesText(payload: unknown): string { const p = payload as | { output_text?: string; output?: Array<{ content?: Array<{ type?: string; text?: string }> }> } | undefined; if (!p) return ""; if (typeof p.output_text === "string") return p.output_text; if (!Array.isArray(p.output)) return ""; const chunks: string[] = []; p.output.forEach((item) => { (item.content || []).forEach((part) => { if (part && part.type === "output_text" && typeof part.text === "string") chunks.push(part.text); else if (part && typeof part.text === "string") chunks.push(part.text); }); }); return chunks.join(""); } // ── JSON helpers ─────────────────────────────────────────────────────────── function stripJsonNoise(text: string): string { return String(text || "") .replace(//gi, "") .replace(/^```(?:json)?\s*/i, "") .replace(/```\s*$/i, "") .trim(); } function extractJsonObject(text: string): string { const start = text.indexOf("{"); const end = text.lastIndexOf("}"); return start >= 0 && end > start ? text.slice(start, end + 1) : ""; } function extractBalancedJsonObjects(text: string): string[] { const out: string[] = []; let start = -1; let depth = 0; let inString = false; let escaped = false; for (let i = 0; i < text.length; i += 1) { const ch = text[i]; if (inString) { if (escaped) escaped = false; else if (ch === "\\") escaped = true; else if (ch === '"') inString = false; continue; } if (ch === '"') { inString = true; continue; } if (ch === "{") { if (depth === 0) start = i; depth += 1; continue; } if (ch === "}" && depth > 0) { depth -= 1; if (depth === 0 && start >= 0) { out.push(text.slice(start, i + 1)); start = -1; } } } return out; } export function parseJsonText(text: string): unknown { const raw = String(text || "").trim(); if (!raw) throw new Error("模型返回为空,无法解析 JSON。"); const cleaned = stripJsonNoise(raw); const candidates = [raw, cleaned, ...extractBalancedJsonObjects(cleaned).reverse(), extractJsonObject(cleaned)].filter( Boolean ); const seen: Record = {}; for (const cand of candidates) { if (seen[cand]) continue; seen[cand] = true; try { return JSON.parse(cand); } catch (_) { // try next } } throw new Error("模型返回的内容不是合法 JSON。"); } // ── HTTP error helpers ───────────────────────────────────────────────────── function shouldFallbackJsonMode( error: AiError, body: Record, options: RequestOptions, tried: boolean ): boolean { if (tried || !body || !body.response_format || options.jsonModeFallback === false) return false; return Number(error?.status) === 400; } function shouldFallbackBetaParameters( error: AiError, tried: boolean, options: RequestOptions ): boolean { if (options.disableBetaParameterFallback) return false; if (tried || Number(error?.status) !== 400) return false; return /beta[-\s]?limitations|temperature[\s\S]*top_p[\s\S]*(?:fixed|1)|presence_penalty|frequency_penalty/i.test( error?.message || "" ); } function applyBetaParameterLimits( body: Record, apiMode: "chat" | "responses" ): Record { const next: Record = { ...body, temperature: 1 }; if (apiMode === "chat") { next.top_p = 1; next.n = 1; } delete next.presence_penalty; delete next.frequency_penalty; return next; } async function parseErrorResponse(response: Response): Promise { const status = response.status; let message = "HTTP " + status; try { const payload = await response.json(); const p = payload as { error?: { message?: string; code?: string }; message?: string }; if (p && p.error) { message += " - " + (p.error.message || p.error.code || JSON.stringify(p.error)); } else if (p && p.message) { message += " - " + p.message; } } catch (_) { try { const text = await response.text(); if (text) message += " - " + text.slice(0, 300); } catch (_ignore) { // ignore } } const err = new AiError(message, status); err.isAuthError = status === 401 || status === 403; err.retryable = status === 429 || (status >= 500 && status < 600); return err; } function isRetryable(error: AiError | Error | undefined): boolean { if (!error) return false; if (error instanceof AiError && typeof error.retryable === "boolean") return error.retryable; const status = Number((error as AiError).status) || 0; return status === 429 || (status >= 500 && status < 600); } function delay(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } // ── Main client ──────────────────────────────────────────────────────────── export interface AiClientOptions { /** Optional UPSTREAM_PROXY URI (socks5:// or http://). */ upstreamProxy?: string; } export class AiClient { private dispatcher?: Dispatcher; constructor(opts: AiClientOptions = {}) { if (opts.upstreamProxy) { this.dispatcher = new ProxyAgent(opts.upstreamProxy); } } private fetchImpl: typeof undiciFetch = (url, init) => undiciFetch(url, this.dispatcher ? { ...init, dispatcher: this.dispatcher } : init); /** Call model and return parsed JSON. Recovers from common malformed outputs. */ async requestJson( config: AiClientConfig, messages: ChatMessage[], options: RequestOptions = {} ): Promise { const raw = await this.requestText(config, messages, options); try { return parseJsonText(raw.text); } catch (error) { if ((error as AiError).isTruncated) throw error; const next = new AiError( "模型返回非 JSON,已截取原始返回用于诊断。可切换 Responses API 或关闭 thinking 后重试。", 0 ); next.retryable = false; next.isJsonParseError = true; next.rawPreview = safePreview(raw.text); next.endpoint = raw.endpoint; next.apiMode = raw.apiMode; next.original = error; throw next; } } /** Call model and return raw text. Handles retries + JSON-mode fallback + streaming. */ async requestText( config: AiClientConfig, messages: ChatMessage[], options: RequestOptions = {} ): Promise { const apiMode = resolveApiMode(config.baseURL, config.apiMode); const endpoint = resolveEndpoint(config.baseURL, apiMode); const effectiveConfig: AiClientConfig = { ...config, apiMode }; let body = buildRequestBody(effectiveConfig, messages, { ...options, endpoint }); const headers = buildHeaders(effectiveConfig, endpoint); const retries = options.retries == null ? 3 : Number(options.retries); let attempt = 0; let lastError: AiError | Error | undefined; let jsonModeFallbackTried = false; let betaParameterFallbackTried = false; while (attempt <= retries) { try { const response = await this.fetchImpl(endpoint, { method: "POST", headers, body: JSON.stringify(body), }); if (!response.ok) { const httpError = await parseErrorResponse(response as unknown as Response); if (shouldFallbackBetaParameters(httpError, betaParameterFallbackTried, options)) { betaParameterFallbackTried = true; body = applyBetaParameterLimits(body, apiMode); attempt = 0; continue; } if ( shouldFallbackJsonMode(httpError, body, options, jsonModeFallbackTried) ) { jsonModeFallbackTried = true; body = { ...body }; delete body.response_format; attempt = 0; continue; } throw httpError; } const text = options.stream && response.body ? await readStream(response as unknown as Response, apiMode, options) : await readJsonPayload(response as unknown as Response, apiMode); return { text, endpoint, fetchURL: endpoint, apiMode }; } catch (error) { lastError = error as AiError; if (!isRetryable(lastError) || attempt >= retries) throw lastError; attempt += 1; const retryDelay = Math.pow(2, attempt - 1) * 1000; if (typeof options.onRetry === "function") { options.onRetry(lastError as AiError, retryDelay, attempt); } await delay(retryDelay); } } throw lastError || new AiError("请求失败。"); } } // ── Streaming reader (SSE) ───────────────────────────────────────────────── async function readStream( response: Response, apiMode: "chat" | "responses", options: RequestOptions ): Promise { if (!response.body) return ""; const reader = (response.body as ReadableStream).getReader(); const decoder = new TextDecoder(); let buffer = ""; let collected = ""; let reasoning = ""; const consumeEvent = (rawEvent: string) => { const parsed = parseSSEEvent(rawEvent); if (!parsed) return; if (apiMode === "chat") { const choice = (parsed.choices && parsed.choices[0]) as | { delta?: { reasoning_content?: unknown; content?: unknown } } | undefined; const delta = choice?.delta; if (!delta) return; const reasoningDelta = normalizeText(delta.reasoning_content); if (reasoningDelta) { reasoning += reasoningDelta; options.onReasoningDelta?.(reasoning, reasoningDelta); } const contentDelta = normalizeText(delta.content); if (contentDelta) { collected += contentDelta; options.onTextDelta?.(collected, contentDelta); } return; } if ( parsed.type === "response.output_text.delta" && typeof parsed.delta === "string" ) { collected += parsed.delta; options.onTextDelta?.(collected, parsed.delta); } if ( parsed.type === "response.refusal.delta" && typeof parsed.delta === "string" ) { collected += parsed.delta; options.onTextDelta?.(collected, parsed.delta); } if (parsed.type === "error") { throw new AiError( (parsed.error && parsed.error.message) || "流式响应出错。", 0 ); } }; while (true) { const step = await reader.read(); if (step.done) break; buffer += decoder.decode(step.value, { stream: true }); let boundary = buffer.indexOf("\n\n"); while (boundary !== -1) { const rawEvent = buffer.slice(0, boundary); buffer = buffer.slice(boundary + 2); consumeEvent(rawEvent); boundary = buffer.indexOf("\n\n"); } } buffer += decoder.decode(); if (buffer.trim()) consumeEvent(buffer); return collected; } interface SSEEvent { choices?: Array; type?: string; delta?: unknown; error?: { message?: string }; [key: string]: unknown; } function parseSSEEvent(rawEvent: string): SSEEvent | null { const dataLines: string[] = []; rawEvent.split(/\r?\n/).forEach((line) => { if (line.indexOf("data:") === 0) dataLines.push(line.slice(5).trimStart()); }); const payload = dataLines.join("\n").trim(); if (!payload || payload === "[DONE]") return null; try { return JSON.parse(payload) as SSEEvent; } catch (_) { return null; } } async function readJsonPayload( response: Response, apiMode: "chat" | "responses" ): Promise { const payload = await response.json(); return apiMode === "chat" ? extractChatText(payload) : extractResponsesText(payload); } // Re-export safePreview for use by Judge / runner when persisting error previews. export { safePreview };