Webhooks
Webhooks 是与 MLOps 相关的功能的基础。它们允许您侦听特定仓库或您感兴趣的特定用户/组织所属的所有仓库上的新更改。本指南首先将说明如何以编程方式管理 Webhooks。然后,我们将了解如何利用 huggingface_hub
创建一个侦听 Webhooks 的服务器并将其部署到空间。
本指南假设您熟悉 Huggingface Hub 上 Webhooks 的概念。要详细了解 Webhooks 本身,您应该先阅读此 指南。
管理 Webhooks
huggingface_hub
允许您以编程方式管理您的 Webhooks。您可以列出您现有的 Webhooks,创建新的 Webhooks,以及更新、启用、禁用或删除它们。本节将指导您使用 Hugging Face Hub 的 API 功能完成这些过程。
创建 Webhook
要创建一个新的 Webhook,请使用 create_webhook() 并指定应将有效负载发送到的 URL、要监视的事件,以及可选地设置用于安全性的域名和密钥。
from huggingface_hub import create_webhook
# Example: Creating a webhook
webhook = create_webhook(
url="https://webhook.site/your-custom-url",
watched=[{"type": "user", "name": "your-username"}, {"type": "org", "name": "your-org-name"}],
domains=["repo", "discussion"],
secret="your-secret"
)
列出 Webhook
要查看您已配置的所有 Webhook,您可以使用 list_webhooks() 列出它们。这有助于查看它们的 ID、URL 和状态。
from huggingface_hub import list_webhooks
# Example: Listing all webhooks
webhooks = list_webhooks()
for webhook in webhooks:
print(webhook)
更新 Webhook
如果您需要更改现有 Webhook 的配置,例如 URL 或其监视的事件,您可以使用 update_webhook() 更新它。
from huggingface_hub import update_webhook
# Example: Updating a webhook
updated_webhook = update_webhook(
webhook_id="your-webhook-id",
url="https://new.webhook.site/url",
watched=[{"type": "user", "name": "new-username"}],
domains=["repo"]
)
启用和禁用 Webhook
您可能希望暂时禁用 Webhook 而无需删除它。这可以通过使用 disable_webhook() 来完成,并且稍后可以使用 enable_webhook() 重新启用 Webhook。
from huggingface_hub import enable_webhook, disable_webhook
# Example: Enabling a webhook
enabled_webhook = enable_webhook("your-webhook-id")
print("Enabled:", enabled_webhook)
# Example: Disabling a webhook
disabled_webhook = disable_webhook("your-webhook-id")
print("Disabled:", disabled_webhook)
删除 Webhook
当不再需要 Webhook 时,可以使用 delete_webhook() 永久删除它。
from huggingface_hub import delete_webhook
# Example: Deleting a webhook
delete_webhook("your-webhook-id")
Webhook 服务器
我们将在本指南部分中使用的基类是 WebhooksServer()。它是一个用于轻松配置服务器的类,该服务器可以接收来自 Huggingface Hub 的 Webhook。该服务器基于 Gradio 应用程序。它具有一个 UI,用于向您或您的用户显示说明,以及一个用于侦听 Webhook 的 API。
要查看 Webhook 服务器的运行示例,请查看 Spaces CI Bot。它是一个在 Space 上打开 PR 时启动临时环境的 Space。
这是一个 实验性功能。这意味着我们仍在努力改进 API。将来可能会在未经事先通知的情况下引入重大更改。请确保在您的需求中固定 huggingface_hub
的版本。
创建端点
实现 Webhook 端点就像装饰函数一样简单。让我们看一个第一个示例来解释主要概念
# app.py
from huggingface_hub import webhook_endpoint, WebhookPayload
@webhook_endpoint
async def trigger_training(payload: WebhookPayload) -> None:
if payload.repo.type == "dataset" and payload.event.action == "update":
# Trigger a training job if a dataset is updated
...
将此代码段保存在名为 'app.py'
的文件中,并使用 'python app.py'
运行它。您应该会看到如下消息
Webhook secret is not defined. This means your webhook endpoints will be open to everyone. To add a secret, set `WEBHOOK_SECRET` as environment variable or pass it at initialization: `app = WebhooksServer(webhook_secret='my_secret', ...)` For more details about webhook secrets, please refer to https://huggingface.co/docs/hub/webhooks#webhook-secret. Running on local URL: http://127.0.0.1:7860 Running on public URL: https://1fadb0f52d8bf825fc.gradio.live This share link expires in 72 hours. For free permanent hosting and GPU upgrades (NEW!), check out Spaces: https://huggingface.co/spaces Webhooks are correctly setup and ready to use: - POST https://1fadb0f52d8bf825fc.gradio.live/webhooks/trigger_training Go to https://huggingface.co/settings/webhooks to setup your webhooks.
干得好!您刚刚启动了一个 Webhook 服务器!让我们详细了解一下发生了什么
- 通过使用 webhook_endpoint() 装饰函数,
WebhooksServer()
对象已在后台创建。如您所见,此服务器是在 http://127.0.0.1:7860 上运行的 Gradio 应用程序。如果您在浏览器中打开此 URL,您将看到一个包含有关已注册 Webhook 的说明的登录页面。 - Gradio 应用程序在后台是一个 FastAPI 服务器。一个新的 POST 路由
/webhooks/trigger_training
已添加到其中。这是将侦听 Webhook 并触发时运行trigger_training
函数的路由。FastAPI 将自动解析有效负载并将其作为 WebhookPayload 对象传递给函数。这是一个pydantic
对象,其中包含有关触发 Webhook 的事件的所有信息。 - Gradio 应用程序还打开了一个隧道以接收来自互联网的请求。这是有趣的部分:您可以在 https://huggingface.co/settings/webhooks 上配置一个 Webhook,指向您的本地机器。这对于调试您的 Webhook 服务器并在部署到 Space 之前快速迭代非常有用。
- 最后,日志还告诉您您的服务器当前不受密钥保护。这对于本地调试来说没有问题,但需要牢记以备将来使用。
默认情况下,服务器将在脚本末尾启动。如果您在笔记本中运行它,则可以通过调用 decorated_function.run()
手动启动服务器。由于使用了唯一的服务器,因此即使您有多个端点,也只需启动一次服务器。
配置 Webhook
现在您已运行了一个 Webhook 服务器,您希望配置一个 Webhook 以开始接收消息。转到 https://huggingface.co/settings/webhooks,点击“添加新的 Webhook”并配置您的 Webhook。设置您想要监视的目标存储库和 Webhook URL,这里为 https://1fadb0f52d8bf825fc.gradio.live/webhooks/trigger_training
。
就是这样!您现在可以通过更新目标存储库(例如推送提交)来触发该 Webhook。查看您的 Webhook 的“活动”选项卡以查看已触发的事件。现在您拥有了一个可用的设置,您可以对其进行测试并快速迭代。如果您修改代码并重新启动服务器,您的公共 URL 可能会更改。如有必要,请确保更新 Hub 上的 Webhook 配置。
部署到 Space
现在您已经拥有了一个可工作的 Webhook 服务器,目标是将其部署到一个 Space 中。访问 https://huggingface.co/new-space 创建一个 Space。为其命名,选择 Gradio SDK 并点击“创建 Space”。将您的代码上传到 Space 中名为 app.py
的文件中。您的 Space 将自动启动!有关 Spaces 的更多详细信息,请参考此 指南。
您的 Webhook 服务器现在正在一个公共 Space 上运行。在大多数情况下,您需要使用密钥对其进行保护。转到您的 Space 设置 > “仓库密钥”部分 > “添加密钥”。将 WEBHOOK_SECRET
环境变量设置为您的任意值。返回 Webhook 设置 并设置 Webhook 配置中的密钥。现在,只有具有正确密钥的请求才会被您的服务器接受。
就是这样!您的 Space 现在已准备好接收来自 Hub 的 Webhook。请记住,如果您在免费的“cpu-basic”硬件上运行 Space,则在 48 小时的空闲时间后它将被关闭。如果您需要一个永久的 Space,则应考虑将其设置为 升级后的硬件。
高级用法
以上指南解释了设置 WebhooksServer() 的最快方法。在本节中,我们将了解如何进一步自定义它。
多个端点
您可以在同一服务器上注册多个端点。例如,您可能希望有一个端点触发训练作业,另一个端点触发模型评估。您可以通过添加多个 @webhook_endpoint
装饰器来实现此目的。
# app.py
from huggingface_hub import webhook_endpoint, WebhookPayload
@webhook_endpoint
async def trigger_training(payload: WebhookPayload) -> None:
if payload.repo.type == "dataset" and payload.event.action == "update":
# Trigger a training job if a dataset is updated
...
@webhook_endpoint
async def trigger_evaluation(payload: WebhookPayload) -> None:
if payload.repo.type == "model" and payload.event.action == "update":
# Trigger an evaluation job if a model is updated
...
这将创建两个端点。
(...) Webhooks are correctly setup and ready to use: - POST https://1fadb0f52d8bf825fc.gradio.live/webhooks/trigger_training - POST https://1fadb0f52d8bf825fc.gradio.live/webhooks/trigger_evaluation
自定义服务器
为了获得更大的灵活性,您还可以直接创建一个 WebhooksServer() 对象。如果您想自定义服务器的登录页面,这将非常有用。您可以通过传递一个 Gradio UI 来覆盖默认的 UI。例如,您可以为用户添加说明或添加一个表单以手动触发 Webhook。创建 WebhooksServer() 时,您可以使用 add_webhook()
装饰器注册新的 Webhook。
这是一个完整的示例。
import gradio as gr
from fastapi import Request
from huggingface_hub import WebhooksServer, WebhookPayload
# 1. Define UI
with gr.Blocks() as ui:
...
# 2. Create WebhooksServer with custom UI and secret
app = WebhooksServer(ui=ui, webhook_secret="my_secret_key")
# 3. Register webhook with explicit name
@app.add_webhook("/say_hello")
async def hello(payload: WebhookPayload):
return {"message": "hello"}
# 4. Register webhook with implicit name
@app.add_webhook
async def goodbye(payload: WebhookPayload):
return {"message": "goodbye"}
# 5. Start server (optional)
app.run()
- 我们使用 Gradio 块定义了一个自定义 UI。此 UI 将显示在服务器的登录页面上。
- 我们使用自定义 UI 和密钥创建了一个 WebhooksServer() 对象。密钥是可选的,可以使用
WEBHOOK_SECRET
环境变量设置。 - 我们使用显式名称注册了一个 Webhook。这将在
/webhooks/say_hello
创建一个端点。 - 我们使用隐式名称注册了一个 Webhook。这将在
/webhooks/goodbye
创建一个端点。 - 我们启动服务器。这是可选的,因为您的服务器将在脚本结束时自动启动。