454 lines
14 KiB
JavaScript
454 lines
14 KiB
JavaScript
/**
|
|
* Groq Chat Completions for the agent loop. Host is pinned to api.groq.com.
|
|
* Tools still run locally; only chat tokens leave the machine.
|
|
*/
|
|
|
|
const fs = require('fs');
|
|
const https = require('https');
|
|
const path = require('path');
|
|
const completeWatch = require('./complete-watch.js');
|
|
const toolParse = require('./tool-parse.js');
|
|
|
|
const GROQ_HOST = 'api.groq.com';
|
|
const GROQ_PATH = '/openai/v1/chat/completions';
|
|
const MAX_IMAGE_BYTES = 4 * 1024 * 1024;
|
|
const VISION_MODELS = /^qwen\/qwen3\.[68]-27b$/i;
|
|
const NO_PARALLEL = /gpt-oss|qwen3\.8-27b/i;
|
|
|
|
let httpsRequest = https.request;
|
|
const activeRequests = new Set();
|
|
|
|
function setHttpsRequest(fn) {
|
|
httpsRequest = typeof fn === 'function' ? fn : https.request;
|
|
}
|
|
|
|
function groqHost() {
|
|
return GROQ_HOST;
|
|
}
|
|
|
|
function visionModel(model) {
|
|
return VISION_MODELS.test(String(model || ''));
|
|
}
|
|
|
|
function redact(text) {
|
|
return String(text || '')
|
|
.replace(/Bearer\s+\S+/gi, 'Bearer [redacted]')
|
|
.replace(/gsk_[A-Za-z0-9]+/g, 'gsk_[redacted]');
|
|
}
|
|
|
|
function fail(message) {
|
|
return new Error(redact(message));
|
|
}
|
|
|
|
function mimeForPath(file) {
|
|
const ext = path.extname(file).toLowerCase();
|
|
if (ext === '.png') return 'image/png';
|
|
if (ext === '.webp') return 'image/webp';
|
|
if (ext === '.gif') return 'image/gif';
|
|
return 'image/jpeg';
|
|
}
|
|
|
|
function imagePart(file) {
|
|
const buf = fs.readFileSync(file);
|
|
if (buf.length > MAX_IMAGE_BYTES) throw new Error('Groq vision still is too large');
|
|
return {
|
|
type: 'image_url',
|
|
image_url: { url: 'data:' + mimeForPath(file) + ';base64,' + buf.toString('base64') },
|
|
};
|
|
}
|
|
|
|
function toOpenAiTools(tools) {
|
|
if (!tools || !Array.isArray(tools) || !tools.length) return undefined;
|
|
const out = [];
|
|
for (const t of tools) {
|
|
const fn = t && t.function && t.type === 'function' ? t.function : t;
|
|
const name = fn && fn.name;
|
|
if (!name) continue;
|
|
const parameters = fn.parameters && fn.parameters.type === 'object'
|
|
? fn.parameters
|
|
: { type: 'object', properties: (fn.parameters && fn.parameters.properties) || {} };
|
|
out.push({
|
|
type: 'function',
|
|
function: {
|
|
name,
|
|
description: fn.description || name,
|
|
parameters,
|
|
},
|
|
});
|
|
}
|
|
return out.length ? out : undefined;
|
|
}
|
|
|
|
function contentWithImages(msg, vision) {
|
|
const text = msg.content == null ? '' : String(msg.content);
|
|
const attachments = Array.isArray(msg.attachments) ? msg.attachments : [];
|
|
if (!vision || !attachments.length) return text;
|
|
const parts = [];
|
|
if (text) parts.push({ type: 'text', text });
|
|
for (const att of attachments) {
|
|
const file = att && (att.path || att.file);
|
|
if (!file) continue;
|
|
try {
|
|
parts.push(imagePart(file));
|
|
} catch (err) {
|
|
parts.push({ type: 'text', text: '[image unavailable: ' + redact(err.message) + ']' });
|
|
}
|
|
}
|
|
return parts.length ? parts : text;
|
|
}
|
|
|
|
function mapToolCalls(calls) {
|
|
return (calls || []).map((c, i) => {
|
|
const name = c.name || (c.function && c.function.name);
|
|
const args = c.arguments != null ? c.arguments : c.args != null ? c.args : (c.function && c.function.arguments);
|
|
return {
|
|
id: c.id || 'call_' + i,
|
|
type: 'function',
|
|
function: {
|
|
name,
|
|
arguments: typeof args === 'string' ? args : JSON.stringify(args || {}),
|
|
},
|
|
};
|
|
}).filter((c) => c.function.name);
|
|
}
|
|
|
|
function reasoningEffort(model) {
|
|
if (/gpt-oss/i.test(model)) return 'low';
|
|
if (/qwen3\.[68]-27b/i.test(model)) return 'none';
|
|
return null;
|
|
}
|
|
|
|
function assertChatModel(model) {
|
|
if (/compound/i.test(model)) {
|
|
throw new Error('groq/compound runs tools on Groq. Pick a chat model so Jarvis tools stay on this computer.');
|
|
}
|
|
}
|
|
|
|
function toOpenAiMessages(history, { vision } = {}) {
|
|
const out = [];
|
|
for (const msg of Array.isArray(history) ? history : []) {
|
|
if (!msg || !msg.role) continue;
|
|
if (msg.role === 'tool' || msg.role === 'function') {
|
|
out.push({
|
|
role: 'tool',
|
|
tool_call_id: msg.tool_call_id || msg.toolCallId || '',
|
|
content: msg.content == null ? '' : typeof msg.content === 'string' ? msg.content : JSON.stringify(msg.content),
|
|
...(msg.name ? { name: msg.name } : {}),
|
|
});
|
|
continue;
|
|
}
|
|
if (msg.role === 'assistant' && Array.isArray(msg.tool_calls) && msg.tool_calls.length) {
|
|
out.push({
|
|
role: 'assistant',
|
|
content: msg.content ? String(msg.content) : null,
|
|
tool_calls: mapToolCalls(msg.tool_calls),
|
|
});
|
|
continue;
|
|
}
|
|
if (msg.role === 'system' || msg.role === 'user' || msg.role === 'assistant') {
|
|
out.push({ role: msg.role, content: contentWithImages(msg, vision) });
|
|
}
|
|
}
|
|
return out;
|
|
}
|
|
|
|
function parseArgs(raw) {
|
|
if (raw && typeof raw === 'object' && !Array.isArray(raw)) return raw;
|
|
if (typeof raw !== 'string') return raw == null ? {} : { value: raw };
|
|
const trimmed = raw.trim();
|
|
if (!trimmed) return {};
|
|
try {
|
|
const parsed = JSON.parse(trimmed);
|
|
if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) return parsed;
|
|
return { value: parsed };
|
|
} catch (_) {
|
|
return { value: trimmed };
|
|
}
|
|
}
|
|
|
|
function harvestedCalls(buckets) {
|
|
const calls = [];
|
|
for (const c of buckets) {
|
|
if (!c || !c.name) continue;
|
|
calls.push({
|
|
id: c.id || '',
|
|
name: c.name,
|
|
arguments: parseArgs(c.arguments),
|
|
});
|
|
}
|
|
return calls;
|
|
}
|
|
|
|
function applyDelta(delta, state, onEvent) {
|
|
if (!delta || typeof delta !== 'object') return;
|
|
if (delta.content) {
|
|
state.text += delta.content;
|
|
if (onEvent) onEvent({ type: 'contentDelta', delta: delta.content });
|
|
}
|
|
const thinking = delta.reasoning || delta.reasoning_content;
|
|
if (thinking) {
|
|
state.thinking += thinking;
|
|
if (onEvent) onEvent({ type: 'thinkingDelta', delta: thinking });
|
|
}
|
|
if (!Array.isArray(delta.tool_calls)) return;
|
|
for (const part of delta.tool_calls) {
|
|
const i = Number.isInteger(part.index) ? part.index : state.toolCalls.length;
|
|
if (!state.toolCalls[i]) state.toolCalls[i] = { id: '', name: '', arguments: '' };
|
|
const slot = state.toolCalls[i];
|
|
if (part.id) slot.id = part.id;
|
|
const fn = part.function || {};
|
|
if (fn.name) slot.name += fn.name;
|
|
if (fn.arguments) slot.arguments += fn.arguments;
|
|
}
|
|
}
|
|
|
|
function consumeSse(chunk, carry, onEvent) {
|
|
let rest = carry + chunk;
|
|
let idx;
|
|
while ((idx = rest.indexOf('\n')) >= 0) {
|
|
let line = rest.slice(0, idx);
|
|
rest = rest.slice(idx + 1);
|
|
if (line.endsWith('\r')) line = line.slice(0, -1);
|
|
if (!line.startsWith('data:')) continue;
|
|
const data = line.slice(5).trim();
|
|
if (!data) continue;
|
|
if (data === '[DONE]') return { rest, done: true };
|
|
try {
|
|
onEvent(JSON.parse(data));
|
|
} catch (_) {}
|
|
}
|
|
return { rest, done: false };
|
|
}
|
|
|
|
function postChat(body, apiKey, { timeoutMs, abortHolder }) {
|
|
return new Promise((resolve, reject) => {
|
|
const payload = JSON.stringify(body);
|
|
let req;
|
|
req = httpsRequest({
|
|
hostname: GROQ_HOST,
|
|
path: GROQ_PATH,
|
|
method: 'POST',
|
|
headers: {
|
|
Authorization: 'Bearer ' + apiKey,
|
|
'Content-Type': 'application/json',
|
|
Accept: 'text/event-stream',
|
|
'Content-Length': Buffer.byteLength(payload),
|
|
},
|
|
}, (res) => {
|
|
const chunks = [];
|
|
res.on('data', (d) => chunks.push(d));
|
|
res.on('end', () => {
|
|
activeRequests.delete(req);
|
|
const raw = Buffer.concat(chunks.map((c) => Buffer.isBuffer(c) ? c : Buffer.from(c))).toString('utf8');
|
|
resolve({ status: res.statusCode || 0, raw, headers: res.headers || {} });
|
|
});
|
|
res.on('error', (err) => {
|
|
activeRequests.delete(req);
|
|
reject(fail(err.message));
|
|
});
|
|
});
|
|
abortHolder.req = req;
|
|
activeRequests.add(req);
|
|
req.on('error', (err) => {
|
|
activeRequests.delete(req);
|
|
reject(fail(err.message));
|
|
});
|
|
if (timeoutMs > 0 && typeof req.setTimeout === 'function') {
|
|
req.setTimeout(timeoutMs, () => {
|
|
try { req.destroy(fail('Groq request timed out')); } catch (_) {}
|
|
});
|
|
}
|
|
req.write(payload);
|
|
req.end();
|
|
});
|
|
}
|
|
|
|
function streamChat(body, apiKey, { onChunk, timeoutMs, abortHolder, watch }) {
|
|
return new Promise((resolve, reject) => {
|
|
const payload = JSON.stringify(body);
|
|
let req;
|
|
let settled = false;
|
|
const finish = (err, value) => {
|
|
if (settled) return;
|
|
settled = true;
|
|
activeRequests.delete(req);
|
|
if (err) reject(err);
|
|
else resolve(value);
|
|
};
|
|
req = httpsRequest({
|
|
hostname: GROQ_HOST,
|
|
path: GROQ_PATH,
|
|
method: 'POST',
|
|
headers: {
|
|
Authorization: 'Bearer ' + apiKey,
|
|
'Content-Type': 'application/json',
|
|
Accept: 'text/event-stream',
|
|
'Content-Length': Buffer.byteLength(payload),
|
|
},
|
|
}, (res) => {
|
|
if ((res.statusCode || 0) >= 400) {
|
|
const chunks = [];
|
|
res.on('data', (d) => chunks.push(d));
|
|
res.on('end', () => {
|
|
const raw = Buffer.concat(chunks.map((c) => Buffer.isBuffer(c) ? c : Buffer.from(c))).toString('utf8');
|
|
finish(fail('Groq HTTP ' + res.statusCode + ': ' + raw.slice(0, 400)));
|
|
});
|
|
res.on('error', (err) => finish(fail(err.message)));
|
|
return;
|
|
}
|
|
let carry = '';
|
|
res.on('data', (d) => {
|
|
if (watch && watch.timedOut()) {
|
|
try { req.destroy(); } catch (_) {}
|
|
return;
|
|
}
|
|
if (watch) watch.bump();
|
|
const text = Buffer.isBuffer(d) ? d.toString('utf8') : String(d);
|
|
const parsed = consumeSse(text, carry, onChunk);
|
|
carry = parsed.rest;
|
|
if (parsed.done) {
|
|
finish(null, true);
|
|
try { req.destroy(); } catch (_) {}
|
|
}
|
|
});
|
|
res.on('end', () => finish(null, true));
|
|
res.on('error', (err) => finish(fail(err.message)));
|
|
});
|
|
abortHolder.req = req;
|
|
activeRequests.add(req);
|
|
req.on('error', (err) => finish(fail(err.message)));
|
|
if (timeoutMs > 0 && typeof req.setTimeout === 'function') {
|
|
req.setTimeout(timeoutMs, () => {
|
|
try { req.destroy(); } catch (_) {}
|
|
finish(fail('Groq request timed out'));
|
|
});
|
|
}
|
|
req.write(payload);
|
|
req.end();
|
|
});
|
|
}
|
|
|
|
function cancel() {
|
|
for (const req of activeRequests) {
|
|
try { req.destroy(); } catch (_) {}
|
|
}
|
|
activeRequests.clear();
|
|
}
|
|
|
|
async function complete(opts, onEvent) {
|
|
const apiKey = String((opts && opts.apiKey) || '').trim();
|
|
const model = String((opts && opts.model) || '').trim();
|
|
if (!apiKey) throw new Error('Groq agent inference needs a Groq API key. Set it in Settings or GROQ_API_KEY.');
|
|
if (!model) throw new Error('Groq agent inference needs a groqModel.');
|
|
assertChatModel(model);
|
|
const vision = visionModel(model);
|
|
const tools = toOpenAiTools(opts && opts.tools);
|
|
const messages = toOpenAiMessages((opts && opts.history) || (opts && opts.messages) || [], { vision });
|
|
const effort = reasoningEffort(model);
|
|
const body = {
|
|
model,
|
|
messages,
|
|
stream: opts && opts.stream === false ? false : true,
|
|
temperature: 0.6,
|
|
...(effort ? { reasoning_effort: effort } : {}),
|
|
};
|
|
if (tools) {
|
|
body.tools = tools;
|
|
body.tool_choice = 'auto';
|
|
if (NO_PARALLEL.test(model)) body.parallel_tool_calls = false;
|
|
}
|
|
const abortHolder = { req: null };
|
|
const abortRun = () => {
|
|
try { if (abortHolder.req) abortHolder.req.destroy(); } catch (_) {}
|
|
};
|
|
let settleTimeout;
|
|
const timedOutGate = new Promise((resolve) => { settleTimeout = () => resolve('timeout'); });
|
|
const watch = completeWatch.attachCompleteWatch({
|
|
timeoutMs: opts && opts.timeoutMs,
|
|
idleMs: opts && opts.idleMs,
|
|
abort: abortRun,
|
|
onTimeout: settleTimeout,
|
|
});
|
|
const state = { text: '', thinking: '', toolCalls: [] };
|
|
try {
|
|
if (body.stream === false) {
|
|
const res = await Promise.race([
|
|
postChat(body, apiKey, { timeoutMs: (opts && opts.timeoutMs) || 0, abortHolder }),
|
|
timedOutGate.then(() => null),
|
|
]);
|
|
if (!res || watch.timedOut()) {
|
|
return { text: '', thinking: '', toolCalls: [], stats: null, requestId: null, stopReason: 'timeout' };
|
|
}
|
|
if (res.status >= 400) throw fail('Groq HTTP ' + res.status + ': ' + res.raw.slice(0, 400));
|
|
let parsed;
|
|
try { parsed = JSON.parse(res.raw); } catch (err) { throw fail('Groq returned invalid JSON'); }
|
|
const msg = parsed && parsed.choices && parsed.choices[0] && parsed.choices[0].message;
|
|
if (msg) {
|
|
if (msg.content) state.text = String(msg.content);
|
|
if (msg.reasoning || msg.reasoning_content) state.thinking = String(msg.reasoning || msg.reasoning_content);
|
|
if (Array.isArray(msg.tool_calls)) {
|
|
for (const call of msg.tool_calls) {
|
|
state.toolCalls.push({
|
|
id: call.id || '',
|
|
name: (call.function && call.function.name) || '',
|
|
arguments: (call.function && call.function.arguments) || '',
|
|
});
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
const consume = streamChat(body, apiKey, {
|
|
timeoutMs: (opts && opts.timeoutMs) || 0,
|
|
abortHolder,
|
|
watch,
|
|
onChunk: (evt) => {
|
|
const choice = evt && evt.choices && evt.choices[0];
|
|
if (!choice) return;
|
|
applyDelta(choice.delta || {}, state, onEvent);
|
|
if (choice.message) applyDelta(choice.message, state, onEvent);
|
|
},
|
|
});
|
|
consume.catch(() => {});
|
|
const raced = await Promise.race([consume.then(() => 'ok'), timedOutGate]);
|
|
if (raced !== 'ok' && raced !== 'timeout') throw raced;
|
|
}
|
|
const toolCalls = harvestedCalls(state.toolCalls);
|
|
let text = state.text;
|
|
let thinking = state.thinking;
|
|
if (tools && tools.length) {
|
|
const recovered = toolParse.recover({ text, thinking, tools: opts.tools, existing: toolCalls });
|
|
if (!toolCalls.length && recovered.calls.length) {
|
|
for (const call of recovered.calls) toolCalls.push(call);
|
|
}
|
|
text = recovered.text;
|
|
} else {
|
|
text = toolParse.stripToolMarkup(text);
|
|
}
|
|
return {
|
|
text,
|
|
thinking,
|
|
toolCalls,
|
|
stats: null,
|
|
requestId: null,
|
|
stopReason: watch.timedOut() ? 'timeout' : 'stop',
|
|
};
|
|
} finally {
|
|
watch.clear();
|
|
activeRequests.delete(abortHolder.req);
|
|
}
|
|
}
|
|
|
|
module.exports = {
|
|
GROQ_HOST,
|
|
GROQ_PATH,
|
|
groqHost,
|
|
visionModel,
|
|
reasoningEffort,
|
|
toOpenAiTools,
|
|
toOpenAiMessages,
|
|
setHttpsRequest,
|
|
complete,
|
|
cancel,
|
|
redact,
|
|
};
|