# This file was auto-generated by Fern from our API Definition. import datetime as dt import typing import urllib.parse from json.decoder import JSONDecodeError from ..core.api_error import ApiError from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.jsonable_encoder import jsonable_encoder from ..core.remove_none_from_dict import remove_none_from_dict from ..core.request_options import RequestOptions from ..core.unchecked_base_model import construct_type from ..errors.bad_request_error import BadRequestError from ..errors.forbidden_error import ForbiddenError from ..errors.internal_server_error import InternalServerError from ..errors.not_found_error import NotFoundError from ..errors.service_unavailable_error import ServiceUnavailableError from ..errors.unauthorized_error import UnauthorizedError from .finetuning.types.create_finetuned_model_response import CreateFinetunedModelResponse from .finetuning.types.delete_finetuned_model_response import DeleteFinetunedModelResponse from .finetuning.types.error import Error from .finetuning.types.finetuned_model import FinetunedModel from .finetuning.types.get_finetuned_model_response import GetFinetunedModelResponse from .finetuning.types.list_events_response import ListEventsResponse from .finetuning.types.list_finetuned_models_response import ListFinetunedModelsResponse from .finetuning.types.list_training_step_metrics_response import ListTrainingStepMetricsResponse from .finetuning.types.settings import Settings from .finetuning.types.status import Status from .finetuning.types.update_finetuned_model_response import UpdateFinetunedModelResponse # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class FinetuningClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list_finetuned_models( self, *, page_size: typing.Optional[int] = None, page_token: typing.Optional[str] = None, order_by: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ListFinetunedModelsResponse: """ Parameters: - page_size: typing.Optional[int]. Maximum number of results to be returned by the server. If 0, defaults to 50. - page_token: typing.Optional[str]. Request a specific page of the list results. - order_by: typing.Optional[str]. Comma separated list of fields. For example: "created_at,name". The default sorting order is ascending. To specify descending order for a field, append " desc" to the field name. For example: "created_at desc,name". Supported sorting fields: - created_at (default) - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import Client client = Client( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) client.finetuning.list_finetuned_models() """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "finetuning/finetuned-models"), params=jsonable_encoder( remove_none_from_dict( { "page_size": page_size, "page_token": page_token, "order_by": order_by, **( request_options.get("additional_query_parameters", {}) if request_options is not None else {} ), } ) ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(ListFinetunedModelsResponse, construct_type(type_=ListFinetunedModelsResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def create_finetuned_model( self, *, request: FinetunedModel, request_options: typing.Optional[RequestOptions] = None ) -> CreateFinetunedModelResponse: """ Parameters: - request: FinetunedModel. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import Client from cohere.finetuning import BaseModel, FinetunedModel, Settings client = Client( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) client.finetuning.create_finetuned_model( request=FinetunedModel( name="api-test", settings=Settings( base_model=BaseModel( base_type="BASE_TYPE_GENERATIVE", ), dataset_id="my-dataset-id", ), ), ) """ _response = self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "finetuning/finetuned-models"), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), json=jsonable_encoder(request) if request_options is None or request_options.get("additional_body_parameters") is None else { **jsonable_encoder(request), **(jsonable_encoder(remove_none_from_dict(request_options.get("additional_body_parameters", {})))), }, headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(CreateFinetunedModelResponse, construct_type(type_=CreateFinetunedModelResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def get_finetuned_model( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> GetFinetunedModelResponse: """ Parameters: - id: str. The fine-tuned model ID. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import Client client = Client( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) client.finetuning.get_finetuned_model( id="id", ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"finetuning/finetuned-models/{jsonable_encoder(id)}" ), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(GetFinetunedModelResponse, construct_type(type_=GetFinetunedModelResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def delete_finetuned_model( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> DeleteFinetunedModelResponse: """ Parameters: - id: str. The fine-tuned model ID. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import Client client = Client( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) client.finetuning.delete_finetuned_model( id="id", ) """ _response = self._client_wrapper.httpx_client.request( "DELETE", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"finetuning/finetuned-models/{jsonable_encoder(id)}" ), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(DeleteFinetunedModelResponse, construct_type(type_=DeleteFinetunedModelResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def update_finetuned_model( self, id: str, *, name: str, creator_id: typing.Optional[str] = OMIT, organization_id: typing.Optional[str] = OMIT, settings: Settings, status: typing.Optional[Status] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, updated_at: typing.Optional[dt.datetime] = OMIT, completed_at: typing.Optional[dt.datetime] = OMIT, last_used: typing.Optional[dt.datetime] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> UpdateFinetunedModelResponse: """ Parameters: - id: str. FinetunedModel ID. - name: str. FinetunedModel name (e.g. `foobar`). - creator_id: typing.Optional[str]. User ID of the creator. - organization_id: typing.Optional[str]. Organization ID. - settings: Settings. FinetunedModel settings such as dataset, hyperparameters... - status: typing.Optional[Status]. Current stage in the life-cycle of the fine-tuned model. - created_at: typing.Optional[dt.datetime]. Creation timestamp. - updated_at: typing.Optional[dt.datetime]. Latest update timestamp. - completed_at: typing.Optional[dt.datetime]. Timestamp for the completed fine-tuning. - last_used: typing.Optional[dt.datetime]. Timestamp for the latest request to this fine-tuned model. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import Client from cohere.finetuning import BaseModel, Settings client = Client( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) client.finetuning.update_finetuned_model( id="id", name="name", settings=Settings( base_model=BaseModel( base_type="BASE_TYPE_UNSPECIFIED", ), dataset_id="dataset_id", ), ) """ _request: typing.Dict[str, typing.Any] = {"name": name, "settings": settings} if creator_id is not OMIT: _request["creator_id"] = creator_id if organization_id is not OMIT: _request["organization_id"] = organization_id if status is not OMIT: _request["status"] = status if created_at is not OMIT: _request["created_at"] = created_at if updated_at is not OMIT: _request["updated_at"] = updated_at if completed_at is not OMIT: _request["completed_at"] = completed_at if last_used is not OMIT: _request["last_used"] = last_used _response = self._client_wrapper.httpx_client.request( "PATCH", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"finetuning/finetuned-models/{jsonable_encoder(id)}" ), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), json=jsonable_encoder(_request) if request_options is None or request_options.get("additional_body_parameters") is None else { **jsonable_encoder(_request), **(jsonable_encoder(remove_none_from_dict(request_options.get("additional_body_parameters", {})))), }, headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(UpdateFinetunedModelResponse, construct_type(type_=UpdateFinetunedModelResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def list_events( self, finetuned_model_id: str, *, page_size: typing.Optional[int] = None, page_token: typing.Optional[str] = None, order_by: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ListEventsResponse: """ Parameters: - finetuned_model_id: str. The parent fine-tuned model ID. - page_size: typing.Optional[int]. Maximum number of results to be returned by the server. If 0, defaults to 50. - page_token: typing.Optional[str]. Request a specific page of the list results. - order_by: typing.Optional[str]. Comma separated list of fields. For example: "created_at,name". The default sorting order is ascending. To specify descending order for a field, append " desc" to the field name. For example: "created_at desc,name". Supported sorting fields: - created_at (default) - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import Client client = Client( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) client.finetuning.list_events( finetuned_model_id="finetuned_model_id", ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"finetuning/finetuned-models/{jsonable_encoder(finetuned_model_id)}/events", ), params=jsonable_encoder( remove_none_from_dict( { "page_size": page_size, "page_token": page_token, "order_by": order_by, **( request_options.get("additional_query_parameters", {}) if request_options is not None else {} ), } ) ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(ListEventsResponse, construct_type(type_=ListEventsResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def list_training_step_metrics( self, finetuned_model_id: str, *, page_size: typing.Optional[int] = None, page_token: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ListTrainingStepMetricsResponse: """ Parameters: - finetuned_model_id: str. The parent fine-tuned model ID. - page_size: typing.Optional[int]. Maximum number of results to be returned by the server. If 0, defaults to 50. - page_token: typing.Optional[str]. Request a specific page of the list results. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import Client client = Client( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) client.finetuning.list_training_step_metrics( finetuned_model_id="finetuned_model_id", ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"finetuning/finetuned-models/{jsonable_encoder(finetuned_model_id)}/metrics", ), params=jsonable_encoder( remove_none_from_dict( { "page_size": page_size, "page_token": page_token, **( request_options.get("additional_query_parameters", {}) if request_options is not None else {} ), } ) ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(ListTrainingStepMetricsResponse, construct_type(type_=ListTrainingStepMetricsResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) class AsyncFinetuningClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list_finetuned_models( self, *, page_size: typing.Optional[int] = None, page_token: typing.Optional[str] = None, order_by: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ListFinetunedModelsResponse: """ Parameters: - page_size: typing.Optional[int]. Maximum number of results to be returned by the server. If 0, defaults to 50. - page_token: typing.Optional[str]. Request a specific page of the list results. - order_by: typing.Optional[str]. Comma separated list of fields. For example: "created_at,name". The default sorting order is ascending. To specify descending order for a field, append " desc" to the field name. For example: "created_at desc,name". Supported sorting fields: - created_at (default) - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import AsyncClient client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) await client.finetuning.list_finetuned_models() """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "finetuning/finetuned-models"), params=jsonable_encoder( remove_none_from_dict( { "page_size": page_size, "page_token": page_token, "order_by": order_by, **( request_options.get("additional_query_parameters", {}) if request_options is not None else {} ), } ) ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(ListFinetunedModelsResponse, construct_type(type_=ListFinetunedModelsResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def create_finetuned_model( self, *, request: FinetunedModel, request_options: typing.Optional[RequestOptions] = None ) -> CreateFinetunedModelResponse: """ Parameters: - request: FinetunedModel. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import AsyncClient from cohere.finetuning import BaseModel, FinetunedModel, Settings client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) await client.finetuning.create_finetuned_model( request=FinetunedModel( name="api-test", settings=Settings( base_model=BaseModel( base_type="BASE_TYPE_GENERATIVE", ), dataset_id="my-dataset-id", ), ), ) """ _response = await self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "finetuning/finetuned-models"), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), json=jsonable_encoder(request) if request_options is None or request_options.get("additional_body_parameters") is None else { **jsonable_encoder(request), **(jsonable_encoder(remove_none_from_dict(request_options.get("additional_body_parameters", {})))), }, headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(CreateFinetunedModelResponse, construct_type(type_=CreateFinetunedModelResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def get_finetuned_model( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> GetFinetunedModelResponse: """ Parameters: - id: str. The fine-tuned model ID. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import AsyncClient client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) await client.finetuning.get_finetuned_model( id="id", ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"finetuning/finetuned-models/{jsonable_encoder(id)}" ), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(GetFinetunedModelResponse, construct_type(type_=GetFinetunedModelResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def delete_finetuned_model( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> DeleteFinetunedModelResponse: """ Parameters: - id: str. The fine-tuned model ID. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import AsyncClient client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) await client.finetuning.delete_finetuned_model( id="id", ) """ _response = await self._client_wrapper.httpx_client.request( "DELETE", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"finetuning/finetuned-models/{jsonable_encoder(id)}" ), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(DeleteFinetunedModelResponse, construct_type(type_=DeleteFinetunedModelResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def update_finetuned_model( self, id: str, *, name: str, creator_id: typing.Optional[str] = OMIT, organization_id: typing.Optional[str] = OMIT, settings: Settings, status: typing.Optional[Status] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, updated_at: typing.Optional[dt.datetime] = OMIT, completed_at: typing.Optional[dt.datetime] = OMIT, last_used: typing.Optional[dt.datetime] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> UpdateFinetunedModelResponse: """ Parameters: - id: str. FinetunedModel ID. - name: str. FinetunedModel name (e.g. `foobar`). - creator_id: typing.Optional[str]. User ID of the creator. - organization_id: typing.Optional[str]. Organization ID. - settings: Settings. FinetunedModel settings such as dataset, hyperparameters... - status: typing.Optional[Status]. Current stage in the life-cycle of the fine-tuned model. - created_at: typing.Optional[dt.datetime]. Creation timestamp. - updated_at: typing.Optional[dt.datetime]. Latest update timestamp. - completed_at: typing.Optional[dt.datetime]. Timestamp for the completed fine-tuning. - last_used: typing.Optional[dt.datetime]. Timestamp for the latest request to this fine-tuned model. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import AsyncClient from cohere.finetuning import BaseModel, Settings client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) await client.finetuning.update_finetuned_model( id="id", name="name", settings=Settings( base_model=BaseModel( base_type="BASE_TYPE_UNSPECIFIED", ), dataset_id="dataset_id", ), ) """ _request: typing.Dict[str, typing.Any] = {"name": name, "settings": settings} if creator_id is not OMIT: _request["creator_id"] = creator_id if organization_id is not OMIT: _request["organization_id"] = organization_id if status is not OMIT: _request["status"] = status if created_at is not OMIT: _request["created_at"] = created_at if updated_at is not OMIT: _request["updated_at"] = updated_at if completed_at is not OMIT: _request["completed_at"] = completed_at if last_used is not OMIT: _request["last_used"] = last_used _response = await self._client_wrapper.httpx_client.request( "PATCH", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"finetuning/finetuned-models/{jsonable_encoder(id)}" ), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), json=jsonable_encoder(_request) if request_options is None or request_options.get("additional_body_parameters") is None else { **jsonable_encoder(_request), **(jsonable_encoder(remove_none_from_dict(request_options.get("additional_body_parameters", {})))), }, headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(UpdateFinetunedModelResponse, construct_type(type_=UpdateFinetunedModelResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def list_events( self, finetuned_model_id: str, *, page_size: typing.Optional[int] = None, page_token: typing.Optional[str] = None, order_by: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ListEventsResponse: """ Parameters: - finetuned_model_id: str. The parent fine-tuned model ID. - page_size: typing.Optional[int]. Maximum number of results to be returned by the server. If 0, defaults to 50. - page_token: typing.Optional[str]. Request a specific page of the list results. - order_by: typing.Optional[str]. Comma separated list of fields. For example: "created_at,name". The default sorting order is ascending. To specify descending order for a field, append " desc" to the field name. For example: "created_at desc,name". Supported sorting fields: - created_at (default) - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import AsyncClient client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) await client.finetuning.list_events( finetuned_model_id="finetuned_model_id", ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"finetuning/finetuned-models/{jsonable_encoder(finetuned_model_id)}/events", ), params=jsonable_encoder( remove_none_from_dict( { "page_size": page_size, "page_token": page_token, "order_by": order_by, **( request_options.get("additional_query_parameters", {}) if request_options is not None else {} ), } ) ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(ListEventsResponse, construct_type(type_=ListEventsResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def list_training_step_metrics( self, finetuned_model_id: str, *, page_size: typing.Optional[int] = None, page_token: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ListTrainingStepMetricsResponse: """ Parameters: - finetuned_model_id: str. The parent fine-tuned model ID. - page_size: typing.Optional[int]. Maximum number of results to be returned by the server. If 0, defaults to 50. - page_token: typing.Optional[str]. Request a specific page of the list results. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from cohere.client import AsyncClient client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) await client.finetuning.list_training_step_metrics( finetuned_model_id="finetuned_model_id", ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"finetuning/finetuned-models/{jsonable_encoder(finetuned_model_id)}/metrics", ), params=jsonable_encoder( remove_none_from_dict( { "page_size": page_size, "page_token": page_token, **( request_options.get("additional_query_parameters", {}) if request_options is not None else {} ), } ) ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else self._client_wrapper.get_timeout(), retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return typing.cast(ListTrainingStepMetricsResponse, construct_type(type_=ListTrainingStepMetricsResponse, object_=_response.json())) # type: ignore if _response.status_code == 400: raise BadRequestError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 401: raise UnauthorizedError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) if _response.status_code == 403: raise ForbiddenError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 404: raise NotFoundError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 500: raise InternalServerError( typing.cast(typing.Any, construct_type(type_=typing.Any, object_=_response.json())) # type: ignore ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast(Error, construct_type(type_=Error, object_=_response.json())) # type: ignore ) try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json)