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 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) -> adaptive_sdk.graphql_client.DescribeDatasetDataset | None:

Get details for dataset.

Arguments:
  • key: Dataset key.
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_config: adaptive_sdk.input_types.SampleConfigInput, models: List[str], judge_model: str, method: Literal['custom', 'answer_relevancy', 'context_relevancy', 'faithfulness'], 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_config: Input data configuration.
  • models: Models to evaluate.
  • judge_model: Model key of judge.
  • method: Eval method (built in method, or custom eval).
  • custom_eval_config: Only required if method=="custom".
  • name: Optional name for evaluation job.
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: dict[str, str] | None = 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 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-2-2b'], 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, config: adaptive_sdk.input_types.AdaptRequestConfigInput, 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.
  • config: Training config.
  • name: Name for training job.
def cancel(self, job_id: str) -> str:

Cancel ongoing training job.

def list( self) -> List[adaptive_sdk.graphql_client.ListTrainingJobsTrainingJobs]:

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 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) -> adaptive_sdk.graphql_client.DescribeDatasetDataset | None:

Get details for dataset.

Arguments:
  • key: Dataset key.
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_config: adaptive_sdk.input_types.SampleConfigInput, models: List[str], judge_model: str, method: Literal['custom', 'answer_relevancy', 'context_relevancy', 'faithfulness'], 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_config: Input data configuration.
  • models: Models to evaluate.
  • judge_model: Model key of judge.
  • method: Eval method (built in method, or custom eval).
  • custom_eval_config: Configuration for custom eval. Only required if method=="custom".
  • name: Optional name for evaluation job.
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: dict[str, 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.
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 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: 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, config: adaptive_sdk.input_types.AdaptRequestConfigInput, 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.
  • config: Training config.
  • name: Name for training job.
async def cancel(self, job_id: str) -> str:

Cancel ongoing training job.

async def list( self) -> List[adaptive_sdk.graphql_client.ListTrainingJobsTrainingJobs]:

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