mirror of
https://github.com/Nezumi-2711/9router.git
synced 2026-09-22 13:38:31 +00:00
feat(request-details): implement observability settings and enhance request detail tracking
- Added new observability settings in the dashboard for max records, batch size, flush interval, and max JSON size. - Introduced `extractRequestConfig` function to capture full request configurations. - Enhanced error handling by saving detailed request information on failures. - Updated usage tracking to include new token metrics. - Modified streaming functions to support detailed content and reasoning tracking.
This commit is contained in:
decolua
+51
-22
@@ -28,6 +28,7 @@ const STREAM_MODE = {
|
||||
* @param {string} options.model - Model name
|
||||
* @param {string} options.connectionId - Connection ID for usage tracking
|
||||
* @param {object} options.body - Request body (for input token estimation)
|
||||
* @param {function} options.onStreamComplete - Callback when stream completes (content, usage)
|
||||
*/
|
||||
export function createSSEStream(options = {}) {
|
||||
const {
|
||||
@@ -39,20 +40,25 @@ export function createSSEStream(options = {}) {
|
||||
toolNameMap = null,
|
||||
model = null,
|
||||
connectionId = null,
|
||||
body = null
|
||||
body = null,
|
||||
onStreamComplete = null
|
||||
} = options;
|
||||
|
||||
let buffer = "";
|
||||
let usage = null;
|
||||
|
||||
// State for translate mode
|
||||
const state = mode === STREAM_MODE.TRANSLATE ? { ...initState(sourceFormat), provider, toolNameMap } : null;
|
||||
|
||||
// Track content length for usage estimation (both modes)
|
||||
let totalContentLength = 0;
|
||||
let accumulatedContent = "";
|
||||
let accumulatedThinking = "";
|
||||
let ttftAt = null;
|
||||
|
||||
return new TransformStream({
|
||||
transform(chunk, controller) {
|
||||
if (!ttftAt) {
|
||||
ttftAt = Date.now();
|
||||
}
|
||||
const text = sharedDecoder.decode(chunk, { stream: true });
|
||||
buffer += text;
|
||||
reqLogger?.appendProviderChunk?.(text);
|
||||
@@ -79,9 +85,15 @@ export function createSSEStream(options = {}) {
|
||||
}
|
||||
|
||||
const delta = parsed.choices?.[0]?.delta;
|
||||
const content = delta?.content || delta?.reasoning_content;
|
||||
const content = delta?.content;
|
||||
const reasoning = delta?.reasoning_content;
|
||||
if (content && typeof content === "string") {
|
||||
totalContentLength += content.length;
|
||||
accumulatedContent += content;
|
||||
}
|
||||
if (reasoning && typeof reasoning === "string") {
|
||||
totalContentLength += reasoning.length;
|
||||
accumulatedThinking += reasoning;
|
||||
}
|
||||
|
||||
const extracted = extractUsage(parsed);
|
||||
@@ -134,30 +146,39 @@ export function createSSEStream(options = {}) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Track content length for estimation (from various formats)
|
||||
// Include both regular content and reasoning/thinking content
|
||||
|
||||
// Claude format
|
||||
// Claude format - content
|
||||
if (parsed.delta?.text) {
|
||||
totalContentLength += parsed.delta.text.length;
|
||||
accumulatedContent += parsed.delta.text;
|
||||
}
|
||||
// Claude format - thinking
|
||||
if (parsed.delta?.thinking) {
|
||||
totalContentLength += parsed.delta.thinking.length;
|
||||
accumulatedThinking += parsed.delta.thinking;
|
||||
}
|
||||
|
||||
// OpenAI format
|
||||
// OpenAI format - content
|
||||
if (parsed.choices?.[0]?.delta?.content) {
|
||||
totalContentLength += parsed.choices[0].delta.content.length;
|
||||
accumulatedContent += parsed.choices[0].delta.content;
|
||||
}
|
||||
// OpenAI format - reasoning
|
||||
if (parsed.choices?.[0]?.delta?.reasoning_content) {
|
||||
totalContentLength += parsed.choices[0].delta.reasoning_content.length;
|
||||
accumulatedThinking += parsed.choices[0].delta.reasoning_content;
|
||||
}
|
||||
|
||||
// Gemini format - may have multiple parts
|
||||
// Gemini format
|
||||
if (parsed.candidates?.[0]?.content?.parts) {
|
||||
for (const part of parsed.candidates[0].content.parts) {
|
||||
if (part.text && typeof part.text === "string") {
|
||||
totalContentLength += part.text.length;
|
||||
// Check if this is thinking content
|
||||
if (part.thought === true) {
|
||||
accumulatedThinking += part.text;
|
||||
} else {
|
||||
accumulatedContent += part.text;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -220,7 +241,6 @@ export function createSSEStream(options = {}) {
|
||||
controller.enqueue(sharedEncoder.encode(output));
|
||||
}
|
||||
|
||||
// Estimate usage if provider didn't return valid usage (PASSTHROUGH is always OpenAI format)
|
||||
if (!hasValidUsage(usage) && totalContentLength > 0) {
|
||||
usage = estimateUsage(body, totalContentLength, FORMATS.OPENAI);
|
||||
}
|
||||
@@ -230,16 +250,21 @@ export function createSSEStream(options = {}) {
|
||||
} else {
|
||||
appendRequestLog({ model, provider, connectionId, tokens: null, status: "200 OK" }).catch(() => { });
|
||||
}
|
||||
|
||||
if (onStreamComplete) {
|
||||
onStreamComplete({
|
||||
content: accumulatedContent,
|
||||
thinking: accumulatedThinking
|
||||
}, usage, ttftAt);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// Translate mode: process remaining buffer
|
||||
if (buffer.trim()) {
|
||||
const parsed = parseSSELine(buffer.trim());
|
||||
if (parsed && !parsed.done) {
|
||||
const translated = translateResponse(targetFormat, sourceFormat, parsed, state);
|
||||
|
||||
// Log OpenAI intermediate chunks
|
||||
if (translated?._openaiIntermediate) {
|
||||
for (const item of translated._openaiIntermediate) {
|
||||
const openaiOutput = formatSSE(item, FORMATS.OPENAI);
|
||||
@@ -257,10 +282,8 @@ export function createSSEStream(options = {}) {
|
||||
}
|
||||
}
|
||||
|
||||
// Flush remaining events (only once at stream end)
|
||||
const flushed = translateResponse(targetFormat, sourceFormat, null, state);
|
||||
|
||||
// Log OpenAI intermediate chunks for flushed events
|
||||
if (flushed?._openaiIntermediate) {
|
||||
for (const item of flushed._openaiIntermediate) {
|
||||
const openaiOutput = formatSSE(item, FORMATS.OPENAI);
|
||||
@@ -276,12 +299,10 @@ export function createSSEStream(options = {}) {
|
||||
}
|
||||
}
|
||||
|
||||
// Send [DONE] and log usage
|
||||
const doneOutput = "data: [DONE]\n\n";
|
||||
reqLogger?.appendConvertedChunk?.(doneOutput);
|
||||
controller.enqueue(sharedEncoder.encode(doneOutput));
|
||||
|
||||
// Estimate usage if provider didn't return valid usage (for translate mode)
|
||||
if (!hasValidUsage(state?.usage) && totalContentLength > 0) {
|
||||
state.usage = estimateUsage(body, totalContentLength, sourceFormat);
|
||||
}
|
||||
@@ -291,6 +312,13 @@ export function createSSEStream(options = {}) {
|
||||
} else {
|
||||
appendRequestLog({ model, provider, connectionId, tokens: null, status: "200 OK" }).catch(() => { });
|
||||
}
|
||||
|
||||
if (onStreamComplete) {
|
||||
onStreamComplete({
|
||||
content: accumulatedContent,
|
||||
thinking: accumulatedThinking
|
||||
}, state?.usage, ttftAt);
|
||||
}
|
||||
} catch (error) {
|
||||
console.log("Error in flush:", error);
|
||||
}
|
||||
@@ -298,8 +326,7 @@ export function createSSEStream(options = {}) {
|
||||
});
|
||||
}
|
||||
|
||||
// Convenience functions for backward compatibility
|
||||
export function createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider = null, reqLogger = null, toolNameMap = null, model = null, connectionId = null, body = null) {
|
||||
export function createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider = null, reqLogger = null, toolNameMap = null, model = null, connectionId = null, body = null, onStreamComplete = null) {
|
||||
return createSSEStream({
|
||||
mode: STREAM_MODE.TRANSLATE,
|
||||
targetFormat,
|
||||
@@ -309,17 +336,19 @@ export function createSSETransformStreamWithLogger(targetFormat, sourceFormat, p
|
||||
toolNameMap,
|
||||
model,
|
||||
connectionId,
|
||||
body
|
||||
body,
|
||||
onStreamComplete
|
||||
});
|
||||
}
|
||||
|
||||
export function createPassthroughStreamWithLogger(provider = null, reqLogger = null, model = null, connectionId = null, body = null) {
|
||||
export function createPassthroughStreamWithLogger(provider = null, reqLogger = null, model = null, connectionId = null, body = null, onStreamComplete = null) {
|
||||
return createSSEStream({
|
||||
mode: STREAM_MODE.PASSTHROUGH,
|
||||
provider,
|
||||
reqLogger,
|
||||
model,
|
||||
connectionId,
|
||||
body
|
||||
body,
|
||||
onStreamComplete
|
||||
});
|
||||
}
|
||||
|
||||
@@ -312,11 +312,11 @@ export function logUsage(provider, usage, model = null, connectionId = null) {
|
||||
|
||||
// Save to usage DB
|
||||
const tokens = {
|
||||
input: inTokens,
|
||||
output: outTokens,
|
||||
cacheRead: cacheRead || 0,
|
||||
cacheCreation: cacheCreation || 0,
|
||||
reasoning: reasoning || 0
|
||||
prompt_tokens: inTokens,
|
||||
completion_tokens: outTokens,
|
||||
cache_read_input_tokens: cacheRead || 0,
|
||||
cache_creation_input_tokens: cacheCreation || 0,
|
||||
reasoning_tokens: reasoning || 0
|
||||
};
|
||||
saveRequestUsage({ model, provider, connectionId, tokens }).catch(() => { });
|
||||
appendRequestLog({ model, provider, connectionId, tokens, status: "200 OK" }).catch(() => { });
|
||||
|
||||
Reference in New Issue
Block a user