Method serialize_many

trinity/common/experience.py:346–402

Serialize a list of experiences into a compact bytes payload. Tensor fields are packed with safetensors while non-tensor fields are packed as metadata via pickle.

(cls, experiences: List[Experience])

    @classmethod
    def serialize_many(cls, experiences: List[Experience]) -> bytes:
        """Serialize a list of experiences into a compact bytes payload.

        Tensor fields are packed with safetensors while non-tensor fields are packed
        as metadata via pickle.
        """
        metadata = {"version": cls._SER_VERSION, "num_items": len(experiences), "items": []}
        tensor_data = {}
        seen_storages: set = set()

        for index, exp in enumerate(experiences):
            item_meta = {}
            for field_name in cls._META_FIELDS:
                value = getattr(exp, field_name)
                if field_name == "eid" and value is not None:
                    item_meta[field_name] = value.to_dict() if isinstance(value, EID) else value
                else:
                    item_meta[field_name] = value

            item_meta["custom_fields"] = cls._serialize_custom_fields(exp.custom_fields)

            for field_name in cls._TENSOR_FIELDS:
                value = getattr(exp, field_name)
                if value is None:
                    continue
                t = value.detach().contiguous().cpu()
                storage_ptr = t.untyped_storage().data_ptr()
                if storage_ptr in seen_storages:
                    t = t.clone()
                seen_storages.add(storage_ptr)
                tensor_data[f"{index}:{field_name}"] = t

            if exp.multi_modal_inputs is None:
                item_meta["multi_modal_input_keys"] = []
            else:
                mm_keys = list(exp.multi_modal_inputs.keys())
                item_meta["multi_modal_input_keys"] = mm_keys
                for key in mm_keys:
                    value = exp.multi_modal_inputs[key]
                    t = value.detach().contiguous().cpu()
                    storage_ptr = t.untyped_storage().data_ptr()
                    if storage_ptr in seen_storages:
                        t = t.clone()
                    seen_storages.add(storage_ptr)
                    tensor_data[f"{index}:multi_modal_inputs:{key}"] = t

            metadata["items"].append(item_meta)

        metadata_bytes = pickle.dumps(metadata, protocol=pickle.HIGHEST_PROTOCOL)
        tensor_bytes = st_save(tensor_data)
        header = (
            cls._SER_MAGIC
            + struct.pack("<B", cls._SER_VERSION)
            + struct.pack("<Q", len(metadata_bytes))
            + struct.pack("<Q", len(tensor_bytes))
        )
        return header + metadata_bytes + tensor_bytes