beimingwu
/
learnware
Not watched
1
0
Fork 0
Code
Releases 1 Wiki evaluate Activity
Issues 0 Pull Requests 0 Datasets Model Cloudbrain HPC
Browse Source

Merge pull request #157 from Learnware-LAMDA/text_benchmark 

[MNT] add text benchmark
tags/v0.3.2
bxdd GitHub 2 years ago
parent
6b2015432a cf4ad803b2
commit
ffd39a29a4
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 468 additions and 491 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +5
    -5
      docs/advanced/anchor.rst
  2. +1
    -2
      docs/components/spec.rst
  3. +7
    -7
      docs/start/exp.rst
  4. +58
    -0
      examples/dataset_text_workflow/README.md
  5. +62
    -0
      examples/dataset_text_workflow/config.py
  6. +0
    -29
      examples/dataset_text_workflow/example_files/example_init.py
  7. +0
    -8
      examples/dataset_text_workflow/example_files/example_yaml.yaml
  8. +0
    -4
      examples/dataset_text_workflow/example_files/requirements.txt
  9. +0
    -16
      examples/dataset_text_workflow/get_data.py
  10. +0
    -274
      examples/dataset_text_workflow/main.py
  11. +0
    -104
      examples/dataset_text_workflow/utils.py
  12. +285
    -0
      examples/dataset_text_workflow/workflow.py
  13. +3
    -3
      learnware/specification/module.py
  14. +21
    -15
      learnware/specification/regular/image/rkme.py
  15. +21
    -9
      learnware/tests/benchmarks/__init__.py
  16. +2
    -11
      learnware/tests/benchmarks/config.py
  17. +3
    -4
      learnware/tests/data.py

+ 5
- 5
docs/advanced/anchor.rst View File

@@ -4,15 +4,15 @@ Anchor learnware

Anchor learnwares are a small fraction of representative learnwares that helps locate user's requirements through user feedback. The learnware market can choose or generate several learnwares as anchor learnwares corresponding to the specification island. If the user does not have sufficient training data for constructing an RKME requirement, the learnware market can send several anchor learnwares to the user. By feeding her own data to these anchor learnwares, some information such as (precision, recall) or other performance indicators, can be generated and returned to the market. These information could help the market identify potentially helpful models, e.g., by identifying models that are far from anchors exhibiting poor performance whereas close to anchors exhibiting relatively better performance in the specification island.

To fulfill the anchor learnware method, you need to implement the following functions in the folder ``Learnware/market/anchor``:
To fulfill the anchor learnware method, you need to implement the following functions in ``anchor.py``:

- First, you should design how the market chooses or generates anchor learnwares. This can be realized by selecting prototype models through functional space clustering, and more interesting designs can be explored. The class ``AnchoredOrganizer`` in ``organizer.py`` is designed for it. The function ``AnchoredOrganizer.update_anchor_learnware_list`` is reserved for choosing or generating anchor learnwares. The functions ``AnchoredOrganizer._update_anchor_learnware`` and ``AnchoredOrganizer._delete_anchor_learnware`` have been completed as auxiliary.
- First, you should design how the market chooses or generates anchor learnwares. This can be realized by selecting prototype models through functional space clustering, and more interesting designs can be explored. The function ``AnchoredMarket.update_anchor_learnware_list`` is reserved for it. The functions ``AnchoredMarket._update_anchor_learnware`` and ``AnchoredMarket._delete_anchor_learnware`` have been completed as auxiliary.

- Second, when a user comes with no RKME(or other statistical) specifications, the market should choose several anchor learnwares and send them to the user. This process is done by ``AnchoredSearcher.search_anchor_learnware`` in ``searcher.py``. Besides the list of anchor learnwares, it also returns an item specifying which performance indicator should the user return.
- Second, when a user comes with no RKME(or other statistical) specifications, the market should choose several anchor learnwares and send them to the user. This process is done by ``AnchoredMarket.search_anchor_learnware``, and the chosen anchors are stored in ``AnchoredUserInfo`` by ``AnchoredUserInfo.add_anchor_learnware``.
- Third, by feeding the user's data to these anchor learnwares, the returned information is calculated and stored in ``AnchoredUserInfo`` in ``user_info.py``. The user should add the anchor learnwares into it using ``AnchoredUserInfo.add_anchor_learnware_ids``, and then fill in performance indicator using ``AnchoredUserInfo.update_stat_info``.
- Third, the market should specify which performance indicator should the user return. By feeding the user's data to these anchor learnwares, the returned information is calculated and stored in ``AnchoredUserInfo`` by ``AnchoredUserInfo.update_stat_info``.
- Fourth, according to the returned information from the user, the market should identify the helpful learnwares for the user. This process is done in ``AnchoredSearcher.search_learnware`` in ``searcher.py``, which returns the recommended comination of helpful learnwares and the list of helpful learnwares.
- Fourth, according to the returned information from the user, the market should identify the helpful learnwares for the user. This process is done in ``AnchoredMarket.search_learnware``.




+ 1
- 2
docs/components/spec.rst View File

