将用户、工作区和数据集迁移到 Argilla 2.x

本指南将帮助您将任务迁移到 Argilla V2。这些不包括 FeedbackDataset，它只是最新的可扩展数据集的临时命名约定。特定于任务的数据集是用于特定任务的数据集，例如文本分类、Token 分类等。如果您想了解 SDK 迁移的背景故事，请参阅SDK 迁移博文。此外，我们将提供关于如何在新的 Argilla V2 格式中维护您的 User 和 Workspace 的指导。

注意

旧数据集包括：DatasetForTextClassification、DatasetForTokenClassification 和 DatasetForText2Text。

FeedbackDataset 不需要迁移，因为它们已经采用 Argilla V2 格式。无论如何，由于 2.x 版本包含对搜索索引结构的更改，您应该通过启用 docker 环境变量 REINDEX_DATASET 来重新索引数据集（如果您在 HF Space 中运行 Argilla，则此步骤会自动执行）。有关更多详细信息，请参阅服务器配置文档部分。

要遵循本指南，您需要具备以下先决条件

• 运行旧数据集的 argilla 1.* 服务器实例。
• 运行的 argilla >=1.29 服务器实例。如果您没有，可以按照此Argilla 指南创建一个。
• 您的环境中安装了 argilla sdk 包。

警告

本指南将在新服务器上重新创建所有 User 和 Workspace。因此，它们将使用新的密码和 ID 创建。如果您想保留相同的密码和 ID，您可以将数据集复制到临时 v2 实例，然后将您当前的实例升级到 v2.0，并将数据集复制回您的原始实例。

如果您当前的旧数据集位于 Argilla 1.29 之后发布的服务器上，您可以选择在同一服务器上将您的旧数据集重新创建为新数据集。然后您可以将服务器升级到 Argilla 2.0 并继续在那里工作。您的旧数据集在新服务器上将不可见，但如果您需要访问它们，它们将保留在存储层中。

为了迁移指南，您需要安装新的 argilla 包。这包括一个新的 v1 模块，允许您连接到 Argilla V1 服务器。

pip install "argilla>=2.0.0"

迁移用户和工作区

本指南将引导您完成两个步骤

1. 从 Argilla V1 服务器使用新的 argilla 包检索旧的用户和工作区。
2. 基于 name 作为唯一标识符在 Argilla V2 服务器上重新创建用户和工作区。

步骤 1：检索旧的用户和工作区

您可以使用 v1 模块连接到 Argilla V1 服务器。

import argilla.v1 as rg_v1

# Initialize the API with an Argilla server less than 2.0
api_url = "<your-url>"
api_key = "<your-api-key>"
rg_v1.init(api_url, api_key)

接下来，从 Argilla V1 服务器加载数据集 User 和 Workspaces

users_v1 = rg_v1.User.list()
workspaces_v1 = rg_v1.Workspace.list()

步骤 2：重新创建用户和工作区

要在 Argilla V2 服务器上重新创建用户和工作区，您可以使用 argilla 包。

首先，实例化 Argilla 类以连接到 Argilla V2 服务器

import argilla as rg

client = rg.Argilla()

接下来，在 Argilla V2 服务器上重新创建用户和工作区

for workspace in workspaces_v1:
    rg.Workspace(
        id=workspace.id,
        name=workspace.name,
    ).create()

for user in users_v1:
    user_v2 = rg.User(
        id=user.id,
        username=user.username,
        first_name=user.first_name,
        last_name=user.last_name,
        role=user.role,
        password="<your_chosen_password>" # (1)
    ).create()

    if user.role == "owner":
       continue

    for workspace in user.workspaces:
        workspace_v2 = client.workspaces(name=workspace.name)
        if workspace_v2 is None:
            continue
        user.add_to_workspace(workspace_v2)

1. 您需要为用户选择一个新密码，要以编程方式执行此操作，您可以使用 uuid 包生成随机密码。请注意跟踪您选择的密码，因为您以后将无法检索它们。

现在您已成功将用户和工作区迁移到 Argilla V2，可以继续执行下一步。

迁移数据集

本指南将引导您完成三个步骤

1. 从 Argilla V1 服务器使用新的 argilla 包检索旧数据集。
2. 在 Argilla V2 格式中定义新数据集。
3. 将数据集记录上传到新的 Argilla V2 数据集格式和属性。

步骤 1：检索旧数据集

您可以使用 v1 模块连接到 Argilla V1 服务器。

import argilla.v1 as rg_v1

# Initialize the API with an Argilla server less than 2.0
api_url = "<your-url>"
api_key = "<your-api-key>"
rg_v1.init(api_url, api_key)

接下来，从 Argilla V1 服务器加载数据集设置和记录

dataset_name = "news-programmatic-labeling"
workspace = "demo"

