mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-07-31 04:12:10 +03:00
refactor(executors): extract pure JSONL stream translation from huggingchat (#6016)
Extract the pure JSONL->OpenAI-SSE translation (sseChunk, parseJsonlLine, streamJsonlToOpenAi, readJsonlResponse) verbatim into the leaf huggingchat/jsonlStream.ts. They consume a passed-in ReadableStream (no fetch/network/state). Host imports back the two it uses; all module-private (no re-export). Host 812 -> 594 LOC. Byte-identical bodies (verbatim), leaf has zero imports (no cycle). Cookie/auth/multipart/execute untouched. Adds a split-guard; consumer tests stay green (executor-huggingchat 6, huggingchat-model-catalog 3).
This commit is contained in:
Diego Rodrigues de Sa e Souza
committed by
GitHub
GitHub
parent
f7dea02853
commit
20a67d4456
@@ -27,6 +27,7 @@ import {
|
||||
import { FETCH_TIMEOUT_MS } from "../config/constants.ts";
|
||||
import { buildErrorBody, sanitizeErrorMessage } from "../utils/error.ts";
|
||||
import { normalizeSessionCookieHeader } from "@/lib/providers/webCookieAuth";
|
||||
import { streamJsonlToOpenAi, readJsonlResponse } from "./huggingchat/jsonlStream.ts";
|
||||
|
||||
const HUGGINGFACE_BASE = "https://huggingface.co";
|
||||
const CONVERSATION_URL = `${HUGGINGFACE_BASE}/chat/conversation`;
|
||||
@@ -108,10 +109,6 @@ function buildConversationPrompt(messages: Array<Record<string, unknown>>): {
|
||||
};
|
||||
}
|
||||
|
||||
function sseChunk(data: unknown): string {
|
||||
return `data: ${JSON.stringify(data)}\n\n`;
|
||||
}
|
||||
|
||||
function estimateTokens(text: string): number {
|
||||
return Math.max(1, Math.ceil((text || "").length / 4));
|
||||
}
|
||||
@@ -245,222 +242,6 @@ function mergeCookieHeaderWithSetCookie(cookieHeader: string, setCookieHeaders:
|
||||
return [...cookieMap.entries()].map(([name, value]) => `${name}=${value}`).join("; ");
|
||||
}
|
||||
|
||||
function parseJsonlLine(line: string): {
|
||||
token?: string;
|
||||
done?: boolean;
|
||||
error?: string;
|
||||
text?: string;
|
||||
} {
|
||||
try {
|
||||
const event = JSON.parse(line);
|
||||
|
||||
if (event.type === "stream" && typeof event.token === "string") {
|
||||
const token = event.token.replace(/\0/g, "");
|
||||
if (token) return { token };
|
||||
}
|
||||
|
||||
if (event.type === "finalAnswer" && typeof event.text === "string") {
|
||||
return { text: event.text, done: true };
|
||||
}
|
||||
|
||||
if (event.type === "status") {
|
||||
if (event.status === "error") {
|
||||
return { error: event.message || "HuggingChat generation error" };
|
||||
}
|
||||
if (event.status === "finished") {
|
||||
return { done: true };
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Skip non-JSON lines
|
||||
}
|
||||
|
||||
return {};
|
||||
}
|
||||
|
||||
async function* streamJsonlToOpenAi(
|
||||
body: ReadableStream<Uint8Array>,
|
||||
model: string,
|
||||
id: string,
|
||||
created: number,
|
||||
signal?: AbortSignal | null
|
||||
): AsyncGenerator<string> {
|
||||
const reader = body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = "";
|
||||
let emittedRole = false;
|
||||
let fullText = "";
|
||||
let finished = false;
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
if (signal?.aborted) break;
|
||||
|
||||
const { value, done } = await reader.read();
|
||||
if (done) break;
|
||||
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
|
||||
const lines = buffer.split("\n");
|
||||
buffer = lines.pop() || "";
|
||||
|
||||
for (const line of lines) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
|
||||
const parsed = parseJsonlLine(trimmed);
|
||||
|
||||
if (parsed.error) {
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
|
||||
});
|
||||
yield "data: [DONE]\n\n";
|
||||
finished = true;
|
||||
return;
|
||||
}
|
||||
|
||||
if (parsed.token) {
|
||||
if (!emittedRole) {
|
||||
emittedRole = true;
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }],
|
||||
});
|
||||
}
|
||||
|
||||
fullText += parsed.token;
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: { content: parsed.token }, finish_reason: null }],
|
||||
});
|
||||
}
|
||||
|
||||
if (parsed.text) {
|
||||
const remaining = parsed.text.slice(fullText.length);
|
||||
if (remaining) {
|
||||
if (!emittedRole) {
|
||||
emittedRole = true;
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }],
|
||||
});
|
||||
}
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: { content: remaining }, finish_reason: null }],
|
||||
});
|
||||
}
|
||||
finished = true;
|
||||
break;
|
||||
}
|
||||
|
||||
if (parsed.done) {
|
||||
finished = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (finished) break;
|
||||
}
|
||||
|
||||
if (!finished && buffer.trim()) {
|
||||
const parsed = parseJsonlLine(buffer.trim());
|
||||
if (parsed.token && !signal?.aborted) {
|
||||
if (!emittedRole) {
|
||||
emittedRole = true;
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }],
|
||||
});
|
||||
}
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: { content: parsed.token }, finish_reason: null }],
|
||||
});
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
|
||||
if (!signal?.aborted) {
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
|
||||
});
|
||||
yield "data: [DONE]\n\n";
|
||||
}
|
||||
}
|
||||
|
||||
async function readJsonlResponse(
|
||||
body: ReadableStream<Uint8Array>,
|
||||
signal?: AbortSignal | null
|
||||
): Promise<string> {
|
||||
const reader = body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = "";
|
||||
let fullText = "";
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
if (signal?.aborted) break;
|
||||
|
||||
const { value, done } = await reader.read();
|
||||
if (done) break;
|
||||
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
|
||||
const lines = buffer.split("\n");
|
||||
buffer = lines.pop() || "";
|
||||
|
||||
for (const line of lines) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
|
||||
const parsed = parseJsonlLine(trimmed);
|
||||
if (parsed.token) fullText += parsed.token;
|
||||
if (parsed.text) return parsed.text;
|
||||
if (parsed.error) throw new Error(parsed.error);
|
||||
}
|
||||
}
|
||||
|
||||
if (buffer.trim()) {
|
||||
const parsed = parseJsonlLine(buffer.trim());
|
||||
if (parsed.text) return parsed.text;
|
||||
if (parsed.token) fullText += parsed.token;
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
|
||||
return fullText;
|
||||
}
|
||||
|
||||
// -- Executor ----------------------------------------------------------------
|
||||
|
||||
export class HuggingChatExecutor extends BaseExecutor {
|
||||
|
||||
221
open-sse/executors/huggingchat/jsonlStream.ts
Normal file
221
open-sse/executors/huggingchat/jsonlStream.ts
Normal file
@@ -0,0 +1,221 @@
|
||||
// Pure JSONL stream translation (HuggingChat NDJSON -> OpenAI SSE). Verbatim from huggingchat.ts.
|
||||
|
||||
export function sseChunk(data: unknown): string {
|
||||
return `data: ${JSON.stringify(data)}\n\n`;
|
||||
}
|
||||
|
||||
export function parseJsonlLine(line: string): {
|
||||
token?: string;
|
||||
done?: boolean;
|
||||
error?: string;
|
||||
text?: string;
|
||||
} {
|
||||
try {
|
||||
const event = JSON.parse(line);
|
||||
|
||||
if (event.type === "stream" && typeof event.token === "string") {
|
||||
const token = event.token.replace(/\0/g, "");
|
||||
if (token) return { token };
|
||||
}
|
||||
|
||||
if (event.type === "finalAnswer" && typeof event.text === "string") {
|
||||
return { text: event.text, done: true };
|
||||
}
|
||||
|
||||
if (event.type === "status") {
|
||||
if (event.status === "error") {
|
||||
return { error: event.message || "HuggingChat generation error" };
|
||||
}
|
||||
if (event.status === "finished") {
|
||||
return { done: true };
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Skip non-JSON lines
|
||||
}
|
||||
|
||||
return {};
|
||||
}
|
||||
|
||||
export async function* streamJsonlToOpenAi(
|
||||
body: ReadableStream<Uint8Array>,
|
||||
model: string,
|
||||
id: string,
|
||||
created: number,
|
||||
signal?: AbortSignal | null
|
||||
): AsyncGenerator<string> {
|
||||
const reader = body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = "";
|
||||
let emittedRole = false;
|
||||
let fullText = "";
|
||||
let finished = false;
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
if (signal?.aborted) break;
|
||||
|
||||
const { value, done } = await reader.read();
|
||||
if (done) break;
|
||||
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
|
||||
const lines = buffer.split("\n");
|
||||
buffer = lines.pop() || "";
|
||||
|
||||
for (const line of lines) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
|
||||
const parsed = parseJsonlLine(trimmed);
|
||||
|
||||
if (parsed.error) {
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
|
||||
});
|
||||
yield "data: [DONE]\n\n";
|
||||
finished = true;
|
||||
return;
|
||||
}
|
||||
|
||||
if (parsed.token) {
|
||||
if (!emittedRole) {
|
||||
emittedRole = true;
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }],
|
||||
});
|
||||
}
|
||||
|
||||
fullText += parsed.token;
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: { content: parsed.token }, finish_reason: null }],
|
||||
});
|
||||
}
|
||||
|
||||
if (parsed.text) {
|
||||
const remaining = parsed.text.slice(fullText.length);
|
||||
if (remaining) {
|
||||
if (!emittedRole) {
|
||||
emittedRole = true;
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }],
|
||||
});
|
||||
}
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: { content: remaining }, finish_reason: null }],
|
||||
});
|
||||
}
|
||||
finished = true;
|
||||
break;
|
||||
}
|
||||
|
||||
if (parsed.done) {
|
||||
finished = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (finished) break;
|
||||
}
|
||||
|
||||
if (!finished && buffer.trim()) {
|
||||
const parsed = parseJsonlLine(buffer.trim());
|
||||
if (parsed.token && !signal?.aborted) {
|
||||
if (!emittedRole) {
|
||||
emittedRole = true;
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }],
|
||||
});
|
||||
}
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: { content: parsed.token }, finish_reason: null }],
|
||||
});
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
|
||||
if (!signal?.aborted) {
|
||||
yield sseChunk({
|
||||
id,
|
||||
object: "chat.completion.chunk",
|
||||
created,
|
||||
model,
|
||||
choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
|
||||
});
|
||||
yield "data: [DONE]\n\n";
|
||||
}
|
||||
}
|
||||
|
||||
export async function readJsonlResponse(
|
||||
body: ReadableStream<Uint8Array>,
|
||||
signal?: AbortSignal | null
|
||||
): Promise<string> {
|
||||
const reader = body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = "";
|
||||
let fullText = "";
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
if (signal?.aborted) break;
|
||||
|
||||
const { value, done } = await reader.read();
|
||||
if (done) break;
|
||||
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
|
||||
const lines = buffer.split("\n");
|
||||
buffer = lines.pop() || "";
|
||||
|
||||
for (const line of lines) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
|
||||
const parsed = parseJsonlLine(trimmed);
|
||||
if (parsed.token) fullText += parsed.token;
|
||||
if (parsed.text) return parsed.text;
|
||||
if (parsed.error) throw new Error(parsed.error);
|
||||
}
|
||||
}
|
||||
|
||||
if (buffer.trim()) {
|
||||
const parsed = parseJsonlLine(buffer.trim());
|
||||
if (parsed.text) return parsed.text;
|
||||
if (parsed.token) fullText += parsed.token;
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
|
||||
return fullText;
|
||||
}
|
||||
32
tests/unit/huggingchat-jsonl-split.test.ts
Normal file
32
tests/unit/huggingchat-jsonl-split.test.ts
Normal file
@@ -0,0 +1,32 @@
|
||||
import { test } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { readFileSync } from "node:fs";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { dirname, join } from "node:path";
|
||||
|
||||
// Split-guard for the huggingchat executor JSONL-stream extraction.
|
||||
// The pure JSONL->OpenAI-SSE translation lives in huggingchat/jsonlStream.ts
|
||||
// (consumes a passed-in ReadableStream; no host state/fetch/auth). Host imports it back.
|
||||
const HERE = dirname(fileURLToPath(import.meta.url));
|
||||
const EXE = join(HERE, "../../open-sse/executors");
|
||||
const HOST = join(EXE, "huggingchat.ts");
|
||||
const LEAF = join(EXE, "huggingchat/jsonlStream.ts");
|
||||
|
||||
test("leaf hosts the jsonl translators and does not import the host", () => {
|
||||
const src = readFileSync(LEAF, "utf8");
|
||||
assert.match(src, /export async function\* streamJsonlToOpenAi\b/);
|
||||
assert.match(src, /export async function readJsonlResponse\b/);
|
||||
assert.match(src, /export function (sseChunk|parseJsonlLine)\b/);
|
||||
assert.doesNotMatch(src, /from "\.\.\/huggingchat\.ts"/);
|
||||
});
|
||||
|
||||
test("host imports the jsonl translators back from the leaf", () => {
|
||||
const host = readFileSync(HOST, "utf8");
|
||||
assert.match(host, /from "\.\/huggingchat\/jsonlStream\.ts"/);
|
||||
});
|
||||
|
||||
test("parseJsonlLine parses a jsonl data line", async () => {
|
||||
const { parseJsonlLine } = await import("../../open-sse/executors/huggingchat/jsonlStream.ts");
|
||||
assert.equal(typeof parseJsonlLine, "function");
|
||||
assert.doesNotThrow(() => parseJsonlLine("not json"));
|
||||
});
|
||||
Reference in New Issue
Block a user