for o in post_model_outputs_response.outputs
                ]
            )
        return embeddings
[docs]    def embed_query(self, text: str) -> List[float]:
        """Call out to Clarifai's embedding models.
        Args:
            text: The text to embed.
        Returns:
            Embeddings for the text.
        """
        try:
            from clarifai_grpc.grpc.api import (
                resources_pb2,
                service_pb2,
            )
            from clarifai_grpc.grpc.api.status import status_code_pb2
        except ImportError:
            raise ImportError(
                "Could not import clarifai python package. "
                "Please install it with `pip install clarifai`."
            )
        post_model_outputs_request = service_pb2.PostModelOutputsRequest(
            user_app_id=self.userDataObject,
            model_id=self.model_id,
            version_id=self.model_version_id,
            inputs=[
                resources_pb2.Input(
                    data=resources_pb2.Data(text=resources_pb2.Text(raw=text))
                )
            ],
        )
        post_model_outputs_response = self.stub.PostModelOutputs(
            post_model_outputs_request
        )
        if post_model_outputs_response.status.code != status_code_pb2.SUCCESS:
            logger.error(post_model_outputs_response.status)
            first_output_failure = (
                post_model_outputs_response.outputs[0].status
                if len(post_model_outputs_response.outputs[0])
                else None
            )
            raise Exception(
                f"Post model outputs failed, status: "
                f"{post_model_outputs_response.status}, first output failure: "
                f"{first_output_failure}"
            )
        embeddings = [
            list(o.data.embeddings[0].vector)
            for o in post_model_outputs_response.outputs