|
| 1 | +/** |
| 2 | + * @description Parse OpenAI-compatible `chat/completions` streaming (SSE) bodies |
| 3 | + * @author Zhengqbbb <zhengqbbb@gmail.com> |
| 4 | + * @license MIT |
| 5 | + */ |
| 6 | + |
| 7 | +import { Buffer } from 'node:buffer' |
| 8 | +import { Readable } from 'node:stream' |
| 9 | +import type { ReadableStream as WebReadableStream } from 'node:stream/web' |
| 10 | + |
| 11 | +/** |
| 12 | + * Normalize `fetch` response body to a Node.js readable stream. |
| 13 | + */ |
| 14 | +export function bodyToNodeReadable(body: unknown): NodeJS.ReadableStream { |
| 15 | + if (body == null) |
| 16 | + throw new Error('Response has no body') |
| 17 | + if (typeof (body as WebReadableStream).getReader === 'function') |
| 18 | + return Readable.fromWeb(body as WebReadableStream) |
| 19 | + return body as NodeJS.ReadableStream |
| 20 | +} |
| 21 | + |
| 22 | +/** |
| 23 | + * Append only user-visible completion tokens from `delta.content`. |
| 24 | + * Skips reasoning / `reasoning_content` (not present on `content` in typical deltas). |
| 25 | + */ |
| 26 | +export function appendVisibleDelta(acc: string, delta: { content?: unknown } | undefined): string { |
| 27 | + if (!delta) |
| 28 | + return acc |
| 29 | + const c = delta.content |
| 30 | + if (c == null) |
| 31 | + return acc |
| 32 | + if (typeof c === 'string') |
| 33 | + return acc + c |
| 34 | + if (Array.isArray(c)) { |
| 35 | + let s = acc |
| 36 | + for (const p of c) { |
| 37 | + if (p && typeof p === 'object' && (p as { type?: string, text?: string }).type === 'text') { |
| 38 | + const t = (p as { text?: string }).text |
| 39 | + if (typeof t === 'string') |
| 40 | + s += t |
| 41 | + } |
| 42 | + } |
| 43 | + return s |
| 44 | + } |
| 45 | + return acc |
| 46 | +} |
| 47 | + |
| 48 | +interface StreamChoiceChunk { index?: number, delta?: { content?: unknown } } |
| 49 | + |
| 50 | +interface NonStreamChoice { index?: number, message?: { content?: unknown } } |
| 51 | + |
| 52 | +async function readableToUtf8String(stream: NodeJS.ReadableStream): Promise<string> { |
| 53 | + const chunks: Buffer[] = [] |
| 54 | + for await (const chunk of stream as AsyncIterable<string | Buffer>) { |
| 55 | + if (Buffer.isBuffer(chunk)) |
| 56 | + chunks.push(chunk) |
| 57 | + else if (typeof chunk === 'string') |
| 58 | + chunks.push(Buffer.from(chunk)) |
| 59 | + else |
| 60 | + chunks.push(Buffer.from(String(chunk))) |
| 61 | + } |
| 62 | + return Buffer.concat(chunks as readonly Uint8Array[]).toString('utf8') |
| 63 | +} |
| 64 | + |
| 65 | +/** |
| 66 | + * Parse a non-streaming `chat/completions` JSON body when `stream: true` was ignored. |
| 67 | + * @returns subjects slice, or `undefined` if the body is not a usable completion object. |
| 68 | + */ |
| 69 | +function trySubjectsFromNonStreamCompletionJson( |
| 70 | + body: string, |
| 71 | + choiceCount: number, |
| 72 | +): string[] | undefined { |
| 73 | + const t = body.trim() |
| 74 | + if (!t.startsWith('{')) |
| 75 | + return undefined |
| 76 | + let json: unknown |
| 77 | + try { |
| 78 | + json = JSON.parse(t) |
| 79 | + } |
| 80 | + catch { |
| 81 | + return undefined |
| 82 | + } |
| 83 | + if (!json || typeof json !== 'object') |
| 84 | + return undefined |
| 85 | + const o = json as { choices?: NonStreamChoice[], error?: { message?: string } } |
| 86 | + if (o.error) |
| 87 | + throw new Error(o.error.message || 'OpenAI API error') |
| 88 | + if (!Array.isArray(o.choices)) |
| 89 | + return undefined |
| 90 | + |
| 91 | + const buffers = Array.from({ length: choiceCount }, () => '') |
| 92 | + let maxIndexSeen = -1 |
| 93 | + for (const ch of o.choices) { |
| 94 | + const idx = typeof ch.index === 'number' ? ch.index : 0 |
| 95 | + if (idx >= 0 && idx < choiceCount) { |
| 96 | + buffers[idx] = appendVisibleDelta('', { content: ch.message?.content }) |
| 97 | + maxIndexSeen = Math.max(maxIndexSeen, idx) |
| 98 | + } |
| 99 | + } |
| 100 | + if (maxIndexSeen < 0) |
| 101 | + return undefined |
| 102 | + return buffers.slice(0, maxIndexSeen + 1) |
| 103 | +} |
| 104 | + |
| 105 | +function collectSubjectsFromSseLines( |
| 106 | + body: string, |
| 107 | + choiceCount: number, |
| 108 | +): { buffers: string[], maxIndexSeen: number } { |
| 109 | + const buffers = Array.from({ length: choiceCount }, () => '') |
| 110 | + let maxIndexSeen = -1 |
| 111 | + for (const line of body.split(/\r?\n/)) { |
| 112 | + const trimmed = line.trim() |
| 113 | + if (!trimmed.startsWith('data:')) |
| 114 | + continue |
| 115 | + const payload = trimmed.slice(5).trim() |
| 116 | + if (payload === '[DONE]') |
| 117 | + continue |
| 118 | + try { |
| 119 | + const json = JSON.parse(payload) as { |
| 120 | + error?: { message?: string } |
| 121 | + choices?: StreamChoiceChunk[] |
| 122 | + } |
| 123 | + if (json.error) |
| 124 | + throw new Error(json.error.message || 'OpenAI stream error') |
| 125 | + |
| 126 | + for (const ch of json.choices ?? []) { |
| 127 | + const idx = typeof ch.index === 'number' ? ch.index : 0 |
| 128 | + if (idx >= 0 && idx < choiceCount) { |
| 129 | + buffers[idx] = appendVisibleDelta(buffers[idx], ch.delta) |
| 130 | + maxIndexSeen = Math.max(maxIndexSeen, idx) |
| 131 | + } |
| 132 | + } |
| 133 | + } |
| 134 | + catch (e) { |
| 135 | + if (e instanceof SyntaxError) |
| 136 | + continue |
| 137 | + throw e |
| 138 | + } |
| 139 | + } |
| 140 | + return { buffers, maxIndexSeen } |
| 141 | +} |
| 142 | + |
| 143 | +/** |
| 144 | + * Read an OpenAI-style `chat/completions` response body and return one finished string per choice. |
| 145 | + * Primary path: SSE lines (`data: {...}`) with `choices[].delta`, bucketed by `choices[].index` |
| 146 | + * up to `choiceCount` (requested `n`). Returned length is `maxSeenIndex + 1` (capped by `choiceCount`), |
| 147 | + * mirroring non-stream `choices.length` when fewer parallel completions appear. |
| 148 | + * Fallback: if no choice index ever appears (e.g. provider ignores `stream: true` and returns one JSON object), |
| 149 | + * the full body is parsed as a non-streaming completion using `choices[].message.content`. |
| 150 | + */ |
| 151 | +export async function readChatCompletionStreamToSubjects( |
| 152 | + input: NodeJS.ReadableStream, |
| 153 | + choiceCount: number, |
| 154 | +): Promise<string[]> { |
| 155 | + if (choiceCount < 1) |
| 156 | + throw new Error('choiceCount must be at least 1') |
| 157 | + |
| 158 | + const body = await readableToUtf8String(input) |
| 159 | + const { buffers, maxIndexSeen } = collectSubjectsFromSseLines(body, choiceCount) |
| 160 | + |
| 161 | + if (maxIndexSeen >= 0) |
| 162 | + return buffers.slice(0, maxIndexSeen + 1) |
| 163 | + |
| 164 | + const fromJson = trySubjectsFromNonStreamCompletionJson(body, choiceCount) |
| 165 | + if (fromJson !== undefined) |
| 166 | + return fromJson |
| 167 | + |
| 168 | + throw new Error( |
| 169 | + 'Chat completions response had no streamed choice deltas and is not a parseable non-streaming JSON body with choices (or choices were empty).', |
| 170 | + ) |
| 171 | +} |
0 commit comments