Files
NexQuant/rdagent/components/proposal/__init__.py
T
you-n-g 5d5084b87e feat: new-york-city-taxi-fare-prediction_template (#488)
* copy init version

* feat: new-york-city-taxi-fare-prediction_template

* add move to linear model

* Add more details about docker

* auto lint

* auto lint with new black
2024-11-15 16:47:44 +08:00

131 lines
4.4 KiB
Python

from abc import abstractmethod
from pathlib import Path
from typing import Tuple
from jinja2 import Environment, StrictUndefined
from rdagent.core.experiment import Experiment
from rdagent.core.prompts import Prompts
from rdagent.core.proposal import (
Hypothesis,
Hypothesis2Experiment,
HypothesisGen,
Scenario,
Trace,
)
from rdagent.oai.llm_utils import APIBackend
prompt_dict = Prompts(file_path=Path(__file__).parent / "prompts.yaml")
class LLMHypothesisGen(HypothesisGen):
def __init__(self, scen: Scenario):
super().__init__(scen)
# The following methods are scenario related so they should be implemented in the subclass
@abstractmethod
def prepare_context(self, trace: Trace) -> Tuple[dict, bool]: ...
@abstractmethod
def convert_response(self, response: str) -> Hypothesis: ...
def gen(self, trace: Trace) -> Hypothesis:
context_dict, json_flag = self.prepare_context(trace)
system_prompt = (
Environment(undefined=StrictUndefined)
.from_string(prompt_dict["hypothesis_gen"]["system_prompt"])
.render(
targets=self.targets,
scenario=self.scen.get_scenario_all_desc(filtered_tag="hypothesis_and_experiment"),
hypothesis_output_format=context_dict["hypothesis_output_format"],
hypothesis_specification=context_dict["hypothesis_specification"],
)
)
user_prompt = (
Environment(undefined=StrictUndefined)
.from_string(prompt_dict["hypothesis_gen"]["user_prompt"])
.render(
targets=self.targets,
hypothesis_and_feedback=context_dict["hypothesis_and_feedback"],
RAG=context_dict["RAG"],
)
)
resp = APIBackend().build_messages_and_create_chat_completion(user_prompt, system_prompt, json_mode=json_flag)
hypothesis = self.convert_response(resp)
return hypothesis
class FactorHypothesisGen(LLMHypothesisGen):
def __init__(self, scen: Scenario):
super().__init__(scen)
self.targets = "factors"
class ModelHypothesisGen(LLMHypothesisGen):
def __init__(self, scen: Scenario):
super().__init__(scen)
self.targets = "model tuning"
class FactorAndModelHypothesisGen(LLMHypothesisGen):
def __init__(self, scen: Scenario):
super().__init__(scen)
self.targets = "feature engineering and model building"
class LLMHypothesis2Experiment(Hypothesis2Experiment[Experiment]):
@abstractmethod
def prepare_context(self, hypothesis: Hypothesis, trace: Trace) -> Tuple[dict, bool]: ...
@abstractmethod
def convert_response(self, response: str, trace: Trace) -> Experiment: ...
def convert(self, hypothesis: Hypothesis, trace: Trace) -> Experiment:
context, json_flag = self.prepare_context(hypothesis, trace)
system_prompt = (
Environment(undefined=StrictUndefined)
.from_string(prompt_dict["hypothesis2experiment"]["system_prompt"])
.render(
targets=self.targets,
scenario=trace.scen.get_scenario_all_desc(filtered_tag="hypothesis_and_experiment"),
experiment_output_format=context["experiment_output_format"],
)
)
user_prompt = (
Environment(undefined=StrictUndefined)
.from_string(prompt_dict["hypothesis2experiment"]["user_prompt"])
.render(
targets=self.targets,
target_hypothesis=context["target_hypothesis"],
hypothesis_and_feedback=context["hypothesis_and_feedback"],
target_list=context["target_list"],
RAG=context["RAG"],
)
)
resp = APIBackend().build_messages_and_create_chat_completion(user_prompt, system_prompt, json_mode=json_flag)
return self.convert_response(resp, trace)
class FactorHypothesis2Experiment(LLMHypothesis2Experiment):
def __init__(self):
super().__init__()
self.targets = "factors"
class ModelHypothesis2Experiment(LLMHypothesis2Experiment):
def __init__(self):
super().__init__()
self.targets = "model tuning"
class FactorAndModelHypothesis2Experiment(LLMHypothesis2Experiment):
def __init__(self):
super().__init__()
self.targets = "feature engineering and model building"