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；我很想了解它！