164 lines
3.5 KiB
JavaScript
164 lines
3.5 KiB
JavaScript
// Import dotenv file if not in production
|
|
if (process.env.NODE_ENV !== "production") {
|
|
require("dotenv").config();
|
|
}
|
|
|
|
const {env, pipeline} = require("@xenova/transformers");
|
|
|
|
const models = require("./models");
|
|
const db = models.db;
|
|
const redis = models.redis;
|
|
const {sequelize, Sequelize} = db;
|
|
const Intent = db.intent;
|
|
|
|
async function getIntents() {
|
|
try {
|
|
const intents = await Intent.findAll();
|
|
|
|
console.log(intents);
|
|
|
|
return intents;
|
|
} catch (err) {
|
|
console.error("getIntents error:", err);
|
|
}
|
|
}
|
|
|
|
const config = require("./config");
|
|
|
|
let extractor;
|
|
|
|
function normalizeText(text) {
|
|
return text
|
|
.toLowerCase()
|
|
.replace(/[^\w\s]/g, " ")
|
|
.replace(/\s+/g, " ")
|
|
.trim();
|
|
}
|
|
|
|
function cosineSimilarity(a, b) {
|
|
let dot = 0;
|
|
let magA = 0;
|
|
let magB = 0;
|
|
|
|
for (let i = 0; i < a.length; i++) {
|
|
dot += a[i] * b[i];
|
|
magA += a[i] * a[i];
|
|
magB += b[i] * b[i];
|
|
}
|
|
|
|
return dot / (Math.sqrt(magA) * Math.sqrt(magB));
|
|
}
|
|
|
|
async function initModel() {
|
|
extractor = await pipeline("feature-extraction", `${config.NLP_TRANSFORMER}`, {
|
|
dtype: 'fp32',
|
|
quantized: false,
|
|
device: "cpu",
|
|
});
|
|
console.log("MiniLM model loaded");
|
|
}
|
|
|
|
async function embed(text) {
|
|
const output = await extractor(normalizeText(text), {
|
|
pooling: "mean",
|
|
normalize: true,
|
|
});
|
|
|
|
return Array.from(output.data);
|
|
}
|
|
|
|
async function classifyIntent(text) {
|
|
const inputEmbedding = await embed(text);
|
|
|
|
const INTENTS = await getIntents();
|
|
|
|
let best = {
|
|
name: "unknown",
|
|
confidence: 0.5,
|
|
matched: null,
|
|
};
|
|
|
|
for (const intent of INTENTS) {
|
|
for (const example of intent.examples) {
|
|
const exampleEmbedding = await embed(example);
|
|
const score = cosineSimilarity(inputEmbedding, exampleEmbedding);
|
|
|
|
if (score > best.confidence) {
|
|
best = {
|
|
name: intent.name,
|
|
confidence: score,
|
|
matched: example,
|
|
};
|
|
}
|
|
}
|
|
}
|
|
|
|
return best;
|
|
}
|
|
|
|
async function handleIntent(streamKey, msg) {
|
|
const text = msg.message?.text;
|
|
const deviceId = msg.message?.device_id;
|
|
|
|
if (!text) return;
|
|
|
|
const intent = await classifyIntent(text);
|
|
|
|
await redis.client.xAdd(streamKey, '*', {
|
|
transcript_id: msg.id,
|
|
command: intent.name,
|
|
confidence: String(intent.confidence),
|
|
matched: intent.matched || "",
|
|
device_id: deviceId || ""
|
|
});
|
|
|
|
console.log("Handling intent", intent);
|
|
}
|
|
|
|
async function processor(group, consumer, streamKey) {
|
|
await redis.client.connect().catch(console.error);
|
|
|
|
while (true) {
|
|
const res = await redis.client.xReadGroup(
|
|
group,
|
|
consumer,
|
|
[{key: streamKey, id: '>'}], // '>' means undelivered messages
|
|
{COUNT: 5, BLOCK:3000}
|
|
);
|
|
|
|
if (!res) continue;
|
|
|
|
for (const stream of res) {
|
|
for (const msg of stream.messages) {
|
|
try {
|
|
console.log(`[${consumer}] Processing:`, msg);
|
|
|
|
await handleIntent("commands", msg);
|
|
|
|
await redis.client.xAck(streamKey, group, msg.id);
|
|
} catch (err) {
|
|
console.error("Failed to process message:", err);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
const main = async () => {
|
|
// Sync DB (not forced in production)
|
|
const doForce = config.FORCE_DB_SYNC === "true" && config.NODE_ENV !== "production";
|
|
await sequelize.authenticate();
|
|
await sequelize.sync({force: doForce});
|
|
|
|
if (doForce) console.info("Database dropped and resynced (force=true).");
|
|
|
|
await initModel();
|
|
|
|
await redis.ensureGroup("transcripts", 'nlp');
|
|
await redis.ensureGroup("commands", 'nlp');
|
|
|
|
processor('nlp', `${config.DEVICE_ID}`, "transcripts").catch(console.error);
|
|
}
|
|
|
|
main().catch(console.error);
|