当无头 BI 遇上调度：腾讯音乐 SuperSonic 集成 DolphinScheduler CLI 的联动实战

点亮⭐️

https://github.com/apache/

点击蓝字 关注我们

来源 | PiflowX公众号

项目简介

SuperSonic

SuperSonic 是腾讯音乐开源的下一代 AI+BI 平台，统一了 Chat BI（由 LLM 驱动）和 Headless BI（由语义层驱动）两种范式。

核心特性：

• Chat BI 界面：业务用户可通过自然语言查询数据，并以图表形式可视化结果

• Headless BI 界面：分析工程师可构建语义数据模型（指标/维度/标签的定义及其含义和关系）

• 可扩展架构：通过 Java SPI 机制支持自定义实现

• 语义增强：将数据语义（业务术语、列值等）注入 Prompt，减少 LLM 幻觉

• 查询简化：将复杂 SQL 语法（JOIN、公式等）从 LLM 卸载到语义层

架构组件：

GitHub: https://github.com/tencentmusic/supersonic (⭐ 4.9k)

Apache DolphinScheduler

Apache DolphinScheduler 是 Apache 基金会的现代数据编排平台，专注于处理数据管道中的复杂任务依赖。

核心特性：

• 易于部署：支持 Standalone、Cluster、Docker、Kubernetes 四种部署模式

• 易于使用：通过 Web UI、Python SDK 或 Open API 创建和管理工作流

• 高可靠：去中心化、多 Master 多 Worker 架构，原生支持水平扩展

• 高性能：性能是其他编排平台的数倍，每天可处理数千万任务

• 云原生：支持跨多云和数据中心编排工作流

• 版本控制：提供工作流和任务实例的版本控制

• 灵活状态控制：支持随时暂停、停止和恢复工作流

• 多租户支持

GitHub: https://github.com/apache/dolphinscheduler (⭐ 14.3k)

dsctl (DolphinScheduler CLI)

dsctl 是一个命令行工具，用于管理 DolphinScheduler 资源。

主要功能：

• 项目管理（project）
• 工作流管理（workflow）
• 任务管理（task）
• 数据源管理（datasource）
• 环境管理（environment）
• 用户管理（user）
• 监控和审计（monitor/audit）

示例命令：

dsctl project list                    # 列出所有项目
dsctl workflow run daily-etl          # 运行工作流
dsctl workflow-instance watch 123     # 监控工作流实例
dsctl task-instance log 456 --raw     # 获取任务日志

集成背景

为什么需要集成？

集成价值：

1. 提升效率：减少 UI 操作步骤，一句话完成复杂操作

2. 降低门槛：非技术人员可通过自然语言操作工作流

3. 统一入口：在 SuperSonic 对话界面中统一管理数据查询和工作流操作

架构设计

用户自然语言输入
    ↓
WorkflowParser（ChatQueryParser SPI）
    ├── accept(): 检查 Agent 是否配置了 DSCTL 工具
    └── parse(): 调用 LLM 将自然语言映射为 dsctl 子命令
                 结果写入 SemanticParseInfo.properties
    ↓
WorkflowExecutor（ChatQueryExecutor SPI）
    ├── accept(): 检查 queryMode == "WORKFLOW_CTL"
    └── execute(): 调用 ProcessBuilder 执行 dsctl 命令
                   注入 DS_API_URL / DS_TOKEN 环境变量
                   返回 QueryResult.textResult（Markdown 格式）
    ↓
前端 ChatItem（修改白名单，支持 WORKFLOW_CTL 模式渲染）

数据流：

1. 用户输入自然语言 → SuperSonic Chat API
2. WorkflowParser接收请求 → 调用 LLM 生成 dsctl 命令
3. WorkflowExecutor执行命令 → 调用 dsctl 进程
4. dsctl 调用 DolphinScheduler API → 返回结果
5. 结果格式化为 Markdown → 前端渲染

后端实现

文件 1：AgentToolType.java（修改）

路径：chat/server/src/main/java/com/tencent/supersonic/chat/server/agent/AgentToolType.java

package com.tencent.supersonic.chat.server.agent;

import java.util.HashMap;
import java.util.Map;

publicenumAgentToolType {
    DATASET("Text2SQL数据集"),
    PLUGIN("第三方插件"),
    WORK_FLOW_CTL("工作流命令行");

    privatefinal String title;

    AgentToolType(String title) {
        this.title = title;
    }

    publicstatic Map<AgentToolType, String> getToolTypes() {
        Map<AgentToolType, String> map = newHashMap<>();
        map.put(DATASET, DATASET.title);
        map.put(PLUGIN, PLUGIN.title);
        map.put(DSCTL, DSCTL.title);   // 新增
        return map;
    }
}

注册后，前端 getToolTypes() API 会自动返回新类型，Agent 管理页面的工具类型下拉框中会出现 "DolphinScheduler CLI"。

文件 2：WorkflowTool.java（新建）

路径：chat/server/src/main/java/com/tencent/supersonic/chat/server/agent/WorkflowTool.java

package com.tencent.supersonic.chat.server.agent;

import java.util.List;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
publicclassWorkflowToolextendsAgentTool {

/** dsctl 可执行文件路径，默认 "dsctl" */
privateStringdsctlPath="dsctl";

/** 默认使用的 DolphinScheduler 项目名 */
private String defaultProject;

private String dsApiUrl;

private String dsToken;

/** 示例问题，用于 embedding 召回 */
private List<String> exampleQuestions;
}

文件 3：WorkflowParser.java（新建）

路径：chat/server/src/main/java/com/tencent/supersonic/chat/server/parser/WorkflowParser.java

package com.tencent.supersonic.chat.server.parser;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.tencent.supersonic.chat.server.agent.AgentToolType;
import com.tencent.supersonic.chat.server.agent.WorkflowTool;
import com.tencent.supersonic.chat.server.pojo.ParseContext;
import com.tencent.supersonic.common.pojo.ChatApp;
import com.tencent.supersonic.common.pojo.enums.AppModule;
import com.tencent.supersonic.common.util.ChatAppManager;
import com.tencent.supersonic.headless.api.pojo.SemanticParseInfo;
import com.tencent.supersonic.headless.api.pojo.response.ParseResp;
import dev.langchain4j.model.chat.ChatModel;
import dev.langchain4j.model.input.Prompt;
import dev.langchain4j.model.input.PromptTemplate;
import dev.langchain4j.model.output.structured.Description;
import dev.langchain4j.provider.ModelProvider;
import dev.langchain4j.service.AiServices;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;

/**
 * Agent 管理页面 └── 添加工具: type=DSCTL, dsctlPath=/usr/bin/dsctl, defaultProject=etl-prod ↓ 存入
 * Agent.toolConfig (JSON) WorkflowParser.accept() └── agent.getTools(AgentToolType.DSCTL) 非空 → 激活
 * WorkflowParser.parse() └── 读取 DsctlTool 配置 → LLM 解析命令 → 写入 parseInfo.properties
 * WorkflowExecutor.execute() └── 读取 dsctl_path + dsctl_cmd → Runtime.exec() → 返回结果
 */
@Slf4j
publicclassWorkflowParserimplementsChatQueryParser {

publicstaticfinalStringQUERY_MODE="WORKFLOW_CTL";
publicstaticfinalStringAPP_KEY="WORKFLOW_PARSER";

/** 缓存 dsctl schema 结果，避免重复执行 */
privatestaticvolatile String cachedSchema;
privatestaticvolatilelong schemaCacheTime;
privatestaticfinallongSCHEMA_CACHE_TTL_MS=5 * 60 * 1000; // 5 minutes

privatestaticfinalStringINSTRUCTION=
          """
          # Role: You are a DolphinScheduler CLI expert using dsctl.
    
          # Task: Convert user's natural language into dsctl commands.
    
          # dsctl Usage:
          {{dsctl_help}}
    
          # Available Commands:
          {{dsctl_commands}}
    
          # Command Generation Rules:
          1. Match user intent to the appropriate command
          2. For resource selectors, use discovery commands listed below
          3. Include required arguments and options
          4. Output ONLY the dsctl command
    
          # Intent Patterns:
          - "list/view/show" + resource → dsctl <resource> list
          - "get/detail" + resource → dsctl <resource> get <selector>
          - "create/new" + resource → dsctl <resource> create --name <name>
          - "delete/remove" + resource → dsctl <resource> delete <selector> --force
          - "run/execute" + workflow → dsctl workflow run <workflow>
          - "stop" + instance → dsctl workflow-instance stop <id>
          - "watch/monitor" + instance → dsctl workflow-instance watch <id>
          - "log" + task → dsctl task-instance log <id> --raw
    
          # Resource Selectors:
          - project: name or code (discover: dsctl project list)
          - workflow: name or code (discover: dsctl workflow list)
          - task: name or code (discover: dsctl task list)
          - workflow-instance: numeric id (discover: dsctl workflow-instance list)
          - task-instance: numeric id (discover: dsctl task-instance list)
    
          # Output Rules:
          1. Output ONLY the dsctl command
          2. If intent is unclear, output: unknown
          3. Do NOT include explanations
    
          # Question: {{question}}
          # Command:
          """;

publicWorkflowParser() {
    ChatAppManager.register(
            APP_KEY,
            ChatApp.builder()
                    .prompt(INSTRUCTION)
                    .name("工作流命令行")
                    .appModule(AppModule.CHAT)
                    .description("将自然语言转换为工作流命令")
                    .enable(true)
                    .build());
  }

/** 结构化输出类，@Description 告诉 AiServices 每个字段的含义 */
@Data
staticclassDsctlCommand {
    @Description("the dsctl subcommand without 'dsctl' prefix, e.g. 'workflow run daily-etl'")
    private String command;

    @Description("brief explanation in Chinese of what this command does")
    private String thought;
  }

/** 提取接口，AiServices 自动生成实现 */
interfaceDsctlCommandExtractor {
    DsctlCommand extractCommand(String text);
  }

@Override
publicbooleanaccept(ParseContext parseContext) {
    // 只有 Agent 配置了 Workflow Cli 工具才激活
    List<String> tools = parseContext.getAgent().getTools(AgentToolType.WORK_FLOW_CTL);
    return !CollectionUtils.isEmpty(tools);
  }

@Override
publicvoidparse(ParseContext parseContext) {
    // 读取 Agent 上配置的 workflowTool（取第一个）
    List<String> toolJsonList = parseContext.getAgent().getTools(AgentToolType.WORK_FLOW_CTL);
    WorkflowToolworkflowTool= JSONObject.parseObject(toolJsonList.getFirst(), WorkflowTool.class);

    // 读取 ChatApp 配置（LLM 模型绑定 + prompt 模板）
    ChatAppchatApp= parseContext.getAgent().getChatAppConfig().get(APP_KEY);
    if (Objects.isNull(chatApp) || !chatApp.isEnable()) {
      log.warn("WorkflowParser ChatApp [{}] is not enabled, skip.", APP_KEY);
      return;
    }

    // 构建 prompt，替换占位符
    Map<String, Object> variables = newHashMap<>();

    // 1. 执行 dsctl --help 获取使用说明
    StringdsctlHelp= executeDsctlCommand(workflowTool.getDsctlPath(), workflowTool, "--help");

    // 2. 执行 dsctl schema 获取命令列表并格式化（带缓存）
    StringdsctlSchema= getCachedOrExecuteSchema(workflowTool);
    StringformattedCommands= formatCommandsFromSchema(dsctlSchema);

    variables.put("dsctl_help", dsctlHelp);
    variables.put("dsctl_commands", formattedCommands);
    variables.put("question", parseContext.getRequest().getQueryText());

    Promptprompt= PromptTemplate.from(chatApp.getPrompt()).apply(variables);

    // 调用 LLM，使用 AiServices 做结构化输出
    ChatModelmodel= ModelProvider.getChatModel(chatApp.getChatModelConfig());
    DsctlCommandExtractorextractor= AiServices.create(DsctlCommandExtractor.class, model);
    DsctlCommandcmd= extractor.extractCommand(prompt.toUserMessage().singleText());

    log.info(
            "WorkflowParser input=[{}] -> command=[{}] thought=[{}]",
            parseContext.getRequest().getQueryText(),
            cmd == null ? "null" : cmd.getCommand(),
            cmd == null ? "null" : cmd.getThought());

    if (cmd == null || "unknown".equalsIgnoreCase(cmd.getCommand())) {
      // 无法识别，不设置 parseInfo，交给后续 parser 处理
      return;
    }

    // 将解析结果写入 parseInfo，供 DsctlExecutor 使用
    SemanticParseInfoparseInfo=newSemanticParseInfo();
    parseInfo.setQueryMode(QUERY_MODE);
    parseInfo.setId(1);

    Map<String, Object> props = newHashMap<>();
    // e.g. "workflow run daily-etl"
    props.put("workflow_ctl_cmd", cmd.getCommand());
    // LLM 的解释
    props.put("workflow_ctl_thought", cmd.getThought());
    // 可执行文件路径
    props.put("workflow_ctl_path", workflowTool.getDsctlPath());
    if (workflowTool.getDefaultProject() != null) {
      props.put("default_project", workflowTool.getDefaultProject());
    }

    if (workflowTool.getDsApiUrl() != null) {
      props.put("ds_api_url", workflowTool.getDsApiUrl());
    }
    if (workflowTool.getDsToken() != null) {
      props.put("ds_token", workflowTool.getDsToken());
    }

    parseInfo.setProperties(props);

    parseContext.getResponse().getSelectedParses().add(parseInfo);
    parseContext.getResponse().setState(ParseResp.ParseState.COMPLETED);
  }

/**
   * 执行 dsctl 命令并返回输出
   *
   * @param dsctlPath dsctl 可执行文件路径
   * @param workflowTool 工作流工具配置（包含环境变量）
   * @param args 命令参数
   * @return 命令输出
   */
private String executeDsctlCommand(String dsctlPath, WorkflowTool workflowTool, String... args) {
    try {
      String[] cmdArray = newString[args.length + 1];
      cmdArray[0] = dsctlPath;
      System.arraycopy(args, 0, cmdArray, 1, args.length);

      ProcessBuilderpb=newProcessBuilder(cmdArray);
      pb.redirectErrorStream(true);

      // 添加环境变量
      Map<String, String> env = pb.environment();
      if (workflowTool.getDsApiUrl() != null) {
        env.put("DS_API_URL", workflowTool.getDsApiUrl());
      }
      if (workflowTool.getDsToken() != null) {
        env.put("DS_API_TOKEN", workflowTool.getDsToken());
      }

      Processprocess= pb.start();

      String output;
      try (BufferedReaderreader=
              newBufferedReader(newInputStreamReader(process.getInputStream()))) {
        output = reader.lines().collect(Collectors.joining("\n"));
      }

      booleanfinished= process.waitFor(30, TimeUnit.SECONDS);
      if (!finished) {
        process.destroyForcibly();
        log.warn("dsctl command timed out: {}", String.join(" ", cmdArray));
        return"";
      }

      if (process.exitValue() != 0) {
        log.warn("dsctl command failed with exit code {}: {}", process.exitValue(), output);
      }

      return output;
    } catch (Exception e) {
      log.error("Failed to execute dsctl command: {}", e.getMessage(), e);
      return"";
    }
  }

/**
   * 获取缓存的 schema 或执行新命令
   *
   * @param workflowTool 工作流工具配置
   * @return dsctl schema JSON
   */
private String getCachedOrExecuteSchema(WorkflowTool workflowTool) {
    longnow= System.currentTimeMillis();
    if (cachedSchema != null && (now - schemaCacheTime) < SCHEMA_CACHE_TTL_MS) {
      log.debug("Using cached dsctl schema");
      return cachedSchema;
    }

    synchronized (WorkflowParser.class) {
      // Double-check locking
      if (cachedSchema != null && (now - schemaCacheTime) < SCHEMA_CACHE_TTL_MS) {
        return cachedSchema;
      }

      Stringschema= executeDsctlCommand(workflowTool.getDsctlPath(), workflowTool, "schema");
      if (schema != null && !schema.isBlank()) {
        cachedSchema = schema;
        schemaCacheTime = now;
      }
      return schema;
    }
  }

/**
   * 从 dsctl schema JSON 中格式化命令列表
   *
   * @param schemaJson dsctl schema 的 JSON 输出
   * @return 格式化后的命令列表
   */
private String formatCommandsFromSchema(String schemaJson) {
    if (schemaJson == null || schemaJson.isBlank()) {
      return getDefaultCommands();
    }

    try {
      JSONObjectschema= JSON.parseObject(schemaJson);
      JSONObjectdata= schema.getJSONObject("data");
      if (data == null) {
        return getDefaultCommands();
      }

      JSONArraycommands= data.getJSONArray("commands");
      if (commands == null) {
        return getDefaultCommands();
      }

      StringBuildersb=newStringBuilder();

      for (inti=0; i < commands.size(); i++) {
        JSONObjectgroup= commands.getJSONObject(i);
        if (!"group".equals(group.getString("kind"))) {
          continue;
        }

        StringgroupName= group.getString("name");
        StringgroupSummary= group.getString("summary");

        sb.append("\n## ").append(groupName).append("\n");
        sb.append("# ").append(groupSummary).append("\n");

        JSONArraycmdList= group.getJSONArray("commands");
        if (cmdList != null) {
          for (intj=0; j < cmdList.size(); j++) {
            JSONObjectcmd= cmdList.getJSONObject(j);
            StringcmdName= cmd.getString("name");
            Stringsummary= cmd.getString("summary");

            // Build command syntax
            StringBuildersyntax=newStringBuilder();
            syntax.append("dsctl ").append(groupName).append(" ").append(cmdName);

            // Add required arguments
            JSONArrayargs= cmd.getJSONArray("arguments");
            if (args != null) {
              for (intk=0; k < args.size(); k++) {
                JSONObjectarg= args.getJSONObject(k);
                if (arg.getBooleanValue("required")) {
                  syntax.append(" <").append(arg.getString("name")).append(">");
                }
              }
            }

            sb.append("- ").append(syntax).append(": ").append(summary).append("\n");

            // Add discovery command if available
            if (args != null) {
              for (intk=0; k < args.size(); k++) {
                JSONObjectarg= args.getJSONObject(k);
                Stringdiscovery= arg.getString("discovery_command");
                if (discovery != null && !discovery.isEmpty()) {
                  sb.append("  Discovery: ").append(discovery).append("\n");
                  break;
                }
              }
            }
          }
        }
      }

      return sb.toString();
    } catch (Exception e) {
      log.error("Failed to parse dsctl schema: {}", e.getMessage(), e);
      return getDefaultCommands();
    }
  }

/**
   * 返回默认的命令列表（当 schema 解析失败时使用）
   *
   * @return 默认命令列表
   */
private String getDefaultCommands() {
    return"""
        - dsctl project list: List projects
        - dsctl project get <project>: Get one project
        - dsctl project create --name <name>: Create a project
        - dsctl project update <project>: Update a project
        - dsctl project delete <project>: Delete a project
        - dsctl workflow list: List workflows
        - dsctl workflow get <workflow>: Get one workflow
        - dsctl workflow run <workflow>: Run a workflow
        - dsctl workflow delete <workflow>: Delete a workflow
        - dsctl workflow-instance list: List instances
        - dsctl workflow-instance get <id>: Get one instance
        - dsctl workflow-instance watch <id>: Watch instance
        - dsctl workflow-instance stop <id>: Stop instance
        - dsctl workflow-instance digest <id>: Get instance summary
        - dsctl task list --workflow <workflow>: List tasks
        - dsctl task get <task>: Get one task
        - dsctl task-instance list --workflow-instance <id>: List task instances
        - dsctl task-instance log <id>: Get task log
        - dsctl cluster list: List clusters
        - dsctl datasource list: List datasources
        - dsctl environment list: List environments
        - dsctl user list: List users
        - dsctl tenant list: List tenants
        - dsctl schedule list: List schedules
        - dsctl doctor: Run diagnostics
        """;
  }
}

关键设计说明：

• ChatAppManager.register()

  在构造函数中注册 prompt 模板，使其在 Agent 管理页面中可配置、可绑定不同 LLM

• AiServices.create()

  自动在 prompt 中追加 JSON schema 约束，将 LLM 输出反序列化为 DsctlCommand 对象

• accept()

  通过检查 Agent 是否配置了 DSCTL 类型工具来决定是否激活，实现按 Agent 粒度控制

文件 4：WorkflowExecutor.java（新建）

路径：chat/server/src/main/java/com/tencent/supersonic/chat/server/executor/WorkflowExecutor.java

package com.tencent.supersonic.chat.server.executor;

import com.tencent.supersonic.chat.api.pojo.response.QueryResult;
import com.tencent.supersonic.chat.server.parser.WorkflowParser;
import com.tencent.supersonic.chat.server.pojo.ExecuteContext;
import com.tencent.supersonic.headless.api.pojo.response.QueryState;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;

@Slf4j
publicclassWorkflowExecutorimplementsChatQueryExecutor {

@Override
publicbooleanaccept(ExecuteContext executeContext) {
    return WorkflowParser.QUERY_MODE.equals(executeContext.getParseInfo().getQueryMode());
  }

@Override
public QueryResult execute(ExecuteContext executeContext) {
    Map<String, Object> props = executeContext.getParseInfo().getProperties();
    StringdsctlPath= (String) props.getOrDefault("workflow_ctl_path", "dsctl");
    StringsubCmd= (String) props.get("workflow_ctl_cmd");
    Stringthought= (String) props.getOrDefault("workflow_ctl_thought", "");

    QueryResultresult=newQueryResult();
    result.setQueryId(executeContext.getRequest().getQueryId());
    result.setQueryMode(WorkflowParser.QUERY_MODE);
    result.setQueryMode("PLAIN_TEXT");

    if (subCmd == null || subCmd.isBlank()) {
      result.setQueryState(QueryState.INVALID);
      result.setErrorMsg("无法识别 dsctl 命令");
      return result;
    }

    try {
      Stringoutput= runDsctl(dsctlPath, subCmd, props);
      result.setQueryState(QueryState.SUCCESS);
      // textResult 会显示在对话界面中
      result.setTextResult(buildTextResult(thought, subCmd, output));
    } catch (Exception e) {
      log.error("WorkflowExecutor failed, cmd=[dsctl {}]", subCmd, e);
      result.setQueryState(QueryState.SEARCH_EXCEPTION);
      result.setErrorMsg("dsctl 执行失败: " + e.getMessage());
    }

    return result;
  }

/** 执行 dsctl 子命令，返回标准输出内容。 超时时间 60 秒（watch 类命令可能需要更长，可按需调整）。 */
private String runDsctl(String dsctlPath, String subCmd, Map<String, Object> props)
      throws Exception {
    StringfullCmd= dsctlPath + " " + subCmd;
    log.info("DsctlExecutor executing: {}", fullCmd);

    ProcessBuilderpb=newProcessBuilder(fullCmd.split("\\s+"));
    pb.redirectErrorStream(true);

    // 显式注入环境变量到子进程
    Map<String, String> env = pb.environment();
    StringdsApiUrl= (String) props.get("ds_api_url");
    StringdsToken= (String) props.get("ds_token");
    if (dsApiUrl != null && !dsApiUrl.isBlank()) {
      env.put("DS_API_URL", dsApiUrl);
    }
    if (dsToken != null && !dsToken.isBlank()) {
      env.put("DS_API_TOKEN", dsToken);
    }

    Processprocess= pb.start();
    booleanfinished= process.waitFor(60, TimeUnit.SECONDS);
    if (!finished) {
      process.destroyForcibly();
      thrownewRuntimeException("dsctl 命令超时（60s）: " + fullCmd);
    }

    try (BufferedReaderreader=
        newBufferedReader(
            newInputStreamReader(process.getInputStream(), StandardCharsets.UTF_8))) {
      StringBuildersb=newStringBuilder();
      String line;
      while ((line = reader.readLine()) != null) {
        sb.append(line).append("\n");
      }
      return sb.toString().trim();
    }
  }

private String buildTextResult(String thought, String subCmd, String output) {
    StringBuildersb=newStringBuilder();
    if (thought != null && !thought.isBlank()) {
      sb.append("**").append(thought).append("**\n\n");
    }
    sb.append("```\n$ dsctl ").append(subCmd).append("\n");
    sb.append(output).append("\n```");
    return sb.toString();
  }
}

关键设计说明：

• ProcessBuilder.environment()

  显式注入 DS_API_URL / DS_API_TOKEN，解决 JVM 子进程不继承 shell 环境变量的问题

• redirectErrorStream(true)

  将 stderr 合并到 stdout，方便统一处理错误输出

• buildTextResult()

  将输出格式化为 Markdown，供前端渲染

文件 5：spring.factories（修改）

路径：launchers/standalone/src/main/resources/META-INF/spring.factories

在现有内容基础上追加（DsctlParser 放在 PlainTextParser 前，DsctlExecutor 放在 SqlExecutor 前）：

com.tencent.supersonic.chat.server.parser.ChatQueryParser=\
    com.tencent.supersonic.chat.server.parser.NL2PluginParser, \
    com.tencent.supersonic.chat.server.parser.NL2SQLParser,\
    com.tencent.supersonic.chat.server.parser.WorkflowParser,\
    com.tencent.supersonic.chat.server.parser.PlainTextParser

com.tencent.supersonic.chat.server.executor.ChatQueryExecutor=\
    com.tencent.supersonic.chat.server.executor.PluginExecutor, \
    com.tencent.supersonic.chat.server.executor.WorkflowExecutor,\
    com.tencent.supersonic.chat.server.executor.SqlExecutor,\
    com.tencent.supersonic.chat.server.executor.PlainTextExecutor

前端实现

文件 6：type.ts（修改）

路径：webapp/packages/supersonic-fe/src/pages/Agent/type.ts

export enumAgentToolTypeEnum {
NL2SQL_RULE = 'NL2SQL_RULE',
NL2SQL_LLM = 'NL2SQL_LLM',
PLUGIN = 'PLUGIN',
DATASET = 'DATASET',
WORK_FLOW_CTL= 'WORK_FLOW_CTL',   // 新增
}

exporttypeAgentToolType = {
  id?: string;
type: AgentToolTypeEnum;
name: string;
  queryModes?: QueryModeEnum[];
  plugins?: number[];
  metricOptions?: MetricOptionType[];
  exampleQuestions?: string[];
  modelIds?: number[];
// 新增 WORK_FLOW_CTL相关字段
  dsctlPath?: string;
  dsApiUrl?: string;
  dsToken?: string;
  defaultProject?: string;
};

文件 7：ToolModal.tsx（修改）

路径：webapp/packages/supersonic-fe/src/pages/Agent/ToolModal.tsx

在现有 PLUGIN 的 FormItem 之后追加 WORK_FLOW_CTL 的表单字段：

{toolType === AgentToolTypeEnum.WORK_FLOW_CTL&& (
<>
    <FormItem
      name="dsApiUrl"
      label="DS API 地址"
      rules={[{required:true, message: '请输入DolphinSchedulerAPI地址' }]}
    >
      <Inputplaceholder="例如：http://ds-host:12345"allowClear />
    </FormItem>
    <FormItemname="dsToken"label="DS Token">
      <Input.Passwordplaceholder="DolphinScheduler API Token"allowClear />
    </FormItem>
    <FormItemname="dsctlPath"label="dsctl 路径">
      <Inputplaceholder="dsctl 可执行文件路径，默认 dsctl"allowClear />
    </FormItem>
    <FormItemname="defaultProject"label="默认项目">
      <Inputplaceholder="默认使用的 DolphinScheduler 项目名"allowClear />
    </FormItem>
</>
)}

