rg.Settings

rg.Settings 用于定义 Argilla Dataset 的设置。这些设置可用于配置数据集的行为，例如字段、问题、指南、元数据和向量。Settings 类被传递给 Dataset 类，用于在服务器上创建数据集。一旦创建，数据集的设置就无法更改。

使用示例

使用设置创建新数据集

要使用设置创建新数据集，请实例化 Settings 类并将其传递给 Dataset 类。

import argilla as rg

settings = rg.Settings(
    guidelines="Select the sentiment of the prompt.",
    fields=[rg.TextField(name="prompt", use_markdown=True)],
    questions=[rg.LabelQuestion(name="sentiment", labels=["positive", "negative"])],
)

dataset = rg.Dataset(name="sentiment_analysis", settings=settings)

# Create the dataset on the server
dataset.create()

要定义字段、问题、元数据、向量或分布的设置，请参阅 rg.TextField、rg.LabelQuestion、rg.TermsMetadataProperty 和 rg.VectorField、rg.TaskDistribution 类文档。

向设置添加或删除属性

设置对象可以在创建数据集之前进行修改，方法是使用 settings.add 和 settings.<>.remove 方法添加、替换或删除属性

import argilla as rg

settings = rg.Settings(
    guidelines="Select the sentiment of the prompt.",
    fields=[rg.TextField(name="prompt", use_markdown=True)],
    questions=[rg.LabelQuestion(name="sentiment", labels=["positive", "negative"])],
)

# Adding a new property
settings.add(rg.TextField(name="response", use_markdown=True))

# Replace an existing property by other property type
settings.add(rg.TextQuestion(name="response", use_markdown=False))

# Remove an existing property
settings.questions.remove("response")

使用内置模板创建设置

Argilla 提供了内置模板，用于为常见数据集类型创建设置。要使用模板，请使用 Settings 类的类方法。有三个内置模板可用于分类、排序和评分任务。模板设置还包括默认指南和映射。

分类任务

您可以使用 rg.Settings.for_classification 类方法定义分类任务。这将创建一个包含文本字段和标签问题的数据集。您可以使用带有 image 或 text 的 field_type 参数选择字段类型。

settings = rg.Settings.for_classification(labels=["positive", "negative"]) # (1)

这将返回一个包含以下设置的 Settings 对象

settings = Settings(
    guidelines="Select a label for the document.",
    fields=[rg.TextField(field_type)(name="text")],
    questions=[LabelQuestion(name="label", labels=labels)],
    mapping={"input": "text", "output": "label", "document": "text"},
)

排序任务

您可以使用 rg.Settings.for_ranking 类方法定义排序任务。这将创建一个包含文本字段和排序问题的数据集。

settings = rg.Settings.for_ranking()

这将返回一个包含以下设置的 Settings 对象

settings = Settings(
    guidelines="Rank the responses.",
    fields=[
        rg.TextField(name="instruction"),
        rg.TextField(name="response1"),
        rg.TextField(name="response2"),
    ],
    questions=[RankingQuestion(name="ranking", values=["response1", "response2"])],
    mapping={
        "input": "instruction",
        "prompt": "instruction",
        "chosen": "response1",
        "rejected": "response2",
    },
)

评分任务

您可以使用 rg.Settings.for_rating 类方法定义评分任务。这将创建一个包含文本字段和评分问题的数据集。

settings = rg.Settings.for_rating()

这将返回一个包含以下设置的 Settings 对象

settings = Settings(
    guidelines="Rate the response.",
    fields=[
        rg.TextField(name="instruction"),
        rg.TextField(name="response"),
    ],
    questions=[RatingQuestion(name="rating", values=[1, 2, 3, 4, 5])],
    mapping={
        "input": "instruction",
        "prompt": "instruction",
        "output": "response",
        "score": "rating",
    },
)

---

Settings

基类：DefaultSettingsMixin, Resource

Argilla 数据集的 Settings 类。

此类用于定义数据集在 UI 中的表示形式。

源代码位于 src/argilla/settings/_resource.py

