Fix streaming + tool call interleaving: properly accumulate tool_call deltas by index, execute completed tools immediately, and detect termination via finish_reason instead of missing chunk.done flag.
This commit is contained in:
+111
-39
@@ -41,7 +41,7 @@ export class OpenAi extends LLMProvider {
|
||||
wire.push({
|
||||
role: 'assistant',
|
||||
content: null,
|
||||
tool_calls: [{id: h.id, type: 'function', function: {name: h.name, arguments: JSON.stringify(h.args)}}],
|
||||
tool_calls: [{id: h.id, type: 'function', function: {name: h.name, arguments: JSON.stringify(h.args)}],
|
||||
}, {
|
||||
role: 'tool',
|
||||
tool_call_id: h.id,
|
||||
@@ -100,70 +100,142 @@ export class OpenAi extends LLMProvider {
|
||||
|
||||
let usage: any, msg: any = {content: '', tool_calls: []};
|
||||
if(options.stream) {
|
||||
const toolCallState: Record<number, any> = {};
|
||||
let pendingToolCalls = 0;
|
||||
|
||||
for await (const chunk of resp) {
|
||||
if(controller.signal.aborted) break;
|
||||
|
||||
if(chunk.usage) usage = chunk.usage;
|
||||
|
||||
// Handle content deltas
|
||||
if(chunk.choices[0]?.delta?.content) {
|
||||
msg.content += chunk.choices[0].delta.content;
|
||||
options.stream({text: chunk.choices[0].delta.content});
|
||||
options.stream?.({text: chunk.choices[0].delta.content});
|
||||
}
|
||||
|
||||
// Handle tool_call deltas
|
||||
if(chunk.choices[0]?.delta?.tool_calls) {
|
||||
for(const deltaTC of chunk.choices[0].delta.tool_calls) {
|
||||
const existing = msg.tool_calls.find((tc: any) => tc.index === deltaTC.index);
|
||||
if(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 {
|
||||
const tc = toolCallState[deltaTC.index];
|
||||
|
||||
if(!tc && deltaTC.id) {
|
||||
// New tool call delta
|
||||
toolCallState[deltaTC.index] = {
|
||||
id: deltaTC.id,
|
||||
name: deltaTC.function?.name || '',
|
||||
arguments: deltaTC.function?.arguments || '',
|
||||
complete: false
|
||||
};
|
||||
msg.tool_calls.push({
|
||||
index: deltaTC.index,
|
||||
id: deltaTC.id || '',
|
||||
function: {name: deltaTC.function?.name || '', arguments: deltaTC.function?.arguments || ''}
|
||||
});
|
||||
pendingToolCalls++;
|
||||
} else if(tc && deltaTC.function?.name) {
|
||||
// Update existing tool call
|
||||
if(deltaTC.id) tc.id = deltaTC.id;
|
||||
if(deltaTC.function.name) tc.name = deltaTC.function.name;
|
||||
if(deltaTC.function.arguments) tc.arguments += deltaTC.function.arguments;
|
||||
}
|
||||
|
||||
if(tc && deltaTC.function?.arguments && !tc.complete) {
|
||||
tc.complete = true;
|
||||
pendingToolCalls--;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Execute completed tools immediately when all deltas are received
|
||||
if(!pendingToolCalls && msg.tool_calls.length > 0) {
|
||||
const completedTools: any[] = [];
|
||||
|
||||
for(const tc of msg.tool_calls) {
|
||||
const tcState = toolCallState[tc.index];
|
||||
if(!tcState) continue;
|
||||
|
||||
const entry: any = {role: 'tool', id: tc.id, name: tcState.name, args: JSONAttemptParse(tcState.arguments, {}), content: undefined, timestamp: Date.now()};
|
||||
history.push(entry);
|
||||
completedTools.push({tc, entry});
|
||||
|
||||
const tool = tools.find(findByProp('name', tcState.name));
|
||||
if(options.stream) options.stream?.({tool: tcState.name});
|
||||
|
||||
if(!tool) {
|
||||
entry.error = 'Tool not found';
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
const toolStream = options.stream && ((chunk: any) => {
|
||||
if(chunk.done) { terminal = true; return; }
|
||||
options.stream?.(chunk);
|
||||
});
|
||||
const result = await tool.fn(entry.args, toolStream, this.ai, tc.id);
|
||||
entry.content = typeof result === 'object' ? JSONSanitize(result) : result;
|
||||
} catch(err: any) {
|
||||
entry.error = err?.message || err?.toString() || 'Unknown';
|
||||
}
|
||||
}
|
||||
|
||||
if(options.stream) options.stream?.({done: true});
|
||||
|
||||
for(const {entry} of completedTools) {
|
||||
if(entry.error) {
|
||||
options.stream?.({error: entry.error});
|
||||
} else if(entry.content) {
|
||||
options.stream?.({toolResult: true});
|
||||
}
|
||||
}
|
||||
|
||||
msg.tool_calls = []; // Clear after execution
|
||||
}
|
||||
}
|
||||
|
||||
// Handle finish_reason for proper termination detection
|
||||
if(chunk.choices[0]?.finish_reason) {
|
||||
const finishReason = chunk.choices[0].finish_reason;
|
||||
if(finishReason === 'stop' || finishReason === 'tool_calls' || finishReason === 'length') {
|
||||
terminal = true;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
usage = resp.usage;
|
||||
msg = resp.choices[0].message;
|
||||
if(msg.tool_calls) {
|
||||
// Non-streaming: execute tools immediately
|
||||
for(const tc of msg.tool_calls) {
|
||||
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);
|
||||
|
||||
const tool = tools.find(findByProp('name', tc.function.name));
|
||||
if(!tool) {
|
||||
entry.error = 'Tool not found';
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await tool.fn(entry.args, undefined, this.ai, tc.id);
|
||||
entry.content = typeof result === 'object' ? JSONSanitize(result) : result;
|
||||
} catch(err: any) {
|
||||
entry.error = err?.message || err?.toString() || 'Unknown';
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const duration = Date.now() - callStart;
|
||||
const tps = usage?.completion_tokens && duration > 0 ? usage.completion_tokens / (duration / 1000) : 0;
|
||||
|
||||
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);
|
||||
return {tc, entry};
|
||||
});
|
||||
|
||||
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; }
|
||||
try {
|
||||
const toolStream = options.stream && ((chunk: any) => {
|
||||
if(chunk.done) { terminal = true; return; }
|
||||
options.stream!(chunk);
|
||||
});
|
||||
const result = await tool.fn(entry.args, toolStream, this.ai, tc.id);
|
||||
entry.content = typeof result === 'object' ? JSONSanitize(result) : result;
|
||||
} catch(err: any) {
|
||||
entry.error = err?.message || err?.toString() || 'Unknown';
|
||||
}
|
||||
}));
|
||||
} else {
|
||||
terminal = true;
|
||||
const text = (msg.content || '').trim();
|
||||
if(text) history.push({role: 'assistant', content: text, timestamp: Date.now(), duration, tps});
|
||||
// Capture assistant messages (before or after tools)
|
||||
if(msg.content?.trim()) {
|
||||
history.push({role: 'assistant', content: msg.content.trim(), timestamp: Date.now(), duration, tps});
|
||||
}
|
||||
|
||||
} while(!terminal && !controller.signal.aborted);
|
||||
|
||||
if(options.stream) options.stream({done: true});
|
||||
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();
|
||||
@@ -173,4 +245,4 @@ export class OpenAi extends LLMProvider {
|
||||
}
|
||||
}), {abort: () => controller.abort()});
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user