**feat: introduce unified augmentation pipeline system**
### Changes: - Added `Pipeline` class in `src/pipeline.ts` to unify primary and streamlined augmentation pipelines: - Supports multiple execution modes: sequential, parallel, threaded, first success, and first result. - Provides methods for registering, unregistering, and managing augmentations by type. - Implements flexible execution options, including error handling, timeouts, and threading control. - Introduced comprehensive augmentation registry for managing sense, dialog, memory, cognition, and WebSocket-based augmentations. - Added global pipeline instance with legacy method support for backward compatibility. - Enhanced augmentation execution capabilities: - Support for static and streaming data processing. - Pipeline creation for reusable augmentation workflows. - Threaded execution for improved performance when supported. - Added default pipeline integration with augmentation registry to resolve circular dependencies. ### Purpose: Implemented a unified augmentation pipeline to streamline augmentation workflows, enhance flexibility, and simplify execution management. The changes improve performance, reduce complexity, and provide robust support for legacy methods.
This commit is contained in:
parent
3860140c91
commit
e980d33a34
5 changed files with 1605 additions and 5 deletions
628
src/augmentationFactory.ts
Normal file
628
src/augmentationFactory.ts
Normal file
|
|
@ -0,0 +1,628 @@
|
|||
/**
|
||||
* Augmentation Factory
|
||||
*
|
||||
* This module provides a simplified factory for creating augmentations with minimal boilerplate.
|
||||
* It reduces the complexity of creating and using augmentations by providing a fluent API
|
||||
* and handling common patterns automatically.
|
||||
*/
|
||||
|
||||
import {
|
||||
IAugmentation,
|
||||
AugmentationType,
|
||||
AugmentationResponse,
|
||||
ISenseAugmentation,
|
||||
IConduitAugmentation,
|
||||
ICognitionAugmentation,
|
||||
IMemoryAugmentation,
|
||||
IPerceptionAugmentation,
|
||||
IDialogAugmentation,
|
||||
IActivationAugmentation,
|
||||
IWebSocketSupport,
|
||||
WebSocketConnection
|
||||
} from './types/augmentations.js'
|
||||
import { registerAugmentation } from './augmentationRegistry.js'
|
||||
|
||||
/**
|
||||
* Options for creating an augmentation
|
||||
*/
|
||||
export interface AugmentationOptions {
|
||||
name: string
|
||||
description?: string
|
||||
enabled?: boolean
|
||||
autoRegister?: boolean
|
||||
autoInitialize?: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Base class for all augmentations created with the factory
|
||||
* Handles common functionality like initialization, shutdown, and status
|
||||
*/
|
||||
class BaseAugmentation implements IAugmentation {
|
||||
readonly name: string
|
||||
readonly description: string
|
||||
enabled: boolean = true
|
||||
protected isInitialized: boolean = false
|
||||
|
||||
constructor(options: AugmentationOptions) {
|
||||
this.name = options.name
|
||||
this.description = options.description || `${options.name} augmentation`
|
||||
this.enabled = options.enabled !== false
|
||||
}
|
||||
|
||||
async initialize(): Promise<void> {
|
||||
if (this.isInitialized) return
|
||||
this.isInitialized = true
|
||||
}
|
||||
|
||||
async shutDown(): Promise<void> {
|
||||
this.isInitialized = false
|
||||
}
|
||||
|
||||
async getStatus(): Promise<'active' | 'inactive' | 'error'> {
|
||||
return this.isInitialized ? 'active' : 'inactive'
|
||||
}
|
||||
|
||||
protected async ensureInitialized(): Promise<void> {
|
||||
if (!this.isInitialized) {
|
||||
await this.initialize()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Factory for creating sense augmentations
|
||||
*/
|
||||
export function createSenseAugmentation(
|
||||
options: AugmentationOptions & {
|
||||
processRawData?: (
|
||||
rawData: Buffer | string,
|
||||
dataType: string
|
||||
) =>
|
||||
| Promise<AugmentationResponse<{ nouns: string[]; verbs: string[] }>>
|
||||
| AugmentationResponse<{
|
||||
nouns: string[]
|
||||
verbs: string[]
|
||||
}>
|
||||
listenToFeed?: (
|
||||
feedUrl: string,
|
||||
callback: (data: { nouns: string[]; verbs: string[] }) => void
|
||||
) => Promise<void>
|
||||
}
|
||||
): ISenseAugmentation {
|
||||
const augmentation = new BaseAugmentation(options) as unknown as ISenseAugmentation
|
||||
|
||||
// Implement the sense augmentation methods
|
||||
augmentation.processRawData = async (
|
||||
rawData: Buffer | string,
|
||||
dataType: string
|
||||
) => {
|
||||
await augmentation.ensureInitialized()
|
||||
|
||||
if (options.processRawData) {
|
||||
const result = options.processRawData(rawData, dataType)
|
||||
return result instanceof Promise ? await result : result
|
||||
}
|
||||
|
||||
return {
|
||||
success: false,
|
||||
data: { nouns: [], verbs: [] },
|
||||
error: 'processRawData not implemented'
|
||||
}
|
||||
}
|
||||
|
||||
augmentation.listenToFeed = async (
|
||||
feedUrl: string,
|
||||
callback: (data: { nouns: string[]; verbs: string[] }) => void
|
||||
) => {
|
||||
await augmentation.ensureInitialized()
|
||||
|
||||
if (options.listenToFeed) {
|
||||
return options.listenToFeed(feedUrl, callback)
|
||||
}
|
||||
|
||||
throw new Error('listenToFeed not implemented')
|
||||
}
|
||||
|
||||
// Auto-register if requested
|
||||
if (options.autoRegister) {
|
||||
registerAugmentation(augmentation)
|
||||
|
||||
// Auto-initialize if requested
|
||||
if (options.autoInitialize) {
|
||||
augmentation.initialize().catch((error) => {
|
||||
console.error(
|
||||
`Failed to initialize augmentation ${augmentation.name}:`,
|
||||
error
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return augmentation
|
||||
}
|
||||
|
||||
/**
|
||||
* Factory for creating conduit augmentations
|
||||
*/
|
||||
export function createConduitAugmentation(
|
||||
options: AugmentationOptions & {
|
||||
establishConnection?: (
|
||||
targetSystemId: string,
|
||||
config: Record<string, unknown>
|
||||
) =>
|
||||
| Promise<AugmentationResponse<WebSocketConnection>>
|
||||
| AugmentationResponse<WebSocketConnection>
|
||||
readData?: (
|
||||
query: Record<string, unknown>,
|
||||
options?: Record<string, unknown>
|
||||
) => Promise<AugmentationResponse<unknown>> | AugmentationResponse<unknown>
|
||||
writeData?: (
|
||||
data: Record<string, unknown>,
|
||||
options?: Record<string, unknown>
|
||||
) => Promise<AugmentationResponse<unknown>> | AugmentationResponse<unknown>
|
||||
monitorStream?: (
|
||||
streamId: string,
|
||||
callback: (data: unknown) => void
|
||||
) => Promise<void>
|
||||
}
|
||||
): IConduitAugmentation {
|
||||
const augmentation = new BaseAugmentation(options) as unknown as IConduitAugmentation
|
||||
|
||||
// Implement the conduit augmentation methods
|
||||
augmentation.establishConnection = async (
|
||||
targetSystemId: string,
|
||||
config: Record<string, unknown>
|
||||
) => {
|
||||
await augmentation.ensureInitialized()
|
||||
|
||||
if (options.establishConnection) {
|
||||
const result = options.establishConnection(targetSystemId, config)
|
||||
return result instanceof Promise ? await result : result
|
||||
}
|
||||
|
||||
return {
|
||||
success: false,
|
||||
data: null as any,
|
||||
error: 'establishConnection not implemented'
|
||||
}
|
||||
}
|
||||
|
||||
augmentation.readData = async (
|
||||
query: Record<string, unknown>,
|
||||
opts?: Record<string, unknown>
|
||||
) => {
|
||||
await augmentation.ensureInitialized()
|
||||
|
||||
if (options.readData) {
|
||||
const result = options.readData(query, opts)
|
||||
return result instanceof Promise ? await result : result
|
||||
}
|
||||
|
||||
return {
|
||||
success: false,
|
||||
data: null,
|
||||
error: 'readData not implemented'
|
||||
}
|
||||
}
|
||||
|
||||
augmentation.writeData = async (
|
||||
data: Record<string, unknown>,
|
||||
opts?: Record<string, unknown>
|
||||
) => {
|
||||
await augmentation.ensureInitialized()
|
||||
|
||||
if (options.writeData) {
|
||||
const result = options.writeData(data, opts)
|
||||
return result instanceof Promise ? await result : result
|
||||
}
|
||||
|
||||
return {
|
||||
success: false,
|
||||
data: null,
|
||||
error: 'writeData not implemented'
|
||||
}
|
||||
}
|
||||
|
||||
augmentation.monitorStream = async (
|
||||
streamId: string,
|
||||
callback: (data: unknown) => void
|
||||
) => {
|
||||
await augmentation.ensureInitialized()
|
||||
|
||||
if (options.monitorStream) {
|
||||
return options.monitorStream(streamId, callback)
|
||||
}
|
||||
|
||||
throw new Error('monitorStream not implemented')
|
||||
}
|
||||
|
||||
// Auto-register if requested
|
||||
if (options.autoRegister) {
|
||||
registerAugmentation(augmentation)
|
||||
|
||||
// Auto-initialize if requested
|
||||
if (options.autoInitialize) {
|
||||
augmentation.initialize().catch((error) => {
|
||||
console.error(
|
||||
`Failed to initialize augmentation ${augmentation.name}:`,
|
||||
error
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return augmentation
|
||||
}
|
||||
|
||||
/**
|
||||
* Factory for creating memory augmentations
|
||||
*/
|
||||
export function createMemoryAugmentation(
|
||||
options: AugmentationOptions & {
|
||||
storeData?: (
|
||||
key: string,
|
||||
data: unknown,
|
||||
options?: Record<string, unknown>
|
||||
) => Promise<AugmentationResponse<boolean>> | AugmentationResponse<boolean>
|
||||
retrieveData?: (
|
||||
key: string,
|
||||
options?: Record<string, unknown>
|
||||
) => Promise<AugmentationResponse<unknown>> | AugmentationResponse<unknown>
|
||||
updateData?: (
|
||||
key: string,
|
||||
data: unknown,
|
||||
options?: Record<string, unknown>
|
||||
) => Promise<AugmentationResponse<boolean>> | AugmentationResponse<boolean>
|
||||
deleteData?: (
|
||||
key: string,
|
||||
options?: Record<string, unknown>
|
||||
) => Promise<AugmentationResponse<boolean>> | AugmentationResponse<boolean>
|
||||
listDataKeys?: (
|
||||
pattern?: string,
|
||||
options?: Record<string, unknown>
|
||||
) =>
|
||||
| Promise<AugmentationResponse<string[]>>
|
||||
| AugmentationResponse<string[]>
|
||||
search?: (
|
||||
query: unknown,
|
||||
k?: number,
|
||||
options?: Record<string, unknown>
|
||||
) =>
|
||||
| Promise<
|
||||
AugmentationResponse<
|
||||
Array<{ id: string; score: number; data: unknown }>
|
||||
>
|
||||
>
|
||||
| AugmentationResponse<
|
||||
Array<{ id: string; score: number; data: unknown }>
|
||||
>
|
||||
}
|
||||
): IMemoryAugmentation {
|
||||
const augmentation = new BaseAugmentation(options) as unknown as IMemoryAugmentation
|
||||
|
||||
// Implement the memory augmentation methods
|
||||
augmentation.storeData = async (
|
||||
key: string,
|
||||
data: unknown,
|
||||
opts?: Record<string, unknown>
|
||||
) => {
|
||||
await augmentation.ensureInitialized()
|
||||
|
||||
if (options.storeData) {
|
||||
const result = options.storeData(key, data, opts)
|
||||
return result instanceof Promise ? await result : result
|
||||
}
|
||||
|
||||
return {
|
||||
success: false,
|
||||
data: false,
|
||||
error: 'storeData not implemented'
|
||||
}
|
||||
}
|
||||
|
||||
augmentation.retrieveData = async (
|
||||
key: string,
|
||||
opts?: Record<string, unknown>
|
||||
) => {
|
||||
await augmentation.ensureInitialized()
|
||||
|
||||
if (options.retrieveData) {
|
||||
const result = options.retrieveData(key, opts)
|
||||
return result instanceof Promise ? await result : result
|
||||
}
|
||||
|
||||
return {
|
||||
success: false,
|
||||
data: null,
|
||||
error: 'retrieveData not implemented'
|
||||
}
|
||||
}
|
||||
|
||||
augmentation.updateData = async (
|
||||
key: string,
|
||||
data: unknown,
|
||||
opts?: Record<string, unknown>
|
||||
) => {
|
||||
await augmentation.ensureInitialized()
|
||||
|
||||
if (options.updateData) {
|
||||
const result = options.updateData(key, data, opts)
|
||||
return result instanceof Promise ? await result : result
|
||||
}
|
||||
|
||||
return {
|
||||
success: false,
|
||||
data: false,
|
||||
error: 'updateData not implemented'
|
||||
}
|
||||
}
|
||||
|
||||
augmentation.deleteData = async (
|
||||
key: string,
|
||||
opts?: Record<string, unknown>
|
||||
) => {
|
||||
await augmentation.ensureInitialized()
|
||||
|
||||
if (options.deleteData) {
|
||||
const result = options.deleteData(key, opts)
|
||||
return result instanceof Promise ? await result : result
|
||||
}
|
||||
|
||||
return {
|
||||
success: false,
|
||||
data: false,
|
||||
error: 'deleteData not implemented'
|
||||
}
|
||||
}
|
||||
|
||||
augmentation.listDataKeys = async (
|
||||
pattern?: string,
|
||||
opts?: Record<string, unknown>
|
||||
) => {
|
||||
await augmentation.ensureInitialized()
|
||||
|
||||
if (options.listDataKeys) {
|
||||
const result = options.listDataKeys(pattern, opts)
|
||||
return result instanceof Promise ? await result : result
|
||||
}
|
||||
|
||||
return {
|
||||
success: false,
|
||||
data: [],
|
||||
error: 'listDataKeys not implemented'
|
||||
}
|
||||
}
|
||||
|
||||
augmentation.search = async (
|
||||
query: unknown,
|
||||
k?: number,
|
||||
opts?: Record<string, unknown>
|
||||
) => {
|
||||
await augmentation.ensureInitialized()
|
||||
|
||||
if (options.search) {
|
||||
const result = options.search(query, k, opts)
|
||||
return result instanceof Promise ? await result : result
|
||||
}
|
||||
|
||||
return {
|
||||
success: false,
|
||||
data: [],
|
||||
error: 'search not implemented'
|
||||
}
|
||||
}
|
||||
|
||||
// Auto-register if requested
|
||||
if (options.autoRegister) {
|
||||
registerAugmentation(augmentation)
|
||||
|
||||
// Auto-initialize if requested
|
||||
if (options.autoInitialize) {
|
||||
augmentation.initialize().catch((error) => {
|
||||
console.error(
|
||||
`Failed to initialize augmentation ${augmentation.name}:`,
|
||||
error
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return augmentation
|
||||
}
|
||||
|
||||
/**
|
||||
* Factory for creating WebSocket-enabled augmentations
|
||||
* This can be combined with other augmentation factories to create WebSocket-enabled versions
|
||||
*/
|
||||
export function addWebSocketSupport<T extends IAugmentation>(
|
||||
augmentation: T,
|
||||
options: {
|
||||
connectWebSocket?: (
|
||||
url: string,
|
||||
protocols?: string | string[]
|
||||
) => Promise<WebSocketConnection>
|
||||
sendWebSocketMessage?: (
|
||||
connectionId: string,
|
||||
data: unknown
|
||||
) => Promise<void>
|
||||
onWebSocketMessage?: (
|
||||
connectionId: string,
|
||||
callback: (data: unknown) => void
|
||||
) => Promise<void>
|
||||
offWebSocketMessage?: (
|
||||
connectionId: string,
|
||||
callback: (data: unknown) => void
|
||||
) => Promise<void>
|
||||
closeWebSocket?: (
|
||||
connectionId: string,
|
||||
code?: number,
|
||||
reason?: string
|
||||
) => Promise<void>
|
||||
}
|
||||
): T & IWebSocketSupport {
|
||||
const wsAugmentation = augmentation as T & IWebSocketSupport
|
||||
|
||||
// Add WebSocket methods
|
||||
wsAugmentation.connectWebSocket = async (
|
||||
url: string,
|
||||
protocols?: string | string[]
|
||||
) => {
|
||||
await (augmentation as any).ensureInitialized?.()
|
||||
|
||||
if (options.connectWebSocket) {
|
||||
return options.connectWebSocket(url, protocols)
|
||||
}
|
||||
|
||||
throw new Error('connectWebSocket not implemented')
|
||||
}
|
||||
|
||||
wsAugmentation.sendWebSocketMessage = async (
|
||||
connectionId: string,
|
||||
data: unknown
|
||||
) => {
|
||||
await (augmentation as any).ensureInitialized?.()
|
||||
|
||||
if (options.sendWebSocketMessage) {
|
||||
return options.sendWebSocketMessage(connectionId, data)
|
||||
}
|
||||
|
||||
throw new Error('sendWebSocketMessage not implemented')
|
||||
}
|
||||
|
||||
wsAugmentation.onWebSocketMessage = async (
|
||||
connectionId: string,
|
||||
callback: (data: unknown) => void
|
||||
) => {
|
||||
await (augmentation as any).ensureInitialized?.()
|
||||
|
||||
if (options.onWebSocketMessage) {
|
||||
return options.onWebSocketMessage(connectionId, callback)
|
||||
}
|
||||
|
||||
throw new Error('onWebSocketMessage not implemented')
|
||||
}
|
||||
|
||||
wsAugmentation.offWebSocketMessage = async (
|
||||
connectionId: string,
|
||||
callback: (data: unknown) => void
|
||||
) => {
|
||||
await (augmentation as any).ensureInitialized?.()
|
||||
|
||||
if (options.offWebSocketMessage) {
|
||||
return options.offWebSocketMessage(connectionId, callback)
|
||||
}
|
||||
|
||||
throw new Error('offWebSocketMessage not implemented')
|
||||
}
|
||||
|
||||
wsAugmentation.closeWebSocket = async (
|
||||
connectionId: string,
|
||||
code?: number,
|
||||
reason?: string
|
||||
) => {
|
||||
await (augmentation as any).ensureInitialized?.()
|
||||
|
||||
if (options.closeWebSocket) {
|
||||
return options.closeWebSocket(connectionId, code, reason)
|
||||
}
|
||||
|
||||
throw new Error('closeWebSocket not implemented')
|
||||
}
|
||||
|
||||
return wsAugmentation
|
||||
}
|
||||
|
||||
/**
|
||||
* Simplified function to execute an augmentation method with automatic error handling
|
||||
* This provides a more concise way to execute augmentation methods compared to the full pipeline
|
||||
*/
|
||||
export async function executeAugmentation<T, R>(
|
||||
augmentation: IAugmentation,
|
||||
method: string,
|
||||
...args: any[]
|
||||
): Promise<AugmentationResponse<R>> {
|
||||
try {
|
||||
if (!augmentation.enabled) {
|
||||
return {
|
||||
success: false,
|
||||
data: null as any,
|
||||
error: `Augmentation ${augmentation.name} is disabled`
|
||||
}
|
||||
}
|
||||
|
||||
if (typeof (augmentation as any)[method] !== 'function') {
|
||||
return {
|
||||
success: false,
|
||||
data: null as any,
|
||||
error: `Method ${method} not found on augmentation ${augmentation.name}`
|
||||
}
|
||||
}
|
||||
|
||||
const result = await (augmentation as any)[method](...args)
|
||||
return result
|
||||
} catch (error) {
|
||||
console.error(`Error executing ${method} on ${augmentation.name}:`, error)
|
||||
return {
|
||||
success: false,
|
||||
data: null as any,
|
||||
error: error instanceof Error ? error.message : String(error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Dynamically load augmentations from a module at runtime
|
||||
* This allows for lazy-loading augmentations when needed instead of at build time
|
||||
*/
|
||||
export async function loadAugmentationModule(
|
||||
modulePromise: Promise<any>,
|
||||
options: {
|
||||
autoRegister?: boolean
|
||||
autoInitialize?: boolean
|
||||
} = {}
|
||||
): Promise<IAugmentation[]> {
|
||||
try {
|
||||
const module = await modulePromise
|
||||
const augmentations: IAugmentation[] = []
|
||||
|
||||
// Extract augmentations from the module
|
||||
for (const key in module) {
|
||||
const exported = module[key]
|
||||
|
||||
// Skip non-objects and null
|
||||
if (!exported || typeof exported !== 'object') {
|
||||
continue
|
||||
}
|
||||
|
||||
// Check if it's an augmentation
|
||||
if (
|
||||
typeof exported.name === 'string' &&
|
||||
typeof exported.initialize === 'function' &&
|
||||
typeof exported.shutDown === 'function' &&
|
||||
typeof exported.getStatus === 'function'
|
||||
) {
|
||||
augmentations.push(exported)
|
||||
|
||||
// Auto-register if requested
|
||||
if (options.autoRegister) {
|
||||
registerAugmentation(exported)
|
||||
|
||||
// Auto-initialize if requested
|
||||
if (options.autoInitialize) {
|
||||
exported.initialize().catch((error: Error) => {
|
||||
console.error(
|
||||
`Failed to initialize augmentation ${exported.name}:`,
|
||||
error
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return augmentations
|
||||
} catch (error) {
|
||||
console.error('Error loading augmentation module:', error)
|
||||
return []
|
||||
}
|
||||
}
|
||||
|
|
@ -5,9 +5,21 @@
|
|||
* It replaces the dynamic loading mechanism in pluginLoader.ts.
|
||||
*/
|
||||
|
||||
import { AugmentationPipeline, augmentationPipeline } from './augmentationPipeline.js'
|
||||
import { IPipeline } from './types/pipelineTypes.js'
|
||||
import { AugmentationType, IAugmentation } from './types/augmentations.js'
|
||||
|
||||
// Forward declaration of the pipeline instance to avoid circular dependency
|
||||
// The actual pipeline will be provided when initializeAugmentationPipeline is called
|
||||
let defaultPipeline: IPipeline | null = null
|
||||
|
||||
/**
|
||||
* Sets the default pipeline instance
|
||||
* This function should be called from pipeline.ts after the pipeline is created
|
||||
*/
|
||||
export function setDefaultPipeline(pipeline: IPipeline): void {
|
||||
defaultPipeline = pipeline
|
||||
}
|
||||
|
||||
/**
|
||||
* Registry of all available augmentations
|
||||
*/
|
||||
|
|
@ -36,10 +48,18 @@ export function registerAugmentation<T extends IAugmentation>(augmentation: T):
|
|||
*
|
||||
* @param pipeline Optional custom pipeline to use instead of the default
|
||||
* @returns The pipeline that was initialized
|
||||
* @throws Error if no pipeline is provided and the default pipeline hasn't been set
|
||||
*/
|
||||
export function initializeAugmentationPipeline(
|
||||
pipeline: AugmentationPipeline = augmentationPipeline
|
||||
): AugmentationPipeline {
|
||||
pipelineInstance?: IPipeline
|
||||
): IPipeline {
|
||||
// Use the provided pipeline or fall back to the default
|
||||
const pipeline = pipelineInstance || defaultPipeline
|
||||
|
||||
if (!pipeline) {
|
||||
throw new Error('No pipeline provided and default pipeline not set. Call setDefaultPipeline first.')
|
||||
}
|
||||
|
||||
// Register all augmentations with the pipeline
|
||||
for (const augmentation of availableAugmentations) {
|
||||
if (augmentation.enabled) {
|
||||
|
|
|
|||
919
src/pipeline.ts
Normal file
919
src/pipeline.ts
Normal file
|
|
@ -0,0 +1,919 @@
|
|||
/**
|
||||
* Unified Pipeline
|
||||
*
|
||||
* This module combines the functionality of the primary augmentation pipeline and the streamlined pipeline
|
||||
* into a single, unified pipeline system. It provides both the registry functionality of the primary pipeline
|
||||
* and the simplified execution API of the streamlined pipeline.
|
||||
*/
|
||||
|
||||
import {
|
||||
IAugmentation,
|
||||
AugmentationType,
|
||||
AugmentationResponse,
|
||||
IWebSocketSupport,
|
||||
BrainyAugmentations
|
||||
} from './types/augmentations.js'
|
||||
import { AugmentationRegistry, IPipeline } from './types/pipelineTypes.js'
|
||||
import { isThreadingAvailable } from './utils/environment.js'
|
||||
import { executeInThread } from './utils/workerUtils.js'
|
||||
import { executeAugmentation } from './augmentationFactory.js'
|
||||
import { setDefaultPipeline } from './augmentationRegistry.js'
|
||||
|
||||
/**
|
||||
* Execution mode for the pipeline
|
||||
*/
|
||||
export enum ExecutionMode {
|
||||
SEQUENTIAL = 'sequential',
|
||||
PARALLEL = 'parallel',
|
||||
FIRST_SUCCESS = 'firstSuccess',
|
||||
FIRST_RESULT = 'firstResult',
|
||||
THREADED = 'threaded' // Execute in separate threads when available
|
||||
}
|
||||
|
||||
/**
|
||||
* Options for pipeline execution
|
||||
*/
|
||||
export interface PipelineOptions {
|
||||
mode?: ExecutionMode;
|
||||
timeout?: number;
|
||||
stopOnError?: boolean;
|
||||
forceThreading?: boolean; // Force threading even if not in THREADED mode
|
||||
disableThreading?: boolean; // Disable threading even if in THREADED mode
|
||||
}
|
||||
|
||||
/**
|
||||
* Default pipeline options
|
||||
*/
|
||||
const DEFAULT_PIPELINE_OPTIONS: PipelineOptions = {
|
||||
mode: ExecutionMode.SEQUENTIAL,
|
||||
timeout: 30000,
|
||||
stopOnError: false,
|
||||
forceThreading: false,
|
||||
disableThreading: false
|
||||
}
|
||||
|
||||
/**
|
||||
* Result of a pipeline execution
|
||||
*/
|
||||
export interface PipelineResult<T> {
|
||||
results: AugmentationResponse<T>[];
|
||||
errors: Error[];
|
||||
successful: AugmentationResponse<T>[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Pipeline class
|
||||
*
|
||||
* Manages multiple augmentations of each type and provides methods to execute them.
|
||||
* Implements the IPipeline interface to avoid circular dependencies.
|
||||
*/
|
||||
export class Pipeline implements IPipeline {
|
||||
private registry: AugmentationRegistry = {
|
||||
sense: [],
|
||||
conduit: [],
|
||||
cognition: [],
|
||||
memory: [],
|
||||
perception: [],
|
||||
dialog: [],
|
||||
activation: [],
|
||||
webSocket: []
|
||||
}
|
||||
|
||||
/**
|
||||
* Register an augmentation with the pipeline
|
||||
*
|
||||
* @param augmentation The augmentation to register
|
||||
* @returns The pipeline instance for chaining
|
||||
*/
|
||||
public register<T extends IAugmentation>(augmentation: T): Pipeline {
|
||||
let registered = false
|
||||
|
||||
// Check for specific augmentation types
|
||||
if (this.isAugmentationType<BrainyAugmentations.ISenseAugmentation>(
|
||||
augmentation,
|
||||
'processRawData',
|
||||
'listenToFeed'
|
||||
)) {
|
||||
this.registry.sense.push(augmentation)
|
||||
registered = true
|
||||
} else if (this.isAugmentationType<BrainyAugmentations.IConduitAugmentation>(
|
||||
augmentation,
|
||||
'establishConnection',
|
||||
'readData',
|
||||
'writeData',
|
||||
'monitorStream'
|
||||
)) {
|
||||
this.registry.conduit.push(augmentation)
|
||||
registered = true
|
||||
} else if (this.isAugmentationType<BrainyAugmentations.ICognitionAugmentation>(
|
||||
augmentation,
|
||||
'reason',
|
||||
'infer',
|
||||
'executeLogic'
|
||||
)) {
|
||||
this.registry.cognition.push(augmentation)
|
||||
registered = true
|
||||
} else if (this.isAugmentationType<BrainyAugmentations.IMemoryAugmentation>(
|
||||
augmentation,
|
||||
'storeData',
|
||||
'retrieveData',
|
||||
'updateData',
|
||||
'deleteData',
|
||||
'listDataKeys'
|
||||
)) {
|
||||
this.registry.memory.push(augmentation)
|
||||
registered = true
|
||||
} else if (this.isAugmentationType<BrainyAugmentations.IPerceptionAugmentation>(
|
||||
augmentation,
|
||||
'interpret',
|
||||
'organize',
|
||||
'generateVisualization'
|
||||
)) {
|
||||
this.registry.perception.push(augmentation)
|
||||
registered = true
|
||||
} else if (this.isAugmentationType<BrainyAugmentations.IDialogAugmentation>(
|
||||
augmentation,
|
||||
'processUserInput',
|
||||
'generateResponse',
|
||||
'manageContext'
|
||||
)) {
|
||||
this.registry.dialog.push(augmentation)
|
||||
registered = true
|
||||
} else if (this.isAugmentationType<BrainyAugmentations.IActivationAugmentation>(
|
||||
augmentation,
|
||||
'triggerAction',
|
||||
'generateOutput',
|
||||
'interactExternal'
|
||||
)) {
|
||||
this.registry.activation.push(augmentation)
|
||||
registered = true
|
||||
}
|
||||
|
||||
// Check if the augmentation supports WebSocket
|
||||
if (this.isAugmentationType<IWebSocketSupport>(
|
||||
augmentation,
|
||||
'connectWebSocket',
|
||||
'sendWebSocketMessage',
|
||||
'onWebSocketMessage',
|
||||
'closeWebSocket'
|
||||
)) {
|
||||
this.registry.webSocket.push(augmentation as IWebSocketSupport)
|
||||
registered = true
|
||||
}
|
||||
|
||||
// If the augmentation wasn't registered as any known type, throw an error
|
||||
if (!registered) {
|
||||
throw new Error(`Unknown augmentation type: ${augmentation.name}`)
|
||||
}
|
||||
|
||||
return this
|
||||
}
|
||||
|
||||
/**
|
||||
* Unregister an augmentation from the pipeline
|
||||
*
|
||||
* @param augmentationName The name of the augmentation to unregister
|
||||
* @returns The pipeline instance for chaining
|
||||
*/
|
||||
public unregister(augmentationName: string): Pipeline {
|
||||
let found = false
|
||||
|
||||
// Remove from all registries
|
||||
for (const type in this.registry) {
|
||||
const typedRegistry = this.registry[type as keyof AugmentationRegistry]
|
||||
const index = typedRegistry.findIndex(aug => aug.name === augmentationName)
|
||||
|
||||
if (index !== -1) {
|
||||
typedRegistry.splice(index, 1)
|
||||
found = true
|
||||
}
|
||||
}
|
||||
|
||||
return this
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize all registered augmentations
|
||||
*
|
||||
* @returns A promise that resolves when all augmentations are initialized
|
||||
*/
|
||||
public async initialize(): Promise<void> {
|
||||
const allAugmentations = this.getAllAugmentations()
|
||||
|
||||
await Promise.all(
|
||||
allAugmentations.map(augmentation =>
|
||||
augmentation.initialize().catch(error => {
|
||||
console.error(`Failed to initialize augmentation ${augmentation.name}:`, error)
|
||||
})
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Shut down all registered augmentations
|
||||
*
|
||||
* @returns A promise that resolves when all augmentations are shut down
|
||||
*/
|
||||
public async shutDown(): Promise<void> {
|
||||
const allAugmentations = this.getAllAugmentations()
|
||||
|
||||
await Promise.all(
|
||||
allAugmentations.map(augmentation =>
|
||||
augmentation.shutDown().catch(error => {
|
||||
console.error(`Failed to shut down augmentation ${augmentation.name}:`, error)
|
||||
})
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all registered augmentations
|
||||
*
|
||||
* @returns An array of all registered augmentations
|
||||
*/
|
||||
public getAllAugmentations(): IAugmentation[] {
|
||||
// Create a Set to avoid duplicates (an augmentation might be in multiple registries)
|
||||
const allAugmentations = new Set<IAugmentation>([
|
||||
...this.registry.sense,
|
||||
...this.registry.conduit,
|
||||
...this.registry.cognition,
|
||||
...this.registry.memory,
|
||||
...this.registry.perception,
|
||||
...this.registry.dialog,
|
||||
...this.registry.activation,
|
||||
...this.registry.webSocket
|
||||
])
|
||||
|
||||
// Convert back to array
|
||||
return Array.from(allAugmentations)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all augmentations of a specific type
|
||||
*
|
||||
* @param type The type of augmentation to get
|
||||
* @returns An array of all augmentations of the specified type
|
||||
*/
|
||||
public getAugmentationsByType(type: AugmentationType): IAugmentation[] {
|
||||
switch (type) {
|
||||
case AugmentationType.SENSE:
|
||||
return [...this.registry.sense]
|
||||
case AugmentationType.CONDUIT:
|
||||
return [...this.registry.conduit]
|
||||
case AugmentationType.COGNITION:
|
||||
return [...this.registry.cognition]
|
||||
case AugmentationType.MEMORY:
|
||||
return [...this.registry.memory]
|
||||
case AugmentationType.PERCEPTION:
|
||||
return [...this.registry.perception]
|
||||
case AugmentationType.DIALOG:
|
||||
return [...this.registry.dialog]
|
||||
case AugmentationType.ACTIVATION:
|
||||
return [...this.registry.activation]
|
||||
case AugmentationType.WEBSOCKET:
|
||||
return [...this.registry.webSocket]
|
||||
default:
|
||||
return []
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all available augmentation types
|
||||
*
|
||||
* @returns An array of all augmentation types that have at least one registered augmentation
|
||||
*/
|
||||
public getAvailableAugmentationTypes(): AugmentationType[] {
|
||||
const availableTypes: AugmentationType[] = []
|
||||
|
||||
if (this.registry.sense.length > 0) availableTypes.push(AugmentationType.SENSE)
|
||||
if (this.registry.conduit.length > 0) availableTypes.push(AugmentationType.CONDUIT)
|
||||
if (this.registry.cognition.length > 0) availableTypes.push(AugmentationType.COGNITION)
|
||||
if (this.registry.memory.length > 0) availableTypes.push(AugmentationType.MEMORY)
|
||||
if (this.registry.perception.length > 0) availableTypes.push(AugmentationType.PERCEPTION)
|
||||
if (this.registry.dialog.length > 0) availableTypes.push(AugmentationType.DIALOG)
|
||||
if (this.registry.activation.length > 0) availableTypes.push(AugmentationType.ACTIVATION)
|
||||
if (this.registry.webSocket.length > 0) availableTypes.push(AugmentationType.WEBSOCKET)
|
||||
|
||||
return availableTypes
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all WebSocket-supporting augmentations
|
||||
*
|
||||
* @returns An array of all augmentations that support WebSocket connections
|
||||
*/
|
||||
public getWebSocketAugmentations(): IWebSocketSupport[] {
|
||||
return [...this.registry.webSocket]
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if an augmentation is of a specific type
|
||||
*
|
||||
* @param augmentation The augmentation to check
|
||||
* @param methods The methods that should be present on the augmentation
|
||||
* @returns True if the augmentation is of the specified type
|
||||
*/
|
||||
private isAugmentationType<T extends IAugmentation>(
|
||||
augmentation: IAugmentation,
|
||||
...methods: (keyof T)[]
|
||||
): augmentation is T {
|
||||
// First check that the augmentation has all the required base methods
|
||||
const baseMethodsExist = [
|
||||
'initialize',
|
||||
'shutDown',
|
||||
'getStatus'
|
||||
].every(method => typeof (augmentation as any)[method] === 'function')
|
||||
|
||||
if (!baseMethodsExist) {
|
||||
return false
|
||||
}
|
||||
|
||||
// Then check that it has all the specific methods for this type
|
||||
return methods.every(method => typeof (augmentation as any)[method] === 'function')
|
||||
}
|
||||
|
||||
/**
|
||||
* Determines if threading should be used based on options and environment
|
||||
*
|
||||
* @param options The pipeline options
|
||||
* @returns True if threading should be used, false otherwise
|
||||
*/
|
||||
private shouldUseThreading(options: PipelineOptions): boolean {
|
||||
// If threading is explicitly disabled, don't use it
|
||||
if (options.disableThreading) {
|
||||
return false
|
||||
}
|
||||
|
||||
// If threading is explicitly forced, use it if available
|
||||
if (options.forceThreading) {
|
||||
return isThreadingAvailable()
|
||||
}
|
||||
|
||||
// If in THREADED mode, use threading if available
|
||||
if (options.mode === ExecutionMode.THREADED) {
|
||||
return isThreadingAvailable()
|
||||
}
|
||||
|
||||
// Otherwise, don't use threading
|
||||
return false
|
||||
}
|
||||
|
||||
/**
|
||||
* Executes a method on multiple augmentations using the specified execution mode
|
||||
*
|
||||
* @param augmentations The augmentations to execute the method on
|
||||
* @param method The method to execute
|
||||
* @param args The arguments to pass to the method
|
||||
* @param options Options for the execution
|
||||
* @returns A promise that resolves with the results
|
||||
*/
|
||||
public async execute<T>(
|
||||
augmentations: IAugmentation[],
|
||||
method: string,
|
||||
args: any[] = [],
|
||||
options: PipelineOptions = {}
|
||||
): Promise<PipelineResult<T>> {
|
||||
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
|
||||
const enabledAugmentations = augmentations.filter(aug => aug.enabled !== false)
|
||||
|
||||
if (enabledAugmentations.length === 0) {
|
||||
return { results: [], errors: [], successful: [] }
|
||||
}
|
||||
|
||||
const result: PipelineResult<T> = {
|
||||
results: [],
|
||||
errors: [],
|
||||
successful: []
|
||||
}
|
||||
|
||||
// Create a function to execute with timeout
|
||||
const executeWithTimeout = async (
|
||||
augmentation: IAugmentation
|
||||
): Promise<AugmentationResponse<T>> => {
|
||||
try {
|
||||
// Create a timeout promise if a timeout is specified
|
||||
if (opts.timeout) {
|
||||
const timeoutPromise = new Promise<never>((_, reject) => {
|
||||
setTimeout(() => {
|
||||
reject(
|
||||
new Error(`Timeout executing ${method} on ${augmentation.name}`)
|
||||
)
|
||||
}, opts.timeout)
|
||||
})
|
||||
|
||||
// Check if threading should be used
|
||||
const useThreading = this.shouldUseThreading(opts)
|
||||
|
||||
// Execute the method on the augmentation, using threading if appropriate
|
||||
let methodPromise: Promise<AugmentationResponse<T>>
|
||||
|
||||
if (useThreading) {
|
||||
// Execute in a separate thread
|
||||
try {
|
||||
// Create a function that can be serialized and executed in a worker
|
||||
const workerFn = (...workerArgs: any[]) => {
|
||||
// This function will be stringified and executed in the worker
|
||||
// It needs to be self-contained
|
||||
const augFn = augmentation[method as string] as Function
|
||||
return augFn.apply(augmentation, workerArgs)
|
||||
}
|
||||
|
||||
methodPromise = executeInThread<AugmentationResponse<T>>(workerFn, ...args)
|
||||
} catch (threadError) {
|
||||
console.warn(`Failed to execute in thread, falling back to main thread: ${threadError}`)
|
||||
// Fall back to executing in the main thread
|
||||
methodPromise = Promise.resolve((augmentation[method] as Function)(...args) as AugmentationResponse<T>)
|
||||
}
|
||||
} else {
|
||||
// Execute in the main thread
|
||||
methodPromise = Promise.resolve((augmentation[method] as Function)(...args) as AugmentationResponse<T>)
|
||||
}
|
||||
|
||||
// Race the method promise against the timeout promise
|
||||
return await Promise.race([methodPromise, timeoutPromise])
|
||||
} else {
|
||||
// No timeout, just execute the method
|
||||
return await executeAugmentation<any, T>(augmentation, method, ...args)
|
||||
}
|
||||
} catch (error) {
|
||||
result.errors.push(
|
||||
error instanceof Error ? error : new Error(String(error))
|
||||
)
|
||||
return {
|
||||
success: false,
|
||||
data: null as any,
|
||||
error: error instanceof Error ? error.message : String(error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Execute based on the specified mode
|
||||
switch (opts.mode) {
|
||||
case ExecutionMode.PARALLEL:
|
||||
case ExecutionMode.THREADED:
|
||||
// Execute all augmentations in parallel
|
||||
result.results = await Promise.all(
|
||||
enabledAugmentations.map((aug) => executeWithTimeout(aug))
|
||||
)
|
||||
break
|
||||
|
||||
case ExecutionMode.FIRST_SUCCESS:
|
||||
// Execute augmentations sequentially until one succeeds
|
||||
for (const augmentation of enabledAugmentations) {
|
||||
const response = await executeWithTimeout(augmentation)
|
||||
result.results.push(response)
|
||||
|
||||
if (response.success) {
|
||||
break
|
||||
}
|
||||
}
|
||||
break
|
||||
|
||||
case ExecutionMode.FIRST_RESULT:
|
||||
// Execute augmentations sequentially until one returns a non-null result
|
||||
for (const augmentation of enabledAugmentations) {
|
||||
const response = await executeWithTimeout(augmentation)
|
||||
result.results.push(response)
|
||||
|
||||
if (
|
||||
response.success &&
|
||||
response.data !== null &&
|
||||
response.data !== undefined
|
||||
) {
|
||||
break
|
||||
}
|
||||
}
|
||||
break
|
||||
|
||||
case ExecutionMode.SEQUENTIAL:
|
||||
default:
|
||||
// Execute augmentations sequentially
|
||||
for (const augmentation of enabledAugmentations) {
|
||||
const response = await executeWithTimeout(augmentation)
|
||||
result.results.push(response)
|
||||
|
||||
// Check if we need to stop on error
|
||||
if (opts.stopOnError && !response.success) {
|
||||
break
|
||||
}
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
// Filter successful results
|
||||
result.successful = result.results.filter((r) => r.success)
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Executes a method on augmentations of a specific type
|
||||
*
|
||||
* @param type The type of augmentations to execute the method on
|
||||
* @param method The method to execute
|
||||
* @param args The arguments to pass to the method
|
||||
* @param options Options for the execution
|
||||
* @returns A promise that resolves with the results
|
||||
*/
|
||||
public async executeByType<T>(
|
||||
type: AugmentationType,
|
||||
method: string,
|
||||
args: any[] = [],
|
||||
options: PipelineOptions = {}
|
||||
): Promise<PipelineResult<T>> {
|
||||
const augmentations = this.getAugmentationsByType(type)
|
||||
return this.execute<T>(augmentations, method, args, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Executes a method on a single augmentation with automatic error handling
|
||||
*
|
||||
* @param augmentation The augmentation to execute the method on
|
||||
* @param method The method to execute
|
||||
* @param args The arguments to pass to the method
|
||||
* @returns A promise that resolves with the result
|
||||
*/
|
||||
public async executeSingle<T>(
|
||||
augmentation: IAugmentation,
|
||||
method: string,
|
||||
...args: any[]
|
||||
): Promise<AugmentationResponse<T>> {
|
||||
return executeAugmentation<any, T>(augmentation, method, ...args)
|
||||
}
|
||||
|
||||
/**
|
||||
* Process static data through a pipeline of augmentations
|
||||
*
|
||||
* @param data The data to process
|
||||
* @param pipeline An array of processing steps, each with an augmentation, method, and optional args transformer
|
||||
* @param options Options for the execution
|
||||
* @returns A promise that resolves with the final result
|
||||
*/
|
||||
public async processStaticData<T, R = any>(
|
||||
data: T,
|
||||
pipeline: Array<{
|
||||
augmentation: IAugmentation
|
||||
method: string
|
||||
transformArgs?: (data: any, prevResult?: any) => any[]
|
||||
}>,
|
||||
options: PipelineOptions = {}
|
||||
): Promise<AugmentationResponse<R>> {
|
||||
let currentData = data
|
||||
let prevResult: any = undefined
|
||||
|
||||
for (const step of pipeline) {
|
||||
// Transform args if a transformer is provided, otherwise use the current data as the only arg
|
||||
const args = step.transformArgs
|
||||
? step.transformArgs(currentData, prevResult)
|
||||
: [currentData]
|
||||
|
||||
// Execute the method
|
||||
const result = await this.executeSingle<any>(
|
||||
step.augmentation,
|
||||
step.method,
|
||||
...args
|
||||
)
|
||||
|
||||
// If the step failed, return the error
|
||||
if (!result.success) {
|
||||
return result as AugmentationResponse<R>
|
||||
}
|
||||
|
||||
// Update the current data for the next step
|
||||
currentData = result.data
|
||||
prevResult = result
|
||||
}
|
||||
|
||||
// Return the final result
|
||||
return prevResult as AugmentationResponse<R>
|
||||
}
|
||||
|
||||
/**
|
||||
* Process streaming data through a pipeline of augmentations
|
||||
*
|
||||
* @param source The source augmentation that provides the data stream
|
||||
* @param sourceMethod The method on the source augmentation that provides the data stream
|
||||
* @param sourceArgs The arguments to pass to the source method
|
||||
* @param pipeline An array of processing steps, each with an augmentation, method, and optional args transformer
|
||||
* @param callback Function to call with the results of processing each data item
|
||||
* @param options Options for the execution
|
||||
* @returns A promise that resolves when the pipeline is set up
|
||||
*/
|
||||
public async processStreamingData<T>(
|
||||
source: IAugmentation,
|
||||
sourceMethod: string,
|
||||
sourceArgs: any[],
|
||||
pipeline: Array<{
|
||||
augmentation: IAugmentation
|
||||
method: string
|
||||
transformArgs?: (data: any, prevResult?: any) => any[]
|
||||
}>,
|
||||
callback: (result: AugmentationResponse<T>) => void,
|
||||
options: PipelineOptions = {}
|
||||
): Promise<void> {
|
||||
// Create a chain of processors
|
||||
const processData = async (data: any) => {
|
||||
let currentData = data
|
||||
let prevResult: any = undefined
|
||||
|
||||
for (const step of pipeline) {
|
||||
// Transform args if a transformer is provided, otherwise use the current data as the only arg
|
||||
const args = step.transformArgs
|
||||
? step.transformArgs(currentData, prevResult)
|
||||
: [currentData]
|
||||
|
||||
// Execute the method
|
||||
const result = await this.executeSingle<any>(
|
||||
step.augmentation,
|
||||
step.method,
|
||||
...args
|
||||
)
|
||||
|
||||
// If the step failed, return the error
|
||||
if (!result.success) {
|
||||
callback(result as AugmentationResponse<T>)
|
||||
return
|
||||
}
|
||||
|
||||
// Update the current data for the next step
|
||||
currentData = result.data
|
||||
prevResult = result
|
||||
}
|
||||
|
||||
// Call the callback with the final result
|
||||
callback(prevResult as AugmentationResponse<T>)
|
||||
}
|
||||
|
||||
// The last argument to the source method should be a callback that receives the data
|
||||
const dataCallback = (data: any) => {
|
||||
processData(data).catch((error) => {
|
||||
console.error('Error processing streaming data:', error)
|
||||
callback({
|
||||
success: false,
|
||||
data: null as any,
|
||||
error: error instanceof Error ? error.message : String(error)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
// Execute the source method with the provided args and the data callback
|
||||
await this.executeSingle(source, sourceMethod, ...sourceArgs, dataCallback)
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a reusable pipeline for processing data
|
||||
*
|
||||
* @param pipeline An array of processing steps
|
||||
* @param options Options for the execution
|
||||
* @returns A function that processes data through the pipeline
|
||||
*/
|
||||
public createPipeline<T, R>(
|
||||
pipeline: Array<{
|
||||
augmentation: IAugmentation
|
||||
method: string
|
||||
transformArgs?: (data: any, prevResult?: any) => any[]
|
||||
}>,
|
||||
options: PipelineOptions = {}
|
||||
): (data: T) => Promise<AugmentationResponse<R>> {
|
||||
return (data: T) => this.processStaticData<T, R>(data, pipeline, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a reusable streaming pipeline
|
||||
*
|
||||
* @param source The source augmentation
|
||||
* @param sourceMethod The method on the source augmentation
|
||||
* @param pipeline An array of processing steps
|
||||
* @param options Options for the execution
|
||||
* @returns A function that sets up the streaming pipeline
|
||||
*/
|
||||
public createStreamingPipeline<T, R>(
|
||||
source: IAugmentation,
|
||||
sourceMethod: string,
|
||||
pipeline: Array<{
|
||||
augmentation: IAugmentation
|
||||
method: string
|
||||
transformArgs?: (data: any, prevResult?: any) => any[]
|
||||
}>,
|
||||
options: PipelineOptions = {}
|
||||
): (
|
||||
sourceArgs: any[],
|
||||
callback: (result: AugmentationResponse<R>) => void
|
||||
) => Promise<void> {
|
||||
return (
|
||||
sourceArgs: any[],
|
||||
callback: (result: AugmentationResponse<R>) => void
|
||||
) =>
|
||||
this.processStreamingData<R>(
|
||||
source,
|
||||
sourceMethod,
|
||||
sourceArgs,
|
||||
pipeline,
|
||||
callback,
|
||||
options
|
||||
)
|
||||
}
|
||||
|
||||
// Legacy methods for backward compatibility
|
||||
|
||||
/**
|
||||
* Execute a sense pipeline (legacy method)
|
||||
*/
|
||||
public async executeSensePipeline<
|
||||
M extends keyof BrainyAugmentations.ISenseAugmentation & string,
|
||||
R extends BrainyAugmentations.ISenseAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never
|
||||
>(
|
||||
method: M & (BrainyAugmentations.ISenseAugmentation[M] extends (...args: any[]) => any ? M : never),
|
||||
args: Parameters<Extract<BrainyAugmentations.ISenseAugmentation[M], (...args: any[]) => any>>,
|
||||
options: PipelineOptions = {}
|
||||
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
||||
const result = await this.executeByType<R>(AugmentationType.SENSE, method, args, options)
|
||||
return result.results.map(r => Promise.resolve(r))
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a conduit pipeline (legacy method)
|
||||
*/
|
||||
public async executeConduitPipeline<
|
||||
M extends keyof BrainyAugmentations.IConduitAugmentation & string,
|
||||
R extends BrainyAugmentations.IConduitAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never
|
||||
>(
|
||||
method: M & (BrainyAugmentations.IConduitAugmentation[M] extends (...args: any[]) => any ? M : never),
|
||||
args: Parameters<Extract<BrainyAugmentations.IConduitAugmentation[M], (...args: any[]) => any>>,
|
||||
options: PipelineOptions = {}
|
||||
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
||||
const result = await this.executeByType<R>(AugmentationType.CONDUIT, method, args, options)
|
||||
return result.results.map(r => Promise.resolve(r))
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a cognition pipeline (legacy method)
|
||||
*/
|
||||
public async executeCognitionPipeline<
|
||||
M extends keyof BrainyAugmentations.ICognitionAugmentation & string,
|
||||
R extends BrainyAugmentations.ICognitionAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never
|
||||
>(
|
||||
method: M & (BrainyAugmentations.ICognitionAugmentation[M] extends (...args: any[]) => any ? M : never),
|
||||
args: Parameters<Extract<BrainyAugmentations.ICognitionAugmentation[M], (...args: any[]) => any>>,
|
||||
options: PipelineOptions = {}
|
||||
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
||||
const result = await this.executeByType<R>(AugmentationType.COGNITION, method, args, options)
|
||||
return result.results.map(r => Promise.resolve(r))
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a memory pipeline (legacy method)
|
||||
*/
|
||||
public async executeMemoryPipeline<
|
||||
M extends keyof BrainyAugmentations.IMemoryAugmentation & string,
|
||||
R extends BrainyAugmentations.IMemoryAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never
|
||||
>(
|
||||
method: M & (BrainyAugmentations.IMemoryAugmentation[M] extends (...args: any[]) => any ? M : never),
|
||||
args: Parameters<Extract<BrainyAugmentations.IMemoryAugmentation[M], (...args: any[]) => any>>,
|
||||
options: PipelineOptions = {}
|
||||
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
||||
const result = await this.executeByType<R>(AugmentationType.MEMORY, method, args, options)
|
||||
return result.results.map(r => Promise.resolve(r))
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a perception pipeline (legacy method)
|
||||
*/
|
||||
public async executePerceptionPipeline<
|
||||
M extends keyof BrainyAugmentations.IPerceptionAugmentation & string,
|
||||
R extends BrainyAugmentations.IPerceptionAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never
|
||||
>(
|
||||
method: M & (BrainyAugmentations.IPerceptionAugmentation[M] extends (...args: any[]) => any ? M : never),
|
||||
args: Parameters<Extract<BrainyAugmentations.IPerceptionAugmentation[M], (...args: any[]) => any>>,
|
||||
options: PipelineOptions = {}
|
||||
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
||||
const result = await this.executeByType<R>(AugmentationType.PERCEPTION, method, args, options)
|
||||
return result.results.map(r => Promise.resolve(r))
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a dialog pipeline (legacy method)
|
||||
*/
|
||||
public async executeDialogPipeline<
|
||||
M extends keyof BrainyAugmentations.IDialogAugmentation & string,
|
||||
R extends BrainyAugmentations.IDialogAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never
|
||||
>(
|
||||
method: M & (BrainyAugmentations.IDialogAugmentation[M] extends (...args: any[]) => any ? M : never),
|
||||
args: Parameters<Extract<BrainyAugmentations.IDialogAugmentation[M], (...args: any[]) => any>>,
|
||||
options: PipelineOptions = {}
|
||||
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
||||
const result = await this.executeByType<R>(AugmentationType.DIALOG, method, args, options)
|
||||
return result.results.map(r => Promise.resolve(r))
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute an activation pipeline (legacy method)
|
||||
*/
|
||||
public async executeActivationPipeline<
|
||||
M extends keyof BrainyAugmentations.IActivationAugmentation & string,
|
||||
R extends BrainyAugmentations.IActivationAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never
|
||||
>(
|
||||
method: M & (BrainyAugmentations.IActivationAugmentation[M] extends (...args: any[]) => any ? M : never),
|
||||
args: Parameters<Extract<BrainyAugmentations.IActivationAugmentation[M], (...args: any[]) => any>>,
|
||||
options: PipelineOptions = {}
|
||||
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
|
||||
const result = await this.executeByType<R>(AugmentationType.ACTIVATION, method, args, options)
|
||||
return result.results.map(r => Promise.resolve(r))
|
||||
}
|
||||
}
|
||||
|
||||
// Create and export a default instance of the pipeline
|
||||
export const pipeline = new Pipeline()
|
||||
|
||||
// Set the default pipeline instance for the augmentation registry
|
||||
// This breaks the circular dependency between pipeline.ts and augmentationRegistry.ts
|
||||
setDefaultPipeline(pipeline)
|
||||
|
||||
// Re-export the legacy pipeline for backward compatibility
|
||||
export const augmentationPipeline = pipeline
|
||||
|
||||
// Re-export the streamlined execution functions for backward compatibility
|
||||
export const executeStreamlined = <T>(
|
||||
augmentations: IAugmentation[],
|
||||
method: string,
|
||||
args: any[] = [],
|
||||
options: PipelineOptions = {}
|
||||
): Promise<PipelineResult<T>> => {
|
||||
return pipeline.execute<T>(augmentations, method, args, options)
|
||||
}
|
||||
|
||||
export const executeByType = <T>(
|
||||
type: AugmentationType,
|
||||
method: string,
|
||||
args: any[] = [],
|
||||
options: PipelineOptions = {}
|
||||
): Promise<PipelineResult<T>> => {
|
||||
return pipeline.executeByType<T>(type, method, args, options)
|
||||
}
|
||||
|
||||
export const executeSingle = <T>(
|
||||
augmentation: IAugmentation,
|
||||
method: string,
|
||||
...args: any[]
|
||||
): Promise<AugmentationResponse<T>> => {
|
||||
return pipeline.executeSingle<T>(augmentation, method, ...args)
|
||||
}
|
||||
|
||||
export const processStaticData = <T, R = any>(
|
||||
data: T,
|
||||
pipelineSteps: Array<{
|
||||
augmentation: IAugmentation
|
||||
method: string
|
||||
transformArgs?: (data: any, prevResult?: any) => any[]
|
||||
}>,
|
||||
options: PipelineOptions = {}
|
||||
): Promise<AugmentationResponse<R>> => {
|
||||
return pipeline.processStaticData<T, R>(data, pipelineSteps, options)
|
||||
}
|
||||
|
||||
export const processStreamingData = <T>(
|
||||
source: IAugmentation,
|
||||
sourceMethod: string,
|
||||
sourceArgs: any[],
|
||||
pipelineSteps: Array<{
|
||||
augmentation: IAugmentation
|
||||
method: string
|
||||
transformArgs?: (data: any, prevResult?: any) => any[]
|
||||
}>,
|
||||
callback: (result: AugmentationResponse<T>) => void,
|
||||
options: PipelineOptions = {}
|
||||
): Promise<void> => {
|
||||
return pipeline.processStreamingData<T>(source, sourceMethod, sourceArgs, pipelineSteps, callback, options)
|
||||
}
|
||||
|
||||
export const createPipeline = <T, R>(
|
||||
pipelineSteps: Array<{
|
||||
augmentation: IAugmentation
|
||||
method: string
|
||||
transformArgs?: (data: any, prevResult?: any) => any[]
|
||||
}>,
|
||||
options: PipelineOptions = {}
|
||||
): (data: T) => Promise<AugmentationResponse<R>> => {
|
||||
return pipeline.createPipeline<T, R>(pipelineSteps, options)
|
||||
}
|
||||
|
||||
export const createStreamingPipeline = <T, R>(
|
||||
source: IAugmentation,
|
||||
sourceMethod: string,
|
||||
pipelineSteps: Array<{
|
||||
augmentation: IAugmentation
|
||||
method: string
|
||||
transformArgs?: (data: any, prevResult?: any) => any[]
|
||||
}>,
|
||||
options: PipelineOptions = {}
|
||||
): (
|
||||
sourceArgs: any[],
|
||||
callback: (result: AugmentationResponse<R>) => void
|
||||
) => Promise<void> => {
|
||||
return pipeline.createStreamingPipeline<T, R>(source, sourceMethod, pipelineSteps, options)
|
||||
}
|
||||
|
||||
// For backward compatibility with StreamlinedExecutionMode
|
||||
export const StreamlinedExecutionMode = ExecutionMode
|
||||
export type StreamlinedPipelineOptions = PipelineOptions
|
||||
export type StreamlinedPipelineResult<T> = PipelineResult<T>
|
||||
|
|
@ -111,10 +111,10 @@ export namespace BrainyAugmentations {
|
|||
* @param rawData The raw, unstructured data (e.g., text, image buffer, audio stream)
|
||||
* @param dataType The type of raw data (e.g., 'text', 'image', 'audio')
|
||||
*/
|
||||
processRawData(rawData: Buffer | string, dataType: string): AugmentationResponse<{
|
||||
processRawData(rawData: Buffer | string, dataType: string): Promise<AugmentationResponse<{
|
||||
nouns: string[]
|
||||
verbs: string[]
|
||||
}>
|
||||
}>>
|
||||
|
||||
/**
|
||||
* Registers a listener for real-time data feeds.
|
||||
|
|
|
|||
33
src/types/pipelineTypes.ts
Normal file
33
src/types/pipelineTypes.ts
Normal file
|
|
@ -0,0 +1,33 @@
|
|||
/**
|
||||
* Pipeline Types
|
||||
*
|
||||
* This module provides shared types for the pipeline system to avoid circular dependencies.
|
||||
*/
|
||||
|
||||
import {
|
||||
BrainyAugmentations,
|
||||
IWebSocketSupport,
|
||||
IAugmentation
|
||||
} from './augmentations.js'
|
||||
|
||||
/**
|
||||
* Type definitions for the augmentation registry
|
||||
*/
|
||||
export type AugmentationRegistry = {
|
||||
sense: BrainyAugmentations.ISenseAugmentation[];
|
||||
conduit: BrainyAugmentations.IConduitAugmentation[];
|
||||
cognition: BrainyAugmentations.ICognitionAugmentation[];
|
||||
memory: BrainyAugmentations.IMemoryAugmentation[];
|
||||
perception: BrainyAugmentations.IPerceptionAugmentation[];
|
||||
dialog: BrainyAugmentations.IDialogAugmentation[];
|
||||
activation: BrainyAugmentations.IActivationAugmentation[];
|
||||
webSocket: IWebSocketSupport[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Interface for the Pipeline class
|
||||
* This is used to break circular dependencies between pipeline.ts and augmentationRegistry.ts
|
||||
*/
|
||||
export interface IPipeline {
|
||||
register<T extends IAugmentation>(augmentation: T): IPipeline;
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue