mirror of
https://github.com/Nezumi-2711/9router.git
synced 2026-09-22 13:38:31 +00:00
Feature/ai observability dashboard (#79)
* feat: add AI request details feature with latency tracking
Add comprehensive request history and debugging capability to the Usage dashboard:
**Storage Layer** (usageDb.js):
- Add saveRequestDetail() for storing full request/response details
- Implement FIFO queue with 1000-record limit in request-details.json
- Auto-sanitize sensitive headers (authorization, api-key, cookie, token)
- Add getRequestDetails() with pagination and filtering support
- Add getRequestDetailById() for single record lookup
**Pipeline Integration** (chatCore.js):
- Track request start time and calculate total latency
- Record TTFT (Time To First Token) and total latency for all requests
- Capture full request details (messages, model, parameters)
- Save response content for non-streaming, mark streaming responses
- Handle error cases with detailed error information
- Async non-blocking saves to avoid impacting request performance
**API Layer** (/api/usage/request-details):
- GET endpoint with pagination (page, pageSize: 1-100)
- Filter by provider, model, connectionId, status, date range
- Returns { details: [...], pagination: {...} } format
**UI Components**:
- Drawer.js: Right slide-out panel with backdrop blur and ESC close
- Pagination.js: Full pagination with page size selector (10/20/50)
- RequestDetailsTab.js: Complete table view with filters and detail drawer
**Dashboard Integration**:
- Add "Details" tab to Usage page (4th tab after Overview/Logger/Limits)
- Table columns: Timestamp, Model, Provider, Input Tokens, Output Tokens, Latency (TTFT/Total), Action
- Provider filter dropdown (9 providers supported)
- Date range filters (start/end datetime)
- Click "Detail" button to view full request/response JSON in slide-out drawer
**Features**:
- Real-time latency monitoring (TTFT & Total)
- Complete request/response inspection for debugging
- Filterable and searchable request history
- Responsive design with mobile-friendly filters
- Data security with automatic header sanitization
- Performance: async saves don't block request pipeline
**Files Created/Modified**:
- src/lib/usageDb.js (modified)
- open-sse/handlers/chatCore.js (modified)
- src/app/api/usage/request-details/route.js (new)
- src/shared/components/Drawer.js (new)
- src/shared/components/Pagination.js (new)
- src/app/(dashboard)/dashboard/usage/components/RequestDetailsTab.js (new)
- src/app/(dashboard)/dashboard/usage/page.js (modified)
Closes: AI Observability Dashboard feature
* feat: enhance request details with full config and streaming content capture
Improve Request Details feature to capture comprehensive request parameters
and actual streaming response content:
**Request Configuration Enhancement** (chatCore.js):
- Add extractRequestConfig() helper function to capture all request parameters
- Include temperature controls: temperature, top_p, top_k
- Include token limits: max_tokens, max_completion_tokens
- Include thinking/reasoning modes: thinking, reasoning, enable_thinking
- Include OpenAI parameters: presence_penalty, frequency_penalty, seed, stop,
tools, tool_choice, response_format, n, logprobs, top_logprobs, logit_bias,
user, parallel_tool_calls, prediction, store, metadata
- Apply to all request types: non-streaming, streaming, and error cases
**Streaming Content Capture** (chatCore.js & stream.js):
- Add onStreamComplete callback mechanism to stream processors
- Accumulate content from all formats: OpenAI, Claude, Gemini
- Track content from delta.content, delta.reasoning_content, delta.text,
delta.thinking, and Gemini content.parts
- Save initial record with "[Streaming in progress...]" marker
- Update record with actual content when stream completes
- Include usage tokens when available from stream
**Files Modified**:
- open-sse/handlers/chatCore.js - extractRequestConfig() + streaming capture
- open-sse/utils/stream.js - onStreamComplete callback + content accumulation
**Benefits**:
- View complete request configuration in Request Details (thinking mode, etc.)
- See actual streaming response content instead of placeholder
- Better debugging and observability for AI requests
Refs: #request-details-enhancement
* feat: separate thinking/reasoning content from response content
Improve Request Details to display thinking process separately from final response:
**Backend Changes**:
- stream.js: Capture content and thinking separately in streaming mode
- Add accumulatedThinking variable alongside accumulatedContent
- Route delta.content to content, delta.reasoning_content to thinking
- Support OpenAI (reasoning_content), Claude (thinking), Gemini (part.thought)
- Update onStreamComplete callback to return { content, thinking } object
- chatCore.js: Update response structure to include thinking field
- Non-streaming: Extract thinking from reasoning_content field
- Streaming: Receive { content, thinking } from stream callback
- Error responses: Include thinking: null
- Initial streaming save: Include thinking: null
**Frontend Changes**:
- RequestDetailsTab.js: Display thinking and content in separate sections
- Add amber/yellow themed "Thinking Process" section with psychology icon
- Show "Final Response" label when thinking is present
- Use distinct visual styling for thinking (amber bg) vs content (gray bg)
- Only show thinking section when thinking content exists
**Benefits**:
- Users can clearly see model's reasoning process vs final answer
- Better debugging for models with thinking capabilities (Claude, o1, etc.)
- Visual distinction makes it easy to identify thinking vs response
Refs: #thinking-content-separation
* fix: map Claude thinking to reasoning_content field
Fix Claude thinking content to be properly captured as reasoning_content
instead of regular content, enabling separate display in Request Details:
**Changes**:
- claude-to-openai.js: Use reasoning_content field for thinking blocks
- thinking start: send { reasoning_content: "" } instead of { content: "```\n```" }
- thinking delta: map to reasoning_content instead of content
- thinking stop: send { reasoning_content: "" } instead of { content: "```\n```" }
**Why This Matters**:
- Previously Claude thinking was sent as `content` field, mixed with actual response
- Now thinking uses `reasoning_content` field, matching OpenAI's o1 format
- stream.js can now properly route thinking to accumulatedThinking variable
- Request Details UI will show Claude thinking in separate "Thinking Process" section
**Supported Thinking Formats**:
- OpenAI: delta.reasoning_content → thinking
- Claude: delta.thinking → reasoning_content (now fixed)
- Gemini: part.thought === true → thinking
Refs: #claude-thinking-fix
* feat(observability): capture and display full 4-layer request chain
Capture complete request/response chain in AI Request Details:
- Add providerRequest field (translated request sent to provider)
- Add providerResponse field (raw provider response, streaming indicator)
- Update chatCore.js at all 5 saveRequestDetail() call sites
- Reorganize UI into 4 collapsible sections with Material icons
- Preserve backward compatibility for old records
- Add distinct styling for streaming indicator
* fix(observability): resolve React duplicate key warning in request details table
- Use composite key (detail.id + index) to ensure unique keys
- Prevents React warnings when database contains duplicate IDs from old ID generation
* fix(observability): display actual content in streaming request details
Change providerResponse field for streaming requests from placeholder
"[Streaming - raw response not captured]" to actual final content.
This improves debugging experience by showing the real AI response
in the "Provider Response (Raw)" section instead of a confusing
placeholder message.
Files changed:
- open-sse/handlers/chatCore.js: Save contentObj.content to providerResponse
- src/app/.../RequestDetailsTab.js: Remove special handling for placeholder
* refactor(observability): migrate request details to SQLite for improved concurrency
- Replace LowDB JSON storage with better-sqlite3
- Enable WAL mode for true concurrent read/write support
- Add 5 indexes to accelerate queries (timestamp, provider, model, connection_id, status)
- Perform pagination at the database level to reduce memory footprint
- Maintain 1000 record limit with automatic cleanup of old data
- Ensure API compatibility via re-exports, requiring no caller changes
Performance improvements:
- Concurrent Writes: Lock-free WAL mode prevents data contention
- Query Efficiency: Index-based searches replace full dataset loading
- Data Integrity: Atomic operations prevent file corruption
* fix(observability): resolve pagination statistics display issues
- Fix issue where totalItems=0 showed 'Showing 1 to 0 of 0 results'
- Hide pagination controls when totalItems=0 or totalPages<=1
- Standardize API response fields: pagination.total -> pagination.totalItems
Before: Incorrect stats shown for empty data, and pager visible even for single-page results
After: Stats hidden for empty data, pager hidden when navigation is unnecessary
* feat(observability): display friendly provider names in request details
- Add /api/usage/providers endpoint to dynamically fetch provider list with names
- Replace hardcoded provider options with dynamic loading from database
- Display friendly provider names instead of IDs in both table and detail drawer
- Support custom provider nodes (e.g., OpenAI-compatible) with user-defined names
- Add provider name caching to optimize performance
* fix(observability): use INSERT OR REPLACE for request details to handle streaming updates
* fix(observability): resolve zero-token display issue by ensuring streaming usage capture and fixing key mismatch
* fix(observability): separate TTFT and total latency calculation for streaming requests
* feat(observability): implement SQLite write queue and JSON size limits
- Added in-memory buffer and batch writing for SQLite to prevent lock contention
- Implemented with configurable 1MB limit to prevent DB bloat
- Added dashboard UI for observability performance and data management settings
- Integrated graceful shutdown handlers to prevent data loss
* fix(observability): resolve ReferenceError by declaring dbInstance
This commit is contained in:
Blade
+186
-10
@@ -10,7 +10,7 @@ import { getModelTargetFormat, PROVIDER_ID_TO_ALIAS } from "../config/providerMo
|
||||
import { createErrorResult, parseUpstreamError, formatProviderError } from "../utils/error.js";
|
||||
import { HTTP_STATUS } from "../config/constants.js";
|
||||
import { handleBypassRequest } from "../utils/bypassHandler.js";
|
||||
import { saveRequestUsage, trackPendingRequest, appendRequestLog } from "@/lib/usageDb.js";
|
||||
import { saveRequestUsage, trackPendingRequest, appendRequestLog, saveRequestDetail } from "@/lib/usageDb.js";
|
||||
import { getExecutor } from "../executors/index.js";
|
||||
|
||||
/**
|
||||
@@ -225,6 +225,38 @@ function extractUsageFromResponse(responseBody, provider) {
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract full request configuration from body
|
||||
* Captures all relevant parameters for request details
|
||||
*/
|
||||
function extractRequestConfig(body, stream) {
|
||||
const config = {
|
||||
messages: body.messages || [],
|
||||
model: body.model,
|
||||
stream: stream
|
||||
};
|
||||
|
||||
// Add all optional configuration parameters
|
||||
const optionalParams = [
|
||||
'temperature', 'top_p', 'top_k',
|
||||
'max_tokens', 'max_completion_tokens',
|
||||
'thinking', 'reasoning', 'enable_thinking',
|
||||
'presence_penalty', 'frequency_penalty',
|
||||
'seed', 'stop', 'tools', 'tool_choice',
|
||||
'response_format', 'prediction', 'store', 'metadata',
|
||||
'n', 'logprobs', 'top_logprobs', 'logit_bias',
|
||||
'user', 'parallel_tool_calls'
|
||||
];
|
||||
|
||||
for (const param of optionalParams) {
|
||||
if (body[param] !== undefined) {
|
||||
config[param] = body[param];
|
||||
}
|
||||
}
|
||||
|
||||
return config;
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert OpenAI-style SSE chunks into a single non-streaming JSON response.
|
||||
* Used as a fallback when upstream returns text/event-stream for stream=false.
|
||||
@@ -315,6 +347,7 @@ function parseSSEToOpenAIResponse(rawSSE, fallbackModel) {
|
||||
*/
|
||||
export async function handleChatCore({ body, modelInfo, credentials, log, onCredentialsRefreshed, onRequestSuccess, onDisconnect, clientRawRequest, connectionId, userAgent }) {
|
||||
const { provider, model } = modelInfo;
|
||||
const requestStartTime = Date.now();
|
||||
|
||||
const sourceFormat = detectFormat(body);
|
||||
|
||||
@@ -407,6 +440,26 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
|
||||
} catch (error) {
|
||||
trackPendingRequest(model, provider, connectionId, false);
|
||||
appendRequestLog({ model, provider, connectionId, status: `FAILED ${error.name === "AbortError" ? 499 : HTTP_STATUS.BAD_GATEWAY}` }).catch(() => { });
|
||||
|
||||
const errorDetail = {
|
||||
provider: provider || "unknown",
|
||||
model: model || "unknown",
|
||||
connectionId: connectionId || undefined,
|
||||
timestamp: new Date().toISOString(),
|
||||
latency: { ttft: 0, total: Date.now() - requestStartTime },
|
||||
tokens: { prompt_tokens: 0, completion_tokens: 0 },
|
||||
request: extractRequestConfig(body, stream),
|
||||
providerRequest: translatedBody || null,
|
||||
providerResponse: null,
|
||||
response: {
|
||||
error: error.message || String(error),
|
||||
status: error.name === "AbortError" ? 499 : 502,
|
||||
thinking: null
|
||||
},
|
||||
status: "error"
|
||||
};
|
||||
saveRequestDetail(errorDetail).catch(() => {});
|
||||
|
||||
if (error.name === "AbortError") {
|
||||
streamController.handleError(error);
|
||||
return createErrorResult(499, "Request aborted");
|
||||
@@ -463,6 +516,26 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
|
||||
trackPendingRequest(model, provider, connectionId, false);
|
||||
const { statusCode, message, retryAfterMs } = await parseUpstreamError(providerResponse, provider);
|
||||
appendRequestLog({ model, provider, connectionId, status: `FAILED ${statusCode}` }).catch(() => { });
|
||||
|
||||
const errorDetail = {
|
||||
provider: provider || "unknown",
|
||||
model: model || "unknown",
|
||||
connectionId: connectionId || undefined,
|
||||
timestamp: new Date().toISOString(),
|
||||
latency: { ttft: 0, total: Date.now() - requestStartTime },
|
||||
tokens: { prompt_tokens: 0, completion_tokens: 0 },
|
||||
request: extractRequestConfig(body, stream),
|
||||
providerRequest: finalBody || translatedBody || null,
|
||||
providerResponse: null,
|
||||
response: {
|
||||
error: message,
|
||||
status: statusCode,
|
||||
thinking: null
|
||||
},
|
||||
status: "error"
|
||||
};
|
||||
saveRequestDetail(errorDetail).catch(() => {});
|
||||
|
||||
const errMsg = formatProviderError(new Error(message), provider, model, statusCode);
|
||||
console.log(`${COLORS.red}[ERROR] ${errMsg}${COLORS.reset}`);
|
||||
|
||||
@@ -531,6 +604,37 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
|
||||
translatedResponse.usage = filterUsageForFormat(buffered, sourceFormat);
|
||||
}
|
||||
|
||||
const totalLatency = Date.now() - requestStartTime;
|
||||
const requestDetail = {
|
||||
provider: provider || "unknown",
|
||||
model: model || "unknown",
|
||||
connectionId: connectionId || undefined,
|
||||
timestamp: new Date().toISOString(),
|
||||
latency: {
|
||||
ttft: totalLatency,
|
||||
total: totalLatency
|
||||
},
|
||||
tokens: usage || { prompt_tokens: 0, completion_tokens: 0 },
|
||||
request: extractRequestConfig(body, stream),
|
||||
providerRequest: finalBody || translatedBody || null,
|
||||
providerResponse: responseBody || null,
|
||||
response: {
|
||||
content: translatedResponse?.choices?.[0]?.message?.content ||
|
||||
translatedResponse?.content ||
|
||||
null,
|
||||
thinking: translatedResponse?.choices?.[0]?.message?.reasoning_content ||
|
||||
translatedResponse?.reasoning_content ||
|
||||
null,
|
||||
finish_reason: translatedResponse?.choices?.[0]?.finish_reason || "unknown"
|
||||
},
|
||||
status: "success"
|
||||
};
|
||||
|
||||
// Async save (don't block response)
|
||||
saveRequestDetail(requestDetail).catch(err => {
|
||||
console.error("[RequestDetail] Failed to save:", err.message);
|
||||
});
|
||||
|
||||
return {
|
||||
success: true,
|
||||
response: new Response(JSON.stringify(translatedResponse), {
|
||||
@@ -556,31 +660,103 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
|
||||
"Access-Control-Allow-Origin": "*"
|
||||
};
|
||||
|
||||
// Create transform stream with logger for streaming response
|
||||
let streamContent = "";
|
||||
let streamUsage = null;
|
||||
const streamDetailId = `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`;
|
||||
|
||||
const onStreamComplete = (contentObj, usage, ttftAt) => {
|
||||
// contentObj is object { content, thinking }
|
||||
streamUsage = usage;
|
||||
|
||||
const updatedDetail = {
|
||||
provider: provider || "unknown",
|
||||
model: model || "unknown",
|
||||
connectionId: connectionId || undefined,
|
||||
timestamp: new Date().toISOString(),
|
||||
latency: {
|
||||
ttft: ttftAt ? ttftAt - requestStartTime : Date.now() - requestStartTime,
|
||||
total: Date.now() - requestStartTime
|
||||
},
|
||||
tokens: usage || { prompt_tokens: 0, completion_tokens: 0 },
|
||||
request: extractRequestConfig(body, stream),
|
||||
providerRequest: finalBody || translatedBody || null,
|
||||
providerResponse: contentObj.content || "[Empty streaming response]",
|
||||
response: {
|
||||
content: contentObj.content || "[Empty streaming response]",
|
||||
thinking: contentObj.thinking || null,
|
||||
type: "streaming"
|
||||
},
|
||||
status: "success",
|
||||
id: streamDetailId
|
||||
};
|
||||
|
||||
saveRequestDetail(updatedDetail).catch(err => {
|
||||
console.error("[RequestDetail] Failed to update streaming content:", err.message);
|
||||
});
|
||||
|
||||
// Save usage stats for dashboard
|
||||
if (usage && typeof usage === 'object') {
|
||||
const msg = `[${new Date().toLocaleTimeString("en-US", { hour12: false, hour: "2-digit", minute: "2-digit" })}] 📊 [STREAM USAGE] ${provider.toUpperCase()} | in=${usage?.prompt_tokens || 0} | out=${usage?.completion_tokens || 0}${connectionId ? ` | account=${connectionId.slice(0, 8)}...` : ""}`;
|
||||
console.log(`${COLORS.green}${msg}${COLORS.reset}`);
|
||||
|
||||
saveRequestUsage({
|
||||
provider: provider || "unknown",
|
||||
model: model || "unknown",
|
||||
tokens: usage,
|
||||
timestamp: new Date().toISOString(),
|
||||
connectionId: connectionId || undefined
|
||||
}).catch(err => {
|
||||
console.error("Failed to save streaming usage stats:", err.message);
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
let transformStream;
|
||||
// For Codex provider, translate response from openai-responses to openai (Chat Completions) format
|
||||
// UNLESS client is Droid CLI which expects openai-responses format back
|
||||
const isDroidCLI = userAgent?.toLowerCase().includes('droid') || userAgent?.toLowerCase().includes('codex-cli');
|
||||
const needsCodexTranslation = provider === 'codex'
|
||||
&& targetFormat === 'openai-responses'
|
||||
&& !isDroidCLI;
|
||||
|
||||
if (needsCodexTranslation) {
|
||||
// Codex returns openai-responses, translate to openai (Chat Completions) that clients expect
|
||||
log?.debug?.("STREAM", `Codex translation mode: openai-responses → openai`);
|
||||
transformStream = createSSETransformStreamWithLogger('openai-responses', 'openai', provider, reqLogger, toolNameMap, model, connectionId, body);
|
||||
transformStream = createSSETransformStreamWithLogger('openai-responses', 'openai', provider, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete);
|
||||
} else if (needsTranslation(targetFormat, sourceFormat)) {
|
||||
// Standard translation for other providers
|
||||
log?.debug?.("STREAM", `Translation mode: ${targetFormat} → ${sourceFormat}`);
|
||||
transformStream = createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider, reqLogger, toolNameMap, model, connectionId, body);
|
||||
transformStream = createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete);
|
||||
} else {
|
||||
log?.debug?.("STREAM", `Standard passthrough mode`);
|
||||
transformStream = createPassthroughStreamWithLogger(provider, reqLogger, model, connectionId, body);
|
||||
transformStream = createPassthroughStreamWithLogger(provider, reqLogger, model, connectionId, body, onStreamComplete);
|
||||
}
|
||||
|
||||
// Pipe response through transform with disconnect detection
|
||||
const transformedBody = pipeWithDisconnect(providerResponse, transformStream, streamController);
|
||||
|
||||
const totalLatency = Date.now() - requestStartTime;
|
||||
const streamingDetail = {
|
||||
provider: provider || "unknown",
|
||||
model: model || "unknown",
|
||||
connectionId: connectionId || undefined,
|
||||
timestamp: new Date().toISOString(),
|
||||
latency: {
|
||||
ttft: 0,
|
||||
total: Date.now() - requestStartTime
|
||||
},
|
||||
tokens: { prompt_tokens: 0, completion_tokens: 0 },
|
||||
request: extractRequestConfig(body, stream),
|
||||
providerRequest: finalBody || translatedBody || null,
|
||||
providerResponse: "[Streaming - raw response not captured]",
|
||||
response: {
|
||||
content: "[Streaming in progress...]",
|
||||
thinking: null,
|
||||
type: "streaming"
|
||||
},
|
||||
status: "success",
|
||||
id: streamDetailId
|
||||
};
|
||||
|
||||
saveRequestDetail(streamingDetail).catch(err => {
|
||||
console.error("[RequestDetail] Failed to save streaming request:", err.message);
|
||||
});
|
||||
|
||||
return {
|
||||
success: true,
|
||||
response: new Response(transformedBody, {
|
||||
|
||||
@@ -64,7 +64,7 @@ export function claudeToOpenAIResponse(chunk, state) {
|
||||
if (delta?.type === "text_delta" && delta.text) {
|
||||
results.push(createChunk(state, { content: delta.text }));
|
||||
} else if (delta?.type === "thinking_delta" && delta.thinking) {
|
||||
results.push(createChunk(state, { content: delta.thinking }));
|
||||
results.push(createChunk(state, { reasoning_content: delta.thinking }));
|
||||
} else if (delta?.type === "input_json_delta" && delta.partial_json) {
|
||||
const toolCall = state.toolCalls.get(chunk.index);
|
||||
if (toolCall) {
|
||||
@@ -83,7 +83,7 @@ export function claudeToOpenAIResponse(chunk, state) {
|
||||
|
||||
case "content_block_stop": {
|
||||
if (state.inThinkingBlock && chunk.index === state.currentBlockIndex) {
|
||||
results.push(createChunk(state, { content: "</think>" }));
|
||||
results.push(createChunk(state, { reasoning_content: "" }));
|
||||
state.inThinkingBlock = false;
|
||||
}
|
||||
state.textBlockStarted = false;
|
||||
|
||||
+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