rg.Record

Record 对象用于表示 Argilla 中的单个记录。它包含字段、建议、响应、元数据和向量。

使用示例

创建记录

要创建记录，您可以使用 Record 类并将其传递给 Dataset.records.log 方法。Record 类需要一个 fields 参数，它是一个字段名称和值的字典。字段名称必须与数据集的 Settings 对象中的字段名称匹配才能被接受。

dataset.records.log(
    records=[
        rg.Record(
            fields={"text": "Hello World, how are you?"},
        ),
    ]
) # (1)

1. Argilla 数据集包含一个名为 text 的字段，与此处的键匹配。

要创建带有图像字段的记录，请将图像作为远程 url、本地图像文件路径或 PIL 对象传递给记录对象。字段名称必须在数据集的 Settings 对象中定义为 rg.ImageField 才能被接受。图像将存储在 Argilla 数据库中，并作为重新缩放的 PIL 对象返回。

dataset.records.log(
    records=[
        rg.Record(
            fields={"image": "https://example.com/image.jpg"}, # (1)
        ),
    ]
)

1. 图像可以作为远程 url、本地文件路径或 PIL 对象引用。

注意

图像将存储在 Argilla 数据库中，并可能影响数据集的存储使用量。图像应小于 5mb，数据集应包含少于 10,000 张图像。

访问记录属性

Record 对象具有 suggestions、responses、metadata 和 vectors 属性，可以在数据集中迭代记录时直接访问。

for record in dataset.records(
    with_suggestions=True,
    with_responses=True,
    with_metadata=True,
    with_vectors=True
    ):
    print(record.suggestions)
    print(record.responses)
    print(record.metadata)
    print(record.vectors)

记录属性也可以在数据集中迭代记录时更新。

for record in dataset.records(with_metadata=True):
    record.metadata = {"department": "toys"}

为了使更改生效，用户必须在 Dataset 对象上调用 update 方法，或者将更新后的记录传递给 Dataset.records.log。所有核心记录属性都可以通过这种方式更新。查看它们各自的文档以获取更多信息：Suggestions，Responses，Metadata，Vectors。

---

Record

基类：Resource

用于与 Argilla 记录交互的类。Record 是数据集中的单个样本。记录接收响应和建议形式的反馈。记录包含字段、元数据和向量。

属性

名称	类型	描述
id	Union[str, UUID]	记录的 ID。
fields	RecordFields	记录的字段。
metadata	RecordMetadata	记录的元数据。
vectors	RecordVectors	记录的向量。
responses	RecordResponses	记录的响应。
suggestions	RecordSuggestions	记录的建议。
dataset	Dataset	记录所属的数据集。
_server_id	UUID	Argilla 服务器为记录生成的 ID。

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

