File size: 4,510 Bytes
4a755c4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | import type {
ChatMessage,
ChatCompletionResponse,
ChatCompletionChunk,
ChatToolDefinition,
ChatToolChoice,
Platform,
} from '@freellmapi/shared/types.js';
export interface CompletionOptions {
model?: string;
temperature?: number;
max_tokens?: number;
top_p?: number;
tools?: ChatToolDefinition[];
tool_choice?: ChatToolChoice;
parallel_tool_calls?: boolean;
}
export abstract class BaseProvider {
abstract readonly platform: Platform;
abstract readonly name: string;
/** Providers whose free tier needs no API key (e.g. Kilo's anonymous gateway).
* When true, the gateway stores a sentinel key row so routing still considers
* the platform "configured", and the provider omits the Authorization header
* on outgoing requests. Defaults to false; set by subclasses. */
keyless = false;
abstract chatCompletion(
apiKey: string,
messages: ChatMessage[],
modelId: string,
options?: CompletionOptions,
): Promise<ChatCompletionResponse>;
abstract streamChatCompletion(
apiKey: string,
messages: ChatMessage[],
modelId: string,
options?: CompletionOptions,
): AsyncGenerator<ChatCompletionChunk>;
abstract validateKey(apiKey: string): Promise<boolean>;
protected async fetchWithTimeout(
url: string,
init: RequestInit,
timeoutMs = 15000,
): Promise<Response> {
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), timeoutMs);
try {
return await fetch(url, { ...init, signal: controller.signal });
} finally {
clearTimeout(timeout);
}
}
protected makeId(): string {
return `chatcmpl-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
}
/**
* Shared SSE reader for OpenAI-wire streaming endpoints (#231 audit).
*
* Hardened against the upstream failure modes observed live:
* - Inactivity timeout: fetchWithTimeout's abort timer dies the moment
* response HEADERS arrive, so a provider that stalls mid-body used to
* hang the client forever. Each read now has its own deadline.
* - Abrupt EOF: a stream that ends without `[DONE]` AND without any
* `finish_reason` is a truncated generation, not a completion. It used
* to end the generator silently (truncation logged as success); it now
* throws a retryable error so the proxy can fail over or report it.
* Providers that skip `[DONE]` but do send a terminal finish_reason
* (several compat shims) still complete normally.
*
* Malformed data lines are skipped, matching previous behavior.
*/
protected async *readSseStream(
res: Response,
inactivityTimeoutMs = 90000,
): AsyncGenerator<ChatCompletionChunk> {
const reader = res.body?.getReader();
if (!reader) throw new Error('No response body');
const decoder = new TextDecoder();
let buffer = '';
let sawFinishReason = false;
try {
while (true) {
let timer: ReturnType<typeof setTimeout> | undefined;
const result = await Promise.race([
reader.read(),
new Promise<never>((_, reject) => {
timer = setTimeout(
() => reject(new Error(`${this.name} stream stalled: no data for ${inactivityTimeoutMs}ms (timeout)`)),
inactivityTimeoutMs,
);
}),
]).finally(() => clearTimeout(timer));
const { done, value } = result;
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split('\n');
buffer = lines.pop() ?? '';
for (const line of lines) {
const trimmed = line.trim();
if (!trimmed || !trimmed.startsWith('data: ')) continue;
const data = trimmed.slice(6);
if (data === '[DONE]') return;
try {
const chunk = JSON.parse(data) as ChatCompletionChunk;
if (chunk.choices?.some(c => c.finish_reason != null)) sawFinishReason = true;
yield chunk;
} catch {
// Skip malformed chunks
}
}
}
} finally {
reader.cancel().catch(() => { /* upstream already gone */ });
}
if (!sawFinishReason) {
throw new Error(`${this.name} stream ended unexpectedly (no [DONE], no finish_reason) — connection reset or truncated upstream`);
}
}
}
|