Class DashScopeToEmbeddings

api/dashscope_client.py:834–929

Component that converts document sequences to embedding vector sequences, specifically optimized for DashScope API

class DashScopeToEmbeddings(DataComponent):
    """Component that converts document sequences to embedding vector sequences, specifically optimized for DashScope API"""

    def __init__(self, embedder, batch_size: int = 100, force_recreate_db: bool = False, embedding_cache_file_name: str = "default") -> None:
        super().__init__(batch_size=batch_size)
        self.embedder = embedder
        self.batch_size = batch_size
        self.batch_embedder = DashScopeBatchEmbedder(embedder=embedder, batch_size=batch_size, embedding_cache_file_name=embedding_cache_file_name)
        self.force_recreate_db = force_recreate_db

    def __call__(self, input: List[Document]) -> List[Document]:
        """
        Process list of documents, generating embedding vectors for each document
        
        Args:
            input: List of input documents
            
        Returns:
            List of documents containing embedding vectors
        """
        output = deepcopy(input)
        
        # Convert to text list
        embedder_input: List[str] = [chunk.text for chunk in output]
        
        log.info(f"Starting to process embeddings for {len(embedder_input)} documents")
        
        # Batch process embeddings
        outputs: List[EmbedderOutput] = self.batch_embedder(
            input=embedder_input, 
            force_recreate=self.force_recreate_db
        )
        
        # Validate output
        total_embeddings = 0
        error_batches = 0
        
        for batch_output in outputs:
            if batch_output.error:
                error_batches += 1
                log.error(f"Found error batch: {batch_output.error}")
            elif batch_output.data:
                total_embeddings += len(batch_output.data)
            
        log.info(f"Embedding statistics: total {total_embeddings} valid embeddings, {error_batches} error batches")
        
        # Assign embedding vectors back to documents
        doc_idx = 0
        for batch_idx, batch_output in tqdm(
            enumerate(outputs), 
            desc="Assigning embedding vectors to documents",
            disable=False
        ):
            if batch_output.error:
                # Create empty vectors for documents in error batches
                batch_size_actual = min(self.batch_size, len(output) - doc_idx)
                log.warning(f"Creating empty vectors for {batch_size_actual} documents in batch {batch_idx}")