Tag: references

Step by step build a complete Data Quality Skill from scratch.
“Please analyze this Excel file and give me a data quality report.”
we will build skill like this.

data-quality-skill/
│
├── SKILL.md
│
├── scripts/
│   ├── profile_excel.py
│   ├── check_nulls.py
│   ├── check_duplicates.py
│   └── generate_report.py
│
├── references/
│   └── data-quality-rules.md
│
└── resources/
    └── report_template.md

2. Design the capability first, then the workflow, and then the scripts.

3. Agent 到底要完成什么任务？

Agent should do:

• Identify the input file – 找到输入文件
• Profile the dataset – Profile 数据集
• Check NULL values – 检查 NULL
• Check duplicate rows – 检查重复
• Check basic schema – 检查基本 Schema
• Interpret the results – 解释结果
• Generate a report – 生成报告

4. Design Workflow

User Request   用户请求
     ↓
Identify Input   确定输入
     ↓
Profile Dataset   分析数据集
     ↓
Run Quality Checks   执行质量检查
     ├── NULL check   NULL 检查
     ├── Duplicate check   重复检查
     └── Schema check   Schema 检查
     ↓
Interpret Results   解释结果
     ↓
Generate Report   生成报告

5. 第一步：创建目录

</> Bash

mkdir -p data-quality-skill/scripts
mkdir -p data-quality-skill/references
mkdir -p data-quality-skill/resources

## looks this (SKILL.md is file rather than directory)

data-quality-skill/
├── SKILL.md
├── scripts/
├── references/
└── resources/

6. 第二步：profile_excel.py

我们先做第一个真正的执行工具, 它负责：获取数据事实。

</> Python

"""
profile_excel.py

Purpose / 目的:
    Profile an Excel dataset and return deterministic statistics.
    对 Excel 数据集进行 Profile，并返回确定性的统计结果。

Why this script exists / 为什么需要这个脚本:
    The LLM should reason about data-quality results,
    but it should NOT manually calculate dataset statistics.

    LLM 应该负责解释数据质量结果，
    而不是自己手工计算数据统计信息。
"""

from pathlib import Path
import json
import sys

import pandas as pd


def profile_excel(file_path: str) -> dict:
    """
    Read an Excel file and calculate basic profiling information.

    读取 Excel 文件并计算基础 Profile 信息。
    """

    # ---------------------------------------------------------
    # Validate the input path.
    # 验证输入文件路径。
    # ---------------------------------------------------------
    path = Path(file_path)

    if not path.exists():
        raise FileNotFoundError(
            f"File does not exist: {file_path}"
        )

    # ---------------------------------------------------------
    # Read the Excel file.
    # 读取 Excel 文件。
    #
    # pandas performs deterministic data processing.
    # pandas 执行确定性的数据处理。
    # ---------------------------------------------------------
    df = pd.read_excel(path)

    # ---------------------------------------------------------
    # Build structured profiling results.
    # 构建结构化 Profile 结果。
    #
    # The result is deliberately JSON-friendly.
    # 结果故意设计成 JSON-friendly 格式。
    # ---------------------------------------------------------
    result = {
        # Total number of rows.
        # 总行数。
        "row_count": int(len(df)),

        # Total number of columns.
        # 总列数。
        "column_count": int(len(df.columns)),

        # Column names.
        # 列名。
        "columns": list(df.columns),

        # Data type of every column.
        # 每一列的数据类型。
        "data_types": {
            column: str(dtype)
            for column, dtype in df.dtypes.items()
        },

        # NULL count for every column.
        # 每一列的 NULL 数量。
        "null_counts": {
            column: int(count)
            for column, count in df.isnull().sum().items()
        },

        # Total number of completely duplicated rows.
        # 完全重复的行数量。
        "duplicate_rows": int(df.duplicated().sum()),
    }

    return result


def main():
    """
    Command-line entry point.
    命令行程序入口。

    This makes the script independently executable.
    这样这个脚本可以独立执行。
    """

    # ---------------------------------------------------------
    # Validate command-line arguments.
    # 验证命令行参数。
    # ---------------------------------------------------------
    if len(sys.argv) != 2:
        print(
            "Usage: python profile_excel.py <excel_file>",
            file=sys.stderr
        )
        sys.exit(1)

    file_path = sys.argv[1]

    try:
        # -----------------------------------------------------
        # Execute the profiling logic.
        # 执行 Profile 逻辑。
        # -----------------------------------------------------
        result = profile_excel(file_path)

        # -----------------------------------------------------
        # Convert Python dictionary into JSON.
        # 把 Python dictionary 转换成 JSON。
        #
        # JSON is easy for Agents and other programs to consume.
        # JSON 很容易被 Agent 和其他程序消费。
        # -----------------------------------------------------
        print(
            json.dumps(
                {
                    "status": "success",
                    "data": result
                },
                indent=2
            )
        )

    except Exception as exc:

        # -----------------------------------------------------
        # Return structured error information.
        # 返回结构化错误信息。
        # -----------------------------------------------------
        print(
            json.dumps(
                {
                    "status": "error",
                    "error": str(exc)
                },
                indent=2
            ),
            file=sys.stderr
        )

        sys.exit(1)


# -------------------------------------------------------------
# Only execute main() when this file is run directly.
# 只有直接运行这个文件时才执行 main()。
# -------------------------------------------------------------
if __name__ == "__main__":
    main()