class Settings(DefaultSettingsMixin, Resource):
    """
    Settings class for Argilla Datasets.

    This class is used to define the representation of a Dataset within the UI.
    """

    def __init__(
        self,
        fields: Optional[List[Field]] = None,
        questions: Optional[List[QuestionType]] = None,
        vectors: Optional[List[VectorField]] = None,
        metadata: Optional[List[MetadataType]] = None,
        guidelines: Optional[str] = None,
        allow_extra_metadata: bool = False,
        distribution: Optional[TaskDistribution] = None,
        mapping: Optional[Dict[str, Union[str, Sequence[str]]]] = None,
        _dataset: Optional["Dataset"] = None,
    ) -> None:
        """
        Args:
            fields (List[Field]): A list of Field objects that represent the fields in the Dataset.
            questions (List[Union[LabelQuestion, MultiLabelQuestion, RankingQuestion, TextQuestion, RatingQuestion]]):
                A list of Question objects that represent the questions in the Dataset.
            vectors (List[VectorField]): A list of VectorField objects that represent the vectors in the Dataset.
            metadata (List[MetadataField]): A list of MetadataField objects that represent the metadata in the Dataset.
            guidelines (str): A string containing the guidelines for the Dataset.
            allow_extra_metadata (bool): A boolean that determines whether or not extra metadata is allowed in the
                Dataset. Defaults to False.
            distribution (TaskDistribution): The annotation task distribution configuration.
                Default to DEFAULT_TASK_DISTRIBUTION
            mapping (Dict[str, Union[str, Sequence[str]]]): A dictionary that maps incoming data names to Argilla dataset attributes in DatasetRecords.
        """
        super().__init__(client=_dataset._client if _dataset else None)

        self._dataset = _dataset
        self._distribution = distribution or TaskDistribution.default()
        self._mapping = mapping
        self.__guidelines = self.__process_guidelines(guidelines)
        self.__allow_extra_metadata = allow_extra_metadata

        self.__questions = SettingsProperties(self, questions)
        self.__fields = SettingsProperties(self, fields)
        self.__vectors = SettingsProperties(self, vectors)
        self.__metadata = SettingsProperties(self, metadata)

    #####################
    # Properties        #
    #####################

    @property
    def fields(self) -> "SettingsProperties":
        return self.__fields

    @fields.setter
    def fields(self, fields: List[Field]):
        self.__fields = SettingsProperties(self, fields)

    @property
    def questions(self) -> "SettingsProperties":
        return self.__questions

    @questions.setter
    def questions(self, questions: List[QuestionType]):
        self.__questions = SettingsProperties(self, questions)

    @property
    def vectors(self) -> "SettingsProperties":
        return self.__vectors

    @vectors.setter
    def vectors(self, vectors: List[VectorField]):
        self.__vectors = SettingsProperties(self, vectors)

    @property
    def metadata(self) -> "SettingsProperties":
        return self.__metadata

    @metadata.setter
    def metadata(self, metadata: List[MetadataType]):
        self.__metadata = SettingsProperties(self, metadata)

    @property
    def guidelines(self) -> str:
        return self.__guidelines

    @guidelines.setter
    def guidelines(self, guidelines: str):
        self.__guidelines = self.__process_guidelines(guidelines)

    @property
    def allow_extra_metadata(self) -> bool:
        return self.__allow_extra_metadata

    @allow_extra_metadata.setter
    def allow_extra_metadata(self, value: bool):
        self.__allow_extra_metadata = value

    @property
    def distribution(self) -> TaskDistribution:
        return self._distribution

    @distribution.setter
    def distribution(self, value: TaskDistribution) -> None:
        self._distribution = value

    @property
    def mapping(self) -> Dict[str, Union[str, Sequence[str]]]:
        return self._mapping

    @mapping.setter
    def mapping(self, value: Dict[str, Union[str, Sequence[str]]]):
        self._mapping = value

    @property
    def dataset(self) -> "Dataset":
        return self._dataset

    @dataset.setter
    def dataset(self, dataset: "Dataset"):
        self._dataset = dataset
        self._client = dataset._client

    @cached_property
    def schema(self) -> dict:
        schema_dict = {}

        for field in self.fields:
            schema_dict[field.name] = field

        for question in self.questions:
            schema_dict[question.name] = question

        for vector in self.vectors:
            schema_dict[vector.name] = vector

        for metadata in self.metadata:
            schema_dict[metadata.name] = metadata

        return schema_dict

    @cached_property
    def schema_by_id(self) -> Dict[UUID, Union[Field, QuestionType, MetadataType, VectorField]]:
        return {v.id: v for v in self.schema.values()}

    def validate(self) -> None:
        self._validate_empty_settings()
        self._validate_duplicate_names()

        for field in self.fields:
            field.validate()

    #####################
    #  Public methods   #
    #####################

    def get(self) -> "Settings":
        self.fields = self._fetch_fields()
        self.questions = self._fetch_questions()
        self.vectors = self._fetch_vectors()
        self.metadata = self._fetch_metadata()
        self.__fetch_dataset_related_attributes()

        self._update_last_api_call()
        return self

    def create(self) -> "Settings":
        self.validate()

        self._update_dataset_related_attributes()
        self.__fields._create()
        self.__questions._create()
        self.__vectors._create()
        self.__metadata._create()

        self._update_last_api_call()
        return self

    def update(self) -> "Resource":
        self.validate()

        self._update_dataset_related_attributes()
        self.__fields._update()
        self.__questions._update()
        self.__vectors._update()
        self.__metadata._update()
        self.__questions._update()

        self._update_last_api_call()
        return self

    def serialize(self):
        try:
            return {
                "guidelines": self.guidelines,
                "questions": self.__questions.serialize(),
                "fields": self.__fields.serialize(),
                "vectors": self.vectors.serialize(),
                "metadata": self.metadata.serialize(),
                "allow_extra_metadata": self.allow_extra_metadata,
                "distribution": self.distribution.to_dict(),
                "mapping": self.mapping,
            }
        except Exception as e:
            raise ArgillaSerializeError(f"Failed to serialize the settings. {e.__class__.__name__}") from e

    def to_json(self, path: Union[Path, str]) -> None:
        """Save the settings to a file on disk

        Parameters:
            path (str): The path to save the settings to
        """
        if not isinstance(path, Path):
            path = Path(path)
        if path.exists():
            raise FileExistsError(f"File {path} already exists")
        with open(path, "w") as file:
            json.dump(self.serialize(), file)

    @classmethod
    def from_json(cls, path: Union[Path, str]) -> "Settings":
        """Load the settings from a file on disk"""

        with open(path, "r") as file:
            settings_dict = json.load(file)
            return cls._from_dict(settings_dict)

    @classmethod
    def from_hub(
        cls,
        repo_id: str,
        subset: Optional[str] = None,
        feature_mapping: Optional[Dict[str, Literal["question", "field", "metadata"]]] = None,
        **kwargs,
    ) -> "Settings":
        """Load the settings from the Hub

        Parameters:
            repo_id (str): The ID of the repository to load the settings from on the Hub.
            subset (Optional[str]): The subset of the repository to load the settings from.
            feature_mapping (Dict[str, Literal["question", "field", "metadata"]]): A dictionary that maps incoming column names to Argilla attributes.
        """

        settings = build_settings_from_repo_id(repo_id=repo_id, feature_mapping=feature_mapping, subset=subset)
        return settings

    def __eq__(self, other: "Settings") -> bool:
        if not (other and isinstance(other, Settings)):
            return False
        return self.serialize() == other.serialize()  # TODO: Create proper __eq__ methods for fields and questions

    def add(
        self, property: Union[Field, VectorField, MetadataType, QuestionType], override: bool = True
    ) -> Union[Field, VectorField, MetadataType, QuestionType]:
        """
        Add a property to the settings

        Args:
            property: The property to add
            override: If True, override the existing property with the same name. Otherwise, raise an error.  Defaults to True.

        Returns:
            The added property

        """
        # review all settings properties and remove any existing property with the same name
        for attributes in [self.fields, self.questions, self.vectors, self.metadata]:
            for prop in attributes:
                if prop.name == property.name:
                    message = f"Property with name {property.name!r} already exists in settings as {prop.__class__.__name__!r}"
                    if override:
                        warnings.warn(message + ". Overriding the existing property.")
                        attributes.remove(prop)
                    else:
                        raise SettingsError(message)

        if isinstance(property, FieldBase):
            self.fields.add(property)
        elif isinstance(property, QuestionBase):
            self.questions.add(property)
        elif isinstance(property, VectorField):
            self.vectors.add(property)
        elif isinstance(property, MetadataPropertyBase):
            self.metadata.add(property)
        else:
            raise ValueError(f"Unsupported property type: {type(property).__name__}")
        return property

    #####################
    #  Repr Methods     #
    #####################

    def __repr__(self) -> str:
        return (
            f"Settings(guidelines={self.guidelines}, allow_extra_metadata={self.allow_extra_metadata}, "
            f"distribution={self.distribution}, "
            f"fields={self.fields}, questions={self.questions}, vectors={self.vectors}, metadata={self.metadata})"
        )

    #####################
    #  Private methods  #
    #####################

    @classmethod
    def _from_dict(cls, settings_dict: dict) -> "Settings":
        fields = settings_dict.get("fields", [])
        vectors = settings_dict.get("vectors", [])
        metadata = settings_dict.get("metadata", [])
        guidelines = settings_dict.get("guidelines")
        distribution = settings_dict.get("distribution")
        allow_extra_metadata = settings_dict.get("allow_extra_metadata")
        mapping = settings_dict.get("mapping")

        questions = [_question_from_dict(question) for question in settings_dict.get("questions", [])]
        fields = [_field_from_dict(field) for field in fields]
        vectors = [VectorField.from_dict(vector) for vector in vectors]
        metadata = [MetadataField.from_dict(metadata) for metadata in metadata]

        if distribution:
            distribution = TaskDistribution.from_dict(distribution)

        if mapping:
            mapping = cls._validate_mapping(mapping)

        return cls(
            questions=questions,
            fields=fields,
            vectors=vectors,
            metadata=metadata,
            guidelines=guidelines,
            allow_extra_metadata=allow_extra_metadata,
            distribution=distribution,
            mapping=mapping,
        )

    def _copy(self) -> "Settings":
        instance = self.__class__._from_dict(self.serialize())
        return instance

    def _fetch_fields(self) -> List[Field]:
        models = self._client.api.fields.list(dataset_id=self._dataset.id)
        return [_field_from_model(model) for model in models]

    def _fetch_questions(self) -> List[QuestionType]:
        models = self._client.api.questions.list(dataset_id=self._dataset.id)
        return [question_from_model(model) for model in models]

    def _fetch_vectors(self) -> List[VectorField]:
        models = self.dataset._client.api.vectors.list(self.dataset.id)
        return [VectorField.from_model(model) for model in models]

    def _fetch_metadata(self) -> List[MetadataType]:
        models = self._client.api.metadata.list(dataset_id=self._dataset.id)
        return [MetadataField.from_model(model) for model in models]

    def __fetch_dataset_related_attributes(self):
        # This flow may be a bit weird, but it's the only way to update the dataset related attributes
        # Everything is point that we should have several settings-related endpoints in the API to handle this.
        # POST /api/v1/datasets/{dataset_id}/settings
        # {
        #   "guidelines": ....,
        #   "allow_extra_metadata": ....,
        # }
        # But this is not implemented yet, so we need to update the dataset model directly
        dataset_model = self._client.api.datasets.get(self._dataset.id)

        self.guidelines = dataset_model.guidelines
        self.allow_extra_metadata = dataset_model.allow_extra_metadata

        if dataset_model.distribution:
            self.distribution = TaskDistribution.from_model(dataset_model.distribution)

    def _update_dataset_related_attributes(self):
        # This flow may be a bit weird, but it's the only way to update the dataset related attributes
        # Everything is point that we should have several settings-related endpoints in the API to handle this.
        # POST /api/v1/datasets/{dataset_id}/settings
        # {
        #   "guidelines": ....,
        #   "allow_extra_metadata": ....,
        # }
        # But this is not implemented yet, so we need to update the dataset model directly
        dataset_model = DatasetModel(
            id=self._dataset.id,
            name=self._dataset.name,
            guidelines=self.guidelines,
            allow_extra_metadata=self.allow_extra_metadata,
            distribution=self.distribution._api_model(),
        )
        self._client.api.datasets.update(dataset_model)

    def _validate_empty_settings(self):
        if not all([self.fields, self.questions]):
            message = "Fields and questions are required"
            raise SettingsError(message=message)

    def _validate_duplicate_names(self) -> None:
        dataset_properties_by_name = {}

        for properties in [self.fields, self.questions, self.vectors, self.metadata]:
            for property in properties:
                if property.name in dataset_properties_by_name:
                    raise SettingsError(
                        f"names of dataset settings must be unique, "
                        f"but the name {property.name!r} is used by {type(property).__name__!r} and {type(dataset_properties_by_name[property.name]).__name__!r} "
                    )
                dataset_properties_by_name[property.name] = property

    @classmethod
    def _validate_mapping(cls, mapping: Dict[str, Union[str, Sequence[str]]]) -> dict:
        validate_mapping = {}
        for key, value in mapping.items():
            if isinstance(value, str):
                validate_mapping[key] = value
            elif isinstance(value, list) or isinstance(value, tuple):
                validate_mapping[key] = tuple(value)
            else:
                raise SettingsError(f"Invalid mapping value for key {key!r}: {value}")

        return validate_mapping

    def __process_guidelines(self, guidelines):
        if guidelines is None:
            return guidelines

        if not isinstance(guidelines, str):
            raise SettingsError("Guidelines must be a string or a path to a file")

        if os.path.exists(guidelines):
            with open(guidelines, "r") as file:
                return file.read()

        return guidelines

