Class StreamingProcessor

rag_system/utils/batch_processor.py:135–186

Process items one at a time with minimal memory usage

            yield items[i:i + self.batch_size]

class StreamingProcessor:
    """Process items one at a time with minimal memory usage"""
    
    def __init__(self, enable_gc_interval: int = 100):
        self.enable_gc_interval = enable_gc_interval
        
    def process_streaming(
        self,
        items: List[Any],
        process_func: Callable,
        operation_name: str = "Streaming Processing",
        **kwargs
    ) -> List[Any]:
        """
        Process items one at a time with minimal memory footprint
        
        Args:
            items: List of items to process
            process_func: Function to process each item
            operation_name: Name for progress reporting
            **kwargs: Additional arguments passed to process_func
            
        Returns:
            List of results
        """
        if not items:
            logger.info(f"{operation_name}: No items to process")
            return []
            
        tracker = ProgressTracker(len(items), operation_name)
        results = []
        
        logger.info(f"Starting {operation_name} for {len(items)} items (streaming)")
        
        with timer(f"{operation_name} (streaming)"):
            for i, item in enumerate(items):
                try:
                    result = process_func(item, **kwargs)
                    results.append(result)
                    tracker.update(1)
                    
                except Exception as e:
                    logger.error(f"Error processing item {i}: {e}")
                    tracker.update(1, errors=1)
                    continue
                    
                # Periodic garbage collection
                if self.enable_gc_interval and (i + 1) % self.enable_gc_interval == 0:
                    gc.collect()
                    
        tracker.finish()
        return results

# Utility functions for common batch operations
def batch_chunks_by_document(chunks: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]: