Use AI VibeFlow to produce vector data

Console

1. Log on to the OSS console.

2. In the left-side navigation pane, click AI VibeFlow.

3. On the AI VibeFlow page, click Create DataPipeline.

4. In the Create DataPipeline panel, configure the following items as instructed, and then click OK.

   • Basic settings

     • Rule Name: up to 64 characters. Uppercase and lowercase letters, digits, hyphens (-), and underscores (_) are supported. The name must be unique within the same account and region.

     • Rule Description: optional, up to 200 characters.

   • Source data settings

     • Source Bucket: first select the region, and then select the target bucket.

     • Source File Scope: All Files processes all objects in the bucket; Specified Object Name Prefix processes only the objects that match the prefix.

     • Source File Types: you can select multiple types from Text, Image, and Video. The selection must match the capabilities of the vector model you use later.

     • Processing Mode: Process only existing initial data (files that already exist when the pipeline is created); Process only subsequent incremental data (newly uploaded files trigger processing automatically); Process both existing and subsequent incremental data.

   • Vectorization settings

     • Vector Model Source: currently only Alibaba Cloud Model Studio is supported.

     • API KEY: enter the Model Studio API key, which starts with sk- and is 35 to 128 characters in length.

     • Vector Model: text models such as text-embedding-v4 (default) and text-embedding-v3; multimodal models such as tongyi-embedding-vision-plus and tongyi-embedding-vision-flash. Make sure that the model output dimensions are the same as the vector index dimensions.

     • Video Frame Capture Frequency: can be configured only for multimodal models. The value range is (0, 1], and the default value is 1.0. A smaller value captures fewer frames.

     • Cascade Deletion Settings: Retain the corresponding vector data when the source object is deleted (default); Delete the corresponding vector data when the source object is deleted.

   • Destination data settings

     • Vector Bucket: must be located in the same region as the source bucket.

     • Vector Index: select the target index from the index list of the selected vector bucket. The index dimensions must be the same as the output dimensions of the vector model.

     • Vector Key Naming Rule: optional. If you enter a prefix, the format is {prefix}/{ObjectKey}. If you leave it empty, the vector key is the same as the ObjectKey.

     • Use ObjectTag as metadata: optional, up to 10 keys, used for subsequent scalar filtering during retrieval.

     • Use UserMeta as metadata: optional, up to 10 keys. The system automatically adds the X-OSS-META- prefix.

   • Exception handling and authorization

     • Exception Handling Method: Skip the exception and continue with subsequent processing tasks; Skip the exception and continue with subsequent processing tasks, and obtain the exception details (you must configure a bucket and prefix to receive error messages; the receiving bucket must be located in the same region as the source bucket and cannot be the source bucket itself); Terminate subsequent tasks immediately when an exception occurs.

     • Authorized Role: select Create Role to automatically create and bind oss-embedding-{uuid}.

   • Confirm the settings: check all settings and click OK after you confirm that they are correct.

ossutil

<?xml version="1.0" encoding="UTF-8" ?>
<DataPipelineConfiguration>
  <Sources>
      <InputBucket>my-bucket</InputBucket>
      <InputDataScope>All</InputDataScope>
  </Sources>
  <DataPipelineEmbeddingConfiguration>
      <EmbeddingProvider>bailian</EmbeddingProvider>
      <ApiKey>sk-xxxx</ApiKey>
      <Model>qwen2.5-vl-embedding</Model>
  </DataPipelineEmbeddingConfiguration>
  <Destination>
      <VectorBucketName>my-vector-bucket</VectorBucketName>
      <VectorIndexNames>my-index</VectorIndexNames>
  </Destination>
  <DataPipelineError>
      <ErrorMode>ignore</ErrorMode>
  </DataPipelineError>
</DataPipelineConfiguration>

ossutil api do-data-pipe-line-action --action putDataPipelineConfiguration \
  --parameters dataPipelineName=my-data-pipeline \
  --parameters role=acs:ram::<AccountId>:role/my-data-pipeline-role \
  --body file://config.xml \
  --region cn-hangzhou

SDK

