# This file was auto-generated by Fern from our API Definition. import typing from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.request_options import RequestOptions from ..types.experiment_group_aggregations_response import ExperimentGroupAggregationsResponse from ..types.experiment_group_response import ExperimentGroupResponse from ..types.experiment_item import ExperimentItem from ..types.experiment_item_bulk_record_experiment_item_bulk_write_view import ( ExperimentItemBulkRecordExperimentItemBulkWriteView, ) from ..types.experiment_item_public import ExperimentItemPublic from ..types.experiment_page_public import ExperimentPagePublic from ..types.experiment_public import ExperimentPublic from ..types.experiment_score import ExperimentScore from ..types.experiment_score_write import ExperimentScoreWrite from ..types.experiment_update import ExperimentUpdate from ..types.experiment_update_status import ExperimentUpdateStatus from ..types.experiment_update_type import ExperimentUpdateType from ..types.feedback_score_names_public import FeedbackScoreNamesPublic from ..types.json_list_string_write import JsonListStringWrite from ..types.json_node import JsonNode from ..types.prompt_version_link_write import PromptVersionLinkWrite from .raw_client import AsyncRawExperimentsClient, RawExperimentsClient from .types.experiment_write_evaluation_method import ExperimentWriteEvaluationMethod from .types.experiment_write_status import ExperimentWriteStatus from .types.experiment_write_type import ExperimentWriteType # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class ExperimentsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._raw_client = RawExperimentsClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> RawExperimentsClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- RawExperimentsClient """ return self._raw_client def batch_update_experiments( self, *, ids: typing.Sequence[str], update: ExperimentUpdate, merge_tags: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update multiple experiments Parameters ---------- ids : typing.Sequence[str] List of experiment IDs to update (max 1000) update : ExperimentUpdate merge_tags : typing.Optional[bool] If true, merge tags with existing tags instead of replacing them. Default: false request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi from Opik import ExperimentUpdate client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.batch_update_experiments(ids=['ids'], update=ExperimentUpdate(), ) """ _response = self._raw_client.batch_update_experiments( ids=ids, update=update, merge_tags=merge_tags, request_options=request_options ) return _response.data def find_experiments( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, dataset_id: typing.Optional[str] = None, optimization_id: typing.Optional[str] = None, types: typing.Optional[str] = None, name: typing.Optional[str] = None, dataset_deleted: typing.Optional[bool] = None, prompt_id: typing.Optional[str] = None, project_id: typing.Optional[str] = None, project_deleted: typing.Optional[bool] = None, sorting: typing.Optional[str] = None, filters: typing.Optional[str] = None, experiment_ids: typing.Optional[str] = None, force_sorting: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ExperimentPagePublic: """ Find experiments Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] dataset_id : typing.Optional[str] optimization_id : typing.Optional[str] types : typing.Optional[str] name : typing.Optional[str] dataset_deleted : typing.Optional[bool] prompt_id : typing.Optional[str] project_id : typing.Optional[str] project_deleted : typing.Optional[bool] sorting : typing.Optional[str] filters : typing.Optional[str] experiment_ids : typing.Optional[str] force_sorting : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ExperimentPagePublic Experiments resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.find_experiments() """ _response = self._raw_client.find_experiments( page=page, size=size, dataset_id=dataset_id, optimization_id=optimization_id, types=types, name=name, dataset_deleted=dataset_deleted, prompt_id=prompt_id, project_id=project_id, project_deleted=project_deleted, sorting=sorting, filters=filters, experiment_ids=experiment_ids, force_sorting=force_sorting, request_options=request_options, ) return _response.data def create_experiment( self, *, id: typing.Optional[str] = OMIT, dataset_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, metadata: typing.Optional[JsonListStringWrite] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, type: typing.Optional[ExperimentWriteType] = OMIT, evaluation_method: typing.Optional[ExperimentWriteEvaluationMethod] = OMIT, optimization_id: typing.Optional[str] = OMIT, status: typing.Optional[ExperimentWriteStatus] = OMIT, experiment_scores: typing.Optional[typing.Sequence[ExperimentScoreWrite]] = OMIT, prompt_version: typing.Optional[PromptVersionLinkWrite] = OMIT, prompt_versions: typing.Optional[typing.Sequence[PromptVersionLinkWrite]] = OMIT, dataset_version_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Create experiment Parameters ---------- id : typing.Optional[str] dataset_name : typing.Optional[str] project_id : typing.Optional[str] Project ID. Takes precedence over project_name when both are provided. project_name : typing.Optional[str] Project name. Creates project if it doesn't exist. Ignored when project_id is provided. name : typing.Optional[str] metadata : typing.Optional[JsonListStringWrite] tags : typing.Optional[typing.Sequence[str]] type : typing.Optional[ExperimentWriteType] evaluation_method : typing.Optional[ExperimentWriteEvaluationMethod] optimization_id : typing.Optional[str] status : typing.Optional[ExperimentWriteStatus] experiment_scores : typing.Optional[typing.Sequence[ExperimentScoreWrite]] prompt_version : typing.Optional[PromptVersionLinkWrite] prompt_versions : typing.Optional[typing.Sequence[PromptVersionLinkWrite]] dataset_version_id : typing.Optional[str] ID of the dataset version this experiment is linked to. If not provided at creation, experiment will be automatically linked to the latest version. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.create_experiment() """ _response = self._raw_client.create_experiment( id=id, dataset_name=dataset_name, project_id=project_id, project_name=project_name, name=name, metadata=metadata, tags=tags, type=type, evaluation_method=evaluation_method, optimization_id=optimization_id, status=status, experiment_scores=experiment_scores, prompt_version=prompt_version, prompt_versions=prompt_versions, dataset_version_id=dataset_version_id, request_options=request_options, ) return _response.data def create_experiment_items( self, *, experiment_items: typing.Sequence[ExperimentItem], request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Create experiment items Parameters ---------- experiment_items : typing.Sequence[ExperimentItem] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi from Opik import ExperimentItem client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.create_experiment_items(experiment_items=[ExperimentItem(experiment_id='experiment_id', dataset_item_id='dataset_item_id', trace_id='trace_id', )], ) """ _response = self._raw_client.create_experiment_items( experiment_items=experiment_items, request_options=request_options ) return _response.data def delete_experiment_items( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Delete experiment items Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.delete_experiment_items(ids=['ids'], ) """ _response = self._raw_client.delete_experiment_items(ids=ids, request_options=request_options) return _response.data def delete_experiments_by_id( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Delete experiments by id Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.delete_experiments_by_id(ids=['ids'], ) """ _response = self._raw_client.delete_experiments_by_id(ids=ids, request_options=request_options) return _response.data def experiment_items_bulk( self, *, experiment_name: str, dataset_name: str, items: typing.Sequence[ExperimentItemBulkRecordExperimentItemBulkWriteView], experiment_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Record experiment items in bulk with traces, spans, and feedback scores. Maximum request size is 4MB. Parameters ---------- experiment_name : str dataset_name : str items : typing.Sequence[ExperimentItemBulkRecordExperimentItemBulkWriteView] experiment_id : typing.Optional[str] Optional experiment ID. If provided, items will be added to the existing experiment and experimentName will be ignored. If not provided or experiment with that ID doesn't exist, a new experiment will be created with the given experimentName request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi from Opik import ExperimentItemBulkRecordExperimentItemBulkWriteView client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.experiment_items_bulk(experiment_name='experiment_name', dataset_name='dataset_name', items=[ExperimentItemBulkRecordExperimentItemBulkWriteView(dataset_item_id='dataset_item_id', )], ) """ _response = self._raw_client.experiment_items_bulk( experiment_name=experiment_name, dataset_name=dataset_name, items=items, experiment_id=experiment_id, request_options=request_options, ) return _response.data def find_feedback_score_names( self, *, experiment_ids: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> FeedbackScoreNamesPublic: """ Find Feedback Score names Parameters ---------- experiment_ids : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- FeedbackScoreNamesPublic Feedback Scores resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.find_feedback_score_names() """ _response = self._raw_client.find_feedback_score_names( experiment_ids=experiment_ids, request_options=request_options ) return _response.data def find_experiment_groups( self, *, groups: typing.Optional[str] = None, types: typing.Optional[str] = None, name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, project_deleted: typing.Optional[bool] = None, filters: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ExperimentGroupResponse: """ Find experiments grouped by specified fields Parameters ---------- groups : typing.Optional[str] types : typing.Optional[str] name : typing.Optional[str] project_id : typing.Optional[str] project_deleted : typing.Optional[bool] filters : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ExperimentGroupResponse Experiment groups Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.find_experiment_groups() """ _response = self._raw_client.find_experiment_groups( groups=groups, types=types, name=name, project_id=project_id, project_deleted=project_deleted, filters=filters, request_options=request_options, ) return _response.data def find_experiment_groups_aggregations( self, *, groups: typing.Optional[str] = None, types: typing.Optional[str] = None, name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, project_deleted: typing.Optional[bool] = None, filters: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ExperimentGroupAggregationsResponse: """ Find experiments grouped by specified fields with aggregation metrics Parameters ---------- groups : typing.Optional[str] types : typing.Optional[str] name : typing.Optional[str] project_id : typing.Optional[str] project_deleted : typing.Optional[bool] filters : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ExperimentGroupAggregationsResponse Experiment groups with aggregations Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.find_experiment_groups_aggregations() """ _response = self._raw_client.find_experiment_groups_aggregations( groups=groups, types=types, name=name, project_id=project_id, project_deleted=project_deleted, filters=filters, request_options=request_options, ) return _response.data def finish_experiments( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Finish experiments and trigger alert events Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.finish_experiments(ids=['ids'], ) """ _response = self._raw_client.finish_experiments(ids=ids, request_options=request_options) return _response.data def get_experiment_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> ExperimentPublic: """ Get experiment by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ExperimentPublic Experiment resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.get_experiment_by_id(id='id', ) """ _response = self._raw_client.get_experiment_by_id(id, request_options=request_options) return _response.data def update_experiment( self, id: str, *, name: typing.Optional[str] = OMIT, metadata: typing.Optional[JsonNode] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_add: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_remove: typing.Optional[typing.Sequence[str]] = OMIT, type: typing.Optional[ExperimentUpdateType] = OMIT, status: typing.Optional[ExperimentUpdateStatus] = OMIT, experiment_scores: typing.Optional[typing.Sequence[ExperimentScore]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update experiment by id Parameters ---------- id : str name : typing.Optional[str] metadata : typing.Optional[JsonNode] tags : typing.Optional[typing.Sequence[str]] Tags tags_to_add : typing.Optional[typing.Sequence[str]] Tags to add tags_to_remove : typing.Optional[typing.Sequence[str]] Tags to remove type : typing.Optional[ExperimentUpdateType] status : typing.Optional[ExperimentUpdateStatus] The status of the experiment experiment_scores : typing.Optional[typing.Sequence[ExperimentScore]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.update_experiment(id='id', ) """ _response = self._raw_client.update_experiment( id, name=name, metadata=metadata, tags=tags, tags_to_add=tags_to_add, tags_to_remove=tags_to_remove, type=type, status=status, experiment_scores=experiment_scores, request_options=request_options, ) return _response.data def get_experiment_item_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> ExperimentItemPublic: """ Get experiment item by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ExperimentItemPublic Experiment item resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.experiments.get_experiment_item_by_id(id='id', ) """ _response = self._raw_client.get_experiment_item_by_id(id, request_options=request_options) return _response.data def stream_experiment_items( self, *, experiment_name: str, limit: typing.Optional[int] = OMIT, last_retrieved_id: typing.Optional[str] = OMIT, truncate: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Iterator[bytes]: """ Stream experiment items Parameters ---------- experiment_name : str limit : typing.Optional[int] last_retrieved_id : typing.Optional[str] truncate : typing.Optional[bool] Truncate image included in either input, output or metadata request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.Iterator[bytes] Experiment items stream or error during process """ with self._raw_client.stream_experiment_items( experiment_name=experiment_name, limit=limit, last_retrieved_id=last_retrieved_id, truncate=truncate, request_options=request_options, ) as r: yield from r.data def stream_experiments( self, *, name: str, limit: typing.Optional[int] = OMIT, last_retrieved_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Iterator[bytes]: """ Stream experiments Parameters ---------- name : str limit : typing.Optional[int] last_retrieved_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.Iterator[bytes] Experiments stream or error during process """ with self._raw_client.stream_experiments( name=name, limit=limit, last_retrieved_id=last_retrieved_id, request_options=request_options ) as r: yield from r.data class AsyncExperimentsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._raw_client = AsyncRawExperimentsClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> AsyncRawExperimentsClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- AsyncRawExperimentsClient """ return self._raw_client async def batch_update_experiments( self, *, ids: typing.Sequence[str], update: ExperimentUpdate, merge_tags: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update multiple experiments Parameters ---------- ids : typing.Sequence[str] List of experiment IDs to update (max 1000) update : ExperimentUpdate merge_tags : typing.Optional[bool] If true, merge tags with existing tags instead of replacing them. Default: false request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi from Opik import ExperimentUpdate import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.batch_update_experiments(ids=['ids'], update=ExperimentUpdate(), ) asyncio.run(main()) """ _response = await self._raw_client.batch_update_experiments( ids=ids, update=update, merge_tags=merge_tags, request_options=request_options ) return _response.data async def find_experiments( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, dataset_id: typing.Optional[str] = None, optimization_id: typing.Optional[str] = None, types: typing.Optional[str] = None, name: typing.Optional[str] = None, dataset_deleted: typing.Optional[bool] = None, prompt_id: typing.Optional[str] = None, project_id: typing.Optional[str] = None, project_deleted: typing.Optional[bool] = None, sorting: typing.Optional[str] = None, filters: typing.Optional[str] = None, experiment_ids: typing.Optional[str] = None, force_sorting: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ExperimentPagePublic: """ Find experiments Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] dataset_id : typing.Optional[str] optimization_id : typing.Optional[str] types : typing.Optional[str] name : typing.Optional[str] dataset_deleted : typing.Optional[bool] prompt_id : typing.Optional[str] project_id : typing.Optional[str] project_deleted : typing.Optional[bool] sorting : typing.Optional[str] filters : typing.Optional[str] experiment_ids : typing.Optional[str] force_sorting : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ExperimentPagePublic Experiments resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.find_experiments() asyncio.run(main()) """ _response = await self._raw_client.find_experiments( page=page, size=size, dataset_id=dataset_id, optimization_id=optimization_id, types=types, name=name, dataset_deleted=dataset_deleted, prompt_id=prompt_id, project_id=project_id, project_deleted=project_deleted, sorting=sorting, filters=filters, experiment_ids=experiment_ids, force_sorting=force_sorting, request_options=request_options, ) return _response.data async def create_experiment( self, *, id: typing.Optional[str] = OMIT, dataset_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, metadata: typing.Optional[JsonListStringWrite] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, type: typing.Optional[ExperimentWriteType] = OMIT, evaluation_method: typing.Optional[ExperimentWriteEvaluationMethod] = OMIT, optimization_id: typing.Optional[str] = OMIT, status: typing.Optional[ExperimentWriteStatus] = OMIT, experiment_scores: typing.Optional[typing.Sequence[ExperimentScoreWrite]] = OMIT, prompt_version: typing.Optional[PromptVersionLinkWrite] = OMIT, prompt_versions: typing.Optional[typing.Sequence[PromptVersionLinkWrite]] = OMIT, dataset_version_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Create experiment Parameters ---------- id : typing.Optional[str] dataset_name : typing.Optional[str] project_id : typing.Optional[str] Project ID. Takes precedence over project_name when both are provided. project_name : typing.Optional[str] Project name. Creates project if it doesn't exist. Ignored when project_id is provided. name : typing.Optional[str] metadata : typing.Optional[JsonListStringWrite] tags : typing.Optional[typing.Sequence[str]] type : typing.Optional[ExperimentWriteType] evaluation_method : typing.Optional[ExperimentWriteEvaluationMethod] optimization_id : typing.Optional[str] status : typing.Optional[ExperimentWriteStatus] experiment_scores : typing.Optional[typing.Sequence[ExperimentScoreWrite]] prompt_version : typing.Optional[PromptVersionLinkWrite] prompt_versions : typing.Optional[typing.Sequence[PromptVersionLinkWrite]] dataset_version_id : typing.Optional[str] ID of the dataset version this experiment is linked to. If not provided at creation, experiment will be automatically linked to the latest version. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.create_experiment() asyncio.run(main()) """ _response = await self._raw_client.create_experiment( id=id, dataset_name=dataset_name, project_id=project_id, project_name=project_name, name=name, metadata=metadata, tags=tags, type=type, evaluation_method=evaluation_method, optimization_id=optimization_id, status=status, experiment_scores=experiment_scores, prompt_version=prompt_version, prompt_versions=prompt_versions, dataset_version_id=dataset_version_id, request_options=request_options, ) return _response.data async def create_experiment_items( self, *, experiment_items: typing.Sequence[ExperimentItem], request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Create experiment items Parameters ---------- experiment_items : typing.Sequence[ExperimentItem] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi from Opik import ExperimentItem import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.create_experiment_items(experiment_items=[ExperimentItem(experiment_id='experiment_id', dataset_item_id='dataset_item_id', trace_id='trace_id', )], ) asyncio.run(main()) """ _response = await self._raw_client.create_experiment_items( experiment_items=experiment_items, request_options=request_options ) return _response.data async def delete_experiment_items( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Delete experiment items Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.delete_experiment_items(ids=['ids'], ) asyncio.run(main()) """ _response = await self._raw_client.delete_experiment_items(ids=ids, request_options=request_options) return _response.data async def delete_experiments_by_id( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Delete experiments by id Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.delete_experiments_by_id(ids=['ids'], ) asyncio.run(main()) """ _response = await self._raw_client.delete_experiments_by_id(ids=ids, request_options=request_options) return _response.data async def experiment_items_bulk( self, *, experiment_name: str, dataset_name: str, items: typing.Sequence[ExperimentItemBulkRecordExperimentItemBulkWriteView], experiment_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Record experiment items in bulk with traces, spans, and feedback scores. Maximum request size is 4MB. Parameters ---------- experiment_name : str dataset_name : str items : typing.Sequence[ExperimentItemBulkRecordExperimentItemBulkWriteView] experiment_id : typing.Optional[str] Optional experiment ID. If provided, items will be added to the existing experiment and experimentName will be ignored. If not provided or experiment with that ID doesn't exist, a new experiment will be created with the given experimentName request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi from Opik import ExperimentItemBulkRecordExperimentItemBulkWriteView import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.experiment_items_bulk(experiment_name='experiment_name', dataset_name='dataset_name', items=[ExperimentItemBulkRecordExperimentItemBulkWriteView(dataset_item_id='dataset_item_id', )], ) asyncio.run(main()) """ _response = await self._raw_client.experiment_items_bulk( experiment_name=experiment_name, dataset_name=dataset_name, items=items, experiment_id=experiment_id, request_options=request_options, ) return _response.data async def find_feedback_score_names( self, *, experiment_ids: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> FeedbackScoreNamesPublic: """ Find Feedback Score names Parameters ---------- experiment_ids : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- FeedbackScoreNamesPublic Feedback Scores resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.find_feedback_score_names() asyncio.run(main()) """ _response = await self._raw_client.find_feedback_score_names( experiment_ids=experiment_ids, request_options=request_options ) return _response.data async def find_experiment_groups( self, *, groups: typing.Optional[str] = None, types: typing.Optional[str] = None, name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, project_deleted: typing.Optional[bool] = None, filters: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ExperimentGroupResponse: """ Find experiments grouped by specified fields Parameters ---------- groups : typing.Optional[str] types : typing.Optional[str] name : typing.Optional[str] project_id : typing.Optional[str] project_deleted : typing.Optional[bool] filters : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ExperimentGroupResponse Experiment groups Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.find_experiment_groups() asyncio.run(main()) """ _response = await self._raw_client.find_experiment_groups( groups=groups, types=types, name=name, project_id=project_id, project_deleted=project_deleted, filters=filters, request_options=request_options, ) return _response.data async def find_experiment_groups_aggregations( self, *, groups: typing.Optional[str] = None, types: typing.Optional[str] = None, name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, project_deleted: typing.Optional[bool] = None, filters: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ExperimentGroupAggregationsResponse: """ Find experiments grouped by specified fields with aggregation metrics Parameters ---------- groups : typing.Optional[str] types : typing.Optional[str] name : typing.Optional[str] project_id : typing.Optional[str] project_deleted : typing.Optional[bool] filters : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ExperimentGroupAggregationsResponse Experiment groups with aggregations Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.find_experiment_groups_aggregations() asyncio.run(main()) """ _response = await self._raw_client.find_experiment_groups_aggregations( groups=groups, types=types, name=name, project_id=project_id, project_deleted=project_deleted, filters=filters, request_options=request_options, ) return _response.data async def finish_experiments( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Finish experiments and trigger alert events Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.finish_experiments(ids=['ids'], ) asyncio.run(main()) """ _response = await self._raw_client.finish_experiments(ids=ids, request_options=request_options) return _response.data async def get_experiment_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> ExperimentPublic: """ Get experiment by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ExperimentPublic Experiment resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.get_experiment_by_id(id='id', ) asyncio.run(main()) """ _response = await self._raw_client.get_experiment_by_id(id, request_options=request_options) return _response.data async def update_experiment( self, id: str, *, name: typing.Optional[str] = OMIT, metadata: typing.Optional[JsonNode] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_add: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_remove: typing.Optional[typing.Sequence[str]] = OMIT, type: typing.Optional[ExperimentUpdateType] = OMIT, status: typing.Optional[ExperimentUpdateStatus] = OMIT, experiment_scores: typing.Optional[typing.Sequence[ExperimentScore]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update experiment by id Parameters ---------- id : str name : typing.Optional[str] metadata : typing.Optional[JsonNode] tags : typing.Optional[typing.Sequence[str]] Tags tags_to_add : typing.Optional[typing.Sequence[str]] Tags to add tags_to_remove : typing.Optional[typing.Sequence[str]] Tags to remove type : typing.Optional[ExperimentUpdateType] status : typing.Optional[ExperimentUpdateStatus] The status of the experiment experiment_scores : typing.Optional[typing.Sequence[ExperimentScore]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.update_experiment(id='id', ) asyncio.run(main()) """ _response = await self._raw_client.update_experiment( id, name=name, metadata=metadata, tags=tags, tags_to_add=tags_to_add, tags_to_remove=tags_to_remove, type=type, status=status, experiment_scores=experiment_scores, request_options=request_options, ) return _response.data async def get_experiment_item_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> ExperimentItemPublic: """ Get experiment item by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ExperimentItemPublic Experiment item resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.experiments.get_experiment_item_by_id(id='id', ) asyncio.run(main()) """ _response = await self._raw_client.get_experiment_item_by_id(id, request_options=request_options) return _response.data async def stream_experiment_items( self, *, experiment_name: str, limit: typing.Optional[int] = OMIT, last_retrieved_id: typing.Optional[str] = OMIT, truncate: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.AsyncIterator[bytes]: """ Stream experiment items Parameters ---------- experiment_name : str limit : typing.Optional[int] last_retrieved_id : typing.Optional[str] truncate : typing.Optional[bool] Truncate image included in either input, output or metadata request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.AsyncIterator[bytes] Experiment items stream or error during process """ async with self._raw_client.stream_experiment_items( experiment_name=experiment_name, limit=limit, last_retrieved_id=last_retrieved_id, truncate=truncate, request_options=request_options, ) as r: async for data in r.data: yield data async def stream_experiments( self, *, name: str, limit: typing.Optional[int] = OMIT, last_retrieved_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.AsyncIterator[bytes]: """ Stream experiments Parameters ---------- name : str limit : typing.Optional[int] last_retrieved_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.AsyncIterator[bytes] Experiments stream or error during process """ async with self._raw_client.stream_experiments( name=name, limit=limit, last_retrieved_id=last_retrieved_id, request_options=request_options ) as r: async for data in r.data: yield data