polymech/mono
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

kbot cache

Browse Source
This commit is contained in:
lovebird 2025-04-07 10:46:09 +02:00
parent 1c3c340d4c
commit 0a3ee28ca2
10 changed files with 233 additions and 944 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
packages/kbot/.vscode/launch.json vendored
View File

@ -723,6 +723,24 @@
"!**/node_modules/**" "!**/node_modules/**"
], ],
"outputCapture": "std" "outputCapture": "std"
},
{
"type": "node",
"request": "launch",
"name": "examples:iterator-factory",
"skipFiles": ["<node_internals>/**"],
"cwd": "${workspaceFolder}",
"runtimeExecutable": "npm",
"runtimeArgs": ["run", "examples:iterator-factory"],
"outFiles": ["${workspaceFolder}/**/*.js", "${workspaceFolder}/**/*.ts"],
"console": "integratedTerminal",
"internalConsoleOptions": "neverOpen",
"sourceMaps": true,
"resolveSourceMapLocations": [
"${workspaceFolder}/**",
"!**/node_modules/**"
],
"outputCapture": "std"
} }
] ]
} }

119
packages/kbot/dist-in/examples/core/iterator-factory-example.js
View File

File diff suppressed because one or more lines are too long

41
packages/kbot/dist-in/iterator.d.ts vendored
View File

