feat: stream Reginald's replies token-by-token
The 26b is slow (~30s/call); the chat now shows the answer forming instead of freezing until it's done. The final prose streams over SSE; tool-calling turns stay structured (no partial tokens), so streaming kicks in for the narration. core (@commitea/core): - chat-client.complete gains an optional onToken — when set, it requests stream:true and parses the OpenAI SSE stream, emitting content deltas and assembling streamed tool-call argument fragments into the final result. - GiteaHttpResponse exposes the optional `body` stream (real fetch has it; stubs don't). agent-loop threads onToken to each completion. app: - model:chat forwards each delta to the renderer (event.sender.send); preload exposes model.onToken(cb) → unsubscribe. useChat accumulates the live stream into a growing bubble (with a cursor), replaced by the authoritative final content when the turn resolves. Unconfigured → scripted reply, unchanged. Verified: 118 core tests green (2 streaming: SSE content deltas + tool-call fragment assembly), desktop typecheck clean, 14 fixture e2e green. Live: a real turn against gemma-4-26b assembles the correct answer via the streaming path (live-reginald green) — the reply now renders token-by-token. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -31,18 +31,20 @@ function stringify(result: unknown): string {
|
||||
}
|
||||
|
||||
export async function runAgentTurn(opts: {
|
||||
complete: (messages: ChatMessage[], tools?: ToolDecl[]) => Promise<CompletionResult>
|
||||
complete: (messages: ChatMessage[], tools?: ToolDecl[], onToken?: (delta: string) => void) => Promise<CompletionResult>
|
||||
messages: ChatMessage[]
|
||||
tools: ToolDecl[]
|
||||
execute: ToolExecutor
|
||||
maxSteps?: number
|
||||
/** Streams content deltas as the model produces prose (final-answer streaming). */
|
||||
onToken?: (delta: string) => void
|
||||
}): Promise<AgentTurn> {
|
||||
const maxSteps = opts.maxSteps ?? DEFAULT_MAX_STEPS
|
||||
const convo: ChatMessage[] = [...opts.messages]
|
||||
const steps: AgentStep[] = []
|
||||
|
||||
for (let step = 0; step < maxSteps; step++) {
|
||||
const { content, toolCalls } = await opts.complete(convo, opts.tools)
|
||||
const { content, toolCalls } = await opts.complete(convo, opts.tools, opts.onToken)
|
||||
if (toolCalls.length === 0) {
|
||||
convo.push({ role: 'assistant', content })
|
||||
return { content, steps, messages: convo }
|
||||
@@ -63,7 +65,7 @@ export async function runAgentTurn(opts: {
|
||||
}
|
||||
|
||||
// Out of tool budget — force a final prose answer with tools withheld.
|
||||
const final = await opts.complete(convo, [])
|
||||
const final = await opts.complete(convo, [], opts.onToken)
|
||||
convo.push({ role: 'assistant', content: final.content })
|
||||
return { content: final.content, steps, messages: convo }
|
||||
}
|
||||
|
||||
@@ -25,6 +25,48 @@ describe('createChatClient', () => {
|
||||
return { fetch, calls }
|
||||
}
|
||||
|
||||
function sseStub(chunks: string[]) {
|
||||
const calls: { url: string; body: unknown }[] = []
|
||||
const fetch: FetchLike = (url, init) => {
|
||||
calls.push({ url, body: init?.body ? JSON.parse(init.body) : undefined })
|
||||
const enc = new TextEncoder()
|
||||
const body = new ReadableStream<Uint8Array>({
|
||||
start(c) {
|
||||
for (const ch of chunks) c.enqueue(enc.encode(ch))
|
||||
c.close()
|
||||
},
|
||||
})
|
||||
return Promise.resolve({ ok: true, status: 200, body, json: () => Promise.resolve({}), text: () => Promise.resolve('') })
|
||||
}
|
||||
return { fetch, calls }
|
||||
}
|
||||
|
||||
it('streams content deltas via onToken and returns the assembled result', async () => {
|
||||
const { fetch, calls } = sseStub([
|
||||
'data: {"choices":[{"delta":{"content":"Right "}}]}\n\n',
|
||||
'data: {"choices":[{"delta":{"content":"now: #2."}}]}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
])
|
||||
const client = createChatClient({ baseUrl: 'http://x/v1', model: 'm' }, fetch)
|
||||
const tokens: string[] = []
|
||||
const res = await client.complete([{ role: 'user', content: 'now?' }], undefined, (d) => tokens.push(d))
|
||||
|
||||
expect(tokens).toEqual(['Right ', 'now: #2.'])
|
||||
expect(res.content).toBe('Right now: #2.')
|
||||
expect((calls[0].body as { stream?: boolean }).stream).toBe(true)
|
||||
})
|
||||
|
||||
it('assembles a streamed tool call from argument deltas', async () => {
|
||||
const { fetch } = sseStub([
|
||||
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c1","function":{"name":"query_project","arguments":"{\\"view\\""}}]}}]}\n\n',
|
||||
'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":":\\"focus\\"}"}}]}}]}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
])
|
||||
const client = createChatClient({ baseUrl: 'http://x/v1', model: 'm' }, fetch)
|
||||
const res = await client.complete([{ role: 'user', content: 'x' }], [{ name: 'query_project', description: '', parameters: {} }], () => {})
|
||||
expect(res.toolCalls).toEqual([{ id: 'c1', name: 'query_project', arguments: '{"view":"focus"}' }])
|
||||
})
|
||||
|
||||
it('POSTs to /chat/completions and parses content', async () => {
|
||||
const { fetch, calls } = stub({ choices: [{ message: { content: 'the focus is #2' } }] })
|
||||
const client = createChatClient({ baseUrl: 'http://localhost:1234/v1', model: 'gemma' }, fetch)
|
||||
|
||||
@@ -48,8 +48,12 @@ export interface CompletionResult {
|
||||
toolCalls: ToolCall[]
|
||||
}
|
||||
|
||||
/** Called with each streamed content delta (final-prose streaming). */
|
||||
export type OnToken = (delta: string) => void
|
||||
|
||||
export interface ChatClient {
|
||||
complete(messages: ChatMessage[], tools?: ToolDecl[]): Promise<CompletionResult>
|
||||
/** When `onToken` is given, the response streams (SSE) and each content delta is emitted. */
|
||||
complete(messages: ChatMessage[], tools?: ToolDecl[], onToken?: OnToken): Promise<CompletionResult>
|
||||
}
|
||||
|
||||
/** Map our message shape to the OpenAI wire shape. */
|
||||
@@ -85,7 +89,7 @@ export function createChatClient(config: ModelConfig, fetchImpl: FetchLike): Cha
|
||||
const url = `${config.baseUrl.replace(/\/+$/, '')}/chat/completions`
|
||||
|
||||
return {
|
||||
async complete(messages, tools) {
|
||||
async complete(messages, tools, onToken) {
|
||||
const body: Record<string, unknown> = {
|
||||
model: config.model,
|
||||
messages: messages.map(toWireMessage),
|
||||
@@ -95,11 +99,13 @@ export function createChatClient(config: ModelConfig, fetchImpl: FetchLike): Cha
|
||||
body.tools = tools.map(toWireTool)
|
||||
body.tool_choice = 'auto'
|
||||
}
|
||||
const stream = !!onToken
|
||||
if (stream) body.stream = true
|
||||
const res = await fetchImpl(url, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
Accept: 'application/json',
|
||||
Accept: stream ? 'text/event-stream' : 'application/json',
|
||||
...(config.apiKey ? { Authorization: `Bearer ${config.apiKey}` } : {}),
|
||||
},
|
||||
body: JSON.stringify(body),
|
||||
@@ -108,6 +114,8 @@ export function createChatClient(config: ModelConfig, fetchImpl: FetchLike): Cha
|
||||
const text = await res.text().catch(() => '')
|
||||
throw new Error(`model completion failed (${res.status}): ${text.slice(0, 200)}`)
|
||||
}
|
||||
if (stream && res.body) return readStream(res.body, onToken!)
|
||||
|
||||
const json = (await res.json()) as { choices?: { message: RawChoiceMessage }[] }
|
||||
const msg = json.choices?.[0]?.message
|
||||
return {
|
||||
@@ -121,3 +129,56 @@ export function createChatClient(config: ModelConfig, fetchImpl: FetchLike): Cha
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/** Streamed tool-call delta: name arrives first, arguments accumulate across chunks. */
|
||||
interface RawToolDelta {
|
||||
index: number
|
||||
id?: string
|
||||
function?: { name?: string; arguments?: string }
|
||||
}
|
||||
|
||||
/** Parse an OpenAI SSE stream: emit content deltas via onToken, accumulate the final result. */
|
||||
async function readStream(body: ReadableStream<Uint8Array>, onToken: OnToken): Promise<CompletionResult> {
|
||||
const reader = body.getReader()
|
||||
const decoder = new TextDecoder()
|
||||
let buffer = ''
|
||||
let content = ''
|
||||
const toolAcc: { id: string; name: string; arguments: string }[] = []
|
||||
|
||||
const handle = (data: string) => {
|
||||
if (data === '[DONE]') return
|
||||
let chunk: { choices?: { delta?: { content?: string; tool_calls?: RawToolDelta[] } }[] }
|
||||
try {
|
||||
chunk = JSON.parse(data)
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
const delta = chunk.choices?.[0]?.delta
|
||||
if (!delta) return
|
||||
if (delta.content) {
|
||||
content += delta.content
|
||||
onToken(delta.content)
|
||||
}
|
||||
for (const tc of delta.tool_calls ?? []) {
|
||||
const slot = (toolAcc[tc.index] ??= { id: '', name: '', arguments: '' })
|
||||
if (tc.id) slot.id = tc.id
|
||||
if (tc.function?.name) slot.name = tc.function.name
|
||||
if (tc.function?.arguments) slot.arguments += tc.function.arguments
|
||||
}
|
||||
}
|
||||
|
||||
for (;;) {
|
||||
const { done, value } = 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 t = line.trim()
|
||||
if (t.startsWith('data:')) handle(t.slice(5).trim())
|
||||
}
|
||||
}
|
||||
if (buffer.trim().startsWith('data:')) handle(buffer.trim().slice(5).trim())
|
||||
|
||||
return { content, toolCalls: toolAcc.filter((t) => t.name).map((t) => ({ id: t.id, name: t.name, arguments: t.arguments })) }
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user