1882 lines
68 KiB
Python
1882 lines
68 KiB
Python
# 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)
|