@@ -10,6 +10,103 @@ import {
10101111const DEFAULT_MAX_SDK_RETRY_WAIT_SECONDS = 60;
121213+function hasReadableSseData(block: string): boolean {
14+const dataLines = block
15+.split(/\r\n|\n|\r/)
16+.filter((line) => line === "data" || line.startsWith("data:"))
17+.map((line) => {
18+if (line === "data") {
19+return "";
20+}
21+const value = line.slice("data:".length);
22+return value.startsWith(" ") ? value.slice(1) : value;
23+});
24+return dataLines.length > 0 && dataLines.join("\n").trim().length > 0;
25+}
26+27+function findSseEventBoundary(buffer: string): { index: number; length: number } | undefined {
28+let best: { index: number; length: number } | undefined;
29+for (const delimiter of ["\r\n\r\n", "\n\n", "\r\r"]) {
30+const index = buffer.indexOf(delimiter);
31+if (index === -1) {
32+continue;
33+}
34+if (!best || index < best.index) {
35+best = { index, length: delimiter.length };
36+}
37+}
38+return best;
39+}
40+41+function sanitizeOpenAISdkSseResponse(response: Response): Response {
42+const contentType = response.headers.get("content-type") ?? "";
43+if (!response.body || !/\btext\/event-stream\b/i.test(contentType)) {
44+return response;
45+}
46+47+const source = response.body;
48+const decoder = new TextDecoder();
49+const encoder = new TextEncoder();
50+let reader: ReadableStreamDefaultReader<Uint8Array> | undefined;
51+let buffer = "";
52+53+const enqueueSanitized = (
54+controller: ReadableStreamDefaultController<Uint8Array>,
55+text: string,
56+) => {
57+buffer += text;
58+for (;;) {
59+const boundary = findSseEventBoundary(buffer);
60+if (!boundary) {
61+return;
62+}
63+const block = buffer.slice(0, boundary.index);
64+const separator = buffer.slice(boundary.index, boundary.index + boundary.length);
65+buffer = buffer.slice(boundary.index + boundary.length);
66+// OpenAI's SDK currently tries to JSON.parse event-only or blank-data SSE
67+// messages. Drop those malformed keepalive-style blocks before it parses.
68+if (hasReadableSseData(block)) {
69+controller.enqueue(encoder.encode(`${block}${separator}`));
70+}
71+}
72+};
73+74+const sanitizedBody = new ReadableStream<Uint8Array>({
75+start() {
76+reader = source.getReader();
77+},
78+async pull(controller) {
79+try {
80+const chunk = await reader?.read();
81+if (!chunk || chunk.done) {
82+const tail = decoder.decode();
83+if (tail) {
84+enqueueSanitized(controller, tail);
85+}
86+if (buffer && hasReadableSseData(buffer)) {
87+controller.enqueue(encoder.encode(buffer));
88+}
89+buffer = "";
90+controller.close();
91+return;
92+}
93+enqueueSanitized(controller, decoder.decode(chunk.value, { stream: true }));
94+} catch (error) {
95+controller.error(error);
96+}
97+},
98+async cancel(reason) {
99+await reader?.cancel(reason);
100+},
101+});
102+103+return new Response(sanitizedBody, {
104+status: response.status,
105+statusText: response.statusText,
106+headers: response.headers,
107+});
108+}
109+13110function parseRetryAfterSeconds(headers: Headers): number | undefined {
14111const retryAfterMs = headers.get("retry-after-ms");
15112if (retryAfterMs) {
@@ -218,6 +315,7 @@ export function buildGuardedModelFetch(model: Model<Api>, timeoutMs?: number): t
218315 headers,
219316});
220317}
318+response = sanitizeOpenAISdkSseResponse(response);
221319return buildManagedResponse(response, result.release);
222320};
223321}
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。