class Record(Resource):
    """The class for interacting with Argilla Records. A `Record` is a single sample
    in a dataset. Records receives feedback in the form of responses and suggestions.
    Records contain fields, metadata, and vectors.

    Attributes:
        id (Union[str, UUID]): The id of the record.
        fields (RecordFields): The fields of the record.
        metadata (RecordMetadata): The metadata of the record.
        vectors (RecordVectors): The vectors of the record.
        responses (RecordResponses): The responses of the record.
        suggestions (RecordSuggestions): The suggestions of the record.
        dataset (Dataset): The dataset to which the record belongs.
        _server_id (UUID): An id for the record generated by the Argilla server.
    """

    _model: RecordModel

    def __init__(
        self,
        id: Optional[Union[UUID, str]] = None,
        fields: Optional[Dict[str, FieldValue]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        vectors: Optional[Dict[str, VectorValue]] = None,
        responses: Optional[List[Response]] = None,
        suggestions: Optional[List[Suggestion]] = None,
        _server_id: Optional[UUID] = None,
        _dataset: Optional["Dataset"] = None,
    ):
        """Initializes a Record with fields, metadata, vectors, responses, suggestions, external_id, and id.
        Records are typically defined as flat dictionary objects with fields, metadata, vectors, responses, and suggestions
        and passed to Dataset.DatasetRecords.add() as a list of dictionaries.

        Args:
            id: An id for the record. If not provided, a UUID will be generated.
            fields: A dictionary of fields for the record.
            metadata: A dictionary of metadata for the record.
            vectors: A dictionary of vectors for the record.
            responses: A list of Response objects for the record.
            suggestions: A list of Suggestion objects for the record.
            _server_id: An id for the record. (Read-only and set by the server)
            _dataset: The dataset object to which the record belongs.
        """

        if fields is None and metadata is None and vectors is None and responses is None and suggestions is None:
            raise ValueError("At least one of fields, metadata, vectors, responses, or suggestions must be provided.")
        if fields is None and id is None:
            raise ValueError("If fields are not provided, an id must be provided.")
        if fields == {} and id is None:
            raise ValueError("If fields are an empty dictionary, an id must be provided.")

        self._dataset = _dataset
        self._model = RecordModel(external_id=id, id=_server_id)
        self.__fields = RecordFields(fields=fields, record=self)
        self.__vectors = RecordVectors(vectors=vectors)
        self.__metadata = RecordMetadata(metadata=metadata)
        self.__responses = RecordResponses(responses=responses, record=self)
        self.__suggestions = RecordSuggestions(suggestions=suggestions, record=self)

    def __repr__(self) -> str:
        return (
            f"Record(id={self.id},status={self.status},fields={self.fields},metadata={self.metadata},"
            f"suggestions={self.suggestions},responses={self.responses})"
        )

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

    @property
    def id(self) -> str:
        return self._model.external_id

    @id.setter
    def id(self, value: str) -> None:
        self._model.external_id = value

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

    @dataset.setter
    def dataset(self, value: "Dataset") -> None:
        self._dataset = value

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

    @property
    def responses(self) -> "RecordResponses":
        return self.__responses

    @property
    def suggestions(self) -> "RecordSuggestions":
        return self.__suggestions

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

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

    @property
    def status(self) -> str:
        return self._model.status

    @property
    def _server_id(self) -> Optional[UUID]:
        return self._model.id

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

    def get(self) -> "Record":
        """Retrieves the record from the server."""
        model = self._client.api.records.get(self._server_id)
        instance = self.from_model(model, dataset=self.dataset)
        self.__dict__ = instance.__dict__

        return self

    def api_model(self) -> RecordModel:
        return RecordModel(
            id=self._model.id,
            external_id=self._model.external_id,
            fields=self.fields.to_dict(),
            metadata=self.metadata.api_models(),
            vectors=self.vectors.api_models(),
            responses=self.responses.api_models(),
            suggestions=self.suggestions.api_models(),
            status=self.status,
        )

    def serialize(self) -> Dict[str, Any]:
        """Serializes the Record to a dictionary for interaction with the API"""
        serialized_model = self._model.model_dump()
        serialized_suggestions = [suggestion.serialize() for suggestion in self.__suggestions]
        serialized_responses = [response.serialize() for response in self.__responses]
        serialized_model["responses"] = serialized_responses
        serialized_model["suggestions"] = serialized_suggestions

        return serialized_model

    def to_dict(self) -> Dict[str, Dict]:
        """Converts a Record object to a dictionary for export.
        Returns:
            A dictionary representing the record where the keys are "fields",
            "metadata", "suggestions", and "responses". Each field and question is
            represented as a key-value pair in the dictionary of the respective key. i.e.
            `{"fields": {"prompt": "...", "response": "..."}, "responses": {"rating": "..."},
        """
        id = str(self.id) if self.id else None
        server_id = str(self._model.id) if self._model.id else None
        status = self.status
        fields = self.fields.to_dict()
        metadata = self.metadata.to_dict()
        suggestions = self.suggestions.to_dict()
        responses = self.responses.to_dict()
        vectors = self.vectors.to_dict()

        # TODO: Review model attributes when to_dict and serialize methods are unified
        return {
            "id": id,
            "fields": fields,
            "metadata": metadata,
            "suggestions": suggestions,
            "responses": responses,
            "vectors": vectors,
            "status": status,
            "_server_id": server_id,
        }

    @classmethod
    def from_dict(cls, data: Dict[str, Dict], dataset: Optional["Dataset"] = None) -> "Record":
        """Converts a dictionary to a Record object.
        Args:
            data: A dictionary representing the record.
            dataset: The dataset object to which the record belongs.
        Returns:
            A Record object.
        """
        fields = data.get("fields", {})
        metadata = data.get("metadata", {})
        suggestions = data.get("suggestions", {})
        responses = data.get("responses", {})
        vectors = data.get("vectors", {})
        record_id = data.get("id", None)
        _server_id = data.get("_server_id", None)

        suggestions = [Suggestion(question_name=question_name, **value) for question_name, value in suggestions.items()]
        responses = [
            Response(question_name=question_name, **value)
            for question_name, _responses in responses.items()
            for value in _responses
        ]

        return cls(
            id=record_id,
            fields=fields,
            suggestions=suggestions,
            responses=responses,
            vectors=vectors,
            metadata=metadata,
            _dataset=dataset,
            _server_id=_server_id,
        )

    @classmethod
    def from_model(cls, model: RecordModel, dataset: "Dataset") -> "Record":
        """Converts a RecordModel object to a Record object.
        Args:
            model: A RecordModel object.
            dataset: The dataset object to which the record belongs.
        Returns:
            A Record object.
        """
        instance = cls(
            id=model.external_id,
            fields=model.fields,
            metadata={meta.name: meta.value for meta in model.metadata},
            vectors={vector.name: vector.vector_values for vector in model.vectors},
            _dataset=dataset,
            responses=[],
            suggestions=[],
        )

        # set private attributes
        instance._dataset = dataset
        instance._model = model

        # Responses and suggestions are computed separately based on the record model
        instance.responses.from_models(model.responses)
        instance.suggestions.from_models(model.suggestions)

        return instance

    @property
    def _client(self) -> Optional["Argilla"]:
        if self._dataset:
            return self.dataset._client

    @property
    def _api(self) -> Optional["RecordsAPI"]:
        if self._client:
            return self._client.api.records

__init__(id=None, fields=None, metadata=None, vectors=None, responses=None, suggestions=None, _server_id=None, _dataset=None)

使用字段、元数据、向量、响应、建议、external_id 和 id 初始化 Record。记录通常定义为扁平字典对象，包含字段、元数据、向量、响应和建议，并作为字典列表传递给 Dataset.DatasetRecords.add()。

参数

名称	类型	描述	默认值
id	Optional[Union[UUID, str]]	记录的 ID。如果未提供，将生成 UUID。	无
fields	Optional[Dict[str, FieldValue]]	记录的字段字典。	无
metadata	Optional[Dict[str, Any]]	记录的元数据字典。	无
vectors	Optional[Dict[str, VectorValue]]	记录的向量字典。	无
responses	Optional[List[Response]]	记录的 Response 对象列表。	无
suggestions	Optional[List[Suggestion]]	记录的 Suggestion 对象列表。	无
_server_id	Optional[UUID]	记录的 ID。（只读，由服务器设置）	无
_dataset	Optional[Dataset]	记录所属的数据集对象。	无

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

def __init__(
    self,
    id: Optional[Union[UUID, str]] = None,
    fields: Optional[Dict[str, FieldValue]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    vectors: Optional[Dict[str, VectorValue]] = None,
    responses: Optional[List[Response]] = None,
    suggestions: Optional[List[Suggestion]] = None,
    _server_id: Optional[UUID] = None,
    _dataset: Optional["Dataset"] = None,
):
    """Initializes a Record with fields, metadata, vectors, responses, suggestions, external_id, and id.
    Records are typically defined as flat dictionary objects with fields, metadata, vectors, responses, and suggestions
    and passed to Dataset.DatasetRecords.add() as a list of dictionaries.

    Args:
        id: An id for the record. If not provided, a UUID will be generated.
        fields: A dictionary of fields for the record.
        metadata: A dictionary of metadata for the record.
        vectors: A dictionary of vectors for the record.
        responses: A list of Response objects for the record.
        suggestions: A list of Suggestion objects for the record.
        _server_id: An id for the record. (Read-only and set by the server)
        _dataset: The dataset object to which the record belongs.
    """

    if fields is None and metadata is None and vectors is None and responses is None and suggestions is None:
        raise ValueError("At least one of fields, metadata, vectors, responses, or suggestions must be provided.")
    if fields is None and id is None:
        raise ValueError("If fields are not provided, an id must be provided.")
    if fields == {} and id is None:
        raise ValueError("If fields are an empty dictionary, an id must be provided.")

    self._dataset = _dataset
    self._model = RecordModel(external_id=id, id=_server_id)
    self.__fields = RecordFields(fields=fields, record=self)
    self.__vectors = RecordVectors(vectors=vectors)
    self.__metadata = RecordMetadata(metadata=metadata)
    self.__responses = RecordResponses(responses=responses, record=self)
    self.__suggestions = RecordSuggestions(suggestions=suggestions, record=self)

get()

从服务器检索记录。

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

def get(self) -> "Record":
    """Retrieves the record from the server."""
    model = self._client.api.records.get(self._server_id)
    instance = self.from_model(model, dataset=self.dataset)
    self.__dict__ = instance.__dict__

    return self

serialize()

将 Record 序列化为字典，以便与 API 交互

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

def serialize(self) -> Dict[str, Any]:
    """Serializes the Record to a dictionary for interaction with the API"""
    serialized_model = self._model.model_dump()
    serialized_suggestions = [suggestion.serialize() for suggestion in self.__suggestions]
    serialized_responses = [response.serialize() for response in self.__responses]
    serialized_model["responses"] = serialized_responses
    serialized_model["suggestions"] = serialized_suggestions

    return serialized_model

to_dict()

将 Record 对象转换为字典以进行导出。 返回：表示记录的字典，其中键为“fields”、“metadata”、“suggestions”和“responses”。每个字段和问题在相应键的字典中表示为键值对。 例如：`{"fields": {"prompt": "...", "response": "..."}, "responses": {"rating": "..."},

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

def to_dict(self) -> Dict[str, Dict]:
    """Converts a Record object to a dictionary for export.
    Returns:
        A dictionary representing the record where the keys are "fields",
        "metadata", "suggestions", and "responses". Each field and question is
        represented as a key-value pair in the dictionary of the respective key. i.e.
        `{"fields": {"prompt": "...", "response": "..."}, "responses": {"rating": "..."},
    """
    id = str(self.id) if self.id else None
    server_id = str(self._model.id) if self._model.id else None
    status = self.status
    fields = self.fields.to_dict()
    metadata = self.metadata.to_dict()
    suggestions = self.suggestions.to_dict()
    responses = self.responses.to_dict()
    vectors = self.vectors.to_dict()

    # TODO: Review model attributes when to_dict and serialize methods are unified
    return {
        "id": id,
        "fields": fields,
        "metadata": metadata,
        "suggestions": suggestions,
        "responses": responses,
        "vectors": vectors,
        "status": status,
        "_server_id": server_id,
    }

from_dict(data, dataset=None) classmethod

将字典转换为 Record 对象。 Args: data: 表示记录的字典。 dataset: 记录所属的数据集对象。 返回：Record 对象。

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

@classmethod
def from_dict(cls, data: Dict[str, Dict], dataset: Optional["Dataset"] = None) -> "Record":
    """Converts a dictionary to a Record object.
    Args:
        data: A dictionary representing the record.
        dataset: The dataset object to which the record belongs.
    Returns:
        A Record object.
    """
    fields = data.get("fields", {})
    metadata = data.get("metadata", {})
    suggestions = data.get("suggestions", {})
    responses = data.get("responses", {})
    vectors = data.get("vectors", {})
    record_id = data.get("id", None)
    _server_id = data.get("_server_id", None)

    suggestions = [Suggestion(question_name=question_name, **value) for question_name, value in suggestions.items()]
    responses = [
        Response(question_name=question_name, **value)
        for question_name, _responses in responses.items()
        for value in _responses
    ]

    return cls(
        id=record_id,
        fields=fields,
        suggestions=suggestions,
        responses=responses,
        vectors=vectors,
        metadata=metadata,
        _dataset=dataset,
        _server_id=_server_id,
    )

from_model(model, dataset) classmethod

将 RecordModel 对象转换为 Record 对象。 Args: model: RecordModel 对象。 dataset: 记录所属的数据集对象。 返回：Record 对象。

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

@classmethod
def from_model(cls, model: RecordModel, dataset: "Dataset") -> "Record":
    """Converts a RecordModel object to a Record object.
    Args:
        model: A RecordModel object.
        dataset: The dataset object to which the record belongs.
    Returns:
        A Record object.
    """
    instance = cls(
        id=model.external_id,
        fields=model.fields,
        metadata={meta.name: meta.value for meta in model.metadata},
        vectors={vector.name: vector.vector_values for vector in model.vectors},
        _dataset=dataset,
        responses=[],
        suggestions=[],
    )

    # set private attributes
    instance._dataset = dataset
    instance._model = model

    # Responses and suggestions are computed separately based on the record model
    instance.responses.from_models(model.responses)
    instance.suggestions.from_models(model.suggestions)

    return instance