import com.aliyun.sdk.service.oss2.credentials.CredentialsProvider;
import com.aliyun.sdk.service.oss2.credentials.EnvironmentVariableCredentialsProvider;
import com.aliyun.sdk.service.oss2.dataprocess.OSSDataProcessClient;
import com.aliyun.sdk.service.oss2.dataprocess.models.*;
import java.util.Arrays;
import java.util.Collections;

public class PutDataPipelineConfigurationSample {
    public static void main(String[] args) throws Exception {
        String role = "acs:ram::<AccountId>:role/my-data-pipeline-role";
        CredentialsProvider provider = new EnvironmentVariableCredentialsProvider();
        try (OSSDataProcessClient client = OSSDataProcessClient.newBuilder()
                .region("cn-hangzhou")
                .credentialsProvider(provider)
                .build()) {

            DataPipelineSourceFilterConfiguration filterConfig =
                    DataPipelineSourceFilterConfiguration.newBuilder()
                            .prefixSet(Arrays.asList("data/"))
                            .objectMediaTypes(Arrays.asList("image", "video"))
                            .build();

            DataPipelineSource source = DataPipelineSource.newBuilder()
                    .inputBucket("my-bucket")
                    .inputDataScope("All")
                    .ignoreDelete(true)
                    .filterConfiguration(filterConfig)
                    .build();

            DataPipelineEmbeddingConfiguration embeddingConfig =
                    DataPipelineEmbeddingConfiguration.newBuilder()
                            .embeddingProvider("bailian")
                            .apiKey("sk-xxxx")
                            .model("qwen2.5-vl-embedding")
                            .fps(1.0f)
                            .build();

            DataPipelineDestination destination = DataPipelineDestination.newBuilder()
                    .vectorBucketName("my-vector-bucket")
                    .vectorIndexNames(Collections.singletonList("my-index"))
                    .build();

            PutDataPipelineConfigurationConfiguration config =
                    PutDataPipelineConfigurationConfiguration.newBuilder()
                            .dataPipelineDescription("Data pipeline sample")
                            .sources(Collections.singletonList(source))
                            .dataPipelineEmbeddingConfiguration(embeddingConfig)
                            .destination(destination)
                            .build();

            PutDataPipelineConfigurationResult result = client.putDataPipelineConfiguration(
                    PutDataPipelineConfigurationRequest.newBuilder()
                            .dataPipelineName("my-data-pipeline")
                            .role(role)
                            .putDataPipelineConfigurationConfiguration(config)
                            .build());

            System.out.printf("Status code:%d, request id:%s%n",
                    result.statusCode(), result.requestId());

        } catch (Exception e) {
            System.out.printf("error:%n%s", e);
        }
    }
}

package main

import (
	"context"
	"log"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/dataprocess"
)

func main() {
	region := "cn-hangzhou"
	role := "acs:ram::<AccountId>:role/my-data-pipeline-role"
	apiKey := "sk-xxxx"
	dataPipelineName := "my-data-pipeline"

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := dataprocess.NewClient(cfg)

	result, err := client.PutDataPipelineConfiguration(context.TODO(), &dataprocess.PutDataPipelineConfigurationRequest{
		DataPipelineName: oss.Ptr(dataPipelineName),
		Role:             oss.Ptr(role),
		DataPipelineConfiguration: &dataprocess.DataPipelineConfiguration{
			DataPipelineDescription: oss.Ptr("Vectorize business data using the BERT multimodal model"),
			Sources: []dataprocess.DataPipelineSource{
				{
					InputBucket:    oss.Ptr("bucket"),
					InputDataScope: oss.Ptr("All"),
					FilterConfiguration: &dataprocess.DataPipelineSourceFilterConfiguration{
						PrefixSet:        []string{"prefix1"},
						ObjectMediaTypes: []string{"text"},
					},
				},
			},
			DataPipelineEmbeddingConfiguration: &dataprocess.DataPipelineEmbeddingConfiguration{
				ApiKey:            oss.Ptr(apiKey),
				EmbeddingProvider: oss.Ptr("bailian"),
				FPS:               oss.Ptr(float64(1)),
				Model:             oss.Ptr("qwen2.5-vl-embedding"),
			},
			Destination: &dataprocess.DataPipelineDestination{
				VectorBucketName:    oss.Ptr("my-vector-bucket"),
				VectorIndexNames:    []string{"index"},
				VectorKeyPrefix:     oss.Ptr("prefix"),
				ObjectTagToMetadata: []string{"key1"},
				UsermetaToMetadata:  []string{"x-oss-meta-key1"},
			},
			DataPipelineError: &dataprocess.DataPipelineError{
				ErrorMode:   oss.Ptr("ignoreAndRecord"),
				ErrorBucket: oss.Ptr("my-error-bucket"),
				ErrorPrefix: oss.Ptr("error-output/"),
			},
		},
	})

	if err != nil {
		log.Fatalf("failed to put pipeline configuration %v", err)
	}
	log.Printf("put pipeline configuration result:%#v\n", result)
}

API

Call the PutDataPipelineConfiguration operation to create a DataPipeline.

Console

1. On the AI VibeFlow page, click the Rule Name of the target DataPipeline or click Details in the Actions column.

2. On the details page, view the complete information, including the basic settings, source data settings, vectorization settings, destination settings, exception handling settings, and authorization settings.

ossutil

ossutil api do-data-pipe-line-action --action getDataPipelineConfiguration \
  --parameters dataPipelineName=my-data-pipeline \
  --region cn-hangzhou

SDK

import com.aliyun.sdk.service.oss2.credentials.CredentialsProvider;
import com.aliyun.sdk.service.oss2.credentials.EnvironmentVariableCredentialsProvider;
import com.aliyun.sdk.service.oss2.dataprocess.OSSDataProcessClient;
import com.aliyun.sdk.service.oss2.dataprocess.models.*;

public class GetDataPipelineConfigurationSample {
    public static void main(String[] args) throws Exception {
        CredentialsProvider provider = new EnvironmentVariableCredentialsProvider();
        try (OSSDataProcessClient client = OSSDataProcessClient.newBuilder()
                .region("cn-hangzhou")
                .credentialsProvider(provider)
                .build()) {

            GetDataPipelineConfigurationResult result = client.getDataPipelineConfiguration(
                    GetDataPipelineConfigurationRequest.newBuilder()
                            .dataPipelineName("my-data-pipeline")
                            .build());

            System.out.printf("Status code:%d, request id:%s%n",
                    result.statusCode(), result.requestId());

            DataPipelineConfiguration cfg = result.dataPipelineConfiguration();
            if (cfg != null) {
                System.out.printf("Pipeline name:%s, status:%s, role:%s, createTime:%s%n",
                        cfg.dataPipelineName(), cfg.status(),
                        cfg.dataPipelineRole(), cfg.createTime());
            }

        } catch (Exception e) {
            System.out.printf("error:%n%s", e);
        }
    }
}

package main

import (
	"context"
	"log"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/dataprocess"
)

func main() {
	region := "cn-hangzhou"
	dataPipelineName := "my-data-pipeline"

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := dataprocess.NewClient(cfg)

	result, err := client.GetDataPipelineConfiguration(context.TODO(), &dataprocess.GetDataPipelineConfigurationRequest{
		DataPipelineName: oss.Ptr(dataPipelineName),
	})

	if err != nil {
		log.Fatalf("failed to get pipeline configuration %v", err)
	}
	log.Printf("get pipeline configuration result:%#v\n", result)
}

API

Call the GetDataPipelineConfiguration operation to get the DataPipeline configuration.

Console

On the AI VibeFlow page, you can directly view the list of all DataPipelines of the current account in the selected region, including information such as the rule name, description, status, and creation time.

ossutil

ossutil api do-data-pipe-line-action --action listDataPipelineConfigurations \
  --parameters maxResults=100 \
  --region cn-hangzhou

SDK

import com.aliyun.sdk.service.oss2.credentials.CredentialsProvider;
import com.aliyun.sdk.service.oss2.credentials.EnvironmentVariableCredentialsProvider;
import com.aliyun.sdk.service.oss2.dataprocess.OSSDataProcessClient;
import com.aliyun.sdk.service.oss2.dataprocess.models.*;

public class ListDataPipelineConfigurationsSample {
    public static void main(String[] args) throws Exception {
        CredentialsProvider provider = new EnvironmentVariableCredentialsProvider();
        try (OSSDataProcessClient client = OSSDataProcessClient.newBuilder()
                .region("cn-hangzhou")
                .credentialsProvider(provider)
                .build()) {

            String nextToken = null;
            do {
                ListDataPipelineConfigurationsRequest.Builder reqBuilder =
                        ListDataPipelineConfigurationsRequest.newBuilder().maxResults(100);
                if (nextToken != null) {
                    reqBuilder.nextToken(nextToken);
                }

                ListDataPipelineConfigurationsResult result =
                        client.listDataPipelineConfigurations(reqBuilder.build());

                System.out.printf("Status code:%d, request id:%s%n",
                        result.statusCode(), result.requestId());

                if (result.dataPipelineConfigurations() != null) {
                    for (DataPipelineConfiguration cfg : result.dataPipelineConfigurations()) {
                        System.out.printf("Pipeline: name=%s, status=%s, createTime=%s%n",
                                cfg.dataPipelineName(), cfg.status(), cfg.createTime());
                    }
                }

                nextToken = result.nextToken();
            } while (nextToken != null && !nextToken.isEmpty());

        } catch (Exception e) {
            System.out.printf("error:%n%s", e);
        }
    }
}

package main

import (
	"context"
	"log"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/dataprocess"
)

func main() {
	region := "cn-hangzhou"

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := dataprocess.NewClient(cfg)

	p := client.NewListDataPipelineConfigurationsPaginator(&dataprocess.ListDataPipelineConfigurationsRequest{})

	var i int
	log.Println("Data Pipeline Configurations:")
	for p.HasNext() {
		i++

		page, err := p.NextPage(context.TODO())
		if err != nil {
			log.Fatalf("failed to get page %v, %v", i, err)
		}

		for _, data := range page.DataPipelineConfigurations {
			log.Printf("Data Pipeline Name:%v, Data Pipeline Description:%v, Data Pipeline Role:%v, Status:%v, Phase:%v\n", oss.ToString(data.DataPipelineName), oss.ToString(data.DataPipelineDescription), oss.ToString(data.DataPipelineRole), oss.ToString(data.Status), oss.ToString(data.Phase))
		}
	}
}

API

Call the ListDataPipelineConfigurations operation to list DataPipelines.

Console

On the AI VibeFlow page, find the target DataPipeline and click Pause in the Actions column.

ossutil

ossutil api do-data-pipe-line-action --action pauseDataPipeline \
  --parameters dataPipelineName=my-data-pipeline \
  --region cn-hangzhou

SDK

import com.aliyun.sdk.service.oss2.credentials.CredentialsProvider;
import com.aliyun.sdk.service.oss2.credentials.EnvironmentVariableCredentialsProvider;
import com.aliyun.sdk.service.oss2.dataprocess.OSSDataProcessClient;
import com.aliyun.sdk.service.oss2.dataprocess.models.*;

public class PauseDataPipelineSample {
    public static void main(String[] args) throws Exception {
        CredentialsProvider provider = new EnvironmentVariableCredentialsProvider();
        try (OSSDataProcessClient client = OSSDataProcessClient.newBuilder()
                .region("cn-hangzhou")
                .credentialsProvider(provider)
                .build()) {

            PauseDataPipelineResult result = client.pauseDataPipeline(
                    PauseDataPipelineRequest.newBuilder()
                            .dataPipelineName("my-data-pipeline")
                            .build());

            System.out.printf("Status code:%d, request id:%s%n",
                    result.statusCode(), result.requestId());

        } catch (Exception e) {
            System.out.printf("error:%n%s", e);
        }
    }
}

package main

import (
	"context"
	"log"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/dataprocess"
)

func main() {
	region := "cn-hangzhou"
	dataPipelineName := "my-data-pipeline"

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := dataprocess.NewClient(cfg)

	result, err := client.PauseDataPipeline(context.TODO(), &dataprocess.PauseDataPipelineRequest{
		DataPipelineName: oss.Ptr(dataPipelineName),
	})

	if err != nil {
		log.Fatalf("failed to pause pipeline %v", err)
	}
	log.Printf("pause pipeline result:%#v\n", result)
}

API

Call the PauseDataPipeline operation to pause a DataPipeline.

Console

On the AI VibeFlow page, find the target DataPipeline and click Restart in the Actions column.

ossutil

ossutil api do-data-pipe-line-action --action restartDataPipeline \
  --parameters dataPipelineName=my-data-pipeline \
  --region cn-hangzhou

SDK

import com.aliyun.sdk.service.oss2.credentials.CredentialsProvider;
import com.aliyun.sdk.service.oss2.credentials.EnvironmentVariableCredentialsProvider;
import com.aliyun.sdk.service.oss2.dataprocess.OSSDataProcessClient;
import com.aliyun.sdk.service.oss2.dataprocess.models.*;

public class RestartDataPipelineSample {
    public static void main(String[] args) throws Exception {
        CredentialsProvider provider = new EnvironmentVariableCredentialsProvider();
        try (OSSDataProcessClient client = OSSDataProcessClient.newBuilder()
                .region("cn-hangzhou")
                .credentialsProvider(provider)
                .build()) {

            RestartDataPipelineResult result = client.restartDataPipeline(
                    RestartDataPipelineRequest.newBuilder()
                            .dataPipelineName("my-data-pipeline")
                            .build());

            System.out.printf("Status code:%d, request id:%s%n",
                    result.statusCode(), result.requestId());

        } catch (Exception e) {
            System.out.printf("error:%n%s", e);
        }
    }
}

package main

import (
	"context"
	"log"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/dataprocess"
)

func main() {
	region := "cn-hangzhou"
	dataPipelineName := "my-data-pipeline"

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := dataprocess.NewClient(cfg)

	result, err := client.RestartDataPipeline(context.TODO(), &dataprocess.RestartDataPipelineRequest{
		DataPipelineName: oss.Ptr(dataPipelineName),
	})

	if err != nil {
		log.Fatalf("failed to restart pipeline %v", err)
	}
	log.Printf("restart pipeline result:%#v\n", result)
}

API

Call the RestartDataPipeline operation to restart a DataPipeline.

Console

On the AI VibeFlow page, find the target DataPipeline, click Delete in the Actions column, and confirm to complete the deletion.

ossutil

ossutil api do-data-pipe-line-action --action deleteDataPipelineConfiguration \
  --parameters dataPipelineName=my-data-pipeline \
  --region cn-hangzhou

SDK

import com.aliyun.sdk.service.oss2.credentials.CredentialsProvider;
import com.aliyun.sdk.service.oss2.credentials.EnvironmentVariableCredentialsProvider;
import com.aliyun.sdk.service.oss2.dataprocess.OSSDataProcessClient;
import com.aliyun.sdk.service.oss2.dataprocess.models.*;

public class DeleteDataPipelineConfigurationSample {
    public static void main(String[] args) throws Exception {
        CredentialsProvider provider = new EnvironmentVariableCredentialsProvider();
        try (OSSDataProcessClient client = OSSDataProcessClient.newBuilder()
                .region("cn-hangzhou")
                .credentialsProvider(provider)
                .build()) {

            DeleteDataPipelineConfigurationResult result = client.deleteDataPipelineConfiguration(
                    DeleteDataPipelineConfigurationRequest.newBuilder()
                            .dataPipelineName("my-data-pipeline")
                            .build());

            System.out.printf("Status code:%d, request id:%s%n",
                    result.statusCode(), result.requestId());

        } catch (Exception e) {
            System.out.printf("error:%n%s", e);
        }
    }
}

package main

import (
	"context"
	"log"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/dataprocess"
)

func main() {
	region := "cn-hangzhou"
	dataPipelineName := "my-data-pipeline"

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region)

	client := dataprocess.NewClient(cfg)

	result, err := client.DeleteDataPipelineConfiguration(context.TODO(), &dataprocess.DeleteDataPipelineConfigurationRequest{
		DataPipelineName: oss.Ptr(dataPipelineName),
	})

	if err != nil {
		log.Fatalf("failed to delete pipeline configuration %v", err)
	}
	log.Printf("delete pipeline configuration result:%#v\n", result)
}

API

Call the DeleteDataPipelineConfiguration operation to delete a DataPipeline.