RuntimeError: The size of tensor a (1718) must match the size of tensor b (600) at non-singleton dimension 1

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/scratch/hemanth/Hemanth/llms-data/RAG_pipeline/main.py", line 201, in <module>
    speech=tts.synthesize(text,"./output2.mp3",)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/scratch/hemanth/Hemanth/llms-data/RAG_pipeline/chatbottool/texttospeech.py", line 101, in synthesize
    raise RuntimeError(f"Speech synthesis failed: {str(e)}")
RuntimeError: Speech synthesis failed: The size of tensor a (1718) must match the size of tensor b (600) at non-singleton dimension 1

from typing import Union, Optional, List, Dict, Any
import logging
import torch
from transformers import SpeechT5Processor, SpeechT5ForTextToSpeech, SpeechT5HifiGan
from transformers.modeling_utils import PreTrainedModel
from datasets import load_dataset
import soundfile as sf
from abc import ABC, abstractmethod

logger = logging.getLogger(__name__)

class BaseTextToSpeechPipeline(ABC):
    @abstractmethod
    def synthesize(self, *args, **kwargs):
        pass

    @abstractmethod
    def batch_synthesize(self, *args, **kwargs):
        pass

class TextToSpeech(BaseTextToSpeechPipeline):
    """A pipeline for text-to-speech synthesis using pre-trained models."""

    def __init__(
        self,
        cache_dir: Optional[str] = "./cache",
        model_name: str = "microsoft/speecht5_tts",
        vocoder_name: str = "microsoft/speecht5_hifigan",
        device: Optional[str] = None,
        torch_dtype: Optional[torch.dtype] = None,
        speaker_embeddings_dataset: str = "Matthijs/cmu-arctic-xvectors"
    ) -> None:
        """
        Initialize the TextToSpeech pipeline.

        Args:
            cache_dir: Directory to cache model files.
            model_name: Name of the pre-trained TTS model.
            vocoder_name: Name of the pre-trained vocoder model.
            device: Device to run the model on. If None, automatically selects.
            torch_dtype: Torch data type for the model. If None, automatically selects.
            speaker_embeddings_dataset: Dataset containing speaker embeddings.
        """
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        self.torch_dtype = torch_dtype or (torch.float16 if torch.cuda.is_available() else torch.float32)
        
        try:
            self.processor = SpeechT5Processor.from_pretrained(model_name, cache_dir=cache_dir)
            self.model: PreTrainedModel = SpeechT5ForTextToSpeech.from_pretrained(
                model_name,
                torch_dtype=self.torch_dtype,
                cache_dir=cache_dir
            ).to(self.device)
            self.vocoder = SpeechT5HifiGan.from_pretrained(vocoder_name, cache_dir=cache_dir).to(self.device)
            
            embeddings_dataset = load_dataset(speaker_embeddings_dataset, split="validation",cache_dir="")
            self.speaker_embeddings = torch.tensor(embeddings_dataset[7306]["xvector"]).unsqueeze(0).to(self.device)
        except Exception as e:
            logger.error(f"Failed to load models or datasets: {str(e)}")
            raise

    def synthesize(
        self,
        text: str,
        output_path: Optional[str] = None,
        speaker_embedding: Optional[torch.Tensor] = None,
        **kwargs: Any
    ) -> torch.Tensor:
        """
        Synthesize speech from the input text.

        Args:
            text: Input text to synthesize.
            output_path: Path to save the synthesized audio. If None, audio is not saved.
            speaker_embedding: Custom speaker embedding. If None, default is used.
            **kwargs: Additional keyword arguments for model.generate_speech().

        Returns:
            Synthesized speech as a torch.Tensor.

        Raises:
            RuntimeError: If synthesis fails.
        """
        try:
            inputs = self.processor(text=text, return_tensors="pt").to(self.device)
            embedding = speaker_embedding if speaker_embedding is not None else self.speaker_embeddings

            speech = self.model.generate_speech(
                inputs["input_ids"], 
                embedding, 
                vocoder=self.vocoder,
                **kwargs
            )

            if output_path:
                sf.write(output_path, speech.cpu().numpy(), samplerate=16000)

            return speech
        except Exception as e:
            logger.error(f"Speech synthesis failed: {str(e)}")
            raise RuntimeError(f"Speech synthesis failed: {str(e)}")

    def batch_synthesize(
        self,
        texts: List[str],
        output_paths: Optional[List[str]] = None,
        speaker_embeddings: Optional[List[torch.Tensor]] = None,
        **kwargs: Any
    ) -> List[torch.Tensor]:
        """
        Synthesize speech for a batch of input texts.

        Args:
            texts: List of input texts to synthesize.
            output_paths: List of paths to save the synthesized audios. If None, audios are not saved.
            speaker_embeddings: List of custom speaker embeddings. If None, default is used for all.
            **kwargs: Additional keyword arguments for model.generate_speech().

        Returns:
            List of synthesized speeches as torch.Tensors.

        Raises:
            RuntimeError: If batch synthesis fails.
        """
        try:
            speeches = []
            for i, text in enumerate(texts):
                embedding = speaker_embeddings[i] if speaker_embeddings else self.speaker_embeddings
                output_path = output_paths[i] if output_paths else None
                speech = self.synthesize(text, output_path, embedding, **kwargs)
                speeches.append(speech)
            return speeches
        except Exception as e:
            logger.error(f"Batch speech synthesis failed: {str(e)}")
            raise RuntimeError(f"Batch speech synthesis failed: {str(e)}")

    def __call__(
        self,
        text: Union[str, List[str]],
        output_path: Optional[Union[str, List[str]]] = None,
        speaker_embedding: Optional[Union[torch.Tensor, List[torch.Tensor]]] = None,
        **kwargs: Any
    ) -> Union[torch.Tensor, List[torch.Tensor]]:
        """
        Call the pipeline directly.

        Args:
            text: Input text or list of texts to synthesize.
            output_path: Path or list of paths to save the synthesized audio(s).
            speaker_embedding: Custom speaker embedding or list of embeddings.
            **kwargs: Additional keyword arguments for synthesize().

        Returns:
            Synthesized speech or list of synthesized speeches as torch.Tensors.

        Raises:
            ValueError: If inputs are not properly formatted.
        """
        if isinstance(text, str):
            return self.synthesize(text, output_path, speaker_embedding, **kwargs)
        elif isinstance(text, list):
            if output_path is not None and not isinstance(output_path, list):
                raise ValueError("output_path must be a list when synthesizing multiple texts.")
            if speaker_embedding is not None and not isinstance(speaker_embedding, list):
                raise ValueError("speaker_embedding must be a list when synthesizing multiple texts.")
            return self.batch_synthesize(text, output_path, speaker_embedding, **kwargs)
        else:
            raise ValueError("Input must be a single text or a list of texts.")
        

# #example
# tts=TextToSpeech(device="cuda", torch_dtype=torch.float32)
# text="""
# text: hemanta is a bad boy, but goood person: Path or list of paths to save the synthesized audio(s).speaker_embedding: Custom speaker embedding or list of embeddings.**kwargs: Additional keyword arguments for synthesize().
# """
# tts.synthesize(text,"./output2.mp3",)