gavindiaz/NexQuant
1
0
Fork 0
mirror of https://github.com/NicolasBohn/NexQuant.git synced 2026-07-27 23:47:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: a unified CoSTEER to fit more scenarios (#491)

Browse Source 
* Use ExtendedBaseSettings to replace BaseSettings

* update a more general way to pass the default setting

* update all code

* fix CI

* fix CI

* fix qlib scenario

* fix CI

* fix CI

* fix CI & add data science interfaces

* remove redundant code

* abandon costeer knowledge base v1

---------

Co-authored-by: Xu Yang <xuyang1@microsoft.com>
Co-authored-by: XianBW <36835909+XianBW@users.noreply.github.com>
This commit is contained in:
Xu Yang
2024-11-25 16:27:34 +08:00
committed by GitHub
parent a4602bdcc2
commit fce241b9f9
72 changed files with 1793 additions and 1817 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rdagent/components/coder/model_coder/CoSTEER/__init__.py
-94
View File
@@ -1,94 +0,0 @@
import pickle
from pathlib import Path
from rdagent.components.coder.model_coder.conf import MODEL_IMPL_SETTINGS
from rdagent.components.coder.model_coder.CoSTEER.evaluators import (
ModelCoderMultiEvaluator,
)
from rdagent.components.coder.model_coder.CoSTEER.evolvable_subjects import (
ModelEvolvingItem,
)
from rdagent.components.coder.model_coder.CoSTEER.evolving_agent import ModelRAGEvoAgent
from rdagent.components.coder.model_coder.CoSTEER.evolving_strategy import (
ModelCoderEvolvingStrategy,
)
from rdagent.components.coder.model_coder.CoSTEER.knowledge_management import (
ModelKnowledgeBase,
ModelRAGStrategy,
)
from rdagent.components.coder.model_coder.model import ModelExperiment
from rdagent.core.developer import Developer
from rdagent.core.evolving_agent import RAGEvoAgent
class ModelCoSTEER(Developer[ModelExperiment]):
def __init__(
self,
*args,
with_knowledge: bool = True,
with_feedback: bool = True,
knowledge_self_gen: bool = True,
filter_final_evo: bool = True,
**kwargs,
) -> None:
super().__init__(*args, **kwargs)
self.max_loop = MODEL_IMPL_SETTINGS.max_loop
self.knowledge_base_path = (
Path(MODEL_IMPL_SETTINGS.knowledge_base_path)
if MODEL_IMPL_SETTINGS.knowledge_base_path is not None
else None
)
self.new_knowledge_base_path = (
Path(MODEL_IMPL_SETTINGS.new_knowledge_base_path)
if MODEL_IMPL_SETTINGS.new_knowledge_base_path is not None
else None
)
self.with_knowledge = with_knowledge
self.with_feedback = with_feedback
self.knowledge_self_gen = knowledge_self_gen
self.filter_final_evo = filter_final_evo
self.evolving_strategy = ModelCoderEvolvingStrategy(scen=self.scen)
self.model_evaluator = ModelCoderMultiEvaluator(scen=self.scen)
def load_or_init_knowledge_base(self, former_knowledge_base_path: Path = None, component_init_list: list = []):
if former_knowledge_base_path is not None and former_knowledge_base_path.exists():
model_knowledge_base = pickle.load(open(former_knowledge_base_path, "rb"))
if not isinstance(model_knowledge_base, ModelKnowledgeBase):
raise ValueError("The former knowledge base is not compatible with the current version")
else:
model_knowledge_base = ModelKnowledgeBase()
return model_knowledge_base
def develop(self, exp: ModelExperiment) -> ModelExperiment:
# init knowledge base
model_knowledge_base = self.load_or_init_knowledge_base(
former_knowledge_base_path=self.knowledge_base_path,
component_init_list=[],
)
# init rag method
self.rag = ModelRAGStrategy(model_knowledge_base)
# init intermediate items
model_experiment = ModelEvolvingItem.from_experiment(exp)
self.evolve_agent = ModelRAGEvoAgent(
max_loop=self.max_loop,
evolving_strategy=self.evolving_strategy,
rag=self.rag,
with_knowledge=self.with_knowledge,
with_feedback=self.with_feedback,
knowledge_self_gen=self.knowledge_self_gen,
)
model_experiment = self.evolve_agent.multistep_evolve(
model_experiment,
self.model_evaluator,
filter_final_evo=self.filter_final_evo,
)
# save new knowledge base
if self.new_knowledge_base_path is not None:
pickle.dump(model_knowledge_base, open(self.new_knowledge_base_path, "wb"))
exp.sub_workspace_list = model_experiment.sub_workspace_list
return exp
rdagent/components/coder/model_coder/CoSTEER/evolvable_subjects.py
-36
View File
@@ -1,36 +0,0 @@
from rdagent.components.coder.model_coder.model import (
ModelExperiment,
ModelFBWorkspace,
ModelTask,
)
from rdagent.core.evolving_framework import EvolvableSubjects
from rdagent.log import rdagent_logger as logger
class ModelEvolvingItem(ModelExperiment, EvolvableSubjects):
"""
Intermediate item of model implementation.
"""
def __init__(
self,
sub_tasks: list[ModelTask],
sub_gt_implementations: list[ModelFBWorkspace] = None,
):
ModelExperiment.__init__(self, sub_tasks=sub_tasks)
if sub_gt_implementations is not None and len(
sub_gt_implementations,
) != len(self.sub_tasks):
self.sub_gt_implementations = None
logger.warning(
"The length of sub_gt_implementations is not equal to the length of sub_tasks, set sub_gt_implementations to None",
)
else:
self.sub_gt_implementations = sub_gt_implementations
@classmethod
def from_experiment(cls, exp: ModelExperiment) -> "ModelEvolvingItem":
ei = cls(sub_tasks=exp.sub_tasks)
ei.based_experiments = exp.based_experiments
ei.experiment_workspace = exp.experiment_workspace
return ei
rdagent/components/coder/model_coder/CoSTEER/evolving_agent.py
-19
View File
@@ -1,19 +0,0 @@
from rdagent.components.coder.model_coder.CoSTEER.evaluators import ModelCoderFeedback
from rdagent.components.coder.model_coder.CoSTEER.evolvable_subjects import (
ModelEvolvingItem,
)
from rdagent.core.evaluation import Feedback
from rdagent.core.evolving_agent import RAGEvoAgent
from rdagent.core.evolving_framework import EvolvableSubjects
class ModelRAGEvoAgent(RAGEvoAgent):
def filter_evolvable_subjects_by_feedback(self, evo: EvolvableSubjects, feedback: Feedback) -> EvolvableSubjects:
assert isinstance(evo, ModelEvolvingItem)
assert isinstance(feedback, list)
assert len(evo.sub_workspace_list) == len(feedback)
for index in range(len(evo.sub_workspace_list)):
if not feedback[index].final_decision:
evo.sub_workspace_list[index].clear()
return evo
rdagent/components/coder/model_coder/CoSTEER/evolving_strategy.py
-158
View File
@@ -1,158 +0,0 @@
import json
from copy import deepcopy
from pathlib import Path
from jinja2 import Environment, StrictUndefined
from rdagent.components.coder.model_coder.conf import MODEL_IMPL_SETTINGS
from rdagent.components.coder.model_coder.CoSTEER.evolvable_subjects import (
ModelEvolvingItem,
)
from rdagent.components.coder.model_coder.CoSTEER.knowledge_management import (
ModelQueriedKnowledge,
)
from rdagent.components.coder.model_coder.model import (
ModelExperiment,
ModelFBWorkspace,
ModelTask,
)
from rdagent.core.conf import RD_AGENT_SETTINGS
from rdagent.core.evolving_framework import EvolvingStrategy
from rdagent.core.prompts import Prompts
from rdagent.core.utils import multiprocessing_wrapper
from rdagent.oai.llm_conf import LLM_SETTINGS
from rdagent.oai.llm_utils import APIBackend
from rdagent.scenarios.kaggle.experiment.kaggle_experiment import KG_MODEL_MAPPING
coder_prompts = Prompts(file_path=Path(__file__).parent.parent / "prompts.yaml")
class ModelCoderEvolvingStrategy(EvolvingStrategy):
def implement_one_model(
self,
target_task: ModelTask,
queried_knowledge: ModelQueriedKnowledge = None,
current_exp: ModelExperiment = None, # Add this parameter
) -> str:
model_information_str = target_task.get_task_information()
model_type = target_task.model_type
if len(current_exp.based_experiments) == 0:
current_code = None
else:
current_code = ""
sota_exp_code_dict = current_exp.based_experiments[-1].experiment_workspace.code_dict
if target_task.version == 2:
if model_type in KG_MODEL_MAPPING:
current_code = sota_exp_code_dict.get(KG_MODEL_MAPPING[model_type], None)
elif "model.py" in sota_exp_code_dict:
current_code = sota_exp_code_dict["model.py"]
else:
current_code = None
elif target_task.version == 1:
current_code = sota_exp_code_dict.get("model.py", None)
if queried_knowledge is not None and model_information_str in queried_knowledge.success_task_to_knowledge_dict:
return queried_knowledge.success_task_to_knowledge_dict[model_information_str].implementation
elif queried_knowledge is not None and model_information_str in queried_knowledge.failed_task_info_set:
return None
else:
queried_similar_successful_knowledge = (
queried_knowledge.working_task_to_similar_successful_knowledge_dict[model_information_str]
if queried_knowledge is not None
else []
)
queried_former_failed_knowledge = (
queried_knowledge.working_task_to_former_failed_knowledge_dict[model_information_str]
if queried_knowledge is not None
else []
)
queried_former_failed_knowledge_to_render = queried_former_failed_knowledge
system_prompt = (
Environment(undefined=StrictUndefined)
.from_string(
coder_prompts["evolving_strategy_model_coder"]["system"],
)
.render(
scenario=self.scen.get_scenario_all_desc(filtered_tag=target_task.model_type),
queried_former_failed_knowledge=queried_former_failed_knowledge_to_render,
current_code=current_code,
)
)
queried_similar_successful_knowledge_to_render = queried_similar_successful_knowledge
for _ in range(10): # max attempt to reduce the length of user_prompt
user_prompt = (
Environment(undefined=StrictUndefined)
.from_string(
coder_prompts["evolving_strategy_model_coder"]["user"],
)
.render(
model_information_str=model_information_str,
queried_similar_successful_knowledge=queried_similar_successful_knowledge_to_render,
queried_former_failed_knowledge=queried_former_failed_knowledge_to_render,
)
.strip("\n")
)
if (
APIBackend().build_messages_and_calculate_token(
user_prompt=user_prompt,
system_prompt=system_prompt,
)
< LLM_SETTINGS.chat_token_limit
):
break
elif len(queried_former_failed_knowledge_to_render) > 1:
queried_former_failed_knowledge_to_render = queried_former_failed_knowledge_to_render[1:]
elif len(queried_similar_successful_knowledge_to_render) > 1:
queried_similar_successful_knowledge_to_render = queried_similar_successful_knowledge_to_render[1:]
code = json.loads(
APIBackend(
use_chat_cache=MODEL_IMPL_SETTINGS.coder_use_cache
).build_messages_and_create_chat_completion(
user_prompt=user_prompt,
system_prompt=system_prompt,
json_mode=True,
),
)["code"]
return code
def evolve(
self,
*,
evo: ModelEvolvingItem,
queried_knowledge: ModelQueriedKnowledge | None = None,
**kwargs,
) -> ModelEvolvingItem:
# 1.找出需要evolve的model
to_be_finished_task_index = []
for index, target_model_task in enumerate(evo.sub_tasks):
target_model_task_desc = target_model_task.get_task_information()
if target_model_task_desc in queried_knowledge.success_task_to_knowledge_dict:
evo.sub_workspace_list[index] = queried_knowledge.success_task_to_knowledge_dict[
target_model_task_desc
].implementation
elif (
target_model_task_desc not in queried_knowledge.success_task_to_knowledge_dict
and target_model_task_desc not in queried_knowledge.failed_task_info_set
):
to_be_finished_task_index.append(index)
result = multiprocessing_wrapper(
[
(self.implement_one_model, (evo.sub_tasks[target_index], queried_knowledge, evo))
for target_index in to_be_finished_task_index
],
n=RD_AGENT_SETTINGS.multi_proc_n,
)
for index, target_index in enumerate(to_be_finished_task_index):
evo.sub_workspace_list[target_index] = ModelFBWorkspace(target_task=evo.sub_tasks[target_index])
evo.sub_workspace_list[target_index].inject_code(**{"model.py": result[index]})
evo.corresponding_selection = to_be_finished_task_index
return evo
rdagent/components/coder/model_coder/CoSTEER/knowledge_management.py
-171
View File
@@ -1,171 +0,0 @@
from pathlib import Path
from rdagent.components.coder.model_coder.conf import MODEL_IMPL_SETTINGS
from rdagent.components.coder.model_coder.CoSTEER.evaluators import ModelCoderFeedback
from rdagent.components.coder.model_coder.model import ModelTask
from rdagent.core.evolving_framework import (
EvolvableSubjects,
EvolvingKnowledgeBase,
EvoStep,
Knowledge,
QueriedKnowledge,
RAGStrategy,
)
from rdagent.core.experiment import Workspace
from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list
class ModelKnowledge(Knowledge):
def __init__(
self,
target_task: ModelTask,
implementation: Workspace,
feedback: ModelCoderFeedback,
) -> None:
"""
Initialize a ModelKnowledge object. The ModelKnowledge object is used to store a model implementation without the ground truth code and value.
Args:
model (Model): The model object associated with the KnowledgeManagement.
Returns:
None
"""
self.target_task = target_task
self.implementation = implementation.copy()
self.feedback = feedback
def get_implementation_and_feedback_str(self) -> str:
return f"""------------------Model implementation code:------------------
{self.implementation.code}
------------------Model implementation feedback:------------------
{self.feedback!s}
"""
class ModelQueriedKnowledge(QueriedKnowledge):
def __init__(self, success_task_to_knowledge_dict: dict = {}, failed_task_info_set: set = set()) -> None:
self.success_task_to_knowledge_dict = success_task_to_knowledge_dict
self.failed_task_info_set = failed_task_info_set
self.working_task_to_former_failed_knowledge_dict = dict()
self.working_task_to_similar_successful_knowledge_dict = dict()
class ModelKnowledgeBase(EvolvingKnowledgeBase):
def __init__(self, path: str | Path = None) -> None:
self.implementation_trace: dict[str, ModelKnowledge] = dict()
self.success_task_info_set: set[str] = set()
self.task_to_embedding = dict()
super().__init__(path)
def query(self) -> QueriedKnowledge | None:
"""
Query the knowledge base to get the queried knowledge. So far is handled in RAG strategy.
"""
raise NotImplementedError
class ModelRAGStrategy(RAGStrategy):
def __init__(self, knowledgebase: ModelKnowledgeBase) -> None:
super().__init__(knowledgebase)
self.current_generated_trace_count = 0
def generate_knowledge(
self,
evolving_trace: list[EvoStep],
*,
return_knowledge: bool = False,
) -> Knowledge | None:
if len(evolving_trace) == self.current_generated_trace_count:
return
else:
for trace_index in range(
self.current_generated_trace_count,
len(evolving_trace),
):
evo_step = evolving_trace[trace_index]
implementations = evo_step.evolvable_subjects
feedback = evo_step.feedback
for task_index in range(len(implementations.sub_tasks)):
target_task = implementations.sub_tasks[task_index]
target_task_information = target_task.get_task_information()
implementation = implementations.sub_workspace_list[task_index]
single_feedback = feedback[task_index]
if single_feedback is None:
continue
single_knowledge = ModelKnowledge(
target_task=target_task,
implementation=implementation,
feedback=single_feedback,
)
if target_task_information not in self.knowledgebase.success_task_info_set:
self.knowledgebase.implementation_trace.setdefault(
target_task_information,
[],
).append(single_knowledge)
if single_feedback.final_decision == True:
self.knowledgebase.success_task_info_set.add(
target_task_information,
)
self.current_generated_trace_count = len(evolving_trace)
def query(
self,
evo: EvolvableSubjects,
evolving_trace: list[EvoStep],
) -> QueriedKnowledge | None:
query_former_trace_limit = MODEL_IMPL_SETTINGS.query_former_trace_limit
query_similar_success_limit = MODEL_IMPL_SETTINGS.query_similar_success_limit
fail_task_trial_limit = MODEL_IMPL_SETTINGS.fail_task_trial_limit
queried_knowledge = ModelQueriedKnowledge()
for target_model_task in evo.sub_tasks:
target_model_task_information = target_model_task.get_task_information()
if target_model_task_information in self.knowledgebase.success_task_info_set:
queried_knowledge.success_task_to_knowledge_dict[target_model_task_information] = (
self.knowledgebase.implementation_trace[target_model_task_information][-1]
)
elif (
len(
self.knowledgebase.implementation_trace.setdefault(
target_model_task_information,
[],
),
)
>= fail_task_trial_limit
):
queried_knowledge.failed_task_info_set.add(target_model_task_information)
else:
queried_knowledge.working_task_to_former_failed_knowledge_dict[target_model_task_information] = (
self.knowledgebase.implementation_trace.setdefault(
target_model_task_information,
[],
)[-query_former_trace_limit:]
)
knowledge_base_success_task_list = list(
self.knowledgebase.success_task_info_set,
)
similarity = calculate_embedding_distance_between_str_list(
[target_model_task_information],
knowledge_base_success_task_list,
)[0]
similar_indexes = sorted(
range(len(similarity)),
key=lambda i: similarity[i],
reverse=True,
)[:query_similar_success_limit]
similar_successful_knowledge = [
self.knowledgebase.implementation_trace.setdefault(
knowledge_base_success_task_list[index],
[],
)[-1]
for index in similar_indexes
]
queried_knowledge.working_task_to_similar_successful_knowledge_dict[target_model_task_information] = (
similar_successful_knowledge
)
return queried_knowledge
rdagent/components/coder/model_coder/__init__.py
+21
View File
@@ -0,0 +1,21 @@
from rdagent.components.coder.CoSTEER import CoSTEER
from rdagent.components.coder.CoSTEER.config import CoSTEER_SETTINGS
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiEvaluator
from rdagent.components.coder.model_coder.evaluators import ModelCoSTEEREvaluator
from rdagent.components.coder.model_coder.evolving_strategy import (
ModelMultiProcessEvolvingStrategy,
)
from rdagent.core.scenario import Scenario
class ModelCoSTEER(CoSTEER):
def __init__(
self,
scen: Scenario,
*args,
**kwargs,
) -> None:
eva = CoSTEERMultiEvaluator(ModelCoSTEEREvaluator(scen=scen), scen=scen)
es = ModelMultiProcessEvolvingStrategy(scen=scen, settings=CoSTEER_SETTINGS)
super().__init__(*args, settings=CoSTEER_SETTINGS, eva=eva, es=es, evolving_version=2, scen=scen, **kwargs)
rdagent/components/coder/model_coder/conf.py
-23
View File
@@ -1,23 +0,0 @@
from pathlib import Path
from typing import Union
from pydantic_settings import BaseSettings
class ModelImplSettings(BaseSettings):
class Config:
env_prefix = "MODEL_CODER_" # Use MODEL_CODER_ as prefix for environment variables
coder_use_cache: bool = False
knowledge_base_path: Union[str, None] = None
new_knowledge_base_path: Union[str, None] = None
max_loop: int = 10
query_former_trace_limit: int = 5
query_similar_success_limit: int = 5
fail_task_trial_limit: int = 20
MODEL_IMPL_SETTINGS = ModelImplSettings()
rdagent/components/coder/model_coder/CoSTEER/evaluators.py → rdagent/components/coder/model_coder/eva_utils.py
+2 -162
View File
@@ -1,27 +1,18 @@
import json
import random
from pathlib import Path
from typing import List, Tuple
from typing import Tuple
import numpy as np
from jinja2 import Environment, StrictUndefined
from rdagent.components.coder.model_coder.conf import MODEL_IMPL_SETTINGS
from rdagent.components.coder.model_coder.CoSTEER.evolvable_subjects import (
ModelEvolvingItem,
)
from rdagent.components.coder.model_coder.model import ModelFBWorkspace, ModelTask
from rdagent.core.conf import RD_AGENT_SETTINGS
from rdagent.core.evaluation import Evaluator
from rdagent.core.evolving_framework import QueriedKnowledge
from rdagent.core.experiment import Task, Workspace
from rdagent.core.prompts import Prompts
from rdagent.core.utils import multiprocessing_wrapper
from rdagent.log import rdagent_logger as logger
from rdagent.oai.llm_conf import LLM_SETTINGS
from rdagent.oai.llm_utils import APIBackend
evaluate_prompts = Prompts(file_path=Path(__file__).parent.parent / "prompts.yaml")
evaluate_prompts = Prompts(file_path=Path(__file__).parent / "prompts.yaml")
def shape_evaluator(prediction: np.ndarray, target_shape: Tuple = None) -> Tuple[str, bool]:
@@ -193,154 +184,3 @@ class ModelFinalEvaluator(Evaluator):
final_evaluation_dict["final_feedback"],
final_evaluation_dict["final_decision"],
)
class ModelCoderFeedback:
"""This feedback includes all the content to the model coder"""
def __init__(
self,
execution_feedback: str,
shape_feedback: str,
value_feedback: str,
code_feedback: str,
final_feedback: str,
final_decision: bool,
):
self.execution_feedback: str = execution_feedback
self.shape_feedback: str = shape_feedback
self.value_feedback: str = value_feedback
self.code_feedback: str = code_feedback
self.final_feedback: str = final_feedback
self.final_decision: str = final_decision
def __str__(self) -> str:
return f"""------------------Model Execution Feedback------------------
{self.execution_feedback}
------------------Model Shape Feedback------------------
{self.shape_feedback}
------------------Model Value Feedback------------------
{self.value_feedback}
------------------Model Code Feedback------------------
{self.code_feedback}
------------------Model Final Feedback------------------
{self.final_feedback}
------------------Model Final Decision------------------
This implementation is {'SUCCESS' if self.final_decision else 'FAIL'}.
"""
class ModelCoderEvaluator(Evaluator):
def evaluate(
self,
target_task: Task,
implementation: Workspace,
gt_implementation: Workspace,
queried_knowledge: QueriedKnowledge = None,
**kwargs,
) -> ModelCoderFeedback:
target_task_information = target_task.get_task_information()
if (
queried_knowledge is not None
and target_task_information in queried_knowledge.success_task_to_knowledge_dict
):
return queried_knowledge.success_task_to_knowledge_dict[target_task_information].feedback
elif queried_knowledge is not None and target_task_information in queried_knowledge.failed_task_info_set:
return ModelCoderFeedback(
execution_feedback="This task has failed too many times, skip implementation.",
shape_feedback="This task has failed too many times, skip implementation.",
value_feedback="This task has failed too many times, skip implementation.",
code_feedback="This task has failed too many times, skip implementation.",
final_feedback="This task has failed too many times, skip implementation.",
final_decision=False,
)
assert isinstance(target_task, ModelTask)
# NOTE: Use fixed input to test the model to avoid randomness
batch_size = 8
num_features = 30
num_timesteps = 40
input_value = 0.4
param_init_value = 0.6
assert isinstance(implementation, ModelFBWorkspace)
model_execution_feedback, gen_np_array = implementation.execute(
batch_size=batch_size,
num_features=num_features,
num_timesteps=num_timesteps,
input_value=input_value,
param_init_value=param_init_value,
)
if gt_implementation is not None:
assert isinstance(gt_implementation, ModelFBWorkspace)
_, gt_np_array = gt_implementation.execute(
batch_size=batch_size,
num_features=num_features,
num_timesteps=num_timesteps,
input_value=input_value,
param_init_value=param_init_value,
)
else:
gt_np_array = None
shape_feedback, shape_decision = shape_evaluator(
gen_np_array,
(batch_size, self.scen.model_output_channel if hasattr(self.scen, "model_output_channel") else 1),
)
value_feedback, value_decision = value_evaluator(gen_np_array, gt_np_array)
code_feedback, _ = ModelCodeEvaluator(scen=self.scen).evaluate(
target_task=target_task,
implementation=implementation,
gt_implementation=gt_implementation,
model_execution_feedback=model_execution_feedback,
model_value_feedback="\n".join([shape_feedback, value_feedback]),
)
final_feedback, final_decision = ModelFinalEvaluator(scen=self.scen).evaluate(
target_task=target_task,
implementation=implementation,
gt_implementation=gt_implementation,
model_execution_feedback=model_execution_feedback,
model_value_feedback=value_feedback,
model_code_feedback=code_feedback,
)
return ModelCoderFeedback(
execution_feedback=model_execution_feedback,
shape_feedback=shape_feedback,
value_feedback=value_feedback,
code_feedback=code_feedback,
final_feedback=final_feedback,
final_decision=final_decision,
)
class ModelCoderMultiEvaluator(Evaluator):
def evaluate(
self,
evo: ModelEvolvingItem,
queried_knowledge: QueriedKnowledge = None,
**kwargs,
) -> List[ModelCoderFeedback]:
multi_implementation_feedback = multiprocessing_wrapper(
[
(
ModelCoderEvaluator(scen=self.scen).evaluate,
(
evo.sub_tasks[index],
evo.sub_workspace_list[index],
evo.sub_gt_implementations[index] if evo.sub_gt_implementations is not None else None,
queried_knowledge,
),
)
for index in range(len(evo.sub_tasks))
],
n=RD_AGENT_SETTINGS.multi_proc_n,
)
final_decision = [
None if single_feedback is None else single_feedback.final_decision
for single_feedback in multi_implementation_feedback
]
logger.info(f"Final decisions: {final_decision} True count: {final_decision.count(True)}")
return multi_implementation_feedback
rdagent/components/coder/model_coder/evaluators.py
+103
View File
@@ -0,0 +1,103 @@
from rdagent.components.coder.CoSTEER.evaluators import (
CoSTEEREvaluator,
CoSTEERMultiFeedback,
CoSTEERSingleFeedback,
)
from rdagent.components.coder.model_coder.eva_utils import (
ModelCodeEvaluator,
ModelFinalEvaluator,
shape_evaluator,
value_evaluator,
)
from rdagent.components.coder.model_coder.model import ModelFBWorkspace, ModelTask
from rdagent.core.evolving_framework import QueriedKnowledge
from rdagent.core.experiment import Task, Workspace
ModelSingleFeedback = CoSTEERSingleFeedback
ModelMultiFeedback = CoSTEERMultiFeedback
class ModelCoSTEEREvaluator(CoSTEEREvaluator):
def evaluate(
self,
target_task: Task,
implementation: Workspace,
gt_implementation: Workspace,
queried_knowledge: QueriedKnowledge = None,
**kwargs,
) -> ModelSingleFeedback:
target_task_information = target_task.get_task_information()
if (
queried_knowledge is not None
and target_task_information in queried_knowledge.success_task_to_knowledge_dict
):
return queried_knowledge.success_task_to_knowledge_dict[target_task_information].feedback
elif queried_knowledge is not None and target_task_information in queried_knowledge.failed_task_info_set:
return ModelSingleFeedback(
execution_feedback="This task has failed too many times, skip implementation.",
shape_feedback="This task has failed too many times, skip implementation.",
value_feedback="This task has failed too many times, skip implementation.",
code_feedback="This task has failed too many times, skip implementation.",
final_feedback="This task has failed too many times, skip implementation.",
final_decision=False,
)
assert isinstance(target_task, ModelTask)
# NOTE: Use fixed input to test the model to avoid randomness
batch_size = 8
num_features = 30
num_timesteps = 40
input_value = 0.4
param_init_value = 0.6
assert isinstance(implementation, ModelFBWorkspace)
model_execution_feedback, gen_np_array = implementation.execute(
batch_size=batch_size,
num_features=num_features,
num_timesteps=num_timesteps,
input_value=input_value,
param_init_value=param_init_value,
)
if gt_implementation is not None:
assert isinstance(gt_implementation, ModelFBWorkspace)
_, gt_np_array = gt_implementation.execute(
batch_size=batch_size,
num_features=num_features,
num_timesteps=num_timesteps,
input_value=input_value,
param_init_value=param_init_value,
)
else:
gt_np_array = None
shape_feedback, shape_decision = shape_evaluator(
gen_np_array,
(batch_size, self.scen.model_output_channel if hasattr(self.scen, "model_output_channel") else 1),
)
value_feedback, value_decision = value_evaluator(gen_np_array, gt_np_array)
code_feedback, _ = ModelCodeEvaluator(scen=self.scen).evaluate(
target_task=target_task,
implementation=implementation,
gt_implementation=gt_implementation,
model_execution_feedback=model_execution_feedback,
model_value_feedback="\n".join([shape_feedback, value_feedback]),
)
final_feedback, final_decision = ModelFinalEvaluator(scen=self.scen).evaluate(
target_task=target_task,
implementation=implementation,
gt_implementation=gt_implementation,
model_execution_feedback=model_execution_feedback,
model_value_feedback=value_feedback,
model_code_feedback=code_feedback,
)
return ModelSingleFeedback(
execution_feedback=model_execution_feedback,
shape_feedback=shape_feedback,
value_feedback=value_feedback,
code_feedback=code_feedback,
final_feedback=final_feedback,
final_decision=final_decision,
value_generated_flag=(gen_np_array is not None),
final_decision_based_on_gt=(gt_implementation is not None),
)
rdagent/components/coder/model_coder/evolving_strategy.py
+106
View File
@@ -0,0 +1,106 @@
import json
from pathlib import Path
from jinja2 import Environment, StrictUndefined
from rdagent.components.coder.CoSTEER.config import CoSTEER_SETTINGS
from rdagent.components.coder.CoSTEER.evolving_strategy import (
MultiProcessEvolvingStrategy,
)
from rdagent.components.coder.CoSTEER.knowledge_management import (
CoSTEERQueriedKnowledge,
CoSTEERQueriedKnowledgeV2,
)
from rdagent.components.coder.model_coder.model import (
ModelExperiment,
ModelFBWorkspace,
ModelTask,
)
from rdagent.core.prompts import Prompts
from rdagent.oai.llm_conf import LLM_SETTINGS
from rdagent.oai.llm_utils import APIBackend
coder_prompts = Prompts(file_path=Path(__file__).parent / "prompts.yaml")
class ModelMultiProcessEvolvingStrategy(MultiProcessEvolvingStrategy):
def implement_one_task(
self,
target_task: ModelTask,
queried_knowledge: CoSTEERQueriedKnowledge = None,
) -> str:
model_information_str = target_task.get_task_information()
queried_similar_successful_knowledge = (
queried_knowledge.task_to_similar_task_successful_knowledge[model_information_str]
if queried_knowledge is not None
else []
)
queried_former_failed_knowledge = (
queried_knowledge.task_to_former_failed_traces[model_information_str]
if queried_knowledge is not None
else []
)
queried_former_failed_knowledge_to_render = (
queried_former_failed_knowledge[0]
if isinstance(queried_knowledge, CoSTEERQueriedKnowledgeV2)
else queried_former_failed_knowledge
)
system_prompt = (
Environment(undefined=StrictUndefined)
.from_string(
coder_prompts["evolving_strategy_model_coder"]["system"],
)
.render(
scenario=self.scen.get_scenario_all_desc(filtered_tag=target_task.model_type),
queried_former_failed_knowledge=queried_former_failed_knowledge_to_render,
current_code=target_task.base_code,
)
)
queried_similar_successful_knowledge_to_render = queried_similar_successful_knowledge
for _ in range(10): # max attempt to reduce the length of user_prompt
user_prompt = (
Environment(undefined=StrictUndefined)
.from_string(
coder_prompts["evolving_strategy_model_coder"]["user"],
)
.render(
model_information_str=model_information_str,
queried_similar_successful_knowledge=queried_similar_successful_knowledge_to_render,
queried_former_failed_knowledge=queried_former_failed_knowledge_to_render,
)
.strip("\n")
)
if (
APIBackend().build_messages_and_calculate_token(
user_prompt=user_prompt,
system_prompt=system_prompt,
)
< LLM_SETTINGS.chat_token_limit
):
break
elif len(queried_former_failed_knowledge_to_render) > 1:
queried_former_failed_knowledge_to_render = queried_former_failed_knowledge_to_render[1:]
elif len(queried_similar_successful_knowledge_to_render) > 1:
queried_similar_successful_knowledge_to_render = queried_similar_successful_knowledge_to_render[1:]
code = json.loads(
APIBackend(use_chat_cache=CoSTEER_SETTINGS.coder_use_cache).build_messages_and_create_chat_completion(
user_prompt=user_prompt,
system_prompt=system_prompt,
json_mode=True,
),
)["code"]
return code
def assign_code_list_to_evo(self, code_list, evo):
for index in range(len(evo.sub_tasks)):
if code_list[index] is None:
continue
if evo.sub_workspace_list[index] is None:
evo.sub_workspace_list[index] = ModelFBWorkspace(target_task=evo.sub_tasks[index])
evo.sub_workspace_list[index].inject_code(**{"model.py": code_list[index]})
return evo
rdagent/components/coder/model_coder/model.py
+3 -3
View File
@@ -4,14 +4,14 @@ import traceback
from pathlib import Path
from typing import Dict, Optional
from rdagent.components.coder.model_coder.conf import MODEL_IMPL_SETTINGS
from rdagent.core.experiment import Experiment, FBWorkspace, Task
from rdagent.components.coder.CoSTEER.task import CoSTEERTask
from rdagent.core.experiment import Experiment, FBWorkspace
from rdagent.core.utils import cache_with_pickle
from rdagent.oai.llm_utils import md5_hash
from rdagent.utils.env import KGDockerEnv, QTDockerEnv
class ModelTask(Task):
class ModelTask(CoSTEERTask):
def __init__(
self,
name: str,
rdagent/components/coder/model_coder/prompts.yaml
+1 -1
View File
@@ -83,7 +83,7 @@ evolving_strategy_model_coder:
--------------Correct code to similar models:---------------
{% for similar_successful_knowledge in queried_similar_successful_knowledge %}
=====Model {{loop.index}}:=====
{{ similar_successful_knowledge.target_task.get_model_information() }}
{{ similar_successful_knowledge.target_task.get_task_information() }}
=====Code:=====
{{ similar_successful_knowledge.implementation.code }}
{% endfor %}