settings_v1 = rg_v1.load_dataset_settings(dataset_name, workspace)
records_v1 = rg_v1.load(dataset_name, workspace)
hf_dataset = records_v1.to_datasets()

您的旧数据集现在已加载到 hf_dataset 对象中。

步骤 2：定义新数据集

在 Argilla V2 格式中定义新数据集。新的数据集格式在 argilla 包中定义。您可以使用 Settings 和 Dataset 类创建一个新数据集

首先，实例化 Argilla 类以连接到 Argilla V2 服务器

import argilla as rg

client = rg.Argilla()

接下来，定义新的数据集设置

对于单标签分类对于多标签分类对于 Token 分类对于文本生成

settings = rg.Settings(
    fields=[
        rg.TextField(name="text"), # (1)
    ],
    questions=[
        rg.LabelQuestion(name="label", labels=settings_v1.label_schema),
    ],
    metadata=[
        rg.TermsMetadataProperty(name="split"), # (2)
    ],
    vectors=[
        rg.VectorField(name='mini-lm-sentence-transformers', dimensions=384), # (3)
    ],
)

1. DatasetForTextClassification 中的默认字段是 text，但请确保您提供 record.inputs 中包含的所有字段。
2. 确保您提供数据集中所有相关的元数据字段。
3. 确保您提供数据集中所有相关的向量。

settings = rg.Settings(
    fields=[
        rg.TextField(name="text"), # (1)
    ],
    questions=[
        rg.MultiLabelQuestion(name="labels", labels=settings_v1.label_schema),
    ],
    metadata=[
        rg.TermsMetadataProperty(name="split"), # (2)
    ],
    vectors=[
        rg.VectorField(name='mini-lm-sentence-transformers', dimensions=384), # (3)
    ],
)

1. DatasetForTextClassification 中的默认字段是 text，但我们应该提供 record.inputs 中包含的所有字段。
2. 确保您提供数据集中所有相关的元数据字段。
3. 确保您提供数据集中所有相关的向量。

settings = rg.Settings(
    fields=[
        rg.TextField(name="text"),
    ],
    questions=[
        rg.SpanQuestion(name="spans", labels=settings_v1.label_schema),
    ],
    metadata=[
        rg.TermsMetadataProperty(name="split"), # (1)
    ],
    vectors=[
        rg.VectorField(name='mini-lm-sentence-transformers', dimensions=384), # (2)
    ],
)

1. 确保您提供数据集中所有相关的元数据字段。
2. 确保您提供数据集中所有相关的向量。

settings = rg.Settings(
    fields=[
        rg.TextField(name="text"),
    ],
    questions=[
        rg.TextQuestion(name="text_generation"),
    ],
    metadata=[
        rg.TermsMetadataProperty(name="split"), # (1)
    ],
    vectors=[
        rg.VectorField(name='mini-lm-sentence-transformers', dimensions=384), # (2)
    ],
)

1. 我们应该提供数据集中所有相关的元数据字段。
2. 我们应该提供数据集中所有相关的向量。

最后，在 Argilla V2 服务器上创建新数据集

dataset = rg.Dataset(name=dataset_name, workspace=workspace, settings=settings)
dataset.create()

注意

如果已存在同名数据集，则 create 方法将引发异常。您可以检查数据集是否存在并在创建新数据集之前将其删除。

dataset = client.datasets(name=dataset_name, workspace=workspace)

if dataset is not None:
    dataset.delete()

步骤 3：上传数据集记录

要将记录上传到新服务器，我们将需要将记录从 Argilla V1 格式转换为 Argilla V2 格式。新的 argilla sdk 包使用通用的 Record 类，但旧数据集具有特定的记录类。我们将需要将记录转换为通用的 Record 类。

这是一组示例函数，用于转换单标签和多标签分类的记录。您可以修改这些函数以适应您的数据集。

对于单标签分类对于多标签分类对于 Token 分类对于文本生成

def map_to_record_for_single_label(data: dict, users_by_name: dict, current_user: rg.User) -> rg.Record:
    """ This function maps a text classification record dictionary to the new Argilla record."""
    suggestions = []
    responses = []

    if prediction := data.get("prediction"):
        label, score = prediction[0].values()
        agent = data["prediction_agent"]
        suggestions.append(
            rg.Suggestion(
                question_name="label", # (1)
                value=label,
                score=score,
                agent=agent
            )
        )

    if annotation := data.get("annotation"):
        user_id = users_by_name.get(data["annotation_agent"], current_user).id
        responses.append(
            rg.Response(
                question_name="label", # (2)
                value=annotation,
                user_id=user_id
            )
        )

    return rg.Record(
        id=data["id"],
        fields=data["inputs"],
        # The inputs field should be a dictionary with the same keys as the `fields` in the settings
        metadata=data["metadata"],
        # The metadata field should be a dictionary with the same keys as the `metadata` in the settings
        vectors=data.get("vectors") or {},
        suggestions=suggestions,
        responses=responses,
    )

