Tool pipelines;

src/main/java/de/extio/lmdatasetprep/Execution.java

@@ -0,0 +1,205 @@
+package de.extio.lmdatasetprep;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class Execution {
+	
+	public static final AtomicInteger startedFileConsumers = new AtomicInteger();
+	
+	public static final AtomicInteger finishedFileConsumers = new AtomicInteger();
+	
+	public static final AtomicBoolean shutdown = new AtomicBoolean();
+	
+	public static final ConcurrentMap<String, ConcurrentMap<String, BlockingQueue<WorkPacket>>> work = new ConcurrentHashMap<>();
+	
+	public static final BlockingQueue<CompletableFuture<Void>> tasks;
+	
+	public static final ScheduledExecutorService scheduler;
+	
+	public static final ExecutorService executorService;
+	
+	private static final Logger LOGGER = LoggerFactory.getLogger(Execution.class);
+	
+	static {
+		final int EXECUTORS = Integer.parseInt(LmDatasetPreparerApplication.applicationContext.getEnvironment().getProperty("agent.threads"));
+		
+		tasks = new LinkedBlockingQueue<>(EXECUTORS * 2);
+		
+		executorService = Executors.newFixedThreadPool(EXECUTORS);
+		
+		scheduler = Executors.newScheduledThreadPool(1);
+		scheduler.scheduleAtFixedRate(() -> {
+			tasks.removeIf(future -> future.isDone());
+		}, 0, 1, TimeUnit.SECONDS);
+	}
+	
+	public static void transform(final String source, final Function<WorkPacket, List<Runnable>> createTasks) {
+		if (source.startsWith("queue")) {
+			transformQueue(source, createTasks);
+		}
+		else {
+			transformDirectory(source, createTasks);
+		}
+	}
+	
+	private static void transformDirectory(final String source, final Function<WorkPacket, List<Runnable>> createTasks) {
+		startedFileConsumers.incrementAndGet();
+		try (Stream<Path> stream = Files.list(Paths.get(source))) {
+			stream.filter(p -> Files.isRegularFile(p))
+					.map(p -> {
+						try {
+							return new WorkPacket(p, Files.readString(p));
+						}
+						catch (final IOException e) {
+							LOGGER.error("IO exception", e);
+							return null;
+						}
+					})
+					.forEach(p -> {
+						for (final Runnable r : createTasks.apply(p)) {
+							try {
+								tasks.put(CompletableFuture.runAsync(r, executorService));
+							}
+							catch (final InterruptedException e) {
+								throw new RuntimeException("Interrupted", e);
+							}
+						}
+					});
+		}
+		catch (final IOException e) {
+			LOGGER.error("IO exception", e);
+		}
+		finally {
+			finishedFileConsumers.incrementAndGet();
+		}
+	}
+	
+	private static void transformQueue(final String source, final Function<WorkPacket, List<Runnable>> createTasks) {
+		final var queue = work.computeIfAbsent(source, k -> new ConcurrentHashMap<>()).computeIfAbsent(Thread.currentThread().getName(), p -> new LinkedBlockingQueue<>());
+		final Supplier<WorkPacket> getNext = () -> {
+			try {
+				WorkPacket next;
+				while ((next = queue.poll(1, TimeUnit.SECONDS)) == null) {
+					if (shutdown.get()) {
+						return null;
+					}
+				}
+				return next;
+			}
+			catch (final InterruptedException e) {
+				return null;
+			}
+		};
+		
+		WorkPacket next;
+		while ((next = getNext.get()) != null) {
+			for (final Runnable r : createTasks.apply(next)) {
+				try {
+					tasks.put(CompletableFuture.runAsync(r, executorService));
+				}
+				catch (final InterruptedException e) {
+					throw new RuntimeException("Interrupted", e);
+				}
+			}
+		}
+	}
+	
+	public static Path suffixFilename(final Path f, final String... suffixes) {
+		final StringBuilder sb = new StringBuilder();
+		for (final String suffix : suffixes) {
+			if (suffix.charAt(0) != '.') {
+				sb.append('_');
+				sb.append(suffix);
+			}
+		}
+		
+		final String[] parts = f.getFileName().toString().split("[.]");
+		String last;
+		if (suffixes[suffixes.length - 1].charAt(0) == '.') {
+			last = suffixes[suffixes.length - 1];
+		}
+		else {
+			last = "." + parts[1];
+		}
+		
+		final var parent = f.getParent();
+		if (parent != null) {
+			return f.getParent().resolve(parts[0] + sb.toString() + last);
+		}
+		return Path.of(parts[0] + sb.toString() + last);
+	}
+	
+	public static void streamOut(final Path f, final String propertyPrefix, final Properties properties, final Consumer<OutputStream> consumer) {
+		final var destination = properties.getProperty(propertyPrefix);
+		if (destination.startsWith("queue")) {
+			streamOutToQueue(f, destination, consumer);
+		}
+		else {
+			streamOutToFile(f, destination, consumer);
+		}
+		
+	}
+	
+	private static void streamOutToFile(final Path f, final String destination, final Consumer<OutputStream> consumer) {
+		final var out = Path.of(destination).resolve(f);
+		if (Files.exists(out)) {
+			LOGGER.info("File already exists: {} ", out);
+			return;
+		}
+		
+		final var tmp = out.resolveSibling(out.getFileName() + ".tmp");
+		try (var fos = Files.newOutputStream(tmp)) {
+			consumer.accept(fos);
+		}
+		catch (final IOException e1) {
+			LOGGER.error("IO exception", e1);
+			return;
+		}
+		try {
+			Files.move(tmp, out);
+		}
+		catch (final IOException e) {
+			LOGGER.error("IO exception", e);
+		}
+	}
+	
+	private static void streamOutToQueue(final Path f, final String destination, final Consumer<OutputStream> consumer) {
+		try (var bos = new ByteArrayOutputStream()) {
+			consumer.accept(bos);
+			final var text = bos.toString(StandardCharsets.UTF_8);
+			work.computeIfAbsent(destination, k -> new ConcurrentHashMap<>()).values().forEach(queue -> queue.add(new WorkPacket(f, text)));
+		}
+		catch (final IOException e) {
+			LOGGER.error("IO exception", e);
+		}
+	}
+	
+	public record WorkPacket(Path file, String text) {
+	}
+}

