Skip to main content

Webhook 指南：为模型和数据集自动执行元数据质量审查

tip

Webhook 现已对所有用户开放！

本指南将带你创建一个系统，用于监听 Hub 上某个用户或组织的模型/数据集变动，并为发生变动的仓库生成“元数据审查”报告。

我们要构建什么，为什么要这样做？​

在深入技术细节之前，先简要回顾目标与动机。

模型卡片和数据集卡片是记录机器学习模型与数据集的关键工具。Hugging Face Hub 使用包含 YAML 头部的 README.md 来生成模型和数据集卡片。该 YAML 部分定义了模型或数据集的元数据。例如：

---
language: 
  - "List of ISO 639-1 code for your language"
  - lang1
  - lang2
tags:
- tag1
- tag2
license: "any valid license identifier"
datasets:
- dataset1
---

这些元数据包含模型或数据集的关键信息，帮助潜在用户了解其用途与适用范围。例如，许可证字段指明模型或数据集的使用条款。Hub 用户也可以基于 YAML 中定义的字段筛选符合条件的模型或数据集。

由于此元数据对潜在用户至关重要，确保其填写完整十分必要。在团队或组织中，成员对该 YAML 元数据块的重要性理解程度可能不一。与其由某个成员手动负责审查，不如通过自动化手段辅助完成。当仓库内容发生更新时，我们希望自动生成或更新一份元数据审查报告。对于元数据质量而言，该系统的作用类似于 CI/CD。

Metadata review

你也可以在这里查看示例报告。

使用 Hub 客户端库创建元数据审查卡片​

huggingface_hub 是一个与 Hub 交互的 Python 库。我们可以使用其中的 DatasetCard.load 或 ModelCard.load 方法来下载模型/数据集卡片。在本例中，我们会调用这些方法并返回一个 Python 字典，其中包含模型或数据集卡片 YAML 中定义的元数据。我们先编写一个简短的 Python 函数封装这些方法并处理异常：

from huggingface_hub import DatasetCard, ModelCard
from huggingface_hub.utils import EntryNotFoundError 

def load_repo_card_metadata(repo_type, repo_name):
    if repo_type == "dataset":
        try:
            return DatasetCard.load(repo_name).data.to_dict()
        except EntryNotFoundError:
            return {}
    if repo_type == "model":
        try:
            return ModelCard.load(repo_name).data.to_dict()
        except EntryNotFoundError:
            return {}

该函数会返回一个包含仓库元数据的字典（若未定义元数据，则返回空字典）。

{'license': 'afl-3.0'}

创建元数据审查报告​

在获取到元数据字典后，我们需要生成一份“审查卡片”。这里，我们假设存在一组希望填写的元数据字段（例如必须填写 license）。我们会统计所需字段的覆盖情况，计算一个百分比得分，表示元数据的完整程度。

由于已经得到元数据字典，可以遍历字典并检查特定键是否存在。若缺失，则将该键的值置为 None。

def create_metadata_key_dict(card_data, repo_type: str):
    shared_keys = ["tags", "license"]
    if repo_type == "model":
        model_keys = ["library_name", "datasets", "metrics", "co2", "pipeline_tag"]
        shared_keys.extend(model_keys)
        keys = shared_keys
        return {key: card_data.get(key) for key in keys}
    if repo_type == "dataset":
        # [...]

该函数返回一个包含我们期望元数据字段的字典；值要么为实际元数据，要么为 None，表示在 YAML 中缺失。

{'tags': None,
 'license': 'afl-3.0',
 'library_name': None,
 'datasets': None,
 'metrics': None,
 'co2': None,
 'pipeline_tag': None}

接下来便可生成元数据报告。为简洁起见，这里不贴出完整代码。完整示例可在 Hugging Face Spaces 仓库中找到。

我们编写一个函数生成 Markdown 表格，用于更清晰地展示元数据覆盖情况：

def create_metadata_breakdown_table(desired_metadata_dictionary):
    # [...]
    return tabulate(
        table_data, tablefmt="github", headers=("Metadata Field", "Provided Value")
    )

再编写一个函数计算得分（所需元数据字段中已填写的比例）：

