From c684b655c30932b0f0fe54ba8155826a388f480a Mon Sep 17 00:00:00 2001 From: Alexander Kuzmik Date: Wed, 2 Sep 2026 11:39:45 +0200 Subject: [PATCH] [NA] [SDK] feat: allow dataset item insertion without deduplication Add a `deduplication` flag to every Python SDK method that writes dataset items. With `deduplication=False` the whole dedup path is bypassed: the existing items are not downloaded from the backend and no content hashes are computed or compared, which is significantly faster on large datasets. Also raise the default `num_threads` for `Dataset.insert` from 1 to 4, and cache the backend-version probe that gates parallel upload so a loop of inserts does not pay a round trip per call. Co-Authored-By: Claude Opus 5 (1M context) --- .../evaluation/advanced/manage_datasets.mdx | 22 ++- .../src/opik/api_objects/dataset/dataset.py | 104 +++++++----- .../dataset/test_suite/test_suite.py | 38 ++++- .../dataset/test_dataset_client.py | 153 +++++++++++++++++- .../dataset/test_suite/test_test_suite.py | 16 ++ 5 files changed, 284 insertions(+), 49 deletions(-) diff --git a/apps/opik-documentation/documentation/fern/docs-v2/evaluation/advanced/manage_datasets.mdx b/apps/opik-documentation/documentation/fern/docs-v2/evaluation/advanced/manage_datasets.mdx index 368d0368a67..cf8bce10a4d 100644 --- a/apps/opik-documentation/documentation/fern/docs-v2/evaluation/advanced/manage_datasets.mdx +++ b/apps/opik-documentation/documentation/fern/docs-v2/evaluation/advanced/manage_datasets.mdx @@ -277,12 +277,32 @@ dataset.insert([ Opik automatically deduplicates items that are inserted into a dataset when using the Python SDK. This means that you can insert the same item multiple times without duplicating it in the dataset. This combined with the `get or create - dataset` methods means that you can use the SDK to manage your datasets in a "fire and forget" manner. + dataset` methods means that you can use the SDK to manage your datasets in a "fire and forget" manner. It can be + turned off with `deduplication=False`, see [Disabling deduplication](#disabling-deduplication). +#### Disabling deduplication + +Deduplication requires the Python SDK to download the dataset's existing items once so it can compare their +content hashes against the items you are inserting. On large datasets that download dominates the insert. If you +already know your items are unique — for example when populating a fresh dataset, or when you generate ids +yourself — pass `deduplication=False` to skip that work entirely: nothing is downloaded, no hashes are computed, +and every item you pass is sent as-is. + +```python title="Python" language="python" +dataset.insert(items, deduplication=False) +``` + +The flag is available on every Python SDK method that writes items — `insert`, `update`, `insert_from_json`, +`insert_from_pandas` and `read_jsonl_from_file` — as well as on the equivalent `TestSuite` methods. With +deduplication disabled, inserting the same content twice produces two separate dataset items. + When using the SDK to insert items, a new dataset version is automatically created. If you insert items in multiple batches within a single `insert()` call, they are grouped into one version. + The Python SDK uploads those batches on 4 worker threads by default; use `num_threads=1` to upload them + sequentially instead. Parallel upload requires a recent Opik backend — against older ones the SDK falls back + to a sequential upload and logs a warning. Once the items have been inserted, you can view them in the Opik UI: diff --git a/sdks/python/src/opik/api_objects/dataset/dataset.py b/sdks/python/src/opik/api_objects/dataset/dataset.py index ccff2e9658f..db238ec5906 100644 --- a/sdks/python/src/opik/api_objects/dataset/dataset.py +++ b/sdks/python/src/opik/api_objects/dataset/dataset.py @@ -611,14 +611,18 @@ def _insert_batch_with_retry( ) LOGGER.debug("Successfully sent dataset items batch of size %d", len(batch)) - def _resolve_num_threads(self, num_threads: int) -> int: - """Downgrade to a sequential upload when the backend predates parallel support. + @functools.cached_property + def _parallel_insert_supported(self) -> bool: + """Whether the backend tolerates concurrent batches sharing a batch_group_id. - Older backends race on concurrent batches that share a batch_group_id, - so parallelism is only safe from + Older backends race on them, so parallelism is only safe from ``constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT`` onwards. When the version cannot be determined at all — unreachable endpoint, non-semver - build string — we fall back to sequential rather than risk the race. + build string — we report unsupported rather than risk the race. + + Cached because the backend cannot change version mid-session, and + parallel upload is the default: probing per ``insert`` would add a + round trip to every call in a loop. """ try: backend_version = self._rest_client.version()["version"] @@ -634,7 +638,7 @@ def _resolve_num_threads(self, num_threads: int) -> int: constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT, exc_info=True, ) - return 1 + return False if not supported: LOGGER.warning( @@ -644,9 +648,8 @@ def _resolve_num_threads(self, num_threads: int) -> int: backend_version, constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT, ) - return 1 - return num_threads + return supported def _send_batches( self, @@ -681,11 +684,10 @@ def _send_batches( for future in futures.as_completed(submitted): future.result() - def __internal_api__insert_items_as_dataclasses__( - self, - items: List[dataset_item.DatasetItem], - num_threads: int = 1, - ) -> None: + def _deduplicate( + self, items: List[dataset_item.DatasetItem] + ) -> List[dataset_item.DatasetItem]: + """Drop items whose content hash was already seen locally or on the backend.""" # Lazy-sync against the backend the first time we insert into a # dataset that was fetched from the backend (list or get-by-name # factory), so content-hash dedup still works without paying an @@ -693,7 +695,6 @@ def __internal_api__insert_items_as_dataclasses__( if not self._hashes_synced: self.__internal_api__sync_hashes__() - # Remove duplicates if they already exist deduplicated_items: List[dataset_item.DatasetItem] = [] for item in items: item_hash = item.content_hash() @@ -709,7 +710,23 @@ def __internal_api__insert_items_as_dataclasses__( self._hashes.add(item_hash) self._id_to_hash[item.id] = item_hash - rest_items = [self._convert_to_rest_item(item) for item in deduplicated_items] + return deduplicated_items + + def __internal_api__insert_items_as_dataclasses__( + self, + items: List[dataset_item.DatasetItem], + num_threads: int = 1, + deduplication: bool = True, + ) -> None: + if deduplication: + items_to_send = self._deduplicate(items) + else: + # Nothing was hashed, so the local cache no longer describes the + # backend; force a re-sync before the next deduplicated insert. + items_to_send = items + self._hashes_synced = False + + rest_items = [self._convert_to_rest_item(item) for item in items_to_send] batches = sequence_splitter.split_into_batches( rest_items, @@ -724,42 +741,44 @@ def __internal_api__insert_items_as_dataclasses__( # Invalidate the cached count so it will be fetched from backend on next access self._dataset_items_count = None - def insert(self, items: Sequence[Dict[str, Any]], num_threads: int = 1) -> None: + def insert( + self, + items: Sequence[Dict[str, Any]], + num_threads: int = 4, + deduplication: bool = True, + ) -> None: """ Insert new items into the dataset. A new dataset version will be created. Args: items: List of dicts (which will be converted to dataset items) to add to the dataset. + deduplication: Whether to skip items whose content already exists + in the dataset. Pass ``False`` to insert every item as-is + without any duplicate checking, which is significantly faster + on large datasets. num_threads: Number of worker threads used to upload the item - batches. Must be a positive integer. With ``1`` (the default) - batches are uploaded sequentially. With more than ``1`` the - batches of this single ``insert`` are uploaded in parallel; - they all land in one dataset version. If any batch fails the - call re-raises; there is no rollback, so batches that already - succeeded remain persisted. Items are keyed by their ``id``, so - parallel and sequential inserts of the same items produce - identical dataset content; the input order is not a read-back - guarantee. Requires an Opik backend of at least - ``constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT``; against - older backends, or when the backend version cannot be - determined, the upload falls back to sequential and logs a - warning. + batches. Must be a positive integer, defaults to ``4``; pass + ``1`` to upload sequentially. All batches land in a single + dataset version. If a batch fails the call raises, and the + batches that already succeeded stay persisted. Older Opik + backends do not support parallel upload and fall back to a + sequential one. """ if isinstance(num_threads, bool) or not isinstance(num_threads, int): raise ValueError("num_threads must be a positive integer") if num_threads < 1: raise ValueError("num_threads must be a positive integer") - if num_threads > 1: - num_threads = self._resolve_num_threads(num_threads) + if num_threads > 1 and not self._parallel_insert_supported: + num_threads = 1 dataset_items: List[dataset_item.DatasetItem] = [ # type: ignore (dataset_item.DatasetItem(**item) if isinstance(item, dict) else item) for item in items ] self.__internal_api__insert_items_as_dataclasses__( - dataset_items, num_threads=num_threads + dataset_items, num_threads=num_threads, deduplication=deduplication ) @property @@ -794,12 +813,14 @@ def __internal_api__sync_hashes__(self) -> None: self._hashes_synced = True LOGGER.debug("Finish hash sync in dataset") - def update(self, items: List[Dict[str, Any]]) -> None: + def update(self, items: List[Dict[str, Any]], deduplication: bool = True) -> None: """ Update existing items in the dataset. Args: items: List of DatasetItem objects to update in the dataset. You need to provide the full item object as it will override what has been supplied previously. + deduplication: Whether to skip items whose content already exists in + the dataset. See :meth:`insert` for details. Raises: DatasetItemUpdateOperationRequiresItemId: If any item in the list is missing an id. @@ -810,7 +831,7 @@ def update(self, items: List[Dict[str, Any]]) -> None: "Missing id for dataset item to update: %s", item ) - self.insert(items) + self.insert(items, deduplication=deduplication) def _delete_batch_with_retry( self, @@ -912,6 +933,7 @@ def insert_from_json( json_array: str, keys_mapping: Optional[Dict[str, str]] = None, ignore_keys: Optional[List[str]] = None, + deduplication: bool = True, ) -> None: """ Args: @@ -921,6 +943,8 @@ def insert_from_json( Example: {'Expected output': 'expected_output'} ignore_keys: if your json dicts contain keys that are not needed for DatasetItem construction - pass them as ignore_keys argument + deduplication: Whether to skip items whose content already exists in + the dataset. See :meth:`insert` for details. """ keys_mapping = {} if keys_mapping is None else keys_mapping ignore_keys = [] if ignore_keys is None else ignore_keys @@ -929,13 +953,14 @@ def insert_from_json( json_array, keys_mapping=keys_mapping, ignore_keys=ignore_keys ) - self.insert(new_items) + self.insert(new_items, deduplication=deduplication) def read_jsonl_from_file( self, file_path: str, keys_mapping: Optional[Dict[str, str]] = None, ignore_keys: Optional[List[str]] = None, + deduplication: bool = True, ) -> None: """ Read JSONL from a file and insert it into the dataset. @@ -946,17 +971,20 @@ def read_jsonl_from_file( Example: {'Expected output': 'expected_output'} ignore_keys: if your json dicts contain keys that are not needed for DatasetItem construction - pass them as ignore_keys argument + deduplication: Whether to skip items whose content already exists in + the dataset. See :meth:`insert` for details. """ keys_mapping = {} if keys_mapping is None else keys_mapping ignore_keys = [] if ignore_keys is None else ignore_keys new_items = converters.from_jsonl_file(file_path, keys_mapping, ignore_keys) - self.insert(new_items) + self.insert(new_items, deduplication=deduplication) def insert_from_pandas( self, dataframe: "pd.DataFrame", keys_mapping: Optional[Dict[str, str]] = None, ignore_keys: Optional[List[str]] = None, + deduplication: bool = True, ) -> None: """ Requires: `pandas` library to be installed. @@ -967,13 +995,15 @@ def insert_from_pandas( Example: {'Expected output': 'expected_output'} ignore_keys: if your dataframe contains columns that are not needed for DatasetItem construction - pass them as ignore_keys argument + deduplication: Whether to skip items whose content already exists in + the dataset. See :meth:`insert` for details. """ keys_mapping = {} if keys_mapping is None else keys_mapping ignore_keys = [] if ignore_keys is None else ignore_keys new_items = converters.from_pandas(dataframe, keys_mapping, ignore_keys) - self.insert(new_items) + self.insert(new_items, deduplication=deduplication) def get_version_view(self, version_name: str) -> DatasetVersion: """ diff --git a/sdks/python/src/opik/api_objects/dataset/test_suite/test_suite.py b/sdks/python/src/opik/api_objects/dataset/test_suite/test_suite.py index 8894ad999d4..013b38af4e8 100644 --- a/sdks/python/src/opik/api_objects/dataset/test_suite/test_suite.py +++ b/sdks/python/src/opik/api_objects/dataset/test_suite/test_suite.py @@ -434,6 +434,7 @@ def insert_from_json( json_array: str, keys_mapping: Optional[Dict[str, str]] = None, ignore_keys: Optional[List[str]] = None, + deduplication: bool = True, ) -> None: """ Insert test suite items from a JSON string. @@ -452,17 +453,23 @@ def insert_from_json( keys_mapping: Maps JSON keys to the target keys listed above. Example: ``{"test_data": "data", "checks": "assertions"}`` ignore_keys: Keys in the JSON dicts to skip during import. + deduplication: Whether to skip items whose content already exists in + the suite. See :meth:`insert` for details. """ keys_mapping = {} if keys_mapping is None else keys_mapping ignore_keys = [] if ignore_keys is None else ignore_keys - self.insert(converters.from_json(json_array, keys_mapping, ignore_keys)) + self.insert( + converters.from_json(json_array, keys_mapping, ignore_keys), + deduplication=deduplication, + ) def insert_from_pandas( self, dataframe: "pd.DataFrame", keys_mapping: Optional[Dict[str, str]] = None, ignore_keys: Optional[List[str]] = None, + deduplication: bool = True, ) -> None: """ Insert test suite items from a pandas DataFrame. @@ -483,17 +490,23 @@ def insert_from_pandas( keys_mapping: Maps column names to the target keys listed above. Example: ``{"test_data": "data", "checks": "assertions"}`` ignore_keys: Column names in the DataFrame to skip during import. + deduplication: Whether to skip items whose content already exists in + the suite. See :meth:`insert` for details. """ keys_mapping = {} if keys_mapping is None else keys_mapping ignore_keys = [] if ignore_keys is None else ignore_keys - self.insert(converters.from_pandas(dataframe, keys_mapping, ignore_keys)) + self.insert( + converters.from_pandas(dataframe, keys_mapping, ignore_keys), + deduplication=deduplication, + ) def insert_from_jsonl_file( self, file_path: str, keys_mapping: Optional[Dict[str, str]] = None, ignore_keys: Optional[List[str]] = None, + deduplication: bool = True, ) -> None: """ Read JSONL from a file and insert items into the test suite. @@ -512,11 +525,16 @@ def insert_from_jsonl_file( keys_mapping: Maps JSON keys to the target keys listed above. Example: ``{"test_data": "data", "checks": "assertions"}`` ignore_keys: Keys in the JSON objects to skip during import. + deduplication: Whether to skip items whose content already exists in + the suite. See :meth:`insert` for details. """ keys_mapping = {} if keys_mapping is None else keys_mapping ignore_keys = [] if ignore_keys is None else ignore_keys - self.insert(converters.from_jsonl_file(file_path, keys_mapping, ignore_keys)) + self.insert( + converters.from_jsonl_file(file_path, keys_mapping, ignore_keys), + deduplication=deduplication, + ) def update_test_settings( self, @@ -643,6 +661,7 @@ def get_global_assertions(self) -> List[str]: def update( self, items: List[suite_types.TestSuiteItem], + deduplication: bool = True, ) -> None: """ Update existing items in the test suite. @@ -653,6 +672,8 @@ def update( Args: items: List of item dicts to update. Each must contain ``"id"``. + deduplication: Whether to skip items whose content already exists in + the suite. See :meth:`insert` for details. Raises: DatasetItemUpdateOperationRequiresItemId: If any item is missing @@ -664,17 +685,22 @@ def update( "Missing id for test suite item to update: %s", item ) - self.insert(items) + self.insert(items, deduplication=deduplication) def insert( self, items: List[suite_types.TestSuiteItem], + deduplication: bool = True, ) -> None: """ Insert test cases into the test suite. Args: items: List of test case items to add. + deduplication: Whether to skip items whose content already exists + in the suite. Pass ``False`` to insert every item as-is + without any duplicate checking, which is significantly faster + on large suites. Example: >>> suite.insert([ @@ -689,7 +715,9 @@ def insert( validators.validate_suite_items(items) ds_items = [converters.suite_item_dict_to_dataset_item(item) for item in items] - self._dataset.__internal_api__insert_items_as_dataclasses__(ds_items) + self._dataset.__internal_api__insert_items_as_dataclasses__( + ds_items, deduplication=deduplication + ) def __internal_api__run_optimization_suite__( self, diff --git a/sdks/python/tests/unit/api_objects/dataset/test_dataset_client.py b/sdks/python/tests/unit/api_objects/dataset/test_dataset_client.py index 67ce7a3bbcf..a60758cc29c 100644 --- a/sdks/python/tests/unit/api_objects/dataset/test_dataset_client.py +++ b/sdks/python/tests/unit/api_objects/dataset/test_dataset_client.py @@ -1,6 +1,7 @@ import threading import time -from unittest.mock import Mock +from typing import Optional +from unittest.mock import Mock, patch import pytest @@ -112,6 +113,102 @@ def test_insert_deduplication__three_dicts_passed__one_unique__two_duplicates__t assert len(inserted_rest_items) == 2, "Two items should be inserted" +def test_insert__deduplication_disabled__duplicates_are_inserted(): + mock_rest_client = Mock() + + dataset = Dataset( + name="test_dataset", + description="Test description", + project_name="Test project", + rest_client=mock_rest_client, + ) + + item_dict = { + "input": {"key": "value"}, + "expected_output": {"key": "output"}, + "metadata": {"key": "meta"}, + } + + dataset.insert([item_dict, item_dict], deduplication=False) + + call_args = mock_rest_client.datasets.create_or_update_dataset_items.call_args + assert len(call_args[1]["items"]) == 2, ( + "Both identical items must be sent when deduplication is disabled" + ) + + +def test_insert__deduplication_disabled__backend_items_are_not_downloaded(): + mock_rest_client = Mock() + + dataset = Dataset( + name="test_dataset", + description="Test description", + project_name="Test project", + rest_client=mock_rest_client, + ) + # The state `get_dataset`/`get_datasets` leave behind: the backend holds + # items this object has not hashed yet. + dataset.__internal_api__hashes_synced__ = False + + dataset.insert(_make_items(3), deduplication=False) + + mock_rest_client.datasets.stream_dataset_items.assert_not_called() + + +def test_insert__deduplication_disabled__next_deduplicated_insert_syncs_hashes(): + mock_rest_client = Mock() + + dataset = Dataset( + name="test_dataset", + description="Test description", + project_name="Test project", + rest_client=mock_rest_client, + ) + + dataset.insert(_make_items(3), deduplication=False) + assert not dataset.__internal_api__hashes_synced__, ( + "Skipping deduplication leaves the local hash cache stale" + ) + + with patch.object(dataset, "__internal_api__sync_hashes__") as sync_hashes: + dataset.insert(_make_items(3)) + + sync_hashes.assert_called_once() + + +def test_update__deduplication_disabled__unchanged_item_is_still_sent(): + mock_rest_client = Mock() + + dataset = Dataset( + name="test_dataset", + description="Test description", + project_name="Test project", + rest_client=mock_rest_client, + ) + + item = { + "input": {"key": "value"}, + "expected_output": {"key": "output"}, + "metadata": {"key": "meta"}, + } + dataset.insert([item]) + + inserted_id = mock_rest_client.datasets.create_or_update_dataset_items.call_args[1][ + "items" + ][0].id + + dataset.update([{"id": inserted_id, **item}], deduplication=False) + + assert mock_rest_client.datasets.create_or_update_dataset_items.call_count == 2 + updated_items = mock_rest_client.datasets.create_or_update_dataset_items.call_args[ + 1 + ]["items"] + assert len(updated_items) == 1, ( + "An update with unchanged content must still be sent when deduplication " + "is disabled" + ) + + def test_update__happyflow(): mock_rest_client = Mock() @@ -312,8 +409,9 @@ def test_insert__invalid_num_threads__raises_before_upload(bad_value): def _batches_overlapped( monkeypatch, mock_rest_client: Mock, - num_threads: int, + num_threads: Optional[int], expect_overlap: bool, + item_count: int = _GATE_ITEM_COUNT, ) -> bool: """Insert through the public API and report whether batches ran concurrently. @@ -321,6 +419,10 @@ def _batches_overlapped( boundary rather than at the gate's internal decision — so this still fails if insert() stops honouring the worker count downstream. + ``num_threads=None`` omits the argument so the call exercises the default. + ``item_count`` sets how many batches there are, which must not exceed the + worker count when overlap is expected — the barrier is sized to it. + ``expect_overlap`` selects how uploads are held, because the two expectations fail in opposite directions and need opposite instruments. @@ -339,12 +441,13 @@ def _batches_overlapped( hold once per batch. """ _small_batches(monkeypatch, size=_GATE_BATCH_SIZE) + batch_count = item_count // _GATE_BATCH_SIZE lock = threading.Lock() in_flight = 0 peak_in_flight = 0 barrier = ( - threading.Barrier(_GATE_BATCH_COUNT, timeout=_OVERLAP_TIMEOUT_SECONDS) + threading.Barrier(batch_count, timeout=_OVERLAP_TIMEOUT_SECONDS) if expect_overlap else None ) @@ -375,11 +478,15 @@ def tracked_upload(*args, **kwargs): project_name="Test project", rest_client=mock_rest_client, ) - dataset.insert(_make_items(_GATE_ITEM_COUNT), num_threads=num_threads) + items = _make_items(item_count) + if num_threads is None: + dataset.insert(items) + else: + dataset.insert(items, num_threads=num_threads) assert ( mock_rest_client.datasets.create_or_update_dataset_items.call_count - == _GATE_BATCH_COUNT + == batch_count ), "Every batch must be uploaded regardless of the thread count" return peak_in_flight > 1 @@ -451,5 +558,39 @@ def test_insert__sequential__uploads_sequentially_without_probing_version(monkey monkeypatch, mock_rest_client, num_threads=1, expect_overlap=False ), "num_threads=1 must upload sequentially" - # The default path must not pay an extra request. + # An explicitly sequential upload cannot race, so it must not pay the probe. mock_rest_client.version.assert_not_called() + + +# 4 batches, which is the default worker count — so all of them fit in flight +# together and the barrier can prove the default really uses the pool. +_DEFAULT_THREADS_ITEM_COUNT = _GATE_BATCH_SIZE * 4 + + +def test_insert__num_threads_not_given__uploads_concurrently_by_default(monkeypatch): + assert _batches_overlapped( + monkeypatch, + _mock_rest_client(), + num_threads=None, + expect_overlap=True, + item_count=_DEFAULT_THREADS_ITEM_COUNT, + ), "insert() must upload in parallel without being asked to" + + +def test_insert__repeated_inserts__backend_version_probed_once(monkeypatch): + _small_batches(monkeypatch, size=_GATE_BATCH_SIZE) + mock_rest_client = _mock_rest_client() + dataset = Dataset( + name="test_dataset", + description="Test description", + project_name="Test project", + rest_client=mock_rest_client, + ) + + for _ in range(3): + dataset.insert(_make_items(4), deduplication=False) + + assert mock_rest_client.version.call_count == 1, ( + "Parallel upload is the default, so the version gate must be probed once " + "per dataset rather than once per insert" + ) diff --git a/sdks/python/tests/unit/api_objects/dataset/test_suite/test_test_suite.py b/sdks/python/tests/unit/api_objects/dataset/test_suite/test_test_suite.py index 310cc0fc299..09531e2d997 100644 --- a/sdks/python/tests/unit/api_objects/dataset/test_suite/test_test_suite.py +++ b/sdks/python/tests/unit/api_objects/dataset/test_suite/test_test_suite.py @@ -147,6 +147,22 @@ def test_insert__with_assertions_shorthand__creates_evaluator_items(self): assert len(item.evaluators) == 1 assert item.evaluators[0].type == "llm_judge" + @pytest.mark.parametrize("deduplication", [True, False]) + def test_insert__deduplication_flag__forwarded_to_the_dataset(self, deduplication): + mock_dataset = _create_mock_dataset() + suite = test_suite.TestSuite( + name="test_suite", + dataset_=mock_dataset, + ) + + suite.insert( + [{"data": {"input": "test"}}], + deduplication=deduplication, + ) + + call_args = mock_dataset.__internal_api__insert_items_as_dataclasses__.call_args + assert call_args[1]["deduplication"] is deduplication + def test_resolve_evaluators__with_both_assertions_and_evaluators__raises_value_error( self, ):