997 lines
38 KiB
TypeScript
997 lines
38 KiB
TypeScript
import { inject } from '@adonisjs/core'
|
||
import OpenAI from 'openai'
|
||
import type { ChatCompletionChunk, ChatCompletionMessageParam } from 'openai/resources/chat/completions.js'
|
||
import type { Stream } from 'openai/streaming.js'
|
||
import { NomadOllamaModel } from '../../types/ollama.js'
|
||
import { EMBEDDING_MODEL_NAME, FALLBACK_RECOMMENDED_OLLAMA_MODELS } from '../../constants/ollama.js'
|
||
import fs from 'node:fs/promises'
|
||
import path from 'node:path'
|
||
import logger from '@adonisjs/core/services/logger'
|
||
import axios from 'axios'
|
||
import { DownloadModelJob } from '#jobs/download_model_job'
|
||
import { SERVICE_NAMES } from '../../constants/service_names.js'
|
||
import transmit from '@adonisjs/transmit/services/main'
|
||
import Fuse, { IFuseOptions } from 'fuse.js'
|
||
import { BROADCAST_CHANNELS } from '../../constants/broadcast.js'
|
||
import env from '#start/env'
|
||
import { NOMAD_API_DEFAULT_BASE_URL } from '../../constants/misc.js'
|
||
import KVStore from '#models/kv_store'
|
||
|
||
const NOMAD_MODELS_API_PATH = '/api/v1/ollama/models'
|
||
const MODELS_CACHE_FILE = path.join(process.cwd(), 'storage', 'ollama-models-cache.json')
|
||
const CACHE_MAX_AGE_MS = 24 * 60 * 60 * 1000 // 24 hours
|
||
|
||
export type NomadInstalledModel = {
|
||
name: string
|
||
size: number
|
||
digest?: string
|
||
details?: Record<string, any>
|
||
}
|
||
|
||
export type NomadChatResponse = {
|
||
message: { content: string; thinking?: string }
|
||
done: boolean
|
||
model: string
|
||
}
|
||
|
||
export type NomadChatStreamChunk = {
|
||
message: { content: string; thinking?: string }
|
||
done: boolean
|
||
}
|
||
|
||
type ChatInput = {
|
||
model: string
|
||
messages: Array<{ role: 'system' | 'user' | 'assistant'; content: string }>
|
||
think?: boolean | 'medium'
|
||
stream?: boolean
|
||
numCtx?: number
|
||
}
|
||
|
||
@inject()
|
||
export class OllamaService {
|
||
private openai: OpenAI | null = null
|
||
private baseUrl: string | null = null
|
||
private initPromise: Promise<void> | null = null
|
||
private isOllamaNative: boolean | null = null
|
||
private activeDownloads: Map<string, Promise<{ success: boolean; message: string; retryable?: boolean }>> = new Map()
|
||
|
||
constructor() {}
|
||
|
||
private async _initialize() {
|
||
if (!this.initPromise) {
|
||
this.initPromise = (async () => {
|
||
// Check KVStore for a custom base URL (remote Ollama, LM Studio, llama.cpp, etc.)
|
||
const customUrl = (await KVStore.getValue('ai.remoteOllamaUrl')) as string | null
|
||
if (customUrl && customUrl.trim()) {
|
||
this.baseUrl = customUrl.trim().replace(/\/$/, '')
|
||
} else {
|
||
// Fall back to the local Ollama container managed by Docker
|
||
const dockerService = new (await import('./docker_service.js')).DockerService()
|
||
const ollamaUrl = await dockerService.getServiceURL(SERVICE_NAMES.OLLAMA)
|
||
if (!ollamaUrl) {
|
||
throw new Error('Ollama service is not installed or running.')
|
||
}
|
||
this.baseUrl = ollamaUrl.trim().replace(/\/$/, '')
|
||
}
|
||
|
||
this.openai = new OpenAI({
|
||
apiKey: 'nomad', // Required by SDK; not validated by Ollama/LM Studio/llama.cpp
|
||
baseURL: `${this.baseUrl}/v1`,
|
||
})
|
||
})()
|
||
}
|
||
return this.initPromise
|
||
}
|
||
|
||
private async _ensureDependencies() {
|
||
if (!this.openai) {
|
||
await this._initialize()
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Downloads a model from Ollama with progress tracking. Only works with Ollama backends.
|
||
* Use dispatchModelDownload() for background job processing where possible.
|
||
*
|
||
* @param signal Optional AbortSignal — when triggered, the underlying axios stream is cancelled
|
||
* and the method returns a non-retryable failure so callers can mark the job
|
||
* unrecoverable in BullMQ and avoid the 40-attempt retry storm.
|
||
* @param jobId Optional BullMQ job id — included in progress broadcasts so the frontend can
|
||
* correlate Transmit events to a cancellable job.
|
||
*/
|
||
async downloadModel(
|
||
model: string,
|
||
progressCallback?: (
|
||
percent: number,
|
||
bytes?: { downloadedBytes: number; totalBytes: number }
|
||
) => void,
|
||
signal?: AbortSignal,
|
||
jobId?: string
|
||
): Promise<{ success: boolean; message: string; retryable?: boolean }> {
|
||
// Deduplicate concurrent downloads of the same model
|
||
const existing = this.activeDownloads.get(model)
|
||
if (existing) {
|
||
logger.info(`[OllamaService] Download already in progress for "${model}", waiting on existing download.`)
|
||
return existing
|
||
}
|
||
|
||
const downloadPromise = this._doDownloadModel(model, progressCallback, signal, jobId)
|
||
this.activeDownloads.set(model, downloadPromise)
|
||
try {
|
||
return await downloadPromise
|
||
} finally {
|
||
this.activeDownloads.delete(model)
|
||
}
|
||
}
|
||
|
||
private async _doDownloadModel(
|
||
model: string,
|
||
progressCallback?: (
|
||
percent: number,
|
||
bytes?: { downloadedBytes: number; totalBytes: number }
|
||
) => void,
|
||
signal?: AbortSignal,
|
||
jobId?: string
|
||
): Promise<{ success: boolean; message: string; retryable?: boolean }> {
|
||
await this._ensureDependencies()
|
||
if (!this.baseUrl) {
|
||
return { success: false, message: 'AI service is not initialized.' }
|
||
}
|
||
|
||
try {
|
||
// See if model is already installed
|
||
const installedModels = await this.getModels()
|
||
if (installedModels && installedModels.some((m) => m.name === model)) {
|
||
logger.info(`[OllamaService] Model "${model}" is already installed.`)
|
||
return { success: true, message: 'Model is already installed.' }
|
||
}
|
||
|
||
// Model pulling is an Ollama-only operation. Non-Ollama backends (LM Studio, llama.cpp, etc.)
|
||
// return HTTP 200 for unknown endpoints, so the pull would appear to succeed but do nothing.
|
||
if (this.isOllamaNative === false) {
|
||
logger.warn(
|
||
`[OllamaService] Non-Ollama backend detected — skipping model pull for "${model}". Load the model manually in your AI host.`
|
||
)
|
||
return {
|
||
success: false,
|
||
message: `Model "${model}" is not available in your AI host. Please load it manually (model pulling is only supported for Ollama backends).`,
|
||
}
|
||
}
|
||
|
||
// Stream pull via Ollama native API. axios supports `signal` natively for AbortController
|
||
// integration — when triggered, the request errors with code 'ERR_CANCELED' which we detect
|
||
// in the catch block below to return a non-retryable cancel result.
|
||
const pullResponse = await axios.post(
|
||
`${this.baseUrl}/api/pull`,
|
||
{ model, stream: true },
|
||
{ responseType: 'stream', timeout: 0, signal }
|
||
)
|
||
|
||
// Ollama's pull API reports progress per-digest (each blob). A single model can contain
|
||
// multiple blobs (weights, tokenizer, template, etc.) and each is reported in turn.
|
||
// Aggregate across all digests so the UI shows a single monotonically-increasing total,
|
||
// matching the behavior of the content download progress (Active Downloads section).
|
||
const digestProgress = new Map<string, { completed: number; total: number }>()
|
||
|
||
// Throttle broadcasts to once per BROADCAST_THROTTLE_MS — Ollama can emit hundreds of
|
||
// progress events per second for fast connections, which would flood the Transmit SSE
|
||
// channel and cause jittery speed calculations on the frontend.
|
||
const BROADCAST_THROTTLE_MS = 500
|
||
let lastBroadcastAt = 0
|
||
|
||
await new Promise<void>((resolve, reject) => {
|
||
let buffer = ''
|
||
// If the abort fires after headers are received but mid-stream, axios's signal handling
|
||
// destroys the stream which surfaces as an 'error' event — wire the signal listener so
|
||
// the promise rejects promptly with a recognizable cancel reason.
|
||
const onAbort = () => {
|
||
const err: any = new Error('Download cancelled')
|
||
err.code = 'ERR_CANCELED'
|
||
pullResponse.data.destroy(err)
|
||
}
|
||
if (signal) {
|
||
if (signal.aborted) {
|
||
onAbort()
|
||
return
|
||
}
|
||
signal.addEventListener('abort', onAbort, { once: true })
|
||
}
|
||
|
||
pullResponse.data.on('data', (chunk: Buffer) => {
|
||
buffer += chunk.toString()
|
||
const lines = buffer.split('\n')
|
||
buffer = lines.pop() || ''
|
||
for (const line of lines) {
|
||
if (!line.trim()) continue
|
||
try {
|
||
const parsed = JSON.parse(line)
|
||
if (parsed.completed && parsed.total && parsed.digest) {
|
||
// Update this digest's progress — take the max seen value so transient
|
||
// out-of-order updates don't make the aggregate jump backwards.
|
||
const existing = digestProgress.get(parsed.digest)
|
||
digestProgress.set(parsed.digest, {
|
||
completed: Math.max(existing?.completed ?? 0, parsed.completed),
|
||
total: Math.max(existing?.total ?? 0, parsed.total),
|
||
})
|
||
|
||
// Compute aggregate across all known blobs
|
||
let aggCompleted = 0
|
||
let aggTotal = 0
|
||
for (const { completed, total } of digestProgress.values()) {
|
||
aggCompleted += completed
|
||
aggTotal += total
|
||
}
|
||
|
||
const percent = aggTotal > 0
|
||
? parseFloat(((aggCompleted / aggTotal) * 100).toFixed(2))
|
||
: 0
|
||
|
||
// Throttle broadcasts. Always call the progressCallback though — the worker
|
||
// uses it to update job state in Redis, which should reflect the latest view.
|
||
const now = Date.now()
|
||
if (now - lastBroadcastAt >= BROADCAST_THROTTLE_MS) {
|
||
lastBroadcastAt = now
|
||
this.broadcastDownloadProgress(model, percent, jobId, {
|
||
downloadedBytes: aggCompleted,
|
||
totalBytes: aggTotal,
|
||
})
|
||
}
|
||
if (progressCallback) {
|
||
progressCallback(percent, {
|
||
downloadedBytes: aggCompleted,
|
||
totalBytes: aggTotal,
|
||
})
|
||
}
|
||
}
|
||
} catch {
|
||
// ignore parse errors on partial lines
|
||
}
|
||
}
|
||
})
|
||
pullResponse.data.on('end', () => {
|
||
if (signal) signal.removeEventListener('abort', onAbort)
|
||
resolve()
|
||
})
|
||
pullResponse.data.on('error', (err: any) => {
|
||
if (signal) signal.removeEventListener('abort', onAbort)
|
||
reject(err)
|
||
})
|
||
})
|
||
|
||
logger.info(`[OllamaService] Model "${model}" downloaded successfully.`)
|
||
return { success: true, message: 'Model downloaded successfully.' }
|
||
} catch (error) {
|
||
// Detect axios cancel (signal-triggered abort). Don't broadcast an error event for
|
||
// user-initiated cancels — the cancel handler in DownloadService already broadcasts
|
||
// a cancelled state. Returning retryable: false prevents BullMQ retries.
|
||
const isCancelled =
|
||
axios.isCancel(error) ||
|
||
(error as any)?.code === 'ERR_CANCELED' ||
|
||
(error as any)?.name === 'CanceledError'
|
||
if (isCancelled) {
|
||
logger.info(`[OllamaService] Model "${model}" download cancelled by user.`)
|
||
return { success: false, message: 'Download cancelled', retryable: false }
|
||
}
|
||
|
||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||
logger.error(
|
||
`[OllamaService] Failed to download model "${model}": ${errorMessage}`
|
||
)
|
||
|
||
// Check for version mismatch (Ollama 412 response)
|
||
const isVersionMismatch = errorMessage.includes('newer version of Ollama')
|
||
const userMessage = isVersionMismatch
|
||
? 'This model requires a newer version of Ollama. Please update AI Assistant from the Apps page.'
|
||
: `Failed to download model: ${errorMessage}`
|
||
|
||
// Broadcast failure to connected clients so UI can show the error
|
||
this.broadcastDownloadError(model, userMessage)
|
||
|
||
return { success: false, message: userMessage, retryable: !isVersionMismatch }
|
||
}
|
||
}
|
||
|
||
async dispatchModelDownload(modelName: string): Promise<{ success: boolean; message: string }> {
|
||
try {
|
||
logger.info(`[OllamaService] Dispatching model download for ${modelName} via job queue`)
|
||
|
||
await DownloadModelJob.dispatch({
|
||
modelName,
|
||
})
|
||
|
||
return {
|
||
success: true,
|
||
message:
|
||
'Model download has been queued successfully. It will start shortly after Ollama and Open WebUI are ready (if not already).',
|
||
}
|
||
} catch (error) {
|
||
logger.error(
|
||
`[OllamaService] Failed to dispatch model download for ${modelName}: ${error instanceof Error ? error.message : error}`
|
||
)
|
||
return {
|
||
success: false,
|
||
message: 'Failed to queue model download. Please try again.',
|
||
}
|
||
}
|
||
}
|
||
|
||
public async chat(chatRequest: ChatInput): Promise<NomadChatResponse> {
|
||
await this._ensureDependencies()
|
||
if (!this.openai) {
|
||
throw new Error('AI client is not initialized.')
|
||
}
|
||
|
||
const params: any = {
|
||
model: chatRequest.model,
|
||
messages: chatRequest.messages as ChatCompletionMessageParam[],
|
||
stream: false,
|
||
}
|
||
if (chatRequest.think) {
|
||
params.think = chatRequest.think
|
||
}
|
||
if (chatRequest.numCtx) {
|
||
params.num_ctx = chatRequest.numCtx
|
||
}
|
||
|
||
const response = await this.openai.chat.completions.create(params)
|
||
const choice = response.choices[0]
|
||
|
||
return {
|
||
message: {
|
||
content: choice.message.content ?? '',
|
||
thinking: (choice.message as any).thinking ?? undefined,
|
||
},
|
||
done: true,
|
||
model: response.model,
|
||
}
|
||
}
|
||
|
||
public async chatStream(chatRequest: ChatInput): Promise<AsyncIterable<NomadChatStreamChunk>> {
|
||
await this._ensureDependencies()
|
||
if (!this.openai) {
|
||
throw new Error('AI client is not initialized.')
|
||
}
|
||
|
||
const params: any = {
|
||
model: chatRequest.model,
|
||
messages: chatRequest.messages as ChatCompletionMessageParam[],
|
||
stream: true,
|
||
}
|
||
if (chatRequest.think) {
|
||
params.think = chatRequest.think
|
||
}
|
||
if (chatRequest.numCtx) {
|
||
params.num_ctx = chatRequest.numCtx
|
||
}
|
||
|
||
const stream = (await this.openai.chat.completions.create(params)) as unknown as Stream<ChatCompletionChunk>
|
||
|
||
// Returns how many trailing chars of `text` could be the start of `tag`
|
||
function partialTagSuffix(tag: string, text: string): number {
|
||
for (let len = Math.min(tag.length - 1, text.length); len >= 1; len--) {
|
||
if (text.endsWith(tag.slice(0, len))) return len
|
||
}
|
||
return 0
|
||
}
|
||
|
||
async function* normalize(): AsyncGenerator<NomadChatStreamChunk> {
|
||
// Stateful parser for <think>...</think> tags that may be split across chunks.
|
||
// Ollama provides thinking natively via delta.thinking; OpenAI-compatible backends
|
||
// (LM Studio, llama.cpp, etc.) embed them inline in delta.content.
|
||
let tagBuffer = ''
|
||
let inThink = false
|
||
|
||
for await (const chunk of stream) {
|
||
const delta = chunk.choices[0]?.delta
|
||
const nativeThinking: string = (delta as any)?.thinking ?? ''
|
||
const rawContent: string = delta?.content ?? ''
|
||
|
||
// Parse <think> tags out of the content stream
|
||
tagBuffer += rawContent
|
||
let parsedContent = ''
|
||
let parsedThinking = ''
|
||
|
||
while (tagBuffer.length > 0) {
|
||
if (inThink) {
|
||
const closeIdx = tagBuffer.indexOf('</think>')
|
||
if (closeIdx !== -1) {
|
||
parsedThinking += tagBuffer.slice(0, closeIdx)
|
||
tagBuffer = tagBuffer.slice(closeIdx + 8)
|
||
inThink = false
|
||
} else {
|
||
const hold = partialTagSuffix('</think>', tagBuffer)
|
||
parsedThinking += tagBuffer.slice(0, tagBuffer.length - hold)
|
||
tagBuffer = tagBuffer.slice(tagBuffer.length - hold)
|
||
break
|
||
}
|
||
} else {
|
||
const openIdx = tagBuffer.indexOf('<think>')
|
||
if (openIdx !== -1) {
|
||
parsedContent += tagBuffer.slice(0, openIdx)
|
||
tagBuffer = tagBuffer.slice(openIdx + 7)
|
||
inThink = true
|
||
} else {
|
||
const hold = partialTagSuffix('<think>', tagBuffer)
|
||
parsedContent += tagBuffer.slice(0, tagBuffer.length - hold)
|
||
tagBuffer = tagBuffer.slice(tagBuffer.length - hold)
|
||
break
|
||
}
|
||
}
|
||
}
|
||
|
||
yield {
|
||
message: {
|
||
content: parsedContent,
|
||
thinking: nativeThinking + parsedThinking,
|
||
},
|
||
done: chunk.choices[0]?.finish_reason !== null && chunk.choices[0]?.finish_reason !== undefined,
|
||
}
|
||
}
|
||
}
|
||
|
||
return normalize()
|
||
}
|
||
|
||
public async checkModelHasThinking(modelName: string): Promise<boolean> {
|
||
await this._ensureDependencies()
|
||
if (!this.baseUrl) return false
|
||
|
||
try {
|
||
const response = await axios.post(
|
||
`${this.baseUrl}/api/show`,
|
||
{ model: modelName },
|
||
{ timeout: 5000 }
|
||
)
|
||
return Array.isArray(response.data?.capabilities) && response.data.capabilities.includes('thinking')
|
||
} catch {
|
||
// Non-Ollama backends don't expose /api/show — assume no thinking support
|
||
return false
|
||
}
|
||
}
|
||
|
||
public async deleteModel(modelName: string): Promise<{ success: boolean; message: string }> {
|
||
await this._ensureDependencies()
|
||
if (!this.baseUrl) {
|
||
return { success: false, message: 'AI service is not initialized.' }
|
||
}
|
||
|
||
try {
|
||
await axios.delete(`${this.baseUrl}/api/delete`, {
|
||
data: { model: modelName },
|
||
timeout: 10000,
|
||
})
|
||
return { success: true, message: `Model "${modelName}" deleted.` }
|
||
} catch (error) {
|
||
logger.error(
|
||
`[OllamaService] Failed to delete model "${modelName}": ${error instanceof Error ? error.message : error}`
|
||
)
|
||
return { success: false, message: 'Failed to delete model. This may not be an Ollama backend.' }
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Hard char cap per embed input, applied as a runtime safety net regardless of
|
||
* which backend path runs. The chunker in RagService caps at MAX_SAFE_TOKENS=1600
|
||
* (3200 chars at the conservative 2 chars/token estimate), but dense technical
|
||
* content has been observed to slip past on multi-batch ZIM ingestion (#881).
|
||
*
|
||
* 4000 chars ≈ 1000–2000 tokens depending on density, which keeps us comfortably
|
||
* under nomic-embed-text:v1.5's default 2048-token context even on the OpenAI-compat
|
||
* fallback path (which can't pass `truncate:true`/`num_ctx` to the model).
|
||
*/
|
||
public static readonly EMBED_MAX_INPUT_CHARS = 4000
|
||
|
||
/**
|
||
* Aggressive 2048-safe character cap, applied only on a context-length retry. nomic-embed-text:v1.5
|
||
* defaults to a 2048-token context, and on the OpenAI-compat fallback path (or an older Ollama that
|
||
* ignores num_ctx for embeddings) we cannot widen it at request time. 2000 chars stays under 2048
|
||
* tokens even for the densest content (~1 char/token code/markup), so an oversized chunk gets
|
||
* truncated-and-kept instead of silently dropped from Qdrant — and the embed job stops re-embedding
|
||
* the whole file 30x on the one bad chunk (#881).
|
||
*/
|
||
public static readonly EMBED_CONTEXT_SAFE_CHARS = 2000
|
||
|
||
/**
|
||
* True if the error is the model rejecting input that exceeds its context window
|
||
* ("input length exceeds the context length"). Matches both the native /api/embed axios error
|
||
* shape and the OpenAI-compat BadRequestError. Drives the truncate-and-retry here and the
|
||
* non-retryable classification in EmbedFileJob (#881).
|
||
*/
|
||
public static isContextLengthError(err: unknown): boolean {
|
||
const parts: string[] = []
|
||
if (err instanceof Error && err.message) parts.push(err.message)
|
||
const anyErr = err as any
|
||
const data = anyErr?.response?.data
|
||
if (data) parts.push(typeof data === 'string' ? data : JSON.stringify(data))
|
||
if (anyErr?.error) parts.push(typeof anyErr.error === 'string' ? anyErr.error : JSON.stringify(anyErr.error))
|
||
const haystack = parts.join(' ').toLowerCase()
|
||
return (
|
||
(haystack.includes('context length') && haystack.includes('exceed')) ||
|
||
haystack.includes('input length exceeds')
|
||
)
|
||
}
|
||
|
||
/**
|
||
* Generate embeddings for the given input strings.
|
||
* Tries the Ollama native /api/embed endpoint first, falls back to /v1/embeddings.
|
||
*
|
||
* If the first attempt fails because a chunk exceeds the model's context window, retries once
|
||
* with an aggressive 2048-safe truncation (EMBED_CONTEXT_SAFE_CHARS) so the chunk is embedded
|
||
* (start-of-chunk) rather than silently dropped from Qdrant (#881).
|
||
*/
|
||
public async embed(model: string, input: string[]): Promise<{ embeddings: number[][] }> {
|
||
await this._ensureDependencies()
|
||
if (!this.baseUrl || !this.openai) {
|
||
throw new Error('AI service is not initialized.')
|
||
}
|
||
|
||
const cap = (arr: string[], max: number) => arr.map((s) => (s.length > max ? s.slice(0, max) : s))
|
||
|
||
// Generous pre-cap (#881): fine for the native path (num_ctx=8192) but can still exceed a
|
||
// 2048-context fallback on dense content. The context-length retry below is the hard backstop.
|
||
const safeInput = cap(input, OllamaService.EMBED_MAX_INPUT_CHARS)
|
||
|
||
try {
|
||
return await this._embedWithFallback(model, safeInput)
|
||
} catch (err) {
|
||
if (!OllamaService.isContextLengthError(err)) throw err
|
||
// One or more chunks exceeded the model's context even after the pre-cap — typically an
|
||
// older Ollama that ignores num_ctx for embeddings, or the OpenAI-compat fallback path.
|
||
// Retry once, truncated hard enough to fit a 2048-token context at any density, so the
|
||
// chunk is embedded (truncated) instead of dropped and the job doesn't storm.
|
||
const hardCapped = cap(input, OllamaService.EMBED_CONTEXT_SAFE_CHARS)
|
||
const reduced = hardCapped.reduce((n, s, i) => (s.length < safeInput[i].length ? n + 1 : n), 0)
|
||
logger.warn(
|
||
'[OllamaService] embed: context-length overflow; retrying %d/%d inputs hard-capped at %d chars',
|
||
reduced,
|
||
input.length,
|
||
OllamaService.EMBED_CONTEXT_SAFE_CHARS
|
||
)
|
||
return await this._embedWithFallback(model, hardCapped)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Single embed attempt: native /api/embed first, then the OpenAI-compat /v1/embeddings fallback.
|
||
* Both paths request num_ctx/truncate (Ollama's OpenAI-compat shim forwards them). A context-length
|
||
* error from the native path is re-thrown rather than falling back, because the fallback has a
|
||
* smaller effective context and would only fail the same way — the caller (embed) retries it
|
||
* truncated instead.
|
||
*/
|
||
private async _embedWithFallback(model: string, input: string[]): Promise<{ embeddings: number[][] }> {
|
||
try {
|
||
// Pass num_ctx explicitly so we don't depend on the embedding model's modelfile defaults.
|
||
// Some installs ship nomic-embed-text:v1.5 with num_ctx=2048; 8192 matches its RoPE-extrapolated
|
||
// max. truncate:true is a server-side net for any chunk that still overshoots.
|
||
const response = await axios.post(
|
||
`${this.baseUrl}/api/embed`,
|
||
{
|
||
model,
|
||
input,
|
||
truncate: true,
|
||
options: { num_ctx: 8192 },
|
||
},
|
||
{ timeout: 60000 }
|
||
)
|
||
// Some backends (e.g. LM Studio) return HTTP 200 for unknown endpoints with an incompatible
|
||
// body — validate explicitly before accepting the result.
|
||
if (!Array.isArray(response.data?.embeddings)) {
|
||
throw new Error('Invalid /api/embed response — missing embeddings array')
|
||
}
|
||
return { embeddings: response.data.embeddings }
|
||
} catch (err) {
|
||
// Let context-length errors bubble so embed() can retry with a smaller cap; the fallback
|
||
// endpoint (smaller effective context, no num_ctx honored on older Ollama) can't help here.
|
||
if (OllamaService.isContextLengthError(err)) throw err
|
||
// Log the original error so we know *why* we fell back. Earlier bare catches here masked
|
||
// recurring failures for months (#369, #670, #881).
|
||
logger.warn(
|
||
'[OllamaService] /api/embed failed, falling back to /v1/embeddings: %s',
|
||
err instanceof Error ? err.message : String(err)
|
||
)
|
||
// Fall back to OpenAI-compatible /v1/embeddings. Explicitly request float format — some
|
||
// backends (e.g. LM Studio) don't reliably implement the base64 the OpenAI SDK defaults to.
|
||
// truncate/num_ctx are forwarded by Ollama's OpenAI-compat shim; the SDK types omit them,
|
||
// hence the cast. We only ever talk to a local Ollama here, not real OpenAI.
|
||
const results = await this.openai!.embeddings.create({
|
||
model,
|
||
input,
|
||
encoding_format: 'float',
|
||
truncate: true,
|
||
options: { num_ctx: 8192 },
|
||
} as any)
|
||
return { embeddings: results.data.map((e) => e.embedding as number[]) }
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Returns true if Ollama is currently running an embedding model with non-zero VRAM
|
||
* (i.e., GPU-offloaded). Returns false if the model is running CPU-only OR if it's
|
||
* not currently loaded OR if /api/ps is unreachable.
|
||
*
|
||
* Used by EmbedFileJob to pace continuation batches when the embedding model is
|
||
* CPU-bound — sustained 100% CPU on a multi-batch ZIM ingestion can starve other
|
||
* services (sshd, etc.) hard enough to require a power-cycle. AMD ROCm installs
|
||
* hit this today because Ollama's ROCm build doesn't accelerate nomic-bert; on
|
||
* NVIDIA, nomic-embed-text runs at 100% GPU and pacing is unnecessary.
|
||
*
|
||
* Only the Ollama-native endpoint is supported — backends that expose
|
||
* `/v1/embeddings` (LM Studio, llama.cpp) don't surface placement info.
|
||
*/
|
||
public async isEmbeddingGpuAccelerated(): Promise<boolean> {
|
||
await this._ensureDependencies()
|
||
if (!this.baseUrl) return false
|
||
|
||
try {
|
||
const response = await axios.get(`${this.baseUrl}/api/ps`, { timeout: 5000 })
|
||
const models: Array<{ name?: string; size_vram?: number }> = response.data?.models ?? []
|
||
// Match any loaded model whose name signals it's an embedding model.
|
||
// nomic-embed-text, mxbai-embed-large, snowflake-arctic-embed, etc. all follow this convention.
|
||
return models.some(
|
||
(m) => m.name?.toLowerCase().includes('embed') && (m.size_vram ?? 0) > 0
|
||
)
|
||
} catch (err: any) {
|
||
// /api/ps unreachable (Ollama down, non-native backend, etc.) — fail closed: assume CPU,
|
||
// which means we'll pace. Better to over-pace than risk box-killing CPU saturation.
|
||
logger.warn(
|
||
`[OllamaService] Could not check embedding placement via /api/ps: ${err?.message ?? err}`
|
||
)
|
||
return false
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Enforces the "at most one chat model resident in VRAM" invariant by firing
|
||
* `keep_alive: 0` against every currently-loaded model except (a) the
|
||
* embedding model (always exempt) and (b) `targetModel` (the one we want
|
||
* loaded next — leaving it alone preserves a hot model when the target is
|
||
* already loaded).
|
||
*
|
||
* Best-effort: queries `/api/ps` and POSTs unload hints in parallel. Network
|
||
* or Ollama errors are swallowed and logged — neither chat nor page-load
|
||
* should fail just because the unload housekeeping didn't go through.
|
||
*
|
||
* Returns the list of model names that were sent the unload hint, so the
|
||
* caller (and tests) can confirm what actually happened.
|
||
*
|
||
* Pass `targetModel: null` to unload every chat model (used for the future
|
||
* "free up VRAM" path; not exposed yet but the helper supports it).
|
||
*
|
||
* Note that `keep_alive: 0` is a post-completion hint, not a force-kill —
|
||
* Ollama defers eviction until the runner is idle, so in-flight inference
|
||
* on the same model is never interrupted. See the design doc for the race
|
||
* analysis behind this.
|
||
*/
|
||
public async unloadAllChatModelsExcept(targetModel: string | null): Promise<string[]> {
|
||
await this._ensureDependencies()
|
||
if (!this.baseUrl) return []
|
||
|
||
let loadedModels: string[] = []
|
||
try {
|
||
const response = await axios.get(`${this.baseUrl}/api/ps`, { timeout: 5000 })
|
||
loadedModels = (response.data?.models ?? [])
|
||
.map((m: { name?: string }) => m.name)
|
||
.filter((name: unknown): name is string => typeof name === 'string')
|
||
} catch (err: any) {
|
||
logger.warn(
|
||
`[OllamaService] unloadAllChatModelsExcept: /api/ps unreachable, skipping unload sweep: ${err?.message ?? err}`
|
||
)
|
||
return []
|
||
}
|
||
|
||
const toUnload = loadedModels.filter(
|
||
(name) => name !== EMBEDDING_MODEL_NAME && name !== targetModel
|
||
)
|
||
|
||
await Promise.all(
|
||
toUnload.map(async (modelName) => {
|
||
try {
|
||
await axios.post(
|
||
`${this.baseUrl}/api/generate`,
|
||
{ model: modelName, prompt: '', keep_alive: 0 },
|
||
{ timeout: 10000 }
|
||
)
|
||
} catch (err: any) {
|
||
logger.warn(
|
||
`[OllamaService] Failed to send unload hint for ${modelName}: ${err?.message ?? err}`
|
||
)
|
||
}
|
||
})
|
||
)
|
||
|
||
if (toUnload.length > 0) {
|
||
logger.info(
|
||
`[OllamaService] Sent unload hint for ${toUnload.length} chat model(s): ${toUnload.join(', ')}`
|
||
)
|
||
}
|
||
return toUnload
|
||
}
|
||
|
||
public async getModels(includeEmbeddings = false): Promise<NomadInstalledModel[]> {
|
||
await this._ensureDependencies()
|
||
if (!this.baseUrl) {
|
||
throw new Error('AI service is not initialized.')
|
||
}
|
||
|
||
try {
|
||
// Prefer the Ollama native endpoint which includes size and metadata
|
||
const response = await axios.get(`${this.baseUrl}/api/tags`, { timeout: 5000 })
|
||
// LM Studio returns HTTP 200 for unknown endpoints with an incompatible body — validate explicitly
|
||
if (!Array.isArray(response.data?.models)) {
|
||
throw new Error('Not an Ollama-compatible /api/tags response')
|
||
}
|
||
this.isOllamaNative = true
|
||
const models: NomadInstalledModel[] = response.data.models
|
||
if (includeEmbeddings) return models
|
||
return models.filter((m) => !m.name.includes('embed'))
|
||
} catch {
|
||
// Fall back to the OpenAI-compatible /v1/models endpoint (LM Studio, llama.cpp, etc.)
|
||
this.isOllamaNative = false
|
||
logger.info('[OllamaService] /api/tags unavailable, falling back to /v1/models')
|
||
try {
|
||
const modelList = await this.openai!.models.list()
|
||
const models: NomadInstalledModel[] = modelList.data.map((m) => ({ name: m.id, size: 0 }))
|
||
if (includeEmbeddings) return models
|
||
return models.filter((m) => !m.name.includes('embed'))
|
||
} catch (err) {
|
||
logger.error(
|
||
`[OllamaService] Failed to list models: ${err instanceof Error ? err.message : err}`
|
||
)
|
||
return []
|
||
}
|
||
}
|
||
}
|
||
|
||
async getAvailableModels(
|
||
{
|
||
sort,
|
||
recommendedOnly,
|
||
query,
|
||
limit,
|
||
force,
|
||
}: {
|
||
sort?: 'pulls' | 'name'
|
||
recommendedOnly?: boolean
|
||
query: string | null
|
||
limit?: number
|
||
force?: boolean
|
||
} = {
|
||
sort: 'pulls',
|
||
recommendedOnly: false,
|
||
query: null,
|
||
limit: 15,
|
||
}
|
||
): Promise<{ models: NomadOllamaModel[]; hasMore: boolean } | null> {
|
||
try {
|
||
const models = await this.retrieveAndRefreshModels(sort, force)
|
||
if (!models || models.length === 0) {
|
||
logger.warn(
|
||
'[OllamaService] Returning fallback recommended models due to failure in fetching available models'
|
||
)
|
||
return {
|
||
models: FALLBACK_RECOMMENDED_OLLAMA_MODELS,
|
||
hasMore: false,
|
||
}
|
||
}
|
||
|
||
if (!recommendedOnly) {
|
||
const filteredModels = query ? this.fuseSearchModels(models, query) : models
|
||
return {
|
||
models: filteredModels.slice(0, limit || 15),
|
||
hasMore: filteredModels.length > (limit || 15),
|
||
}
|
||
}
|
||
|
||
const sortedByPulls = sort === 'pulls' ? models : this.sortModels(models, 'pulls')
|
||
const firstThree = sortedByPulls.slice(0, 3)
|
||
|
||
const recommendedModels = firstThree.map((model) => {
|
||
return {
|
||
...model,
|
||
tags: model.tags && model.tags.length > 0 ? [model.tags[0]] : [],
|
||
}
|
||
})
|
||
|
||
if (query) {
|
||
const filteredRecommendedModels = this.fuseSearchModels(recommendedModels, query)
|
||
return {
|
||
models: filteredRecommendedModels,
|
||
hasMore: filteredRecommendedModels.length > (limit || 15),
|
||
}
|
||
}
|
||
|
||
return {
|
||
models: recommendedModels,
|
||
hasMore: recommendedModels.length > (limit || 15),
|
||
}
|
||
} catch (error) {
|
||
logger.error(
|
||
`[OllamaService] Failed to get available models: ${error instanceof Error ? error.message : error}`
|
||
)
|
||
return null
|
||
}
|
||
}
|
||
|
||
private async retrieveAndRefreshModels(
|
||
sort?: 'pulls' | 'name',
|
||
force?: boolean
|
||
): Promise<NomadOllamaModel[] | null> {
|
||
try {
|
||
if (!force) {
|
||
const cachedModels = await this.readModelsFromCache()
|
||
// An empty cached array (e.g. written from a transient empty upstream
|
||
// response) must not be treated as valid data — fall through to a
|
||
// fresh fetch and, failing that, the fallback list.
|
||
if (cachedModels && cachedModels.length > 0) {
|
||
logger.info('[OllamaService] Using cached available models data')
|
||
return this.sortModels(cachedModels, sort)
|
||
}
|
||
} else {
|
||
logger.info('[OllamaService] Force refresh requested, bypassing cache')
|
||
}
|
||
|
||
logger.info('[OllamaService] Fetching fresh available models from API')
|
||
|
||
const baseUrl = env.get('NOMAD_API_URL') || NOMAD_API_DEFAULT_BASE_URL
|
||
const fullUrl = new URL(NOMAD_MODELS_API_PATH, baseUrl).toString()
|
||
|
||
const response = await axios.get(fullUrl, { timeout: 10000 })
|
||
if (!response.data || !Array.isArray(response.data.models)) {
|
||
logger.warn(
|
||
`[OllamaService] Invalid response format when fetching available models: ${JSON.stringify(response.data)}`
|
||
)
|
||
return null
|
||
}
|
||
|
||
const rawModels = response.data.models as NomadOllamaModel[]
|
||
|
||
const noCloud = rawModels
|
||
.map((model) => ({
|
||
...model,
|
||
tags: model.tags.filter((tag) => !tag.cloud),
|
||
}))
|
||
.filter((model) => model.tags.length > 0)
|
||
|
||
// A successful-but-empty upstream response (0 models, or all filtered out
|
||
// as cloud-only) is a soft failure: return null so the caller serves the
|
||
// fallback list, and don't poison the 24h cache with an empty array.
|
||
if (noCloud.length === 0) {
|
||
logger.warn(
|
||
'[OllamaService] Nomad API returned no usable (non-cloud) models; using fallback'
|
||
)
|
||
return null
|
||
}
|
||
|
||
await this.writeModelsToCache(noCloud)
|
||
return this.sortModels(noCloud, sort)
|
||
} catch (error) {
|
||
logger.error(
|
||
`[OllamaService] Failed to retrieve models from Nomad API: ${error instanceof Error ? error.message : error}`
|
||
)
|
||
return null
|
||
}
|
||
}
|
||
|
||
private async readModelsFromCache(): Promise<NomadOllamaModel[] | null> {
|
||
try {
|
||
const stats = await fs.stat(MODELS_CACHE_FILE)
|
||
const cacheAge = Date.now() - stats.mtimeMs
|
||
|
||
if (cacheAge > CACHE_MAX_AGE_MS) {
|
||
logger.info('[OllamaService] Cache is stale, will fetch fresh data')
|
||
return null
|
||
}
|
||
|
||
const cacheData = await fs.readFile(MODELS_CACHE_FILE, 'utf-8')
|
||
const models = JSON.parse(cacheData) as NomadOllamaModel[]
|
||
|
||
if (!Array.isArray(models)) {
|
||
logger.warn('[OllamaService] Invalid cache format, will fetch fresh data')
|
||
return null
|
||
}
|
||
|
||
return models
|
||
} catch (error) {
|
||
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') {
|
||
logger.warn(
|
||
`[OllamaService] Error reading cache: ${error instanceof Error ? error.message : error}`
|
||
)
|
||
}
|
||
return null
|
||
}
|
||
}
|
||
|
||
private async writeModelsToCache(models: NomadOllamaModel[]): Promise<void> {
|
||
try {
|
||
await fs.mkdir(path.dirname(MODELS_CACHE_FILE), { recursive: true })
|
||
await fs.writeFile(MODELS_CACHE_FILE, JSON.stringify(models, null, 2), 'utf-8')
|
||
logger.info('[OllamaService] Successfully cached available models')
|
||
} catch (error) {
|
||
logger.warn(
|
||
`[OllamaService] Failed to write models cache: ${error instanceof Error ? error.message : error}`
|
||
)
|
||
}
|
||
}
|
||
|
||
private sortModels(models: NomadOllamaModel[], sort?: 'pulls' | 'name'): NomadOllamaModel[] {
|
||
if (sort === 'pulls') {
|
||
models.sort((a, b) => {
|
||
const parsePulls = (pulls: string) => {
|
||
const multiplier = pulls.endsWith('K')
|
||
? 1_000
|
||
: pulls.endsWith('M')
|
||
? 1_000_000
|
||
: pulls.endsWith('B')
|
||
? 1_000_000_000
|
||
: 1
|
||
return parseFloat(pulls) * multiplier
|
||
}
|
||
return parsePulls(b.estimated_pulls) - parsePulls(a.estimated_pulls)
|
||
})
|
||
} else if (sort === 'name') {
|
||
models.sort((a, b) => a.name.localeCompare(b.name))
|
||
}
|
||
|
||
models.forEach((model) => {
|
||
if (model.tags && Array.isArray(model.tags)) {
|
||
model.tags.sort((a, b) => {
|
||
const parseSize = (size: string) => {
|
||
const multiplier = size.endsWith('KB')
|
||
? 1 / 1_000
|
||
: size.endsWith('MB')
|
||
? 1 / 1_000_000
|
||
: size.endsWith('GB')
|
||
? 1
|
||
: size.endsWith('TB')
|
||
? 1_000
|
||
: 0
|
||
return parseFloat(size) * multiplier
|
||
}
|
||
return parseSize(a.size) - parseSize(b.size)
|
||
})
|
||
}
|
||
})
|
||
|
||
return models
|
||
}
|
||
|
||
private broadcastDownloadError(model: string, error: string) {
|
||
transmit.broadcast(BROADCAST_CHANNELS.OLLAMA_MODEL_DOWNLOAD, {
|
||
model,
|
||
percent: -1,
|
||
error,
|
||
timestamp: new Date().toISOString(),
|
||
})
|
||
}
|
||
|
||
private broadcastDownloadProgress(
|
||
model: string,
|
||
percent: number,
|
||
jobId?: string,
|
||
bytes?: { downloadedBytes: number; totalBytes: number }
|
||
) {
|
||
// Conditional spread on jobId/bytes — Transmit's Broadcastable type rejects fields whose
|
||
// value is `undefined`, so we omit each key entirely when its value isn't available.
|
||
transmit.broadcast(BROADCAST_CHANNELS.OLLAMA_MODEL_DOWNLOAD, {
|
||
model,
|
||
percent,
|
||
...(jobId ? { jobId } : {}),
|
||
...(bytes ? { downloadedBytes: bytes.downloadedBytes, totalBytes: bytes.totalBytes } : {}),
|
||
timestamp: new Date().toISOString(),
|
||
})
|
||
logger.info(`[OllamaService] Download progress for model "${model}": ${percent}%`)
|
||
}
|
||
|
||
private fuseSearchModels(models: NomadOllamaModel[], query: string): NomadOllamaModel[] {
|
||
const options: IFuseOptions<NomadOllamaModel> = {
|
||
ignoreDiacritics: true,
|
||
keys: ['name', 'description', 'tags.name'],
|
||
threshold: 0.3,
|
||
}
|
||
|
||
const fuse = new Fuse(models, options)
|
||
|
||
return fuse.search(query).map((result) => result.item)
|
||
}
|
||
}
|