使用 Argilla Webhook¶

本指南概述了如何在 Argilla 中创建和使用 Webhook。

Webhook 允许应用程序在特定事件发生时向其他应用程序提交实时信息。与传统的 API 不同，您无需非常频繁地轮询数据即可实时获取数据。这使得 Webhook 对于提供者和消费者来说都更加高效。

在 Argilla 中创建 Webhook 监听器¶

Python SDK 提供了一种在 Argilla 中创建 Webhook 的简单方法。它允许您专注于 Webhook 的用例，而不是实现细节。您只需要使用 webhook_listener 装饰器创建您的事件处理函数。

import argilla as rg

from datetime import datetime
from argilla import webhook_listener

@webhook_listener(events="dataset.created")
async def my_webhook_handler(dataset: rg.Dataset, type: str, timestamp: datetime):
    print(dataset, type, timestamp)

在上面的示例中，我们创建了一个监听 dataset.created 事件的 Webhook。

您可以在事件部分找到事件列表。

Python SDK 将自动在 Argilla 中创建 Webhook 并监听指定的事件。当事件被触发时，将使用事件数据调用 my_webhook_handler 函数。SDK 还会将传入的 Webhook 事件解析为正确的资源对象（rg.Dataset、rg.Record 和 rg.Response）。SDK 还会处理请求身份验证和错误处理。

运行 Webhook 服务器¶

在底层，SDK 使用 FastAPI 框架创建 Webhook 服务器和 POST 端点以接收 Webhook 事件。

要运行 Webhook，您需要在代码中定义 Webhook 服务器，并使用 uvicorn 命令启动它。

# my_webhook.py file
from argilla import get_webhook_server

server = get_webhook_server()

uvicorn my_webhook:server

您可以通过访问 http://localhost:8000/docs 来浏览 Swagger UI 以探索您定义的 Webhook。

uvicorn 命令将在默认端口 8000 上启动 Webhook 服务器。

默认情况下，Python SDK 将使用服务器 URL http://127.0.0.1:8000/ 注册 Webhook。如果您想使用不同的服务器 URL，您可以设置 WEBHOOK_SERVER_URL 环境变量。

export WEBHOOK_SERVER_URL=http://my-webhook-server.com

所有传入的 Webhook 事件都将发送到指定的服务器 URL。

Webhook 管理¶

Python SDK 提供了一种在 Argilla 中管理 Webhook 的简单方法。您可以使用 SDK 创建、列出、更新和删除 Webhook。

创建 Webhook¶

要在 Argilla 中创建新的 Webhook，您可以在 Webhook 类中定义它，然后调用 create 方法。

import argilla as rg

client = rg.Argilla(api_url="<api_url>", api_key="<api_key>")

webhook = rg.Webhook(
    url="http://127.0.0.1:8000",
    events=["dataset.created"],
    description="My webhook"
)

webhook.create()

列出 Webhook¶

您可以通过访问 Argilla 类上的 webhooks 属性并迭代它们来列出 Argilla 中所有现有的 Webhook。

import argilla as rg

client = rg.Argilla(api_url="<api_url>", api_key="<api_key>")

for webhook in client.webhooks:
    print(webhook)

更新 Webhook¶

您可以使用 update 方法更新 Webhook。

import argilla as rg

client = rg.Argilla(api_url="<api_url>", api_key="<api_key>")

webhook = rg.Webhook(
    url="http://127.0.0.1:8000",
    events=["dataset.created"],
    description="My webhook"
).create()

webhook.events = ["dataset.updated"]
webhook.update()

您应该使用 IP 地址而不是 localhost，因为 Webhook 验证期望 URL 中包含顶级域名 (TLD)。

删除 Webhook¶

您可以使用 delete 方法删除 Webhook。

import argilla as rg

client = rg.Argilla(api_url="<api_url>", api_key="<api_key>")

for webhook in client.webhooks:
    webhook.delete()

在 Hugging Face Space 中部署 Webhook 服务器¶

您可以在 Hugging Face Space 中部署您的 Webhook。您可以访问此链接以探索在 Hugging Face Space 中部署的 Webhook 服务器示例。

事件¶

以下是您可以在 Argilla 中监听的事件列表，按资源类型分组。

数据集事件¶

dataset.created：数据集资源已创建。
dataset.updated：数据集资源已更新。
dataset.deleted：数据集资源已删除。
dataset.published：数据集资源已发布。

记录事件¶

record.created：记录资源已创建。
record.updated：记录资源已更新。
record.deleted：记录资源已删除。
record.completed：记录资源已完成 (status="completed")。

响应事件¶

response.created：响应资源已创建。
response.updated：响应资源已更新。
response.deleted：响应资源已删除。