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/*）。

限制与配额