Repinoid
12d29ee98d
search(resourceBaseSegment): суффиксы-подресурсы перебираются по убыванию длины (KNOWN_RESOURCE_SUBS). Иначе "_user" матчился раньше "_sub_user" и nubes_s3_sub_user давал "s3_sub" вместо "s3" — страница ресурса не находилась. assist(llmChat): реально применяет ASSIST_LLM_TIMEOUT_MS — таймаут ОДНОГО вызова (покрывает и чтение тела ответа). Таймаут возвращает отдельный код llm_timeout, отмена вызывающим (дедлайн/закрытие соединения) — по-прежнему aborted, чтобы stage2 различал причины. stage1: убран дублирующий AbortController (таймаут теперь только в llmChat). orchestrator: llm_timeout -> HTTP 504.
489 lines
22 KiB
JavaScript
489 lines
22 KiB
JavaScript
'use strict';
|
||
|
||
/**
|
||
* assist.js — каркас AI-ассистента (чистая механика, без интеграции в server.js).
|
||
*
|
||
* Только встроенные модули Node.js (>= 18): http/https/fetch/AbortController.
|
||
* Никаких внешних npm-зависимостей.
|
||
*
|
||
* Экспортируемые функции:
|
||
* - validateInput(req)
|
||
* - createRateLimiter(opts)
|
||
* - createSemaphore(max)
|
||
* - llmChat({...})
|
||
* - normalizeQuery(q)
|
||
* - jsonError(res, status, code, message, headers)
|
||
*/
|
||
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
// Конфигурация из переменных окружения (с дефолтами)
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
const ENV = {
|
||
ASSIST_MAX_CONCURRENT: intEnv('ASSIST_MAX_CONCURRENT', 4),
|
||
ASSIST_TOTAL_DEADLINE_MS: intEnv('ASSIST_TOTAL_DEADLINE_MS', 40000),
|
||
ASSIST_LLM_TIMEOUT_MS: intEnv('ASSIST_LLM_TIMEOUT_MS', 25000),
|
||
ASSIST_RATE_PER_WINDOW: intEnv('ASSIST_RATE_PER_WINDOW', 6),
|
||
ASSIST_RATE_WINDOW_MS: intEnv('ASSIST_RATE_WINDOW_MS', 60000),
|
||
ASSIST_REPLICAS: intEnv('ASSIST_REPLICAS', 1),
|
||
ASSIST_MAX_BODY: intEnv('ASSIST_MAX_BODY', 2048),
|
||
LLM_BASE_URL: process.env.LLM_BASE_URL || '',
|
||
LLM_API_KEY: process.env.LLM_API_KEY || '',
|
||
LLM_MODEL: process.env.LLM_MODEL || 'gpt-oss-120b',
|
||
};
|
||
|
||
/** Читает целочисленную переменную окружения с фолбэком и нижней границей > 0. */
|
||
function intEnv(name, fallback) {
|
||
const raw = process.env[name];
|
||
if (raw === undefined || raw === null || raw === '') return fallback;
|
||
const n = Number.parseInt(raw, 10);
|
||
if (!Number.isFinite(n)) return fallback;
|
||
return n;
|
||
}
|
||
|
||
/** Маскирует API-ключ: показывает не более 4 символов, остальное `***`. */
|
||
function maskKey(key) {
|
||
if (!key) return '(empty)';
|
||
if (key.length <= 4) return '***';
|
||
return key.slice(0, 4) + '***';
|
||
}
|
||
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
// Чтение тела запроса с потоковым счётчиком байт
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
/**
|
||
* Читает поток (req) со счётчиком байт; при превышении maxBytes обрывает
|
||
* чтение через req.destroy() и резолвится {ok:false, code:413}.
|
||
*
|
||
* @param {import('http').IncomingMessage} req
|
||
* @param {{maxBytes:number}} opts
|
||
* @returns {Promise<{ok:true, body:string} | {ok:false, code:number, message:string}>}
|
||
*/
|
||
function readBodyWithLimit(req, { maxBytes }) {
|
||
return new Promise((resolve) => {
|
||
const chunks = [];
|
||
let total = 0;
|
||
let settled = false;
|
||
|
||
const done = (result) => {
|
||
if (settled) return;
|
||
settled = true;
|
||
resolve(result);
|
||
};
|
||
|
||
req.on('data', (chunk) => {
|
||
if (settled) return;
|
||
total += chunk.length;
|
||
if (total > maxBytes) {
|
||
// Превышен лимит в процессе чтения — обрываем поток.
|
||
req.destroy();
|
||
done({ ok: false, code: 413, message: 'body too large' });
|
||
return;
|
||
}
|
||
chunks.push(chunk);
|
||
});
|
||
|
||
req.on('end', () => {
|
||
done({ ok: true, body: Buffer.concat(chunks).toString('utf8') });
|
||
});
|
||
|
||
req.on('error', () => {
|
||
// На уже завершённом запросе — ничего не делаем.
|
||
if (settled) return;
|
||
done({ ok: false, code: 400, message: 'bad request body' });
|
||
});
|
||
});
|
||
}
|
||
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
// 1. validateInput
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
/**
|
||
* Валидирует POST-запрос: проверяет Content-Type, размер тела, читает и
|
||
* разбирает JSON, извлекает непустой `question` (или `query`).
|
||
*
|
||
* @param {import('http').IncomingMessage} req
|
||
* @returns {Promise<{ok:true, question:string} |
|
||
* {ok:false, code:number, message:string}>}
|
||
*/
|
||
async function validateInput(req) {
|
||
// Content-Type должен начинаться с application/json.
|
||
const contentType = req.headers['content-type'] || '';
|
||
if (!contentType.toLowerCase().startsWith('application/json')) {
|
||
return { ok: false, code: 415, message: 'unsupported media type' };
|
||
}
|
||
|
||
// Content-Length больше лимита — отказ БЕЗ чтения тела.
|
||
const maxBody = ENV.ASSIST_MAX_BODY;
|
||
const contentLengthHeader = req.headers['content-length'];
|
||
if (contentLengthHeader !== undefined) {
|
||
const declared = Number.parseInt(contentLengthHeader, 10);
|
||
if (Number.isFinite(declared) && declared > maxBody) {
|
||
return { ok: false, code: 413, message: 'body too large' };
|
||
}
|
||
}
|
||
|
||
// Читаем тело со счётчиком (обрыв при превышении в процессе).
|
||
const readResult = await readBodyWithLimit(req, { maxBytes: maxBody });
|
||
if (!readResult.ok) return readResult;
|
||
|
||
let parsed;
|
||
try {
|
||
parsed = JSON.parse(readResult.body);
|
||
} catch {
|
||
return { ok: false, code: 400, message: 'invalid json' };
|
||
}
|
||
|
||
// Извлекаем question/query; после trim — непустая строка.
|
||
const raw = (parsed && (parsed.question ?? parsed.query)) || '';
|
||
const question = typeof raw === 'string' ? raw.trim() : '';
|
||
if (!question) {
|
||
return { ok: false, code: 400, message: 'question is required' };
|
||
}
|
||
|
||
return { ok: true, question };
|
||
}
|
||
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
// 2. createRateLimiter
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
/**
|
||
* Фабрика in-memory rate-limiter со скользящим окном по ключу (например, IP).
|
||
*
|
||
* @param {{
|
||
* limit?:number, // базовый лимит запросов за окно
|
||
* windowMs?:number, // размер окна, мс
|
||
* replicas?:number, // делитель: реальный лимит = baseLimit / replicas
|
||
* cleanupIntervalMs?:number // период ленивой чистки устаревших ключей
|
||
* }} opts
|
||
* @returns {{check:(key:string)=>{ok:boolean, retryAfterSec?:number}}}
|
||
*/
|
||
function createRateLimiter(opts = {}) {
|
||
const {
|
||
limit = ENV.ASSIST_RATE_PER_WINDOW,
|
||
windowMs = ENV.ASSIST_RATE_WINDOW_MS,
|
||
replicas = ENV.ASSIST_REPLICAS,
|
||
cleanupIntervalMs = windowMs,
|
||
} = opts;
|
||
|
||
// Реальный лимит с учётом количества реплик (не меньше 1).
|
||
const effectiveLimit = Math.max(1, Math.floor(limit / Math.max(1, replicas)));
|
||
|
||
// key -> массив таймстемпов (мс) запросов в текущем окне.
|
||
const buckets = new Map();
|
||
|
||
/**
|
||
* Проверяет, можно ли пропустить запрос для ключа, и регистрирует его.
|
||
*
|
||
* @param {string} key
|
||
* @returns {{ok:boolean, retryAfterSec?:number}}
|
||
*/
|
||
function check(key) {
|
||
const now = Date.now();
|
||
const windowStart = now - windowMs;
|
||
|
||
let bucket = buckets.get(key);
|
||
if (!bucket) {
|
||
bucket = [];
|
||
buckets.set(key, bucket);
|
||
}
|
||
|
||
// Отбрасываем таймстемпы вне текущего окна (скользящее окно).
|
||
while (bucket.length && bucket[0] <= windowStart) {
|
||
bucket.shift();
|
||
}
|
||
|
||
if (bucket.length >= effectiveLimit) {
|
||
const oldest = bucket[0];
|
||
const retryAfterMs = Math.max(0, oldest + windowMs - now);
|
||
return { ok: false, retryAfterSec: Math.ceil(retryAfterMs / 1000) };
|
||
}
|
||
|
||
bucket.push(now);
|
||
return { ok: true };
|
||
}
|
||
|
||
// Периодическая чистка полностью устаревших ключей.
|
||
const cleanupTimer = setInterval(() => {
|
||
const now = Date.now();
|
||
const windowStart = now - windowMs;
|
||
for (const [key, bucket] of buckets) {
|
||
while (bucket.length && bucket[0] <= windowStart) {
|
||
bucket.shift();
|
||
}
|
||
if (bucket.length === 0) {
|
||
buckets.delete(key);
|
||
}
|
||
}
|
||
}, cleanupIntervalMs);
|
||
|
||
// Не даём интервалу удерживать процесс (для чистоты в тестах/CLI).
|
||
if (typeof cleanupTimer.unref === 'function') {
|
||
cleanupTimer.unref();
|
||
}
|
||
|
||
return { check };
|
||
}
|
||
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
// 3. createSemaphore
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
/**
|
||
* Простой неблокирующий семафор на max одновременных «вхождений».
|
||
*
|
||
* @param {number} max
|
||
* @returns {{tryAcquire:()=>boolean, release:()=>void}}
|
||
*/
|
||
function createSemaphore(max) {
|
||
const cap = Math.max(1, Math.floor(max));
|
||
let used = 0;
|
||
|
||
return {
|
||
/** Пытается занять слот. Вернёт false, если слотов нет. */
|
||
tryAcquire() {
|
||
if (used >= cap) return false;
|
||
used += 1;
|
||
return true;
|
||
},
|
||
/** Освобождает занятый слот. */
|
||
release() {
|
||
if (used > 0) used -= 1;
|
||
},
|
||
};
|
||
}
|
||
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
// 4. llmChat
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
/**
|
||
* Вызывает LLM через OpenAI-совместимый эндпоинт /v1/chat/completions.
|
||
*
|
||
* @param {{
|
||
* messages:Array, // массив сообщений {role, content}
|
||
* maxTokens?:number,
|
||
* controller?:AbortController, // сигнал отмены вызывающего (дедлайн/закрытие соединения)
|
||
* timeoutMs?:number, // таймаут ОДНОГО вызова ЛЛМ (по умолч. ASSIST_LLM_TIMEOUT_MS)
|
||
* baseUrl?:string,
|
||
* apiKey?:string,
|
||
* model?:string,
|
||
* temperature?:number,
|
||
* responseFormat?:{type:string} // напр. {type:'json_object'}
|
||
* usageRef?:object // мутабельный объект для сбора токенов {prompt,completion,total}
|
||
* }} params
|
||
* @returns {Promise<string>} — содержимое choices[0].message.content
|
||
*/
|
||
async function llmChat({
|
||
messages,
|
||
maxTokens,
|
||
controller,
|
||
timeoutMs = ENV.ASSIST_LLM_TIMEOUT_MS,
|
||
baseUrl = ENV.LLM_BASE_URL,
|
||
apiKey = ENV.LLM_API_KEY,
|
||
model = ENV.LLM_MODEL,
|
||
temperature,
|
||
responseFormat,
|
||
usageRef,
|
||
}) {
|
||
const base = String(baseUrl).replace(/\/+$/, '');
|
||
const url = `${base}/v1/chat/completions`;
|
||
|
||
const payload = { model, messages };
|
||
if (maxTokens !== undefined && maxTokens !== null) {
|
||
payload.max_tokens = maxTokens;
|
||
}
|
||
if (temperature !== undefined && temperature !== null) {
|
||
payload.temperature = temperature;
|
||
}
|
||
if (responseFormat !== undefined && responseFormat !== null) {
|
||
payload.response_format = responseFormat;
|
||
}
|
||
|
||
// Составной сигнал отмены. fetch принимает РОВНО ОДИН signal, поэтому, когда
|
||
// задан индивидуальный таймаут вызова (ASSIST_LLM_TIMEOUT_MS), создаём
|
||
// дочерний AbortController, который абортится ЛИБО по таймауту, ЛИБО при
|
||
// аборте вызывающего (дедлайн всего запроса / закрытие соединения клиентом).
|
||
// Без таймаута используем сигнал вызывающего напрямую. Дочерний контроллер
|
||
// покрывает ВЕСЬ вызов, включая чтение тела ответа (response.json()).
|
||
let signal = controller ? controller.signal : undefined;
|
||
let timeoutTimer = null;
|
||
let onParentAbort = null;
|
||
if (timeoutMs && Number.isFinite(timeoutMs) && timeoutMs > 0) {
|
||
const ownController = new AbortController();
|
||
signal = ownController.signal;
|
||
if (controller) {
|
||
onParentAbort = () => ownController.abort();
|
||
if (controller.signal.aborted) ownController.abort();
|
||
else controller.signal.addEventListener('abort', onParentAbort, { once: true });
|
||
}
|
||
timeoutTimer = setTimeout(() => ownController.abort(), timeoutMs);
|
||
}
|
||
|
||
try {
|
||
let response;
|
||
try {
|
||
response = await fetch(url, {
|
||
method: 'POST',
|
||
headers: {
|
||
Authorization: `Bearer ${apiKey}`,
|
||
'Content-Type': 'application/json',
|
||
},
|
||
body: JSON.stringify(payload),
|
||
signal,
|
||
});
|
||
} catch (err) {
|
||
if (err && err.name === 'AbortError') {
|
||
// Различаем причину: отмена вызывающим vs. собственный таймаут
|
||
// вызова (иначе stage2 не смог бы отличить таймаут от ухода
|
||
// клиента и корректно решить, повторять ли вызов).
|
||
if (controller && controller.signal.aborted) {
|
||
throw { code: 'aborted', message: 'llm request aborted (AbortError)' };
|
||
}
|
||
throw { code: 'llm_timeout', message: `llm call exceeded timeout ${timeoutMs} ms` };
|
||
}
|
||
throw new Error('llm request failed (network)');
|
||
}
|
||
|
||
// Не-200 → ошибка со статусом (без apiKey в тексте!).
|
||
if (!response.ok) {
|
||
const status = response.status;
|
||
const retryAfter = response.headers.get('retry-after');
|
||
const err = { code: 'llm_http_error', status, retryAfter: undefined };
|
||
|
||
if (status === 429) {
|
||
err.code = 'rate_limited';
|
||
// retry-after может быть секундами или HTTP-датой.
|
||
if (retryAfter) {
|
||
const secs = parseRetryAfter(retryAfter);
|
||
if (secs !== null) err.retryAfter = secs;
|
||
}
|
||
}
|
||
|
||
// Читаем тело ошибки, но никогда не упоминаем apiKey.
|
||
let detail = '';
|
||
try {
|
||
detail = await response.text();
|
||
if (detail && detail.length > 500) detail = detail.slice(0, 500);
|
||
} catch {
|
||
detail = '';
|
||
}
|
||
if (detail) err.message = `llm error ${status}: ${detail}`;
|
||
else err.message = `llm error ${status}`;
|
||
|
||
throw err;
|
||
}
|
||
|
||
const data = await response.json();
|
||
|
||
// Собираем токены (usage) для отображения пользователю.
|
||
if (usageRef && data && data.usage) {
|
||
usageRef.prompt = (usageRef.prompt || 0) + (data.usage.prompt_tokens || 0);
|
||
usageRef.completion = (usageRef.completion || 0) + (data.usage.completion_tokens || 0);
|
||
usageRef.total = (usageRef.total || 0) + (data.usage.total_tokens || 0);
|
||
}
|
||
|
||
const content = data && data.choices && data.choices[0] &&
|
||
data.choices[0].message && data.choices[0].message.content;
|
||
|
||
if (typeof content !== 'string' || content.trim() === '') {
|
||
throw { code: 'empty_content', message: 'llm returned empty content' };
|
||
}
|
||
|
||
return content;
|
||
} catch (err) {
|
||
// Аборт мог прилететь не только из fetch, но и при чтении тела ответа
|
||
// (response.json() выше) — нормализуем и этот случай тем же образом.
|
||
if (err && err.name === 'AbortError') {
|
||
if (controller && controller.signal.aborted) {
|
||
throw { code: 'aborted', message: 'llm request aborted (AbortError)' };
|
||
}
|
||
throw { code: 'llm_timeout', message: `llm call exceeded timeout ${timeoutMs} ms` };
|
||
}
|
||
throw err;
|
||
} finally {
|
||
// Снимаем таймер и слушателя родителя на всех путях (успех/ошибка),
|
||
// чтобы не оставлять висящих таймеров и подписок на сигнале.
|
||
if (timeoutTimer) clearTimeout(timeoutTimer);
|
||
if (controller && onParentAbort) controller.signal.removeEventListener('abort', onParentAbort);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Разбирает значение заголовка Retry-After (секунды или HTTP-дата) в секунды.
|
||
*
|
||
* @param {string} value
|
||
* @returns {number|null}
|
||
*/
|
||
function parseRetryAfter(value) {
|
||
if (!value) return null;
|
||
const num = Number.parseInt(value, 10);
|
||
if (Number.isFinite(num)) return num;
|
||
const date = Date.parse(value);
|
||
if (Number.isFinite(date)) {
|
||
return Math.max(0, Math.ceil((date - Date.now()) / 1000));
|
||
}
|
||
return null;
|
||
}
|
||
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
// 5. normalizeQuery
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
/**
|
||
* Нормализует запрос для использования в качестве ключа кэша:
|
||
* lowercase, схлопывание пробелов, обрезка краёв.
|
||
*
|
||
* @param {string} q
|
||
* @returns {string}
|
||
*/
|
||
function normalizeQuery(q) {
|
||
return String(q).toLowerCase().replace(/\s+/g, ' ').trim();
|
||
}
|
||
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
// 6. jsonError
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
/**
|
||
* Пишет JSON-ошибку вида {"error":code,"message":...} в HTTP-ответ.
|
||
*
|
||
* @param {import('http').ServerResponse} res
|
||
* @param {number} status
|
||
* @param {string|number} code
|
||
* @param {string} message
|
||
* @param {Record<string,string|number>|object} [headers]
|
||
*/
|
||
function jsonError(res, status, code, message, headers = {}) {
|
||
const body = JSON.stringify({ error: code, message });
|
||
const finalHeaders = {
|
||
'Content-Type': 'application/json; charset=utf-8',
|
||
'Content-Length': Buffer.byteLength(body),
|
||
...headers,
|
||
};
|
||
if (!res.headersSent) {
|
||
res.writeHead(status, finalHeaders);
|
||
}
|
||
res.end(body);
|
||
}
|
||
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
// Экспорт
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
module.exports = {
|
||
validateInput,
|
||
createRateLimiter,
|
||
createSemaphore,
|
||
llmChat,
|
||
normalizeQuery,
|
||
jsonError,
|
||
};
|
||
|
||
// Дополнительно: ENV и maskKey доступны для внутренних нужд/тестов.
|
||
module.exports.ENV = ENV;
|
||
module.exports.maskKey = maskKey;
|