Class DashScopeBatchEmbedder

api/dashscope_client.py:736–831

Batch embedder specifically designed for DashScope API

# Batch Embedding Components for DashScope
class DashScopeBatchEmbedder(DataComponent):
    """Batch embedder specifically designed for DashScope API"""

    def __init__(self, embedder, batch_size: int = 100, embedding_cache_file_name: str = "default") -> None:
        super().__init__(batch_size=batch_size)
        self.embedder = embedder
        self.batch_size = batch_size
        if self.batch_size > 25:
            log.warning(f"DashScope batch embedder initialization, batch size: {self.batch_size}, note that DashScope batch embedding size cannot exceed 25, automatically set to 25")
            self.batch_size = 25
        self.cache_path = f'./embedding_cache/{embedding_cache_file_name}_{self.embedder.__class__.__name__}_dashscope_embeddings.pkl'

    def call(
        self, input: BatchEmbedderInputType, model_kwargs: Optional[Dict] = {}, force_recreate: bool = False
    ) -> BatchEmbedderOutputType:
        """
        Batch call to DashScope embedder
        
        Args:
            input: List of input texts
            model_kwargs: Model parameters
            force_recreate: Whether to force recreation
            
        Returns:
            Batch embedding output
        """
        # Check cache first
        
        if not force_recreate and os.path.exists(self.cache_path):
            try:
                with open(self.cache_path, 'rb') as f:
                    embeddings = pickle.load(f)
                    log.info(f"Loaded cached DashScope embeddings from: {self.cache_path}")
                return embeddings
            except Exception as e:
                log.warning(f"Failed to load cache file {self.cache_path}: {e}, proceeding with fresh embedding")
        
        if isinstance(input, str):
            input = [input]
        
        n = len(input)
        embeddings: List[EmbedderOutput] = []
        
        log.info(f"Starting DashScope batch embedding processing, total {n} texts, batch size: {self.batch_size}")
        
        for i in tqdm(
            range(0, n, self.batch_size),
            desc="DashScope batch embedding",
            disable=False,
        ):
            batch_input = input[i : min(i + self.batch_size, n)]
            
            try:
                # Use correct calling method: directly call embedder instance
                batch_output = self.embedder(
                    input=batch_input, model_kwargs=model_kwargs
                )
                embeddings.append(batch_output)