Fixed delegate agent history... again
This commit is contained in:
+92
-141
@@ -25,73 +25,38 @@ export class OpenAi extends LLMProvider {
|
||||
return client;
|
||||
}
|
||||
|
||||
private toStandard(history: any[]): LLMMessage[] {
|
||||
for(let i = 0; i < history.length; i++) {
|
||||
const h = history[i];
|
||||
if(h.role === 'assistant' && h.tool_calls) {
|
||||
const items: any[] = [];
|
||||
if(h.content) items.push({role: 'assistant', content: h.content, timestamp: h.timestamp, duration: h.duration, tps: h.tps});
|
||||
items.push(...h.tool_calls.map((tc: any) => ({
|
||||
role: 'tool',
|
||||
id: tc.id,
|
||||
name: tc.function.name,
|
||||
args: JSONAttemptParse(tc.function.arguments, {}),
|
||||
timestamp: h.timestamp,
|
||||
duration: h.duration,
|
||||
tps: h.tps
|
||||
})));
|
||||
history.splice(i, 1, ...items);
|
||||
i += items.length - 1;
|
||||
} else if(h.role === 'tool') {
|
||||
const record = history.find(h2 => h.tool_call_id == h2.id);
|
||||
if(record) {
|
||||
if(h.content?.includes('"error":')) record.error = h.content;
|
||||
else record.content = h.content || '';
|
||||
}
|
||||
history.splice(i, 1);
|
||||
i--;
|
||||
}
|
||||
if(!history[i]?.timestamp) history[i].timestamp = Date.now();
|
||||
}
|
||||
return history;
|
||||
}
|
||||
|
||||
private fromStandard(history: LLMMessage[]): any[] {
|
||||
return history.reduce((result, h) => {
|
||||
/** Convert standard history -> OpenAI wire format */
|
||||
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') {
|
||||
result.push({
|
||||
wire.push({
|
||||
role: 'assistant',
|
||||
content: null,
|
||||
tool_calls: [{ id: h.id, type: 'function', function: { name: h.name, arguments: JSON.stringify(h.args) } }],
|
||||
refusal: null,
|
||||
annotations: [],
|
||||
timestamp: h.timestamp,
|
||||
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,
|
||||
timestamp: h.timestamp,
|
||||
content: h.error || h.content || '',
|
||||
});
|
||||
} else {
|
||||
result.push(h);
|
||||
wire.push({role: h.role, content: h.content});
|
||||
}
|
||||
return result;
|
||||
}, [] as any[]);
|
||||
}
|
||||
return wire;
|
||||
}
|
||||
|
||||
ask(message: string, options: LLMRequest = {}): AbortablePromise<string | any> {
|
||||
const controller = new AbortController();
|
||||
return Object.assign(new Promise<any>(async (res, rej) => {
|
||||
const base = (options.history || []).filter(h => h.role !== 'system');
|
||||
let history = this.fromStandard([
|
||||
...(options.system ? [{role: <any>'system', content: options.system, timestamp: Date.now()}] : []),
|
||||
...base,
|
||||
{role: 'user', content: message, timestamp: Date.now()}
|
||||
]);
|
||||
if(!options.history) options.history = [];
|
||||
const history = options.history;
|
||||
if(message) history.push({role: 'user', content: message, timestamp: Date.now()});
|
||||
|
||||
const tools = options.tools || this.ai.options.llm?.tools || [];
|
||||
const requestParams: any = {
|
||||
model: options.model || this.model,
|
||||
messages: history,
|
||||
stream: !!options.stream,
|
||||
max_completion_tokens: options.max_tokens || this.ai.options.llm?.max_tokens || undefined,
|
||||
temperature: options.temperature || this.ai.options.llm?.temperature || undefined,
|
||||
@@ -111,108 +76,94 @@ 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};
|
||||
let resp: any, terminal = false, duration = 0, tps = 0;
|
||||
do {
|
||||
requestParams.messages = history.map(({timestamp, ...m}) => m);
|
||||
const callStart = Date.now();
|
||||
resp = await this.tokenPool.run(token => this.getClient(token).chat.completions.create(requestParams)).catch(err => {
|
||||
err.message += `\n\nMessages:\n${JSON.stringify(history, null, 2)}`;
|
||||
throw err;
|
||||
});
|
||||
|
||||
let usage: any;
|
||||
if(options.stream) {
|
||||
resp.choices = [{message: {role: 'assistant', content: '', tool_calls: [], timestamp: Date.now()}}];
|
||||
for await (const chunk of resp) {
|
||||
if(controller.signal.aborted) break;
|
||||
if(chunk.usage) usage = chunk.usage;
|
||||
if(chunk.choices[0]?.delta?.content) {
|
||||
resp.choices[0].message.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 = resp.choices[0].message.tool_calls.find(tc => tc.index === deltaTC.index);
|
||||
if(existing) {
|
||||
if(deltaTC.id) existing.id = deltaTC.id;
|
||||
if(deltaTC.type) existing.type = deltaTC.type;
|
||||
if(deltaTC.function) {
|
||||
if(!existing.function) existing.function = {};
|
||||
if(deltaTC.function.name) existing.function.name = deltaTC.function.name;
|
||||
if(deltaTC.function.arguments) existing.function.arguments = (existing.function.arguments || '') + deltaTC.function.arguments;
|
||||
try {
|
||||
let terminal = false;
|
||||
do {
|
||||
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 => {
|
||||
err.message += `\n\nMessages:\n${JSON.stringify(requestParams.messages, null, 2)}`;
|
||||
throw err;
|
||||
});
|
||||
|
||||
let usage: any, 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]?.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 = 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 {
|
||||
msg.tool_calls.push({
|
||||
index: deltaTC.index,
|
||||
id: deltaTC.id || '',
|
||||
function: {name: deltaTC.function?.name || '', arguments: deltaTC.function?.arguments || ''}
|
||||
});
|
||||
}
|
||||
} else {
|
||||
resp.choices[0].message.tool_calls.push({
|
||||
index: deltaTC.index,
|
||||
id: deltaTC.id || '',
|
||||
type: deltaTC.type || 'function',
|
||||
function: {
|
||||
name: deltaTC.function?.name || '',
|
||||
arguments: deltaTC.function?.arguments || ''
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
usage = resp.usage;
|
||||
msg = resp.choices[0].message;
|
||||
}
|
||||
} else {
|
||||
usage = resp.usage;
|
||||
}
|
||||
duration = Date.now() - callStart;
|
||||
tps = usage?.completion_tokens && duration > 0 ? usage.completion_tokens / (duration / 1000) : 0;
|
||||
const duration = Date.now() - callStart;
|
||||
const tps = usage?.completion_tokens && duration > 0 ? usage.completion_tokens / (duration / 1000) : 0;
|
||||
|
||||
if(resp.error) throw new Error(resp.error);
|
||||
const toolCalls = resp.choices[0].message.tool_calls || [];
|
||||
if(toolCalls.length && !controller.signal.aborted) {
|
||||
history.push({...resp.choices[0].message, duration, tps});
|
||||
const results = await Promise.all(toolCalls.map(async (toolCall: any) => {
|
||||
const tool = tools?.find(findByProp('name', toolCall.function.name));
|
||||
if(options.stream) options.stream({tool: toolCall.function.name});
|
||||
if(!tool) return {role: 'tool', tool_call_id: toolCall.id, content: '{"error": "Tool not found"}', timestamp: Date.now()};
|
||||
try {
|
||||
const args = JSONAttemptParse(toolCall.function.arguments, {});
|
||||
const toolStream = options.stream && ((chunk: any) => {
|
||||
if(chunk.done) { terminal = true; return; }
|
||||
options.stream!(chunk);
|
||||
});
|
||||
const result = await tool.fn(args, toolStream, this.ai, toolCall.id);
|
||||
return {role: 'tool', tool_call_id: toolCall.id, content: typeof result == 'object' ? JSONSanitize(result) : result, timestamp: Date.now()};
|
||||
} catch (err: any) {
|
||||
return {role: 'tool', tool_call_id: toolCall.id, content: JSONSanitize({error: err?.message || err?.toString() || 'Unknown'}), timestamp: Date.now()};
|
||||
}
|
||||
}));
|
||||
history.push(...results);
|
||||
requestParams.messages = history;
|
||||
}
|
||||
} while (!terminal && !controller.signal.aborted && resp.choices?.[0]?.message?.tool_calls?.length);
|
||||
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});
|
||||
|
||||
if(!terminal) {
|
||||
const textContent = resp.choices[0].message.content || '';
|
||||
history.push({role: 'assistant', content: textContent.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});
|
||||
}
|
||||
} 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);
|
||||
} catch(err) {
|
||||
rej(err);
|
||||
}
|
||||
|
||||
history = this.toStandard(history);
|
||||
if(options.history) options.history.splice(0, options.history.length, ...history.filter(h => h.role !== 'system'));
|
||||
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) => {
|
||||
if(h.role === 'assistant') return str + (h.content || '');
|
||||
return str;
|
||||
}, '').trim();
|
||||
|
||||
res(options.schema ? JSONAttemptParse(finalContent, finalContent) : finalContent);
|
||||
}), {abort: () => controller.abort()});
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user