Scaffold shre-embed: embedding + semantic search with three-valued health
Express (ESM, no build step) service wrapping Ollama embeddings and Qdrant: - POST /v1/embed, /v1/index, /v1/search - GET /health reports each dependency as ok | absent | unobservable with the exact measurement_surface URL and observed_at; overall ok only when all deps are ok, 503 degraded otherwise - upstream failures on /v1/* return 502 marked unobservable — never an empty result set or indexed:0 - defaults PORT=5499, QDRANT_URL=127.0.0.1:6333, OLLAMA_URL=127.0.0.1:11436 (11436 native Ollama; 11434 is the broken Colima forward on this Mac) - node:test unit tests with injected fetch; run with zero live dependencies - README documents endpoints and a reference launchd plist (not installed) Co-Authored-By: Claude Fable 5 <[email protected]>
This commit is contained in:
Claude
+181
@@ -0,0 +1,181 @@
|
||||
import express from "express";
|
||||
import { httpProbe, probeResult, OK, ABSENT, UNOBSERVABLE } from "./probe.js";
|
||||
import {
|
||||
UpstreamError,
|
||||
embedTexts,
|
||||
ensureCollection,
|
||||
upsertPoints,
|
||||
searchPoints,
|
||||
} from "./upstreams.js";
|
||||
|
||||
/**
|
||||
* Build the express app. `fetchImpl` is injectable so unit tests run without
|
||||
* Qdrant/Ollama up.
|
||||
*/
|
||||
export function createApp(config, { fetchImpl = fetch } = {}) {
|
||||
const app = express();
|
||||
app.use(express.json({ limit: "5mb" }));
|
||||
|
||||
const upstreamOpts = {
|
||||
ollamaUrl: config.ollamaUrl,
|
||||
qdrantUrl: config.qdrantUrl,
|
||||
embedModel: config.embedModel,
|
||||
fetchImpl,
|
||||
timeoutMs: config.upstreamTimeoutMs,
|
||||
};
|
||||
|
||||
// ---- health: three-valued, per-dependency, with exact surfaces ----------
|
||||
app.get("/health", async (_req, res) => {
|
||||
const qdrantSurface = `${config.qdrantUrl}/readyz`;
|
||||
const ollamaSurface = `${config.ollamaUrl}/api/tags`;
|
||||
|
||||
let embedModelResult = null;
|
||||
const [qdrant, ollama] = await Promise.all([
|
||||
httpProbe(qdrantSurface, { fetchImpl, timeoutMs: config.probeTimeoutMs }),
|
||||
httpProbe(ollamaSurface, {
|
||||
fetchImpl,
|
||||
timeoutMs: config.probeTimeoutMs,
|
||||
classify: (_res, bodyText) => {
|
||||
// While we can see the tag list, also classify the embed model:
|
||||
// present -> ok, listed-but-missing -> absent (a true gap, distinct
|
||||
// from "could not look").
|
||||
try {
|
||||
const models = JSON.parse(bodyText)?.models ?? [];
|
||||
const found = models.some(
|
||||
(m) => m?.name === config.embedModel || m?.name?.split(":")[0] === config.embedModel
|
||||
);
|
||||
embedModelResult = probeResult(found ? OK : ABSENT, ollamaSurface, {
|
||||
model: config.embedModel,
|
||||
});
|
||||
} catch {
|
||||
embedModelResult = probeResult(UNOBSERVABLE, ollamaSurface, {
|
||||
model: config.embedModel,
|
||||
reason: "tag list unparseable",
|
||||
});
|
||||
}
|
||||
return null; // do not override the reachability verdict
|
||||
},
|
||||
}),
|
||||
]);
|
||||
if (!embedModelResult) {
|
||||
// Ollama itself was unobservable, so the model's presence is too —
|
||||
// never report it as absent (or fine) when we could not look.
|
||||
embedModelResult = probeResult(UNOBSERVABLE, ollamaSurface, {
|
||||
model: config.embedModel,
|
||||
reason: "ollama unobservable, model presence not measurable",
|
||||
});
|
||||
}
|
||||
|
||||
const deps = { qdrant, ollama, embed_model: embedModelResult };
|
||||
const allOk = Object.values(deps).every((d) => d.status === OK);
|
||||
// Overall is NEVER plain "ok" while any dependency is unobservable/absent.
|
||||
const overall = allOk ? "ok" : "degraded";
|
||||
res.status(allOk ? 200 : 503).json({
|
||||
status: overall,
|
||||
service: "shre-embed",
|
||||
observed_at: new Date().toISOString(),
|
||||
dependencies: deps,
|
||||
});
|
||||
});
|
||||
|
||||
// ---- helpers ------------------------------------------------------------
|
||||
function upstreamErrorBody(err) {
|
||||
return {
|
||||
error: err.message,
|
||||
// The caller must be able to tell "failed" from "answer is empty/zero":
|
||||
status: UNOBSERVABLE,
|
||||
measurement_surface: err.surface,
|
||||
observed_at: new Date().toISOString(),
|
||||
...(err.httpStatus ? { upstream_http_status: err.httpStatus } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
// ---- POST /v1/embed -----------------------------------------------------
|
||||
app.post("/v1/embed", async (req, res) => {
|
||||
const { texts } = req.body ?? {};
|
||||
if (!Array.isArray(texts) || texts.length === 0 || !texts.every((t) => typeof t === "string")) {
|
||||
return res.status(400).json({ error: "body must be {texts: string[]} with at least one text" });
|
||||
}
|
||||
try {
|
||||
const vectors = await embedTexts(texts, upstreamOpts);
|
||||
res.json({
|
||||
vectors,
|
||||
model: config.embedModel,
|
||||
measurement_surface: `${config.ollamaUrl}/api/embeddings`,
|
||||
observed_at: new Date().toISOString(),
|
||||
});
|
||||
} catch (err) {
|
||||
if (err instanceof UpstreamError) return res.status(502).json(upstreamErrorBody(err));
|
||||
throw err;
|
||||
}
|
||||
});
|
||||
|
||||
// ---- POST /v1/index -----------------------------------------------------
|
||||
app.post("/v1/index", async (req, res) => {
|
||||
const { collection, items } = req.body ?? {};
|
||||
if (typeof collection !== "string" || !collection) {
|
||||
return res.status(400).json({ error: "body must include a non-empty 'collection' string" });
|
||||
}
|
||||
if (
|
||||
!Array.isArray(items) ||
|
||||
items.length === 0 ||
|
||||
!items.every((it) => it && it.id !== undefined && typeof it.text === "string")
|
||||
) {
|
||||
return res.status(400).json({ error: "body must include items: [{id, text, payload?}] with at least one item" });
|
||||
}
|
||||
try {
|
||||
const vectors = await embedTexts(items.map((it) => it.text), upstreamOpts);
|
||||
await ensureCollection(collection, vectors[0].length, upstreamOpts);
|
||||
const points = items.map((it, i) => ({
|
||||
id: it.id,
|
||||
vector: vectors[i],
|
||||
payload: { ...(it.payload ?? {}), text: it.text },
|
||||
}));
|
||||
const result = await upsertPoints(collection, points, upstreamOpts);
|
||||
res.json({
|
||||
indexed: items.length,
|
||||
collection,
|
||||
qdrant: result?.result ?? result,
|
||||
measurement_surface: `${config.qdrantUrl}/collections/${collection}/points`,
|
||||
observed_at: new Date().toISOString(),
|
||||
});
|
||||
} catch (err) {
|
||||
if (err instanceof UpstreamError) return res.status(502).json(upstreamErrorBody(err));
|
||||
throw err;
|
||||
}
|
||||
});
|
||||
|
||||
// ---- POST /v1/search ----------------------------------------------------
|
||||
app.post("/v1/search", async (req, res) => {
|
||||
const { collection, query, limit } = req.body ?? {};
|
||||
if (typeof collection !== "string" || !collection || typeof query !== "string" || !query) {
|
||||
return res.status(400).json({ error: "body must include non-empty 'collection' and 'query' strings" });
|
||||
}
|
||||
const lim = Number.isInteger(limit) && limit > 0 ? Math.min(limit, 100) : 10;
|
||||
try {
|
||||
const [vector] = await embedTexts([query], upstreamOpts);
|
||||
const result = await searchPoints(collection, vector, lim, upstreamOpts);
|
||||
res.json({
|
||||
results: result.map((r) => ({ id: r.id, score: r.score, payload: r.payload ?? null })),
|
||||
collection,
|
||||
limit: lim,
|
||||
measurement_surface: `${config.qdrantUrl}/collections/${collection}/points/search`,
|
||||
observed_at: new Date().toISOString(),
|
||||
});
|
||||
} catch (err) {
|
||||
if (err instanceof UpstreamError) return res.status(502).json(upstreamErrorBody(err));
|
||||
throw err;
|
||||
}
|
||||
});
|
||||
|
||||
// express error fallthrough (bad JSON etc.)
|
||||
// eslint-disable-next-line no-unused-vars
|
||||
app.use((err, _req, res, _next) => {
|
||||
if (err?.type === "entity.parse.failed") {
|
||||
return res.status(400).json({ error: "invalid JSON body" });
|
||||
}
|
||||
res.status(500).json({ error: "internal error" });
|
||||
});
|
||||
|
||||
return app;
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
// Config via environment with estate defaults.
|
||||
//
|
||||
// NOTE: default OLLAMA_URL is 127.0.0.1:11436 (native Ollama), NOT 11434.
|
||||
// On this Mac, 127.0.0.1:11434 is a known-broken Colima SSH-mux forward that
|
||||
// accepts connections (tags respond) but hangs on generation/embedding.
|
||||
export function loadConfig(env = process.env) {
|
||||
return {
|
||||
port: Number(env.PORT) > 0 ? Number(env.PORT) : 5499,
|
||||
qdrantUrl: (env.QDRANT_URL || "http://127.0.0.1:6333").replace(/\/+$/, ""),
|
||||
ollamaUrl: (env.OLLAMA_URL || "http://127.0.0.1:11436").replace(/\/+$/, ""),
|
||||
embedModel: env.EMBED_MODEL || "nomic-embed-text",
|
||||
// Per-request timeout for dependency probes and upstream calls (ms).
|
||||
probeTimeoutMs: Number(env.PROBE_TIMEOUT_MS) > 0 ? Number(env.PROBE_TIMEOUT_MS) : 2500,
|
||||
upstreamTimeoutMs: Number(env.UPSTREAM_TIMEOUT_MS) > 0 ? Number(env.UPSTREAM_TIMEOUT_MS) : 30000,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
// Three-valued observability primitives.
|
||||
//
|
||||
// Every probe answer is one of THREE states, never two:
|
||||
// "ok" — the probe saw the thing and it is present/working
|
||||
// "absent" — the probe SAW the surface and the thing is genuinely missing
|
||||
// "unobservable" — the probe COULD NOT SEE (dependency down, timeout, auth
|
||||
// failure, endpoint missing). Never collapsed into ok/absent.
|
||||
//
|
||||
// Every result carries measurement_surface (exactly what was measured) and
|
||||
// observed_at (ISO timestamp).
|
||||
|
||||
export const OK = "ok";
|
||||
export const ABSENT = "absent";
|
||||
export const UNOBSERVABLE = "unobservable";
|
||||
|
||||
export function probeResult(status, measurementSurface, extra = {}) {
|
||||
return {
|
||||
status,
|
||||
measurement_surface: measurementSurface,
|
||||
observed_at: new Date().toISOString(),
|
||||
...extra,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Perform an HTTP GET probe of `url`. Returns a three-valued probeResult:
|
||||
* - ok when the response status passes `okWhen` (default: 2xx)
|
||||
* - unobservable on network error, timeout, or non-passing HTTP status
|
||||
* (a 401/404/500 from the surface means we could not observe the truth,
|
||||
* not that the dependency is healthy or absent).
|
||||
* `classify(response, bodyText)` may optionally downgrade an HTTP-ok response
|
||||
* to ABSENT (e.g. endpoint reachable but required model not installed).
|
||||
*/
|
||||
export async function httpProbe(url, { fetchImpl = fetch, timeoutMs = 2500, classify } = {}) {
|
||||
const started = Date.now();
|
||||
let res;
|
||||
try {
|
||||
res = await fetchImpl(url, { signal: AbortSignal.timeout(timeoutMs) });
|
||||
} catch (err) {
|
||||
return probeResult(UNOBSERVABLE, url, {
|
||||
reason: `fetch failed: ${err?.cause?.code || err?.name || "error"}: ${err?.message}`,
|
||||
latency_ms: Date.now() - started,
|
||||
});
|
||||
}
|
||||
const latency = Date.now() - started;
|
||||
if (!res.ok) {
|
||||
return probeResult(UNOBSERVABLE, url, {
|
||||
reason: `HTTP ${res.status} from probe surface`,
|
||||
http_status: res.status,
|
||||
latency_ms: latency,
|
||||
});
|
||||
}
|
||||
if (classify) {
|
||||
let bodyText = "";
|
||||
try {
|
||||
bodyText = await res.text();
|
||||
} catch {
|
||||
return probeResult(UNOBSERVABLE, url, {
|
||||
reason: "response body unreadable",
|
||||
latency_ms: latency,
|
||||
});
|
||||
}
|
||||
const classified = classify(res, bodyText);
|
||||
if (classified) return probeResult(classified.status, url, { latency_ms: latency, ...classified.extra });
|
||||
}
|
||||
return probeResult(OK, url, { http_status: res.status, latency_ms: latency });
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
import { loadConfig } from "./config.js";
|
||||
import { createApp } from "./app.js";
|
||||
|
||||
const config = loadConfig();
|
||||
const app = createApp(config);
|
||||
|
||||
app.listen(config.port, "127.0.0.1", () => {
|
||||
// eslint-disable-next-line no-console
|
||||
console.log(
|
||||
`shre-embed listening on 127.0.0.1:${config.port} ` +
|
||||
`(qdrant=${config.qdrantUrl} ollama=${config.ollamaUrl} model=${config.embedModel})`
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,126 @@
|
||||
// Thin clients for Ollama (embeddings) and Qdrant (vector store).
|
||||
// All calls take an injected fetchImpl so tests run without either service up.
|
||||
//
|
||||
// Failures are surfaced as UpstreamError carrying the exact measurement
|
||||
// surface (URL) that failed — callers propagate this so an API consumer can
|
||||
// always tell "the answer is X" apart from "the answer was unobservable".
|
||||
|
||||
export class UpstreamError extends Error {
|
||||
constructor(message, { surface, cause, httpStatus } = {}) {
|
||||
super(message);
|
||||
this.name = "UpstreamError";
|
||||
this.surface = surface;
|
||||
this.httpStatus = httpStatus;
|
||||
if (cause) this.cause = cause;
|
||||
}
|
||||
}
|
||||
|
||||
async function postJson(url, body, { fetchImpl, timeoutMs }) {
|
||||
let res;
|
||||
try {
|
||||
res = await fetchImpl(url, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(body),
|
||||
signal: AbortSignal.timeout(timeoutMs),
|
||||
});
|
||||
} catch (err) {
|
||||
throw new UpstreamError(`request failed: ${err?.message}`, { surface: url, cause: err });
|
||||
}
|
||||
let json;
|
||||
try {
|
||||
json = await res.json();
|
||||
} catch {
|
||||
json = undefined;
|
||||
}
|
||||
if (!res.ok) {
|
||||
throw new UpstreamError(`HTTP ${res.status}: ${JSON.stringify(json)?.slice(0, 300)}`, {
|
||||
surface: url,
|
||||
httpStatus: res.status,
|
||||
});
|
||||
}
|
||||
return json;
|
||||
}
|
||||
|
||||
/** Embed each text via Ollama's embeddings API. Returns number[][]. */
|
||||
export async function embedTexts(texts, { ollamaUrl, embedModel, fetchImpl = fetch, timeoutMs = 30000 }) {
|
||||
const url = `${ollamaUrl}/api/embeddings`;
|
||||
const vectors = [];
|
||||
for (const text of texts) {
|
||||
const json = await postJson(url, { model: embedModel, prompt: text }, { fetchImpl, timeoutMs });
|
||||
if (!Array.isArray(json?.embedding)) {
|
||||
throw new UpstreamError("Ollama response missing 'embedding' array", { surface: url });
|
||||
}
|
||||
vectors.push(json.embedding);
|
||||
}
|
||||
return vectors;
|
||||
}
|
||||
|
||||
/** Ensure a Qdrant collection exists with the given vector size (Cosine). */
|
||||
export async function ensureCollection(collection, vectorSize, { qdrantUrl, fetchImpl = fetch, timeoutMs = 30000 }) {
|
||||
const checkUrl = `${qdrantUrl}/collections/${encodeURIComponent(collection)}`;
|
||||
let res;
|
||||
try {
|
||||
res = await fetchImpl(checkUrl, { signal: AbortSignal.timeout(timeoutMs) });
|
||||
} catch (err) {
|
||||
throw new UpstreamError(`request failed: ${err?.message}`, { surface: checkUrl, cause: err });
|
||||
}
|
||||
if (res.ok) return { created: false };
|
||||
if (res.status !== 404) {
|
||||
throw new UpstreamError(`HTTP ${res.status} checking collection`, { surface: checkUrl, httpStatus: res.status });
|
||||
}
|
||||
let createRes;
|
||||
try {
|
||||
createRes = await fetchImpl(checkUrl, {
|
||||
method: "PUT",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ vectors: { size: vectorSize, distance: "Cosine" } }),
|
||||
signal: AbortSignal.timeout(timeoutMs),
|
||||
});
|
||||
} catch (err) {
|
||||
throw new UpstreamError(`request failed: ${err?.message}`, { surface: checkUrl, cause: err });
|
||||
}
|
||||
if (!createRes.ok) {
|
||||
throw new UpstreamError(`HTTP ${createRes.status} creating collection`, { surface: checkUrl, httpStatus: createRes.status });
|
||||
}
|
||||
return { created: true };
|
||||
}
|
||||
|
||||
/** Upsert points into a Qdrant collection. */
|
||||
export async function upsertPoints(collection, points, { qdrantUrl, fetchImpl = fetch, timeoutMs = 30000 }) {
|
||||
const url = `${qdrantUrl}/collections/${encodeURIComponent(collection)}/points?wait=true`;
|
||||
let res;
|
||||
try {
|
||||
res = await fetchImpl(url, {
|
||||
method: "PUT",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ points }),
|
||||
signal: AbortSignal.timeout(timeoutMs),
|
||||
});
|
||||
} catch (err) {
|
||||
throw new UpstreamError(`request failed: ${err?.message}`, { surface: url, cause: err });
|
||||
}
|
||||
let json;
|
||||
try {
|
||||
json = await res.json();
|
||||
} catch {
|
||||
json = undefined;
|
||||
}
|
||||
if (!res.ok) {
|
||||
throw new UpstreamError(`HTTP ${res.status}: ${JSON.stringify(json)?.slice(0, 300)}`, {
|
||||
surface: url,
|
||||
httpStatus: res.status,
|
||||
});
|
||||
}
|
||||
return json;
|
||||
}
|
||||
|
||||
/** Vector search in a Qdrant collection. Returns scored points with payloads. */
|
||||
export async function searchPoints(collection, vector, limit, { qdrantUrl, fetchImpl = fetch, timeoutMs = 30000 }) {
|
||||
const url = `${qdrantUrl}/collections/${encodeURIComponent(collection)}/points/search`;
|
||||
const json = await postJson(url, { vector, limit, with_payload: true }, { fetchImpl, timeoutMs });
|
||||
if (!Array.isArray(json?.result)) {
|
||||
throw new UpstreamError("Qdrant response missing 'result' array", { surface: url });
|
||||
}
|
||||
return json.result;
|
||||
}
|
||||
Reference in New Issue
Block a user