Integrating optimization

learn2rag/compose/__init__.py

@@ -206,8 +206,15 @@ def start(self) -> None:
         try:
             for file in self.content.get('files', []):
                 file_path = Path(file['path']).expanduser().absolute()
-                file_path.parent.mkdir(parents=True, exist_ok=True)
-                file_path.write_text(file['content'])
+                if not file_path.exists() or file.get('force', True):
+                    file_path.parent.mkdir(parents=True, exist_ok=True)
+                    if 'content' in file:
+                        content = file['content']
+                    elif 'src' in file:
+                        content = Path(file['src']).read_text()
+                    else:
+                        raise NotImplementedError(file)
+                    file_path.write_text(content)
         except Exception as e:
             con.rollback()
             raise e

learn2rag/evaluation/example.py

@@ -51,5 +51,6 @@ def process_qa(dataset_name: str, qa_rows: Any) -> None:
     # qa_rows = read_dataset_qa('hotpot_qa', 'distractor', 'validation')
     # process_qa('hotpot_qa', qa_rows.select(range(3)))
 
-    qa_rows = read_dataset_qa('WikiEval', '', 'train')
+    target_path = pathlib.Path('./datasets/WikiEval/source')
+    qa_rows = read_dataset_qa(target_path, split='train')
     process_qa('WikiEval', qa_rows)

learn2rag/evaluation/tools.py

@@ -6,7 +6,8 @@
 import logging
 import datasets
 import json_stream  # type: ignore[import-untyped]
-from typing import Any, Callable
+from typing import Any, Callable, cast
+import pandas as pd
 
 import learn2rag.pipeline.ingestion
 from learn2rag.pipeline.config import opt_config
@@ -78,13 +79,39 @@ def ingest_dataset_documents(dataset_name: str) -> None:
     # FIXME
     # learn2rag.pipeline.ingestion.index(user_config, opt_config)
 
+def read_dataset_qa(file_path: pathlib.Path | str, split: str | None = None) -> Any:
+    target_path = pathlib.Path(file_path)
+    logging.debug(f'Loading dataset from: {target_path}')
+
+    if not target_path.exists():
+        raise FileNotFoundError(f"Dataset path does not exist: {target_path}")
+
+    if target_path.suffix.lower() == '.csv':
+        logging.debug('load csv file')
+        with open(target_path, 'r', encoding='utf-8') as f:
+            first_line = f.readline()
+            try:
+                # Sniff for comma, semicolon, or tab
+                dialect = csv.Sniffer().sniff(first_line, delimiters=",;\t")
+                detected_sep = dialect.delimiter
+            except csv.Error:
+                # Fallback to standard comma if sniffing fails
+                detected_sep = ','
+
+        logging.debug(f'load csv file with delimiter: "{detected_sep}"')
+        dataset_dict = datasets.load_dataset('csv', data_files=str(target_path), sep=detected_sep)
+    else:
+        logging.debug(f'load HF dataset ')
+        dataset_dict = datasets.load_from_disk(str(target_path))
 
-def read_dataset_qa(dataset_name: str, subdirectory: str, split: str | None=None) -> Any:
-    logging.debug(f'{dataset_name=}')
-    dataset_work_dir = pathlib.Path('./datasets') / dataset_name
-    dataset_dict = datasets.load_from_disk(dataset_work_dir / 'source' / subdirectory)
     return dataset_dict[split] if split is not None else dataset_dict
 
