Merge pull request 'STT' (#7) from STT into main

Reviewed-on: Spilum.Net/Many-Tailed-Fox#7

package.json

@@ -12,6 +12,7 @@
   "license": "ISC",
   "type": "commonjs",
   "dependencies": {
+    "@xenova/transformers": "^2.17.2",
     "pg": "^8.22.0",
     "redis": "^6.1.0",
     "sequelize": "^6.37.8"

src/server.js

@@ -3,6 +3,8 @@ if (process.env.NODE_ENV !== "production") {
 	require("dotenv").config();
 }
 
+const {pipeline} = require("@xenova/transformers");
+
 const models = require("./models");
 const db = models.db;
 const redis = models.redis;
@@ -10,20 +12,110 @@ const {sequelize, Sequelize} = db;
 
 const config = require("./config");
 
-const handleIntent = async (streamKey, intent) => {
+const INTENTS = [
-	await redis.client.xAdd(
-		streamKey,
-		'*',
   {
-			transcript_id: intent.id,
+    name: "play_music",
-			command: "[TEST] INTENT_RECEIVED",
+    examples: ["play music", "start music", "resume music", "play songs"],
-			device_id: intent.message.device_id
+  },
+  {
+    name: "stop_music",
+    examples: ["stop music", "pause music", "halt music", "stop playback"],
+  },
+  {
+    name: "open_browser",
+    examples: ["open browser", "launch browser", "start chrome", "open chrome"],
+  },
+  {
+    name: "set_volume",
+    examples: ["set volume to 30", "volume 50", "turn volume up", "turn volume down"],
+  },
+];
+
+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", "Xenova/all-MiniLM-L6-v2");
+}
+
+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);
+
+  let best = {
+    name: "unknown",
+    confidence: 0,
+    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 || "",
+		text,
+	});
+
 	console.log("TODO: Handle Intents", intent);
 }
 
-const processor = async (group, consumer, streamKey) => {
+async function processor(group, consumer, streamKey) {
 	await redis.client.connect().catch(console.error);
 
 	while (true) {
@@ -38,16 +130,18 @@ const processor = async (group, consumer, streamKey) => {
 
 		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);
+				}
 			}
 		}
 	}
-
-	await redis.client.disconnect();
 }
 
 const main = async () => {
@@ -55,12 +149,15 @@ const main = async () => {
 	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");
+	processor('nlp', `${config.DEVICE_ID}`, "transcripts").catch(console.error);
 }
 
 main().catch(console.error);