資料管道（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 億行 | 建立新的向量索引，並建立資料管道繼續處理 |