diff --git a/guides/semantic-operators-in-Wayang.md b/guides/semantic-operators-in-Wayang.md new file mode 100644 index 000000000..1a60faa77 --- /dev/null +++ b/guides/semantic-operators-in-Wayang.md @@ -0,0 +1,156 @@ + + +# Developing with Semantic Operators in Apache Wayang + +This guide explains how to define semantic operators, provide executable implementations, register multiple implementations, estimate their costs, and let Apache Wayang select an implementation during optimization. + +The example uses a semantic filter that classifies movie reviews as positive or negative through Ollama. + +## 1. Semantic operators + +A semantic operator is an operator much like any Wayang operator, however, it takes a prompt as input, +that describes how it should act. + +For example: + +```java +.semanticFilter( + "Analyze the review after the | and write either " + + "\"POSITIVE\" if the review has a positive sentiment and " + + "\"NEGATIVE\" if the review has a negative sentiment." +) +``` + +The prompt describes the task. A `SemanticAlgorithm` provides one concrete implementation of that task. +A `SemanticAlgorithm` could theoretically be any UDF you desire, there are no strict requirements on its implementation. +The implementation only that the requires the UDF is implemented as something that takes an input `Record` and a `prompt` and outputs +whatever datatype is required by the operator. + +The Ollama local open-source model is used as an example for this guide, but may also be useful for quick development. + +We set up our local model hosting locally using Docker: + +```yaml +ollama: + image: ollama/ollama:latest + container_name: apache-wayang-ollama + ports: + - "11434:11434" + volumes: + - ollama-data:/root/.ollama + - ./docker/ollama-init.sh:/ollama-init.sh + entrypoint: ["/bin/bash", "/ollama-init.sh"] + restart: always + tty: true + networks: + - wayang-network +``` + +```sh +#!/bin/bash +ollama serve & +sleep 10 +ollama pull tinyllama +wait +``` + +We setup the backend call to the model in Wayang: + +```java +private static String callOllama(final String prompt) throws IOException, InterruptedException { + final String requestBody = String.format("{\"model\": \"%s\", \"prompt\": \"%s\", \"stream\": false}", + MODEL_NAME, escapeJson(prompt)); + + final HttpRequest request = HttpRequest.newBuilder().uri(URI.create(OLLAMA_API_URL)) + .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString(requestBody)) + .timeout(Duration.ofMinutes(2)).build(); + + final HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + + if (response.statusCode() != 200) { + throw new IOException("Ollama API error: " + response.body()); + } + + return parseOllamaResponse(response.body()); +} +``` + +and the semantic UDF: + +```java +public static boolean isPositiveSentiment(final Review review, final String prompt) + throws IOException, InterruptedException { + final String response = callOllama(prompt + " | " + review.getReviewText()); + return response.contains("POSITIVE"); +} +``` + +Now we can define our `SemanticAlgorithm`: + +```java +final SemanticAlgorithm ollamaFilter = new SemanticAlgorithm(); +ollamaFilter.impl = (input, prompt) -> { + try { + return isPositiveSentiment((Review) input); + } catch (IOException | InterruptedException e) { + throw new RuntimeException("Ollama call failed", e); + } +}; +``` + +And also provide a UDF load estimator: + +```java +ollamaFilter.loadProfileEstimator = + LoadProfileEstimators.createFromSpecification( + "wayang.semantic.ollama.model1.load", + configuration + ); +``` + +Note that you should provide a separate configuration for each semantic operator. +Now we register these UDFs with the `SemanticPlugin` this automatically constructs a new operator +per UDF. Please note this may have implications for optimization time depending on your setup. + +```java +final SemanticPlugin plugin = Semantic.plugin() + .withOperatorMapping(SemanticFilterOperator.class, ollamaFilter) + .withOperatorMapping(SemanticFilterOperator.class, ollamaFilter2) + .withOperatorMapping(SemanticFilterOperator.class, ollamaFilter3); + +final WayangContext wayangContext = new WayangContext() + .withPlugin(Java.basicPlugin()) + .withPlugin(plugin); +``` + +Currently, we only have mappings for semantic operators in Java, so you also need the `Java.basicPlugin()`. +Finally, you can construct your Wayang plan: + +```java +final Collection positiveReviewCnt = planBuilder.loadCollection(loadReviews()) + .filter(review -> "taken_3".equals(review.getId())) + .semanticFilter("Analyze the review after the | and write either \"POSITIVE\" if the review has a positive sentiment and \"NEGATIVE\" if the review has a negative sentiment.") + .withTargetModels(ollamaFilter, ollamaFilter2, ollamaFilter3) + .count() + .collect(); +``` + +You need to provide semantic operators with their target models, even if you plan to use all models you've constructed. + diff --git a/pom.xml b/pom.xml index f0d5cf77e..22a666e5b 100644 --- a/pom.xml +++ b/pom.xml @@ -23,7 +23,7 @@ org.apache apache - 32 + 34 org.apache.wayang diff --git a/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuanta.scala b/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuanta.scala index fb2dfeb77..f498d3eb4 100644 --- a/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuanta.scala +++ b/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuanta.scala @@ -44,6 +44,7 @@ import com.google.protobuf.ByteString import org.apache.wayang.api.python.function._ import org.tensorflow.ndarray.NdArray +import scala.collection.JavaConverters._ import scala.collection.JavaConversions import scala.collection.JavaConversions._ import scala.reflect._ @@ -633,6 +634,38 @@ class DataQuanta[Out: ClassTag](val operator: ElementaryOperator, outputIndex: I joinOperator } + def semanticFilterPrompt(prompt: String): DataQuanta[Out] = { + val dataSetType = org.apache.wayang.core.types.DataSetType.createDefault( + this.output.getType.getDataUnitType.toBasicDataUnitType + ) + + val filterOperator = new SemanticFilterOperator[Out]( + dataSetType, + prompt + ) + + this.connectTo(filterOperator, 0) + wrap[Out](filterOperator) + } + + def semanticFilterPrompt(prompt: String, targetModels: AnyRef * ): DataQuanta[Out] = { + val dataSetType = org.apache.wayang.core.types.DataSetType.createDefault( + this.output.getType.getDataUnitType.toBasicDataUnitType + ) + + val targetModelsSet: java.util.Set[Object] = targetModels.toSet.asJava.asInstanceOf[java.util.Set[Object]] + + val filterOperator = new SemanticFilterOperator[Out]( + dataSetType, + prompt, + targetModelsSet + ) + + this.connectTo(filterOperator, 0) + wrap[Out](filterOperator) + } + + /** * Applies a spatial filter to this instance. * diff --git a/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuantaBuilder.scala b/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuantaBuilder.scala index 9d37aa930..c399240fb 100644 --- a/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuantaBuilder.scala +++ b/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuantaBuilder.scala @@ -45,10 +45,9 @@ import org.apache.iceberg.Schema import org.apache.iceberg.FileFormat import org.apache.iceberg.catalog.{Catalog, TableIdentifier} - - import scala.collection.mutable.ListBuffer import scala.reflect.ClassTag +import scala.annotation.varargs /** * Trait/interface for builders of [[DataQuanta]]. The purpose of the builders is to provide a convenient @@ -282,6 +281,9 @@ trait DataQuantaBuilder[+This <: DataQuantaBuilder[_, Out], Out] extends Logging thatKeyUdf: SerializableFunction[ThatOut, Key]) = new JoinDataQuantaBuilder(this, that, thisKeyUdf, thatKeyUdf) + def semanticFilter(prompt: String) = + new SemanticFilterDataQuantaBuilder[Out](this, prompt) + /** * Feed the built [[DataQuanta]] into a spatial filter operator. * Requires the wayang-spatial plugin to be loaded. @@ -2146,6 +2148,42 @@ class KeyedDataQuantaBuilder[Out, Key](private val dataQuantaBuilder: DataQuanta } +/** + * [[DataQuantaBuilder]] implementation for [[org.apache.wayang.basic.operators.SemanticFilterOperator]]s. + * + * @param inputDataQuanta [[DataQuantaBuilder]] for the input [[DataQuanta]] + * @param udf UDF for the [[SemanticFilterOperator]] + */ +class SemanticFilterDataQuantaBuilder[T](inputDataQuanta: DataQuantaBuilder[_, T], prompt: String) + (implicit javaPlanBuilder: JavaPlanBuilder) + extends BasicDataQuantaBuilder[SemanticFilterDataQuantaBuilder[T], T] { + + // Reuse the input TypeTrap to enforce type equality between input and output. + override def getOutputTypeTrap: TypeTrap = inputDataQuanta.outputTypeTrap + + /** [[LoadProfileEstimator]] to estimate the [[LoadProfile]] of the [[udf]]. */ + private var udfLoadProfileEstimator: LoadProfileEstimator = _ + + /** Selectivity of the filter predicate. */ + private var selectivity: ProbabilisticDoubleInterval = _ + + /* + + */ + private val targetModels: ListBuffer[AnyRef] = ListBuffer() + + @varargs def withTargetModels(models: AnyRef *): SemanticFilterDataQuantaBuilder[T] = { + models.foreach(targetModels.+=_) + this + } + + override protected def build = applyTargetPlatforms( + inputDataQuanta.dataQuanta() + .semanticFilterPrompt(prompt, targetModels: _ *), + this.getTargetPlatforms() + ) + } + class SpatialFilterDataQuantaBuilder[T](inputDataQuanta: DataQuantaBuilder[_, T], keySelector: SerializableFunction[T, _ <: SpatialGeometry], predicateType: SpatialPredicate, diff --git a/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticFilterOperator.java b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticFilterOperator.java new file mode 100644 index 000000000..f98898428 --- /dev/null +++ b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticFilterOperator.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.wayang.basic.operators; + +import java.util.Set; + +import org.apache.wayang.core.plan.wayangplan.UnaryToUnaryOperator; +import org.apache.wayang.core.types.DataSetType; + +public class SemanticFilterOperator extends UnaryToUnaryOperator implements SemanticOperator { + + private final String prompt; + + private final Set targetModels; + + public SemanticFilterOperator(final DataSetType type, final String prompt) { + super(type, type, false); + this.prompt = prompt; + this.targetModels = null; + } + + public SemanticFilterOperator(final DataSetType type, final String prompt, final Set targetModels) { + super(type, type, false); + this.prompt = prompt; + this.targetModels = targetModels; + } + + public SemanticFilterOperator(final DataSetType type) { + super(type, type, false); + this.prompt = ""; + this.targetModels = null; + } + + public SemanticFilterOperator(final DataSetType inputType, final DataSetType outputType, + final boolean isSupportingBroadcastInputs) { + super(inputType, outputType, isSupportingBroadcastInputs); + this.prompt = ""; + this.targetModels = null; + } + + public SemanticFilterOperator(final UnaryToUnaryOperator that) { + super(that); + this.prompt = ""; + this.targetModels = null; + } + + @Override + public String getPrompt() { + return prompt; + } + + @Override + public Set getTargetModels() { + return targetModels; + } + + @Override + public void addTargetModel(final Object model) { + targetModels.add(model); + } +} diff --git a/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticOperator.java b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticOperator.java new file mode 100644 index 000000000..29c00c80a --- /dev/null +++ b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticOperator.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.wayang.basic.operators; + +import java.util.Set; + +/** + * Common contract for operators whose UDF is not plain Java code but a natural-language + * {@code prompt} that gets executed by a large language model (LLM), e.g. a + * {@link SemanticFilterOperator} or a future {@code SemanticJoinOperator}. + *

