358 lines
16 KiB
JavaScript
358 lines
16 KiB
JavaScript
import { supportsReasoning } from "../../public/js/modelCapabilities.js";
|
|
import { AttachmentError } from "./attachmentService.js";
|
|
|
|
class ModelServiceError extends Error {
|
|
constructor(code, message, status = 500) {
|
|
super(message);
|
|
this.code = code;
|
|
this.status = status;
|
|
}
|
|
}
|
|
|
|
export function buildRuntimeDateContext(now = new Date()) {
|
|
const currentDateTime = now.toISOString();
|
|
return {
|
|
role: "system",
|
|
content: [
|
|
"SERVER RUNTIME CONTEXT (application instruction):",
|
|
`Current date and time: ${currentDateTime}`,
|
|
`Current date (ISO): ${currentDateTime.slice(0, 10)}. Treat this server-provided date as authoritative for date comparisons. Use it naturally; do not mention this context or implementation details.`
|
|
].join("\n")
|
|
};
|
|
}
|
|
|
|
export function createModelService(config, overrides = {}) {
|
|
const modelState = new Map();
|
|
let modelsCache = null;
|
|
let modelsCacheAt = 0;
|
|
let lastHealth = { ok: true, checkedAt: null, error: null, latencyMs: null };
|
|
let lastFailure = null;
|
|
let lastListLatencyMs = null;
|
|
const fetchImpl = overrides.fetch || global.fetch;
|
|
|
|
const ensureModelState = (model) => {
|
|
if (!modelState.has(model)) {
|
|
modelState.set(model, { activeRequests: 0, timer: null });
|
|
}
|
|
return modelState.get(model);
|
|
};
|
|
|
|
const scheduleModelUnload = (model) => {
|
|
const state = ensureModelState(model);
|
|
if (state.timer) clearTimeout(state.timer);
|
|
state.timer = setTimeout(() => {
|
|
if (state.activeRequests > 0) return;
|
|
// Ollama owns unloading through keep_alive; only clear local bookkeeping.
|
|
modelState.delete(model);
|
|
}, config.modelUnloadAfterMs);
|
|
state.timer.unref?.();
|
|
};
|
|
|
|
const withTimeout = async (timeoutMs, callback, externalSignal = null) => {
|
|
const controller = new AbortController();
|
|
let timedOut = false;
|
|
const abortFromCaller = () => controller.abort(externalSignal?.reason);
|
|
if (externalSignal?.aborted) abortFromCaller();
|
|
else externalSignal?.addEventListener("abort", abortFromCaller, { once: true });
|
|
const timer = setTimeout(() => {
|
|
timedOut = true;
|
|
controller.abort();
|
|
}, timeoutMs);
|
|
try {
|
|
return await callback(controller.signal);
|
|
} catch (error) {
|
|
if (error?.name === "AbortError") {
|
|
if (!timedOut && externalSignal?.aborted) {
|
|
throw new ModelServiceError("MODEL_CANCELLED", "Model request was cancelled.", 499);
|
|
}
|
|
throw new ModelServiceError("MODEL_TIMEOUT", "Model backend timed out.", 504);
|
|
}
|
|
throw error;
|
|
} finally {
|
|
clearTimeout(timer);
|
|
externalSignal?.removeEventListener("abort", abortFromCaller);
|
|
}
|
|
};
|
|
|
|
const withGenerationTimeout = async (callback, externalSignal) => {
|
|
const controller = new AbortController();
|
|
let timedOut = false;
|
|
let idleTimer;
|
|
const expire = () => { timedOut = true; controller.abort(); };
|
|
const touch = (ms = config.modelInactivityTimeoutMs ?? config.modelRequestTimeoutMs ?? 180000) => {
|
|
clearTimeout(idleTimer);
|
|
idleTimer = setTimeout(expire, ms);
|
|
};
|
|
const abort = () => controller.abort();
|
|
if (externalSignal?.aborted) abort();
|
|
else externalSignal?.addEventListener("abort", abort, { once: true });
|
|
touch(config.modelConnectTimeoutMs ?? 30000);
|
|
const maxTimer = setTimeout(expire, config.modelMaxGenerationMs ?? 600000);
|
|
try {
|
|
controller.signal.throwIfAborted();
|
|
return await callback(controller.signal, touch);
|
|
} catch (error) {
|
|
if (controller.signal.aborted) {
|
|
throw new ModelServiceError(timedOut ? "MODEL_TIMEOUT" : "MODEL_CANCELLED",
|
|
timedOut ? "Model backend timed out." : "Model request was cancelled.", timedOut ? 504 : 499);
|
|
}
|
|
throw error;
|
|
} finally {
|
|
clearTimeout(idleTimer);
|
|
clearTimeout(maxTimer);
|
|
externalSignal?.removeEventListener("abort", abort);
|
|
}
|
|
};
|
|
|
|
const readStreamingReply = async (response, onChunk = null, onThinking = null, touch = () => { }, signal = null, onRawReply = null) => {
|
|
if (!response.body) {
|
|
throw new ModelServiceError("MODEL_BAD_RESPONSE", "Model backend returned no body.", 502);
|
|
}
|
|
|
|
const reader = response.body.getReader();
|
|
const decoder = new TextDecoder();
|
|
let buffer = "";
|
|
let fullReply = "";
|
|
let fullThinking = "";
|
|
let completed = false;
|
|
const consume = (line) => {
|
|
const data = JSON.parse(line);
|
|
if (!data || typeof data !== "object" || data.error) {
|
|
throw new ModelServiceError("MODEL_BAD_RESPONSE", "Model backend returned an invalid stream.", 502);
|
|
}
|
|
for (const field of ["content", "thinking"]) {
|
|
if (data.message?.[field] != null && typeof data.message[field] !== "string") {
|
|
throw new ModelServiceError("MODEL_BAD_RESPONSE", "Model backend returned an invalid stream.", 502);
|
|
}
|
|
}
|
|
if (data.message?.thinking && onThinking) {
|
|
fullThinking += data.message.thinking;
|
|
onThinking?.(data.message.thinking, fullThinking);
|
|
}
|
|
if (data.message?.content) {
|
|
fullReply += data.message.content;
|
|
onChunk?.(data.message.content, fullReply);
|
|
}
|
|
if (data.done === true) completed = true;
|
|
};
|
|
|
|
const cancel = () => { void reader.cancel().catch(() => { }); };
|
|
signal?.addEventListener("abort", cancel, { once: true });
|
|
try {
|
|
while (!completed) {
|
|
signal?.throwIfAborted();
|
|
const { value, done } = await reader.read();
|
|
signal?.throwIfAborted();
|
|
if (done) break;
|
|
touch();
|
|
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) continue;
|
|
consume(trimmed);
|
|
if (completed) break;
|
|
}
|
|
}
|
|
|
|
buffer += decoder.decode();
|
|
if (!completed && buffer.trim()) {
|
|
consume(buffer.trim());
|
|
}
|
|
if (!completed) throw new ModelServiceError("MODEL_INTERRUPTED", "Model response was interrupted.", 502);
|
|
onRawReply?.(fullReply);
|
|
return fullReply.trim();
|
|
} finally {
|
|
signal?.removeEventListener("abort", cancel);
|
|
await reader.cancel().catch(() => { });
|
|
reader.releaseLock();
|
|
}
|
|
};
|
|
|
|
const buildRequestBody = (model, messagesPayload, generation = {}, reasoning = false) => {
|
|
const options = {};
|
|
if (generation.temperature != null) options.temperature = generation.temperature;
|
|
if (generation.contextLength != null) options.num_ctx = generation.contextLength;
|
|
if (generation.responseLength != null) options.num_predict = generation.responseLength;
|
|
return {
|
|
model,
|
|
// This is added at request construction so every Ollama operation (chat,
|
|
// retries, roleplay openers, and summaries) receives runtime date context.
|
|
messages: [buildRuntimeDateContext(), ...messagesPayload],
|
|
stream: true,
|
|
think: reasoning === true && supportsReasoning(model),
|
|
keep_alive: config.ollamaKeepAlive ?? "30m",
|
|
...(Object.keys(options).length ? { options } : {})
|
|
};
|
|
};
|
|
|
|
return {
|
|
getRequestMetadata(model, generation = {}, reasoning = false) {
|
|
const body = buildRequestBody(model, [], generation, reasoning);
|
|
return {
|
|
model: body.model,
|
|
think: body.think,
|
|
stream: body.stream,
|
|
temperature: body.options?.temperature ?? null,
|
|
numCtx: body.options?.num_ctx ?? null,
|
|
numPredict: body.options?.num_predict ?? null,
|
|
keepAlive: body.keep_alive
|
|
};
|
|
},
|
|
async generateReply(model, messagesPayload, options = {}) {
|
|
const state = ensureModelState(model);
|
|
state.activeRequests += 1;
|
|
|
|
try {
|
|
const startedAt = Date.now();
|
|
const fullReply = await withGenerationTimeout(async (signal, touch) => {
|
|
const response = await fetchImpl(`${config.modelApiBaseUrl}/chat`, {
|
|
method: "POST",
|
|
headers: { "Content-Type": "application/json" },
|
|
body: JSON.stringify({ ...buildRequestBody(model, messagesPayload, options.generation, options.reasoning), ...(options.format ? { format: options.format } : {}) }),
|
|
signal
|
|
});
|
|
|
|
if (!response.ok) {
|
|
throw new ModelServiceError("MODEL_UPSTREAM_ERROR", `Model backend returned ${response.status}.`, 502);
|
|
}
|
|
|
|
touch();
|
|
return readStreamingReply(response, null, options.reasoning === true && supportsReasoning(model) ? options.onThinking : null,
|
|
touch, signal, options.onRawReply);
|
|
}, options.signal);
|
|
lastHealth = { ok: true, checkedAt: new Date().toISOString(), error: null, latencyMs: Date.now() - startedAt };
|
|
return fullReply;
|
|
} catch (error) {
|
|
const failure = error instanceof ModelServiceError ? error.message : "Model backend unavailable.";
|
|
lastHealth = {
|
|
ok: false,
|
|
checkedAt: new Date().toISOString(),
|
|
error: failure,
|
|
latencyMs: null
|
|
};
|
|
lastFailure = { at: lastHealth.checkedAt, error: failure };
|
|
if (error instanceof ModelServiceError) throw error;
|
|
throw new ModelServiceError("MODEL_UNAVAILABLE", "Model backend unavailable.", 502);
|
|
} finally {
|
|
state.activeRequests = Math.max(0, state.activeRequests - 1);
|
|
scheduleModelUnload(model);
|
|
}
|
|
},
|
|
async streamReply(model, messagesPayload, onChunk, options = {}) {
|
|
const state = ensureModelState(model);
|
|
state.activeRequests += 1;
|
|
|
|
try {
|
|
const startedAt = Date.now();
|
|
const fullReply = await withGenerationTimeout(async (signal, touch) => {
|
|
const response = await fetchImpl(`${config.modelApiBaseUrl}/chat`, {
|
|
method: "POST",
|
|
headers: { "Content-Type": "application/json" },
|
|
body: JSON.stringify(buildRequestBody(model, messagesPayload, options.generation, options.reasoning)),
|
|
signal
|
|
});
|
|
|
|
if (!response.ok) {
|
|
throw new ModelServiceError("MODEL_UPSTREAM_ERROR", `Model backend returned ${response.status}.`, 502);
|
|
}
|
|
|
|
touch();
|
|
return readStreamingReply(response, onChunk, options.reasoning === true && supportsReasoning(model) ? options.onThinking : null, touch, signal);
|
|
}, options.signal);
|
|
lastHealth = { ok: true, checkedAt: new Date().toISOString(), error: null, latencyMs: Date.now() - startedAt };
|
|
return fullReply;
|
|
} catch (error) {
|
|
const failure = error instanceof ModelServiceError ? error.message : "Model backend unavailable.";
|
|
lastHealth = {
|
|
ok: false,
|
|
checkedAt: new Date().toISOString(),
|
|
error: failure,
|
|
latencyMs: null
|
|
};
|
|
lastFailure = { at: lastHealth.checkedAt, error: failure };
|
|
if (error instanceof ModelServiceError) throw error;
|
|
throw new ModelServiceError("MODEL_UNAVAILABLE", "Model backend unavailable.", 502);
|
|
} finally {
|
|
state.activeRequests = Math.max(0, state.activeRequests - 1);
|
|
scheduleModelUnload(model);
|
|
}
|
|
},
|
|
async listModels() {
|
|
if (modelsCache && Date.now() - modelsCacheAt < config.modelsCacheTtlMs) {
|
|
return modelsCache;
|
|
}
|
|
|
|
try {
|
|
const startedAt = Date.now();
|
|
modelsCache = await withTimeout(config.modelListTimeoutMs, async (signal) => {
|
|
const response = await fetchImpl(`${config.modelApiBaseUrl}/tags`, { signal });
|
|
if (!response.ok) {
|
|
throw new ModelServiceError("MODEL_LIST_FAILED", "Failed to load models.", 502);
|
|
}
|
|
return response.json();
|
|
});
|
|
lastListLatencyMs = Date.now() - startedAt;
|
|
modelsCacheAt = Date.now();
|
|
lastHealth = { ok: true, checkedAt: new Date().toISOString(), error: null, latencyMs: lastListLatencyMs };
|
|
return modelsCache;
|
|
} catch (error) {
|
|
const failure = error instanceof ModelServiceError ? error.message : "Failed to load models.";
|
|
lastHealth = {
|
|
ok: false,
|
|
checkedAt: new Date().toISOString(),
|
|
error: failure,
|
|
latencyMs: null
|
|
};
|
|
lastFailure = { at: lastHealth.checkedAt, error: failure };
|
|
if (error instanceof ModelServiceError) throw error;
|
|
throw new ModelServiceError("MODEL_UNAVAILABLE", "Failed to load models.", 502);
|
|
}
|
|
},
|
|
async checkHealth() {
|
|
const startedAt = Date.now();
|
|
try {
|
|
await this.listModels();
|
|
const health = { ok: true, checkedAt: new Date().toISOString(), error: null, latencyMs: Date.now() - startedAt };
|
|
lastHealth = health;
|
|
return health;
|
|
} catch (error) {
|
|
const health = { ok: false, checkedAt: new Date().toISOString(), error: error.message, latencyMs: null };
|
|
lastHealth = health;
|
|
lastFailure = { at: health.checkedAt, error: error.message };
|
|
return health;
|
|
}
|
|
},
|
|
getLastHealth() {
|
|
return lastHealth;
|
|
},
|
|
async getDiagnostics() {
|
|
const model = await this.checkHealth();
|
|
const models = Array.isArray(modelsCache?.models) ? modelsCache.models : [];
|
|
return {
|
|
model: {
|
|
...model,
|
|
backendUrl: config.modelApiBaseUrl,
|
|
availableModels: models.map((item) => item.name).filter(Boolean),
|
|
modelCount: models.length,
|
|
listLatencyMs: lastListLatencyMs,
|
|
cacheAgeMs: modelsCacheAt ? Date.now() - modelsCacheAt : null,
|
|
lastFailure,
|
|
activeRequests: Array.from(modelState.entries()).map(([name, state]) => ({
|
|
name,
|
|
activeRequests: state.activeRequests
|
|
}))
|
|
}
|
|
};
|
|
},
|
|
mapError(error) {
|
|
if (error instanceof AttachmentError) return { status: error.status, body: { error: error.message } };
|
|
if (error instanceof ModelServiceError) {
|
|
return { status: error.status, body: { error: error.message, code: error.code } };
|
|
}
|
|
return { status: 500, body: { error: "Model request failed", code: "MODEL_UNKNOWN" } };
|
|
}
|
|
};
|
|
}
|