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 元数据中定义的字段作为筛选器，用于识别符合特定标准的模型或数据集。

由于此块中定义的元数据对于我们模型和数据集的潜在用户至关重要，因此我们必须完成此部分。在团队或组织环境中，将模型和数据集推送到 Hub 的用户可能对此 YAML 元数据块的重要性有不同的熟悉程度。虽然团队中的某个人可以负责审查此元数据，但我们可以进行一些自动化来帮助我们解决此问题。结果将是在 Hub 上的存储库更改时自动发布或更新的元数据审查报告。对于我们的元数据质量，此系统类似于CI/CD。

您还可以在此处找到一个示例审查报告。

使用 Hub 客户端库创建模型审查卡片

huggingface_hub 是一个 Python 库，允许您与 Hub 交互。我们可以使用此库通过 DatasetCard.load 或 ModelCard.load 方法从 Hub 下载模型和数据集卡片。特别是，我们将使用这些方法加载一个 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 {}

此函数将返回一个包含与存储库关联的元数据的 Python 字典（如果没有元数据，则返回一个空字典）。

{'license': 'afl-3.0'}

创建我们的元数据审查报告

一旦我们有一个包含与存储库关联的元数据的 Python 字典，我们将为我们的元数据审查创建一个“报告卡”。在这个特定的实例中，我们将通过定义一些我们想要值的元数据字段来审查我们的元数据。例如，我们可能希望确保 license 字段始终已完成。为了评估我们的元数据，我们将计算我们所需字段中有多少元数据字段存在，并根据我们想要看到值的所需元数据字段的覆盖率返回一个百分比分数。

由于我们有一个包含元数据的 Python 字典，我们可以遍历此字典以检查我们所需的键是否存在。如果所需的元数据字段（我们字典中的一个键）缺失，我们将值指定为 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":
        # [...]

此函数将返回一个字典，其中包含表示我们模型或数据集所需的元数据字段的键。字典值将包含该字段输入的元数据，或者如果该元数据字段在 YAML 中缺失，则为 None。

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

有了这个字典，我们就可以创建元数据报告了。为了简洁起见，我们在此处不包含完整的代码，但此 Webhook 的 Hugging Face Spaces 存储库包含完整的代码。

我们创建一个函数，该函数创建一个 Markdown 表，生成我们元数据覆盖率字典中数据的更美观版本。

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

我们还有一个 Python 函数，它生成一个分数（表示所需元数据字段存在的百分比）

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

以及一个 Python 函数，它为我们的元数据审查创建 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

创建 Webhook 以响应 Hub 上的更改

我们现在已经掌握了为模型或数据集创建元数据审查报告的核心功能。下一步是使用 Webhooks 自动响应更改。

在您的用户资料中创建 Webhook

首先，通过访问 https://huggingface.co/settings/webhooks 创建您的 Webhook。

• 输入您的 Webhook 将监听的一些目标存储库（您可能希望将其限制为您自己的存储库或您所属组织的存储库）。
• 输入一个密钥以使您的 Webhook 更安全（如果您不知道该选择什么，您可能需要使用密码生成器来生成足够长的随机字符串作为您的密钥）。
• 我们现在可以为 Webhook URL 参数传递一个虚拟 URL。

您的 Webhook 将如下所示

创建新的 Bot 用户资料

本指南创建了一个单独的用户帐户，该帐户将发布元数据审查。

创建 Webhook 监听器

我们现在需要某种方式来监听 Webhook 事件。有许多工具可以用来监听 Webhook 事件。许多现有服务，如 Zapier 和 IFTTT，可以使用 Webhook 触发操作（例如，它们可以在模型更新时发布一条推文）。在这种情况下，我们将使用 FastAPI 实现我们的 Webhook 监听器。

FastAPI 是一个 Python Web 框架。我们将使用 FastAPI 来创建 Webhook 监听器。特别是，我们需要实现一个在 /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 Spaces 的信息。

最后，我们需要将 Webhook 设置中的 URL 更新为我们 Space 的 URL。我们可以从上下文菜单中获取 Space 的“直接 URL”。点击“嵌入此 Space”并复制“直接 URL”。

获取此 URL 后，我们就可以将其传递给 Webhook 设置中的 Webhook URL 参数了。我们的机器人现在应该在受监控的存储库发生更改时开始发布审查报告！

结论和后续步骤

我们现在有了一个自动元数据审查机器人！以下是您可以根据本指南进行构建的一些想法：

• 我们的机器人进行的元数据审查相对粗糙；您可以添加更复杂的规则来审查元数据。
• 您可以使用完整的 README.md 文件进行审查。
• 您可能希望定义对您的组织特别重要的“规则”，并使用 Webhook 来检查这些规则是否得到遵守。

如果您使用 Webhooks 构建了一个元数据质量应用程序，请标记我 @davanstrien；我很想了解它！