@ -1,35 +1,6 @@
import { IKBotTask } from '@polymech/ai-tools'; import { IKBotTask } from '@polymech/ai-tools';
import { AsyncTransformer, ErrorCallback, FilterCallback } from './async-iterator.js'; import { AsyncTransformer, ErrorCallback, FilterCallback } from './async-iterator.js';
declare let cacheModule: { export declare const removeEmptyObjects: (obj: any) => any;
get_cached_object: (key: any, namespace: string) => Promise<any>;
set_cached_object: (key: any, namespace: string, value: any, options?: any) => Promise<void>;
rm_cached_object: (key: any, namespace: string) => Promise<void>;
} | null;
export declare function registerCacheModule(module: typeof cacheModule): void;
export interface CacheInterface {
get: (key: any) => Promise<any>;
set: (key: any, value: any) => Promise<void>;
delete?: (key: any) => Promise<void>;
}
export declare class NoopCache implements CacheInterface {
get(_key: any): Promise<any>;
set(_key: any, _value: any): Promise<void>;
delete(_key: any): Promise<void>;
}
export declare class DefaultCache implements CacheInterface {
private readonly logger?;
constructor(logger?: {
error: (message: string, error?: any) => void;
});
get(key: any): Promise<any>;
set(key: any, value: any): Promise<void>;
delete(key: any): Promise<void>;
}
export interface CacheOptions {
enabled: boolean;
implementation?: CacheInterface;
namespace?: string;
}
export interface FieldMapping { export interface FieldMapping {
jsonPath: string; jsonPath: string;
targetPath?: string | null; targetPath?: string | null;
@ -41,8 +12,11 @@ export interface IteratorFactory {
transform: (mappings: FieldMapping[]) => Promise<void>; transform: (mappings: FieldMapping[]) => Promise<void>;
createTransformer: (options: IKBotTask) => AsyncTransformer; createTransformer: (options: IKBotTask) => AsyncTransformer;
} }
export declare function createCacheKey(obj: any): string; export declare function createLLMTransformer(options: IKBotTask, logger?: {
export declare function removeEmptyValues(obj: any): any; info: (message: string) => void;
warn: (message: string) => void;
error: (message: string, error?: any) => void;
}): AsyncTransformer;
export declare function createIterator(obj: Record<string, any>, optionsMixin: Partial<IKBotTask>, globalOptions?: { export declare function createIterator(obj: Record<string, any>, optionsMixin: Partial<IKBotTask>, globalOptions?: {
throttleDelay?: number; throttleDelay?: number;
concurrentTasks?: number; concurrentTasks?: number;
@ -51,7 +25,6 @@ export declare function createIterator(obj: Record<string, any>, optionsMixin: P
transformerFactory?: (options: IKBotTask) => AsyncTransformer; transformerFactory?: (options: IKBotTask) => AsyncTransformer;
maxRetries?: number; maxRetries?: number;
retryDelay?: number; retryDelay?: number;
cache?: CacheOptions;
logger?: { logger?: {
error: (message: string, error?: any) => void; error: (message: string, error?: any) => void;
}; };
@ -63,9 +36,7 @@ export declare function transformWithMappings(obj: Record<string, any>, createTr
filterCallback?: FilterCallback; filterCallback?: FilterCallback;
maxRetries?: number; maxRetries?: number;
retryDelay?: number; retryDelay?: number;
cache?: CacheOptions;
logger?: { logger?: {
error: (message: string, error?: any) => void; error: (message: string, error?: any) => void;
}; };
}): Promise<void>; }): Promise<void>;
export {};

390
packages/kbot/dist-in/iterator.js
View File

File diff suppressed because one or more lines are too long

25
packages/kbot/package-lock.json generated
View File

@ -9,6 +9,7 @@
"version": "0.3.5", "version": "0.3.5",
"dependencies": { "dependencies": {
"@polymech/ai-tools": "file:../ai-tools", "@polymech/ai-tools": "file:../ai-tools",
"@polymech/cache": "file:../cache",
"@polymech/commons": "file:../commons", "@polymech/commons": "file:../commons",
"@polymech/core": "file:../core", "@polymech/core": "file:../core",
"@polymech/fs": "file:../fs", "@polymech/fs": "file:../fs",
@ -109,6 +110,26 @@
"puppeteer": "^24.2.1" "puppeteer": "^24.2.1"
} }
}, },
"../cache": {
"name": "@polymech/cache",
"version": "0.4.8",
"license": "BSD-3-Clause",
"dependencies": {
"@polymech/commons": "file:../commons",
"@polymech/core": "file:../core",
"@polymech/fs": "file:../fs",
"@polymech/log": "file:../log",
"@types/node": "^22.10.2",
"cacache": "^19.0.1",
"md5": "^2.3.0",
"p-map": "^7.0.3",
"ssri": "^10.0.1",
"yargs": "^17.7.2"
},
"engines": {
"node": ">= 14.0.0"
}
},
"../commons": { "../commons": {
"name": "@polymech/commons", "name": "@polymech/commons",
"version": "0.2.6", "version": "0.2.6",
@ -1359,6 +1380,10 @@
"resolved": "../ai-tools", "resolved": "../ai-tools",
"link": true "link": true
}, },
"node_modules/@polymech/cache": {
"resolved": "../cache",
"link": true
},
"node_modules/@polymech/commons": { "node_modules/@polymech/commons": {
"resolved": "../commons", "resolved": "../commons",
"link": true "link": true

1
packages/kbot/package.json
View File

@ -41,6 +41,7 @@
}, },
"dependencies": { "dependencies": {
"@polymech/ai-tools": "file:../ai-tools", "@polymech/ai-tools": "file:../ai-tools",
"@polymech/cache": "file:../cache",
"@polymech/commons": "file:../commons", "@polymech/commons": "file:../commons",
"@polymech/core": "file:../core", "@polymech/core": "file:../core",
"@polymech/fs": "file:../fs", "@polymech/fs": "file:../fs",

2
packages/kbot/src/commands/run.ts
View File

@ -144,7 +144,7 @@ export const execute_request = async (
params: ChatCompletionToolRunnerParams<any> params: ChatCompletionToolRunnerParams<any>
): Promise<ProcessRunResult> => { ): Promise<ProcessRunResult> => {
let ret: any = null let ret: any = null
try { try {
switch (options.mode) { switch (options.mode) {
case E_Mode.COMPLETION: case E_Mode.COMPLETION:

130
packages/kbot/src/examples/core/iterator-factory-example.ts
View File

@ -4,40 +4,8 @@ import * as fs from 'fs';
import type { IKBotTask } from '@polymech/ai-tools'; import type { IKBotTask } from '@polymech/ai-tools';
import { E_OPENROUTER_MODEL } from '../../models/cache/openrouter-models.js'; import { E_OPENROUTER_MODEL } from '../../models/cache/openrouter-models.js';
import { E_Mode } from '../../zod_schema.js'; import { E_Mode } from '../../zod_schema.js';
import { AsyncTransformer } from '../../async-iterator.js';
import { FieldMapping, createIterator, DefaultCache, registerCacheModule } from '../../iterator.js';
import { run } from '../../commands/run.js';
// Clear the mock cache for fresh testing import { FieldMapping, createIterator, createLLMTransformer } from '../../iterator.js';
const mockCacheModule = {
_cache: new Map<string, any>(),
get_cached_object: async (key: any, namespace: string) => {
const cacheKey = JSON.stringify({ key, namespace });
console.log(`Cache: Looking up ${cacheKey}`);
return mockCacheModule._cache.get(cacheKey) || null;
},
set_cached_object: async (key: any, namespace: string, value: any) => {
const cacheKey = JSON.stringify({ key, namespace });
console.log(`Cache: Storing in ${cacheKey}`);
mockCacheModule._cache.set(cacheKey, value);
// Show what's being stored for debugging
console.log(`Cache value stored: ${JSON.stringify(value, null, 2).substring(0, 100)}...`);
},
rm_cached_object: async (key: any, namespace: string) => {
const cacheKey = JSON.stringify({ key, namespace });
console.log(`Cache: REMOVING ${cacheKey}`);
// Check if it existed before removal
const existed = mockCacheModule._cache.has(cacheKey);
mockCacheModule._cache.delete(cacheKey);
// Show removal status
console.log(`Cache removal: ${existed ? 'REMOVED EXISTING CACHE VALUE' : 'NO CACHE VALUE FOUND'}`);
}
};
// Register our mock cache module
registerCacheModule(mockCacheModule);
const MODEL = E_OPENROUTER_MODEL.MODEL_OPENROUTER_QUASAR_ALPHA; const MODEL = E_OPENROUTER_MODEL.MODEL_OPENROUTER_QUASAR_ALPHA;
const ROUTER = 'openrouter'; const ROUTER = 'openrouter';
@ -104,36 +72,6 @@ const fieldMappings: FieldMapping[] = [
} }
]; ];
// Create LLM transformer factory
const createLLMTransformer = (options: IKBotTask): AsyncTransformer => {
return async (input: string, jsonPath: string): Promise<string> => {
logger.info(`Transforming field at path: ${jsonPath}`);
logger.info(`Input: ${input}`);
logger.info(`Using prompt: ${options.prompt}`);
const kbotTask: IKBotTask = {
...options,
prompt: `${options.prompt}\n\nText to transform: "${input}"`,
};
try {
const results = await run(kbotTask);
if (results && results.length > 0 && typeof results[0] === 'string') {
const result = results[0].trim();
logger.info(`Result: ${result}`);
return result;
}
logger.warn(`No valid result received for ${jsonPath}, returning original`);
return input;
} catch (error) {
console.error(`Error calling LLM API: ${error.message}`, error);
return input;
}
};
};
// Error handler // Error handler
const errorCallback = (path: string, value: string, error: any) => { const errorCallback = (path: string, value: string, error: any) => {
logger.error(`Error transforming ${path}: ${error.message}`); logger.error(`Error transforming ${path}: ${error.message}`);
@ -175,7 +113,7 @@ export async function factoryExample() {
concurrentTasks: 1, concurrentTasks: 1,
errorCallback, errorCallback,
filterCallback: async () => true, filterCallback: async () => true,
transformerFactory: createLLMTransformer, transformerFactory: (options) => createLLMTransformer(options, logger),
logger: logger, logger: logger,
cache: { cache: {
enabled: true, enabled: true,
@ -210,7 +148,7 @@ export async function factoryExample() {
concurrentTasks: 1, concurrentTasks: 1,
errorCallback, errorCallback,
filterCallback: async () => true, filterCallback: async () => true,
transformerFactory: createLLMTransformer, transformerFactory: (options) => createLLMTransformer(options, logger),
logger: logger, logger: logger,
cache: { cache: {
enabled: true, enabled: true,
@ -222,58 +160,6 @@ export async function factoryExample() {
// Should use cached values // Should use cached values
await iterator2.transform(fieldMappings); await iterator2.transform(fieldMappings);
// Manually set the marketingName for demonstration purposes
data2.products.fruits[0].name = {
value: "apple",
marketingName: "Crimson Orchard Delight"
};
data2.products.fruits[1].name = {
value: "banana",
marketingName: "Golden Delight Banana"
};
// Now create a third instance with caching disabled to demonstrate cache cleanup
console.log("\n\n========================================");
console.log("Third run - with caching disabled (should clean up cache):");
console.log("========================================");
/*
const data3 = JSON.parse(JSON.stringify(exampleData));
const iterator3 = createIterator(
data3,
globalOptionsMixin,
{
throttleDelay: 1000,
concurrentTasks: 1,
errorCallback,
filterCallback: async () => true,
transformerFactory: createLLMTransformer,
logger: logger,
cache: {
enabled: false, // Caching disabled
namespace: 'product-transformations'
}
}
);
// Should transform again and remove the cache
await iterator3.transform(fieldMappings);
// Verify cache was removed by checking if it still exists
console.log("\nVerifying cache removal:");
const cacheKey = {
key: JSON.stringify({
jsonPath: '$..name',
targetPath: 'marketingName',
options: { ...globalOptionsMixin, prompt: 'Generate a more appealing marketing name for this product' }
}),
name: 'product-transformations'
};
const cachedValue = await mockCacheModule.get_cached_object(cacheKey, 'product-transformations');
console.log(`Cache still exists? ${cachedValue !== null ? 'Yes' : 'No'}`);
*/
console.log("\nBefore/After Comparison Example:"); console.log("\nBefore/After Comparison Example:");
console.log(`Original description: ${exampleData.products.fruits[0].description}`); console.log(`Original description: ${exampleData.products.fruits[0].description}`);
console.log(`Transformed description: ${data.products.fruits[0].description}`); console.log(`Transformed description: ${data.products.fruits[0].description}`);
@ -291,21 +177,13 @@ export async function factoryExample() {
} }
} }
console.log("Module loading, checking if this is direct execution"); if (process.argv[1] && process.argv[1].includes('iterator-factory-example')) {
console.log("process.argv[1]:", process.argv[1]);
console.log("import.meta.url:", import.meta.url);
const isDirectExecution = process.argv[1] && process.argv[1].includes('iterator-factory-example');
console.log("Is direct execution:", isDirectExecution);
if (isDirectExecution) {
const isDebug = process.argv.includes('--debug'); const isDebug = process.argv.includes('--debug');
console.log('Starting iterator factory example...'); console.log('Starting iterator factory example...');
console.log(`Command arguments: ${process.argv.slice(2).join(', ')}`); console.log(`Command arguments: ${process.argv.slice(2).join(', ')}`);
console.log(`Debug mode: ${isDebug ? 'enabled' : 'disabled'}`); console.log(`Debug mode: ${isDebug ? 'enabled' : 'disabled'}`);
if (isDebug) { if (isDebug) {
// Set up debug logging
console.log("Running in debug mode with verbose output"); console.log("Running in debug mode with verbose output");
} }

427
packages/kbot/src/iterator.ts
View File

@ -1,89 +1,23 @@
import { IKBotTask } from '@polymech/ai-tools' import { IKBotTask } from '@polymech/ai-tools'
import { AsyncTransformer, ErrorCallback, FilterCallback, defaultError, defaultFilters, testFilters, transformObjectWithOptions } from './async-iterator.js' import { AsyncTransformer, ErrorCallback, FilterCallback, defaultError, defaultFilters, testFilters, transformObjectWithOptions } from './async-iterator.js'
import { JSONPath } from 'jsonpath-plus' import { run } from './commands/run.js'
import { deepClone } from '@polymech/core/objects' import { get_cached_object, set_cached_object, rm_cached_object } from "@polymech/cache"
// Add an optional import that can be provided by the user
let cacheModule: {
get_cached_object: (key: any, namespace: string) => Promise<any>;
set_cached_object: (key: any, namespace: string, value: any, options?: any) => Promise<void>;
rm_cached_object: (key: any, namespace: string) => Promise<void>;
} | null = null;
// Function to register the cache module export const removeEmptyObjects = (obj: any): any => {
export function registerCacheModule(module: typeof cacheModule) { if (obj === null || obj === undefined) return obj;
cacheModule = module; for (const key in obj) {
} const val = obj[key]
if (val === null || val === undefined) continue;
// Cache interface - users can implement to connect different cache systems if (typeof val === 'object' ||
export interface CacheInterface { (key == 'value' && typeof val === 'number' && val === 0 || key == 'base64')
get: (key: any) => Promise<any>; ) {
set: (key: any, value: any) => Promise<void>; obj[key] = removeEmptyObjects(obj[key]);
delete?: (key: any) => Promise<void>; if (Object.keys(obj[key]).length === 0) {
} delete obj[key];
// Default no-op cache implementation
export class NoopCache implements CacheInterface {
async get(_key: any) { return null; }
async set(_key: any, _value: any) { return; }
async delete(_key: any) { return; }
}
// Default cache implementation based on kbot.ts pattern
export class DefaultCache implements CacheInterface {
constructor(private readonly logger?: { error: (message: string, error?: any) => void }) {}
async get(key: any): Promise<any> {
if (!cacheModule) return null;
try {
const namespace = key.name || 'iterator';
const cached = await cacheModule.get_cached_object(key, namespace);
return cached;
} catch (e) {
if (this.logger) {
this.logger.error(`Failed to get cached object for key: ${JSON.stringify(key).substring(0, 50)}`, e);
} else {
console.error(`Cache get error:`, e);
}
return null;
}
}
async set(key: any, value: any): Promise<void> {
if (!cacheModule) return;
try {
const namespace = key.name || 'iterator';
await cacheModule.set_cached_object(key, namespace, value, {});
} catch (e) {
if (this.logger) {
this.logger.error(`Failed to cache object for key: ${JSON.stringify(key).substring(0, 50)}`, e);
} else {
console.error(`Cache set error:`, e);
} }
} }
} }
return obj
async delete(key: any): Promise<void> {
if (!cacheModule) return;
try {
const namespace = key.name || 'iterator';
await cacheModule.rm_cached_object(key, namespace);
} catch (e) {
if (this.logger) {
this.logger.error(`Failed to delete cached object for key: ${JSON.stringify(key).substring(0, 50)}`, e);
} else {
console.error(`Cache delete error:`, e);
}
}
}
}
export interface CacheOptions {
enabled: boolean;
implementation?: CacheInterface;
namespace?: string;
} }
export interface FieldMapping { export interface FieldMapping {
@ -99,45 +33,64 @@ export interface IteratorFactory {
createTransformer: (options: IKBotTask) => AsyncTransformer createTransformer: (options: IKBotTask) => AsyncTransformer
} }
// Utility function to create a cache key from an object export function createLLMTransformer(
export function createCacheKey(obj: any): string { options: IKBotTask,
try { logger?: {
const cleanObj = JSON.parse(JSON.stringify(removeEmptyValues(obj))); info: (message: string) => void;
return JSON.stringify(cleanObj); warn: (message: string) => void;
} catch (e) { error: (message: string, error?: any) => void;
console.error("Error creating cache key:", e);
// Fallback to simpler stringification if error occurs
return JSON.stringify({ hash: String(Date.now()) });
} }
} ): AsyncTransformer {
return async (input: string, jsonPath: string): Promise<string> => {
// Utility to remove null, undefined, and empty objects from an object for cache key generation if (logger) {
export function removeEmptyValues(obj: any): any { logger.info(`Transforming field at path: ${jsonPath}`);
if (obj === null || obj === undefined) { logger.info(`Input: ${input}`);
return undefined; logger.info(`Using prompt: ${options.prompt}`);
}
if (typeof obj !== 'object') {
return obj;
}
if (Array.isArray(obj)) {
const filtered = obj.map(removeEmptyValues).filter(v => v !== undefined);
return filtered.length ? filtered : undefined;
}
const result: Record<string, any> = {};
let hasValues = false;
for (const [key, value] of Object.entries(obj)) {
const cleaned = removeEmptyValues(value);
if (cleaned !== undefined) {
result[key] = cleaned;
hasValues = true;
} }
}
const kbotTask: IKBotTask = {
return hasValues ? result : undefined; ...options,
prompt: `${options.prompt}\n\nText to transform: "${input}"`,
};
// Create cache key for the LLM call
const cacheKeyObj = removeEmptyObjects({
prompt: kbotTask.prompt,
model: kbotTask.model,
router: kbotTask.router,
mode: kbotTask.mode,
filters: [],
tools: []
});
try {
const cachedResponse = await get_cached_object({ ca_options: cacheKeyObj }, 'llm-responses') as { content: string };
if (cachedResponse?.content) {
if (logger) logger.info(`Using cached LLM response for prompt: ${kbotTask.prompt.substring(0, 100)}...`);
return cachedResponse.content;
}
const results = await run(kbotTask)
if (results && results.length > 0 && typeof results[0] === 'string') {
const result = results[0].trim();
await set_cached_object({ ca_options: cacheKeyObj }, 'llm-responses', { content: result }, {});
if (logger) logger.info(`Cached LLM response for prompt: ${kbotTask.prompt.substring(0, 100)}...`);
if (logger) logger.info(`Result: ${result}`);
return result;
}
if (logger) logger.warn(`No valid result received for ${jsonPath}, returning original`);
return input;
} catch (error) {
if (logger) {
logger.error(`Error calling LLM API: ${error.message}`, error);
} else {
console.error(`Error calling LLM API: ${error.message}`, error);
}
return input;
}
};
} }
export function createIterator( export function createIterator(
@ -151,7 +104,6 @@ export function createIterator(
transformerFactory?: (options: IKBotTask) => AsyncTransformer transformerFactory?: (options: IKBotTask) => AsyncTransformer
maxRetries?: number maxRetries?: number
retryDelay?: number retryDelay?: number
cache?: CacheOptions
logger?: { error: (message: string, error?: any) => void } logger?: { error: (message: string, error?: any) => void }
} = {} } = {}
): IteratorFactory { ): IteratorFactory {
@ -163,7 +115,6 @@ export function createIterator(
transformerFactory, transformerFactory,
maxRetries = 3, maxRetries = 3,
retryDelay = 2000, retryDelay = 2000,
cache,
logger logger
} = globalOptions; } = globalOptions;
@ -172,244 +123,35 @@ export function createIterator(
// In real usage, this would be replaced by the provided transformerFactory // In real usage, this would be replaced by the provided transformerFactory
return async (input: string): Promise<string> => input; return async (input: string): Promise<string> => input;
}; };
const createTransformer = transformerFactory || defaultTransformerFactory;
// Default cache settings const createTransformer = transformerFactory || defaultTransformerFactory;
const defaultImplementation = logger ? new DefaultCache(logger) : new DefaultCache();
const cacheImpl = cache?.implementation || defaultImplementation;
const cacheEnabled = cache?.enabled || false;
const cacheNamespace = cache?.namespace || 'iterator';
return { return {
createTransformer, createTransformer,
transform: async (mappings: FieldMapping[]): Promise<void> => { transform: async (mappings: FieldMapping[]): Promise<void> => {
for (const mapping of mappings) { for (const mapping of mappings) {
// Merge the mapping options with the global mixin
const mergedOptions = { ...optionsMixin, ...mapping.options } as IKBotTask; const mergedOptions = { ...optionsMixin, ...mapping.options } as IKBotTask;
const { jsonPath, targetPath = null, maxRetries: mappingRetries, retryDelay: mappingRetryDelay } = mapping; const { jsonPath, targetPath = null, maxRetries: mappingRetries, retryDelay: mappingRetryDelay } = mapping;
const transformer = createTransformer(mergedOptions); const transformer = createTransformer(mergedOptions);
await transformObjectWithOptions(
// Generate cache key for this transformation obj,
if (cacheEnabled) { transformer,
const cacheKeyObj = { {
jsonPath, jsonPath,
targetPath, targetPath,
options: mergedOptions, throttleDelay,
concurrentTasks,
errorCallback,
filterCallback,
maxRetries: mappingRetries || maxRetries, maxRetries: mappingRetries || maxRetries,
retryDelay: mappingRetryDelay || retryDelay retryDelay: mappingRetryDelay || retryDelay
};
const cacheKey = { key: createCacheKey(cacheKeyObj), name: cacheNamespace };
// Check if we have a cached result
try {
const cachedResult = await cacheImpl.get(cacheKey);
if (cachedResult) {
// Apply cached result directly
applyTransformResult(obj, jsonPath, targetPath, cachedResult);
continue; // Skip to next mapping
}
} catch (e) {
errorCallback('cache', 'get', e);
} }
);
// For paths with targetPath, we need special handling for the LLM results
if (targetPath) {
// Get all the values that need transformation
const paths = JSONPath({ path: jsonPath, json: obj, resultType: 'pointer' });
const transformedData: Record<string, any> = {};
for (const path of paths) {
// Extract the value to transform
const keys = path.slice(1).split('/');
let current = obj;
for (const key of keys) {
if (key === '') continue;
if (current === undefined || current === null) break;
current = current[key];
}
// Transform the value if it exists
if (current !== undefined) {
try {
// Perform the transformation using the LLM
const rawValue = typeof current === 'object' && current !== null && 'value' in current
? current.value
: current;
const result = await transformer(String(rawValue), path);
// Store the result with the target path
transformedData[path] = { [targetPath]: result };
} catch (e) {
errorCallback(path, String(current), e);
}
}
}
// Store the result in the cache
if (Object.keys(transformedData).length > 0) {
try {
await cacheImpl.set(cacheKey, transformedData);
} catch (e) {
errorCallback('cache', 'set', e);
}
// Apply the transformations
applyTransformResult(obj, jsonPath, targetPath, transformedData);
}
} else {
// Regular case without targetPath - use normal approach
// Create a deep clone to transform
const objCopy = JSON.parse(JSON.stringify(obj));
// Perform the transformation
await transformObjectWithOptions(
objCopy,
transformer,
{
jsonPath,
targetPath,
throttleDelay,
concurrentTasks,
errorCallback,
filterCallback,
maxRetries: mappingRetries || maxRetries,
retryDelay: mappingRetryDelay || retryDelay
}
);
// Store the result in the cache
try {
// Extract the transformed parts to store in cache
const transformedData = extractTransformedData(objCopy, jsonPath, targetPath);
await cacheImpl.set(cacheKey, transformedData);
} catch (e) {
errorCallback('cache', 'set', e);
}
// Apply the transformation to the original object
applyTransformResult(obj, jsonPath, targetPath, extractTransformedData(objCopy, jsonPath, targetPath));
}
} else {
// Caching is disabled, but check if there are old cached values that should be removed
try {
// Create the same cacheKeyObj as used when caching is enabled
const keyObj = {
jsonPath,
targetPath,
options: mergedOptions,
maxRetries: mappingRetries || maxRetries,
retryDelay: mappingRetryDelay || retryDelay
};
const cacheKey = { key: createCacheKey(keyObj), name: cacheNamespace };
const cachedResult = await cacheImpl.get(cacheKey);
if (cachedResult && cacheImpl.delete) {
// If item exists in cache but caching is disabled, remove it
await cacheImpl.delete(cacheKey);
}
} catch (e) {
// Silently ignore errors when cleaning cache
}
// No caching, just transform the object directly
await transformObjectWithOptions(
obj,
transformer,
{
jsonPath,
targetPath,
throttleDelay,
concurrentTasks,
errorCallback,
filterCallback,
maxRetries: mappingRetries || maxRetries,
retryDelay: mappingRetryDelay || retryDelay
}
);
}
} }
} }
}; };
} }
// Extract transformed data based on the jsonPath and targetPath
function extractTransformedData(obj: Record<string, any>, jsonPath: string, targetPath: string | null): any {
const paths = JSONPath({ path: jsonPath, json: obj, resultType: 'pointer' });
const result: Record<string, any> = {};
for (const p of paths) {
const keys = p.slice(1).split('/');
let current = obj;
// Navigate to the value
for (const key of keys) {
if (key === '') continue;
if (current === undefined || current === null) break;
current = current[key];
}
// Store the value in the result
if (current !== undefined) {
if (targetPath) {
// For target path transformations (like marketingName),
// store the actual transformed value instead of the original value
if (typeof current === 'object' && current !== null && targetPath in current) {
result[p] = { [targetPath]: current[targetPath] };
} else {
// This is the raw value from the LLM - use this as is
result[p] = { [targetPath]: current };
}
} else {
result[p] = current;
}
}
}
return result;
}
// Apply transformed data to the original object
function applyTransformResult(obj: Record<string, any>, jsonPath: string, targetPath: string | null, transformedData: Record<string, any>): void {
for (const [path, value] of Object.entries(transformedData)) {
const keys = path.slice(1).split('/');
let current = obj;
// Navigate to the parent object
for (let i = 0; i < keys.length - 1; i++) {
const key = keys[i];
if (key === '') continue;
if (current[key] === undefined) {
current[key] = {};
}
current = current[key];
}
// Apply the value
const lastKey = keys[keys.length - 1];
if (lastKey !== '') {
if (targetPath) {
// Check if current[lastKey] is an object, if not convert it to an object
if (typeof current[lastKey] !== 'object' || current[lastKey] === null) {
// Save the original value
const originalValue = current[lastKey];
// Convert to object
current[lastKey] = { value: originalValue };
}
// Update the targetPath property with the value from the transformer, not the original value
if (typeof value === 'object' && value !== null && value[targetPath] !== undefined) {
current[lastKey][targetPath] = value[targetPath];
}
} else {
current[lastKey] = value;
}
}
}
}
export async function transformWithMappings( export async function transformWithMappings(
obj: Record<string, any>, obj: Record<string, any>,
createTransformer: (options: IKBotTask) => AsyncTransformer, createTransformer: (options: IKBotTask) => AsyncTransformer,
@ -421,7 +163,6 @@ export async function transformWithMappings(
filterCallback?: FilterCallback filterCallback?: FilterCallback
maxRetries?: number maxRetries?: number
retryDelay?: number retryDelay?: number
cache?: CacheOptions
logger?: { error: (message: string, error?: any) => void } logger?: { error: (message: string, error?: any) => void }
} = {} } = {}
): Promise<void> { ): Promise<void> {
@ -432,13 +173,12 @@ export async function transformWithMappings(
filterCallback = testFilters(defaultFilters()), filterCallback = testFilters(defaultFilters()),
maxRetries = 3, maxRetries = 3,
retryDelay = 2000, retryDelay = 2000,
cache,
logger logger
} = globalOptions; } = globalOptions;
const iterator = createIterator( const iterator = createIterator(
obj, obj,
{}, {},
{ {
throttleDelay, throttleDelay,
concurrentTasks, concurrentTasks,
@ -447,10 +187,9 @@ export async function transformWithMappings(
transformerFactory: createTransformer, transformerFactory: createTransformer,
maxRetries, maxRetries,
retryDelay, retryDelay,
cache,
logger logger
} }
); );
await iterator.transform(mappings); await iterator.transform(mappings);
} }

24
packages/kbot/tests/test-data/core/iterator-factory-data.json
View File

@ -3,29 +3,25 @@
"fruits": [ "fruits": [
{ {
"id": "f1", "id": "f1",
"name": { "name": "apple",
"value": "apple", "description": "A deliciously juicy fruit bursting with natural sweetness, offering a satisfying crunch with every bite—a perfect refreshing snack packed with vitamins and vibrant flavor.",
"marketingName": "Crimson Orchard Delight"
},
"description": "A deliciously sweet, juicy fruit with a satisfying crunch, bursting with refreshing flavor that delights your senses and offers a perfect balance of crisp texture and natural sweetness.",
"details": { "details": {
"color": "red", "color": "red",
"origin": "Worldwide", "origin": "Worldwide",
"nutrition": "Rich in fiber and vitamin C, which supports digestive health, promotes healthy gut bacteria, and boosts immune function. Vitamin C also aids collagen production for skin health and enhances iron absorption." "nutrition": "Rich in fiber and vitamin C, which supports a healthy digestive system, aids in maintaining stable blood sugar levels, strengthens the immune system, and promotes collagen production for healthier skin and faster wound healing."
} },
"marketingName": "Crimson Orchard Essence"
}, },
{ {
"id": "f2", "id": "f2",
"name": { "name": "banana",
"value": "banana", "description": "A vibrant, sun-kissed tropical fruit with a bright yellow hue, boasting a sweet, juicy flavor that bursts with refreshing tropical essence in every delicious bite.",
"marketingName": "Golden Tropic Delight"
},
"description": "A vibrant, sun-kissed yellow tropical fruit with a sweet, juicy flesh and an enticing aroma, bursting with refreshing flavor reminiscent of sunshine and warm island breezes.",
"details": { "details": {
"color": "yellow", "color": "yellow",
"origin": "Southeast Asia", "origin": "Southeast Asia",
"nutrition": "High in potassium, which helps maintain healthy blood pressure levels, supports proper muscle function, and promotes optimal nerve signaling, contributing to overall cardiovascular health and helping reduce the risk of stroke." "nutrition": "High in potassium, which supports healthy nerve and muscle function, helps maintain normal blood pressure, and reduces the risk of stroke by balancing sodium levels in the body for overall cardiovascular health."
} },
"marketingName": "Golden Sun Banana"
} }
] ]
} }