diff --git a/package.json b/package.json index dd7dbfb..7ffd7c9 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@ztimson/ai-utils", - "version": "1.4.1", + "version": "1.4.2", "description": "AI Utility library", "author": "Zak Timson", "license": "MIT", diff --git a/src/antrhopic.ts b/src/antrhopic.ts index 7921585..40e2777 100644 --- a/src/antrhopic.ts +++ b/src/antrhopic.ts @@ -24,49 +24,29 @@ export class Anthropic extends LLMProvider { return client; } - private toStandard(history: any[]): LLMMessage[] { - const timestamp = Date.now(); - const messages: LLMMessage[] = []; - for(let h of history) { - if(typeof h.content == 'string') { - messages.push({timestamp, ...h}); - } else { - const textContent = h.content?.filter((c: any) => c.type == 'text').map((c: any) => c.text).join('\n\n'); - if(textContent) messages.push({role: h.role, content: textContent, timestamp: timestamp, duration: h.duration, tps: h.tps}); - h.content.forEach((c: any) => { - if(c.type == 'tool_use') { - messages.push({role: 'tool', id: c.id, name: c.name, args: c.input, timestamp: h.timestamp, content: undefined, duration: h.duration, tps: h.tps}); - } else if(c.type == 'tool_result') { - const m: any = messages.findLast(m => (m).id == c.tool_use_id); - if(m) m[c.is_error ? 'error' : 'content'] = c.content; - } - }); - } - } - return messages; - } - - private fromStandard(history: LLMMessage[]): any[] { - for(let i = 0; i < history.length; i++) { - if(history[i].role == 'tool') { - const h: any = history[i]; - history.splice(i, 1, + /** Convert standard history -> Anthropic wire format */ + private toWire(history: LLMMessage[]): any[] { + const wire: any[] = []; + for(const h of history) { + if(h.role === 'tool') { + wire.push( {role: 'assistant', content: [{type: 'tool_use', id: h.id, name: h.name, input: h.args}]}, - {role: 'user', content: [{type: 'tool_result', tool_use_id: h.id, is_error: !!h.error, content: h.error || h.content}]} - ) - i++; + {role: 'user', content: [{type: 'tool_result', tool_use_id: h.id, is_error: !!h.error, content: h.error || h.content || ''}]} + ); + } else { + wire.push({role: h.role, content: h.content}); } } - return history; + return wire; } ask(message: string, options: LLMRequest = {}): AbortablePromise { const controller = new AbortController(); - return Object.assign(new Promise(async (res) => { - let history = this.fromStandard([ - ...(options.history || []).filter(h => h.role !== 'system'), - {role: 'user', content: message, timestamp: Date.now()} - ]); + return Object.assign(new Promise(async (res, rej) => { + 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, @@ -80,105 +60,97 @@ export class Anthropic extends LLMProvider { 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]) : [] - }, - fn: undefined + } })), - messages: history, stream: !!options.stream, }; - // Add structured output support if(options.schema) { - requestParams.output_config = { - format: { - type: 'json_schema', - schema: convertSchema(options.schema) - } - }; + requestParams.output_config = {format: {type: 'json_schema', schema: convertSchema(options.schema)}}; } - 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).messages.create(requestParams)).catch(err => { - err.message += `\n\nMessages:\n${JSON.stringify(history, null, 2)}`; - throw err; - }); + try { + let terminal = false; + do { + requestParams.messages = this.toWire(history.filter(h => h.role !== 'system')); - let usage: any; - if(options.stream) { - resp.content = []; - for await (const chunk of resp) { - if(controller.signal.aborted) break; - if(chunk.type === 'content_block_start') { - if(chunk.content_block.type === 'text') { - resp.content.push({type: 'text', text: ''}); - } else if(chunk.content_block.type === 'tool_use') { - resp.content.push({type: 'tool_use', id: chunk.content_block.id, name: chunk.content_block.name, input: ''}); + const callStart = Date.now(); + const resp: any = await this.tokenPool.run(token => this.getClient(token).messages.create(requestParams)).catch(err => { + err.message += `\n\nMessages:\n${JSON.stringify(requestParams.messages, null, 2)}`; + throw err; + }); + + let usage: any, content: any[] = []; + if(options.stream) { + for await (const chunk of resp) { + if(controller.signal.aborted) break; + if(chunk.type === 'content_block_start') { + if(chunk.content_block.type === 'text') content.push({type: 'text', text: ''}); + else if(chunk.content_block.type === 'tool_use') content.push({type: 'tool_use', id: chunk.content_block.id, name: chunk.content_block.name, input: ''}); + } else if(chunk.type === 'content_block_delta') { + if(chunk.delta.type === 'text_delta') { + content.at(-1).text += chunk.delta.text; + options.stream({text: chunk.delta.text}); + } else if(chunk.delta.type === 'input_json_delta') { + content.at(-1).input += chunk.delta.partial_json; + } + } else if(chunk.type === 'content_block_stop') { + const last = content.at(-1); + if(last?.type === 'tool_use') last.input = last.input ? JSONAttemptParse(last.input, {}) : {}; + } else if(chunk.type === 'message_delta') { + if(chunk.usage) usage = chunk.usage; + } else if(chunk.type === 'message_stop') { + break; } - } else if(chunk.type === 'content_block_delta') { - if(chunk.delta.type === 'text_delta') { - const text = chunk.delta.text; - resp.content.at(-1).text += text; - options.stream({text}); - } else if(chunk.delta.type === 'input_json_delta') { - resp.content.at(-1).input += chunk.delta.partial_json; - } - } else if(chunk.type === 'content_block_stop') { - const last = resp.content.at(-1); - if(last?.input != null) last.input = last.input ? JSONAttemptParse(last.input, {}) : {}; - } else if(chunk.type === 'message_delta') { - if(chunk.usage) usage = chunk.usage; - } else if(chunk.type === 'message_stop') { - break; } + } else { + usage = resp.usage; + content = resp.content; } - } else { - usage = resp.usage; - } - duration = Date.now() - callStart; - tps = usage?.output_tokens && duration > 0 ? usage.output_tokens / (duration / 1000) : 0; + const duration = Date.now() - callStart; + const tps = usage?.output_tokens && duration > 0 ? usage.output_tokens / (duration / 1000) : 0; - const toolCalls = resp.content.filter((c: any) => c.type === 'tool_use'); - if(toolCalls.length && !controller.signal.aborted) { - history.push({role: 'assistant', content: resp.content, timestamp: Date.now(), duration, tps}); - const results = await Promise.all(toolCalls.map(async (toolCall: any) => { - const tool = tools.find(findByProp('name', toolCall.name)); - if(options.stream) options.stream({tool: toolCall.name}); - if(!tool) return {tool_use_id: toolCall.id, is_error: true, content: 'Tool not found'}; - try { - const toolStream = options.stream && ((chunk: any) => { - if(chunk.done) { terminal = true; return; } - options.stream!(chunk); - }); - const result = await tool.fn(toolCall.input, toolStream, this.ai, toolCall.id); - return {type: 'tool_result', tool_use_id: toolCall.id, content: typeof result == 'object' ? JSONSanitize(result) : result}; - } catch (err: any) { - return {type: 'tool_result', tool_use_id: toolCall.id, is_error: true, content: err?.message || err?.toString() || 'Unknown'}; - } - })); - history.push({role: 'user', content: results, timestamp: Date.now()}); - requestParams.messages = history; - } - } while (!terminal && !controller.signal.aborted && resp.content.some((c: any) => c.type === 'tool_use')); + const toolCalls = content.filter((c: any) => c.type === 'tool_use'); + if(toolCalls.length && !controller.signal.aborted) { + const text = content.filter((c: any) => c.type === 'text').map((c: any) => c.text).join('\n\n').trim(); + if(text) history.push({role: 'assistant', content: text, timestamp: Date.now(), duration, tps}); - if(!terminal) { - const textContent = resp.content.filter((c: any) => c.type == 'text').map((c: any) => c.text).join('\n\n'); - 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.name, args: tc.input, 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.name)); + if(options.stream) options.stream({tool: tc.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 = content.filter((c: any) => c.type === 'text').map((c: any) => c.text).join('\n\n').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); - 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()}); } } diff --git a/src/llm.ts b/src/llm.ts index 48cd049..ed5e8ac 100644 --- a/src/llm.ts +++ b/src/llm.ts @@ -34,6 +34,10 @@ export type LLMMessage = { content: string | any; /** Timestamp */ timestamp?: number; + /** Response duration in ms */ + duration?: number; + /** Tokens per second */ + tps?: number; } | { /** Tool call */ role: 'tool'; @@ -135,12 +139,15 @@ class LLM { fn: async (args: any, stream: any, ai: any, id?: string) => { if(depth >= MAX_AGENT_DEPTH) return 'Max agent delegation depth exceeded'; - // Opt-in only, self always excluded regardless of whitelist const nested = (a.agents || []) .map(name => allAgents.find(x => x.name === name)) .filter((x): x is Agent => !!x && x.name !== a.name); - const request = this.ask(a.delegate ? '' : `${args.instructions}${args.context ? `\n\n${args.context}` : ''}`, { + // Delegate continues the SAME live conversation - no new user turn needed, + // `history` is always current (shared, mutated in place) by the time this runs + const q = a.delegate ? '' : `${args.instructions}${args.context ? `\n\n${args.context}` : ''}`; + + const request = this.ask(q, { system: `You are a specialized subagent. ${a.delegate ? 'Your output streams directly to the user for the remainder of this turn. You are mid conversation - dispense with greetings.' : 'You are wrapped in a tool call that will be analysis by an LLM - dispense with conversation'} As a subagent, focus on executing your task completely using available tools and returning only the final result - no commentary, questions, or dialogue. @@ -265,7 +272,10 @@ ${a.system}`, promise = (async () => { let tools: AiTool[] = options.tools || this.ai.options.llm?.tools || []; const prompts: string[] = []; + // `history` is the single source of truth from here on - mutated in place by + // this call AND by any nested/delegated agent calls sharing the same array let history = options.history || []; + if(message) history.push({role: 'user', content: message, timestamp: Date.now()}); // MCP const mcp = options.mcp || this.ai.options?.llm?.mcp; @@ -334,7 +344,9 @@ Also relevant but not preloaded (use \`memory_recall\`): ${listed.map(r => r.nam if(aborted) throw Object.assign(new Error('Aborted'), {name: 'AbortError'}); prompts.unshift(options.system || this.ai.options.llm?.system || ''); - request = this.models[m].ask(message, {...options, tools, system: prompts.filter(Boolean).join('\n\n')}); + // Message already appended to shared `history` above - pass '' so the provider + // doesn't push a duplicate user turn + request = this.models[m].ask('', {...options, tools, system: prompts.filter(Boolean).join('\n\n')}); let resp = await request; // Capture meta (duration / tps) diff --git a/src/open-ai.ts b/src/open-ai.ts index b0fb237..7aab26d 100644 --- a/src/open-ai.ts +++ b/src/open-ai.ts @@ -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 { const controller = new AbortController(); return Object.assign(new Promise(async (res, rej) => { - const base = (options.history || []).filter(h => h.role !== 'system'); - let history = this.fromStandard([ - ...(options.system ? [{role: '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()}); } }