wehub-resource-sync
128 lines
4.7 KiB
TypeScript
128 lines
4.7 KiB
TypeScript
/**
|
|
* #3089 — Convert a complete OpenAI-style chat-completion JSON body into an
|
|
* equivalent OpenAI SSE (`chat.completion.chunk`) stream.
|
|
*
|
|
* Some "reasoning" openai-compatible upstreams ignore a `stream: true` request
|
|
* and reply with a single `application/json` chat-completion body instead of an
|
|
* SSE stream. OmniRoute's streaming readiness check only recognizes SSE `data:`
|
|
* frames, so such a body produced a spurious `STREAM_EARLY_EOF` / HTTP 502 even
|
|
* though it carried valid `content` / `reasoning_content`. Synthesizing an SSE
|
|
* stream from that JSON lets the normal streaming pipeline (and the client) get
|
|
* a valid stream that preserves both `content` and `reasoning_content`.
|
|
*
|
|
* Returns "" when the text is not a parseable chat-completion object with at
|
|
* least one choice — callers then fall back to the original (error) handling.
|
|
*/
|
|
import { normalizeOpenAICompatibleFinishReasonString } from "./finishReason.ts";
|
|
import { getUnsupportedReasoningValue } from "./reasoningFields.ts";
|
|
|
|
type JsonRecord = Record<string, unknown>;
|
|
|
|
function isRecord(value: unknown): value is JsonRecord {
|
|
return typeof value === "object" && value !== null && !Array.isArray(value);
|
|
}
|
|
|
|
function nonEmptyString(value: unknown): string {
|
|
return typeof value === "string" && value.length > 0 ? value : "";
|
|
}
|
|
|
|
function addReadableReasoning(message: JsonRecord, delta: JsonRecord): boolean {
|
|
const reasoningContent = nonEmptyString(message.reasoning_content);
|
|
if (reasoningContent) {
|
|
delta.reasoning_content = reasoningContent;
|
|
return true;
|
|
}
|
|
|
|
const reasoning = nonEmptyString(message.reasoning);
|
|
if (reasoning) {
|
|
delta.reasoning = reasoning;
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
function addUnsupportedReasoning(message: JsonRecord, delta: JsonRecord) {
|
|
const reasoningContent = getUnsupportedReasoningValue(message);
|
|
if (reasoningContent) {
|
|
delta.reasoning_content = reasoningContent;
|
|
}
|
|
}
|
|
|
|
function buildReasoningDelta(message: JsonRecord): JsonRecord | null {
|
|
const delta: JsonRecord = {};
|
|
if (Array.isArray(message.reasoning_details)) {
|
|
delta.reasoning_details = message.reasoning_details;
|
|
}
|
|
|
|
if (!addReadableReasoning(message, delta)) {
|
|
addUnsupportedReasoning(message, delta);
|
|
}
|
|
|
|
return Object.keys(delta).length > 0 ? delta : null;
|
|
}
|
|
|
|
function sseEvent(payload: JsonRecord): string {
|
|
return `data: ${JSON.stringify(payload)}\n\n`;
|
|
}
|
|
|
|
export function synthesizeOpenAiSseFromJson(jsonText: string): string {
|
|
let parsed: unknown;
|
|
try {
|
|
parsed = JSON.parse(jsonText);
|
|
} catch {
|
|
return "";
|
|
}
|
|
if (!isRecord(parsed)) return "";
|
|
|
|
const choices = parsed.choices;
|
|
if (!Array.isArray(choices) || choices.length === 0) return "";
|
|
|
|
const id = typeof parsed.id === "string" && parsed.id ? parsed.id : "chatcmpl-omniroute-sse";
|
|
const created = typeof parsed.created === "number" ? parsed.created : 0;
|
|
const model = typeof parsed.model === "string" ? parsed.model : "";
|
|
const base = { id, object: "chat.completion.chunk", created, model };
|
|
|
|
let out = "";
|
|
let emittedAny = false;
|
|
|
|
choices.forEach((choice, fallbackIndex) => {
|
|
if (!isRecord(choice)) return;
|
|
const index = typeof choice.index === "number" ? choice.index : fallbackIndex;
|
|
const message = isRecord(choice.message) ? choice.message : {};
|
|
|
|
// Emit role, reasoning_content, content and tool_calls as SEPARATE sequential
|
|
// deltas — the same shape a real reasoning model streams (reasoning first,
|
|
// then content). Combining them in one delta caused the openai→openai
|
|
// translator to re-split and DUPLICATE reasoning_content across chunks
|
|
// (#3089 follow-up); separate deltas pass through cleanly with no duplication.
|
|
const role = typeof message.role === "string" ? message.role : "assistant";
|
|
const emitDelta = (delta: JsonRecord) => {
|
|
out += sseEvent({ ...base, choices: [{ index, delta, finish_reason: null }] });
|
|
};
|
|
|
|
emitDelta({ role });
|
|
const reasoningDelta = buildReasoningDelta(message);
|
|
if (reasoningDelta) {
|
|
emitDelta(reasoningDelta);
|
|
}
|
|
if (typeof message.content === "string" && message.content.length > 0) {
|
|
emitDelta({ content: message.content });
|
|
}
|
|
if (Array.isArray(message.tool_calls) && message.tool_calls.length > 0) {
|
|
emitDelta({ tool_calls: message.tool_calls });
|
|
}
|
|
|
|
const finishReason = normalizeOpenAICompatibleFinishReasonString(choice.finish_reason);
|
|
const finalChoice: JsonRecord = { index, delta: {}, finish_reason: finishReason };
|
|
const finalChunk: JsonRecord = { ...base, choices: [finalChoice] };
|
|
if (isRecord(parsed.usage)) finalChunk.usage = parsed.usage;
|
|
out += sseEvent(finalChunk);
|
|
emittedAny = true;
|
|
});
|
|
|
|
if (!emittedAny) return "";
|
|
out += "data: [DONE]\n\n";
|
|
return out;
|
|
}
|