__init__(fields=None, questions=None, vectors=None, metadata=None, guidelines=None, allow_extra_metadata=False, distribution=None, mapping=None, _dataset=None)

参数

名称	类型	描述	默认值
fields	List[Field]	表示数据集中的字段的 Field 对象列表。	无
questions	List[Union[LabelQuestion, MultiLabelQuestion, RankingQuestion, TextQuestion, RatingQuestion]]	表示数据集中的问题的 Question 对象列表。	无
vectors	List[VectorField]	表示数据集中的向量的 VectorField 对象列表。	无
metadata	List[MetadataField]	表示数据集中的元数据的 MetadataField 对象列表。	无
guidelines	str	包含数据集指南的字符串。	无
allow_extra_metadata	bool	一个布尔值，确定是否允许在数据集中使用额外的元数据。默认为 False。	False
distribution	TaskDistribution	标注任务分配配置。默认为 DEFAULT_TASK_DISTRIBUTION	无
mapping	Dict[str, Union[str, Sequence[str]]]	一个字典，将传入的数据名称映射到 DatasetRecords 中的 Argilla 数据集属性。	无

源代码位于 src/argilla/settings/_resource.py

def __init__(
    self,
    fields: Optional[List[Field]] = None,
    questions: Optional[List[QuestionType]] = None,
    vectors: Optional[List[VectorField]] = None,
    metadata: Optional[List[MetadataType]] = None,
    guidelines: Optional[str] = None,
    allow_extra_metadata: bool = False,
    distribution: Optional[TaskDistribution] = None,
    mapping: Optional[Dict[str, Union[str, Sequence[str]]]] = None,
    _dataset: Optional["Dataset"] = None,
) -> None:
    """
    Args:
        fields (List[Field]): A list of Field objects that represent the fields in the Dataset.
        questions (List[Union[LabelQuestion, MultiLabelQuestion, RankingQuestion, TextQuestion, RatingQuestion]]):
            A list of Question objects that represent the questions in the Dataset.
        vectors (List[VectorField]): A list of VectorField objects that represent the vectors in the Dataset.
        metadata (List[MetadataField]): A list of MetadataField objects that represent the metadata in the Dataset.
        guidelines (str): A string containing the guidelines for the Dataset.
        allow_extra_metadata (bool): A boolean that determines whether or not extra metadata is allowed in the
            Dataset. Defaults to False.
        distribution (TaskDistribution): The annotation task distribution configuration.
            Default to DEFAULT_TASK_DISTRIBUTION
        mapping (Dict[str, Union[str, Sequence[str]]]): A dictionary that maps incoming data names to Argilla dataset attributes in DatasetRecords.
    """
    super().__init__(client=_dataset._client if _dataset else None)

    self._dataset = _dataset
    self._distribution = distribution or TaskDistribution.default()
    self._mapping = mapping
    self.__guidelines = self.__process_guidelines(guidelines)
    self.__allow_extra_metadata = allow_extra_metadata

    self.__questions = SettingsProperties(self, questions)
    self.__fields = SettingsProperties(self, fields)
    self.__vectors = SettingsProperties(self, vectors)
    self.__metadata = SettingsProperties(self, metadata)

