From 1e8c7c66629cb9ae3542b35419c5ed47e0f9e557 Mon Sep 17 00:00:00 2001 From: ztimson Date: Sat, 19 Sep 2026 13:22:48 -0400 Subject: [PATCH] Fix open-ai early termination --- package.json | 2 +- src/open-ai.ts | 115 +++++++++++++++++++++++++++++++------------------ 2 files changed, 74 insertions(+), 43 deletions(-) diff --git a/package.json b/package.json index 6dbe1c7..fe1c16f 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@ztimson/ai-utils", - "version": "1.6.11", + "version": "1.6.12", "description": "AI Utility library", "author": "Zak Timson", "license": "MIT", diff --git a/src/open-ai.ts b/src/open-ai.ts index feff9c6..d23833e 100644 --- a/src/open-ai.ts +++ b/src/open-ai.ts @@ -36,21 +36,37 @@ export class OpenAi extends LLMProvider { private toWire(history: LLMMessage[], system?: string): any[] { const wire: any[] = []; if(system) wire.push({role: 'system', content: system}); - for(const h of history) { - if(h.role === 'tool') { - wire.push({ - role: 'assistant', - content: null, - tool_calls: [{id: h.id, type: 'function', function: {name: h.name, arguments: JSON.stringify(h.args)}}], - }, { - role: 'tool', - tool_call_id: h.id, - content: h.error || h.content || '', - }); - } else { + for(let i = 0; i < history.length; i++) { + const h = history[i]; + if(h.role !== 'tool') { wire.push({role: h.role, content: this.toWireContent(h.content)}); + continue; } + + const calls: any[] = []; + const results: any[] = []; + while(i < history.length && history[i].role === 'tool') { + const tool = history[i]; + calls.push({ + id: tool.id, + type: 'function', + function: { + name: tool.name, + arguments: JSON.stringify(tool.args) + } + }); + results.push({ + role: 'tool', + tool_call_id: tool.id, + content: tool.error || tool.content || '' + }); + i++; + } + wire.push({role: 'assistant', content: null, tool_calls: calls}); + wire.push(...results); + i--; } + return wire; } @@ -74,8 +90,12 @@ export class OpenAi extends LLMProvider { description: t.description, parameters: { type: 'object', - properties: t.args ? objectMap(t.args, (key, value) => ({...value, required: undefined})) : {}, - required: t.args ? Object.entries(t.args).filter(t => t[1].required).map(t => t[0]) : [] + properties: t.args + ? objectMap(t.args, (key, value) => ({...value, required: undefined})) + : {}, + required: t.args + ? Object.entries(t.args).filter(t => t[1].required).map(t => t[0]) + : [] } } })) @@ -83,7 +103,10 @@ export class OpenAi extends LLMProvider { if(options.schema) { const schema = convertSchema(options.schema); - requestParams.response_format = {type: 'json_schema', json_schema: {name: 'response', strict: true, schema}}; + requestParams.response_format = { + type: 'json_schema', + json_schema: {name: 'response', strict: true, schema} + }; } if(options.stream) requestParams.stream_options = {include_usage: true}; @@ -93,47 +116,57 @@ export class OpenAi extends LLMProvider { requestParams.messages = this.toWire(history.filter(h => h.role !== 'system'), options.system); const callStart = Date.now(); - const resp: any = await this.tokenPool.run(token => this.getClient(token).chat.completions.create(requestParams)).catch(err => { + const resp: any = await this.tokenPool.run(token => + this.getClient(token).chat.completions.create(requestParams) + ).catch(err => { err.message += `\n\nMessages:\n${JSON.stringify(requestParams.messages, null, 2)}`; throw err; }); - let usage: any, finishReason: string | undefined, msg: any = {content: '', tool_calls: []}; + let usage: any; + let finishReason: string | undefined; + let msg: any = {content: '', tool_calls: []}; if(options.stream) { - for await (const chunk of resp) { - if(controller.signal.aborted) break; - if(chunk.usage) usage = chunk.usage; - if(chunk.choices[0]?.finish_reason) finishReason = chunk.choices[0].finish_reason; - if(chunk.choices[0]?.delta?.content) { - msg.content += chunk.choices[0].delta.content; - options.stream({text: chunk.choices[0].delta.content}); - } - if(chunk.choices[0]?.delta?.tool_calls) { - for(const deltaTC of chunk.choices[0].delta.tool_calls) { - const existing = deltaTC.index != null - ? msg.tool_calls.find((tc: any) => tc.index === deltaTC.index) - : (deltaTC.id ? msg.tool_calls.find((tc: any) => tc.id === deltaTC.id) : undefined); - if(existing) { + let streamCompleted = false; + try { + for await (const chunk of resp) { + if(controller.signal.aborted) break; + if(chunk.usage) usage = chunk.usage; + const choice = chunk.choices?.[0]; + if(choice?.finish_reason) finishReason = choice.finish_reason; + if(choice?.delta?.content) { + msg.content += choice.delta.content; + options.stream({text: choice.delta.content}); + } + if(choice?.delta?.tool_calls) { + for(const deltaTC of choice.delta.tool_calls) { + const index = deltaTC.index ?? msg.tool_calls.length; + let existing = msg.tool_calls.find((tc: any) => tc.index === index); + if(!existing) { + existing = {index, id: '', function: {name: '', arguments: ''}}; + msg.tool_calls.push(existing); + } if(deltaTC.id) existing.id = deltaTC.id; if(deltaTC.function?.name) existing.function.name = deltaTC.function.name; if(deltaTC.function?.arguments) existing.function.arguments += deltaTC.function.arguments; - } else { - msg.tool_calls.push({ - index: deltaTC.index, - id: deltaTC.id || '', - function: {name: deltaTC.function?.name || '', arguments: deltaTC.function?.arguments || ''} - }); } } } + streamCompleted = true; + } catch(err) { + if(!controller.signal.aborted) throw err; } + if(streamCompleted && !finishReason) finishReason = msg.tool_calls.length ? 'tool_calls' : 'stop'; } else { usage = resp.usage; finishReason = resp.choices[0].finish_reason; msg = resp.choices[0].message; } + const duration = Date.now() - callStart; - const tps = usage?.completion_tokens && duration > 0 ? usage.completion_tokens / (duration / 1000) : 0; + const tps = usage?.completion_tokens && duration > 0 + ? usage.completion_tokens / (duration / 1000) + : 0; if(finishReason === 'length' && !controller.signal.aborted) { if(msg.content?.trim()) history.push({role: 'assistant', content: msg.content.trim(), timestamp: Date.now(), duration, tps}); @@ -141,13 +174,12 @@ export class OpenAi extends LLMProvider { } if(!finishReason && !controller.signal.aborted) { - throw new Error('[OpenAI] Stream ended prematurely - connection likely dropped'); + throw new Error('[OpenAI] Completion ended without a usable response'); } const toolCalls = msg.tool_calls || []; if(toolCalls.length && !controller.signal.aborted) { if(msg.content?.trim()) history.push({role: 'assistant', content: msg.content.trim(), timestamp: Date.now(), duration, tps}); - const entries = toolCalls.map((tc: any) => { const entry: any = {role: 'tool', id: tc.id, name: tc.function.name, args: JSONAttemptParse(tc.function.arguments, {}), content: undefined, timestamp: Date.now()}; history.push(entry); @@ -157,7 +189,7 @@ export class OpenAi extends LLMProvider { await Promise.all(entries.map(async ({tc, entry}: any) => { const tool = tools.find(findByProp('name', tc.function.name)); if(options.stream) options.stream({tool: tc.function.name}); - if(!tool) { entry.error = 'Tool not found'; return; } + if(!tool) return entry.error = 'Tool not found'; try { const toolStream = options.stream && ((chunk: any) => { if(chunk.done) return; @@ -177,7 +209,6 @@ export class OpenAi extends LLMProvider { } while(!terminal && !controller.signal.aborted); if(options.stream) options.stream({done: true}); - const turnStart = history.map(h => h.role).lastIndexOf('user'); const finalContent = history.slice(turnStart + 1).reduce((str, h) => h.role === 'assistant' ? str + (h.content || '') : str, '').trim(); res(options.schema ? JSONAttemptParse(finalContent, finalContent) : finalContent);