main
1from __future__ import annotations
2
3import json
4from typing import TYPE_CHECKING
5from argparse import ArgumentParser
6
7from ..._utils import get_client, print_model
8from ...._types import Omittable, omit
9from ...._utils import is_given
10from ..._models import BaseModel
11from ....pagination import SyncCursorPage
12from ....types.fine_tuning import (
13 FineTuningJob,
14 FineTuningJobEvent,
15)
16
17if TYPE_CHECKING:
18 from argparse import _SubParsersAction
19
20
21def register(subparser: _SubParsersAction[ArgumentParser]) -> None:
22 sub = subparser.add_parser("fine_tuning.jobs.create")
23 sub.add_argument(
24 "-m",
25 "--model",
26 help="The model to fine-tune.",
27 required=True,
28 )
29 sub.add_argument(
30 "-F",
31 "--training-file",
32 help="The training file to fine-tune the model on.",
33 required=True,
34 )
35 sub.add_argument(
36 "-H",
37 "--hyperparameters",
38 help="JSON string of hyperparameters to use for fine-tuning.",
39 type=str,
40 )
41 sub.add_argument(
42 "-s",
43 "--suffix",
44 help="A suffix to add to the fine-tuned model name.",
45 )
46 sub.add_argument(
47 "-V",
48 "--validation-file",
49 help="The validation file to use for fine-tuning.",
50 )
51 sub.set_defaults(func=CLIFineTuningJobs.create, args_model=CLIFineTuningJobsCreateArgs)
52
53 sub = subparser.add_parser("fine_tuning.jobs.retrieve")
54 sub.add_argument(
55 "-i",
56 "--id",
57 help="The ID of the fine-tuning job to retrieve.",
58 required=True,
59 )
60 sub.set_defaults(func=CLIFineTuningJobs.retrieve, args_model=CLIFineTuningJobsRetrieveArgs)
61
62 sub = subparser.add_parser("fine_tuning.jobs.list")
63 sub.add_argument(
64 "-a",
65 "--after",
66 help="Identifier for the last job from the previous pagination request. If provided, only jobs created after this job will be returned.",
67 )
68 sub.add_argument(
69 "-l",
70 "--limit",
71 help="Number of fine-tuning jobs to retrieve.",
72 type=int,
73 )
74 sub.set_defaults(func=CLIFineTuningJobs.list, args_model=CLIFineTuningJobsListArgs)
75
76 sub = subparser.add_parser("fine_tuning.jobs.cancel")
77 sub.add_argument(
78 "-i",
79 "--id",
80 help="The ID of the fine-tuning job to cancel.",
81 required=True,
82 )
83 sub.set_defaults(func=CLIFineTuningJobs.cancel, args_model=CLIFineTuningJobsCancelArgs)
84
85 sub = subparser.add_parser("fine_tuning.jobs.list_events")
86 sub.add_argument(
87 "-i",
88 "--id",
89 help="The ID of the fine-tuning job to list events for.",
90 required=True,
91 )
92 sub.add_argument(
93 "-a",
94 "--after",
95 help="Identifier for the last event from the previous pagination request. If provided, only events created after this event will be returned.",
96 )
97 sub.add_argument(
98 "-l",
99 "--limit",
100 help="Number of fine-tuning job events to retrieve.",
101 type=int,
102 )
103 sub.set_defaults(func=CLIFineTuningJobs.list_events, args_model=CLIFineTuningJobsListEventsArgs)
104
105
106class CLIFineTuningJobsCreateArgs(BaseModel):
107 model: str
108 training_file: str
109 hyperparameters: Omittable[str] = omit
110 suffix: Omittable[str] = omit
111 validation_file: Omittable[str] = omit
112
113
114class CLIFineTuningJobsRetrieveArgs(BaseModel):
115 id: str
116
117
118class CLIFineTuningJobsListArgs(BaseModel):
119 after: Omittable[str] = omit
120 limit: Omittable[int] = omit
121
122
123class CLIFineTuningJobsCancelArgs(BaseModel):
124 id: str
125
126
127class CLIFineTuningJobsListEventsArgs(BaseModel):
128 id: str
129 after: Omittable[str] = omit
130 limit: Omittable[int] = omit
131
132
133class CLIFineTuningJobs:
134 @staticmethod
135 def create(args: CLIFineTuningJobsCreateArgs) -> None:
136 hyperparameters = json.loads(str(args.hyperparameters)) if is_given(args.hyperparameters) else omit
137 fine_tuning_job: FineTuningJob = get_client().fine_tuning.jobs.create(
138 model=args.model,
139 training_file=args.training_file,
140 hyperparameters=hyperparameters,
141 suffix=args.suffix,
142 validation_file=args.validation_file,
143 )
144 print_model(fine_tuning_job)
145
146 @staticmethod
147 def retrieve(args: CLIFineTuningJobsRetrieveArgs) -> None:
148 fine_tuning_job: FineTuningJob = get_client().fine_tuning.jobs.retrieve(fine_tuning_job_id=args.id)
149 print_model(fine_tuning_job)
150
151 @staticmethod
152 def list(args: CLIFineTuningJobsListArgs) -> None:
153 fine_tuning_jobs: SyncCursorPage[FineTuningJob] = get_client().fine_tuning.jobs.list(
154 after=args.after or omit, limit=args.limit or omit
155 )
156 print_model(fine_tuning_jobs)
157
158 @staticmethod
159 def cancel(args: CLIFineTuningJobsCancelArgs) -> None:
160 fine_tuning_job: FineTuningJob = get_client().fine_tuning.jobs.cancel(fine_tuning_job_id=args.id)
161 print_model(fine_tuning_job)
162
163 @staticmethod
164 def list_events(args: CLIFineTuningJobsListEventsArgs) -> None:
165 fine_tuning_job_events: SyncCursorPage[FineTuningJobEvent] = get_client().fine_tuning.jobs.list_events(
166 fine_tuning_job_id=args.id,
167 after=args.after or omit,
168 limit=args.limit or omit,
169 )
170 print_model(fine_tuning_job_events)