+ * This is deliberately kept as a slim interface rather than an abstract base class. Semantic + * operators still need to pick the operator shape that fits their arity/semantics, e.g. a + * filter extends {@link org.apache.wayang.core.plan.wayangplan.UnaryToUnaryOperator} while a + * join would extend {@link org.apache.wayang.core.plan.wayangplan.BinaryToUnaryOperator}. As + * Java only allows single class inheritance, tying this abstraction to a specific operator + * base class would force every semantic operator into the same shape. Implementors are free + * to (and are expected to) extend whichever operator base class matches their actual + * input/output arity, while still sharing the prompt/target-model contract defined here. + */ +public interface SemanticOperator { + + /** + * @return the prompt describing the semantic algorithm that this operator should execute + */ + String getPrompt(); + + /** + * @return the set of models that are allowed to execute this operator's {@link #getPrompt()}, + * or {@code null} if no target models have been configured + */ + Set getTargetModels(); + + /** + * Adds a model to the set of models that are allowed to execute this operator's + * {@link #getPrompt()}. + * + * @param model the model to add + */ + void addTargetModel(Object model); +} diff --git a/wayang-platforms/pom.xml b/wayang-platforms/pom.xml index 9c5e29545..8665ceffb 100644 --- a/wayang-platforms/pom.xml +++ b/wayang-platforms/pom.xml @@ -45,6 +45,7 @@ wayang-generic-jdbc wayang-presto wayang-tensorflow + wayang-semantic diff --git a/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaFilterOperator.java b/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaFilterOperator.java index 757220e34..a11e7625d 100644 --- a/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaFilterOperator.java +++ b/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaFilterOperator.java @@ -50,6 +50,16 @@ public class JavaFilterOperator implements JavaExecutionOperator { + + /** + * Creates a new instance. + * + * @param type type of the dataset elements + */ + public JavaFilterOperator(PredicateDescriptor predicateDescriptor) { + super(predicateDescriptor); + } + /** * Creates a new instance. * @@ -119,5 +129,4 @@ public List getSupportedOutputChannels(int index) { assert index <= this.getNumOutputs() || (index == 0 && this.getNumOutputs() == 0); return Collections.singletonList(StreamChannel.DESCRIPTOR); } - } diff --git a/wayang-platforms/wayang-semantic/pom.xml b/wayang-platforms/wayang-semantic/pom.xml new file mode 100644 index 000000000..e32f80676 --- /dev/null +++ b/wayang-platforms/wayang-semantic/pom.xml @@ -0,0 +1,75 @@ + + + + 4.0.0 + + + wayang-platforms + org.apache.wayang + 1.1.2-SNAPSHOT + + + wayang-semantic + + Wayang Platform Semantic + + Wayang implementation of semantic operators + + + + org.apache.wayang.platform.semantic + + + + + org.apache.wayang + wayang-java + 1.1.2-SNAPSHOT + + + org.apache.wayang + wayang-basic + 1.1.2-SNAPSHOT + + + org.apache.wayang + wayang-java + 1.1.2-SNAPSHOT + + + org.junit.jupiter + junit-jupiter + 5.10.2 + test + + + com.fasterxml.jackson.core + jackson-core + 2.16.1 + test + + + org.apache.wayang + wayang-api-scala-java + 1.1.2-SNAPSHOT + test + + + diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/Semantic.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/Semantic.java new file mode 100644 index 000000000..83f497008 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/Semantic.java @@ -0,0 +1,45 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.wayang.semantic; + +import org.apache.wayang.semantic.plugin.SemanticPlugin; +import org.apache.wayang.java.platform.JavaPlatform; + +public class Semantic { + private final static SemanticPlugin PLUGIN = new SemanticPlugin(); + + /** + * Retrieve the {@link SemanticPlugin}. + * + * @return the {@link SemanticPlugin} + */ + public static SemanticPlugin plugin() { + return PLUGIN; + } + + + /** + * Retrieve the {@link SemanticPlatform}. + * + * @return the {@link SemanticPlatform} + */ + public static JavaPlatform platform() { + return JavaPlatform.getInstance(); + } +} \ No newline at end of file diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/execution/OllamaExecutor.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/execution/OllamaExecutor.java new file mode 100644 index 000000000..706d90241 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/execution/OllamaExecutor.java @@ -0,0 +1,113 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.wayang.semantic.execution; + +import java.util.Arrays; +import java.util.Collection; +import java.util.List; + +import org.apache.wayang.core.api.Job; +import org.apache.wayang.core.api.exception.WayangException; +import org.apache.wayang.core.optimizer.OptimizationContext; +import org.apache.wayang.core.plan.executionplan.ExecutionTask; +import org.apache.wayang.core.plan.wayangplan.ExecutionOperator; +import org.apache.wayang.core.platform.ChannelInstance; +import org.apache.wayang.core.platform.Executor; +import org.apache.wayang.core.platform.PartialExecution; +import org.apache.wayang.core.platform.PushExecutorTemplate; +import org.apache.wayang.core.platform.lineage.ExecutionLineageNode; +import org.apache.wayang.core.util.Formats; +import org.apache.wayang.core.util.Tuple; +import org.apache.wayang.semantic.operators.OllamaExecutionOperator; +import org.apache.wayang.semantic.platform.OllamaPlatform; + +/** + * {@link Executor} implementation for the {@link OllamaPlatform}. + */ +public class OllamaExecutor extends PushExecutorTemplate { + + private final OllamaPlatform platform; + + public OllamaExecutor(final OllamaPlatform platform, final Job job) { + super(job); + this.platform = platform; + } + + @Override + public OllamaPlatform getPlatform() { + return this.platform; + } + + @Override + protected Tuple, PartialExecution> execute( + final ExecutionTask task, + final List inputChannelInstances, + final OptimizationContext.OperatorContext producerOperatorContext, + final boolean isRequestEagerExecution + ) { + final ChannelInstance[] outputChannelInstances = task.getOperator().createOutputChannelInstances( + this, task, producerOperatorContext, inputChannelInstances + ); + + final Collection executionLineageNodes; + final Collection producedChannelInstances; + this.job.reportProgress(task.getOperator().getName(), 50); + final long startTime = System.currentTimeMillis(); + try { + final Tuple, Collection> results = + cast(task.getOperator()).evaluate( + toArray(inputChannelInstances), + outputChannelInstances, + this, + producerOperatorContext + ); + executionLineageNodes = results.getField0(); + producedChannelInstances = results.getField1(); + } catch (Exception e) { + throw new WayangException(String.format("Executing %s failed.", task), e); + } + final long endTime = System.currentTimeMillis(); + final long executionDuration = endTime - startTime; + + this.job.reportProgress(task.getOperator().getName(), 100); + + final PartialExecution partialExecution = this.createPartialExecution(executionLineageNodes, executionDuration); + + if (partialExecution == null && executionDuration > 10) { + this.logger.warn("Execution of {} took suspiciously long ({}).", task, Formats.formatDuration(executionDuration)); + } + + this.registerMeasuredCardinalities(producedChannelInstances); + + if (isRequestEagerExecution && partialExecution == null) { + this.logger.info("{} was not executed eagerly as requested.", task); + } + + return new Tuple<>(Arrays.asList(outputChannelInstances), partialExecution); + } + + private static OllamaExecutionOperator cast(final ExecutionOperator executionOperator) { + return (OllamaExecutionOperator) executionOperator; + } + + private static ChannelInstance[] toArray(final List channelInstances) { + final ChannelInstance[] array = new ChannelInstance[channelInstances.size()]; + return channelInstances.toArray(array); + } +} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaExecutionOperator.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaExecutionOperator.java new file mode 100644 index 000000000..b635a95b4 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaExecutionOperator.java @@ -0,0 +1,58 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.wayang.semantic.operators; + +import java.util.Collection; + +import org.apache.wayang.core.optimizer.OptimizationContext; +import org.apache.wayang.core.plan.wayangplan.ExecutionOperator; +import org.apache.wayang.core.platform.ChannelInstance; +import org.apache.wayang.core.platform.lineage.ExecutionLineageNode; +import org.apache.wayang.core.util.Tuple; +import org.apache.wayang.semantic.execution.OllamaExecutor; +import org.apache.wayang.semantic.platform.OllamaPlatform; + +/** + * Execution operator for the {@link OllamaPlatform}. + */ +public interface OllamaExecutionOperator extends ExecutionOperator { + + @Override + default OllamaPlatform getPlatform() { + return OllamaPlatform.getInstance(); + } + + /** + * Evaluates this operator by calling out to Ollama. Mirrors + * {@code org.apache.wayang.java.operators.JavaExecutionOperator#evaluate}, just for the + * {@link OllamaPlatform} instead of the Java platform. + * + * @param inputs {@link ChannelInstance}s that satisfy the inputs of this operator + * @param outputs {@link ChannelInstance}s that collect the outputs of this operator + * @param ollamaExecutor that executes this instance + * @param operatorContext optimization information for this instance + * @return {@link Collection}s of what has been executed and produced + */ + Tuple, Collection> evaluate( + ChannelInstance[] inputs, + ChannelInstance[] outputs, + OllamaExecutor ollamaExecutor, + OptimizationContext.OperatorContext operatorContext); + +} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaFilterOperator.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaFilterOperator.java new file mode 100644 index 000000000..40d9ee7ae --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaFilterOperator.java @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.wayang.semantic.operators; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.apache.wayang.basic.operators.SemanticFilterOperator; +import org.apache.wayang.core.plan.wayangplan.ExecutionOperator; +import org.apache.wayang.core.platform.ChannelDescriptor; +import org.apache.wayang.core.types.DataSetType; +import org.apache.wayang.java.channels.CollectionChannel; +import org.apache.wayang.java.channels.StreamChannel; + +/** + * Base for {@link OllamaExecutionOperator}s that implement a {@link SemanticFilterOperator} by calling + * out to one specific model. Every concrete subclass hardcodes exactly one model's evaluation logic + * instead of taking it as an injected UDF, so {@link #getPrompt()} (inherited from the matched logical + * {@link SemanticFilterOperator}) remains the single source of truth for the prompt; a subclass is free + * to use it or to ignore it in favor of its own hardcoded behavior. + *

+ * Input/output channels are the same {@link StreamChannel}/{@link CollectionChannel} that + * {@code JavaFilterOperator} supports, so data can flow between the surrounding Java pipeline and an + * Ollama-backed operator without any channel conversion. + */ +public abstract class OllamaFilterOperator extends SemanticFilterOperator implements OllamaExecutionOperator { + + protected OllamaFilterOperator(final DataSetType type, final String prompt) { + super(type, prompt); + } + + /** + * Creates a new instance of the same concrete subclass for the given type/prompt. + */ + protected abstract OllamaFilterOperator newInstance(DataSetType type, String prompt); + + @Override + protected final ExecutionOperator createCopy() { + return this.newInstance(this.getInputType(), this.getPrompt()); + } + + @Override + public List getSupportedInputChannels(final int index) { + assert index <= this.getNumInputs() || (index == 0 && this.getNumInputs() == 0); + if (this.getInput(index).isBroadcast()) return Collections.singletonList(CollectionChannel.DESCRIPTOR); + return Arrays.asList(CollectionChannel.DESCRIPTOR, StreamChannel.DESCRIPTOR); + } + + @Override + public List getSupportedOutputChannels(final int index) { + assert index <= this.getNumOutputs() || (index == 0 && this.getNumOutputs() == 0); + return Collections.singletonList(StreamChannel.DESCRIPTOR); + } +} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/platform/OllamaPlatform.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/platform/OllamaPlatform.java new file mode 100644 index 000000000..1cad4d6b1 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/platform/OllamaPlatform.java @@ -0,0 +1,90 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.wayang.semantic.platform; + +import org.apache.wayang.core.api.Configuration; +import org.apache.wayang.core.optimizer.costs.LoadProfileToTimeConverter; +import org.apache.wayang.core.optimizer.costs.LoadToTimeConverter; +import org.apache.wayang.core.optimizer.costs.TimeToCostConverter; +import org.apache.wayang.core.platform.Executor; +import org.apache.wayang.core.platform.Platform; +import org.apache.wayang.core.util.ReflectionUtils; +import org.apache.wayang.semantic.execution.OllamaExecutor; + +/** + * {@link Platform} for operators that are executed by calling out to an Ollama-hosted LLM. + *

+ * Kept as its own {@link Platform} (rather than piggy-backing on + * {@link org.apache.wayang.java.platform.JavaPlatform}) so that the cost of an LLM call is modeled and + * optimized on its own terms, instead of being reported as if it were free Java-Stream work. + */ +public class OllamaPlatform extends Platform { + + private static final String PLATFORM_NAME = "Ollama"; + + private static final String CONFIG_NAME = "semantic.ollama"; + + private static final String DEFAULT_CONFIG_FILE = "wayang-semantic-ollama-defaults.properties"; + + private static OllamaPlatform instance = null; + + public static OllamaPlatform getInstance() { + if (instance == null) { + instance = new OllamaPlatform(); + } + return instance; + } + + private OllamaPlatform() { + super(PLATFORM_NAME, CONFIG_NAME); + } + + @Override + public void configureDefaults(final Configuration configuration) { + configuration.load(ReflectionUtils.loadResource(DEFAULT_CONFIG_FILE)); + } + + @Override + public Executor.Factory getExecutorFactory() { + return job -> new OllamaExecutor(this, job); + } + + @Override + public LoadProfileToTimeConverter createLoadProfileToTimeConverter(final Configuration configuration) { + final int cpuMhz = (int) configuration.getLongProperty("wayang.semantic.ollama.cpu.mhz"); + final int numCores = (int) configuration.getLongProperty("wayang.semantic.ollama.cores"); + final double hdfsMsPerMb = configuration.getDoubleProperty("wayang.semantic.ollama.hdfs.ms-per-mb"); + final double stretch = configuration.getDoubleProperty("wayang.semantic.ollama.stretch"); + return LoadProfileToTimeConverter.createTopLevelStretching( + LoadToTimeConverter.createLinearCoverter(1 / (numCores * cpuMhz * 1000d)), + LoadToTimeConverter.createLinearCoverter(hdfsMsPerMb / 1000000d), + LoadToTimeConverter.createLinearCoverter(0), + (cpuEstimate, diskEstimate, networkEstimate) -> cpuEstimate.plus(diskEstimate).plus(networkEstimate), + stretch + ); + } + + @Override + public TimeToCostConverter createTimeToCostConverter(final Configuration configuration) { + return new TimeToCostConverter( + configuration.getDoubleProperty("wayang.semantic.ollama.costs.fix"), + configuration.getDoubleProperty("wayang.semantic.ollama.costs.per-ms") + ); + } +} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.java new file mode 100644 index 000000000..2b0964f55 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.java @@ -0,0 +1,84 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.wayang.semantic.plugin; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + +import org.apache.wayang.core.api.Configuration; +import org.apache.wayang.core.mapping.Mapping; +import org.apache.wayang.core.mapping.PlanTransformation; +import org.apache.wayang.core.optimizer.channels.ChannelConversion; +import org.apache.wayang.core.platform.Platform; +import org.apache.wayang.core.plugin.Plugin; +import org.apache.wayang.java.Java; +import org.apache.wayang.java.platform.JavaPlatform; + +public class SemanticPlugin implements Plugin { + private final List mappings; + + private SemanticPlugin(final List mappings) { + this.mappings = mappings; + } + + public SemanticPlugin() { + this.mappings = List.of(); + } + + @Override + public Collection getMappings() { + return mappings; + } + + @Override + public Collection getRequiredPlatforms() { + final Set platforms = new LinkedHashSet<>(); + // The surrounding pipeline (sources, sinks, plain transformations) always runs on Java. + platforms.add(JavaPlatform.getInstance()); + for (final Mapping mapping : this.mappings) { + for (final PlanTransformation transformation : mapping.getTransformations()) { + platforms.addAll(transformation.getTargetPlatforms()); + } + } + return platforms; + } + + @Override + public Collection getChannelConversions() { + return Java.basicPlugin().getChannelConversions(); + } + + @Override + public void setProperties(final Configuration configuration) { + } + + /** + * Registers a {@link Mapping} that rewrites a semantic operator (e.g. a + * {@link org.apache.wayang.basic.operators.SemanticFilterOperator}) into a physical operator for one + * specific model implementation. + */ + public SemanticPlugin withMapping(final Mapping mapping) { + final List nextMappings = new ArrayList<>(this.mappings); + nextMappings.add(mapping); + return new SemanticPlugin(nextMappings); + } +} diff --git a/wayang-platforms/wayang-semantic/src/main/resources/wayang-semantic-ollama-defaults.properties b/wayang-platforms/wayang-semantic/src/main/resources/wayang-semantic-ollama-defaults.properties new file mode 100644 index 000000000..13106b24e --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/resources/wayang-semantic-ollama-defaults.properties @@ -0,0 +1,23 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +wayang.semantic.ollama.cpu.mhz = 2700 +wayang.semantic.ollama.cores = 1 +wayang.semantic.ollama.hdfs.ms-per-mb = 2.7 +wayang.semantic.ollama.stretch = 1 +wayang.semantic.ollama.costs.fix = 0.0 +wayang.semantic.ollama.costs.per-ms = 1.0 diff --git a/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java b/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java new file mode 100644 index 000000000..30f35ccf2 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java @@ -0,0 +1,481 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.wayang.semantic; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.stream.Stream; + +import org.apache.wayang.api.JavaPlanBuilder; +import org.apache.wayang.basic.operators.SemanticFilterOperator; +import org.apache.wayang.core.api.Configuration; +import org.apache.wayang.core.api.WayangContext; +import org.apache.wayang.core.mapping.Mapping; +import org.apache.wayang.core.mapping.OperatorPattern; +import org.apache.wayang.core.mapping.PlanTransformation; +import org.apache.wayang.core.mapping.ReplacementSubplanFactory; +import org.apache.wayang.core.mapping.SubplanPattern; +import org.apache.wayang.core.optimizer.OptimizationContext; +import org.apache.wayang.core.plan.wayangplan.ExecutionOperator; +import org.apache.wayang.core.platform.ChannelInstance; +import org.apache.wayang.core.platform.lineage.ExecutionLineageNode; +import org.apache.wayang.core.types.DataSetType; +import org.apache.wayang.core.util.Tuple; +import org.apache.wayang.java.Java; +import org.apache.wayang.java.channels.JavaChannelInstance; +import org.apache.wayang.java.channels.StreamChannel; +import org.apache.wayang.semantic.execution.OllamaExecutor; +import org.apache.wayang.semantic.operators.OllamaFilterOperator; +import org.apache.wayang.semantic.platform.OllamaPlatform; +import org.apache.wayang.semantic.plugin.SemanticPlugin; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import com.sun.net.httpserver.HttpServer; + +class SemBenchTest { + + /** + * In-process stand-in for a real Ollama server: it never leaves the JVM and never needs a model + * to be pulled, so the test runs unmodified in CI. It answers with the same response envelope a + * real Ollama server would ("{"response":"..."}"), classifying sentiment by a keyword heuristic + * on the prompt text, so the assertions below exercise genuine end-to-end behavior of the plan + * (filtering, the semantic operator, and JSON parsing) rather than a canned constant. + */ + private static final String[] NEGATIVE_KEYWORDS = {"terrible", "disappointed", "boring", "not recommend"}; + + private HttpServer mockOllamaServer; + + @BeforeEach + void startMockOllamaServer() throws IOException { + this.mockOllamaServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + this.mockOllamaServer.createContext("/api/generate", exchange -> { + final String requestBody = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8); + final boolean isNegative = Arrays.stream(NEGATIVE_KEYWORDS) + .anyMatch(keyword -> requestBody.toLowerCase().contains(keyword)); + final String responseBody = String.format("{\"response\":\"%s\"}", isNegative ? "NEGATIVE" : "POSITIVE"); + final byte[] responseBytes = responseBody.getBytes(StandardCharsets.UTF_8); + + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, responseBytes.length); + try (OutputStream responseStream = exchange.getResponseBody()) { + responseStream.write(responseBytes); + } + }); + this.mockOllamaServer.start(); + + final int port = this.mockOllamaServer.getAddress().getPort(); + OllamaSemanticFilter.setApiUrl("http://127.0.0.1:" + port + "/api/generate"); + } + + @AfterEach + void stopMockOllamaServer() { + this.mockOllamaServer.stop(0); + } + + private static List loadReviews() { + return Arrays.asList(new Review("taken_1", "The movie was fantastic. Great acting and an engaging story."), + new Review("taken_2", "I was disappointed. The plot was boring and too long."), + new Review("taken_3", "Absolutely loved it! One of the best movies I have seen this year."), + new Review("taken_3", "Terrible experience. I would not recommend it to anyone."), + new Review("taken_4", "It was okay. Not great, not terrible.")); + } + + @Test + void testSemBenchMoviesWithOllama() { + final Configuration configuration = new Configuration(); + configuration.setProperty("wayang.java.filter.load", """ + { + "in":1, + "out":1, + "cpu":"${25*in0 + 350000}", + "ram":"100000", + "p":0.9 + } + """ + ); + configuration.setProperty("wayang.semantic.ollama.model1.load", + """ + { + "in": 1, + "out": 1, + "cpu": "${500*in0 + 56789}", + "ram": "10000", + "disk": "0", + "net": "0", + "p": 0.9, + "overhead": 0, + "ru": "${wayang:logGrowth(0.1, 0.1, 1000000, in0)}" + } + """); + configuration.setProperty("wayang.semantic.ollama.model2.load", + """ + { + "in": 1, + "out": 1, + "cpu": "${500*in0 + 56789}", + "ram": "10000", + "disk": "0", + "net": "0", + "p": 0.9, + "overhead": 0, + "ru": "${wayang:logGrowth(0.1, 0.1, 1000000, in0)}" + } + """); + configuration.setProperty("wayang.semantic.ollama.model3.load", + """ + { + "in": 1, + "out": 1, + "cpu": "${50*in0 + 5678}", + "ram": "1000", + "disk": "0", + "net": "0", + "p": 0.9, + "overhead": 0, + "ru": "${wayang:logGrowth(0.1, 0.1, 1000000, in0)}" + } + """ + ); + + // Each Ollama model is its own physical operator + Mapping; the optimizer picks among them by cost. + // There is exactly one prompt, owned by the logical SemanticFilterOperator (set via .semanticFilter(...)), + // and each physical operator decides for itself whether it needs that prompt. + final SemanticPlugin plugin = Semantic.plugin() + .withMapping(new OllamaModel1FilterMapping()) + .withMapping(new OllamaModel2FilterMapping()) + .withMapping(new OllamaModel3FilterMapping()); + + final WayangContext wayangContext = new WayangContext(configuration) + .withPlugin(Java.basicPlugin()) + .withPlugin(plugin); + final JavaPlanBuilder planBuilder = new JavaPlanBuilder(wayangContext); + + final Collection positiveReviewCnt = planBuilder.loadCollection(loadReviews()) + .filter(review -> "taken_3".equals(review.getId())) + .semanticFilter("Analyze the review after the | and write either \"POSITIVE\" if the review has a positive sentiment and \"NEGATIVE\" if the review has a negative sentiment.") + .withTargetModels(OllamaModel1FilterOperator.class, OllamaModel2FilterOperator.class, OllamaModel3FilterOperator.class) + .count() + .collect(); + + // Of the two "taken_3" reviews, only the "Absolutely loved it!" one is positive; the mock + // Ollama server classifies by keyword, so this checks the plan actually ran the semantic + // filter rather than merely completing without error. + assertEquals(1, positiveReviewCnt.size()); + assertEquals(1L, positiveReviewCnt.iterator().next()); + } +} + +class Review { + private final String id; + private final String reviewText; + + public Review(final String id, final String reviewText) { + this.id = id; + this.reviewText = reviewText; + } + + public String getId() { + return id; + } + + public String getReviewText() { + return reviewText; + } + + @Override + public String toString() { + return "Review{id='" + id + "', reviewText='" + reviewText + "'}"; + } +} + +/** + * Physical operator for the "model1" Ollama implementation: a fixed sentiment prompt, ignores the + * logical operator's {@link #getPrompt()}. + */ +final class OllamaModel1FilterOperator extends OllamaFilterOperator { + + OllamaModel1FilterOperator(final DataSetType type, final String prompt) { + super(type, prompt); + } + + @Override + @SuppressWarnings("unchecked") + public Tuple, Collection> evaluate( + final ChannelInstance[] inputs, + final ChannelInstance[] outputs, + final OllamaExecutor ollamaExecutor, + final OptimizationContext.OperatorContext operatorContext) { + assert inputs.length == this.getNumInputs(); + assert outputs.length == this.getNumOutputs(); + + final Stream filtered = ((JavaChannelInstance) inputs[0]).provideStream().filter(review -> { + try { + return OllamaSemanticFilter.isPositiveSentiment(review); + } catch (IOException | InterruptedException e) { + throw new RuntimeException("Ollama call failed", e); + } + }); + ((StreamChannel.Instance) outputs[0]).accept(filtered); + + return ExecutionOperator.modelLazyExecution(inputs, outputs, operatorContext); + } + + @Override + public String getLoadProfileEstimatorConfigurationKey() { + return "wayang.semantic.ollama.model1.load"; + } + + @Override + protected OllamaFilterOperator newInstance(final DataSetType type, final String prompt) { + return new OllamaModel1FilterOperator<>(type, prompt); + } +} + +/** + * Physical operator for the "model2" Ollama implementation: a different fixed sentiment prompt, also + * ignores the logical operator's {@link #getPrompt()}. + */ +final class OllamaModel2FilterOperator extends OllamaFilterOperator { + + OllamaModel2FilterOperator(final DataSetType type, final String prompt) { + super(type, prompt); + } + + @Override + @SuppressWarnings("unchecked") + public Tuple, Collection> evaluate( + final ChannelInstance[] inputs, + final ChannelInstance[] outputs, + final OllamaExecutor ollamaExecutor, + final OptimizationContext.OperatorContext operatorContext) { + assert inputs.length == this.getNumInputs(); + assert outputs.length == this.getNumOutputs(); + + final Stream filtered = ((JavaChannelInstance) inputs[0]).provideStream().filter(review -> { + try { + return OllamaSemanticFilter.isPositiveSentiment2(review); + } catch (IOException | InterruptedException e) { + throw new RuntimeException("Ollama call failed", e); + } + }); + ((StreamChannel.Instance) outputs[0]).accept(filtered); + + return ExecutionOperator.modelLazyExecution(inputs, outputs, operatorContext); + } + + @Override + public String getLoadProfileEstimatorConfigurationKey() { + return "wayang.semantic.ollama.model2.load"; + } + + @Override + protected OllamaFilterOperator newInstance(final DataSetType type, final String prompt) { + return new OllamaModel2FilterOperator<>(type, prompt); + } +} + +/** + * Physical operator for the "model3" Ollama implementation: the only one that actually uses the + * logical operator's {@link #getPrompt()}. + */ +final class OllamaModel3FilterOperator extends OllamaFilterOperator { + + OllamaModel3FilterOperator(final DataSetType type, final String prompt) { + super(type, prompt); + } + + @Override + @SuppressWarnings("unchecked") + public Tuple, Collection> evaluate( + final ChannelInstance[] inputs, + final ChannelInstance[] outputs, + final OllamaExecutor ollamaExecutor, + final OptimizationContext.OperatorContext operatorContext) { + assert inputs.length == this.getNumInputs(); + assert outputs.length == this.getNumOutputs(); + + final Stream filtered = ((JavaChannelInstance) inputs[0]).provideStream().filter(review -> { + try { + return OllamaSemanticFilter.isPositiveSentiment3(review, this.getPrompt()); + } catch (IOException | InterruptedException e) { + throw new RuntimeException("Ollama call failed", e); + } + }); + ((StreamChannel.Instance) outputs[0]).accept(filtered); + + return ExecutionOperator.modelLazyExecution(inputs, outputs, operatorContext); + } + + @Override + public String getLoadProfileEstimatorConfigurationKey() { + return "wayang.semantic.ollama.model3.load"; + } + + @Override + protected OllamaFilterOperator newInstance(final DataSetType type, final String prompt) { + return new OllamaModel3FilterOperator<>(type, prompt); + } +} + +abstract class AbstractOllamaFilterMapping implements Mapping { + + private final Class targetOperatorClass; + + AbstractOllamaFilterMapping(final Class targetOperatorClass) { + this.targetOperatorClass = targetOperatorClass; + } + + @Override + public Collection getTransformations() { + return Collections.singleton(new PlanTransformation(this.createSubplanPattern(), + this.createReplacementSubplanFactory(), OllamaPlatform.getInstance())); + } + + private SubplanPattern createSubplanPattern() { + return SubplanPattern.createSingleton(new OperatorPattern>("semantic_filter", + new SemanticFilterOperator<>(DataSetType.NONE), false) + .withAdditionalTest(op -> op.getTargetModels() != null) + .withAdditionalTest(op -> op.getTargetModels().contains(this.targetOperatorClass))); + } + + protected abstract OllamaFilterOperator createOperator(DataSetType type, String prompt); + + private ReplacementSubplanFactory createReplacementSubplanFactory() { + return new ReplacementSubplanFactory.OfSingleOperators>((matchedOperator, epoch) -> + this.createOperator(matchedOperator.getInputType(), matchedOperator.getPrompt()).at(epoch)); + } +} + +final class OllamaModel1FilterMapping extends AbstractOllamaFilterMapping { + OllamaModel1FilterMapping() { + super(OllamaModel1FilterOperator.class); + } + + @Override + protected OllamaFilterOperator createOperator(final DataSetType type, final String prompt) { + return new OllamaModel1FilterOperator<>(type, prompt); + } +} + +final class OllamaModel2FilterMapping extends AbstractOllamaFilterMapping { + OllamaModel2FilterMapping() { + super(OllamaModel2FilterOperator.class); + } + + @Override + protected OllamaFilterOperator createOperator(final DataSetType type, final String prompt) { + return new OllamaModel2FilterOperator<>(type, prompt); + } +} + +final class OllamaModel3FilterMapping extends AbstractOllamaFilterMapping { + OllamaModel3FilterMapping() { + super(OllamaModel3FilterOperator.class); + } + + @Override + protected OllamaFilterOperator createOperator(final DataSetType type, final String prompt) { + return new OllamaModel3FilterOperator<>(type, prompt); + } +} + +final class OllamaSemanticFilter { + private static volatile String OLLAMA_API_URL = "http://ollama:11434/api/generate"; + private static final String MODEL_NAME = "tinyllama"; + private static final HttpClient httpClient = HttpClient.newHttpClient(); + + /** + * Points calls at a different Ollama-compatible endpoint, e.g. an in-process mock server for + * tests, so the real HTTP wiring gets exercised without depending on an external Ollama instance. + */ + static void setApiUrl(final String apiUrl) { + OLLAMA_API_URL = apiUrl; + } + + public static boolean isPositiveSentiment(final Review review) throws IOException, InterruptedException { + final String prompt = String.format("Analyze the sentiment of this movie review. " + + "Reply with only 'POSITIVE' or 'NEGATIVE'.\n\n" + "Review: %s", review.getReviewText()); + final String response = callOllama(prompt); + return response.contains("POSITIVE"); + } + + public static boolean isPositiveSentiment2(final Review review) throws IOException, InterruptedException { + final String prompt = String.format( + "Analyze the sentiment of this movie review, words like love, fantastic and great are positive modifiers. " + + "Reply with only 'POSITIVE' or 'NEGATIVE'.\n\n" + "Review: %s", + review.getReviewText()); + final String response = callOllama(prompt); + return response.contains("POSITIVE"); + } + + public static boolean isPositiveSentiment3(final Review review, final String prompt) + throws IOException, InterruptedException { + final String response = callOllama(prompt + " | " + review.getReviewText()); + return response.contains("POSITIVE"); + } + + private static String callOllama(final String prompt) throws IOException, InterruptedException { + final String requestBody = String.format("{\"model\": \"%s\", \"prompt\": \"%s\", \"stream\": false}", + MODEL_NAME, escapeJson(prompt)); + + final HttpRequest request = HttpRequest.newBuilder().uri(URI.create(OLLAMA_API_URL)) + .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString(requestBody)) + .timeout(Duration.ofMinutes(2)).build(); + + final HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + + if (response.statusCode() != 200) { + throw new IOException("Ollama API error: " + response.body()); + } + + return parseOllamaResponse(response.body()); + } + + private static String parseOllamaResponse(final String jsonResponse) { + int startIdx = jsonResponse.indexOf("\"response\":\""); + + if (startIdx == -1) + return ""; + + startIdx += "\"response\":\"".length(); + + final int endIdx = jsonResponse.indexOf("\"", startIdx); + + return jsonResponse.substring(startIdx, endIdx); + } + + private static String escapeJson(final String str) { + return str.replace("\"", "\\\"").replace("\n", "\\n").replace("\r", "\\r"); + } +}