Revert "feat(request-details): implement observability settings and enhance request detail tracking"
This reverts commit cbabf5547c.
This commit is contained in:
decolua
@@ -28,7 +28,6 @@ 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 {
|
||||
@@ -40,25 +39,20 @@ export function createSSEStream(options = {}) {
|
||||
toolNameMap = null,
|
||||
model = null,
|
||||
connectionId = null,
|
||||
body = null,
|
||||
onStreamComplete = null
|
||||
body = 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);
|
||||
@@ -85,15 +79,9 @@ export function createSSEStream(options = {}) {
|
||||
}
|
||||
|
||||
const delta = parsed.choices?.[0]?.delta;
|
||||
const content = delta?.content;
|
||||
const reasoning = delta?.reasoning_content;
|
||||
const content = delta?.content || 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);
|
||||
@@ -146,39 +134,30 @@ export function createSSEStream(options = {}) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Claude format - content
|
||||
// Track content length for estimation (from various formats)
|
||||
// Include both regular content and reasoning/thinking content
|
||||
|
||||
// Claude format
|
||||
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 - content
|
||||
// OpenAI format
|
||||
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
|
||||
// Gemini format - may have multiple parts
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -241,6 +220,7 @@ 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);
|
||||
}
|
||||
@@ -250,21 +230,16 @@ 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);
|
||||
@@ -282,8 +257,10 @@ 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);
|
||||
@@ -299,10 +276,12 @@ 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);
|
||||
}
|
||||
@@ -312,13 +291,6 @@ 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);
|
||||
}
|
||||
@@ -326,7 +298,8 @@ export function createSSEStream(options = {}) {
|
||||
});
|
||||
}
|
||||
|
||||
export function createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider = null, reqLogger = null, toolNameMap = null, model = null, connectionId = null, body = null, onStreamComplete = null) {
|
||||
// Convenience functions for backward compatibility
|
||||
export function createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider = null, reqLogger = null, toolNameMap = null, model = null, connectionId = null, body = null) {
|
||||
return createSSEStream({
|
||||
mode: STREAM_MODE.TRANSLATE,
|
||||
targetFormat,
|
||||
@@ -336,19 +309,17 @@ export function createSSETransformStreamWithLogger(targetFormat, sourceFormat, p
|
||||
toolNameMap,
|
||||
model,
|
||||
connectionId,
|
||||
body,
|
||||
onStreamComplete
|
||||
body
|
||||
});
|
||||
}
|
||||
|
||||
export function createPassthroughStreamWithLogger(provider = null, reqLogger = null, model = null, connectionId = null, body = null, onStreamComplete = null) {
|
||||
export function createPassthroughStreamWithLogger(provider = null, reqLogger = null, model = null, connectionId = null, body = null) {
|
||||
return createSSEStream({
|
||||
mode: STREAM_MODE.PASSTHROUGH,
|
||||
provider,
|
||||
reqLogger,
|
||||
model,
|
||||
connectionId,
|
||||
body,
|
||||
onStreamComplete
|
||||
body
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user