to_json(path)

将设置保存到磁盘上的文件

参数

名称	类型	描述	默认值
path	str	要将设置保存到的路径	必需

源代码位于 src/argilla/settings/_resource.py

def to_json(self, path: Union[Path, str]) -> None:
    """Save the settings to a file on disk

    Parameters:
        path (str): The path to save the settings to
    """
    if not isinstance(path, Path):
        path = Path(path)
    if path.exists():
        raise FileExistsError(f"File {path} already exists")
    with open(path, "w") as file:
        json.dump(self.serialize(), file)

from_json(path) classmethod

从磁盘上的文件加载设置

源代码位于 src/argilla/settings/_resource.py

@classmethod
def from_json(cls, path: Union[Path, str]) -> "Settings":
    """Load the settings from a file on disk"""

    with open(path, "r") as file:
        settings_dict = json.load(file)
        return cls._from_dict(settings_dict)

from_hub(repo_id, subset=None, feature_mapping=None, **kwargs) classmethod

从 Hub 加载设置

参数

名称	类型	描述	默认值
repo_id	str	要从 Hub 加载设置的仓库 ID。	必需
subset	Optional[str]	要从中加载设置的仓库子集。	无
feature_mapping	Dict[str, Literal['question', 'field', 'metadata']]	一个字典，将传入的列名称映射到 Argilla 属性。	无

