def compose()

in src/fmeval/model_runners/composers/jumpstart_composer.py [0:0]

• 17 lines of code
• 4 McCabe index (conditional complexity)

    def compose(self, prompt: str) -> Union[JumpStartSerializablePayload, bytes]:
        """
        Composes the payload for the given JumpStartModel from the provided prompt.
        """
        # embedding models input to the endpoint is any string of text dumped in json and encoded in `utf-8` format
        if self._is_embedding_model:
            return json.dumps(prompt).encode("utf-8")
        sagemaker_session = get_sagemaker_session()
        # Default model type is always OPEN_WEIGHTS. See https://tinyurl.com/yc58s6wj
        jumpstart_model_type = JumpStartModelType.OPEN_WEIGHTS
        if is_proprietary_js_model(sagemaker_session.boto_region_name, self._model_id):
            jumpstart_model_type = JumpStartModelType.PROPRIETARY

        payload = _construct_payload(
            prompt,
            model_id=self._model_id,
            model_type=jumpstart_model_type,
            model_version=self._model_version,
            tolerate_deprecated_model=True,
            tolerate_vulnerable_model=True,
        )
        util.require(payload, f"Unable to fetch default model payload for JumpStart model: {self._model_id}")
        return payload