src/main/java/de/extio/lmdatasetprep/LmDatasetPreparerApplication.java

@@ -1,11 +1,19 @@
 package de.extio.lmdatasetprep;
 
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.function.Consumer;
+import java.util.List;
+import java.util.Properties;
+import java.util.stream.Collectors;
 
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanNotOfRequiredTypeException;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -14,7 +22,7 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.annotation.ComponentScan;
 
-import de.extio.lmdatasetprep.preparer.DatasetTool;
+import de.extio.lmdatasetprep.tools.DatasetTool;
 
 @SpringBootApplication
 @EnableCaching
@@ -29,25 +37,65 @@ public class LmDatasetPreparerApplication implements CommandLineRunner, Applicat
 	
 	public static ApplicationContext applicationContext;
 	
-	@SuppressWarnings("unchecked")
 	@Override
 	public void run(final String... args) throws Exception {
-		if (args.length == 0) {
+		if (args.length < 2) {
-			LOGGER.error("Commandline argument is missing: Provide the tool name as argument");
+			LOGGER.error("Commandline arguments are invalid. Usage: <tool properties file> [ <tool 1>, ... ]");
 			this.logAvailableTools();
 			return;
 		}
 		
-		final var bean = applicationContext.getBean(args[0]);
+		LOGGER.info("Loading tools");
-		if (!(bean instanceof Consumer) || !(bean instanceof DatasetTool)) {
+		final List<DatasetTool> tools = new ArrayList<>(args.length - 1);
-			LOGGER.error("Invalid bean {}", args[0]);
+		for (int i = 1; i < args.length; i++) {
-			this.logAvailableTools();
+			try {
-			return;
+				tools.add(applicationContext.getBean(args[i], DatasetTool.class));
+			}
+			catch (NoSuchBeanDefinitionException | BeanNotOfRequiredTypeException e) {
+				LOGGER.error("Invalid bean {}", args[i]);
+				this.logAvailableTools();
+				return;
+			}
 		}
 		
-		LOGGER.info("Executing {}", args[0]);
+		LOGGER.info("Loading tool properties");
+		final Properties toolProps = new Properties();
+		toolProps.load(Files.newInputStream(Path.of(args[0])));
+		
 		try {
-			((Consumer<String[]>) bean).accept(args);
+			LOGGER.info("Executing {}", tools.stream().map(tool -> tool.getClass().getSimpleName()).collect(Collectors.joining(", ")));
+			final List<Thread> threads = new ArrayList<>(tools.size());
+			for (final var tool : tools) {
+				final Thread t = new Thread(() -> tool.accept(toolProps));
+				t.setName(StringUtils.abbreviateMiddle(tool.getClass().getSimpleName(), ".", 15));
+				t.start();
+			}
+			
+			do {
+				Thread.sleep(1000);
+			} while (Execution.startedFileConsumers.get() == 0 || Execution.startedFileConsumers.get() > Execution.finishedFileConsumers.get());
+			LOGGER.info("All file consumers are finished");
+			
+			int cnt = 0;
+			while (cnt < 3) {
+				Thread.sleep(1000);
+				if (!Execution.tasks.isEmpty() || Execution.work.values().stream().anyMatch(q -> q.values().stream().anyMatch(bq -> !bq.isEmpty()))) {
+					cnt = 0;
+				}
+				else {
+					cnt++;
+				}
+			}
+			LOGGER.info("All queues are empty");
+			
+			Execution.shutdown.set(true);
+			for (final var thread : threads) {
+				thread.join();
+			}
+			
+			LOGGER.info("Shutdown");
+			Execution.scheduler.shutdown();
+			Execution.executorService.shutdown();
 		}
 		catch (final Exception e) {
 			LOGGER.error("Error executing the component", e);

src/main/java/de/extio/lmdatasetprep/TextUtils.java

@@ -1,34 +1,20 @@
-package de.extio.lmdatasetprep.preparer;
+package de.extio.lmdatasetprep;
 
-import java.io.IOException;
-import java.io.OutputStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import java.util.stream.Stream;
 
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import de.extio.lmdatasetprep.LmDatasetPreparerApplication;
+public class TextUtils {
 	
-public class Utils {
+	private static final Logger LOGGER = LoggerFactory.getLogger(TextUtils.class);
-	
-	private static final Logger LOGGER = LoggerFactory.getLogger(Utils.class);
 	
 	private static final List<String> PARAGRAPH_DELIMITERS = List.of("\n\n", "\n", ".", "!", "?");
 	
-	static String normalizeText(final String text) {
+	public static String normalizeText(final String text) {
 		String result = text;
 		
 		// Remove excessive newlines, spaces and tabs
@@ -66,7 +52,7 @@ public class Utils {
 		return result;
 	}
 	
-	static String normalizeModelResponse(final String response, final boolean removePreamble) {
+	public static String normalizeModelResponse(final String response, final boolean removePreamble) {
 		var result = StringUtils.replace(response, "\r", "");
 		
 		final int colon = result.indexOf(':');
@@ -81,7 +67,7 @@ public class Utils {
 		return result;
 	}
 	
-	static List<String> splitParagraphs(final String text, final int chunks_norm, final int chunks_var, final boolean slidingWindows) {
+	public static List<String> splitParagraphs(final String text, final int chunks_norm, final int chunks_var, final boolean slidingWindows) {
 		final int CHUNKS_MIN = chunks_norm - chunks_var;
 		final int CHUNKS_MAX = chunks_norm + chunks_var;
 		
@@ -144,104 +130,4 @@ public class Utils {
 		return splits;
 	}
 	
-	static void transformDirectory(final String path, final Function<Path, List<Runnable>> createTasks) {
-		final int EXECUTORS = Integer.parseInt(LmDatasetPreparerApplication.applicationContext.getEnvironment().getProperty("agent.threads"));
-		final AtomicBoolean finished = new AtomicBoolean(false);
-		final BlockingQueue<Runnable> tasks = new LinkedBlockingQueue<>(EXECUTORS * 2);
-		
-		final List<Thread> threads = new ArrayList<>(EXECUTORS);
-		for (int i = 0; i < EXECUTORS; i++) {
-			final Thread executor = new Thread(() -> {
-				while (!Thread.interrupted() && !finished.get()) {
-					try {
-						final Runnable task = tasks.poll(1, TimeUnit.SECONDS);
-						if (task != null) {
-							task.run();
-						}
-					}
-					catch (final InterruptedException e) {
-						break;
-					}
-				}
-			});
-			executor.setName("Executor " + i);
-			executor.start();
-			threads.add(executor);
-		}
-		
-		try (Stream<Path> stream = Files.list(Paths.get(path))) {
-			stream
-					.filter(p -> Files.isRegularFile(p))
-					.forEach(p -> {
-						for (final Runnable r : createTasks.apply(p)) {
-							try {
-								tasks.put(r);
-							}
-							catch (final InterruptedException e) {
-								throw new RuntimeException("Interrupted", e);
-							}
-						}
-					});
-		}
-		catch (final IOException e) {
-			LOGGER.error("Error", e);
-		}
-		
-		while (!tasks.isEmpty()) {
-			try {
-				Thread.sleep(1);
-			}
-			catch (final InterruptedException e) {
-				LOGGER.error("Interrupted", e);
-			}
-		}
-		finished.set(true);
-		
-		for (int i = 0; i < EXECUTORS; i++) {
-			try {
-				threads.get(i).join();
-			}
-			catch (final InterruptedException e) {
-				LOGGER.error("Interrupted", e);
-			}
-		}
-	}
-	
-	static Path suffixFilename(final Path f, final String... suffixes) {
-		final StringBuilder sb = new StringBuilder();
-		for (final String suffix : suffixes) {
-			if (suffix.charAt(0) != '.') {
-				sb.append('_');
-				sb.append(suffix);
-			}
-		}
-		
-		final String[] parts = f.getFileName().toString().split("[.]");
-		String last;
-		if (suffixes[suffixes.length - 1].charAt(0) == '.') {
-			last = suffixes[suffixes.length - 1];
-		}
-		else {
-			last = "." + parts[1];
-		}
-		
-		return f.getParent().resolve(parts[0] + sb.toString() + last);
-	}
-	
-	static void streamOut(final Path f, final Consumer<OutputStream> consumer) {
-		final var tmp = f.resolveSibling(f.getFileName() + ".tmp");
-		try (var fos = Files.newOutputStream(tmp)) {
-			consumer.accept(fos);
-		}
-		catch (final IOException e1) {
-			LOGGER.error("IO exception", e1);
-			return;
-		}
-		try {
-			Files.move(tmp, f);
-		}
-		catch (final IOException e) {
-			LOGGER.error("IO exception", e);
-		}
-	}
 }

src/main/java/de/extio/lmdatasetprep/preparer/DatasetTool.java

@@ -1,6 +0,0 @@
-package de.extio.lmdatasetprep.preparer;
-
-
-public interface DatasetTool {
-	
-}

src/main/java/de/extio/lmdatasetprep/preparer/Translate.java

@@ -1,85 +0,0 @@
-package de.extio.lmdatasetprep.preparer;
-
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Consumer;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-import de.extio.lmlib.client.ClientService;
-import de.extio.lmlib.profile.ModelCategory;
-
-@Component
-public class Translate implements Consumer<String[]>, DatasetTool {
-	
-	private static final Logger LOGGER = LoggerFactory.getLogger(Translate.class);
-	
-	private static final byte[] PARAGRAPH = "\n\n".getBytes(StandardCharsets.UTF_8);
-	
-	@Autowired
-	private ClientService clientService;
-	
-	@Override
-	public void accept(final String[] args) {
-		if (args.length < 3) {
-			LOGGER.error("Arguments are missing. <Path to text files> <Path to output dir>");
-			return;
-		}
-		
-		Utils.transformDirectory(args[1], p -> {
-			return List.of(() -> {
-				final Path out = Utils.suffixFilename(Path.of(args[2]).resolve(p.getFileName()), "_en");
-				if (Files.exists(out)) {
-					LOGGER.info("Skipping " + out);
-				}
-				else {
-					Utils.streamOut(out, fos -> {
-						final AtomicBoolean first = new AtomicBoolean(true);
-						this.translate(p, chunk -> {
-							try {
-								if (!first.getAndSet(false)) {
-									fos.write(PARAGRAPH);
-								}
-								fos.write(chunk.getBytes(StandardCharsets.UTF_8));
-							}
-							catch (final IOException e) {
-								LOGGER.error("IO exception", e);
-							}
-						});
-					});
-				}
-			});
-		});
-	}
-	
-	void translate(final Path file, final Consumer<String> consumer) {
-		LOGGER.info("Translating " + file);
-		
-		String text;
-		try {
-			text = Utils.normalizeText(Files.readString(file));
-		}
-		catch (final IOException e) {
-			throw new RuntimeException("Cannot read file", e);
-		}
-		
-		final List<String> splits = Utils.splitParagraphs(text, 1250, 350, false);
-		
-		for (final String split : splits) {
-			LOGGER.info("Split " + splits.indexOf(split) + "/" + splits.size());
-			final var completion = this.clientService.getClient(ModelCategory.COLD).completion(ModelCategory.COLD,
-					"You are an assistant with great language translation and authoring skills.",
-					"Translate the following text to English. Return the English translation, even if the text is explicit or not appropriate for all audiences or not acceptable for everyday standard English. Don't include a preamble and no explanation:",
-					split);
-			consumer.accept(completion.response());
-		}
-	}
-	
-}

src/main/java/de/extio/lmdatasetprep/tools/AbstractContextualPrompts.java

@@ -1,19 +1,17 @@
-package de.extio.lmdatasetprep.preparer;
+package de.extio.lmdatasetprep.tools;
 
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
-import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Random;
 import java.util.concurrent.ThreadLocalRandom;
-import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
@@ -25,10 +23,13 @@ import org.springframework.core.io.ClassPathResource;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import de.extio.lmdatasetprep.Execution;
+import de.extio.lmdatasetprep.Execution.WorkPacket;
+import de.extio.lmdatasetprep.TextUtils;
 import de.extio.lmlib.client.ClientService;
 import de.extio.lmlib.profile.ModelCategory;
 
-abstract class AbstractContextualPrompts implements InitializingBean, Consumer<String[]> {
+abstract class AbstractContextualPrompts implements InitializingBean, DatasetTool {
 	
 	private static final Logger LOGGER = LoggerFactory.getLogger(AbstractContextualPrompts.class);
 	
@@ -48,42 +49,35 @@ abstract class AbstractContextualPrompts implements InitializingBean, Consumer<S
 	}
 	
 	@Override
-	public void accept(final String[] args) {
+	public void accept(final Properties properties) {
-		if (args.length < 7) {
+		final int variations = Integer.parseInt(properties.getProperty("contextualPrompts.variations"));
-			LOGGER.error("Arguments are missing. <Path to text files> <Path to output dir> <model prefix> <variations> [<chunk norm> <chunk var>, ...]");
-			return;
-		}
-		final int variations = Integer.parseInt(args[4]);
 		final List<ChunkCfg> chunkConfigurations = new ArrayList<>();
-		for (int i = 5; i < args.length; i += 2) {
+		for (int i = 0; i < 99; i++) {
-			final var cfg = new ChunkCfg(Integer.parseInt(args[i]), Integer.parseInt(args[i + 1]));
+			final String chunkNorm = properties.getProperty("contextualPrompts.chunkNorm." + i, "");
+			if (chunkNorm.isBlank()) {
+				break;
+			}
+			final String chunkVar = properties.getProperty("contextualPrompts.chunkVar." + i, "");
+			final var cfg = new ChunkCfg(Integer.parseInt(chunkNorm), Integer.parseInt(chunkVar));
 			chunkConfigurations.add(cfg);
 		}
 		
-		Utils.transformDirectory(args[1], this.createTasks(args, chunkConfigurations, variations)::createTasks);
+		Execution.transform(properties.getProperty("contextualPrompts.source"), this.createTasks(properties, chunkConfigurations, variations)::createTasks);
 	}
 	
 	@FunctionalInterface
 	static interface CreateTasks {
 		
-		List<Runnable> createTasks(Path path);
+		List<Runnable> createTasks(WorkPacket packet);
 		
 	}
 	
-	protected abstract CreateTasks createTasks(String[] args, List<ChunkCfg> chunkConfigurations, int variations);
+	protected abstract CreateTasks createTasks(Properties properties, List<ChunkCfg> chunkConfigurations, int variations);
 	
-	protected List<String> fileToParagraphs(final Path file, final int chunkNorm, final int chunkVar) {
+	protected List<String> fileToParagraphs(final WorkPacket packet, final int chunkNorm, final int chunkVar) {
-		LOGGER.info("Splitting to jsonl " + file + " -> " + chunkNorm + "±" + chunkVar);
+		LOGGER.info("Splitting to jsonl " + packet.file() + " -> " + chunkNorm + "±" + chunkVar);
-		
+		final String text = TextUtils.normalizeText(packet.text());
-		String text;
+		return TextUtils.splitParagraphs(text, chunkNorm, chunkVar, false);
-		try {
-			text = Utils.normalizeText(Files.readString(file));
-		}
-		catch (final IOException e) {
-			throw new RuntimeException("Cannot read file", e);
-		}
-		
-		return Utils.splitParagraphs(text, chunkNorm, chunkVar, false);
 	}
 	
 	protected Map<String, String> createCharacterNameMapping(final String paragraph) {
@@ -141,7 +135,7 @@ abstract class AbstractContextualPrompts implements InitializingBean, Consumer<S
 				instruction,
 				paragraph);
 		
-		return Utils.normalizeModelResponse(completion.response(), true);
+		return TextUtils.normalizeModelResponse(completion.response(), true);
 	}
 	
 	protected void writeJsonLine(final ObjectMapper mapper, final OutputStream fos, final Object line) {

src/main/java/de/extio/lmdatasetprep/tools/DatasetTool.java

@@ -0,0 +1,8 @@
+package de.extio.lmdatasetprep.tools;
+
+import java.util.Properties;
+import java.util.function.Consumer;
+
+public interface DatasetTool extends Consumer<Properties> {
+	
+}

src/main/java/de/extio/lmdatasetprep/tools/MergeJsonl.java

@@ -1,4 +1,4 @@
-package de.extio.lmdatasetprep.preparer;
+package de.extio.lmdatasetprep.tools;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
@@ -7,33 +7,34 @@ import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Properties;
 import java.util.concurrent.ThreadLocalRandom;
-import java.util.function.Consumer;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
 
+import de.extio.lmdatasetprep.Execution;
+
 @Component
-public class MergeJsonl implements Consumer<String[]>, DatasetTool {
+public class MergeJsonl implements DatasetTool {
 	
 	private static final byte[] NEWLINE = "\n".getBytes(StandardCharsets.UTF_8);
 	
 	private static final Logger LOGGER = LoggerFactory.getLogger(MergeJsonl.class);
 	
 	@Override
-	public void accept(final String[] args) {
+	public void accept(final Properties properties) {
-		if (args.length < 3) {
-			LOGGER.error("Arguments are missing. <Path to output dir> [ <Path to jsonl files>, ... ]");
-			return;
-		}
-		
 		final List<String> lines = Collections.synchronizedList(new ArrayList<>());
-		for (int i = 2; i < args.length; i++) {
+		for (int i = 0; i < 99; i++) {
-			Utils.transformDirectory(args[i], p -> {
+			final String source = properties.getProperty("mergeJsonl.source." + i, "");
+			if (source.isBlank()) {
+				break;
+			}
+			Execution.transform(source, p -> {
 				return List.of(() -> {
 					try {
-						lines.addAll(Files.readAllLines(p));
+						lines.addAll(Files.readAllLines(p.file()));
 					}
 					catch (final IOException e) {
 						throw new RuntimeException("Cannot read file", e);
@@ -45,16 +46,18 @@ public class MergeJsonl implements Consumer<String[]>, DatasetTool {
 		
 		Collections.shuffle(lines, ThreadLocalRandom.current());
 		
-		final Path out = Path.of(args[1], "dataset-merged.jsonl");
+		final Path out = Path.of("dataset-merged.jsonl");
-		try (var fos = Files.newOutputStream(out)) {
+		Execution.streamOut(out, "mergeJsonl.destination", properties, fos -> {
-			for (final String line : lines) {
+			try {
-				fos.write(line.getBytes(StandardCharsets.UTF_8));
+				for (final String line : lines) {
-				fos.write(NEWLINE);
+					fos.write(line.getBytes(StandardCharsets.UTF_8));
+					fos.write(NEWLINE);
+				}
 			}
-		}
+			catch (final IOException e1) {
-		catch (final IOException e1) {
+				LOGGER.error("IO exception", e1);
-			LOGGER.error("IO exception", e1);
+			}
-		}
+		});
 	}
 	
 }

src/main/java/de/extio/lmdatasetprep/tools/Rewrite.java

@@ -1,11 +1,11 @@
-package de.extio.lmdatasetprep.preparer;
+package de.extio.lmdatasetprep.tools;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Properties;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
@@ -15,11 +15,14 @@ import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import de.extio.lmdatasetprep.Execution;
+import de.extio.lmdatasetprep.Execution.WorkPacket;
+import de.extio.lmdatasetprep.TextUtils;
 import de.extio.lmlib.client.ClientService;
 import de.extio.lmlib.profile.ModelCategory;
 
 @Component
-public class Rewrite implements Consumer<String[]>, DatasetTool {
+public class Rewrite implements DatasetTool {
 	
 	private static final Logger LOGGER = LoggerFactory.getLogger(Rewrite.class);
 	
@@ -50,40 +53,32 @@ public class Rewrite implements Consumer<String[]>, DatasetTool {
 	private ClientService clientService;
 	
 	@Override
-	public void accept(final String[] args) {
+	public void accept(final Properties properties) {
-		if (args.length < 5) {
+		Execution.transform(properties.getProperty("rewrite.source"),
-			LOGGER.error("Arguments are missing. <Path to text files> <Path to output dir> <Texts per input> <Model suffix>");
+				packet -> {
-			return;
-		}
-		
-		Utils.transformDirectory(args[1],
-				f -> {
 					final List<Runnable> tasks = new ArrayList<>();
-					tasks.add(() -> this.rewriteFile(args, f, 0, "impr", ModelCategory.COLD, IMPROVEMENT_PROMPT));
+					tasks.add(() -> this.rewriteFile(properties, packet, 0, "impr", ModelCategory.COLD, IMPROVEMENT_PROMPT));
-					for (int i = 0; i < Integer.parseInt(args[3]); i++) {
+					for (int i = 0; i < Integer.parseInt(properties.getProperty("rewrite.cnt")); i++) {
 						final int fi = i;
-						tasks.add(() -> this.rewriteFile(args, f, fi, "enh", ModelCategory.COLD, String.format(ENHANCE_PROMPT, ENHANCEMENTS.get(ThreadLocalRandom.current().nextInt(ENHANCEMENTS.size())))));
+						tasks.add(() -> this.rewriteFile(properties, packet, fi, "enh", ModelCategory.COLD, String.format(ENHANCE_PROMPT, ENHANCEMENTS.get(ThreadLocalRandom.current().nextInt(ENHANCEMENTS.size())))));
 					}
 					return tasks;
 				});
 	}
 	
-	private void rewriteFile(final String[] args, final Path f, final int i, final String identifier, final ModelCategory modelCategory, final String prompt) {
+	private void rewriteFile(final Properties properties, final WorkPacket packet, final int i, final String identifier, final ModelCategory modelCategory, final String prompt) {
-		final Path out = Utils.suffixFilename(Path.of(args[2]).resolve(f.getFileName()),
+		final Path out = Execution.suffixFilename(packet.file(),
 				"rewr",
-				args[4],
+				properties.getProperty("rewrite.model"),
 				identifier,
 				modelCategory.toString().toLowerCase(),
 				String.valueOf(i));
-		if (Files.exists(out)) {
-			LOGGER.info("Skipping " + out);
-			return;
-		}
 		
-		LOGGER.info("Rewriting " + f.getFileName() + " " + identifier + " " + modelCategory.name() + " " + i);
+		LOGGER.info("Rewriting " + packet.file().getFileName() + " " + identifier + " " + modelCategory.name() + " " + i);
-		Utils.streamOut(out, fos -> {
+		
+		Execution.streamOut(out, "rewrite.destination", properties, fos -> {
 			final AtomicBoolean first = new AtomicBoolean(true);
-			this.rewrite(f, prompt, modelCategory, chunk -> {
+			this.rewrite(packet, prompt, modelCategory, chunk -> {
 				try {
 					if (!first.getAndSet(false)) {
 						fos.write(PARAGRAPH);
@@ -97,17 +92,9 @@ public class Rewrite implements Consumer<String[]>, DatasetTool {
 		});
 	}
 	
-	void rewrite(final Path file, final String prompt, final ModelCategory modelCategory, final Consumer<String> consumer) {
+	void rewrite(final WorkPacket packet, final String prompt, final ModelCategory modelCategory, final Consumer<String> consumer) {
-		String text;
+		final String text = TextUtils.normalizeText(packet.text());
-		try {
+		final List<String> splits = TextUtils.splitParagraphs(text, 1250, 350, false);
-			text = Utils.normalizeText(Files.readString(file));
-		}
-		catch (final IOException e) {
-			throw new RuntimeException("Cannot read file", e);
-		}
-		
-		final List<String> splits = Utils.splitParagraphs(text, 1250, 350, false);
-		
 		for (final String split : splits) {
 			LOGGER.info("Split " + (splits.indexOf(split) + 1) + "/" + splits.size());
 			final var completion = this.clientService.getClient(modelCategory).completion(modelCategory,

src/main/java/de/extio/lmdatasetprep/tools/Text2Jsonl.java

@@ -1,12 +1,11 @@
-package de.extio.lmdatasetprep.preparer;
+package de.extio.lmdatasetprep.tools;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.function.Consumer;
+import java.util.Properties;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -15,25 +14,24 @@ import org.springframework.stereotype.Component;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import de.extio.lmdatasetprep.Execution;
+import de.extio.lmdatasetprep.Execution.WorkPacket;
+import de.extio.lmdatasetprep.TextUtils;
+
 @Component
-public class Text2Jsonl implements Consumer<String[]>, DatasetTool {
+public class Text2Jsonl implements DatasetTool {
 	
 	private static final Logger LOGGER = LoggerFactory.getLogger(Text2Jsonl.class);
 	
 	private static final byte[] NEWLINE = "\n".getBytes(StandardCharsets.UTF_8);
 	
 	@Override
-	public void accept(final String[] args) {
+	public void accept(final Properties properties) {
-		if (args.length < 5) {
+		final Path out = Path.of("dataset.jsonl");
-			LOGGER.error("Arguments are missing. <Path to text files> <Path to output dir> <chunk norm> <chunk var>");
+		Execution.streamOut(out, "text2Jsonl.destination", properties, fos -> {
-			return;
+			Execution.transform(properties.getProperty("text2Jsonl.source"), p -> {
-		}
-		
-		final Path out = Path.of(args[2], "dataset.jsonl");
-		try (var fos = Files.newOutputStream(out)) {
-			Utils.transformDirectory(args[1], p -> {
 				return List.of(() -> {
-					final List<String> splits = this.splitToJsonl(p, Integer.parseInt(args[3]), Integer.parseInt(args[4]));
+					final List<String> splits = this.splitToJsonl(p, Integer.parseInt(properties.getProperty("text2Jsonl.chunkNorm")), Integer.parseInt(properties.getProperty("text2Jsonl.chunkVar")));
 					synchronized (this) {
 						try {
 							for (final String split : splits) {
@@ -47,27 +45,18 @@ public class Text2Jsonl implements Consumer<String[]>, DatasetTool {
 					}
 				});
 			});
-		}
+		});
-		catch (final IOException e1) {
-			LOGGER.error("IO exception", e1);
-		}
 	}
 	
-	List<String> splitToJsonl(final Path file, final int chunkNorm, final int chunkVar) {
+	List<String> splitToJsonl(final WorkPacket packet, final int chunkNorm, final int chunkVar) {
-		LOGGER.info("Splitting to jsonl " + file);
+		LOGGER.info("Splitting to jsonl " + packet.file());
 		
-		String text;
+		final String text = TextUtils.normalizeText(packet.text());
-		try {
-			text = Utils.normalizeText(Files.readString(file));
-		}
-		catch (final IOException e) {
-			throw new RuntimeException("Cannot read file", e);
-		}
 		
 		final List<String> jsonl = new ArrayList<>();
 		final ObjectMapper mapper = new ObjectMapper();
 		
-		final List<String> paragraphs = Utils.splitParagraphs(text, chunkNorm, chunkVar, true);
+		final List<String> paragraphs = TextUtils.splitParagraphs(text, chunkNorm, chunkVar, true);
 		for (int i = 0; i < paragraphs.size(); i++) {
 			final String split = paragraphs.get(i);
 			try {

src/main/java/de/extio/lmdatasetprep/tools/Text2Jsonl2HistoryInstructCompletion.java

@@ -1,9 +1,9 @@
-package de.extio.lmdatasetprep.preparer;
+package de.extio.lmdatasetprep.tools;
 
-import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Properties;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -11,13 +11,15 @@ import org.springframework.stereotype.Component;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import de.extio.lmdatasetprep.Execution;
+
 @Component
-public class Text2Jsonl2HistoryInstructCompletion extends AbstractContextualPrompts implements DatasetTool {
+public class Text2Jsonl2HistoryInstructCompletion extends AbstractContextualPrompts {
 	
 	private static final Logger LOGGER = LoggerFactory.getLogger(Text2Jsonl2HistoryInstructCompletion.class);
 	
 	@Override
-	protected CreateTasks createTasks(final String[] args, final List<ChunkCfg> chunkConfigurations, final int variations) {
+	protected CreateTasks createTasks(final Properties properties, final List<ChunkCfg> chunkConfigurations, final int variations) {
 		return p -> {
 			final ObjectMapper mapper = new ObjectMapper();
 			final List<Runnable> tasks = new ArrayList<>();
@@ -26,19 +28,15 @@ public class Text2Jsonl2HistoryInstructCompletion extends AbstractContextualProm
 				final List<String> paragraphs = this.fileToParagraphs(p, chunkCfg.chunksNorm(), chunkCfg.chunksVar());
 				
 				for (int i = 0; i < variations; i++) {
-					final Path out = Utils.suffixFilename(Path.of(args[2]).resolve(p.getFileName()),
+					final Path out = Execution.suffixFilename(p.file().getFileName(),
 							"histinstrcompl",
-							args[3],
+							properties.getProperty("contextualPrompts.model"),
 							String.valueOf(chunkCfg.chunksNorm()),
 							String.valueOf(i),
 							".jsonl");
-					if (Files.exists(out)) {
-						LOGGER.info("Skipping {}", out);
-						continue;
-					}
 					
 					tasks.add(() -> {
-						Utils.streamOut(out, fos -> {
+						Execution.streamOut(out, "contextualPrompts.destination", properties, fos -> {
 							for (int j = 1; j < paragraphs.size(); j++) {
 								LOGGER.info("Paragraph " + (j + 1) + "/" + paragraphs.size());
 								

src/main/java/de/extio/lmdatasetprep/tools/Text2Jsonl2WithContextualPrompts.java

@@ -1,10 +1,9 @@
-package de.extio.lmdatasetprep.preparer;
+package de.extio.lmdatasetprep.tools;
 
-import java.io.IOException;
-import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Properties;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -12,13 +11,15 @@ import org.springframework.stereotype.Component;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import de.extio.lmdatasetprep.Execution;
+
 @Component
-public class Text2Jsonl2WithContextualPrompts extends AbstractContextualPrompts implements DatasetTool {
+public class Text2Jsonl2WithContextualPrompts extends AbstractContextualPrompts {
 	
 	private static final Logger LOGGER = LoggerFactory.getLogger(Text2Jsonl2WithContextualPrompts.class);
 	
 	@Override
-	protected CreateTasks createTasks(final String[] args, final List<ChunkCfg> chunkConfigurations, final int variations) {
+	protected CreateTasks createTasks(final Properties properties, final List<ChunkCfg> chunkConfigurations, final int variations) {
 		return p -> {
 			final ObjectMapper mapper = new ObjectMapper();
 			final List<Runnable> tasks = new ArrayList<>();
@@ -27,19 +28,15 @@ public class Text2Jsonl2WithContextualPrompts extends AbstractContextualPrompts
 				final List<String> paragraphs = this.fileToParagraphs(p, chunkCfg.chunksNorm(), chunkCfg.chunksVar());
 				
 				for (int i = 0; i < variations; i++) {
-					final Path out = Utils.suffixFilename(Path.of(args[2]).resolve(p.getFileName()),
+					final Path out = Execution.suffixFilename(p.file().getFileName(),
 							"contextualds",
-							args[3],
+							properties.getProperty("contextualPrompts.model"),
 							String.valueOf(chunkCfg.chunksNorm()),
 							String.valueOf(i),
 							".jsonl");
-					if (Files.exists(out)) {
-						LOGGER.info("Skipping {}", out);
-						continue;
-					}
 					
 					tasks.add(() -> {
-						Utils.streamOut(out, fos -> {
+						Execution.streamOut(out, "contextualPrompts.destination", properties, fos -> {
 							for (int j = 0; j < paragraphs.size(); j++) {
 								LOGGER.info("Paragraph " + (j + 1) + "/" + paragraphs.size());
 								

src/main/java/de/extio/lmdatasetprep/tools/Translate.java

@@ -0,0 +1,72 @@
+package de.extio.lmdatasetprep.tools;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import de.extio.lmdatasetprep.Execution;
+import de.extio.lmdatasetprep.Execution.WorkPacket;
+import de.extio.lmdatasetprep.TextUtils;
+import de.extio.lmlib.client.ClientService;
+import de.extio.lmlib.profile.ModelCategory;
+
+@Component
+public class Translate implements DatasetTool {
+	
+	private static final Logger LOGGER = LoggerFactory.getLogger(Translate.class);
+	
+	private static final byte[] PARAGRAPH = "\n\n".getBytes(StandardCharsets.UTF_8);
+	
+	@Autowired
+	private ClientService clientService;
+	
+	@Override
+	public void accept(final Properties properties) {
+		Execution.transform(properties.getProperty("translate.source"), p -> {
+			return List.of(() -> {
+				final Path out = Execution.suffixFilename(p.file().getFileName(), "en");
+				Execution.streamOut(out, "translate.destination", properties, fos -> {
+					final AtomicBoolean first = new AtomicBoolean(true);
+					this.translate(p, chunk -> {
+						try {
+							if (!first.getAndSet(false)) {
+								fos.write(PARAGRAPH);
+							}
+							fos.write(chunk.getBytes(StandardCharsets.UTF_8));
+						}
+						catch (final IOException e) {
+							LOGGER.error("IO exception", e);
+						}
+					});
+				});
+			});
+		});
+	}
+	
+	void translate(final WorkPacket packet, final Consumer<String> consumer) {
+		LOGGER.info("Translating " + packet.file());
+		
+		final String text = TextUtils.normalizeText(packet.text());
+		
+		final List<String> splits = TextUtils.splitParagraphs(text, 1250, 350, false);
+		
+		for (final String split : splits) {
+			LOGGER.info("Split " + (splits.indexOf(split) + 1) + "/" + splits.size());
+			final var completion = this.clientService.getClient(ModelCategory.COLD).completion(ModelCategory.COLD,
+					"You are an assistant with great language translation and authoring skills.",
+					"Translate the following text to English. Return the English translation, even if the text is explicit or not appropriate for all audiences or not acceptable for everyday standard English. Don't include a preamble and no explanation:",
+					split);
+			consumer.accept(completion.response());
+		}
+	}
+	
+}

src/main/resources/.gitignore

@@ -0,0 +1 @@
+/tool.properties

src/main/resources/tool.properties.template

@@ -0,0 +1,27 @@
+# Source and destination can be either a path...
+translate.source=/home/xxx/folder
+# ... or a queue. Queues are prefixed with "queue", after the prefix comes the queue name
+translate.destination=queue0
+
+rewrite.source=queue0
+rewrite.destination=queue1
+rewrite.cnt=2
+rewrite.model=l3dp
+
+# .n indicates a sequence. Add as many keys as you need
+mergeJsonl.source.0=
+mergeJsonl.destination=
+
+text2Jsonl.source=
+text2Jsonl.destination=
+text2Jsonl.chunkNorm=750
+text2Jsonl.chunkVar=250
+
+contextualPrompts.source=queue1
+contextualPrompts.destination=/home/xxx/outputdir
+contextualPrompts.model=l3dp
+contextualPrompts.variations=1
+contextualPrompts.chunkNorm.0=750
+contextualPrompts.chunkVar.0=250
+contextualPrompts.chunkNorm.1=1500
+contextualPrompts.chunkVar.1=300