1. 确保 question_name 与问题设置中问题的名称匹配。

2. 确保 question_name 与问题设置中问题的名称匹配。

def map_to_record_for_multi_label(data: dict, users_by_name: dict, current_user: rg.User) -> rg.Record:
    """ This function maps a text classification record dictionary to the new Argilla record."""
    suggestions = []
    responses = []

    if prediction := data.get("prediction"):
        labels, scores = zip(*[(pred["label"], pred["score"]) for pred in prediction])
        agent = data["prediction_agent"]
        suggestions.append(
            rg.Suggestion(
                question_name="labels", # (1)
                value=labels,
                score=scores,
                agent=agent
            )
        )

    if annotation := data.get("annotation"):
        user_id = users_by_name.get(data["annotation_agent"], current_user).id
        responses.append(
            rg.Response(
                question_name="labels", # (2)
                value=annotation,
                user_id=user_id
            )
        )

    return rg.Record(
        id=data["id"],
        fields=data["inputs"],
        # The inputs field should be a dictionary with the same keys as the `fields` in the settings
        metadata=data["metadata"],
        # The metadata field should be a dictionary with the same keys as the `metadata` in the settings
        vectors=data.get("vectors") or {},
        suggestions=suggestions,
        responses=responses,
    )

1. 确保 question_name 与问题设置中问题的名称匹配。

2. 确保 question_name 与问题设置中问题的名称匹配。

def map_to_record_for_span(data: dict, users_by_name: dict, current_user: rg.User) -> rg.Record:
    """ This function maps a token classification record dictionary to the new Argilla record."""
    suggestions = []
    responses = []

    if prediction := data.get("prediction"):
        scores = [span["score"] for span in prediction]
        agent = data["prediction_agent"]
        suggestions.append(
            rg.Suggestion(
                question_name="spans", # (1)
                value=prediction,
                score=scores,
                agent=agent
            )
        )

    if annotation := data.get("annotation"):
        user_id = users_by_name.get(data["annotation_agent"], current_user).id
        responses.append(
            rg.Response(
                question_name="spans", # (2)
                value=annotation,
                user_id=user_id
            )
        )

    return rg.Record(
        id=data["id"],
        fields={"text": data["text"]},
        # The inputs field should be a dictionary with the same keys as the `fields` in the settings
        metadata=data["metadata"],
        # The metadata field should be a dictionary with the same keys as the `metadata` in the settings
        vectors=data.get("vectors") or {},
        # The vectors field should be a dictionary with the same keys as the `vectors` in the settings
        suggestions=suggestions,
        responses=responses,
    )

1. 确保 question_name 与问题设置中问题的名称匹配。

2. 确保 question_name 与问题设置中问题的名称匹配。

def map_to_record_for_text_generation(data: dict, users_by_name: dict, current_user: rg.User) -> rg.Record:
    """ This function maps a text2text record dictionary to the new Argilla record."""
    suggestions = []
    responses = []

    if prediction := data.get("prediction"):
        first = prediction[0]
        agent = data["prediction_agent"]
        suggestions.append(
            rg.Suggestion(
                question_name="text_generation", # (1)
                value=first["text"],
                score=first["score"],
                agent=agent
            )
        )

    if annotation := data.get("annotation"):
        # From data[annotation]
        user_id = users_by_name.get(data["annotation_agent"], current_user).id
        responses.append(
            rg.Response(
                question_name="text_generation", # (2)
                value=annotation,
                user_id=user_id
            )
        )

    return rg.Record(
        id=data["id"],
        fields={"text": data["text"]},
        # The inputs field should be a dictionary with the same keys as the `fields` in the settings
        metadata=data["metadata"],
        # The metadata field should be a dictionary with the same keys as the `metadata` in the settings
        vectors=data.get("vectors") or {},
        # The vectors field should be a dictionary with the same keys as the `vectors` in the settings
        suggestions=suggestions,
        responses=responses,
    )

1. 确保 question_name 与问题设置中问题的名称匹配。

2. 确保 question_name 与问题设置中问题的名称匹配。

上面的函数依赖于 users_by_name 字典和 current_user 对象来将响应分配给用户，我们需要加载现有用户。您可以从 Argilla V2 服务器检索用户和当前用户，如下所示

users_by_name = {user.username: user for user in client.users}
current_user = client.me

最后，使用 log 方法和 map 函数将记录上传到新数据集。

records = []

for data in hf_records:
    records.append(map_to_record_for_single_label(data, users_by_name, current_user))

# Upload the records to the new dataset
dataset.records.log(records)

您现在已成功将旧数据集迁移到 Argilla V2。有关如何使用 Argilla SDK 的更多指南，请参阅操作指南。