Added executor that is block-agnostic and executes serialized workflow

This commit is contained in:
Waleed Latif
2025-01-16 12:30:23 -08:00
parent 1a95e7630d
commit 65c7e21386
5 changed files with 530 additions and 8 deletions
+272
View File
@@ -0,0 +1,272 @@
import { Executor } from '../index';
import { SerializedWorkflow } from '@/serializer/types';
import { Tool } from '../types';
import { toolRegistry } from '@/tools/registry';
// Mock tools
class MockTool implements Tool {
constructor(
public name: string,
private mockExecute: (params: Record<string, any>) => Promise<Record<string, any>>,
private mockValidate: (params: Record<string, any>) => boolean | string = () => true
) {}
async execute(params: Record<string, any>): Promise<Record<string, any>> {
return this.mockExecute(params);
}
validateParams(params: Record<string, any>): boolean | string {
return this.mockValidate(params);
}
}
describe('Executor', () => {
beforeEach(() => {
// Reset toolRegistry mock
(toolRegistry as any) = {};
});
describe('Tool Execution', () => {
it('should execute a simple workflow with one tool', async () => {
const mockTool = new MockTool(
'test-tool',
async (params) => ({ result: params.input + ' processed' })
);
(toolRegistry as any)['test-tool'] = mockTool;
const workflow: SerializedWorkflow = {
version: '1.0',
blocks: [{
id: 'block-1',
position: { x: 0, y: 0 },
config: {
tool: 'test-tool',
params: {},
interface: {
inputs: { input: 'string' },
outputs: { result: 'string' }
}
}
}],
connections: []
};
const executor = new Executor(workflow);
const result = await executor.execute('workflow-1', { input: 'test' });
expect(result.success).toBe(true);
expect(result.data).toEqual({ result: 'test processed' });
});
it('should validate tool parameters', async () => {
const mockTool = new MockTool(
'test-tool',
async () => ({}),
(params) => params.required ? true : 'Missing required parameter'
);
(toolRegistry as any)['test-tool'] = mockTool;
const workflow: SerializedWorkflow = {
version: '1.0',
blocks: [{
id: 'block-1',
position: { x: 0, y: 0 },
config: {
tool: 'test-tool',
params: {},
interface: {
inputs: {},
outputs: {}
}
}
}],
connections: []
};
const executor = new Executor(workflow);
const result = await executor.execute('workflow-1', {});
expect(result.success).toBe(false);
expect(result.error).toContain('Missing required parameter');
});
});
describe('Interface Validation', () => {
it('should validate input types', async () => {
const mockTool = new MockTool(
'test-tool',
async (params) => ({ result: params.input })
);
(toolRegistry as any)['test-tool'] = mockTool;
const workflow: SerializedWorkflow = {
version: '1.0',
blocks: [{
id: 'block-1',
position: { x: 0, y: 0 },
config: {
tool: 'test-tool',
params: {},
interface: {
inputs: { input: 'number' },
outputs: { result: 'number' }
}
}
}],
connections: []
};
const executor = new Executor(workflow);
const result = await executor.execute('workflow-1', { input: 'not a number' });
expect(result.success).toBe(false);
expect(result.error).toContain('Invalid type for input');
});
it('should validate tool output against interface', async () => {
const mockTool = new MockTool(
'test-tool',
async () => ({ wrongField: 'wrong type' })
);
(toolRegistry as any)['test-tool'] = mockTool;
const workflow: SerializedWorkflow = {
version: '1.0',
blocks: [{
id: 'block-1',
position: { x: 0, y: 0 },
config: {
tool: 'test-tool',
params: {},
interface: {
inputs: {},
outputs: { result: 'string' }
}
}
}],
connections: []
};
const executor = new Executor(workflow);
const result = await executor.execute('workflow-1', {});
expect(result.success).toBe(false);
expect(result.error).toContain('Tool output missing required field');
});
});
describe('Complex Workflows', () => {
it('should execute a workflow with multiple connected blocks', async () => {
const processorTool = new MockTool(
'processor',
async (params) => ({ processed: params.input.toUpperCase() })
);
const formatterTool = new MockTool(
'formatter',
async (params) => ({ result: `<${params.processed}>` })
);
(toolRegistry as any)['processor'] = processorTool;
(toolRegistry as any)['formatter'] = formatterTool;
const workflow: SerializedWorkflow = {
version: '1.0',
blocks: [
{
id: 'process',
position: { x: 0, y: 0 },
config: {
tool: 'processor',
params: {},
interface: {
inputs: { input: 'string' },
outputs: { processed: 'string' }
}
}
},
{
id: 'format',
position: { x: 100, y: 0 },
config: {
tool: 'formatter',
params: {},
interface: {
inputs: { processed: 'string' },
outputs: { result: 'string' }
}
}
}
],
connections: [{
source: 'process',
target: 'format',
sourceHandle: 'processed',
targetHandle: 'processed'
}]
};
const executor = new Executor(workflow);
const result = await executor.execute('workflow-1', { input: 'test' });
expect(result.success).toBe(true);
expect(result.data).toEqual({ result: '<TEST>' });
});
it('should handle circular dependencies', async () => {
const mockTool = new MockTool(
'test-tool',
async () => ({ output: 'test' })
);
(toolRegistry as any)['test-tool'] = mockTool;
const workflow: SerializedWorkflow = {
version: '1.0',
blocks: [
{
id: 'block-1',
position: { x: 0, y: 0 },
config: {
tool: 'test-tool',
params: {},
interface: {
inputs: { input: 'string' },
outputs: { output: 'string' }
}
}
},
{
id: 'block-2',
position: { x: 100, y: 0 },
config: {
tool: 'test-tool',
params: {},
interface: {
inputs: { input: 'string' },
outputs: { output: 'string' }
}
}
}
],
connections: [
{
source: 'block-1',
target: 'block-2',
sourceHandle: 'output',
targetHandle: 'input'
},
{
source: 'block-2',
target: 'block-1',
sourceHandle: 'output',
targetHandle: 'input'
}
]
};
const executor = new Executor(workflow);
const result = await executor.execute('workflow-1', {});
expect(result.success).toBe(false);
expect(result.error).toContain('Workflow contains cycles');
});
});
});
+225
View File
@@ -0,0 +1,225 @@
import { SerializedWorkflow, SerializedBlock } from '@/serializer/types';
import { ExecutionContext, ExecutionResult, Tool } from './types';
import { toolRegistry } from '@/tools/registry';
export class Executor {
private workflow: SerializedWorkflow;
constructor(workflow: SerializedWorkflow) {
this.workflow = workflow;
}
private async executeBlock(
block: SerializedBlock,
inputs: Record<string, any>,
context: ExecutionContext
): Promise<Record<string, any>> {
// Get the tool specified by the block's tool property
const toolName = block.config.tool;
if (!toolName) {
throw new Error(`Block ${block.id} does not specify a tool`);
}
const tool = toolRegistry[toolName];
if (!tool) {
throw new Error(`Tool not found: ${toolName}`);
}
// Validate interface compatibility
this.validateInterface(block, inputs);
// Merge tool parameters with runtime inputs
const params = {
...block.config.params,
...inputs
};
// Validate the parameters against tool requirements
const validationResult = tool.validateParams(params);
if (typeof validationResult === 'string') {
throw new Error(`Invalid parameters for tool ${toolName}: ${validationResult}`);
}
try {
// Execute the tool and validate its output matches the interface
const result = await tool.execute(params);
this.validateToolOutput(block, result);
return result;
} catch (error) {
throw new Error(`Tool ${toolName} execution failed: ${error instanceof Error ? error.message : 'Unknown error'}`);
}
}
private validateInterface(block: SerializedBlock, inputs: Record<string, any>): void {
const { interface: blockInterface } = block.config;
// Check if all required inputs are provided
for (const [inputName, inputType] of Object.entries(blockInterface.inputs)) {
if (!(inputName in inputs)) {
throw new Error(`Missing required input '${inputName}' of type '${inputType}' for block ${block.id}`);
}
// Basic type validation (can be enhanced for more complex types)
if (!this.validateType(inputs[inputName], inputType)) {
throw new Error(`Invalid type for input '${inputName}' in block ${block.id}. Expected ${inputType}`);
}
}
}
private validateToolOutput(block: SerializedBlock, output: Record<string, any>): void {
const { interface: blockInterface } = block.config;
// Check if all promised outputs are present
for (const [outputName, outputType] of Object.entries(blockInterface.outputs)) {
if (!(outputName in output)) {
throw new Error(`Tool output missing required field '${outputName}' of type '${outputType}' for block ${block.id}`);
}
// Basic type validation (can be enhanced for more complex types)
if (!this.validateType(output[outputName], outputType)) {
throw new Error(`Invalid type for output '${outputName}' in block ${block.id}. Expected ${outputType}`);
}
}
}
private validateType(value: any, expectedType: string): boolean {
switch (expectedType.toLowerCase()) {
case 'string':
return typeof value === 'string';
case 'number':
return typeof value === 'number';
case 'boolean':
return typeof value === 'boolean';
case 'object':
return typeof value === 'object' && value !== null;
case 'array':
return Array.isArray(value);
case 'function':
return typeof value === 'function';
default:
// For complex types like 'Record<string, any>', 'string[]', etc.
// We just do basic object/array validation
return true;
}
}
private determineExecutionOrder(): string[] {
const { blocks, connections } = this.workflow;
const order: string[] = [];
const visited = new Set<string>();
const inDegree = new Map<string, number>();
blocks.forEach(block => inDegree.set(block.id, 0));
connections.forEach(conn => {
const target = conn.target;
inDegree.set(target, (inDegree.get(target) || 0) + 1);
});
const queue = blocks
.filter(block => (inDegree.get(block.id) || 0) === 0)
.map(block => block.id);
while (queue.length > 0) {
const blockId = queue.shift()!;
if (visited.has(blockId)) continue;
visited.add(blockId);
order.push(blockId);
connections
.filter(conn => conn.source === blockId)
.forEach(conn => {
const targetId = conn.target;
inDegree.set(targetId, (inDegree.get(targetId) || 0) - 1);
if (inDegree.get(targetId) === 0) {
queue.push(targetId);
}
});
}
if (order.length !== blocks.length) {
throw new Error('Workflow contains cycles');
}
return order;
}
private resolveInputs(
block: SerializedBlock,
context: ExecutionContext
): Record<string, any> {
const inputs: Record<string, any> = {};
// Get all incoming connections for this block
const incomingConnections = this.workflow.connections.filter(
conn => conn.target === block.id
);
// Map outputs from previous blocks to inputs for this block
incomingConnections.forEach(conn => {
const sourceOutput = context.blockStates.get(conn.source);
if (sourceOutput && conn.sourceHandle && conn.targetHandle) {
inputs[conn.targetHandle] = sourceOutput[conn.sourceHandle];
}
});
// If this is a start block, pass through workflow inputs
if (Object.keys(inputs).length === 0 && context.input) {
return context.input;
}
return inputs;
}
async execute(workflowId: string, input: Record<string, any>): Promise<ExecutionResult> {
const startTime = new Date();
const context: ExecutionContext = {
workflowId,
blockStates: new Map(),
input,
metadata: {
startTime: startTime.toISOString()
}
};
try {
const executionOrder = this.determineExecutionOrder();
for (const blockId of executionOrder) {
const block = this.workflow.blocks.find(b => b.id === blockId);
if (!block) {
throw new Error(`Block ${blockId} not found in workflow`);
}
const blockInputs = this.resolveInputs(block, context);
const result = await this.executeBlock(block, blockInputs, context);
context.blockStates.set(blockId, result);
}
const lastBlockId = executionOrder[executionOrder.length - 1];
const finalOutput = context.blockStates.get(lastBlockId);
const endTime = new Date();
return {
success: true,
data: finalOutput,
metadata: {
duration: endTime.getTime() - startTime.getTime(),
startTime: startTime.toISOString(),
endTime: endTime.toISOString()
}
};
} catch (error) {
const endTime = new Date();
return {
success: false,
data: {},
error: error instanceof Error ? error.message : 'Unknown error occurred',
metadata: {
duration: endTime.getTime() - startTime.getTime(),
startTime: startTime.toISOString(),
endTime: endTime.toISOString()
}
};
}
}
}
+27
View File
@@ -0,0 +1,27 @@
export interface Tool {
name: string;
execute(params: Record<string, any>): Promise<Record<string, any>>;
validateParams(params: Record<string, any>): boolean | string;
}
export interface ToolRegistry {
[key: string]: Tool;
}
export interface ExecutionContext {
workflowId: string;
blockStates: Map<string, any>;
input: Record<string, any>;
metadata?: Record<string, any>;
}
export interface ExecutionResult {
success: boolean;
data: Record<string, any>;
error?: string;
metadata?: {
duration: number;
startTime: string;
endTime: string;
};
}
+6 -6
View File
@@ -128,7 +128,7 @@ describe('Serializer', () => {
data: {
tool: 'model',
params: {
model: 'gpt-4'
model: 'gpt-4o'
},
interface: {
inputs: {},
@@ -142,7 +142,7 @@ describe('Serializer', () => {
expect(block.id).toBe('minimal-1');
expect(block.config.tool).toBe('model');
expect(block.config.params).toEqual({ model: 'gpt-4' });
expect(block.config.params).toEqual({ model: 'gpt-4o' });
expect(block.metadata).toBeUndefined();
});
@@ -173,7 +173,7 @@ describe('Serializer', () => {
data: {
tool: 'model',
params: {
model: 'gpt-4',
model: 'gpt-4o',
systemPrompt: 'Process this data'
},
interface: {
@@ -254,7 +254,7 @@ describe('Serializer', () => {
data: {
tool: 'model',
params: {
model: 'gpt-4',
model: 'gpt-4o',
temperature: 0.7,
maxTokens: 1000,
topP: 0.9,
@@ -273,7 +273,7 @@ describe('Serializer', () => {
const block = serialized.blocks[0];
expect(block.config.params).toEqual({
model: 'gpt-4',
model: 'gpt-4o',
temperature: 0.7,
maxTokens: 1000,
topP: 0.9,
@@ -374,7 +374,7 @@ describe('Serializer', () => {
config: {
tool: 'model',
params: {
model: 'gpt-4'
model: 'gpt-4o'
},
interface: {
inputs: {
-2
View File
@@ -1,5 +1,3 @@
import { BlockConfig } from "@/blocks/types/block";
export interface SerializedWorkflow {
version: string;
blocks: SerializedBlock[];