# This file was auto-generated by Fern from our API Definition. import typing from ..core.client_wrapper import SyncClientWrapper from ..core.request_options import RequestOptions from .finetuning.types.list_finetuned_models_response import ListFinetunedModelsResponse from ..core.unchecked_base_model import construct_type from ..errors.bad_request_error import BadRequestError from ..errors.unauthorized_error import UnauthorizedError from ..errors.forbidden_error import ForbiddenError from ..errors.not_found_error import NotFoundError from ..errors.internal_server_error import InternalServerError from ..errors.service_unavailable_error import ServiceUnavailableError from json.decoder import JSONDecodeError from ..core.api_error import ApiError from .finetuning.types.finetuned_model import FinetunedModel from .finetuning.types.create_finetuned_model_response import CreateFinetunedModelResponse from ..core.serialization import convert_and_respect_annotation_metadata from .finetuning.types.get_finetuned_model_response import GetFinetunedModelResponse from ..core.jsonable_encoder import jsonable_encoder from .finetuning.types.delete_finetuned_model_response import DeleteFinetunedModelResponse from .finetuning.types.settings import Settings from .finetuning.types.status import Status import datetime as dt from .finetuning.types.update_finetuned_model_response import UpdateFinetunedModelResponse from .finetuning.types.list_events_response import ListEventsResponse from .finetuning.types.list_training_step_metrics_response import ListTrainingStepMetricsResponse from ..core.client_wrapper import AsyncClientWrapper # 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. Returns ------- ListFinetunedModelsResponse A successful response. Examples -------- from cohere import Client client = Client( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) client.finetuning.list_finetuned_models() """ _response = self._client_wrapper.httpx_client.request( "v1/finetuning/finetuned-models", method="GET", params={ "page_size": page_size, "page_token": page_token, "order_by": order_by, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: return typing.cast( ListFinetunedModelsResponse, construct_type( type_=ListFinetunedModelsResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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. Returns ------- CreateFinetunedModelResponse A successful response. Examples -------- from cohere import Client from cohere.finetuning.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_CHAT", ), dataset_id="my-dataset-id", ), ), ) """ _response = self._client_wrapper.httpx_client.request( "v1/finetuning/finetuned-models", method="POST", json=convert_and_respect_annotation_metadata(object_=request, annotation=FinetunedModel, direction="write"), request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return typing.cast( CreateFinetunedModelResponse, construct_type( type_=CreateFinetunedModelResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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. Returns ------- GetFinetunedModelResponse A successful response. Examples -------- from cohere 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( f"v1/finetuning/finetuned-models/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return typing.cast( GetFinetunedModelResponse, construct_type( type_=GetFinetunedModelResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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. Returns ------- DeleteFinetunedModelResponse A successful response. Examples -------- from cohere 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( f"v1/finetuning/finetuned-models/{jsonable_encoder(id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return typing.cast( DeleteFinetunedModelResponse, construct_type( type_=DeleteFinetunedModelResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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, settings: Settings, creator_id: typing.Optional[str] = OMIT, organization_id: typing.Optional[str] = OMIT, 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`). settings : Settings FinetunedModel settings such as dataset, hyperparameters... creator_id : typing.Optional[str] User ID of the creator. organization_id : typing.Optional[str] Organization ID. 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] Deprecated: Timestamp for the latest request to this fine-tuned model. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- UpdateFinetunedModelResponse A successful response. Examples -------- from cohere import Client from cohere.finetuning.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", ), ) """ _response = self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(id)}", method="PATCH", json={ "name": name, "creator_id": creator_id, "organization_id": organization_id, "settings": convert_and_respect_annotation_metadata( object_=settings, annotation=Settings, direction="write" ), "status": status, "created_at": created_at, "updated_at": updated_at, "completed_at": completed_at, "last_used": last_used, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return typing.cast( UpdateFinetunedModelResponse, construct_type( type_=UpdateFinetunedModelResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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. Returns ------- ListEventsResponse A successful response. Examples -------- from cohere 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( f"v1/finetuning/finetuned-models/{jsonable_encoder(finetuned_model_id)}/events", method="GET", params={ "page_size": page_size, "page_token": page_token, "order_by": order_by, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: return typing.cast( ListEventsResponse, construct_type( type_=ListEventsResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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. Returns ------- ListTrainingStepMetricsResponse A successful response. Examples -------- from cohere 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( f"v1/finetuning/finetuned-models/{jsonable_encoder(finetuned_model_id)}/training-step-metrics", method="GET", params={ "page_size": page_size, "page_token": page_token, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: return typing.cast( ListTrainingStepMetricsResponse, construct_type( type_=ListTrainingStepMetricsResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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. Returns ------- ListFinetunedModelsResponse A successful response. Examples -------- import asyncio from cohere import AsyncClient client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) async def main() -> None: await client.finetuning.list_finetuned_models() asyncio.run(main()) """ _response = await self._client_wrapper.httpx_client.request( "v1/finetuning/finetuned-models", method="GET", params={ "page_size": page_size, "page_token": page_token, "order_by": order_by, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: return typing.cast( ListFinetunedModelsResponse, construct_type( type_=ListFinetunedModelsResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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. Returns ------- CreateFinetunedModelResponse A successful response. Examples -------- import asyncio from cohere import AsyncClient from cohere.finetuning.finetuning import BaseModel, FinetunedModel, Settings client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) async def main() -> None: await client.finetuning.create_finetuned_model( request=FinetunedModel( name="api-test", settings=Settings( base_model=BaseModel( base_type="BASE_TYPE_CHAT", ), dataset_id="my-dataset-id", ), ), ) asyncio.run(main()) """ _response = await self._client_wrapper.httpx_client.request( "v1/finetuning/finetuned-models", method="POST", json=convert_and_respect_annotation_metadata(object_=request, annotation=FinetunedModel, direction="write"), request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return typing.cast( CreateFinetunedModelResponse, construct_type( type_=CreateFinetunedModelResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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. Returns ------- GetFinetunedModelResponse A successful response. Examples -------- import asyncio from cohere import AsyncClient client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) async def main() -> None: await client.finetuning.get_finetuned_model( id="id", ) asyncio.run(main()) """ _response = await self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return typing.cast( GetFinetunedModelResponse, construct_type( type_=GetFinetunedModelResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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. Returns ------- DeleteFinetunedModelResponse A successful response. Examples -------- import asyncio from cohere import AsyncClient client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) async def main() -> None: await client.finetuning.delete_finetuned_model( id="id", ) asyncio.run(main()) """ _response = await self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return typing.cast( DeleteFinetunedModelResponse, construct_type( type_=DeleteFinetunedModelResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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, settings: Settings, creator_id: typing.Optional[str] = OMIT, organization_id: typing.Optional[str] = OMIT, 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`). settings : Settings FinetunedModel settings such as dataset, hyperparameters... creator_id : typing.Optional[str] User ID of the creator. organization_id : typing.Optional[str] Organization ID. 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] Deprecated: Timestamp for the latest request to this fine-tuned model. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- UpdateFinetunedModelResponse A successful response. Examples -------- import asyncio from cohere import AsyncClient from cohere.finetuning.finetuning import BaseModel, Settings client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) async def main() -> None: await client.finetuning.update_finetuned_model( id="id", name="name", settings=Settings( base_model=BaseModel( base_type="BASE_TYPE_UNSPECIFIED", ), dataset_id="dataset_id", ), ) asyncio.run(main()) """ _response = await self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(id)}", method="PATCH", json={ "name": name, "creator_id": creator_id, "organization_id": organization_id, "settings": convert_and_respect_annotation_metadata( object_=settings, annotation=Settings, direction="write" ), "status": status, "created_at": created_at, "updated_at": updated_at, "completed_at": completed_at, "last_used": last_used, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return typing.cast( UpdateFinetunedModelResponse, construct_type( type_=UpdateFinetunedModelResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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. Returns ------- ListEventsResponse A successful response. Examples -------- import asyncio from cohere import AsyncClient client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) async def main() -> None: await client.finetuning.list_events( finetuned_model_id="finetuned_model_id", ) asyncio.run(main()) """ _response = await self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(finetuned_model_id)}/events", method="GET", params={ "page_size": page_size, "page_token": page_token, "order_by": order_by, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: return typing.cast( ListEventsResponse, construct_type( type_=ListEventsResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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. Returns ------- ListTrainingStepMetricsResponse A successful response. Examples -------- import asyncio from cohere import AsyncClient client = AsyncClient( client_name="YOUR_CLIENT_NAME", token="YOUR_TOKEN", ) async def main() -> None: await client.finetuning.list_training_step_metrics( finetuned_model_id="finetuned_model_id", ) asyncio.run(main()) """ _response = await self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(finetuned_model_id)}/training-step-metrics", method="GET", params={ "page_size": page_size, "page_token": page_token, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: return typing.cast( ListTrainingStepMetricsResponse, construct_type( type_=ListTrainingStepMetricsResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 401: raise UnauthorizedError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 403: raise ForbiddenError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 404: raise NotFoundError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 500: raise InternalServerError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 503: raise ServiceUnavailableError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) _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)