mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-27 15:37:44 +00:00
you-n-g
c707a40073
* scen * scen2 * app * fix * Simplify workflow * We can share more code in new scenarios * rename model to rd loop * Optimize data path * Update rdagent/app/data_mining/model.py * Add TODO * Support GPU * gpu --------- Co-authored-by: SH-Src <suhan.c@outlook.com>
94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
import json
|
|
from pathlib import Path
|
|
from typing import List, Tuple
|
|
|
|
from jinja2 import Environment, StrictUndefined
|
|
|
|
from rdagent.components.coder.model_coder.model import ModelExperiment, ModelTask
|
|
from rdagent.components.proposal.model_proposal import (
|
|
ModelHypothesis,
|
|
ModelHypothesis2Experiment,
|
|
ModelHypothesisGen,
|
|
)
|
|
from rdagent.core.prompts import Prompts
|
|
from rdagent.core.proposal import Hypothesis, Scenario, Trace
|
|
from rdagent.scenarios.data_mining.experiment.model_experiment import DMModelExperiment
|
|
|
|
prompt_dict = Prompts(file_path=Path(__file__).parent.parent.parent / "qlib" / "prompts.yaml")
|
|
|
|
DMModelHypothesis = ModelHypothesis
|
|
|
|
|
|
class DMModelHypothesisGen(ModelHypothesisGen):
|
|
"""
|
|
# NOTE: we can share this class across different data mining scenarios
|
|
# It may better to move the class into components folder like `rdagent/components/proposal/model_proposal.py`
|
|
# Here is the use case:
|
|
|
|
.. code-block:: python
|
|
|
|
class XXXDMModelHypothesisGen(DMModelHypothesisGen):
|
|
prompts: Prompts = a_specifc_prompt_dict
|
|
"""
|
|
def __init__(self, scen: Scenario) -> Tuple[dict, bool]:
|
|
super().__init__(scen)
|
|
|
|
def prepare_context(self, trace: Trace) -> Tuple[dict, bool]:
|
|
hypothesis_feedback = (
|
|
Environment(undefined=StrictUndefined)
|
|
.from_string(prompt_dict["hypothesis_and_feedback"])
|
|
.render(trace=trace)
|
|
)
|
|
context_dict = {
|
|
"hypothesis_and_feedback": hypothesis_feedback,
|
|
"RAG": "",
|
|
"hypothesis_output_format": prompt_dict["hypothesis_output_format"],
|
|
"hypothesis_specification": prompt_dict["model_hypothesis_specification"]
|
|
}
|
|
return context_dict, True
|
|
|
|
def convert_response(self, response: str) -> ModelHypothesis:
|
|
response_dict = json.loads(response)
|
|
hypothesis = DMModelHypothesis(hypothesis=response_dict["hypothesis"], reason=response_dict["reason"])
|
|
return hypothesis
|
|
|
|
|
|
class DMModelHypothesis2Experiment(ModelHypothesis2Experiment):
|
|
def prepare_context(self, hypothesis: Hypothesis, trace: Trace) -> Tuple[dict, bool]:
|
|
scenario = trace.scen.get_scenario_all_desc()
|
|
experiment_output_format = prompt_dict["model_experiment_output_format"]
|
|
|
|
hypothesis_and_feedback = (
|
|
Environment(undefined=StrictUndefined)
|
|
.from_string(prompt_dict["hypothesis_and_feedback"])
|
|
.render(trace=trace)
|
|
)
|
|
|
|
experiment_list: List[ModelExperiment] = [t[1] for t in trace.hist]
|
|
|
|
model_list = []
|
|
for experiment in experiment_list:
|
|
model_list.extend(experiment.sub_tasks)
|
|
|
|
return {
|
|
"target_hypothesis": str(hypothesis),
|
|
"scenario": scenario,
|
|
"hypothesis_and_feedback": hypothesis_and_feedback,
|
|
"experiment_output_format": experiment_output_format,
|
|
"target_list": model_list,
|
|
"RAG": ...,
|
|
}, True
|
|
|
|
def convert_response(self, response: str, trace: Trace) -> ModelExperiment:
|
|
response_dict = json.loads(response)
|
|
tasks = []
|
|
for model_name in response_dict:
|
|
description = response_dict[model_name]["description"]
|
|
architecture = response_dict[model_name]["architecture"]
|
|
hyperparameters = response_dict[model_name]["hyperparameters"]
|
|
model_type = response_dict[model_name]["model_type"]
|
|
tasks.append(ModelTask(model_name, description, architecture, hyperparameters, model_type))
|
|
exp = DMModelExperiment(tasks)
|
|
exp.based_experiments = [t[1] for t in trace.hist if t[2]]
|
|
return exp
|