Cloud Monitor：Pipeline 參考

project：欄位選取

從未經處理資料中選取並重新命名欄位，僅保留需要的欄位。

核心參數

行數變化：不變（1:1）。無擴充列。

extend：欄位擴充

基於運算式計算新列或覆蓋已有列，支援字串處理、數學運算、條件判斷、JSON 提取等豐富的內建函數。

核心參數

行數變化：不變（1:1）。無擴充列（直接輸出新列）。

where：篩選過濾

按條件運算式過濾行，僅保留滿足條件的記錄。支援比較運算、邏輯組合、模式比對等豐富的過濾條件。

核心參數

行數變化：減少或不變。無擴充列。

make-instance：執行個體構建

將離散的事件級日誌按分組鍵彙總為行級樣本執行個體。例如將一次會話中的多條訊息合并為一行完整的對話樣本。

核心參數

行數變化：多行合并為每組一行。無擴充列。

dedup-exact：精確去重

去除完全相同的文本記錄，每組保留文本最長的一條。計算代價最低，建議作為去重的第一步。

核心參數

行數變化：減少或不變。擴充列：__dedup_hash、__dedup_weight、__dedup_rnk。

dedup-fuzzy：近似去重

去除高度相似但不完全相同的文本記錄，每組保留文本最長的一條。通過 threshold 控制相似性判定的嚴格程度。

核心參數

行數變化：減少或不變。擴充列：__dedup_hash、__dedup_weight、__dedup_rnk。

dedup-semantic：語義去重

去除含義相同但表述不同的文本記錄，通過向量距離衡量語義相似性，每組保留一條代表性記錄。

核心參數

行數變化：減少或不變。擴充列：__dedup_emb（向量，可被下遊複用）、__dedup_rid。

embedding：向量產生

對指定文字欄位產生 Embedding 向量，供下遊聚類、相似性計算等運算元使用。

核心參數

行數變化：不變（1:1）。擴充列：{as}（預設 {field}_embedding）。

doc-stats：文檔統計

計算文字欄位的文檔級統計指標，輸出包含字元數、詞數、行數的統計 JSON。

核心參數

輸出格式：JSON 對象，包含 doc_len_char（字元數）、doc_len_words（詞數）、line_counts（行數）三個數值欄位。

行數變化：不變（1:1）。擴充列：{as}（預設 __doc_stats，JSON 類型）。

semantic-cluster：語義聚類

基於 Embedding 向量為資料分配簇 ID，是實現多樣性採樣的關鍵步驟。通常與 sample 運算元配合使用，每簇取樣保證語義多樣性。

核心參數

行數變化：不變（1:1）。擴充列：__cluster_id（簇編號，從 0 開始）。

sample：隨機採樣

從資料中隨機採樣指定比例或數量的記錄。支援按分組列做分層採樣，常與 semantic-cluster 搭配實現"每簇取 N 條"的多樣性採樣。

核心參數

行數變化：減少。無擴充列。

llm-call：LLM 調用

調用大語言模型對每行資料進行智能處理，支援 Prompt 模板渲染、模型選擇和輸出格式解析。

核心參數

行數變化：不變（1:1）。擴充列：{as}（預設 __llm_result）。

agentic-call：智能體調用

調用數字員工對每行資料發起智能對話。與 llm-call 不同，agentic-call 調用使用者構建的數字員工，支援多步推理、知識庫檢索、工具調用等。

核心參數

行數變化：不變（1:1）。擴充列：{as}（預設 __agentic_result）。

頂層欄位

source 配置

定義 Pipeline 從哪裡讀取未經處理資料。

pipeline.nodes 配置

每個運算元的通用結構：

sink 配置

定義處理結果輸出到哪裡。

executePolicy 配置

定義 Pipeline 的調度執行策略。

CreatePipeline

建立一個新的 Pipeline。

POST /pipelines

請求參數

要求的權限：CreatePipeline（acs:log:*:*:project/{project}/pipeline/{name}）+ ram:PassRole。

請求樣本

POST /pipelines HTTP/1.1

{
  "PipelineName": "quality-eval",
  "PipelineType": "DATASET",
  "Description": "對話品質評估管線",
  "Source": {
    "Type": "LogStore",
    "LogStore": {
      "Project": "my-sls-project",
      "LogStore": "agent-logs"
    }
  },
  "Configuration": { ... },
  "Destination": {
    "Type": "dataset",
    "Dataset": {
      "Project": "my-workspace",
      "Dataset": "eval-results"
    }
  }
}

GetPipeline

查詢指定 Pipeline 的詳細資料。

GET /pipelines/{name}

響應包含完整的 Pipeline 配置（與建立時結構一致），以及額外的時間戳記欄位：

要求的權限：GetPipeline（acs:log:*:*:project/{project}/pipeline/{name}）。

UpdatePipeline

更新指定 Pipeline 的配置。請求體結構與 CreatePipeline 一致，限制如下：

• PipelineType 不可修改。

• Source 中的 Project 和 LogStore 不可修改（RoleArn 可更新）。

PUT /pipelines/{name}

要求的權限：UpdatePipeline（acs:log:*:*:project/{project}/pipeline/{name}）+ ram:PassRole。

DeletePipeline

刪除指定 Pipeline。

DELETE /pipelines/{name}

要求的權限：DeletePipeline（acs:log:*:*:project/{project}/pipeline/{name}）。

ListPipelines

查詢專案下的 Pipeline 列表，支援分頁和名稱過濾。

GET /pipelines?size=100&offset=0&name=keyword

響應包含 total（總數）、count（本頁條數）、pipelines（Pipeline 摘要列表，含 name、description、createTime、updateTime）。

要求的權限：ListPipelines（acs:log:*:*:project/{project}/pipeline/*）。

限制與配額