import {snakeCase} from '@ztimson/utils'; import {AbortablePromise, Ai} from './ai.ts'; import {Anthropic} from './antrhopic.ts'; import {OpenAi} from './open-ai.ts'; import {LLMProvider} from './provider.ts'; import {AiTool, AiToolArg} from './tools.ts'; import {fileURLToPath} from 'url'; import {dirname, join} from 'path'; import {spawn} from 'node:child_process'; import {Memory, MemoryCache, MemoryManager, MemoryOptions} from './memory.ts'; export type AnthropicConfig = {proto: 'anthropic', token: string}; export type OpenAiConfig = {proto: 'openai', host?: string, token: string}; export type Agent = { name: string; description?: string; model?: string | null; temperature?: number; system: string; delegate?: boolean; skills?: Skill[] | null; tools?: AiTool[] | null; mcp?: McpServer[] | null; agents?: string[] | null; } export type LLMMessage = { /** Message originator */ role: 'assistant' | 'system' | 'user'; /** Message content */ content: string | any; /** Timestamp */ timestamp?: number; } | { /** Tool call */ role: 'tool'; /** Unique ID for call */ id: string; /** Tool that was run */ name: string; /** Tool arguments */ args: any; /** Tool result */ content: undefined | string; /** Tool error */ error?: undefined | string; /** Timestamp */ timestamp?: number; /** Response duration in ms */ duration?: number; /** Tokens per second */ tps?: number; } export type LLMRequest = { /** Return a parsed JSON object that matches the schema */ schema?: AiToolArg; /** System prompt */ system?: string; /** Message history */ history?: LLMMessage[]; /** Max tokens for request */ max_tokens?: number; /** 0 = Rigid Logic, 1 = Balanced, 2 = Hyper Creative **/ temperature?: number; /** Available tools */ tools?: AiTool[]; /** LLM model */ model?: string; /** Stream response */ stream?: (chunk: {text?: string, tool?: string, done?: true}) => any; /** Compress old messages in the chat to free up context */ compress?: {max: number; min: number}; /** User's memory documents - RAG injected automatically each turn */ memory?: Memory[] | MemoryCache | MemoryOptions; /** Model to use for memory operations */ memoryModel?: string; /** Skill documents the AI can browse and read on demand */ skills?: Skill[]; /** MCP servers to connect and expose as tools */ mcp?: McpServer[]; /** Subagents exposed as delegatable/wrapped tools */ agents?: Agent[]; /** @internal recursion guard for nested agent delegation */ _agentDepth?: number; } export type McpServer = { /** MCP server name for humans */ name: string; /** Host URL */ host: string; /** Server access token */ token?: string; } export type Skill = { /** Name of skill for humans */ name: string; /** Description LLM will use to decide to learn a skill */ description: string; /** Skill instructions */ content: string; } const MAX_AGENT_DEPTH = 5; class LLM { private memoryManager!: MemoryManager; defaultModel!: string; models: {[model: string]: LLMProvider} = {}; constructor(public readonly ai: Ai) { if(!ai.options.llm?.models) return; Object.entries(ai.options.llm.models).forEach(([model, config]) => { if(!this.defaultModel) this.defaultModel = model; if(config.proto == 'anthropic') this.models[model] = new Anthropic(this.ai, config.token, model); else if(config.proto == 'openai') this.models[model] = new OpenAi(this.ai, config.host || null, config.token, model); }); this.memoryManager = new MemoryManager(this); } private setupAgent(agents: Agent[] = [], allAgents: Agent[], pending: Map, aborts: (() => void)[], depth = 0): AiTool[] { return agents.map(a => { const toolName = `${a.delegate ? '' : 'sub'}agent_${snakeCase(a.name)}`; return { name: toolName, description: `${a.delegate ? 'Delegate to ' : ''}Subagent: ${a.description || a.name}`, args: { context: {type: 'string', description: 'Summary of related messages, samples, files, etc...', required: true}, instructions: {type: 'string', description: 'Detailed instructions for subagent to complete', required: true}, }, fn: async (args: any, stream: any, ai: any, id?: string) => { if(depth >= MAX_AGENT_DEPTH) return 'Max agent delegation depth exceeded'; const subHistory: LLMMessage[] = []; // 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 start = Date.now(); const request = this.ask(`${args.instructions}${args.context ? `\n\n${args.context}` : ''}`, { system: `You are a specialized subagent. ${a.delegate ? 'Your output streams directly to the user for the remainder of this turn.' : 'You are wrapped in a tool call that will be analysis by an LLM'} As a subagent, focus on executing your task completely using available tools and returning only the final result - no commentary, questions, or dialogue. ${a.system}`, model: a.model || undefined, temperature: a.temperature, stream: a.delegate ? stream : undefined, history: subHistory, mcp: a.mcp || undefined, skills: a.skills || undefined, tools: a.tools || undefined, agents: nested, _agentDepth: depth + 1, } as any); aborts.push(request.abort); const resp = await request; const duration = Date.now() - start; const assistantTurns = subHistory.filter((h: any) => h.role === 'assistant' && h.duration); const genTime = assistantTurns.reduce((s, h: any) => s + h.duration, 0); const genTokens = assistantTurns.reduce((s, h: any) => s + (h.tps || 0) * (h.duration / 1000), 0); const tps = genTime > 0 ? genTokens / (genTime / 1000) : 0; if(a.delegate) { pending.set(id, {resp, subHistory, duration, tps}); return ''; } return resp; } }; }); } private async setupMcp(servers: McpServer[] = []): Promise<{prompt: string, tools: AiTool[]}> { if(!servers?.length) return {prompt: '', tools: []}; const allTools: AiTool[] = []; await Promise.all(servers.map(async server => { const res = await fetch(`${server.host}/tools`, {headers: server.token ? {Authorization: `Bearer ${server.token}`} : {}}); const mcp: any = await res.json(); if(!mcp?.tools) return; for(const t of mcp.tools) { const args: Record = {}; if(t.inputSchema?.properties) { for(const [key, val] of Object.entries(t.inputSchema.properties)) { args[key] = {type: val.type || 'string', description: val.description || '', required: t.inputSchema.required?.includes(key)}; } } allTools.push({ name: `${server.name}_${t.name}`, description: t.description || '', args, fn: async (a: any) => { const r = await fetch(`${server.host}/tools/call`, { method: 'POST', headers: {'Content-Type': 'application/json', ...(server.token ? {Authorization: `Bearer ${server.token}`} : {})}, body: JSON.stringify({name: t.name, arguments: a}) }); const data: any = await r.json(); return data?.content?.[0]?.text ?? JSON.stringify(data); } }); } })); const list = allTools.map(t => `- ${t.name}: ${t.description}`).join('\n'); return { prompt: `You have access to the following MCP tools:\n${list}`, tools: allTools }; } private setupSkills(skills: Skill[] = []): {prompt: string, tools: AiTool[]} { if(!skills?.length) return {prompt: '', tools: []}; const list = skills.map(s => `- ${s.name}: ${s.description}`).join('\n'); return { prompt: `You have access to the following skill documents, use \`read_skill\` to access them:\n${list}`, tools: [{ name: 'skill_read', description: 'Read the full content of a skill/knowledge document', args: { name: {type: 'string', description: 'Exact skill name', required: true} }, fn: (args: any) => { const skill = skills.find(s => s.name === args.name); if(!skill) return `Skill not found. Available:\n${list}`; return `# ${skill.name}\n${skill.content}`; } }] } } private wrapToolTiming(tools: AiTool[], timings: Map): AiTool[] { return tools.map(t => ({ ...t, fn: async (args: any, stream: any, ai: any, id?: string) => { const start = Date.now(); const result = await t.fn(args, stream, ai, id); const duration = Date.now() - start; const tps = duration > 0 ? this.estimateTokens(result) / (duration / 1000) : 0; if(id) timings.set(id, {duration, tps}); return result; } })); } ask(message: string, options: LLMRequest = {}): AbortablePromise { options = { system: '', ...this.ai.options.llm, models: undefined, history: [], ...options, } const m = options.model || this.defaultModel; if(!this.models[m]) throw new Error(`Model does not exist: ${m}`); let request: AbortablePromise | null = null; let aborted = false; const nestedAborts: (() => void)[] = []; const abort = () => { aborted = true; request?.abort?.(); nestedAborts.forEach(a => a()); }; const promise = (async () => { let tools: AiTool[] = options.tools || this.ai.options.llm?.tools || []; const prompts: string[] = []; let history = options.history || []; // MCP const mcp = options.mcp || this.ai.options?.llm?.mcp; if(mcp?.length) { const m = await this.setupMcp(mcp); prompts.unshift(m.prompt); tools.push(...m.tools); } // Skills const skills = options.skills || this.ai.options?.llm?.skills; if(skills?.length) { const s = this.setupSkills(skills); prompts.unshift(s.prompt); tools.push(...s.tools); } // Agents const agents = options.agents || this.ai.options?.llm?.agents; const pendingDelegates = new Map(); if(agents?.length) tools.push(...this.setupAgent(agents, agents, pendingDelegates, nestedAborts, options._agentDepth || 0)); // Memory const mem = MemoryManager.normalize(options.memory); if(mem) { const mems = mem.memory instanceof MemoryCache ? mem.memory.memories : mem.memory; if(mems.length) { if(mem.inject) { const pool = 15; // candidates considered, cheap since only refs are listed const budget = mem.maxTokens ?? 2000; // actual content injected const relevant = await this.memoryManager.recollect(message, mem.memory, pool); let used = 0; const preloaded: typeof relevant = []; const listed: typeof relevant = []; for(const r of relevant) { const t = this.estimateTokens(r.content); if(used + t <= budget || preloaded.length === 0) { preloaded.push(r); used += t; } else listed.push(r); } prompts.unshift(`You have access to the following memory files: ${mems.map(m => `- ${m.name}: ${m.description}`).join('\n')} ${preloaded.length ? ` Relevant memories have been preloaded: ${preloaded.map(r => ` **${r.name}** ${r.description} ${r.content} `).join('\n---\n')} ` : ''}${listed.length ? ` Also relevant but not preloaded (use \`memory_recall\`): ${listed.map(r => r.name).join(', ')} ` : ''}`.trim()); } if(mem.tool) tools.push(this.memoryManager.tools.read(mem.memory)); } } if(aborted) throw Object.assign(new Error('Aborted'), {name: 'AbortError'}); // Time each tool call's real execution so its history entry gets its own duration/tps const toolTimings = new Map(); tools = this.wrapToolTiming(tools, toolTimings); prompts.unshift(options.system || this.ai.options.llm?.system || ''); request = this.models[m].ask(message, {...options, tools, system: prompts.filter(Boolean).join('\n\n')}); let resp = await request; // Providers stamp duration/tps on assistant entries themselves (from real API usage). // Overwrite tool entries with actual tool-execution timing instead of the LLM call timing. for(const h of history) { if(h.role === 'tool' && toolTimings.has(h.id)) Object.assign(h, toolTimings.get(h.id)); } // Spice delegated agents response into history let lastDelegateResp: string | null = null; if(pendingDelegates.size) { for(let i = 0; i < history.length; i++) { const h: any = history[i]; if(h.role !== 'tool' || !pendingDelegates.has(h.id)) continue; const {resp: delegateResp, subHistory, duration, tps} = pendingDelegates.get(h.id)!; pendingDelegates.delete(h.id); const insert: LLMMessage[] = [...subHistory.filter(sh => sh.role === 'tool'), {role: 'assistant', content: delegateResp, timestamp: Date.now(), duration, tps}]; history.splice(i + 1, 0, ...insert); lastDelegateResp = delegateResp; i += insert.length; } } // If the orchestrator added no commentary of its own, its answer IS the delegate's answer if(typeof resp === 'string' && !resp.trim() && lastDelegateResp !== null) resp = lastDelegateResp; // Trim memory injections from history if(mem?.tool) history.splice(0, history.length, ...history.filter(h => h.role !== 'tool' || h.name !== 'memory_recall')); // Auto-memorize before compressing if(options.compress && this.estimateTokens(history) >= options.compress.max) { if(mem?.update) await this.memoryManager.memorize(history, mem.memory, {model: options.memoryModel || this.defaultModel, ...options}); const compressed = await this.compressHistory(history, options.compress.max, options.compress.min, options); if(options.history) options.history.splice(0, options.history.length, ...compressed); } return resp; })(); return Object.assign(promise, {abort}); } /** * Digest full conversation history into memory documents. * Call on session end to persist the conversation. */ async updateMemory(history: LLMMessage[], memories: Memory[] | MemoryCache, options: LLMRequest = {}): Promise { return this.memoryManager.memorize(history, memories, {model: this.defaultModel, ...options}); } /** * Compress chat history to reduce context size * @param {LLMMessage[]} history Chatlog that will be compressed * @param max Trigger compression once context is larger than max * @param min Leave messages less than the token minimum, summarize the rest * @param {LLMRequest} options LLM options * @returns {Promise} New chat history will summary at index 0 */ async compressHistory(history: LLMMessage[], max: number, min: number, options?: LLMRequest): Promise { if(this.estimateTokens(history) < max) return history; let keep = 0, tokens = 0; for(let m of history.toReversed()) { tokens += this.estimateTokens(m.content); if(tokens < min) keep++; else break; } if(history.length <= keep) return history; const system = history[0].role == 'system' ? history[0] : null, recent = keep == 0 ? [] : history.slice(-keep), process = (keep == 0 ? history : history.slice(0, -keep)).filter(h => h.role === 'assistant' || h.role === 'user'); const summary: any = await this.summarize(process.map(m => `[${m.role}]: ${m.content}`).join('\n\n'), 500, options); const d = Date.now(); const h = [{role: 'tool', name: 'summary', id: `summary_` + d, args: {}, content: `Conversation Summary: ${summary?.summary}`, timestamp: d}, ...recent]; if(system) h.splice(0, 0, system); return h; } /** * Compare the difference between embeddings (calculates the angle between two vectors) * @param {number[]} v1 First embedding / vector comparison * @param {number[]} v2 Second embedding / vector for comparison * @returns {number} Similarity values 0-1: 0 = unique, 1 = identical */ cosineSimilarity(v1: number[], v2: number[]): number { if (v1.length !== v2.length) throw new Error('Vectors must be same length'); let dotProduct = 0, normA = 0, normB = 0; for (let i = 0; i < v1.length; i++) { dotProduct += v1[i] * v2[i]; normA += v1[i] * v1[i]; normB += v2[i] * v2[i]; } const denominator = Math.sqrt(normA) * Math.sqrt(normB); return denominator === 0 ? 0 : dotProduct / denominator; } /** * Chunk text into parts for AI digestion * @param {object | string} target Item that will be chunked (objects get converted) * @param {number} maxTokens Chunking size. More = better context, less = more specific (Search by paragraphs or lines) * @param {number} overlapTokens Includes previous X tokens to provide continuity to AI (In addition to max tokens) * @returns {string[]} Chunked strings */ chunk(target: object | string, maxTokens = 500, overlapTokens = 50): string[] { const objString = (obj: any, path = ''): string[] => { if(!obj) return []; return Object.entries(obj).flatMap(([key, value]) => { const p = path ? `${path}${isNaN(+key) ? `.${key}` : `[${key}]`}` : key; if(typeof value === 'object' && !Array.isArray(value)) return objString(value, p); return `${p}: ${Array.isArray(value) ? value.join(', ') : value}`; }); }; const lines = typeof target === 'object' ? objString(target) : target.toString().split('\n'); const tokens = lines.flatMap(l => [...l.split(/\s+/).filter(Boolean), '\n']); const chunks: string[] = []; for(let i = 0; i < tokens.length;) { let text = '', j = i; while(j < tokens.length) { const next = text + (text ? ' ' : '') + tokens[j]; if(this.estimateTokens(next.replace(/\s*\n\s*/g, '\n')) > maxTokens && text) break; text = next; j++; } const clean = text.replace(/\s*\n\s*/g, '\n').trim(); if(clean) chunks.push(clean); i = Math.max(j - overlapTokens, j === i ? i + 1 : j); } return chunks; } /** * Create a vector representation of a string * @param {object | string} target Item that will be embedded (objects get converted) * @param {maxTokens?: number, overlapTokens?: number} opts Options for embedding such as chunk sizes * @returns {Promise[]>} Chunked embeddings */ embedding(target: object | string, opts: {maxTokens?: number, overlapTokens?: number} = {}): AbortablePromise<{index: number, embedding: number[], text: string, tokens: number}[]> { let {maxTokens = 500, overlapTokens = 50} = opts; let aborted = false; const abort = () => { aborted = true; }; const embed = (text: string): Promise => { return new Promise((resolve, reject) => { if(aborted) return reject(new Error('Aborted')); const args: string[] = [ join(dirname(fileURLToPath(import.meta.url)), 'embedder.js'), this.ai.options.path, this.ai.options?.embedder || 'bge-small-en-v1.5' ]; const proc = spawn('node', args, {stdio: ['pipe', 'pipe', 'ignore']}); proc.stdin.write(text); proc.stdin.end(); let output = ''; proc.stdout.on('data', (data: Buffer) => output += data.toString()); proc.on('close', (code: number) => { if(aborted) return reject(new Error('Aborted')); if(code === 0) { try { const result = JSON.parse(output); resolve(result.embedding); } catch(err) { reject(err); } } else { reject(new Error(`Embedder process exited with code ${code}`)); } }); proc.on('error', reject); }); }; const p = (async () => { const chunks = this.chunk(target, maxTokens, overlapTokens), results: any[] = []; for(let i = 0; i < chunks.length; i++) { if(aborted) break; const text = chunks[i]; const embedding = await embed(text); results.push({index: i, embedding, text, tokens: this.estimateTokens(text)}); } return results; })(); return Object.assign(p, {abort}); } /** * Estimate variable as tokens * @param history Object to size * @returns {number} Rough token count */ estimateTokens(history: any): number { const text = JSON.stringify(history); return Math.ceil((text.length / 4) * 1.2); } /** * Compare the difference between two strings using tensor math * @param target Text that will be checked * @param {string} searchTerms Multiple search terms to check against target * @returns {{avg: number, max: number, similarities: number[]}} Similarity values 0-1: 0 = unique, 1 = identical */ fuzzyMatch(target, ...searchTerms) { if (searchTerms.length < 2) throw new Error('Requires at least 2 strings to compare'); const levenshtein = (a, b) => { const m = a.length, n = b.length; if (!m) return n; if (!n) return m; const dp = Array.from({length: m + 1}, (_, i) => [i, ...Array(n).fill(0)]); for (let j = 0; j <= n; j++) dp[0][j] = j; for (let i = 1; i <= m; i++) { for (let j = 1; j <= n; j++) { dp[i][j] = a[i - 1] === b[j - 1] ? dp[i - 1][j - 1] : 1 + Math.min(dp[i - 1][j - 1], dp[i - 1][j], dp[i][j - 1]); } } return dp[m][n]; }; const similarity = (a, b) => { a = a.toLowerCase(); b = b.toLowerCase(); return 1 - levenshtein(a, b) / Math.max(a.length, b.length, 1); }; const similarities = searchTerms.map(t => similarity(target, t)); return { avg: similarities.reduce((acc, s) => acc + s, 0) / similarities.length, max: Math.max(...similarities), similarities }; } /** * Create a summary of some text * @param {string} text Text to summarize * @param {number} length Max number of words * @param options LLM request options * @returns {Promise} Summary */ async summarize(text: string, length: number = 500, options?: LLMRequest): Promise { let system = `Your job is to summarize the users message using tool calls. Call the \`submit\` tool at least once with the shortest summary possible that's <= ${length} words. The tool call will respond with the token count. Responses are ignored`; if(options?.system) system += '\n\n' + options.system; return new Promise(async (resolve, reject) => { let done = false; const resp = await this.ask(text, { temperature: 0.3, ...options, system, tools: [{ name: 'submit', description: 'Submit summary', args: {summary: {type: 'string', description: 'Text summarization', required: true}}, fn: (args) => { if(!args.summary) return 'No summary provided'; const count = args.summary.split(' ').length; if(count > length) return `Too long: ${length} words`; done = true; resolve(args.summary || null); return `Saved: ${length} words`; } }, ...(options?.tools || [])], }); if(!done) reject(`AI failed to create summary:\n${resp}`); }); } addModel(name: string, config: AnthropicConfig | OpenAiConfig, setDefault = false) { if(config.proto == 'anthropic') this.models[name] = new Anthropic(this.ai, config.token, name); else if(config.proto == 'openai') this.models[name] = new OpenAi(this.ai, config.host || null, config.token, name); if(setDefault || !this.defaultModel) this.defaultModel = name; } removeModel(name: string) { delete this.models[name]; if(this.defaultModel === name) { this.defaultModel = Object.keys(this.models)[0] ?? ''; } } setModels(models: {[model: string]: AnthropicConfig | OpenAiConfig}, replace = true) { if(replace) this.models = {}; Object.entries(models).forEach(([model, config]) => { if(!this.defaultModel) this.defaultModel = model; if(config.proto == 'anthropic') this.models[model] = new Anthropic(this.ai, config.token, model); else if(config.proto == 'openai') this.models[model] = new OpenAi(this.ai, config.host || null, config.token, model); }); this.defaultModel = Object.keys(this.models)[0] ?? ''; } } export default LLM;