@@ -112,8 +112,7 @@ The RBF not only exposes the real data (plotted in the corresponding position in

Text Specification
--------------------------
Different from tabular data, each text input is a string of different length, so we should first transform them to equal-length arrays. Sentence embedding is used here to complete this transformation. We choose the model ``paraphrase-multilingual-MiniLM-L12-v2``, a lightweight multilingual embedding model. Then, we calculate the RKME specification on the embedding,
just like we do with tabular data. Besides, we use the package ``langdetect`` to detect and store the language of the text inputs for further search. We hope to search for the learnware which supports the language of the user task.


System Specification
======================================


+ 7
- 7
docs/start/exp.rst View File

@@ -139,13 +139,13 @@ Model training comprised two parts: the first part involved training a tfidf fea

Our experiments comprises two components:

* ``test_unlabeled`` is designed to evaluate performance when users possess only testing data, searching and reusing learnware available in the market.
* ``test_labeled`` aims to assess performance when users have both testing and limited training data, searching and reusing learnware directly from the market instead of training a model from scratch. This helps determine the amount of training data saved for the user.
* ``unlabeled_text_example`` is designed to evaluate performance when users possess only testing data, searching and reusing learnware available in the market.
* ``labeled_text_example`` aims to assess performance when users have both testing and limited training data, searching and reusing learnware directly from the market instead of training a model from scratch. This helps determine the amount of training data saved for the user.

Results
----------------

* ``test_unlabeled``:
* ``unlabeled_text_example``:

The accuracy of search and reuse is presented in the table below:

@@ -161,7 +161,7 @@ The accuracy of search and reuse is presented in the table below:
| Average Ensemble Reuse (Multiple) | 0.862 ± 0.051 |
+-----------------------------------+---------------------+

* ``test_labeled``:
* ``labeled_text_example``:

We present the change curves in classification error rates for both the user's self-trained model and the multiple learnware reuse(EnsemblePrune), showcasing their performance on the user's test data as the user's training data increases. The average results across 10 users are depicted below:

@@ -210,6 +210,6 @@ Examples for the `20-newsgroup` dataset are available at [examples/dataset_text_
We utilize the `fire` module to construct our experiments. You can execute the experiment with the following commands:

* `python main.py prepare_market`: Prepares the market.
* `python main.py test_unlabeled`: Executes the test_unlabeled experiment; the results will be printed in the terminal.
* `python main.py test_labeled`: Executes the test_labeled experiment; result curves will be automatically saved in the `figs` directory.
* Additionally, you can use `python main.py test_unlabeled True` to combine steps 1 and 2. The same approach applies to running test_labeled directly.
* `python main.py unlabeled_text_example`: Executes the unlabeled_text_example experiment; the results will be printed in the terminal.
* `python main.py labeled_text_example`: Executes the labeled_text_example experiment; result curves will be automatically saved in the `figs` directory.
* Additionally, you can use `python main.py unlabeled_text_example True` to combine steps 1 and 2. The same approach applies to running labeled_text_example directly.

+ 58
- 0
examples/dataset_text_workflow/README.md View File

@@ -0,0 +1,58 @@
# Text Dataset Workflow Example

## Introduction

We conducted experiments on the widely used text benchmark dataset: [``20-newsgroup``](http://qwone.com/~jason/20Newsgroups/).
``20-newsgroup`` is a renowned text classification benchmark with a hierarchical structure, featuring 5 superclasses {comp, rec, sci, talk, misc}.

In the submitting stage, we enumerated all combinations of three superclasses from the five available, randomly sampling 50% of each combination from the training set to create datasets for 50 uploaders.

In the deploying stage, we considered all combinations of two superclasses out of the five, selecting all data for each combination from the testing set as a test dataset for one user. This resulted in 10 users.
The user's own training data was generated using the same sampling procedure as the user test data, despite originating from the training dataset.

Model training comprised two parts: the first part involved training a tfidf feature extractor, and the second part used the extracted text feature vectors to train a naive Bayes classifier.

Our experiments comprises two components:

* ``unlabeled_text_example`` is designed to evaluate performance when users possess only testing data, searching and reusing learnware available in the market.

* ``labeled_text_example`` aims to assess performance when users have both testing and limited training data, searching and reusing learnware directly from the market instead of training a model from scratch. This helps determine the amount of training data saved for the user.


## Run the code

Run the following command to start the ``unlabeled_text_example``.

```bash
python workflow.py unlabeled_text_example
```

Run the following command to start the ``labeled_text_example``.

```bash
python workflow.py labeled_text_example
```

## Results

### ``unlabeled_text_example``:

The accuracy of search and reuse is presented in the table below:

| Metric | Value |
|--------------------------------------|---------------------|
| Mean in Market (Single) | 0.507 ± 0.030 |
| Best in Market (Single) | 0.859 ± 0.051 |
| Top-1 Reuse (Single) | 0.846 ± 0.054 |
| Job Selector Reuse (Multiple) | 0.845 ± 0.053 |
| Average Ensemble Reuse (Multiple) | 0.862 ± 0.051 |

### ``labeled_text_example``:

We present the change curves in classification error rates for both the user's self-trained model and the multiple learnware reuse(EnsemblePrune), showcasing their performance on the user's test data as the user's training data increases. The average results across 10 users are depicted below:

<div style="text-align:center;">
<img src="../../docs/_static/img/text_labeled_curves.png" alt="Text Limited Labeled Data" style="width:50%;" />
</div>

From the figure above, it is evident that when the user's own training data is limited, the performance of multiple learnware reuse surpasses that of the user's own model. As the user's training data grows, it is expected that the user's model will eventually outperform the learnware reuse. This underscores the value of reusing learnware to significantly conserve training data and achieve superior performance when user training data is limited.

+ 62
- 0
examples/dataset_text_workflow/config.py View File

@@ -0,0 +1,62 @@
from learnware.tests.benchmarks import BenchmarkConfig


text_benchmark_config = BenchmarkConfig(
name="20-Newsgroups",
user_num=10,
learnware_ids=[
"00002193",
"00002192",
"00002191",
"00002190",
"00002189",
"00002188",
"00002187",
"00002186",
"00002185",
"00002184",
"00002183",
"00002182",
"00002181",
"00002180",
"00002179",
"00002178",
"00002177",
"00002176",
"00002175",
"00002174",
"00002173",
"00002172",
"00002171",
"00002170",
"00002169",
"00002168",
"00002167",
"00002166",
"00002165",
"00002164",
"00002163",
"00002162",
"00002161",
"00002160",
"00002159",
"00002158",
"00002157",
"00002156",
"00002155",
"00002154",
"00002153",
"00002152",
"00002151",
"00002150",
"00002149",
"00002148",
"00002147",
"00002146",
"00002145",
"00002144",
],
test_data_path="20-Newsgroups/test_data.zip",
train_data_path="20-Newsgroups/train_data.zip",
extra_info_path="20-Newsgroup/extra_info.zip",
)

+ 0
- 29
examples/dataset_text_workflow/example_files/example_init.py View File

@@ -1,29 +0,0 @@
import os
import pickle

import numpy as np

from learnware.model import BaseModel


class Model(BaseModel):
def __init__(self):
super(Model, self).__init__(input_shape=(1,), output_shape=(1,))
dir_path = os.path.dirname(os.path.abspath(__file__))

modelv_path = os.path.join(dir_path, "modelv.pth")
with open(modelv_path, "rb") as f:
self.modelv = pickle.load(f)

modell_path = os.path.join(dir_path, "modell.pth")
with open(modell_path, "rb") as f:
self.modell = pickle.load(f)

def fit(self, X: np.ndarray, y: np.ndarray):
pass

def predict(self, X: np.ndarray) -> np.ndarray:
return self.modell.predict(self.modelv.transform(X))

def finetune(self, X: np.ndarray, y: np.ndarray):
pass

+ 0
- 8
examples/dataset_text_workflow/example_files/example_yaml.yaml View File

@@ -1,8 +0,0 @@
model:
class_name: Model
kwargs: { }
stat_specifications:
- module_path: learnware.specification
class_name: RKMETextSpecification
file_name: rkme.json
kwargs: { }

+ 0
- 4
examples/dataset_text_workflow/example_files/requirements.txt View File

@@ -1,4 +0,0 @@
numpy
pickle
lightgbm
scikit-learn

+ 0
- 16
examples/dataset_text_workflow/get_data.py View File

@@ -1,16 +0,0 @@
import os

import pandas as pd


def get_data(data_root="./data"):
dtrain = pd.read_csv(os.path.join(data_root, "train.csv"))
dtest = pd.read_csv(os.path.join(data_root, "test.csv"))

# returned X(DataFrame), y(Series)
return (
dtrain[["discourse_text", "discourse_type"]],
dtrain["discourse_effectiveness"],
dtest[["discourse_text", "discourse_type"]],
dtest["discourse_effectiveness"],
)

+ 0
- 274
examples/dataset_text_workflow/main.py View File

@@ -1,274 +0,0 @@
import os
import fire
import pickle
import time
import zipfile
from shutil import copyfile, rmtree

import numpy as np

import learnware.specification as specification
from get_data import get_data
from learnware.logger import get_module_logger
from learnware.market import instantiate_learnware_market, BaseUserInfo
from learnware.reuse import JobSelectorReuser, AveragingReuser, EnsemblePruningReuser
from utils import generate_uploader, generate_user, TextDataLoader, train, eval_prediction

logger = get_module_logger("text_test", level="INFO")
origin_data_root = "./data/origin_data"
processed_data_root = "./data/processed_data"
tmp_dir = "./data/tmp"
learnware_pool_dir = "./data/learnware_pool"
dataset = "ae" # argumentative essays
n_uploaders = 7
n_users = 7
n_classes = 3
data_root = os.path.join(origin_data_root, dataset)
data_save_root = os.path.join(processed_data_root, dataset)
user_save_root = os.path.join(data_save_root, "user")
uploader_save_root = os.path.join(data_save_root, "uploader")
model_save_root = os.path.join(data_save_root, "uploader_model")
os.makedirs(data_root, exist_ok=True)
os.makedirs(user_save_root, exist_ok=True)
os.makedirs(uploader_save_root, exist_ok=True)
os.makedirs(model_save_root, exist_ok=True)

output_description = {
"Dimension": 1,
"Description": {
"0": "classify as 0(ineffective), 1(effective), or 2(adequate).",
},
}
semantic_specs = [
{
"Data": {"Values": ["Text"], "Type": "Class"},
"Task": {"Values": ["Classification"], "Type": "Class"},
"Library": {"Values": ["Scikit-learn"], "Type": "Class"},
"Scenario": {"Values": ["Education"], "Type": "Tag"},
"Description": {"Values": "", "Type": "String"},
"Name": {"Values": "learnware_1", "Type": "String"},
"Output": output_description,
"License": {"Values": ["MIT"], "Type": "Class"},
}
]

user_semantic = {
"Data": {"Values": ["Text"], "Type": "Class"},
"Task": {"Values": ["Classification"], "Type": "Class"},
"Library": {"Values": ["Scikit-learn"], "Type": "Class"},
"Scenario": {"Values": ["Education"], "Type": "Tag"},
"Description": {"Values": "", "Type": "String"},
"Name": {"Values": "", "Type": "String"},
"Output": output_description,
"License": {"Values": ["MIT"], "Type": "Class"},
}


class TextDatasetWorkflow:
def _init_text_dataset(self):
self._prepare_data()
self._prepare_model()

def _prepare_data(self):
X_train, y_train, X_test, y_test = get_data(data_root)

generate_uploader(X_train, y_train, n_uploaders=n_uploaders, data_save_root=uploader_save_root)
generate_user(X_test, y_test, n_users=n_users, data_save_root=user_save_root)

def _prepare_model(self):
dataloader = TextDataLoader(data_save_root, train=True)
for i in range(n_uploaders):
logger.info("Train on uploader: %d" % (i))
X, y = dataloader.get_idx_data(i)
vectorizer, lgbm = train(X, y, out_classes=n_classes)

modelv_save_path = os.path.join(model_save_root, "uploader_v_%d.pth" % (i))
modell_save_path = os.path.join(model_save_root, "uploader_l_%d.pth" % (i))

with open(modelv_save_path, "wb") as f:
pickle.dump(vectorizer, f)

with open(modell_save_path, "wb") as f:
pickle.dump(lgbm, f)

logger.info("Model saved to '%s' and '%s'" % (modelv_save_path, modell_save_path))

def _prepare_learnware(
self, data_path, modelv_path, modell_path, init_file_path, yaml_path, env_file_path, save_root, zip_name
):
os.makedirs(save_root, exist_ok=True)
tmp_spec_path = os.path.join(save_root, "rkme.json")

tmp_modelv_path = os.path.join(save_root, "modelv.pth")
tmp_modell_path = os.path.join(save_root, "modell.pth")

tmp_yaml_path = os.path.join(save_root, "learnware.yaml")
tmp_init_path = os.path.join(save_root, "__init__.py")
tmp_env_path = os.path.join(save_root, "requirements.txt")

with open(data_path, "rb") as f:
X = pickle.load(f)
semantic_spec = semantic_specs[0]

st = time.time()

user_spec = specification.RKMETextSpecification()

user_spec.generate_stat_spec_from_data(X=X)
ed = time.time()
logger.info("Stat spec generated in %.3f s" % (ed - st))
user_spec.save(tmp_spec_path)

copyfile(modelv_path, tmp_modelv_path)
copyfile(modell_path, tmp_modell_path)

copyfile(yaml_path, tmp_yaml_path)
copyfile(init_file_path, tmp_init_path)
copyfile(env_file_path, tmp_env_path)
zip_file_name = os.path.join(learnware_pool_dir, "%s.zip" % (zip_name))
with zipfile.ZipFile(zip_file_name, "w", compression=zipfile.ZIP_DEFLATED) as zip_obj:
zip_obj.write(tmp_spec_path, "rkme.json")

zip_obj.write(tmp_modelv_path, "modelv.pth")
zip_obj.write(tmp_modell_path, "modell.pth")

zip_obj.write(tmp_yaml_path, "learnware.yaml")
zip_obj.write(tmp_init_path, "__init__.py")
zip_obj.write(tmp_env_path, "requirements.txt")
rmtree(save_root)
logger.info("New Learnware Saved to %s" % (zip_file_name))
return zip_file_name

def prepare_market(self, regenerate_flag=False):
if regenerate_flag:
self._init_text_dataset()
text_market = instantiate_learnware_market(market_id="ae", rebuild=True)
try:
rmtree(learnware_pool_dir)
except:
pass
os.makedirs(learnware_pool_dir, exist_ok=True)
for i in range(n_uploaders):
data_path = os.path.join(uploader_save_root, "uploader_%d_X.pkl" % (i))

modelv_path = os.path.join(model_save_root, "uploader_v_%d.pth" % (i))
modell_path = os.path.join(model_save_root, "uploader_l_%d.pth" % (i))

init_file_path = "./example_files/example_init.py"
yaml_file_path = "./example_files/example_yaml.yaml"
env_file_path = "./example_files/requirements.txt"
new_learnware_path = self._prepare_learnware(
data_path,
modelv_path,
modell_path,
init_file_path,
yaml_file_path,
env_file_path,
tmp_dir,
"%s_%d" % (dataset, i),
)
semantic_spec = semantic_specs[0]
semantic_spec["Name"]["Values"] = "learnware_%d" % (i)
semantic_spec["Description"]["Values"] = "test_learnware_number_%d" % (i)
text_market.add_learnware(new_learnware_path, semantic_spec)

logger.info("Total Item: %d" % (len(text_market)))

def test(self, regenerate_flag=False):
self.prepare_market(regenerate_flag)
text_market = instantiate_learnware_market(market_id="ae")
print("Total Item: %d" % len(text_market))

select_list = []
avg_list = []
improve_list = []
job_selector_score_list = []
ensemble_score_list = []
pruning_score_list = []
for i in range(n_users):
user_data_path = os.path.join(user_save_root, "user_%d_X.pkl" % (i))
user_label_path = os.path.join(user_save_root, "user_%d_y.pkl" % (i))
with open(user_data_path, "rb") as f:
user_data = pickle.load(f)
with open(user_label_path, "rb") as f:
user_label = pickle.load(f)

user_stat_spec = specification.RKMETextSpecification()
user_stat_spec.generate_stat_spec_from_data(X=user_data)
user_info = BaseUserInfo(semantic_spec=user_semantic, stat_info={"RKMETextSpecification": user_stat_spec})
logger.info("Searching Market for user: %d" % (i))

search_result = text_market.search_learnware(user_info)
single_result = search_result.get_single_results()
multiple_result = search_result.get_multiple_results()

print(f"search result of user{i}:")
print(
f"single model num: {len(single_result)}, max_score: {single_result[0].score}, min_score: {single_result[-1].score}"
)

l = len(single_result)
acc_list = []
for idx in range(l):
learnware = single_result[idx].learnware
score = single_result[idx].score
pred_y = learnware.predict(user_data)
acc = eval_prediction(pred_y, user_label)
acc_list.append(acc)
print(
f"Top1-score: {single_result[0].score}, learnware_id: {single_result[0].learnware.id}, acc: {acc_list[0]}"
)

if len(multiple_result) > 0:
mixture_id = " ".join([learnware.id for learnware in multiple_result[0].learnwares])
print(f"mixture_score: {multiple_result[0].score}, mixture_learnware: {mixture_id}")
mixture_learnware_list = multiple_result[0].learnwares
else:
mixture_learnware_list = [single_result[0].learnware]

# test reuse (job selector)
reuse_baseline = JobSelectorReuser(learnware_list=mixture_learnware_list, herding_num=100)
reuse_predict = reuse_baseline.predict(user_data=user_data)
reuse_score = eval_prediction(reuse_predict, user_label)
job_selector_score_list.append(reuse_score)
print(f"mixture reuse loss(job selector): {reuse_score}")

# test reuse (ensemble)
reuse_ensemble = AveragingReuser(learnware_list=mixture_learnware_list, mode="vote_by_label")
ensemble_predict_y = reuse_ensemble.predict(user_data=user_data)
ensemble_score = eval_prediction(ensemble_predict_y, user_label)
ensemble_score_list.append(ensemble_score)
print(f"mixture reuse accuracy (ensemble): {ensemble_score}")

# test reuse (ensemblePruning)
reuse_pruning = EnsemblePruningReuser(learnware_list=mixture_learnware_list)
pruning_predict_y = reuse_pruning.predict(user_data=user_data)
pruning_score = eval_prediction(pruning_predict_y, user_label)
pruning_score_list.append(pruning_score)
print(f"mixture reuse accuracy (ensemble Pruning): {pruning_score}\n")

select_list.append(acc_list[0])
avg_list.append(np.mean(acc_list))
improve_list.append((acc_list[0] - np.mean(acc_list)) / np.mean(acc_list))

logger.info(
"Accuracy of selected learnware: %.3f +/- %.3f, Average performance: %.3f +/- %.3f"
% (np.mean(select_list), np.std(select_list), np.mean(avg_list), np.std(avg_list))
)
logger.info("Average performance improvement: %.3f" % (np.mean(improve_list)))
logger.info(
"Average Job Selector Reuse Performance: %.3f +/- %.3f"
% (np.mean(job_selector_score_list), np.std(job_selector_score_list))
)
logger.info(
"Averaging Ensemble Reuse Performance: %.3f +/- %.3f"
% (np.mean(ensemble_score_list), np.std(ensemble_score_list))
)
logger.info(
"Selective Ensemble Reuse Performance: %.3f +/- %.3f"
% (np.mean(pruning_score_list), np.std(pruning_score_list))
)


if __name__ == "__main__":
fire.Fire(TextDatasetWorkflow)

+ 0
- 104
examples/dataset_text_workflow/utils.py View File

@@ -1,104 +0,0 @@
import os
import pickle

import numpy as np
import pandas as pd
from lightgbm import LGBMClassifier
from sklearn.feature_extraction.text import TfidfVectorizer


class TextDataLoader:
def __init__(self, data_root, train: bool = True):
self.data_root = data_root
self.train = train

def get_idx_data(self, idx=0):
if self.train:
X_path = os.path.join(self.data_root, "uploader", "uploader_%d_X.pkl" % (idx))
y_path = os.path.join(self.data_root, "uploader", "uploader_%d_y.pkl" % (idx))
if not (os.path.exists(X_path) and os.path.exists(y_path)):
raise Exception("Index Error")
with open(X_path, "rb") as f:
X = pickle.load(f)
with open(y_path, "rb") as f:
y = pickle.load(f)
else:
X_path = os.path.join(self.data_root, "user", "user_%d_X.pkl" % (idx))
y_path = os.path.join(self.data_root, "user", "user_%d_y.pkl" % (idx))
if not (os.path.exists(X_path) and os.path.exists(y_path)):
raise Exception("Index Error")
with open(X_path, "rb") as f:
X = pickle.load(f)
with open(y_path, "rb") as f:
y = pickle.load(f)
return X, y


def generate_uploader(data_x: pd.Series, data_y: pd.Series, n_uploaders=50, data_save_root=None):
if data_save_root is None:
return
os.makedirs(data_save_root, exist_ok=True)

types = data_x["discourse_type"].unique()

for i in range(n_uploaders):
indices = data_x["discourse_type"] == types[i]
selected_X = data_x[indices]["discourse_text"].to_list()
selected_y = data_y[indices].to_list()

X_save_dir = os.path.join(data_save_root, "uploader_%d_X.pkl" % (i))
y_save_dir = os.path.join(data_save_root, "uploader_%d_y.pkl" % (i))
with open(X_save_dir, "wb") as f:
pickle.dump(selected_X, f)
with open(y_save_dir, "wb") as f:
pickle.dump(selected_y, f)

print("Saving to %s" % (X_save_dir))


def generate_user(data_x, data_y, n_users=50, data_save_root=None):
if data_save_root is None:
return
os.makedirs(data_save_root, exist_ok=True)

types = data_x["discourse_type"].unique()

for i in range(n_users):
indices = data_x["discourse_type"] == types[i]
selected_X = data_x[indices]["discourse_text"].to_list()
selected_y = data_y[indices].to_list()

X_save_dir = os.path.join(data_save_root, "user_%d_X.pkl" % (i))
y_save_dir = os.path.join(data_save_root, "user_%d_y.pkl" % (i))
with open(X_save_dir, "wb") as f:
pickle.dump(selected_X, f)
with open(y_save_dir, "wb") as f:
pickle.dump(selected_y, f)

print("Saving to %s" % (X_save_dir))


# Train Uploaders' models
def train(X, y, out_classes):
vectorizer = TfidfVectorizer(stop_words="english")
X_tfidf = vectorizer.fit_transform(X)

lgbm = LGBMClassifier(boosting_type="dart", n_estimators=500, num_leaves=21)
lgbm.fit(X_tfidf, y)

return vectorizer, lgbm


def eval_prediction(pred_y, target_y):
if not isinstance(pred_y, np.ndarray):
pred_y = pred_y.detach().cpu().numpy()
if len(pred_y.shape) == 1:
predicted = np.array(pred_y)
else:
predicted = np.argmax(pred_y, 1)
annos = np.array(target_y)

total = predicted.shape[0]
correct = (predicted == annos).sum().item()

return correct / total

+ 285
- 0
examples/dataset_text_workflow/workflow.py View File

@@ -0,0 +1,285 @@
import os
import fire
import time
import random
import pickle
import tempfile
import numpy as np
import matplotlib.pyplot as plt
from sklearn.metrics import accuracy_score
from sklearn.naive_bayes import MultinomialNB
from sklearn.feature_extraction.text import TfidfVectorizer

from learnware.client import LearnwareClient
from learnware.logger import get_module_logger
from learnware.specification import RKMETextSpecification
from learnware.tests.benchmarks import LearnwareBenchmark
from learnware.market import instantiate_learnware_market, BaseUserInfo
from learnware.reuse import JobSelectorReuser, AveragingReuser, EnsemblePruningReuser
from config import text_benchmark_config

logger = get_module_logger("text_workflow", level="INFO")


class TextDatasetWorkflow:
@staticmethod
def _train_model(X, y):
vectorizer = TfidfVectorizer(stop_words="english")
X_tfidf = vectorizer.fit_transform(X)
clf = MultinomialNB(alpha=0.1)
clf.fit(X_tfidf, y)
return vectorizer, clf

@staticmethod
def _eval_prediction(pred_y, target_y):
if not isinstance(pred_y, np.ndarray):
pred_y = pred_y.detach().cpu().numpy()

pred_y = np.array(pred_y) if len(pred_y.shape) == 1 else np.argmax(pred_y, 1)
target_y = np.array(target_y)
return accuracy_score(target_y, pred_y)

def _plot_labeled_peformance_curves(self, all_user_curves_data):
plt.figure(figsize=(10, 6))
plt.xticks(range(len(self.n_labeled_list)), self.n_labeled_list)

styles = [
{"color": "navy", "linestyle": "-", "marker": "o"},
{"color": "magenta", "linestyle": "-.", "marker": "d"},
]
labels = ["User Model", "Multiple Learnware Reuse (EnsemblePrune)"]

user_mat, pruning_mat = all_user_curves_data
user_mat, pruning_mat = np.array(user_mat), np.array(pruning_mat)
for mat, style, label in zip([user_mat, pruning_mat], styles, labels):
mean_curve, std_curve = 1 - np.mean(mat, axis=0), np.std(mat, axis=0)
plt.plot(mean_curve, **style, label=label)
plt.fill_between(
range(len(mean_curve)),
mean_curve - 0.5 * std_curve,
mean_curve + 0.5 * std_curve,
color=style["color"],
alpha=0.2,
)

plt.xlabel("Labeled Data Size")
plt.ylabel("1 - Accuracy")
plt.title(f"Text Limited Labeled Data")
plt.legend()
plt.tight_layout()
plt.savefig(os.path.join(self.fig_path, "text_labeled_curves.png"), bbox_inches="tight", dpi=700)

def _prepare_market(self, rebuild=False):
client = LearnwareClient()
self.text_benchmark = LearnwareBenchmark().get_benchmark(text_benchmark_config)
self.text_market = instantiate_learnware_market(market_id=self.text_benchmark.name, rebuild=rebuild)
self.user_semantic = client.get_semantic_specification(self.text_benchmark.learnware_ids[0])
self.user_semantic["Name"]["Values"] = ""

if len(self.text_market) == 0 or rebuild == True:
for learnware_id in self.text_benchmark.learnware_ids:
with tempfile.TemporaryDirectory(prefix="text_benchmark_") as tempdir:
zip_path = os.path.join(tempdir, f"{learnware_id}.zip")
for i in range(20):
try:
semantic_spec = client.get_semantic_specification(learnware_id)
client.download_learnware(learnware_id, zip_path)
self.text_market.add_learnware(zip_path, semantic_spec)
break
except:
time.sleep(1)
continue

logger.info("Total Item: %d" % (len(self.text_market)))

def unlabeled_text_example(self, rebuild=False):
self._prepare_market(rebuild)

select_list = []
avg_list = []
best_list = []
improve_list = []
job_selector_score_list = []
ensemble_score_list = []
all_learnwares = self.text_market.get_learnwares()

for i in range(self.text_benchmark.user_num):
user_data, user_label = self.text_benchmark.get_test_data(user_ids=i)

user_stat_spec = RKMETextSpecification()
user_stat_spec.generate_stat_spec_from_data(X=user_data)
user_info = BaseUserInfo(
semantic_spec=self.user_semantic, stat_info={"RKMETextSpecification": user_stat_spec}
)
logger.info("Searching Market for user: %d" % (i))

search_result = self.text_market.search_learnware(user_info)
single_result = search_result.get_single_results()
multiple_result = search_result.get_multiple_results()

print(f"search result of user{i}:")
print(
f"single model num: {len(single_result)}, max_score: {single_result[0].score}, min_score: {single_result[-1].score}"
)

acc_list = []
for idx in range(len(all_learnwares)):
learnware = all_learnwares[idx]
pred_y = learnware.predict(user_data)
acc = self._eval_prediction(pred_y, user_label)
acc_list.append(acc)

learnware = single_result[0].learnware
pred_y = learnware.predict(user_data)
best_acc = self._eval_prediction(pred_y, user_label)
best_list.append(np.max(acc_list))
select_list.append(best_acc)
avg_list.append(np.mean(acc_list))
improve_list.append((best_acc - np.mean(acc_list)) / np.mean(acc_list))
print(f"market mean accuracy: {np.mean(acc_list)}, market best accuracy: {np.max(acc_list)}")
print(
f"Top1-score: {single_result[0].score}, learnware_id: {single_result[0].learnware.id}, acc: {best_acc}"
)

if len(multiple_result) > 0:
mixture_id = " ".join([learnware.id for learnware in multiple_result[0].learnwares])
print(f"mixture_score: {multiple_result[0].score}, mixture_learnware: {mixture_id}")
mixture_learnware_list = multiple_result[0].learnwares
else:
mixture_learnware_list = [single_result[0].learnware]

# test reuse (job selector)
reuse_baseline = JobSelectorReuser(learnware_list=mixture_learnware_list, herding_num=100)
reuse_predict = reuse_baseline.predict(user_data=user_data)
reuse_score = self._eval_prediction(reuse_predict, user_label)
job_selector_score_list.append(reuse_score)
print(f"mixture reuse accuracy (job selector): {reuse_score}")

# test reuse (ensemble)
reuse_ensemble = AveragingReuser(learnware_list=mixture_learnware_list, mode="vote_by_label")
ensemble_predict_y = reuse_ensemble.predict(user_data=user_data)
ensemble_score = self._eval_prediction(ensemble_predict_y, user_label)
ensemble_score_list.append(ensemble_score)
print(f"mixture reuse accuracy (ensemble): {ensemble_score}\n")

logger.info(
"Accuracy of selected learnware: %.3f +/- %.3f, Average performance: %.3f +/- %.3f, Best performance: %.3f +/- %.3f"
% (
np.mean(select_list),
np.std(select_list),
np.mean(avg_list),
np.std(avg_list),
np.mean(best_list),
np.std(best_list),
)
)
logger.info("Average performance improvement: %.3f" % (np.mean(improve_list)))
logger.info(
"Average Job Selector Reuse Performance: %.3f +/- %.3f"
% (np.mean(job_selector_score_list), np.std(job_selector_score_list))
)
logger.info(
"Averaging Ensemble Reuse Performance: %.3f +/- %.3f"
% (np.mean(ensemble_score_list), np.std(ensemble_score_list))
)

def labeled_text_example(self, rebuild=False, train_flag=True):
self.n_labeled_list = [100, 200, 500, 1000, 2000, 4000]
self.repeated_list = [10, 10, 10, 3, 3, 3]
self.root_path = os.path.dirname(os.path.abspath(__file__))
self.fig_path = os.path.join(self.root_path, "figs")
self.curve_path = os.path.join(self.root_path, "curves")

if train_flag:
self._prepare_market(rebuild)
os.makedirs(self.fig_path, exist_ok=True)
os.makedirs(self.curve_path, exist_ok=True)

for i in range(self.text_benchmark.user_num):
user_model_score_mat = []
pruning_score_mat = []
single_score_mat = []
test_x, test_y = self.text_benchmark.get_test_data(user_ids=i)
test_y = np.array(test_y)

train_x, train_y = self.text_benchmark.get_train_data(user_ids=i)
train_y = np.array(train_y)

user_stat_spec = RKMETextSpecification()
user_stat_spec.generate_stat_spec_from_data(X=test_x)
user_info = BaseUserInfo(
semantic_spec=self.user_semantic, stat_info={"RKMETextSpecification": user_stat_spec}
)
logger.info(f"Searching Market for user_{i}")

search_result = self.text_market.search_learnware(user_info)
single_result = search_result.get_single_results()
multiple_result = search_result.get_multiple_results()

learnware = single_result[0].learnware
pred_y = learnware.predict(test_x)
best_acc = self._eval_prediction(pred_y, test_y)
print(f"search result of user_{i}:")
print(
f"single model num: {len(single_result)}, max_score: {single_result[0].score}, min_score: {single_result[-1].score}, single model acc: {best_acc}"
)

if len(multiple_result) > 0:
mixture_id = " ".join([learnware.id for learnware in multiple_result[0].learnwares])
print(f"mixture_score: {multiple_result[0].score}, mixture_learnware: {mixture_id}")
mixture_learnware_list = multiple_result[0].learnwares
else:
mixture_learnware_list = [single_result[0].learnware]
print(len(train_x))

for n_label, repeated in zip(self.n_labeled_list, self.repeated_list):
user_model_score_list, reuse_pruning_score_list = [], []
if n_label > len(train_x):
n_label = len(train_x)
for _ in range(repeated):
x_train, y_train = zip(*random.sample(list(zip(train_x, train_y)), k=n_label))
x_train = list(x_train)
y_train = np.array(list(y_train))

modelv, modell = self._train_model(x_train, y_train)
user_model_predict_y = modell.predict(modelv.transform(test_x))
user_model_score = self._eval_prediction(user_model_predict_y, test_y)
user_model_score_list.append(user_model_score)

reuse_pruning = EnsemblePruningReuser(
learnware_list=mixture_learnware_list, mode="classification"
)
reuse_pruning.fit(x_train, y_train)
reuse_pruning_predict_y = reuse_pruning.predict(user_data=test_x)
reuse_pruning_score = self._eval_prediction(reuse_pruning_predict_y, test_y)
reuse_pruning_score_list.append(reuse_pruning_score)

single_score_mat.append([best_acc] * repeated)
user_model_score_mat.append(user_model_score_list)
pruning_score_mat.append(reuse_pruning_score_list)
print(n_label, np.mean(user_model_score_mat[-1]), np.mean(pruning_score_mat[-1]))

logger.info(f"Saving Curves for User_{i}")
user_curves_data = (single_score_mat, user_model_score_mat, pruning_score_mat)
with open(os.path.join(self.curve_path, f"curve{str(i)}.pkl"), "wb") as f:
pickle.dump(user_curves_data, f)

pruning_curves_data, user_model_curves_data = [], []
for i in range(self.text_benchmark.user_num):
with open(os.path.join(self.curve_path, f"curve{str(i)}.pkl"), "rb") as f:
user_curves_data = pickle.load(f)
(single_score_mat, user_model_score_mat, pruning_score_mat) = user_curves_data
for i in range(len(single_score_mat)):
user_model_score_mat[i] = np.mean(user_model_score_mat[i])
pruning_score_mat[i] = np.mean(pruning_score_mat[i])
if len(user_model_score_mat) < 6:
for i in range(6 - len(user_model_score_mat)):
user_model_score_mat.append(user_model_score_mat[-1])
pruning_score_mat.append(pruning_score_mat[-1])
user_model_curves_data.append(user_model_score_mat[:6])
pruning_curves_data.append(pruning_score_mat[:6])
self._plot_labeled_peformance_curves([user_model_curves_data, pruning_curves_data])


if __name__ == "__main__":
fire.Fire(TextDatasetWorkflow)

+ 3
- 3
learnware/specification/module.py View File

@@ -78,7 +78,7 @@ def generate_rkme_image_spec(
reduce: bool = True,
verbose: bool = True,
cuda_idx: int = None,
**kwargs
**kwargs,
) -> RKMEImageSpecification:
"""
Interface for users to generate Reduced Kernel Mean Embedding (RKME) specification for Image.
@@ -168,7 +168,7 @@ def generate_rkme_text_spec(
# Check input type
if not isinstance(X, list) or not all(isinstance(item, str) for item in X):
raise TypeError("Input data must be a list of strings.")
# Generate rkme text spec
rkme_text_spec = RKMETextSpecification(gamma=gamma, cuda_idx=cuda_idx)
rkme_text_spec.generate_stat_spec_from_data(X, reduced_set_size, step_size, steps, nonnegative_beta, reduce)
@@ -237,7 +237,7 @@ def generate_semantic_spec(
}
if input_description is not None:
semantic_specification["Input"] = input_description
if output_description is not None:
semantic_specification["Output"] = output_description



+ 21
- 15
learnware/specification/regular/image/rkme.py View File

@@ -39,7 +39,7 @@ class RKMEImageSpecification(RegularStatSpecification):
"""
self.RKME_IMAGE_VERSION = 1 # Please maintain backward compatibility.

self.msg=None
self.msg = None

self.z = None
self.beta = None
@@ -169,7 +169,8 @@ class RKMEImageSpecification(RegularStatSpecification):
import torch_optimizer
except ModuleNotFoundError:
raise ModuleNotFoundError(
f"RKMEImageSpecification is not available because 'torch-optimizer' is not installed! Please install it manually.")
f"RKMEImageSpecification is not available because 'torch-optimizer' is not installed! Please install it manually."
)

# Cross-platform by default, unless the spec is specified to be generated specifically for local experiments.
cross_platform = "experimental" not in kwargs or not kwargs["experimental"]
@@ -421,7 +422,9 @@ class RKMEImageSpecification(RegularStatSpecification):
for d in self.get_states():
if d in rkme_load.keys():
if d == "type" and rkme_load[d] != self.type:
raise TypeError(f"The type of loaded RKME ({rkme_load[d]}) is different from the expected type ({self.type})!")
raise TypeError(
f"The type of loaded RKME ({rkme_load[d]}) is different from the expected type ({self.type})!"
)
setattr(self, d, rkme_load[d])

self.beta = self.beta.to(self._device)
@@ -440,9 +443,8 @@ def _get_zca_matrix(X, reg_coef=0.1):


class RandomGenerator:

def __init__(self, seed=0, cross_platform=True):
self.cross_platform=cross_platform
self.cross_platform = cross_platform
self.state = RandomState(seed)

def normal_(self, tensor: torch.Tensor, mean=0.0, std=1.0):
@@ -461,24 +463,24 @@ def deterministic(cross_platform, device):
deterministic_state = torch.backends.cudnn.deterministic
torch.backends.cudnn.deterministic = True
if cross_platform and torch.cuda.is_available():
torch.cuda.set_rng_state(
new_state=torch.cuda.get_rng_state(device.index),
device="cpu")
torch.cuda.set_rng_state(new_state=torch.cuda.get_rng_state(device.index), device="cpu")

yield RandomGenerator(seed=0, cross_platform=cross_platform)

torch.backends.cudnn.deterministic = deterministic_state
if cross_platform and torch.cuda.is_available():
torch.cuda.set_rng_state(
new_state=torch.cuda.get_rng_state(device.index),
device="cuda")
torch.cuda.set_rng_state(new_state=torch.cuda.get_rng_state(device.index), device="cuda")


class _ConvNet_wide(nn.Module):
def __init__(self, channel, random_generator, mu=None, sigma=None, k=2, net_width=128, net_depth=3, im_size=(32, 32)):
def __init__(
self, channel, random_generator, mu=None, sigma=None, k=2, net_width=128, net_depth=3, im_size=(32, 32)
):
self.k = k
super().__init__()
self.features, shape_feat = self._make_layers(channel, net_width, net_depth, im_size, mu, sigma, random_generator)
self.features, shape_feat = self._make_layers(
channel, net_width, net_depth, im_size, mu, sigma, random_generator
)
# self.aggregation = nn.AvgPool2d(kernel_size=shape_feat[1])

def forward(self, x):
@@ -494,7 +496,9 @@ class _ConvNet_wide(nn.Module):
in_channels = channel
shape_feat = [in_channels, im_size[0], im_size[1]]
for d in range(net_depth):
layers += [_build_conv2d_gaussian(in_channels, int(k * net_width), random_generator, 3, 1, mean=mu, std=sigma)]
layers += [
_build_conv2d_gaussian(in_channels, int(k * net_width), random_generator, 3, 1, mean=mu, std=sigma)
]
shape_feat[0] = int(k * net_width)

layers += [nn.ReLU(inplace=True)]
@@ -507,7 +511,9 @@ class _ConvNet_wide(nn.Module):
return nn.Sequential(*layers), shape_feat


def _build_conv2d_gaussian(in_channels, out_channels, random_generator: RandomGenerator, kernel=3, padding=1, mean=None, std=None):
def _build_conv2d_gaussian(
in_channels, out_channels, random_generator: RandomGenerator, kernel=3, padding=1, mean=None, std=None
):
layer = nn.Conv2d(in_channels, out_channels, kernel, padding=padding)
if mean is None:
mean = 0


+ 21
- 9
learnware/tests/benchmarks/__init__.py View File

@@ -12,16 +12,18 @@ from ...config import C

@dataclass
class Benchmark:
learnware_ids: List[str]
name: str
user_num: int
learnware_ids: List[str]
test_X_paths: List[str]
test_y_paths: List[str]
train_X_paths: Optional[List[str]] = None
train_y_paths: Optional[List[str]] = None
extra_info_path: Optional[str] = None

def get_test_data(self, user_ids: Union[str, List[str]]):
if isinstance(user_ids, str):
def get_test_data(self, user_ids: Union[int, List[int]]):
raw_user_ids = user_ids
if isinstance(user_ids, int):
user_ids = [user_ids]

ret = []
@@ -34,13 +36,17 @@ class Benchmark:

ret.append((test_X, test_y))

return ret
if isinstance(raw_user_ids, int):
return ret[0]
else:
return ret

def get_train_data(self, user_ids):
def get_train_data(self, user_ids: Union[int, List[int]]):
if self.train_X_paths is None or self.train_y_paths is None:
return None

if isinstance(user_ids, str):
raw_user_ids = user_ids
if isinstance(user_ids, int):
user_ids = [user_ids]

ret = []
@@ -53,7 +59,10 @@ class Benchmark:

ret.append((train_X, train_y))

return ret
if isinstance(raw_user_ids, int):
return ret[0]
else:
return ret


class LearnwareBenchmark:
@@ -107,7 +116,7 @@ class LearnwareBenchmark:
with zipfile.ZipFile(test_data_zippath, "r") as z_file:
z_file.extractall(save_path)

def _load_cache_data(self, benchmark_config: BenchmarkConfig, data_type: str) -> Tuple(List[str], List[str]):
def _load_cache_data(self, benchmark_config: BenchmarkConfig, data_type: str) -> Tuple[List[str], List[str]]:
"""Load data from local cache path

Parameters
@@ -131,6 +140,8 @@ class LearnwareBenchmark:
X_paths.append(user_X_path)
y_paths.append(user_y_path)

return X_paths, y_paths

def get_benchmark(self, benchmark_config: Union[str, BenchmarkConfig]):
if isinstance(benchmark_config, str):
benchmark_config = self.benchmark_configs[benchmark_config]
@@ -151,8 +162,9 @@ class LearnwareBenchmark:
self._download_data(benchmark_config.extra_info_path, extra_info_path)

return Benchmark(
learnware_ids=benchmark_config.learnware_ids,
name=benchmark_config.name,
user_num=benchmark_config.user_num,
learnware_ids=benchmark_config.learnware_ids,
test_X_paths=test_X_paths,
test_y_paths=test_y_paths,
train_X_paths=train_X_paths,


+ 2
- 11
learnware/tests/benchmarks/config.py View File

@@ -5,20 +5,11 @@ from typing import Optional, List
@dataclass
class BenchmarkConfig:
name: str
learnware_ids: List[str]
user_num: int
learnware_ids: List[str]
test_data_path: str
train_data_path: Optional[str] = None
extra_info_path: Optional[str] = None


benchmark_configs = {
"example": BenchmarkConfig(
name="example",
learnware_ids=["00001951", "00001980", "00001987"],
user_num=3,
test_data_path="example_path1",
train_data_path="example_path2",
extra_info_path="example_path3",
)
}
benchmark_configs = {}

+ 3
- 4
learnware/tests/data.py View File

@@ -1,4 +1,3 @@

import json
import requests
from tqdm import tqdm
@@ -18,12 +17,12 @@ class GetData:
self.chunk_size = chunk_size

def download_file(self, file_path: str, save_path: str):
url = f"{self.host}/engine/download"
url = f"{self.host}/datasets/download_datasets"

response = requests.get(
url,
params={
"file_path": file_path,
"dataset": file_path,
},
stream=True,
)
@@ -37,4 +36,4 @@ class GetData:
with open(save_path, "wb") as f:
for chunk in response.iter_content(chunk_size=self.chunk_size):
f.write(chunk)
bar.update(1)
bar.update(1)

Write Preview
Loading…
Cancel
Save