Agent数据去重清洗

适用人群

核心价值

• 无LLM调用，零AI成本。

• 精确+近似两级去重，覆盖完全相同和字面高度相似的重复。

• 秒级处理大规模日志（单批万级数据秒级完成）。

• 支持跨批次全局去重（可选），避免历史数据重复入库。

JSON API配置

以下为完整的 Pipeline JSON 配置示例，需要将source中的project、logstore和query替换为实际值。

{
  "name": "agent_data_cleaning",
  "description": "Agent 数据去重清洗：精确 + 近似两级去重，快速产出干净数据集",
  "source": {
    "type": "logstore",
    "logstore": {
      "project": "your-project",
      "logstore": "your-agent-logstore",
      "query": "serviceName:your-agent-service and *"
    }
  },
  "pipeline": {
    "nodes": [
      {
        "id": "select_fields",
        "type": "project",
        "parameters": {
          "input": "attributes.input.value",
          "output": "attributes.output.value",
          "model": "attributes.gen_ai.model_name",
          "trace_id": "traceId",
          "span_id": "spanId"
        }
      },
      {
        "id": "extract",
        "type": "extend",
        "parameters": {
          "question": "regexp_extract(input, '(?s)用户提问原文：\\s*(.*?)\\s*,\\s*\"files\"', 1)"
        }
      },
      {
        "id": "filter_empty",
        "type": "where",
        "parameters": {
          "filter": "question IS NOT NULL AND length(question) > 0"
        }
      },
      {
        "id": "exact_dedup",
        "type": "dedup-exact",
        "parameters": {
          "field": "question"
        }
      },
      {
        "id": "fuzzy_dedup",
        "type": "dedup-fuzzy",
        "parameters": {
          "field": "question",
          "threshold": "3"
        }
      }
    ]
  },
  "sink": {
    "type": "dataset",
    "dataset": {
      "workspace": "your-workspace",
      "dataset": "agent_data_cleaned"
    }
  },
  "executePolicy": {
    "mode": "scheduled",
    "scheduled": {
      "fromTime": 1735689600,
      "interval": "15m"
    }
  }
}

数据源参数

Pipeline 节点参数

输出参数

调度参数

前提条件

• 已开通日志服务，并创建了Project和Logstore。关于Pipeline的基本概念，参见 Pipeline。

• Agent 日志已接入目标Logstore。

• 已创建用于存储清洗结果的数据集（Dataset）。

操作步骤

1. 根据实际的Agent 日志Schema，修改project 节点的字段映射。将parameters中的Key保持不变，将 Value 替换为实际日志中对应的字段路径。

2. 修改extend 节点的正则表达式，使其匹配实际日志中用户提问的格式。

   说明

   如果日志中用户提问已经是独立字段（无需正则提取），可以将extend 节点替换为直接引用该字段，或删除此节点并修改后续节点的field参数。

3. 根据需要调整where 节点的过滤条件。例如，添加按服务名或时间范围过滤的条件。

4. 根据数据特征调整dedup-fuzzy的threshold参数。

   说明

   threshold表示允许的SimHash 差异位数。值为1时极其严格（仅去除几乎完全相同的文本），值为5时较宽松（允许更多差异）。建议从默认值3开始，根据效果调整。

5. 将source中的project、logstore和query替换为实际值，将sink中的workspace和dataset替换为目标数据集信息。

6. 通过API提交Pipeline 配置。