umm ai bridge upgrades
This commit is contained in:
@@ -179,6 +179,117 @@ function httpJsonPost(urlString, payload, timeoutMs) {
|
||||
});
|
||||
}
|
||||
|
||||
// Streaming NDJSON POST — reads the server response line-by-line and calls
|
||||
// onSentence(sentenceString) for each complete sentence as it arrives.
|
||||
// This replicates the original Jibo hub pattern: the robot starts speaking
|
||||
// while the server is still generating the rest of the response.
|
||||
// Returns a Promise that resolves to the full reply string.
|
||||
function httpStreamingPost(urlString, payload, onSentence, timeoutMs) {
|
||||
timeoutMs = typeof timeoutMs === "number" ? timeoutMs : 120000;
|
||||
|
||||
var parsed = urlLib.parse(urlString);
|
||||
var isHttps = parsed.protocol === "https:";
|
||||
var bodyStr = JSON.stringify(payload || {});
|
||||
var body = new Buffer(bodyStr, "utf8");
|
||||
|
||||
var requestOptions = {
|
||||
protocol: parsed.protocol,
|
||||
hostname: parsed.hostname,
|
||||
port: parsed.port || (isHttps ? 443 : 80),
|
||||
path: parsed.path || "/",
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
"Content-Length": body.length,
|
||||
},
|
||||
timeout: timeoutMs,
|
||||
};
|
||||
|
||||
return new Promise(function (resolve, reject) {
|
||||
var req = (isHttps ? https : http).request(requestOptions, function (res) {
|
||||
if (res.statusCode && (res.statusCode < 200 || res.statusCode >= 300)) {
|
||||
var errChunks = [];
|
||||
res.on("data", function (d) { errChunks.push(d); });
|
||||
res.on("end", function () {
|
||||
reject(new Error("AI Bridge streaming: HTTP " + res.statusCode + ": " + Buffer.concat(errChunks).toString("utf8")));
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
var fullReply = "";
|
||||
var lineBuf = "";
|
||||
|
||||
res.on("data", function (chunk) {
|
||||
lineBuf += chunk.toString("utf8");
|
||||
|
||||
// Process complete lines (NDJSON = one JSON object per line)
|
||||
var lines = lineBuf.split("\n");
|
||||
// Keep the last (possibly incomplete) line in the buffer
|
||||
lineBuf = lines.pop() || "";
|
||||
|
||||
for (var i = 0; i < lines.length; i++) {
|
||||
var line = lines[i].replace(/^\s+|\s+$/g, "");
|
||||
if (!line) continue;
|
||||
try {
|
||||
var obj = JSON.parse(line);
|
||||
} catch (e) {
|
||||
continue; // skip malformed lines
|
||||
}
|
||||
|
||||
if (obj.done && obj.reply) {
|
||||
fullReply = obj.reply;
|
||||
}
|
||||
if (obj.sentence && !obj.done) {
|
||||
try {
|
||||
onSentence(obj.sentence);
|
||||
} catch (e) {
|
||||
// don't let TTS errors kill the stream
|
||||
}
|
||||
}
|
||||
// Handle error/fallback in the done+sentence case (ollama down)
|
||||
if (obj.done && obj.sentence) {
|
||||
try {
|
||||
onSentence(obj.sentence);
|
||||
} catch (e) {
|
||||
// ignore
|
||||
}
|
||||
if (!fullReply) fullReply = obj.sentence;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
res.on("end", function () {
|
||||
// Process any remaining data in the buffer
|
||||
var remaining = (lineBuf || "").replace(/^\s+|\s+$/g, "");
|
||||
if (remaining) {
|
||||
try {
|
||||
var obj = JSON.parse(remaining);
|
||||
if (obj.reply) fullReply = obj.reply;
|
||||
if (obj.sentence && !obj.done) {
|
||||
try { onSentence(obj.sentence); } catch (e) { /* ignore */ }
|
||||
}
|
||||
if (obj.done && obj.sentence) {
|
||||
try { onSentence(obj.sentence); } catch (e) { /* ignore */ }
|
||||
if (!fullReply) fullReply = obj.sentence;
|
||||
}
|
||||
} catch (e) {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
resolve(fullReply || "");
|
||||
});
|
||||
});
|
||||
|
||||
req.on("error", reject);
|
||||
req.on("timeout", function () {
|
||||
req.destroy(new Error("AI Bridge: streaming request timeout"));
|
||||
});
|
||||
|
||||
req.write(body);
|
||||
req.end();
|
||||
});
|
||||
}
|
||||
|
||||
function httpJsonPostRaw(urlString, payload, timeoutMs) {
|
||||
timeoutMs = typeof timeoutMs === "number" ? timeoutMs : 6000;
|
||||
|
||||
@@ -2481,7 +2592,7 @@ AIBridge.prototype._sendText = function (text, source) {
|
||||
var t0 = Date.now();
|
||||
var url = self.serverBaseUrl.replace(/\/+$/, "") + "/v1/chat/text";
|
||||
if (rlog) {
|
||||
rlog.info("ai-bridge", "sending text", {
|
||||
rlog.info("ai-bridge", "sending text (streaming)", {
|
||||
source: source || "text",
|
||||
chars: String(text).length,
|
||||
url: url,
|
||||
@@ -2489,15 +2600,71 @@ AIBridge.prototype._sendText = function (text, source) {
|
||||
});
|
||||
}
|
||||
|
||||
return httpJsonPost(url, { text: text })
|
||||
.then(function (resp) {
|
||||
// TTS sentence queue: speak sentences in order, overlapping network + TTS.
|
||||
// This is how the original Jibo system achieved sub-5-second perceived latency:
|
||||
// start speaking the first sentence while the LLM is still generating the rest.
|
||||
var ttsQueue = [];
|
||||
var ttsRunning = false;
|
||||
var firstSentenceAt = 0;
|
||||
|
||||
function drainTtsQueue() {
|
||||
if (ttsRunning) return;
|
||||
if (ttsQueue.length === 0) return;
|
||||
ttsRunning = true;
|
||||
var sentence = ttsQueue.shift();
|
||||
if (!firstSentenceAt) {
|
||||
firstSentenceAt = Date.now();
|
||||
if (rlog) {
|
||||
rlog.info("ai-bridge", "text request complete", { ms: Date.now() - t0, ok: !!(resp && resp.reply) });
|
||||
rlog.info("ai-bridge", "first sentence TTS start", {
|
||||
ms: firstSentenceAt - t0,
|
||||
chars: String(sentence).length,
|
||||
sentence: String(sentence).slice(0, 120),
|
||||
});
|
||||
}
|
||||
var reply = resp && resp.reply ? String(resp.reply) : "";
|
||||
if (!reply) return { reply: "" };
|
||||
return self._speak(reply).then(function () {
|
||||
return { reply: reply };
|
||||
}
|
||||
self._speak(sentence).then(function () {
|
||||
ttsRunning = false;
|
||||
drainTtsQueue();
|
||||
}).catch(function () {
|
||||
ttsRunning = false;
|
||||
drainTtsQueue();
|
||||
});
|
||||
}
|
||||
|
||||
function onSentence(sentence) {
|
||||
if (!sentence || !String(sentence).trim()) return;
|
||||
if (rlog) {
|
||||
rlog.info("ai-bridge", "stream sentence received", {
|
||||
ms: Date.now() - t0,
|
||||
chars: String(sentence).length,
|
||||
sentence: String(sentence).slice(0, 120),
|
||||
queueLen: ttsQueue.length,
|
||||
});
|
||||
}
|
||||
ttsQueue.push(String(sentence));
|
||||
drainTtsQueue();
|
||||
}
|
||||
|
||||
return httpStreamingPost(url, { text: text }, onSentence)
|
||||
.then(function (fullReply) {
|
||||
if (rlog) {
|
||||
rlog.info("ai-bridge", "streaming request complete", {
|
||||
ms: Date.now() - t0,
|
||||
firstSentenceMs: firstSentenceAt ? firstSentenceAt - t0 : null,
|
||||
replyChars: String(fullReply).length,
|
||||
});
|
||||
}
|
||||
|
||||
// Wait for remaining TTS queue to finish before releasing in-flight.
|
||||
return new Promise(function (resolve) {
|
||||
function waitForTts() {
|
||||
if (ttsQueue.length === 0 && !ttsRunning) {
|
||||
resolve({ reply: fullReply || "" });
|
||||
} else {
|
||||
setTimeout(waitForTts, 100);
|
||||
}
|
||||
}
|
||||
waitForTts();
|
||||
});
|
||||
})
|
||||
.catch(function (e) {
|
||||
|
||||
@@ -13,6 +13,7 @@ var state = {
|
||||
ws: null,
|
||||
subId: null,
|
||||
reconnectTimer: null,
|
||||
lastProcessed: {},
|
||||
};
|
||||
|
||||
// Robot logger (available on BE runtime)
|
||||
@@ -43,6 +44,51 @@ function parseWsUrl(s) {
|
||||
try { return String(s || '').trim(); } catch (e) { return DEFAULT_WS; }
|
||||
}
|
||||
|
||||
function getCandidateWsUrls(beRuntime) {
|
||||
var list = [];
|
||||
try {
|
||||
var envUrl = process.env.ROSBRIDGE_WS;
|
||||
if (envUrl) list.push(parseWsUrl(envUrl));
|
||||
} catch (e) {}
|
||||
try {
|
||||
var cfg = beRuntime && beRuntime.config && beRuntime.config.rosbridge && beRuntime.config.rosbridge.ws;
|
||||
if (cfg) list.push(parseWsUrl(cfg));
|
||||
} catch (e) {}
|
||||
|
||||
// Common fallbacks
|
||||
list.push('ws://127.0.0.1:9090');
|
||||
list.push(DEFAULT_WS);
|
||||
|
||||
// Attempt gateway-derived host if available
|
||||
try {
|
||||
var os = require('os');
|
||||
var ifaces = os.networkInterfaces();
|
||||
Object.keys(ifaces || {}).forEach(function (k) {
|
||||
(ifaces[k] || []).forEach(function (info) {
|
||||
if (!info || info.internal || info.family !== 'IPv4') return;
|
||||
var parts = String(info.address).split('.');
|
||||
if (parts.length === 4) {
|
||||
// guess the gateway as .1
|
||||
parts[3] = '1';
|
||||
list.push('ws://' + parts.join('.') + ':9090');
|
||||
}
|
||||
});
|
||||
});
|
||||
} catch (e) {}
|
||||
|
||||
// Deduplicate while keeping order
|
||||
var seen = {};
|
||||
var out = [];
|
||||
for (var i = 0; i < list.length; i++) {
|
||||
try {
|
||||
var v = String(list[i] || '').trim();
|
||||
if (!v) continue;
|
||||
if (!seen[v]) { seen[v] = true; out.push(v); }
|
||||
} catch (e) {}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
function sendWs(obj) {
|
||||
try {
|
||||
if (!state.ws || state.ws.readyState !== 1) {
|
||||
@@ -103,8 +149,29 @@ function connect(wsUrl, onMessage) {
|
||||
try { data = JSON.parse(evt.data); } catch (e) { rlogWarn('rosbridge', 'json parse failed', { err: String(e), raw: String(evt && evt.data) }); return; }
|
||||
// rosbridge wraps messages with { op: 'publish', topic: '...', msg: {...} }
|
||||
if (data && data.op === 'publish') {
|
||||
rlogInfo('rosbridge', 'publish received', { topic: data.topic, msg: data.msg });
|
||||
if (data.msg) onMessage && onMessage(data.msg, data.topic);
|
||||
try {
|
||||
var topic = data.topic || 'unknown';
|
||||
// Throttle frequent messages per-topic to avoid blocking the BE event loop.
|
||||
var minMs = parseInt(process.env.ROSBRIDGE_MIN_INTERVAL_MS || '200', 10) || 200;
|
||||
var now = Date.now();
|
||||
var last = state.lastProcessed[topic] || 0;
|
||||
if (now - last < minMs) {
|
||||
rlogInfo('rosbridge', 'throttled publish', { topic: topic, droppedMs: now - last, minMs: minMs });
|
||||
return;
|
||||
}
|
||||
state.lastProcessed[topic] = now;
|
||||
|
||||
// Defer handling so heavy work doesn't block the socket message parser.
|
||||
var handler = function () {
|
||||
try {
|
||||
rlogInfo('rosbridge', 'publish received', { topic: data.topic, msg: data.msg });
|
||||
if (data.msg) onMessage && onMessage(data.msg, data.topic);
|
||||
} catch (e) { rlogWarn('rosbridge', 'publish handler error', { err: String(e) }); }
|
||||
};
|
||||
if (typeof setImmediate === 'function') setImmediate(handler); else setTimeout(handler, 0);
|
||||
} catch (e) {
|
||||
rlogWarn('rosbridge', 'error handling publish', { err: String(e) });
|
||||
}
|
||||
return;
|
||||
}
|
||||
rlogInfo('rosbridge', 'ws message', { op: data && data.op, data: data });
|
||||
@@ -124,6 +191,82 @@ function scheduleReconnect(wsUrl, onMessage) {
|
||||
}, 5000);
|
||||
}
|
||||
|
||||
// Try a list of candidate URLs sequentially until one connects.
|
||||
function connectToCandidates(beRuntime, onMessage) {
|
||||
var candidates = getCandidateWsUrls(beRuntime);
|
||||
var idx = 0;
|
||||
|
||||
function tryNext() {
|
||||
if (state.ws && state.ws.readyState === 1) return; // already connected
|
||||
if (idx >= candidates.length) {
|
||||
rlogWarn('rosbridge', 'no candidates left, will schedule reconnect');
|
||||
scheduleReconnect(candidates[0], onMessage);
|
||||
return;
|
||||
}
|
||||
var url = candidates[idx++];
|
||||
rlogInfo('rosbridge', 'trying candidate', { url: url });
|
||||
|
||||
// attempt connect and use a short timeout to move to next candidate
|
||||
var tried = false;
|
||||
var timeout = setTimeout(function () {
|
||||
if (tried) return;
|
||||
tried = true;
|
||||
try { if (state.ws) state.ws.close(); } catch (e) {}
|
||||
rlogWarn('rosbridge', 'candidate timeout, trying next', { url: url });
|
||||
// small delay before next
|
||||
setTimeout(tryNext, 250);
|
||||
}, 3500);
|
||||
|
||||
try {
|
||||
// reuse existing connect path but attach temporary handlers
|
||||
var prevOnOpen = state.ws && state.ws.onopen;
|
||||
connect(url, function (msg, topic) {
|
||||
clearTimeout(timeout);
|
||||
onMessage && onMessage(msg, topic);
|
||||
});
|
||||
// when open, cancel other attempts
|
||||
(function (u) {
|
||||
var wsInst = state.ws;
|
||||
if (!wsInst) return;
|
||||
var origOnOpen = wsInst.onopen;
|
||||
wsInst.onopen = function (ev) {
|
||||
clearTimeout(timeout);
|
||||
rlogInfo('rosbridge', 'connected candidate', { url: u });
|
||||
try { if (typeof origOnOpen === 'function') origOnOpen.call(wsInst, ev); } catch (e) {}
|
||||
};
|
||||
// if it closes or errors before open, try next
|
||||
var origOnClose = wsInst.onclose;
|
||||
wsInst.onclose = function (ev) {
|
||||
clearTimeout(timeout);
|
||||
if (!tried) {
|
||||
tried = true;
|
||||
rlogWarn('rosbridge', 'candidate closed before ready, next', { url: u });
|
||||
setTimeout(tryNext, 250);
|
||||
}
|
||||
try { if (typeof origOnClose === 'function') origOnClose.call(wsInst, ev); } catch (e) {}
|
||||
};
|
||||
var origOnError = wsInst.onerror;
|
||||
wsInst.onerror = function (err) {
|
||||
clearTimeout(timeout);
|
||||
if (!tried) {
|
||||
tried = true;
|
||||
rlogWarn('rosbridge', 'candidate error, next', { url: u, err: String(err) });
|
||||
try { if (wsInst) wsInst.close(); } catch (e) {}
|
||||
setTimeout(tryNext, 250);
|
||||
}
|
||||
try { if (typeof origOnError === 'function') origOnError.call(wsInst, err); } catch (e) {}
|
||||
};
|
||||
})(url);
|
||||
} catch (e) {
|
||||
clearTimeout(timeout);
|
||||
rlogWarn('rosbridge', 'connect threw, trying next', { url: url, err: String(e) });
|
||||
setTimeout(tryNext, 250);
|
||||
}
|
||||
}
|
||||
|
||||
tryNext();
|
||||
}
|
||||
|
||||
function close() {
|
||||
rlogInfo('rosbridge', 'close requested');
|
||||
try { unsubscribe(); } catch (e) { rlogWarn('rosbridge', 'unsubscribe failed', { err: String(e) }); }
|
||||
@@ -255,8 +398,7 @@ exports.init = function (beRuntime, jibo) {
|
||||
jibo.tts.speak(payload, { mode: jibo.tts.TTSMode ? jibo.tts.TTSMode.SSML : undefined });
|
||||
} else if (jibo && jibo.tts && typeof jibo.tts.speak === 'function') {
|
||||
jibo.tts.speak(String(t), { mode: jibo.tts.TTSMode ? jibo.tts.TTSMode.TEXT : undefined });
|
||||
} else if (beRuntime && beRuntime.api && typeof beRuntime.api.speak === 'function') {
|
||||
beRuntime.api.speak({ text: String(t), mode: useEsml ? 'ssml' : 'text' });
|
||||
} else if (beRuntime && beRuntime.api && typeof beRuntime.api.speak === 'function') { beRuntime.api.speak({ text: String(t), mode: useEsml ? 'ssml' : 'text' });
|
||||
} else if (jibo && jibo.api && typeof jibo.api.speak === 'function') {
|
||||
jibo.api.speak({ text: String(t), mode: useEsml ? 'ssml' : 'text' });
|
||||
} else {
|
||||
|
||||
Reference in New Issue
Block a user