adaptive_sdk.resources

class ABTests(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with AB Tests.

ABTests(client: adaptive_sdk.Adaptive)
def create( self, ab_test_key: str, feedback_key: str, models: List[str], traffic_split: float = 1.0, feedback_type: Literal['metric', 'preference'] = 'metric', auto_deploy: bool = False, use_case: str | None = None) -> adaptive_sdk.graphql_client.AbCampaignCreateData:

Creates a new A/B test in the client's use case.

Arguments:
  • ab_test_key: A unique key to identify the AB test.
  • feedback_key: The feedback key against which the AB test will run.
  • models: The models to include in the AB test; they must be attached to the use case.
  • traffic_split: Percentage of production traffic to route to AB test. traffic_split*100 % of inference requests for the use case will be sent randomly to one of the models included in the AB test.
  • feedback_type: What type of feedback to run the AB test on, metric or preference.
  • auto_deploy: If set to True, when the AB test is completed, the winning model automatically gets promoted to the use case default model.
def cancel(self, key: str) -> str:

Cancel an ongoing AB test.

Arguments:
  • key: The AB test key.
def list( self, active: bool | None = None, status: Optional[Literal['warmup', 'in_progress', 'done', 'cancelled']] = None, use_case: str | None = None) -> Sequence[adaptive_sdk.graphql_client.AbCampaignDetailData]:

List the use case AB tests.

Arguments:
  • active: Filter on active or inactive AB tests.
  • status: Filter on one of the possible AB test status.
def get( self, key: str) -> adaptive_sdk.graphql_client.DescribeAbCampaignAbCampaign | None:

Get the details of an AB test.

Arguments:
  • key: The AB test key.
class Chat(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):
@override
Chat(client: adaptive_sdk.Adaptive)
def create( self, messages: List[adaptive_sdk.input_types.ChatMessage], stream: bool | None = None, model: str | None = None, stop: Optional[List[str]] = None, max_tokens: int | None = None, temperature: float | None = None, top_p: float | None = None, stream_include_usage: bool | None = None, session_id: str | uuid.UUID | None = None, use_case: str | None = None, user: str | uuid.UUID | None = None, ab_campaign: str | None = None, n: int | None = None, labels: Optional[Dict[str, str]] = None) -> Union[adaptive_sdk.rest.ChatResponse, Generator[adaptive_sdk.rest.ChatResponseChunk, NoneType, NoneType]]:

Create a chat completion.

Arguments:
  • messages: Input messages, each dict with keys role and content.
  • stream: If True, partial message deltas will be returned. If stream is over, chunk.choices will be None.
  • model: Target model key for inference. If None, the requests will be routed to the use case's default model.
  • stop: Sequences or where the API will stop generating further tokens.
  • max_tokens: Maximum # of tokens allowed to generate.
  • temperature: Sampling temperature.
  • top_p: Threshold for top-p sampling.
  • stream_include_usage: If set, an additional chunk will be streamed with the token usage statistics for the entire request.
  • user: ID of user making request. If not None, will be logged as metadata for the request.
  • ab_campaign: AB test key. If set, request will be guaranteed to count towards AB test results, no matter the configured traffic_split.
  • n: Number of chat completions to generate for each input messages.
  • labels: Key-value pairs of interaction labels.

Examples:

# streaming chat request
stream_response = client.chat.create(
    model="model_key", messages=[{"role": "user", "content": "Hello from SDK"}], stream=True
)

print("Streaming response: ", end="", flush=True)
for chunk in stream_response:
    if chunk.choices:
        content = chunk.choices[0].delta.content
        print(content, end="", flush=True)
class Completions(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):
Completions(client: adaptive_sdk.Adaptive)
def create( self, prompt: str, model: str | None = None, stream: bool | None = None, session_id: str | uuid.UUID | None = None, use_case: str | None = None, user: str | uuid.UUID | None = None, ab_campaign: str | None = None, n: int | None = None, labels: dict[str, str] | None = None) -> adaptive_sdk.rest.GenerateResponse:
class ComputePools(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with compute pools.

ComputePools(client: adaptive_sdk.Adaptive)
def list(self):
class CustomRecipes(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with custom scripts.

CustomRecipes(client: adaptive_sdk.Adaptive)
def list(self) -> Sequence[adaptive_sdk.graphql_client.CustomScriptData]:
def upload( self, file_path: str, custom_script_key: str, name: str | None = None) -> adaptive_sdk.graphql_client.CustomScriptData:
def run_training_recipe( self, model: str, recipe_key: str, output_name: str, dataset: str | None = None, recipe_args: typing.Any | None = None, name: str | None = None, wait: bool = False, use_case: str | None = None, compute_pool: str | None = None):
class Datasets(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with file datasets.

Datasets(client: adaptive_sdk.Adaptive)
def upload( self, file_path: str, dataset_key: str, name: str | None = None, use_case: str | None = None) -> adaptive_sdk.graphql_client.LoadDatasetCreateDataset:

Upload a dataset from a file. File must be jsonl, where each line should match structure in example below.

Arguments:
  • file_path: Path to jsonl file.
  • dataset_key: New dataset key.
  • name: Optional name to render in UI; if None, defaults to same as dataset_key.

Example:

{"messages": [{"role": "system", "content": "<optional system prompt>"}, {"role": "user", "content": "<user content>"}, {"role": "assistant", "content": "<assistant answer>"}], "completion": "hey"}
def list( self, use_case: str | None = None) -> List[adaptive_sdk.graphql_client.ListDatasetsDatasets]:

List previously uploaded datasets.

def get( self, key: str, use_case: str | None = None) -> adaptive_sdk.graphql_client.DatasetData | None:

Get details for dataset.

Arguments:
  • key: Dataset key.
def generate( self, model: str, file_path: str, dataset_key: str, dataset_type: Literal['RAG'] = 'RAG', chunks_per_new_sample: int = 10, new_system_prompt: str | None = None, name: str | None = None, use_case: str | None = None, compute_pool: str | None = None) -> adaptive_sdk.graphql_client.DatasetData:

Generate a dataset from a file.

Arguments:
  • model: Model to use for data generation.
  • file_path: Path to jsonl file.
  • dataset_key: New dataset key.
  • dataset_type: Type of dataset to generate.
  • chunks_per_new_sample: Number of document chunks that will be included in each generated sample's prompt.
  • new_system_prompt: Optional new system prompt to be prepended to all generated samples.
  • name: Optional name to render in UI; if None, defaults to same as dataset_key.
  • use_case: Use case to associate with the dataset.
  • compute_pool: Compute pool to use for generation.
class Embeddings(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with embeddings.

Embeddings(client: adaptive_sdk.Adaptive)
def create( self, input: str, model: str | None = None, encoding_format: Literal['Float', 'Base64'] = 'Float', use_case: str | None = None, user: str | uuid.UUID | None = None) -> adaptive_sdk.rest.EmbeddingsResponseList:

Creates embeddings inference request.

Arguments:
  • input: Input text to embed.
  • model: Target model key for inference. If None, the requests will be routed to the use case's default model. Request will error if default model is not an embedding model.
  • encoding_format: Encoding format of response.
  • user: ID of user making the requests. If not None, will be logged as metadata for the request.
class EvalJobs(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with evaluation jobs.

@override
EvalJobs(client: adaptive_sdk.Adaptive)
client
def create( self, data_source: Literal['DATASET', 'COMPLETIONS'], data_config: adaptive_sdk.input_types.SampleDatasourceCompletions | adaptive_sdk.input_types.SampleDatasourceDataset, models: List[str], judge_model: str, method: Literal['custom', 'answer_relevancy', 'context_relevancy', 'faithfulness'], generation_params_per_model: Optional[List[Tuple[str, adaptive_sdk.input_types.GenerateParameters]]] = None, custom_eval_config: adaptive_sdk.input_types.CustomRecipe | None = None, name: str | None = None, use_case: str | None = None, compute_pool: str | None = None) -> adaptive_sdk.graphql_client.EvaluationJobData:

Create a new evaluation job.

Arguments:
  • data_source: Source of data to evaluate on; either an uploaded dataset, or logged completions.
  • data_config: Input data configuration.
  • models: Models to evaluate.
  • judge_model: Model key of judge.
  • method: Eval method (built in method, or custom eval).
  • generation_params_per_model: Optional list of models to evaluate with their parameters for generation. The list should contain tuples, where the first element is the model key, and the second element is the generation parameters. If a model is not present in the list, the default generation parameters will be used.
  • custom_eval_config: Custom evaluation configuration. Only required if method=="custom".
  • name: Optional name for evaluation job.
  • use_case: Target use case to associate evaluation job with. Overrides client's default use case.
  • compute_pool: Optional compute pool key where evaluation job will run on.
def cancel(self, job_id: str) -> str:
class Evaluation:
Evaluation(client: adaptive_sdk.Adaptive)
jobs: EvalJobs
class Feedback(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with and log feedback.

Feedback(client: adaptive_sdk.Adaptive)
def register_key( self, key: str, kind: Literal['scalar', 'bool'] = 'scalar', scoring_type: Literal['higher_is_better', 'lower_is_better'] = 'higher_is_better', name: str | None = None, description: str | None = None) -> adaptive_sdk.graphql_client.MetricData:

Register a new feedback key. Feedback can be logged against this key once it is created.

Arguments:
  • key: Feedback key.
  • kind: Feedback kind. If "bool", you can log values 0, 1, True or False only. If "scalar", you can log any integer or float value.
  • scoring_type: Indication of what good means for this feeback key; a higher numeric value (or True) , or a lower numeric value (or False).
  • name Human-readable feedback name that will render in the UI. If None, will be the same as key.
  • description: Description of intended purpose or nuances of feedback. Will render in the UI.
def list_keys(self) -> Sequence[adaptive_sdk.graphql_client.MetricDataAdmin]:

List all feedback keys.

def get_key( self, feedback_key: str) -> adaptive_sdk.graphql_client.MetricData | None:

Get the details of a feedback key.

Arguments:
  • feedback_key: The feedback key.

return self._gql_client.describe_metric(input=feedback_key).metric

def log_metric( self, value: bool | float | int, completion_id: str | uuid.UUID, feedback_key: str, user: str | uuid.UUID | None = None, details: str | None = None) -> adaptive_sdk.rest.FeedbackOutput:

Log metric feedback for a single completion, which can be a float, int or bool depending on the kind of feedback_key it is logged against.

Arguments:
  • value: The feedback values.
  • completion_id: The completion_id to attach the feedback to.
  • feedback_key: The feedback key to log against.
  • user: ID of user submitting feedback. If not None, will be logged as metadata for the request.
  • details: Textual details for the feedback. Can be used to provide further context on the feedback value.
def log_preference( self, feedback_key: str, preferred_completion: str | uuid.UUID | adaptive_sdk.input_types.ComparisonCompletion, other_completion: str | uuid.UUID | adaptive_sdk.input_types.ComparisonCompletion, user: str | uuid.UUID | None = None, prompt: str | None = None, messages: Optional[List[Dict[str, str]]] = None, tied: Optional[Literal['good', 'bad']] = None, use_case: str | None = None) -> adaptive_sdk.rest.ComparisonOutput:

Log preference feedback between 2 completions.

Arguments:
  • feedback_key: The feedback key to log against.
  • preferred_completion: Can be a completion_id or a dict with keys model and text, corresponding the a valid model key and its attributed completion.
  • other_completion: Can be a completion_id or a dict with keys model and text, corresponding the a valid model key and its attributed completion.
  • user: ID of user submitting feedback.
  • prompt: Input text prompt. Ignored if preferred_ and other_completion are completion_ids.
  • messages: Input chat messages, each dict with keys role and content. Ignored if preferred_ and other_completion are completion_ids.
  • tied: Indicator if both completions tied as equally bad or equally good.
class Interactions(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with interactions.

@override
Interactions(client: adaptive_sdk.Adaptive)
def create( self, completion: str, model: str | None = None, prompt: str | None = None, messages: Optional[List[adaptive_sdk.input_types.ChatMessage]] = None, feedbacks: Optional[List[adaptive_sdk.input_types.InteractionFeedbackDict]] = None, user: str | uuid.UUID | None = None, session_id: str | uuid.UUID | None = None, use_case: str | None = None, ab_campaign: str | None = None, labels: Optional[Dict[str, str]] = None, created_at: str | None = None) -> adaptive_sdk.rest.AddInteractionsResponse:

Create/log an interaction.

Arguments:
  • model: Model key.
  • completion: Model completion.
  • prompt: Input text prompt.
  • messages: Input chat messages, each dict should have keys role and content.
  • feedbacks: List of feedbacks, each dict should with keys feedback_key, value and optional(details).
  • user: ID of user making the request. If not None, will be logged as metadata for the interaction.
  • ab_campaign: AB test key. If set, provided feedbacks will count towards AB test results.
  • labels: Key-value pairs of interaction labels.
  • created_at: Timestamp of interaction creation or ingestion.
def list( self, order: Optional[List[adaptive_sdk.input_types.Order]] = None, filters: adaptive_sdk.input_types.ListCompletionsFilterInput | None = None, page: adaptive_sdk.input_types.CursorPageInput | None = None, group_by: Optional[Literal['model', 'prompt']] = None, use_case: str | None = None) -> adaptive_sdk.graphql_client.ListInteractionsCompletions | adaptive_sdk.graphql_client.ListGroupedInteractionsCompletionsGrouped:

List interactions in client's use case.

Arguments:
  • order: Ordering of results.
  • filters: List filters.
  • page: Paging config.
  • group_by: Retrieve interactions grouped by selected dimension.
def get( self, completion_id: str, use_case: str | None = None) -> adaptive_sdk.graphql_client.CompletionData | None:

Get the details for one specific interaction.

Arguments:
  • completion_id: The ID of the completion.
class Judges(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with Judge definitions used to evaluate model completions.

Judges(client: adaptive_sdk.Adaptive)
def create( self, *, criteria: str, judge_model: str, key: str, examples: Optional[List[adaptive_sdk.input_types.JudgeExampleInput]] = None, name: str | None = None, feedback_key: str | None = None, use_case: str | None = None) -> adaptive_sdk.graphql_client.JudgeData:

Create a new custom Judge.

Arguments:
  • criteria: Natural-language explanation of what should be verified in the completion to be considered a pass for this judge.
  • judge_model: Model key of the judge model.
  • key: Unique key for the judge.
  • examples: List of annotated examples used in few-shot prompting for the judge. Each example is a dict with: • input: list of {"role": str, "content": str} messages • output: the assistant response • passes: bool indicating if the output meets the criteria • reasoning (optional): explanation of the decision
  • name: Human-readable judge name; if omitted, the key will be used.
  • feedback_key: Optional feedback key this judge will write feedback to. If omitted, the judge key will be used.
  • use_case: Explicit use-case key. Falls back to client.default_use_case when omitted.
def update( self, *, key: str, name: str | None = None, criteria: str | None = None, examples: Optional[List[adaptive_sdk.input_types.JudgeExampleInput]] = None, judge_model: str | None = None, use_case: str | None = None) -> adaptive_sdk.graphql_client.JudgeData:

Update an existing Judge version. Any field set to None will be left unchanged.

def delete(self, *, key: str, use_case: str | None = None) -> bool:

Delete a Judge. Returns True on success.

def list( self, *, use_case: str | None = None) -> Sequence[adaptive_sdk.graphql_client.JudgeData]:

List all Judges for the given use case.

def list_versions( self, *, key: str, use_case: str | None = None) -> Sequence[adaptive_sdk.graphql_client.JudgeData]:

List all historical versions of a Judge key.

def get( self, *, key: str, version: int | None = None, use_case: str | None = None) -> adaptive_sdk.graphql_client.JudgeData | None:

Retrieve a specific Judge by key (optionally specifying version).

class Models(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with models.

Models(client: adaptive_sdk.Adaptive)
def add_hf_model( self, hf_model_id: Literal['google/gemma-3-4b-it', 'google/gemma-3-12b-it', 'google/gemma-3-27b-it', 'meta-llama/Llama-3.1-8B-Instruct', 'meta-llama/Llama-3.1-70B-Instruct', 'meta-llama/Llama-3.2-1B-Instruct', 'meta-llama/Llama-3.2-3B-Instruct', 'meta-llama/Llama-3.3-70B-Instruct', 'mistralai/Mistral-Small-24B-Instruct-2501', 'Qwen/Qwen2.5-0.5B-Instruct', 'Qwen/Qwen2.5-1.5B-Instruct', 'Qwen/Qwen2.5-3B-Instruct', 'Qwen/Qwen2.5-7B-Instruct', 'Qwen/Qwen2.5-14B-Instruct', 'Qwen/Qwen2.5-32B-Instruct', 'Qwen/Qwen2.5-72B-Instruct', 'Qwen/Qwen2.5-Coder-0.5B-Instruct', 'Qwen/Qwen2.5-Coder-1.5B-Instruct', 'Qwen/Qwen2.5-Coder-3B-Instruct', 'Qwen/Qwen2.5-Coder-7B-Instruct', 'Qwen/Qwen2.5-Coder-14B-Instruct', 'Qwen/Qwen2.5-Coder-32B-Instruct', 'Qwen/Qwen3-0.6B', 'Qwen/Qwen3-1.7B', 'Qwen/Qwen3-4B', 'Qwen/Qwen3-8B', 'Qwen/Qwen3-14B', 'Qwen/Qwen3-32B'], output_model_name: str, output_model_key: str, hf_token: str, compute_pool: str | None = None) -> str:

Add model from the HuggingFace Model hub to Adaptive model registry. It will take several minutes for the model to be downloaded and converted to Adaptive format.

Arguments:
  • hf_model_id: The ID of the selected model repo on HuggingFace Model Hub.
  • output_model_key: The key that will identify the new model in Adaptive.
  • hf_token: Your HuggingFace Token, needed to validate access to gated/restricted model.
def add_external( self, name: str, external_model_id: str, api_key: str, provider: Literal['open_ai', 'google', 'azure'], endpoint: str | None = None) -> adaptive_sdk.graphql_client.ModelData:

Add proprietary external model to Adaptive model registry.

Arguments:
  • name: Adaptive name for the new model.
  • external_model_id: Should match the model id publicly shared by the model provider.
  • api_key: API Key for authentication against external model provider.
  • provider: External proprietary model provider.
def list( self, filter: adaptive_sdk.input_types.ModelFilter | None = None) -> Sequence[adaptive_sdk.graphql_client.ListModelsModels]:

List all models in Adaptive model registry.

def get(self, model) -> adaptive_sdk.graphql_client.ModelData | None:

Get the details for a model.

Arguments:
  • model: Model key.
def attach( self, model: str, wait: bool = False, make_default: bool = False, use_case: str | None = None, placement: adaptive_sdk.input_types.ModelPlacementInput | None = None) -> adaptive_sdk.graphql_client.ModelServiceData:

Attach a model to the client's use case.

Arguments:
  • model: Model key.
  • wait: If the model is not deployed already, attaching it to the use case will automatically deploy it. If True, this call blocks until model is Online.
  • make_default: Make the model the use case's default on attachment.
def detach( self, model: str, use_case: str | None = None) -> adaptive_sdk.graphql_client.ModelServiceData:

Detach model from client's use case.

Arguments:
  • model: Model key.
def update_compute_config( self, model: str, compute_config: adaptive_sdk.input_types.ModelComputeConfigInput) -> adaptive_sdk.graphql_client.ModelData:

Update compute config of model.

def update( self, model: str, is_default: bool | None = None, attached: bool | None = None, desired_online: bool | None = None, use_case: str | None = None, placement: adaptive_sdk.input_types.ModelPlacementInput | None = None) -> adaptive_sdk.graphql_client.ModelServiceData:

Update config of model attached to client's use case.

Arguments:
  • model: Model key.
  • is_default: Change the selection of the model as default for the use case. True to promote to default, False to demote from default. If None, no changes are applied.
  • attached: Whether model should be attached or detached to/from use case. If None, no changes are applied.
  • desired_online: Turn model inference on or off for the client use case. This does not influence the global status of the model, it is use case-bounded. If None, no changes are applied.
def deploy(self, model: str, wait: bool = False) -> str:

Deploy a model, loading it to memory and making it ready for inference.

Arguments:
  • model: Model key.
  • wait: If True, call block until model is in Online state.
def terminate(self, model: str, force: bool = False) -> str:

Terminate model, removing it from memory and making it unavailable to all use cases.

Arguments:
  • model: Model key.
  • force: If model is attached to several use cases, force must equal True in order for the model to be terminated.
class Permissions(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to list permissions.

Permissions(client: adaptive_sdk.Adaptive)
def list(self) -> List[str]:
class RewardServers(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with external reward servers.

RewardServers(client: adaptive_sdk.Adaptive)
def list(self) -> Sequence[adaptive_sdk.graphql_client.RemoteEnvData]:
def add( self, url: str, key: str, name: str | None = None, description: str | None = None) -> adaptive_sdk.graphql_client.RemoteEnvData:
def remove(self, key: str) -> str:
class Roles(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to manage roles.

Roles(client: adaptive_sdk.Adaptive)
def create( self, key: str, permissions: List[str], name: str | None = None) -> adaptive_sdk.graphql_client.CreateRoleCreateRole:

Creates new role.

Arguments:
  • key: Role key.
  • permissions: List of permission identifiers such as use_case:read. You can list all possible permissions with client.permissions.list().
  • name: Role name; if not provided, defaults to key.
class Teams(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to manage teams.

Teams(client: adaptive_sdk.Adaptive)
def create( self, key: str, name: str | None = None) -> adaptive_sdk.graphql_client.CreateTeamCreateTeam:
class Training:
Training(client: adaptive_sdk.Adaptive)
jobs: TrainingJobs
class TrainingJobs(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with training jobs.

TrainingJobs(client: adaptive_sdk.Adaptive)
def create( self, model: str, data_source: Literal['DATASET', 'COMPLETIONS'], data_config: adaptive_sdk.input_types.SampleDatasourceCompletions | adaptive_sdk.input_types.SampleDatasourceDataset, feedback_type: Optional[Literal['DIRECT', 'PREFERENCE']] = None, parameter_efficient: bool = True, alignment_method: Literal['PPO', 'DPO', 'GRPO', 'SFT'] = 'PPO', alignment_objective: adaptive_sdk.input_types.TrainingObjectiveInput | None = None, alignment_params: adaptive_sdk.input_types.TrainingMetadataInputParameters | None = None, base_training_params: adaptive_sdk.input_types.BaseTrainingParamsInput | None = None, job_name: str | None = None, output_model_name: str | None = None, wait: bool = False, use_case: str | None = None, compute_pool: str | None = None) -> adaptive_sdk.graphql_client.CreateTrainingJobCreateTrainingJob:

Create a new training job.

Arguments:
  • model: Model to train.
  • data_source: Source of data to train on; either an uploaded dataset, or logged completions.
  • data_config: Training data configuration.
  • feedback_type: If training on metric training_objective usingcompletions that have both metric and preference feedback logged, you must specify which of those sets to train on.
  • parameter_efficient: If True, training will be parameter-efficient, and the output model will be a lightweight adapter coupled to the selected backbone model.
  • aligment_method: Alignment method selection.
  • alignment_objective: Configuration for the training objective, determines the reward during training.
  • alignment_params: Alignment method-specific hyperparameter configuration.
  • base_training_params: Common training parameters.
  • job_name: Human readable training job name.
  • output_model_name: Name for resulting model.
  • wait: If True, only returns when training is finished.
  • use_case: Target use case to associate training job with.
  • compute_pool: Compute pool where training job will run.
def cancel(self, job_id: str) -> str:

Cancel ongoing training job.

List all training jobs.

def get( self, job_id: str) -> adaptive_sdk.graphql_client.DescribeTrainingJobTrainingJob | None:

Get details for training job.

class UseCase(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with use cases.

UseCase(client: adaptive_sdk.Adaptive)
def create( self, key: str, name: str | None = None, description: str | None = None, team: str | None = None) -> adaptive_sdk.graphql_client.UseCaseData:

Create new use case.

Arguments:
  • key: Use case key.
  • name: Human-readable use case name which will be rendered in the UI. If not set, will be the same as key.
  • description: Description of model which will be rendered in the UI.
def list(self) -> Sequence[adaptive_sdk.graphql_client.UseCaseData]:

List all use cases.

def get( self, use_case: str | None = None) -> adaptive_sdk.graphql_client.UseCaseData | None:

Get details for the client's use case.

class Users(adaptive_sdk.resources.base_resource.SyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to manage users and permissions.

Users(client: adaptive_sdk.Adaptive)
def me(self) -> adaptive_sdk.graphql_client.UserData | None:

Get details of current user.

def list(self) -> Sequence[adaptive_sdk.graphql_client.UserData]:

List all users registered to Adaptive deployment.

def update( self, email: str, team: str, role: str) -> adaptive_sdk.graphql_client.UpdateUserSetTeamMember:

Update team and role for user.

Arguments:
  • email: User email.
  • team: Key of team to which user will be added to.
  • role: Assigned role
class AsyncABTests(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with AB Tests.

AsyncABTests(client: adaptive_sdk.AsyncAdaptive)
async def create( self, ab_test_key: str, feedback_key: str, models: List[str], traffic_split: float = 1.0, feedback_type: Literal['metric', 'preference'] = 'metric', auto_deploy: bool = False, use_case: str | None = None) -> adaptive_sdk.graphql_client.AbCampaignCreateData:

Creates a new A/B test in the client's use case.

Arguments:
  • ab_test_key: A unique key to identify the AB test.
  • feedback_key: The feedback key against which the AB test will run.
  • models: The models to include in the AB test; they must be attached to the use case.
  • traffic_split: Percentage of production traffic to route to AB test. traffic_split*100 % of inference requests for the use case will be sent randomly to one of the models included in the AB test.
  • feedback_type: What type of feedback to run the AB test on, metric (direct) or preference (comparison).
  • auto_deploy: If set to True, when the AB test is completed, the winning model automatically gets promoted to the use case default model.
async def cancel(self, key: str) -> str:

Cancel an ongoing AB test.

Arguments:
  • key: The AB test key.
async def list( self, active: bool | None = None, status: Optional[Literal['warmup', 'in_progress', 'done', 'cancelled']] = None, use_case: str | None = None) -> Sequence[adaptive_sdk.graphql_client.AbCampaignDetailData]:

List the use case AB tests.

Arguments:
  • active: Filter on active or inactive AB tests.
  • status: Filter on one of the possible AB test status.
async def get( self, key: str) -> adaptive_sdk.graphql_client.DescribeAbCampaignAbCampaign | None:

Get the details of an AB test.

Arguments:
  • key: The AB test key.
class AsyncChat(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):
AsyncChat(client: adaptive_sdk.AsyncAdaptive)
async def create( self, messages: List[adaptive_sdk.input_types.ChatMessage], model: str | None = None, stream: bool | None = None, stop: Optional[List[str]] = None, max_tokens: int | None = None, temperature: float | None = None, top_p: float | None = None, stream_include_usage: bool | None = None, session_id: str | uuid.UUID | None = None, use_case: str | None = None, user: str | uuid.UUID | None = None, ab_campaign: str | None = None, n: int | None = None, labels: Optional[Dict[str, str]] = None) -> Union[adaptive_sdk.rest.ChatResponse, AsyncGenerator[adaptive_sdk.rest.ChatResponseChunk, NoneType]]:

Create a chat completion.

Arguments:
  • messages: Input messages, each dict with keys role and content.
  • stream: If True, partial message deltas will be returned.
  • model: Target model key for inference. If None, the requests will be routed to the use case's default model.
  • stop: Sequences or where the API will stop generating further tokens.
  • max_tokens: Maximum # of tokens allowed to generate.
  • temperature: Sampling temperature.
  • top_p: Threshold for top-p sampling.
  • stream_include_usage: If set, an additional chunk will be streamed with the token usaage statistics for the entire request.
  • user: ID of user making request. If not None, will be logged as metadata for the request.
  • ab_campaign: AB test key. If set, request will be guaranteed to count towards AB test results, no matter the configured traffic_split.
  • n: Number of chat completions to generate for each input messages.
  • labels: Key-value pairs of interaction labels.

Examples:

# async streaming chat request
async def async_chat_stream():
    stream_response = aclient.chat.create(
        model="model_key", messages=[{"role": "user", "content": "Hello from SDK"}], stream=True
    )

    print("Async chat streaming response: ", end="", flush=True)
    async for chunk in await stream_response:
        if chunk.choices:
            content = chunk.choices[0].delta.content
            print(content, end="", flush=True)
class AsyncCompletions(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):
@override
AsyncCompletions(client: adaptive_sdk.AsyncAdaptive)
async def create( self, prompt: str, model: str | None = None, stream: bool | None = None, session_id: str | uuid.UUID | None = None, use_case: str | None = None, user: str | uuid.UUID | None = None, ab_campaign: str | None = None, n: int | None = None, labels: dict[str, str] | None = None) -> adaptive_sdk.rest.GenerateResponse:
class AsyncComputePools(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with compute pools.

AsyncComputePools(client: adaptive_sdk.AsyncAdaptive)
async def list(self):
class AsyncCustomRecipes(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with custom scripts.

AsyncCustomRecipes(client: adaptive_sdk.AsyncAdaptive)
async def list(self) -> Sequence[adaptive_sdk.graphql_client.CustomScriptData]:
async def upload( self, file_path: str, custom_script_key: str, name: str | None = None) -> adaptive_sdk.graphql_client.CustomScriptData:
async def run_training_recipe( self, model: str, recipe_key: str, output_name: str, dataset: str | None = None, recipe_args: typing.Any | None = None, name: str | None = None, wait: bool = False, use_case: str | None = None, compute_pool: str | None = None):
class AsyncDatasets(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):
AsyncDatasets(client: adaptive_sdk.AsyncAdaptive)
async def upload( self, file_path: str, dataset_key: str, name: str | None = None, use_case: str | None = None) -> adaptive_sdk.graphql_client.LoadDatasetCreateDataset:

Upload a dataset from a file. File must be jsonl, where each line should match structure in example below.

Arguments:
  • file_path: Path to jsonl file.
  • dataset_key: New dataset key.
  • name: Optional name to render in UI; if None, defaults to same as dataset_key.

Example:

{"messages": [{"role": "system", "content": "<optional system prompt>"}, {"role": "user", "content": "<user content>"}, {"role": "assistant", "content": "<assistant answer>"}], "completion": "hey"}
async def list( self, use_case: str | None = None) -> List[adaptive_sdk.graphql_client.ListDatasetsDatasets]:

List previously uploaded datasets.

async def get( self, key: str, use_case: str | None = None) -> adaptive_sdk.graphql_client.DatasetData | None:

Get details for dataset.

Arguments:
  • key: Dataset key.
async def generate( self, model: str, file_path: str, dataset_key: str, dataset_type: Literal['RAG'] = 'RAG', chunks_per_new_sample: int = 10, new_system_prompt: str | None = None, name: str | None = None, use_case: str | None = None, compute_pool: str | None = None) -> adaptive_sdk.graphql_client.DatasetData:

Generate a dataset from a file.

Arguments:
  • model: Model to use for data generation.
  • file_path: Path to jsonl file.
  • dataset_key: New dataset key.
  • dataset_type: Type of dataset to generate.
  • chunks_per_new_sample: Number of document chunks that will be included in each generated sample's prompt.
  • new_system_prompt: Optional new system prompt to be prepended to all generated samples.
  • name: Optional name to render in UI; if None, defaults to same as dataset_key.
  • use_case: Use case to associate with the dataset.
  • compute_pool: Compute pool to use for generation.
class AsyncEmbeddings(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with embeddings.

AsyncEmbeddings(client: adaptive_sdk.AsyncAdaptive)
async def create( self, input: str, model_key: str | None = None, encoding_format: Literal['Float', 'Base64'] = 'Float', use_case: str | None = None, user: str | uuid.UUID | None = None) -> adaptive_sdk.rest.EmbeddingsResponseList:

Creates embeddings inference request.

Arguments:
  • input: Input text to embed.
  • model: Target model key for inference. If None, the requests will be routed to the use case's default model. Request will error if default model is not an embedding model.
  • encoding_format: Encoding format of response.
  • user: ID of user making the requests. If not None, will be logged as metadata for the request.
class AsyncEvalJobs(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):
AsyncEvalJobs(client: adaptive_sdk.AsyncAdaptive)
client
async def create( self, data_source: Literal['DATASET', 'COMPLETIONS'], data_config: adaptive_sdk.input_types.SampleDatasourceCompletions | adaptive_sdk.input_types.SampleDatasourceDataset, models: List[str], judge_model: str, method: Literal['custom', 'answer_relevancy', 'context_relevancy', 'faithfulness'], generation_params_per_model: Optional[List[Tuple[str, adaptive_sdk.input_types.GenerateParameters]]] = None, custom_eval_config: adaptive_sdk.input_types.CustomRecipe | None = None, name: str | None = None, use_case: str | None = None, compute_pool: str | None = None) -> adaptive_sdk.graphql_client.EvaluationJobData:

Create a new evaluation job.

Arguments:
  • data_source: Source of data to evaluate on; either an uploaded dataset, or logged completions.
  • data_config: Input data configuration.
  • models: Models to evaluate.
  • judge_model: Model key of judge.
  • method: Eval method (built in method, or custom eval).
  • generation_params_per_model: Optional list of models to evaluate with their parameters for generation. The list should contain tuples, where the first element is the model key, and the second element is the generation parameters. If a model is not present in the list, the default generation parameters will be used.
  • custom_eval_config: Custom evaluation configuration. Only required if method=="custom".
  • name: Optional name for evaluation job.
  • use_case: Target use case to associate evaluation job with. Overrides client's default use case.
  • compute_pool: Optional compute pool key where evaluation job will run on.
async def cancel(self, job_id: str) -> str:
async def get( self, job_id: str) -> adaptive_sdk.graphql_client.DescribeEvaluationJobEvaluationJob | None:
class AsyncEvaluation:
AsyncEvaluation(client: adaptive_sdk.AsyncAdaptive)
class AsyncFeedback(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with and log feedback.

AsyncFeedback(client: adaptive_sdk.AsyncAdaptive)
async def register_key( self, key: str, kind: Literal['scalar', 'bool'], scoring_type: Literal['higher_is_better', 'lower_is_better'] = 'higher_is_better', name: str | None = None, description: str | None = None) -> adaptive_sdk.graphql_client.MetricData:

Register a new feedback key. Feedback can be logged against this key once it is created.

Arguments:
  • key: Feedback key.
  • kind: Feedback kind. If "bool", you can log values 0, 1, True or False only. If "scalar", you can log any integer or float value.
  • scoring_type: Indication of what good means for this feeback key; a higher numeric value (or True) , or a lower numeric value (or False).
  • name: Human-readable feedback name that will render in the UI. If None, will be the same as key.
  • description: Description of intended purpose or nuances of feedback. Will render in the UI.
async def list_keys(self) -> Sequence[adaptive_sdk.graphql_client.MetricDataAdmin]:

List all feedback keys.

async def get_key( self, feedback_key: str) -> adaptive_sdk.graphql_client.MetricData | None:

Get the details of a feedback key.

Arguments:
  • feedback_key: The feedback key.
async def log_metric( self, value: bool | float | int, completion_id: str | uuid.UUID, feedback_key: str, user_id: str | uuid.UUID | None = None, details: str | None = None) -> adaptive_sdk.rest.FeedbackOutput:

Log metric feedback for a single completion, which can be a float, int or bool depending on the kind of feedback_key it is logged against.

Arguments:
  • value: The feedback values.
  • completion_id: The completion_id to attach the feedback to.
  • feedback_key: The feedback key to log against.
  • user: ID of user submitting feedback. If not None, will be logged as metadata for the request.
  • details: Textual details for the feedback. Can be used to provide further context on the feedback value.
async def log_preference( self, feedback_key: str, preferred_completion: str | uuid.UUID | adaptive_sdk.input_types.ComparisonCompletion, other_completion: str | uuid.UUID | adaptive_sdk.input_types.ComparisonCompletion, user_id: str | uuid.UUID | None = None, prompt: str | None = None, messages: Optional[List[Dict[str, str]]] = None, tied: Optional[Literal['good', 'bad']] = None, use_case: str | None = None) -> adaptive_sdk.rest.ComparisonOutput:

Log preference feedback between 2 completions.

Arguments:
  • feedback_key: The feedback key to log against.
  • preferred_completion: Can be a completion_id or a dict with keys model and text, corresponding the a valid model key and its attributed completion.
  • other_completion: Can be a completion_id or a dict with keys model and text, corresponding the a valid model key and its attributed completion.
  • user: ID of user submitting feedback.
  • prompt: Input text prompt. Ignored if preferred_ and other_completion are completion_ids.
  • messages: Input chat messages, each dict with keys role and content. Ignored if preferred_ and other_completion are completion_ids.
  • tied: Indicator if both completions tied as equally bad or equally good.
class AsyncInteractions(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with interactions.

AsyncInteractions(client: adaptive_sdk.AsyncAdaptive)
async def create( self, completion: str, model: str | None = None, prompt: str | None = None, messages: Optional[List[adaptive_sdk.input_types.ChatMessage]] = None, feedbacks: Optional[List[adaptive_sdk.input_types.InteractionFeedbackDict]] = None, user: str | uuid.UUID | None = None, session_id: str | uuid.UUID | None = None, use_case: str | None = None, ab_campaign: str | None = None, labels: Optional[Dict[str, str]] = None) -> adaptive_sdk.rest.AddInteractionsResponse:

Create/log an interaction.

Arguments:
  • model: Model key.
  • completion: Model completion.
  • prompt: Input text prompt.
  • messages: Input chat messages, each dict should have keys role and content.
  • feedbacks: List of feedbacks, each dict should with keys feedback_key, value and optional(details).
  • user: ID of user making the request. If not None, will be logged as metadata for the interaction.
  • ab_campaign: AB test key. If set, provided feedbacks will count towards AB test results.
  • labels: Key-value pairs of interaction labels.
  • created_at: Timestamp of interaction creation or ingestion.
async def list( self, order: Optional[List[adaptive_sdk.input_types.Order]] = None, filters: adaptive_sdk.input_types.ListCompletionsFilterInput | None = None, page: adaptive_sdk.input_types.CursorPageInput | None = None, group_by: Optional[Literal['model', 'prompt']] = None, use_case: str | None = None) -> adaptive_sdk.graphql_client.ListInteractionsCompletions | adaptive_sdk.graphql_client.ListGroupedInteractionsCompletionsGrouped:

List interactions in client's use case.

Arguments:
  • order: Ordering of results.
  • filters: List filters.
  • page: Paging config.
  • group_by: Retrieve interactions grouped by selected dimension.
async def get( self, completion_id: str, use_case: str | None = None) -> adaptive_sdk.graphql_client.CompletionData | None:

Get the details for one specific interaction.

Arguments:
  • completion_id: The ID of the completion.
class AsyncJudges(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Asynchronous resource to interact with Judge definitions used to evaluate model completions.

AsyncJudges(client: adaptive_sdk.AsyncAdaptive)
async def create( self, *, criteria: str, judge_model: str, key: str, examples: Optional[List[adaptive_sdk.input_types.JudgeExampleInput]] = None, name: str | None = None, feedback_key: str | None = None, use_case: str | None = None) -> adaptive_sdk.graphql_client.JudgeData:
async def update( self, *, key: str, name: str | None = None, criteria: str | None = None, examples: Optional[List[adaptive_sdk.input_types.JudgeExampleInput]] = None, judge_model: str | None = None, use_case: str | None = None) -> adaptive_sdk.graphql_client.JudgeData:
async def delete(self, *, key: str, use_case: str | None = None) -> bool:
async def list( self, *, use_case: str | None = None) -> Sequence[adaptive_sdk.graphql_client.JudgeData]:
async def list_versions( self, *, key: str, use_case: str | None = None) -> Sequence[adaptive_sdk.graphql_client.JudgeData]:
async def get( self, *, key: str, version: int | None = None, use_case: str | None = None) -> adaptive_sdk.graphql_client.JudgeData | None:
class AsyncModels(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with models.

AsyncModels(client: adaptive_sdk.AsyncAdaptive)
async def add_hf_model( self, hf_model_id: Literal['google/gemma-3-4b-it', 'google/gemma-3-12b-it', 'google/gemma-3-27b-it', 'meta-llama/Llama-3.1-8B-Instruct', 'meta-llama/Llama-3.1-70B-Instruct', 'meta-llama/Llama-3.2-1B-Instruct', 'meta-llama/Llama-3.2-3B-Instruct', 'meta-llama/Llama-3.3-70B-Instruct', 'mistralai/Mistral-Small-24B-Instruct-2501', 'Qwen/Qwen2.5-0.5B-Instruct', 'Qwen/Qwen2.5-1.5B-Instruct', 'Qwen/Qwen2.5-3B-Instruct', 'Qwen/Qwen2.5-7B-Instruct', 'Qwen/Qwen2.5-14B-Instruct', 'Qwen/Qwen2.5-32B-Instruct', 'Qwen/Qwen2.5-72B-Instruct', 'Qwen/Qwen2.5-Coder-0.5B-Instruct', 'Qwen/Qwen2.5-Coder-1.5B-Instruct', 'Qwen/Qwen2.5-Coder-3B-Instruct', 'Qwen/Qwen2.5-Coder-7B-Instruct', 'Qwen/Qwen2.5-Coder-14B-Instruct', 'Qwen/Qwen2.5-Coder-32B-Instruct', 'Qwen/Qwen3-0.6B', 'Qwen/Qwen3-1.7B', 'Qwen/Qwen3-4B', 'Qwen/Qwen3-8B', 'Qwen/Qwen3-14B', 'Qwen/Qwen3-32B'], output_model_name: str, output_model_key: str, hf_token: str, compute_pool: str | None = None):

Add model from the HuggingFace Model hub to Adaptive model registry. It will take several minutes for the model to be downloaded and converted to Adaptive format.

Arguments:
  • hf_model_id: The ID of the selected model repo on HuggingFace Model Hub.
  • output_model_key: The key that will identify the new model in Adaptive.
  • hf_token: Your HuggingFace Token, needed to validate access to gated/restricted model.
async def add_external( self, name: str, external_model_id: str, api_key: str, provider: Literal['open_ai', 'google', 'azure'], endpoint: str | None = None) -> adaptive_sdk.graphql_client.ModelData:

Add proprietary external model to Adaptive model registry.

Arguments:
  • name: Adaptive name for the new model.
  • external_model_id: Should match the model id publicly shared by the model provider.
  • api_key: API Key for authentication against external model provider.
  • provider: External proprietary model provider.
async def list( self, filter: adaptive_sdk.input_types.ModelFilter | None = None) -> Sequence[adaptive_sdk.graphql_client.ListModelsModels]:

List all models in Adaptive model registry.

async def get(self, model) -> adaptive_sdk.graphql_client.ModelData | None:

Get the details for a model.

Arguments:
  • model: Model key.
async def attach( self, model: str, wait: bool = True, make_default: bool = False, use_case: str | None = None, placement: adaptive_sdk.input_types.ModelPlacementInput | None = None) -> adaptive_sdk.graphql_client.ModelServiceData:

Attach a model to the client's use case.

Arguments:
  • model: Model key.
  • wait: If the model is not deployed already, attaching it to the use case will automatically deploy it. If True, this call blocks until model is Online.
  • make_default: Make the model the use case's default on attachment.
async def detach( self, model: str, use_case: str | None = None) -> adaptive_sdk.graphql_client.ModelServiceData:

Detach model from client's use case.

Arguments:
  • model: Model key.
async def update_compute_config( self, model: str, compute_config: adaptive_sdk.input_types.ModelComputeConfigInput) -> adaptive_sdk.graphql_client.ModelData:

Update compute config of model.

async def update( self, model: str, is_default: bool | None = None, attached: bool | None = None, desired_online: bool | None = None, use_case: str | None = None, placement: adaptive_sdk.input_types.ModelPlacementInput | None = None) -> adaptive_sdk.graphql_client.ModelServiceData:

Update config of model attached to client's use case.

Arguments:
  • model: Model key.
  • is_default: Change the selection of the model as default for the use case. True to promote to default, False to demote from default. If None, no changes are applied.
  • attached: Whether model should be attached or detached to/from use case. If None, no changes are applied.
  • desired_online: Turn model inference on or off for the client use case. This does not influence the global status of the model, it is use case-bounded. If None, no changes are applied.
async def deploy(self, model: str, wait: bool = False) -> str:

Deploy a model, loading it to memory and making it ready for inference.

Arguments:
  • model: Model key.
  • wait: If True, call block until model is in Online state.
async def terminate(self, model: str, force: bool = False) -> str:

Terminate model, removing it from memory and making it unavailable to all use cases.

Arguments:
  • model: Model key.
  • force: If model is attached to several use cases, force must equal True in order for the model to be terminated.
class AsyncPermissions(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to list permissions.

AsyncPermissions(client: adaptive_sdk.AsyncAdaptive)
async def list(self) -> List[str]:
class AsyncRewardServers(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Async resource to interact with external rewards servers.

AsyncRewardServers(client: adaptive_sdk.AsyncAdaptive)
async def list(self) -> Sequence[adaptive_sdk.graphql_client.RemoteEnvData]:
async def add( self, url: str, key: str, name: str | None = None, description: str | None = None) -> adaptive_sdk.graphql_client.RemoteEnvData:
async def remove(self, key: str) -> str:
class AsyncRoles(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to manage roles.

AsyncRoles(client: adaptive_sdk.AsyncAdaptive)
async def list(self) -> List[adaptive_sdk.graphql_client.ListRolesRoles]:
async def create( self, key: str, permissions: List[str], name: str | None = None) -> adaptive_sdk.graphql_client.CreateRoleCreateRole:

Creates new role.

Arguments:
  • key: Role key.
  • permissions: List of permission identifiers such as use_case:read. You can list all possible permissions with client.permissions.list().
  • name: Role name; if not provided, defaults to key.
class AsyncTeams(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to manage teams.

AsyncTeams(client: adaptive_sdk.AsyncAdaptive)
async def list(self) -> List[adaptive_sdk.graphql_client.ListTeamsTeams]:
async def create( self, key: str, name: str | None = None) -> adaptive_sdk.graphql_client.CreateTeamCreateTeam:
class AsyncTraining:
AsyncTraining(client: adaptive_sdk.AsyncAdaptive)
class AsyncTrainingJobs(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with training jobs.

AsyncTrainingJobs(client: adaptive_sdk.AsyncAdaptive)
async def create( self, model: str, data_source: Literal['DATASET', 'COMPLETIONS'], data_config: adaptive_sdk.input_types.SampleDatasourceCompletions | adaptive_sdk.input_types.SampleDatasourceDataset, feedback_type: Optional[Literal['DIRECT', 'PREFERENCE']] = None, parameter_efficient: bool = True, alignment_method: Literal['PPO', 'DPO', 'GRPO', 'SFT'] = 'PPO', alignment_objective: adaptive_sdk.input_types.TrainingObjectiveInput | None = None, alignment_params: adaptive_sdk.input_types.TrainingMetadataInputParameters | None = None, base_training_params: adaptive_sdk.input_types.BaseTrainingParamsInput | None = None, job_name: str | None = None, output_model_name: str | None = None, wait: bool = False, use_case: str | None = None, compute_pool: str | None = None) -> adaptive_sdk.graphql_client.CreateTrainingJobCreateTrainingJob:

Create a new training job.

Arguments:
  • model: Model to train.
  • data_source: Source of data to train on; either an uploaded dataset, or logged completions.
  • data_config: Training data configuration.
  • feedback_type: If training on metric training_objective usingcompletions that have both metric and preference feedback logged, you must specify which of those sets to train on.
  • parameter_efficient: If True, training will be parameter-efficient, and the output model will be a lightweight adapter coupled to the selected backbone model.
  • aligment_method: Alignment method selection.
  • alignment_objective: Configuration for the training objective, determines the reward during training.
  • alignment_params: Alignment method-specific hyperparameter configuration.
  • base_training_params: Common training parameters.
  • job_name: Human readable training job name.
  • output_model_name: Name for resulting model.
  • wait: If True, only returns when training is finished.
  • use_case: Target use case to associate training job with.
  • compute_pool: Compute pool where training job will run.
async def cancel(self, job_id: str) -> str:

Cancel ongoing training job.

List all training jobs.

async def get( self, job_id: str) -> adaptive_sdk.graphql_client.DescribeTrainingJobTrainingJob | None:

Get details for training job.

class AsyncUseCase(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to interact with use cases.

AsyncUseCase(client: adaptive_sdk.AsyncAdaptive)
async def create( self, key: str, name: str | None = None, description: str | None = None, team: str | None = None, default_feedback_key: str | None = None) -> adaptive_sdk.graphql_client.UseCaseData:

Create new use case.

Arguments:
  • key: Use case key.
  • name: Human-readable use case name which will be rendered in the UI. If not set, will be the same as key.
  • description: Description of model which will be rendered in the UI.
async def list(self) -> Sequence[adaptive_sdk.graphql_client.UseCaseData]:

List all use cases.

async def get( self, use_case: str | None = None) -> adaptive_sdk.graphql_client.UseCaseData | None:

Get details for the client's use case.

class AsyncUsers(adaptive_sdk.resources.base_resource.AsyncAPIResource, adaptive_sdk.resources.base_resource.UseCaseResource):

Resource to manage users and permissions.

AsyncUsers(client: adaptive_sdk.AsyncAdaptive)
async def me(self) -> adaptive_sdk.graphql_client.UserData | None:

Get details of current user.

async def list(self) -> Sequence[adaptive_sdk.graphql_client.UserData]:

List all users registered to Adaptive deployment.

async def update( self, email: str, team: str, role: str) -> adaptive_sdk.graphql_client.UpdateUserSetTeamMember:

Update team and role for user.

Arguments:
  • email: User email.
  • team: Key of team to which user will be added to.
  • role: Assigned role