def calculate_grade(desired_metadata_dictionary):
    # [...]
    return round(score, 2)

最后生成 Markdown 报告，将得分与表格（以及说明文字）整合起来：

def create_markdown_report(
    desired_metadata_dictionary, repo_name, repo_type, score, update: bool = False
):
    # [...]
    return report

如何自动发布审查报告？​

现已拥有 Markdown 格式的元数据审查报告。接下来使用 huggingface_hub 库将其发布到 Hub。我们编写一个函数，接收来自 Webhook 的数据、解析信息并生成元数据报告。若之前未生成过报告，则创建新讨论；若已有，则在现有线程中发布新评论。

def create_or_update_report(data):
    if parsed_post := parse_webhook_post(data):
        repo_type, repo_name = parsed_post
    else:
        return Response("Unable to parse webhook data", status_code=400)
    # [...]
    return True
tip

:= 是 Python 3.8 引入的赋值表达式（亦称“海象运算符”）。使用该语法不会影响代码逻辑，但能简化写法。可参考 Real Python 的文章了解更多。

创建 Webhook 以响应 Hub 上的变动​

我们已经具备生成元数据审查报告的核心功能，下一步是使用 Webhook 自动响应仓库变更。

在个人资料中创建 Webhook​

访问 https://huggingface.co/settings/webhooks 创建 Webhook。

  • 输入若干要监听的目标仓库（通常应限制在你自己或所属组织的仓库）。
  • 设置密钥以提升安全性（可使用密码生成器生成足够随机的字符串）。
  • 目标 URL 暂时可填写占位地址。

Webhook 设置如下：

webhook settings

创建新的 Bot 用户​

本指南创建了一个独立用户账号用于发布元数据审查结果。

Bot user account

tip

若你的机器人会在 Hub 上与其他用户互动，请清晰地将账号标注为 “Bot”（见上图）。

创建 Webhook 监听器​

需要一个 Webhook 监听程序来接收事件。市面上有很多工具可监听 Webhook（例如 Zapier 和 IFTTT 可以在模型更新时发推文等）。此处我们使用 FastAPI 实现。

FastAPI 是 Python Web 框架，我们可用它创建一个监听 /webhook 路径 POST 请求的路由。认证方式为比较 X-Webhook-Secret 头与在 Docker 容器运行时传入的 WEBHOOK_SECRET。

from fastapi import FastAPI, Request, Response
import os

KEY = os.environ.get("WEBHOOK_SECRET")

app = FastAPI()

@app.post("/webhook")
async def webhook(request: Request):
    if request.method == "POST":
        if request.headers.get("X-Webhook-Secret") != KEY:
            return Response("Invalid secret", status_code=401)
        data = await request.json()
        result = create_or_update_report(data)
        return "Webhook received!" if result else result

上述函数会接收 Webhook 事件，并为变动的仓库创建或更新审查报告。

使用 Spaces 部署 Webhook 应用​

我们的 main.py 文件即为 Webhook 应用全部代码。部署时，我们使用 Space。

Space 采用 Docker 运行。其 Dockerfile 会复制应用文件、安装依赖并启动应用。为给 KEY 变量赋值，我们在 Space 中设置 WEBHOOK_SECRET，其值与之前生成的密钥一致。更多关于 Docker Space 的信息请参考文档。

最后，需将 Webhook 设置的 URL 更新为 Space 的地址。可在上下文菜单中选择 “Embed this Space”，然后复制 “Direct URL”。

direct url

获取 URL 后，将其填入 Webhook 设置中的 Webhook URL。完成后，机器人便会在监听的仓库发生变动时自动发布审查报告！

总结与下一步​

至此，我们搭建了一个自动化的元数据审查机器人。以下是一些扩展思路：

  • 本例的元数据审查较为基础，你可以加入更复杂的审查规则。
  • 可以利用整个 README.md 进行审查。
  • 可为团队设定一些重要规则，并借助 Webhook 检查这些规则是否被遵守。

如果你使用 Webhook 构建了元数据质量应用，请记得 @davanstrien，我很乐意了解你的成果！

Last updated on