文件 8：index.tsx（修改）

路径：webapp/packages/chat-sdk/src/components/ChatItem/index.tsx

在 updateData 函数的白名单中添加 WORKFLOW_CTL，并兼容 code: 1 的响应格式：

const updateData = (res: Result<MsgDataType>) => {
lettip: string = '';
letdata: MsgDataType | undefined = undefined;
const { queryColumns, queryResults, queryState, queryMode, response, chatContext, errorMsg } =
    res.data || {};
setExecuteErrorMsg(errorMsg);
if (res.code === 400 || res.code === 401 || res.code === 412) {
    tip = res.msg;
  } elseif (res.code !== 200 && res.code !== 1) {   // 兼容 code:1
    tip = SEARCH_EXCEPTION_TIP;
  } elseif (queryState !== 'SUCCESS') {
    tip = response && typeof response === 'string' ? response : SEARCH_EXCEPTION_TIP;
  } elseif (
    (queryColumns && queryColumns.length > 0 && queryResults) ||
    queryMode === 'WEB_PAGE' ||
    queryMode === 'WEB_SERVICE' ||
    queryMode === 'PLAIN_TEXT' ||
    queryMode === 'WORKFLOW_CTL'   // 新增
  ) {
    data = res.data;
    tip = '';
  }
// ... 其余逻辑不变

文件 9：ExecuteItem.tsx（修改）

路径：webapp/packages/chat-sdk/src/components/ChatItem/ExecuteItem.tsx

修改 1：标题前缀（第 50 行附近）

const titlePrefix =
  queryMode === 'PLAIN_TEXT' || queryMode === 'WEB_SERVICE' || queryMode === 'WORKFLOW_CTL'
    ? '问答'
    : '数据';

修改 2：渲染逻辑（第 163 行附近），添加 WORKFLOW_CTL 分支，使用 ReactMarkdown 渲染 Markdown 格式的命令输出：

{renderCustomExecuteNode && executeItemNode ? (
  executeItemNode
) : data?.queryMode === 'PLAIN_TEXT' || data?.queryMode === 'WEB_SERVICE' ? (
  data?.textResult
) : data?.queryMode === 'WORKFLOW_CTL' ? (
<ReactMarkdown>{data?.textResult ?? ''}</ReactMarkdown>
) : data?.queryMode === 'WEB_PAGE' ? (
<WebPageid={queryId!}data={data} />
) : (
<ChatMsg
    isSimpleMode={isSimpleMode}
    forceShowTable={showMsgContentTable}
    queryId={queryId}
    question={question}
    data={data}
    chartIndex={chartIndex}
    triggerResize={triggerResize}
    onMsgContentTypeChange={setMsgContentType}
  />
)}

使用配置

前置条件

1. 安装 dsctl：

pip install -e .

1. 配置环境变量：

export DS_API_URL=http://your-dolphinscheduler-host:12345/dolphinscheduler
export DS_API_TOKEN=your-api-token

1. 验证安装：

dsctl doctor

Agent 配置步骤

1. 进入 Agent 管理页面
   
   
   
2. 新建或编辑 Agent，在"工具"Tab 中点击"新增工具"
   
   
   
3. 工具类型选择工作流命令行
   
   
   
4. 填写配置项：
   
   
   

1. 在对话聊天中，进行自然语言对话
   

支持的自然语言指令

扩展要点总结

整个集成无需修改 SuperSonic 框架核心代码，只需实现接口并在 spring.factories 中注册，体现了 SPI 机制的可插拔设计。

参考资料

• SuperSonic GitHub
• Apache DolphinScheduler
• dsctl GitHub
• SuperSonic 文档
• DolphinScheduler 文档