qianfan.model package

class qianfan.model.DeployConfig(*, name: str = '', endpoint_prefix: str = '', description: str = '', replicas: int = 1, pool_type: DeployPoolType = DeployPoolType.PrivateResource, service_type: ServiceType, extras: Any = None)

Bases: BaseModel

description: str

description of service

endpoint_prefix: str

Endpoint custom prefix, will be used to call resource api

extras: Any

name: str

Service name

pool_type: DeployPoolType

resource pool type, public resource will be shared with others.

replicas: int

replicas for model services, related to the capacity in QPS of model service.

default set to 1

service_type: ServiceType

service type, after deploy, Service could behave like the specific type.

class qianfan.model.Model(id: Optional[str] = None, version_id: Optional[str] = None, task_id: Optional[str] = None, job_id: Optional[str] = None, name: Optional[str] = None, **kwargs: Any)

Bases: ExecuteSerializable[Dict, Union[QfResponse, Iterator[QfResponse]]]

auto_complete_info(**kwargs: Any) → None

auto complete Model object’s info. This may override the input model id version id.

Parameters:

**kwargs (Any):

arbitrary arguments

batch_inference(dataset: Dataset, **kwargs: Any) → Dataset

create batch run using specific dataset on qianfan by evaluation ability of platform

Parameters:

dataset (Dataset):

A dataset instance which indicates a dataset on qianfan platform

**kwargs (Any):

Arbitrary keyword arguments

Returns:

Dataset: batch result contained in dataset

deploy(deploy_config: DeployConfig, **kwargs: Any) → Service

model deploy

Parameters:

deploy_config (DeployConfig):

model service deploy config

Returns:

Service: model service instance

dumps() → Optional[bytes]

Serialize the model to bytes.

Returns:

Optional[bytes]:

bytes of this model

exec(input: Optional[Dict] = None, **kwargs: Dict) → Union[QfResponse, Iterator[QfResponse]]

model execution, for different model service type, please input a dict with different keys. Concretely, take

`input={“messages”: [{“role”: “user”,

“content”: “hello world”}]}`

as input, when the model is a chat io Model.

Parameters:

input (Optional[Dict], optional):

input data . Defaults to None.

Raises:

InternalError: model with no service deployed is unable to call exec

Returns:

Union[QfResponse, Iterator[QfResponse]]:

output data

id: Optional[str]

remote model id

job_id: Optional[str]

train job id

loads(data: bytes) → Any

load model instance from bytes

Parameters:

data (bytes):

bytes of this model

Returns:

Any: model instance

name: Optional[str] = None

model name

old_id: Optional[int]

deprecated old model id

old_version_id: Optional[int]

deprecated old model version id

publish(name: str = '', **kwargs: Any) → Model

model publish, before deploying a model, it should be published.

Parameters:

name str:

model name. Defaults to “m_{task_id}{job_id}”.

service: Optional[Service] = None

model service

task_id: Optional[str]

train tkas id

version_id: Optional[str]

remote model version id

class qianfan.model.Service(id: Optional[int] = None, endpoint: Optional[str] = None, model: Optional[Union[Model, str]] = None, deploy_config: Optional[DeployConfig] = None, service_type: Optional[ServiceType] = None)

Bases: ExecuteSerializable[Dict, Union[QfResponse, Iterator[QfResponse]]]

batch_inference(dataset: Dataset, prompt_template: Optional[Prompt] = None, system_prompt: Optional[str] = None, **kwargs: Any) → Dataset

create batch run using specific dataset on qianfan

Args:

dataset (Dataset):

A dataset instance which indicates a dataset on qianfan platform

prompt_template (Optional[Prompt]):

Optional Prompt used as input of llm, default to None. Only used when your Service is a Completion service

system_prompt (Optional[str]):

Optional system text for input using, default to None. Only used when your Service is a ChatCompletion service

**kwargs (Any):

Arbitrary keyword arguments

Returns:

Dataset: batch result contained in dataset

deploy(**kwargs: Any) → Service

deploy_config: Optional[DeployConfig]

service deploy config

dumps() → Optional[bytes]

serialize the model instance to bytes

Returns:

Optional[bytes]:

bytes of the model instance

endpoint: Optional[str]

service endpoint to call

exec(input: Optional[Dict] = None, **kwargs: Dict) → Union[QfResponse, Iterator[QfResponse]]

Parameters:

input (Optional[Union[str, List[str], List[dict]]], optional):

input of execution of service. Defaults to None.

**kwargs: additional args Dict

Raises:

InternalError: unsupported service type

Returns:

Union[str, List[str], List[dict]]:

output

get_res() → Union[ChatCompletion, Completion, Embedding, Text2Image]

convert to the specific model resources. e.g. ChatCompletion, Completion, Embeddings, Text2Image

Returns:

Union[ChatCompletion, Completion, Embedding, Text2Image]:

resource object

id: Optional[int]

remote service id

loads(data: bytes) → Any

load service instance from bytes

Parameters:

data (bytes):

bytes of model instance

Returns:

Any: model instance

model: Optional[Model]

service model instance

service_type: Optional[ServiceType]

service type, for user use service as a execution must specify

property status: str

get the service status

Raises:

InternalError: id not found

Returns:

console_const.ServiceStatus

Submodules

qianfan.model.configs module

class qianfan.model.configs.DeployConfig(*, name: str = '', endpoint_prefix: str = '', description: str = '', replicas: int = 1, pool_type: DeployPoolType = DeployPoolType.PrivateResource, service_type: ServiceType, extras: Any = None)

Bases: BaseModel

description: str

description of service

endpoint_prefix: str

Endpoint custom prefix, will be used to call resource api

extras: Any

name: str

Service name

pool_type: DeployPoolType

resource pool type, public resource will be shared with others.

replicas: int

replicas for model services, related to the capacity in QPS of model service.

default set to 1

service_type: ServiceType

service type, after deploy, Service could behave like the specific type.

qianfan.model.consts module

class qianfan.model.consts.ServiceType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Chat = 'Chat'

Corresponding to the ChatCompletion

Completion = 'Completion'

Corresponding to the Completion

Embedding = 'Embedding'

Corresponding to the Embedding

Text2Image = 'Text2Image'

Corresponding to the `Text2Image

qianfan.model.model module

class qianfan.model.model.Model(id: Optional[str] = None, version_id: Optional[str] = None, task_id: Optional[str] = None, job_id: Optional[str] = None, name: Optional[str] = None, **kwargs: Any)

Bases: ExecuteSerializable[Dict, Union[QfResponse, Iterator[QfResponse]]]

auto_complete_info(**kwargs: Any) → None

auto complete Model object’s info. This may override the input model id version id.

Parameters:

**kwargs (Any):

arbitrary arguments

batch_inference(dataset: Dataset, **kwargs: Any) → Dataset

create batch run using specific dataset on qianfan by evaluation ability of platform

Parameters:

dataset (Dataset):

A dataset instance which indicates a dataset on qianfan platform

**kwargs (Any):

Arbitrary keyword arguments

Returns:

Dataset: batch result contained in dataset

deploy(deploy_config: DeployConfig, **kwargs: Any) → Service

model deploy

Parameters:

deploy_config (DeployConfig):

model service deploy config

Returns:

Service: model service instance

dumps() → Optional[bytes]

Serialize the model to bytes.

Returns:

Optional[bytes]:

bytes of this model

exec(input: Optional[Dict] = None, **kwargs: Dict) → Union[QfResponse, Iterator[QfResponse]]

model execution, for different model service type, please input a dict with different keys. Concretely, take

`input={“messages”: [{“role”: “user”,

“content”: “hello world”}]}`

as input, when the model is a chat io Model.

Parameters:

input (Optional[Dict], optional):

input data . Defaults to None.

Raises:

InternalError: model with no service deployed is unable to call exec

Returns:

Union[QfResponse, Iterator[QfResponse]]:

output data

id: Optional[str]

remote model id

job_id: Optional[str]

train job id

loads(data: bytes) → Any

load model instance from bytes

Parameters:

data (bytes):

bytes of this model

Returns:

Any: model instance

name: Optional[str] = None

model name

old_id: Optional[int]

deprecated old model id

old_version_id: Optional[int]

deprecated old model version id

publish(name: str = '', **kwargs: Any) → Model

model publish, before deploying a model, it should be published.

Parameters:

name str:

model name. Defaults to “m_{task_id}{job_id}”.

service: Optional[Service] = None

model service

task_id: Optional[str]

train tkas id

version_id: Optional[str]

remote model version id

class qianfan.model.model.Service(id: Optional[int] = None, endpoint: Optional[str] = None, model: Optional[Union[Model, str]] = None, deploy_config: Optional[DeployConfig] = None, service_type: Optional[ServiceType] = None)

Bases: ExecuteSerializable[Dict, Union[QfResponse, Iterator[QfResponse]]]

batch_inference(dataset: Dataset, prompt_template: Optional[Prompt] = None, system_prompt: Optional[str] = None, **kwargs: Any) → Dataset

create batch run using specific dataset on qianfan

Args:

dataset (Dataset):

A dataset instance which indicates a dataset on qianfan platform

prompt_template (Optional[Prompt]):

Optional Prompt used as input of llm, default to None. Only used when your Service is a Completion service

system_prompt (Optional[str]):

Optional system text for input using, default to None. Only used when your Service is a ChatCompletion service

**kwargs (Any):

Arbitrary keyword arguments

Returns:

Dataset: batch result contained in dataset

deploy(**kwargs: Any) → Service

deploy_config: Optional[DeployConfig]

service deploy config

dumps() → Optional[bytes]

serialize the model instance to bytes

Returns:

Optional[bytes]:

bytes of the model instance

endpoint: Optional[str]

service endpoint to call

exec(input: Optional[Dict] = None, **kwargs: Dict) → Union[QfResponse, Iterator[QfResponse]]

Parameters:

input (Optional[Union[str, List[str], List[dict]]], optional):

input of execution of service. Defaults to None.

**kwargs: additional args Dict

Raises:

InternalError: unsupported service type

Returns:

Union[str, List[str], List[dict]]:

output

get_res() → Union[ChatCompletion, Completion, Embedding, Text2Image]

convert to the specific model resources. e.g. ChatCompletion, Completion, Embeddings, Text2Image

Returns:

Union[ChatCompletion, Completion, Embedding, Text2Image]:

resource object

id: Optional[int]

remote service id

loads(data: bytes) → Any

load service instance from bytes

Parameters:

data (bytes):

bytes of model instance

Returns:

Any: model instance

model: Optional[Model]

service model instance

service_type: Optional[ServiceType]

service type, for user use service as a execution must specify

property status: str

get the service status

Raises:

InternalError: id not found

Returns:

console_const.ServiceStatus

qianfan.model.model.model_deploy(model: Model, deploy_config: DeployConfig, **kwargs: Any) → Service

model deployment implement, a polling loop will be called after deploy task created.

Parameters:

model (Model):

model to deploy

deploy_config (DeployConfig):

service deploy config, mainly including replicas and pool type.

Returns:

Service: deployed service with endpoint to call