数据管道（DataPipeline）是 AI VibeFlow 中的核心管理对象。可以通过控制台、ossutil、SDK 或 API 创建数据管道，将 OSS 通用 Bucket 中的文本、图片、视频等非结构化数据自动向量化并写入 OSS 向量 Bucket。
说明
当前功能白名单开放，需联系技术支持申请使用。
前提条件
创建数据管道
单个主账号在单个地域最多创建 1000 条数据管道。数据管道创建后不可修改，如需调整请删除后重建。
控制台
登录 OSS 管理控制台。
在左侧导航栏，单击 AI VibeFlow。
在 AI VibeFlow 页面，单击创建数据管道。
在创建数据管道面板中，按照以下说明填写各项配置，然后单击确定。
基础配置
规则名称：最多 64 个字符，支持大小写英文字母、数字、连字符（-）和下划线（_）；同一账号同地域不可重复。
规则描述：可选，最多 200 个字符。
源数据配置
源端 Bucket：先选择地域，再选择目标 Bucket。
源端文件范围：全部文件处理 Bucket 内全部对象；指定文件名前缀仅处理匹配前缀的对象。
源端文件类型：可多选文本、图片、视频，需与后续向量模型能力匹配。
处理模式：仅处理存量起始数据（创建时已存在的文件）；仅处理后续增量数据（新上传文件自动触发）；处理存量和后续增量数据。
向量化配置
向量模型来源：当前仅支持阿里云百炼。
API KEY：输入百炼 API Key，以
sk-开头，长度 35～128 个字符。向量模型：文本模型如
text-embedding-v4（默认）、text-embedding-v3；多模态模型如tongyi-embedding-vision-plus、tongyi-embedding-vision-flash。确保模型输出维度与向量索引维度一致。视频截帧频率：仅多模态模型可配置，取值 (0, 1]，默认 1.0。数值越小抽帧越少。
级联删除设置：原始 Object 删除时，保留对应向量数据（默认）；原始 Object 删除时，删除对应向量数据。
目的端数据配置
向量 Bucket：必须与源端 Bucket 位于同一地域。
向量 Index：从所选向量 Bucket 的索引列表中选择目标索引表，索引维度须与向量模型输出维度相同。
向量 Key 命名方式：可选。填写前缀后格式为
{前缀}/{ObjectKey}；不填时与 ObjectKey 保持一致。将 ObjectTag 作为元数据：可选，最多 10 个键，用于后续标量过滤检索。
将 UserMeta 作为元数据：可选，最多 10 个键，系统自动补齐
X-OSS-META-前缀。
异常处理与授权
异常情况处理方式：跳过异常并继续执行后续处理任务；跳过异常并继续执行后续处理任务，同时获取异常信息详情（需配置错误信息接收 Bucket 与前缀，接收 Bucket 需与源端 Bucket 位于同一地域且不能为源端 Bucket 本身）；一旦异常立即终止后续任务。
授权角色：选择新建角色自动创建
oss-embedding-{uuid}并绑定。
确认配置：检查所有配置项，确认无误后单击确定。
ossutil
使用示例
先将数据管道配置写入 config.xml：
<?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-hangzhouSDK
Java
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);
}
}
}Go
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
调用PutDataPipelineConfiguration接口创建数据管道。
获取数据管道配置
控制台
在 AI VibeFlow 页面，单击目标数据管道的规则名称或操作列的详情。
进入详情页面，查看基础配置、源数据配置、向量化配置、目的端配置、异常处理配置、授权配置等完整信息。
ossutil
使用示例
ossutil api do-data-pipe-line-action --action getDataPipelineConfiguration \
--parameters dataPipelineName=my-data-pipeline \
--region cn-hangzhouSDK
Java
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);
}
}
}Go
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
调用GetDataPipelineConfiguration接口获取数据管道配置。
列举数据管道
控制台
在 AI VibeFlow 页面可直接查看当前账号在所选地域下所有数据管道的列表，包含规则名称、描述、状态、创建时间等信息。
ossutil
使用示例
ossutil api do-data-pipe-line-action --action listDataPipelineConfigurations \
--parameters maxResults=100 \
--region cn-hangzhouSDK
Java
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);
}
}
}Go
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
调用ListDataPipelineConfigurations接口列举数据管道。
暂停数据管道
处于运行中状态的管道支持暂停。暂停后未处理文件保持等待，已写入向量数据不受影响。
控制台
在 AI VibeFlow 页面找到目标数据管道，在操作列单击暂停。
ossutil
使用示例
ossutil api do-data-pipe-line-action --action pauseDataPipeline \
--parameters dataPipelineName=my-data-pipeline \
--region cn-hangzhouSDK
Java
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);
}
}
}Go
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
调用PauseDataPipeline接口暂停数据管道。
重启数据管道
已暂停的管道可重启继续处理。若因索引写满自动暂停，请先创建新索引再重启。
控制台
在 AI VibeFlow 页面找到目标数据管道，在操作列单击重启。
ossutil
使用示例
ossutil api do-data-pipe-line-action --action restartDataPipeline \
--parameters dataPipelineName=my-data-pipeline \
--region cn-hangzhouSDK
Java
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);
}
}
}Go
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
调用RestartDataPipeline接口重启数据管道。
删除数据管道
删除后 OSS 停止向量化处理，已写入向量 Bucket 的向量数据不会被删除。运行中管道请先暂停再删除。删除操作不可撤销。
控制台
在 AI VibeFlow 页面找到目标数据管道，在操作列单击删除，确认后完成删除。
ossutil
使用示例
ossutil api do-data-pipe-line-action --action deleteDataPipelineConfiguration \
--parameters dataPipelineName=my-data-pipeline \
--region cn-hangzhouSDK
Java
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);
}
}
}Go
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
调用DeleteDataPipelineConfiguration接口删除数据管道。
数据管道状态说明
状态 | 说明 | 推荐操作 |
准备中 | 管道已创建，正在初始化 | 稍等片刻自动进入运行中 |
运行中 | 存量数据处理中，或正在监听增量数据 | 正常使用，可查看详情了解进度 |
已完成 | 存量数据全部向量化完成（仅 Historical 模式） | 前往向量 Bucket 执行检索 |
已暂停 | 手动暂停或索引写满自动暂停 | 查看原因，处理后重启 |
失败 | 配置错误（API Key 无效、权限不足等）导致终止 | 查看失败原因，修正后重建管道 |
错误排查
错误记录文件
若创建管道时选择了跳过异常并继续执行后续处理任务，同时获取异常信息详情，向量化失败的文件错误详情会以 Object 形式写入错误信息 Bucket。
错误记录文件的路径格式为：
{ErrorBucket}/{ErrorPrefix}/{DataPipelineName}/{ObjectKey}-{毫秒时间戳}.log示例：若 ErrorBucket 为 my-error-bucket，ErrorPrefix 为 pipeline-errors，管道名为 my-image-pipeline，原始文件 Key 为 images/photo.jpg，则错误记录路径为：
my-error-bucket/pipeline-errors/my-image-pipeline/images/photo.jpg-1735689600000.log错误记录内容
每条错误记录文件包含以下字段：
字段 | 说明 |
ErrorCode | 错误码，用于标识失败原因的类型 |
ErrorMessage | 错误描述，对失败原因的文字说明 |
RequestId | 百炼侧的请求 ID，可提供给阿里云百炼技术支持用于问题定位 |
常见错误及处理建议
错误场景 | 可能原因 | 处理建议 |
向量化请求被拒绝 | API Key 无效或已过期 | 在百炼控制台确认 Key 状态，重建管道 |
向量化请求超限 | RPM/TPM 达到上限 | 联系百炼提高限流，任务会自动排队重试 |
向量写入失败 | 维度与索引不匹配 | 确认所选百炼模型的输出维度与向量索引维度一致；重建数据管道 |
源文件读取失败 | 文件已删除 | 核实源文件是否仍存在；无需处理，管道会跳过并继续 |
向量索引已满 | 索引行数达 20 亿行 | 创建新的向量索引，并新建数据管道继续处理 |