源代码位于 src/argilla/settings/_resource.py

@classmethod
def from_hub(
    cls,
    repo_id: str,
    subset: Optional[str] = None,
    feature_mapping: Optional[Dict[str, Literal["question", "field", "metadata"]]] = None,
    **kwargs,
) -> "Settings":
    """Load the settings from the Hub

    Parameters:
        repo_id (str): The ID of the repository to load the settings from on the Hub.
        subset (Optional[str]): The subset of the repository to load the settings from.
        feature_mapping (Dict[str, Literal["question", "field", "metadata"]]): A dictionary that maps incoming column names to Argilla attributes.
    """

    settings = build_settings_from_repo_id(repo_id=repo_id, feature_mapping=feature_mapping, subset=subset)
    return settings

add(property, override=True)

向设置添加属性

参数

名称	类型	描述	默认值
property	Union[Field, VectorField, MetadataType, QuestionType]	要添加的属性	必需
override	bool	如果为 True，则使用相同的名称覆盖现有属性。否则，引发错误。默认为 True。	True

返回

类型	描述
Union[Field, VectorField, MetadataType, QuestionType]	添加的属性

源代码位于 src/argilla/settings/_resource.py

def add(
    self, property: Union[Field, VectorField, MetadataType, QuestionType], override: bool = True
) -> Union[Field, VectorField, MetadataType, QuestionType]:
    """
    Add a property to the settings

    Args:
        property: The property to add
        override: If True, override the existing property with the same name. Otherwise, raise an error.  Defaults to True.

    Returns:
        The added property

    """
    # review all settings properties and remove any existing property with the same name
    for attributes in [self.fields, self.questions, self.vectors, self.metadata]:
        for prop in attributes:
            if prop.name == property.name:
                message = f"Property with name {property.name!r} already exists in settings as {prop.__class__.__name__!r}"
                if override:
                    warnings.warn(message + ". Overriding the existing property.")
                    attributes.remove(prop)
                else:
                    raise SettingsError(message)

    if isinstance(property, FieldBase):
        self.fields.add(property)
    elif isinstance(property, QuestionBase):
        self.questions.add(property)
    elif isinstance(property, VectorField):
        self.vectors.add(property)
    elif isinstance(property, MetadataPropertyBase):
        self.metadata.add(property)
    else:
        raise ValueError(f"Unsupported property type: {type(property).__name__}")
    return property