[KYUUBI #7379][PR 1/4] Data Agent Engine: module skeleton, configuration, and engine core

Author: wangzhigang1999

build/dist

@@ -288,6 +288,7 @@ mkdir -p "$DISTDIR/externals/engines/trino"
 mkdir -p "$DISTDIR/externals/engines/hive"
 mkdir -p "$DISTDIR/externals/engines/jdbc"
 mkdir -p "$DISTDIR/externals/engines/chat"
+mkdir -p "$DISTDIR/externals/engines/data-agent"
 echo "Kyuubi $VERSION $GITREVSTRING built for" > "$DISTDIR/RELEASE"
 echo "Java $JAVA_VERSION" >> "$DISTDIR/RELEASE"
 echo "Scala $SCALA_VERSION" >> "$DISTDIR/RELEASE"
@@ -363,6 +364,18 @@ for jar in $(ls "$DISTDIR/jars/"); do
   fi
 done
 
+# Copy data-agent engines
+cp "$KYUUBI_HOME/externals/kyuubi-data-agent-engine/target/kyuubi-data-agent-engine_${SCALA_VERSION}-${VERSION}.jar" "$DISTDIR/externals/engines/data-agent/"
+cp -r "$KYUUBI_HOME"/externals/kyuubi-data-agent-engine/target/scala-$SCALA_VERSION/jars/*.jar "$DISTDIR/externals/engines/data-agent/"
+
+# Share the jars w/ server to reduce binary size
+# shellcheck disable=SC2045
+for jar in $(ls "$DISTDIR/jars/"); do
+  if [[ -f "$DISTDIR/externals/engines/data-agent/$jar" ]]; then
+    (cd $DISTDIR/externals/engines/data-agent; ln -snf "../../../jars/$jar" "$DISTDIR/externals/engines/data-agent/$jar")
+  fi
+done
+
 # Copy Kyuubi Spark extension
 # shellcheck disable=SC2068
 for SPARK_EXTENSION_VERSION in ${SPARK_EXTENSION_VERSIONS[@]}; do

docs/configuration/settings.md

@@ -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.kyuubi.engine.dataagent.provider;
+
+import java.util.function.Consumer;
+import org.apache.kyuubi.config.KyuubiConf;
+import org.apache.kyuubi.engine.dataagent.runtime.event.AgentEvent;
+import org.apache.kyuubi.util.reflect.DynConstructors;
+
+/**
+ * A pluggable provider interface for the Data Agent engine. Implementations wire up the ReactAgent
+ * with an LLM model, tools, and middlewares.
+ */
+public interface DataAgentProvider {
+
+  /** Initialize a session for the given user. */
+  void open(String sessionId, String user);
+
+  /**
+   * Run the agent for the given request, emitting events via the consumer. Events include
+   * token-level ContentDelta for streaming, ToolCall/ToolResult for tool invocations, and
+   * AgentFinish when complete.
+   *
+   * @param sessionId the session identifier (maps to conversation memory)
+   * @param request user-facing parameters (question, model override, etc.)
+   * @param onEvent event consumer for streaming events
+   */
+  void run(String sessionId, ProviderRunRequest request, Consumer<AgentEvent> onEvent);
+
+  /**
+   * Close and clean up a single session, releasing session-scoped resources such as conversation
+   * history and session state. Called when one user session ends.
+   */
+  void close(String sessionId);
+
+  /**
+   * Resolve a pending tool approval request. Called when the client sends an approval or denial
+   * response for a tool call that requires human-in-the-loop confirmation.
+   *
+   * @param requestId the request ID from the ApprovalRequest event
+   * @param approved true to approve, false to deny
+   * @return true if the request was found and resolved, false if not found (timed out or invalid)
+   */
+  default boolean resolveApproval(String requestId, boolean approved) {
+    return false;
+  }
+
+  /**
+   * Stop the provider itself, releasing engine-level resources shared across all sessions (e.g.
+   * HTTP connection pools, thread pools). Called once when the entire engine shuts down.
+   */
+  default void stop() {}
+
+  static DataAgentProvider load(KyuubiConf conf) {
+    String providerClass = conf.get(KyuubiConf.ENGINE_DATA_AGENT_PROVIDER());
+    try {
+      return (DataAgentProvider)
+          DynConstructors.builder(DataAgentProvider.class)
+              .impl(providerClass, KyuubiConf.class)
+              .impl(providerClass)
+              .buildChecked()
+              .newInstanceChecked(conf);
+    } catch (ClassCastException e) {
+      throw new IllegalArgumentException(
+          "Class "
+              + providerClass
+              + " is not a child of '"
+              + DataAgentProvider.class.getName()
+              + "'.",
+          e);
+    } catch (Exception e) {
+      throw new IllegalArgumentException("Error while instantiating '" + providerClass + "': ", e);
+    }
+  }
+}

externals/kyuubi-data-agent-engine/pom.xml

@@ -0,0 +1,56 @@
+/*
+ * 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.kyuubi.engine.dataagent.provider;
+
+/**
+ * User-facing request parameters for a provider-level agent invocation. Only contains fields from
+ * the caller (question, model override, etc.). Adding new per-request options does not require
+ * changing the {@link DataAgentProvider} interface.
+ */
+public class ProviderRunRequest {
+
+  private final String question;
+  private String modelName;
+  private String approvalMode;
+
+  public ProviderRunRequest(String question) {
+    this.question = question;
+  }
+
+  public String getQuestion() {
+    return question;
+  }
+
+  public String getModelName() {
+    return modelName;
+  }
+
+  public ProviderRunRequest modelName(String modelName) {
+    this.modelName = modelName;
+    return this;
+  }
+
+  public String getApprovalMode() {
+    return approvalMode;
+  }
+
+  public ProviderRunRequest approvalMode(String approvalMode) {
+    this.approvalMode = approvalMode;
+    return this;
+  }
+}

externals/kyuubi-data-agent-engine/src/main/java/org/apache/kyuubi/engine/dataagent/provider/DataAgentProvider.java

@@ -0,0 +1,62 @@
+/*
+ * 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.kyuubi.engine.dataagent.provider.echo;
+
+import java.util.function.Consumer;
+import org.apache.kyuubi.engine.dataagent.provider.DataAgentProvider;
+import org.apache.kyuubi.engine.dataagent.provider.ProviderRunRequest;
+import org.apache.kyuubi.engine.dataagent.runtime.event.AgentEvent;
+import org.apache.kyuubi.engine.dataagent.runtime.event.AgentFinish;
+import org.apache.kyuubi.engine.dataagent.runtime.event.AgentStart;
+import org.apache.kyuubi.engine.dataagent.runtime.event.ContentComplete;
+import org.apache.kyuubi.engine.dataagent.runtime.event.ContentDelta;
+import org.apache.kyuubi.engine.dataagent.runtime.event.StepEnd;
+import org.apache.kyuubi.engine.dataagent.runtime.event.StepStart;
+
+/** A simple echo provider for testing purposes. Simulates the agent event stream. */
+public class EchoProvider implements DataAgentProvider {
+
+  @Override
+  public void open(String sessionId, String user) {}
+
+  @Override
+  public void run(String sessionId, ProviderRunRequest request, Consumer<AgentEvent> onEvent) {
+    String question = request.getQuestion();
+
+    onEvent.accept(new AgentStart());
+    onEvent.accept(new StepStart(1));
+
+    // Simulate token-level streaming
+    String reply =
+        "[DataAgent Echo] I received your question: "
+            + question
+            + "\n"
+            + "This is the Data Agent engine in echo mode. "
+            + "Please configure an LLM provider (e.g., OPENAI_COMPATIBLE) for actual data analysis.";
+    for (String token : reply.split("(?<=\\s)")) {
+      onEvent.accept(new ContentDelta(token));
+    }
+
+    onEvent.accept(new ContentComplete(reply));
+    onEvent.accept(new StepEnd(1));
+    onEvent.accept(new AgentFinish(1, 0, 0, 0));
+  }
+
+  @Override
+  public void close(String sessionId) {}
+}

externals/kyuubi-data-agent-engine/src/main/java/org/apache/kyuubi/engine/dataagent/provider/ProviderRunRequest.java

@@ -0,0 +1,37 @@
+/*
+ * 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.kyuubi.engine.dataagent.runtime.event;
+
+/** An error occurred during agent execution. */
+public final class AgentError extends AgentEvent {
+  private final String message;
+
+  public AgentError(String message) {
+    super(EventType.ERROR);
+    this.message = message;
+  }
+
+  public String message() {
+    return message;
+  }
+
+  @Override
+  public String toString() {
+    return "AgentError{message='" + message + "'}";
+  }
+}

externals/kyuubi-data-agent-engine/src/main/java/org/apache/kyuubi/engine/dataagent/provider/echo/EchoProvider.java

@@ -0,0 +1,42 @@
+/*
+ * 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.kyuubi.engine.dataagent.runtime.event;
+
+/**
+ * Base class for events emitted by the ReAct agent loop. Each event represents a discrete step in
+ * the agent's reasoning and execution process, enabling real-time token-level streaming to clients.
+ *
+ * <p>Every subclass must declare its {@link EventType}, which also determines the SSE event name
+ * used on the wire. This allows consumers to {@code switch} on the type rather than using {@code
+ * instanceof} chains.
+ *
+ * <p>Package-private constructor restricts subclassing to this package.
+ */
+public abstract class AgentEvent {
+
+  private final EventType eventType;
+
+  AgentEvent(EventType eventType) {
+    this.eventType = eventType;
+  }
+
+  /** Returns the type of this event. */
+  public EventType eventType() {
+    return eventType;
+  }
+}

externals/kyuubi-data-agent-engine/src/main/java/org/apache/kyuubi/engine/dataagent/runtime/event/AgentError.java

@@ -0,0 +1,63 @@
+/*
+ * 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.kyuubi.engine.dataagent.runtime.event;
+
+/** The agent has finished its analysis. */
+public final class AgentFinish extends AgentEvent {
+  private final int totalSteps;
+  private final long promptTokens;
+  private final long completionTokens;
+  private final long totalTokens;
+
+  public AgentFinish(int totalSteps, long promptTokens, long completionTokens, long totalTokens) {
+    super(EventType.AGENT_FINISH);
+    this.totalSteps = totalSteps;
+    this.promptTokens = promptTokens;
+    this.completionTokens = completionTokens;
+    this.totalTokens = totalTokens;
+  }
+
+  public int totalSteps() {
+    return totalSteps;
+  }
+
+  public long promptTokens() {
+    return promptTokens;
+  }
+
+  public long completionTokens() {
+    return completionTokens;
+  }
+
+  public long totalTokens() {
+    return totalTokens;
+  }
+
+  @Override
+  public String toString() {
+    return "AgentFinish{totalSteps="
+        + totalSteps
+        + ", promptTokens="
+        + promptTokens
+        + ", completionTokens="
+        + completionTokens
+        + ", totalTokens="
+        + totalTokens
+        + "}";
+  }
+}