+# def read_dataset_qa(dataset_name: str, subdirectory: str, split: str | None=None) -> Any:
+#     logging.debug(f'{dataset_name=}')
+#     dataset_work_dir = pathlib.Path('./datasets') / dataset_name
+#     dataset_dict = datasets.load_from_disk(dataset_work_dir / 'source' / subdirectory)
+#     return dataset_dict[split] if split is not None else dataset_dict
+
 
 def basic_pipeline(dataset_name: str, question: str) -> dict[str, Any]:
     user_config = {

learn2rag/optimization/__init__.py

@@ -0,0 +1,98 @@
+import logging
+import pathlib
+import os
+import json
+import collections.abc
+import copy
+import yaml
+import argparse
+from typing import Any, Dict, Callable
+from collections.abc import Mapping
+
+from . import baseline_optimization
+from . import retrieval_optimization
+
+OPTIMIZATION_STRATEGIES: Dict[str, Callable[..., Any]] = {
+    "baseline": baseline_optimization.run,
+    "retrieval": retrieval_optimization.run,
+}
+
+
+#TODO : now we need to copy the dataset to here manually it should consider in installation maybe !
+# {storage_path}/datasets/WikiEval/
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Run RAG baseline optimization.")
+    parser.add_argument("task", help="Module task name (e.g., learn2rag.optimization)")
+    parser.add_argument("--logging-config", type=str, help="Path to logging config yml")
+    parser.add_argument("--registry-path", type=str, help="Path to registry.json")
+    parser.add_argument("--dataset", type=str, default="WikiEval")
+    parser.add_argument("--questions", type=int, default=10)
+    parser.add_argument("--trials", type=int, default=10)
+    parser.add_argument(
+        "--strategy",
+        type=str,
+        default="baseline",
+        choices=list(OPTIMIZATION_STRATEGIES.keys()),
+        help="Which optimization algorithm to run"
+    )
+    args, unknown = parser.parse_known_args()
+
+    if args.logging_config and pathlib.Path(args.logging_config).exists():
+        with open(args.logging_config, 'r') as f:
+            config = yaml.safe_load(f)
+            logging.config.dictConfig(config)
+
+    logger = logging.getLogger(__name__)
+    logger.info(f"Running task: {args.task}")
+    logger.info(f"Optimization started for {args.dataset} using strategy: {args.strategy}")
+
+    logging.info(f"Optimization started for {args.dataset}")
+    #TODO : read or get dataset_name and maxquestions and n_trails
+
+    output_dir = pathlib.Path("./optimization/output/")
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    run_optimization = OPTIMIZATION_STRATEGIES[args.strategy]
+    run_optimization(args.dataset, args.questions, args.trials, output_dir, args.registry_path)
+
+    logging.info("optimization is done")
+    results_path = output_dir /args.dataset/ "optimization_results.json"
+    logging.info(f"save optimized results here : {results_path}")
+    if not results_path.exists():
+        logging.warning(f"Optimization results not found at {results_path}")
+        return
+
+    with open(results_path, "r") as f:
+        full_results = json.load(f)
+
+    best_config = full_results.get("best_config", {})
+
+    target_config_path = os.environ.get("PIPELINE_OPT_CONFIG", "learn2rag/pipeline/opt_config.json")
+
+    existing_config = {}
+    if os.path.exists(target_config_path):
+        try:
+            with open(target_config_path, "r", encoding="utf-8") as f:
+                existing_config = json.load(f)
+        except json.JSONDecodeError:
+            logging.error(f"Existing config at {target_config_path} is corrupted. It will be overwritten.")
+
+    updated_config = deep_update(copy.deepcopy(existing_config), best_config)
+
+    pathlib.Path(target_config_path).parent.mkdir(parents=True, exist_ok=True)
+    with open(target_config_path, "w", encoding="utf-8") as f:
+        json.dump(updated_config, f, indent=4)
+
+    logging.info(f"Successfully updated opt_config at: {target_config_path}")
+
+def deep_update(source: dict[str, Any], overrides:Mapping[str, Any]) -> dict[str, Any]:
+    """Recursively updates a dictionary."""
+    for key, value in overrides.items():
+        if isinstance(value, collections.abc.Mapping) and key in source:
+            deep_update(source.get(key, {}), value)
+        else:
+            source[key] = value
+    return source
+
+if __name__ == "__main__":
+    main()