Tianyi Cui
ab19fed77c
The first real LlmAdapter implementations, shipped as a deliberate pair:
same models and wire protocol, completely different internals, so the
StreamChunk protocol is verified across independent implementations.
- dsh-llm-deepseek: hand-rolled fetch + SSE parser + chunk-translation
state machine against the official chat-completions format (thinking
mode via top-level thinking/reasoning_effort; the empty-string
reasoning_content first chunk; usage attached to the finish chunk or
trailing; reasoning_content passback on tool-call turns; disjoint
cache-token accounting).
- dsh-llm-pi-ai: the same endpoint through @earendil-works/pi-ai,
mapping its event vocabulary (parsed tool arguments, in-stream error
events, folded reasoning tokens) onto the same chunks.
The agent loop now honors the in-band error path: an adapter that ends
its stream with finish {kind:error|aborted} (the only option for
adapters that can't throw mid-stream, like pi-ai) is translated into a
step error, so the turn ends error/aborted with a logged error event
instead of a normal completed assistant message. This makes the
StreamChunk error contract real for both adapters; docs/architecture.md
and the StreamChunk doc are updated accordingly.
New yarn test:e2e (vitest.e2e.config.ts, *.e2e.ts) runs key-gated
real-API matrices for both adapters across V4 Flash/Pro and all
thinking/effort levels; it self-skips without DEEPSEEK_API_KEY. Unit
suites run against local node:http mock SSE servers at 100% per-file
coverage.
72 lines
2.7 KiB
TypeScript
72 lines
2.7 KiB
TypeScript
/**
|
|
* Minimal SSE (text/event-stream) parser for the chat-completions stream.
|
|
*
|
|
* Yields each event's `data:` payload as a string, ending with the literal
|
|
* `'[DONE]'` sentinel so the consumer owns end-of-stream flushing. A stream
|
|
* that closes WITHOUT `[DONE]` is a protocol violation → `LlmError`.
|
|
*
|
|
* Handles the wire realities: payloads split across network reads at
|
|
* arbitrary byte positions (including mid-UTF-8), CRLF line endings,
|
|
* multi-`data:` events (joined with newlines per the SSE spec), comment
|
|
* lines, and non-data fields (ignored).
|
|
*
|
|
* @module dsh-llm-deepseek/sse
|
|
*/
|
|
|
|
import { LlmError } from '@deepseek-ai/dsh-llm'
|
|
|
|
/** The terminal payload DeepSeek (and OpenAI) send after the last chunk. */
|
|
export const DONE = '[DONE]'
|
|
|
|
/** Extract the joined data payload from one raw SSE event block. */
|
|
function eventData(block: string): string | undefined {
|
|
const data: string[] = []
|
|
for (const rawLine of block.split('\n')) {
|
|
const line = rawLine.endsWith('\r') ? rawLine.slice(0, -1) : rawLine
|
|
if (line.startsWith('data:')) {
|
|
// The spec strips ONE leading space after the colon.
|
|
data.push(line.startsWith('data: ') ? line.slice(6) : line.slice(5))
|
|
}
|
|
// Comments (':…') and other fields (event:, id:, retry:) are ignored.
|
|
}
|
|
if (data.length === 0) return undefined
|
|
return data.join('\n')
|
|
}
|
|
|
|
/**
|
|
* Parse a byte stream into SSE data payloads. Yields `[DONE]` as the final
|
|
* value and returns; throws `LlmError('STREAM_CLOSED')` when the stream ends
|
|
* without it (truncated response — the model call cannot be trusted).
|
|
*/
|
|
export async function* parseSse(stream: AsyncIterable<Uint8Array>): AsyncGenerator<string> {
|
|
const decoder = new TextDecoder()
|
|
let buffer = ''
|
|
|
|
for await (const bytes of stream) {
|
|
buffer += decoder.decode(bytes, { stream: true })
|
|
// Events are separated by a blank line (\n\n; tolerate \r\n\r\n via the
|
|
// per-line \r strip in eventData and a normalized split here).
|
|
let boundary: number
|
|
while ((boundary = buffer.search(/\r?\n\r?\n/)) !== -1) {
|
|
const matched = /\r?\n\r?\n/.exec(buffer.slice(boundary))
|
|
const block = buffer.slice(0, boundary)
|
|
// matched cannot be null: search() just found the same pattern at 0.
|
|
buffer = buffer.slice(boundary + (matched as RegExpExecArray)[0].length)
|
|
const data = eventData(block)
|
|
if (data === undefined) continue
|
|
yield data
|
|
if (data === DONE) return
|
|
}
|
|
}
|
|
|
|
// Flush any final un-terminated event (servers usually end with \n\n, but
|
|
// a trailing block without one is still parseable).
|
|
buffer += decoder.decode()
|
|
const data = eventData(buffer)
|
|
if (data !== undefined) {
|
|
yield data
|
|
if (data === DONE) return
|
|
}
|
|
throw new LlmError('SSE stream ended without [DONE]', 'STREAM_CLOSED')
|
|
}
|