wehub-resource-sync
384 lines
14 KiB
JavaScript
384 lines
14 KiB
JavaScript
const { v4: uuidv4 } = require("uuid");
|
|
const moment = require("moment");
|
|
|
|
function clientAbortedHandler(resolve, fullText) {
|
|
console.log(
|
|
"\x1b[43m\x1b[34m[STREAM ABORTED]\x1b[0m Client requested to abort stream. Exiting LLM stream handler early."
|
|
);
|
|
resolve(fullText);
|
|
return;
|
|
}
|
|
|
|
/**
|
|
* Handles the default stream response for a chat.
|
|
* @param {import("express").Response} response
|
|
* @param {import('./LLMPerformanceMonitor').MonitoredStream} stream
|
|
* @param {Object} responseProps
|
|
* @returns {Promise<string>}
|
|
*/
|
|
function handleDefaultStreamResponseV2(response, stream, responseProps) {
|
|
const { uuid = uuidv4(), sources = [] } = responseProps;
|
|
|
|
// Why are we doing this?
|
|
// OpenAI do enable the usage metrics in the stream response but:
|
|
// 1. This parameter is not available in our current API version (TODO: update)
|
|
// 2. The usage metrics are not available in _every_ provider that uses this function
|
|
// 3. We need to track the usage metrics for every provider that uses this function - not just OpenAI
|
|
// Other keys are added by the LLMPerformanceMonitor.measureStream method
|
|
let hasUsageMetrics = false;
|
|
let usage = {
|
|
// prompt_tokens can be in this object if the provider supports it - otherwise we manually count it
|
|
// When the stream is created in the LLMProviders `streamGetChatCompletion` `LLMPerformanceMonitor.measureStream` call.
|
|
completion_tokens: 0,
|
|
};
|
|
|
|
return new Promise(async (resolve) => {
|
|
let fullText = "";
|
|
let reasoningText = "";
|
|
|
|
// Establish listener to early-abort a streaming response
|
|
// in case things go sideways or the user does not like the response.
|
|
// We preserve the generated text but continue as if chat was completed
|
|
// to preserve previously generated content.
|
|
const handleAbort = () => {
|
|
stream?.endMeasurement(usage);
|
|
clientAbortedHandler(resolve, fullText);
|
|
};
|
|
response.on("close", handleAbort);
|
|
|
|
// Now handle the chunks from the streamed response and append to fullText.
|
|
try {
|
|
for await (const chunk of stream) {
|
|
const message = chunk?.choices?.[0];
|
|
const token = message?.delta?.content;
|
|
|
|
// Reasoning token can be in different properties depending on the provider.
|
|
// eg: Cerebras uses `reasoning` instead of `reasoning_content` like OpenAI.
|
|
const reasoningToken =
|
|
message?.delta?.reasoning_content || message?.delta?.reasoning;
|
|
|
|
// If we see usage metrics in the chunk, we can use them directly
|
|
// instead of estimating them, but we only want to assign values if
|
|
// the response object is the exact same key:value pair we expect.
|
|
if (
|
|
chunk.hasOwnProperty("usage") && // exists
|
|
!!chunk.usage && // is not null
|
|
Object.values(chunk.usage).length > 0 // has values
|
|
) {
|
|
if (chunk.usage.hasOwnProperty("prompt_tokens")) {
|
|
usage.prompt_tokens = Number(chunk.usage.prompt_tokens);
|
|
}
|
|
|
|
if (chunk.usage.hasOwnProperty("completion_tokens")) {
|
|
hasUsageMetrics = true; // to stop estimating counter
|
|
usage.completion_tokens = Number(chunk.usage.completion_tokens);
|
|
}
|
|
|
|
// Some providers, like Cerebras, return the completion time in the usage metrics.
|
|
// This is used to report the real-time duration of the completion.
|
|
if (chunk.usage.hasOwnProperty("time_info")) {
|
|
usage.duration = chunk.usage.time_info.completion_time;
|
|
}
|
|
}
|
|
|
|
// Reasoning models will always return the reasoning text before the token text.
|
|
if (reasoningToken) {
|
|
// If the reasoning text is empty (''), we need to initialize it
|
|
// and send the first chunk of reasoning text.
|
|
if (reasoningText.length === 0) {
|
|
writeResponseChunk(response, {
|
|
uuid,
|
|
sources: [],
|
|
type: "textResponseChunk",
|
|
textResponse: `<think>${reasoningToken}`,
|
|
close: false,
|
|
error: false,
|
|
});
|
|
reasoningText += `<think>${reasoningToken}`;
|
|
continue;
|
|
} else {
|
|
writeResponseChunk(response, {
|
|
uuid,
|
|
sources: [],
|
|
type: "textResponseChunk",
|
|
textResponse: reasoningToken,
|
|
close: false,
|
|
error: false,
|
|
});
|
|
reasoningText += reasoningToken;
|
|
}
|
|
}
|
|
|
|
// If the reasoning text is not empty, but the reasoning token is empty
|
|
// and the token text is not empty we need to close the reasoning text and begin sending the token text.
|
|
if (!!reasoningText && !reasoningToken && token) {
|
|
writeResponseChunk(response, {
|
|
uuid,
|
|
sources: [],
|
|
type: "textResponseChunk",
|
|
textResponse: `</think>`,
|
|
close: false,
|
|
error: false,
|
|
});
|
|
fullText += `${reasoningText}</think>`;
|
|
reasoningText = "";
|
|
}
|
|
|
|
if (token) {
|
|
fullText += token;
|
|
// If we never saw a usage metric, we can estimate them by number of completion chunks
|
|
if (!hasUsageMetrics) usage.completion_tokens++;
|
|
writeResponseChunk(response, {
|
|
uuid,
|
|
sources: [],
|
|
type: "textResponseChunk",
|
|
textResponse: token,
|
|
close: false,
|
|
error: false,
|
|
});
|
|
}
|
|
|
|
// LocalAi returns '' and others return null on chunks - the last chunk is not "" or null.
|
|
// Either way, the key `finish_reason` must be present to determine ending chunk.
|
|
if (
|
|
message?.hasOwnProperty("finish_reason") && // Got valid message and it is an object with finish_reason
|
|
message.finish_reason !== "" &&
|
|
message.finish_reason !== null
|
|
) {
|
|
writeResponseChunk(response, {
|
|
uuid,
|
|
sources,
|
|
type: "textResponseChunk",
|
|
textResponse: "",
|
|
close: true,
|
|
error: false,
|
|
});
|
|
response.removeListener("close", handleAbort);
|
|
stream?.endMeasurement(usage);
|
|
resolve(fullText);
|
|
break; // Break streaming when a valid finish_reason is first encountered
|
|
}
|
|
}
|
|
} catch (e) {
|
|
console.log(`\x1b[43m\x1b[34m[STREAMING ERROR]\x1b[0m ${e.message}`);
|
|
writeResponseChunk(response, {
|
|
uuid,
|
|
type: "abort",
|
|
textResponse: null,
|
|
sources: [],
|
|
close: true,
|
|
error: e.message,
|
|
});
|
|
stream?.endMeasurement(usage);
|
|
resolve(fullText); // Return what we currently have - if anything.
|
|
}
|
|
});
|
|
}
|
|
|
|
function convertToChatHistory(history = []) {
|
|
const formattedHistory = [];
|
|
for (const record of history) {
|
|
const { prompt, response, createdAt, feedbackScore = null, id } = record;
|
|
const data = JSON.parse(response);
|
|
|
|
// In the event that a bad response was stored - we should skip its entire record
|
|
// because it was likely an error and cannot be used in chats and will fail to render on UI.
|
|
if (typeof prompt !== "string") {
|
|
console.log(
|
|
`[convertToChatHistory] ChatHistory #${record.id} prompt property is not a string - skipping record.`
|
|
);
|
|
continue;
|
|
} else if (typeof data.text !== "string") {
|
|
console.log(
|
|
`[convertToChatHistory] ChatHistory #${record.id} response.text property is not a string - skipping record.`
|
|
);
|
|
continue;
|
|
}
|
|
|
|
formattedHistory.push([
|
|
{
|
|
role: "user",
|
|
content: prompt,
|
|
sentAt: moment(createdAt).unix(),
|
|
attachments: data?.attachments ?? [],
|
|
chatId: id,
|
|
},
|
|
{
|
|
type: data?.type || "chart",
|
|
role: "assistant",
|
|
content: data.text,
|
|
sources: data.sources || [],
|
|
chatId: id,
|
|
sentAt: moment(createdAt).unix(),
|
|
feedbackScore,
|
|
metrics: data?.metrics || {},
|
|
...(data?.outputs?.length > 0 ? { outputs: data.outputs } : {}),
|
|
...(data?.clarifyingQuestions?.length > 0
|
|
? { clarifyingQuestions: data.clarifyingQuestions }
|
|
: {}),
|
|
},
|
|
]);
|
|
}
|
|
|
|
return formattedHistory.flat();
|
|
}
|
|
|
|
/**
|
|
* Render a single saved survey as a tagged Q/A transcript for LLM history.
|
|
* Mirrors the answer-casing rules in formatAnswersForAgent (request-user-input.js)
|
|
* so the model sees the same wording it saw mid-turn when the tool resolved.
|
|
*/
|
|
function formatClarifyingSurveyForPrompt(survey) {
|
|
const questions = Array.isArray(survey?.questions) ? survey.questions : [];
|
|
const result = survey?.result || {};
|
|
if (!questions.length) return "";
|
|
|
|
let body;
|
|
if (result.timedOut) {
|
|
body = "[no response within the time limit]";
|
|
} else if (result.skipped) {
|
|
body = "[user let the agent decide]";
|
|
} else {
|
|
const answers = Array.isArray(result.answers) ? result.answers : [];
|
|
body = questions
|
|
.map((q, i) => {
|
|
const a = answers[i] || { skipped: true };
|
|
let answerText;
|
|
if (a.skipped) answerText = "[user skipped]";
|
|
else if (Array.isArray(a.answer)) answerText = a.answer.join(", ");
|
|
else if (a.answer === null || a.answer === undefined || a.answer === "")
|
|
answerText = "[no answer]";
|
|
else answerText = String(a.answer);
|
|
return `Q: ${q.question}\nA: ${answerText}`;
|
|
})
|
|
.join("\n");
|
|
}
|
|
return `<clarifying_questions>\n${body}\n</clarifying_questions>`;
|
|
}
|
|
|
|
/**
|
|
* Converts a chat history to a prompt history.
|
|
* @param {Object[]} history - The chat history to convert
|
|
* @returns {{role: string, content: string, attachments?: import("..").Attachment}[]}
|
|
*/
|
|
function convertToPromptHistory(history = []) {
|
|
const formattedHistory = [];
|
|
for (const record of history) {
|
|
const { prompt, response } = record;
|
|
const data = JSON.parse(response);
|
|
|
|
// In the event that a bad response was stored - we should skip its entire record
|
|
// because it was likely an error and cannot be used in chats and will fail to render on UI.
|
|
if (typeof prompt !== "string") {
|
|
console.log(
|
|
`[convertToPromptHistory] ChatHistory #${record.id} prompt property is not a string - skipping record.`
|
|
);
|
|
continue;
|
|
} else if (typeof data.text !== "string") {
|
|
console.log(
|
|
`[convertToPromptHistory] ChatHistory #${record.id} response.text property is not a string - skipping record.`
|
|
);
|
|
continue;
|
|
}
|
|
|
|
// If the agent saved one or more clarifying-question surveys on this
|
|
// record, append them to the assistant content so future LLM turns
|
|
// (agent or normal chat) can recall what the user answered.
|
|
let assistantContent = data.text;
|
|
if (data?.clarifyingQuestions?.length > 0) {
|
|
const surveyBlocks = data.clarifyingQuestions
|
|
.map(formatClarifyingSurveyForPrompt)
|
|
.filter(Boolean)
|
|
.join("\n\n");
|
|
if (surveyBlocks)
|
|
assistantContent = `${assistantContent}\n\n${surveyBlocks}`;
|
|
}
|
|
|
|
formattedHistory.push([
|
|
{
|
|
role: "user",
|
|
content: prompt,
|
|
// if there are attachments, add them as a property to the user message so we can reuse them in chat history later if supported by the llm.
|
|
...(data?.attachments?.length > 0
|
|
? { attachments: data?.attachments }
|
|
: {}),
|
|
},
|
|
{
|
|
role: "assistant",
|
|
content: assistantContent,
|
|
},
|
|
]);
|
|
}
|
|
return formattedHistory.flat();
|
|
}
|
|
|
|
/**
|
|
* Safely stringifies any object containing BigInt values
|
|
* @param {*} obj - Anything to stringify that might contain BigInt values
|
|
* @returns {string} JSON string with BigInt values converted to strings
|
|
*/
|
|
function safeJSONStringify(obj) {
|
|
return JSON.stringify(obj, (_, value) => {
|
|
if (typeof value === "bigint") return value.toString();
|
|
return value;
|
|
});
|
|
}
|
|
|
|
function writeResponseChunk(response, data) {
|
|
response.write(`data: ${safeJSONStringify(data)}\n\n`);
|
|
return;
|
|
}
|
|
|
|
/**
|
|
* Formats the chat history to re-use attachments in the chat history
|
|
* that might have existed in the conversation earlier.
|
|
* @param {{role:string, content:string, attachments?: Object[]}[]} chatHistory
|
|
* @param {function} formatterFunction - The function to format the chat history from the llm provider
|
|
* @param {('asProperty'|'spread')} mode - "asProperty" or "spread". Determines how the content is formatted in the message object.
|
|
* @returns {object[]}
|
|
*/
|
|
function formatChatHistory(
|
|
chatHistory = [],
|
|
formatterFunction,
|
|
mode = "asProperty"
|
|
) {
|
|
return chatHistory.map((historicalMessage) => {
|
|
if (
|
|
historicalMessage?.role !== "user" || // Only user messages can have attachments
|
|
!historicalMessage?.attachments || // If there are no attachments, we can skip this
|
|
!historicalMessage.attachments.length // If there is an array but it is empty, we can skip this
|
|
)
|
|
return historicalMessage;
|
|
|
|
// Some providers, like Ollama, expect the content to be embedded in the message object.
|
|
if (mode === "spread") {
|
|
return {
|
|
role: historicalMessage.role,
|
|
...formatterFunction({
|
|
userPrompt: historicalMessage.content,
|
|
attachments: historicalMessage.attachments,
|
|
}),
|
|
};
|
|
}
|
|
|
|
// Most providers expect the content to be a property of the message object formatted like OpenAI models.
|
|
return {
|
|
role: historicalMessage.role,
|
|
content: formatterFunction({
|
|
userPrompt: historicalMessage.content,
|
|
attachments: historicalMessage.attachments,
|
|
}),
|
|
};
|
|
});
|
|
}
|
|
|
|
module.exports = {
|
|
handleDefaultStreamResponseV2,
|
|
convertToChatHistory,
|
|
convertToPromptHistory,
|
|
writeResponseChunk,
|
|
clientAbortedHandler,
|
|
formatChatHistory,
|
|
safeJSONStringify,
|
|
};
|