Implement model (and some factor) coder with evolving (#52)

* store code into FBImplementation

* fix path related bugs

* fix a bug

* fix factor related small bugs

* re-submit all model related code

* new code to model coder

* finish the model evolving code

---------

Co-authored-by: xuyang1 <xuyang1@microsoft.com>
This commit is contained in:
Xu Yang
2024-07-10 15:45:43 +08:00
committed by GitHub
parent 828a624238
commit 720994c8e0
23 changed files with 1230 additions and 284 deletions
@@ -0,0 +1,86 @@
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_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.evolving_agent import RAGEvoAgent
from rdagent.core.task_generator import TaskGenerator
class ModelCoSTEER(TaskGenerator[ModelExperiment]):
def __init__(
self,
*args,
with_knowledge: bool = True,
with_feedback: bool = True,
knowledge_self_gen: 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.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 generate(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(sub_tasks=exp.sub_tasks)
self.evolve_agent = RAGEvoAgent(max_loop=self.max_loop, evolving_strategy=self.evolving_strategy, rag=self.rag)
model_experiment = self.evolve_agent.multistep_evolve(
model_experiment,
self.model_evaluator,
with_knowledge=self.with_knowledge,
with_feedback=self.with_feedback,
knowledge_self_gen=self.knowledge_self_gen,
)
# 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"))
self.knowledge_base = model_knowledge_base
return model_experiment
@@ -0,0 +1,333 @@
import json
import random
from pathlib import Path
from typing import List, Tuple
import numpy as np
import torch
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 ModelImplementation, 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 Implementation, Task
from rdagent.core.log import RDAgentLog
from rdagent.core.prompts import Prompts
from rdagent.core.utils import multiprocessing_wrapper
from rdagent.oai.llm_utils import APIBackend
evaluate_prompts = Prompts(file_path=Path(__file__).parent.parent / "prompts.yaml")
def shape_evaluator(prediction: torch.Tensor, target_shape: Tuple = None) -> Tuple[str, bool]:
if target_shape is None or prediction is None:
return "No output generated from the model. No shape evaluation conducted.", False
pre_shape = prediction.shape
if pre_shape == target_shape:
return "The shape of the output is correct.", True
else:
return f"The shape of the output is incorrect. Expected {target_shape}, but got {pre_shape}.", False
def reshape_tensor(original_tensor, target_shape):
new_tensor = torch.zeros(target_shape)
for i, dim in enumerate(original_tensor.shape):
new_tensor = new_tensor.narrow(i, 0, dim).copy_(original_tensor)
return new_tensor
def value_evaluator(
prediction: torch.Tensor,
target: torch.Tensor,
) -> Tuple[torch.Tensor, bool]:
if target is None or prediction is None:
return "No output generated from the model. No value evaluation conducted.", False
else:
# Calculate the mean absolute difference
diff = torch.mean(torch.abs(target - prediction)).item()
return (
f"The value of the output is correct. The mean absolute difference is {diff}.",
diff < 0.1,
)
class ModelCodeEvaluator(Evaluator):
def evaluate(
self,
target_task: Task,
implementation: Implementation,
gt_implementation: Implementation,
model_execution_feedback: str = "",
model_value_feedback: str = "",
):
assert isinstance(target_task, ModelTask)
assert isinstance(implementation, ModelImplementation)
if gt_implementation is not None:
assert isinstance(gt_implementation, ModelImplementation)
model_task_information = target_task.get_information()
code = implementation.code
system_prompt = (
Environment(undefined=StrictUndefined)
.from_string(evaluate_prompts["evaluator_code_feedback"]["system"])
.render(scenario=self.scen.get_scenario_all_desc() if self.scen is not None else "No scenario description.")
)
execution_feedback_to_render = model_execution_feedback
for _ in range(10): # 10 times to split the content is enough
user_prompt = (
Environment(undefined=StrictUndefined)
.from_string(
evaluate_prompts["evaluator_code_feedback"]["user"],
)
.render(
model_information=model_task_information,
code=code,
model_execution_feedback=execution_feedback_to_render,
model_value_feedback=model_value_feedback,
gt_code=gt_implementation.code if gt_implementation else None,
)
)
if (
APIBackend().build_messages_and_calculate_token(
user_prompt=user_prompt,
system_prompt=system_prompt,
)
> RD_AGENT_SETTINGS.chat_token_limit
):
execution_feedback_to_render = execution_feedback_to_render[len(execution_feedback_to_render) // 2 :]
else:
break
critic_response = APIBackend().build_messages_and_create_chat_completion(
user_prompt=user_prompt,
system_prompt=system_prompt,
json_mode=False,
)
return critic_response, None
class ModelFinalEvaluator(Evaluator):
def evaluate(
self,
target_task: Task,
implementation: Implementation,
gt_implementation: Implementation,
model_execution_feedback: str,
model_value_feedback: str,
model_code_feedback: str,
):
assert isinstance(target_task, ModelTask)
assert isinstance(implementation, ModelImplementation)
if gt_implementation is not None:
assert isinstance(gt_implementation, ModelImplementation)
system_prompt = (
Environment(undefined=StrictUndefined)
.from_string(evaluate_prompts["evaluator_final_feedback"]["system"])
.render(scenario=self.scen.get_scenario_all_desc() if self.scen is not None else "No scenario description.")
)
execution_feedback_to_render = model_execution_feedback
for _ in range(10): # 10 times to split the content is enough
user_prompt = (
Environment(undefined=StrictUndefined)
.from_string(
evaluate_prompts["evaluator_final_feedback"]["user"],
)
.render(
model_information=target_task.get_information(),
model_execution_feedback=execution_feedback_to_render,
model_code_feedback=model_code_feedback,
model_value_feedback=model_value_feedback,
)
)
if (
APIBackend().build_messages_and_calculate_token(
user_prompt=user_prompt,
system_prompt=system_prompt,
)
> RD_AGENT_SETTINGS.chat_token_limit
):
execution_feedback_to_render = execution_feedback_to_render[len(execution_feedback_to_render) // 2 :]
else:
break
final_evaluation_dict = json.loads(
APIBackend().build_messages_and_create_chat_completion(
user_prompt=user_prompt,
system_prompt=system_prompt,
json_mode=True,
),
)
if isinstance(final_evaluation_dict["final_decision"], str) and final_evaluation_dict[
"final_decision"
].lower() in ("true", "false"):
final_evaluation_dict["final_decision"] = bool(final_evaluation_dict["final_decision"])
return (
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: Implementation,
gt_implementation: Implementation,
queried_knowledge: QueriedKnowledge = None,
**kwargs,
) -> ModelCoderFeedback:
target_task_information = target_task.get_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)
batch_size, num_features, num_timesteps = (
random.randint(6, 10),
random.randint(6, 10),
random.randint(6, 10),
)
input_value, param_init_value = random.random(), random.random()
assert isinstance(implementation, ModelImplementation)
model_execution_feedback, gen_tensor = 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, ModelImplementation)
_, gt_tensor = 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_tensor = None
shape_feedback, shape_decision = shape_evaluator(gen_tensor, (batch_size, 1))
value_feedback, value_decision = value_evaluator(gt_tensor, gen_tensor)
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 = []
calls = []
for index in range(len(evo.sub_tasks)):
corresponding_implementation = evo.sub_implementations[index]
corresponding_gt_implementation = (
evo.sub_gt_implementations[index] if evo.sub_gt_implementations is not None else None
)
calls.append(
(
ModelCoderEvaluator(scen=self.scen).evaluate,
(
evo.sub_tasks[index],
corresponding_implementation,
corresponding_gt_implementation,
queried_knowledge,
),
),
)
multi_implementation_feedback = multiprocessing_wrapper(calls, n=MODEL_IMPL_SETTINGS.evo_multi_proc_n)
final_decision = [
None if single_feedback is None else single_feedback.final_decision
for single_feedback in multi_implementation_feedback
]
RDAgentLog().info(f"Final decisions: {final_decision} True count: {final_decision.count(True)}")
return multi_implementation_feedback
@@ -0,0 +1,29 @@
from rdagent.components.coder.model_coder.model import (
ModelExperiment,
ModelImplementation,
ModelTask,
)
from rdagent.core.evolving_framework import EvolvableSubjects
from rdagent.core.log import RDAgentLog
class ModelEvolvingItem(ModelExperiment, EvolvableSubjects):
"""
Intermediate item of model implementation.
"""
def __init__(
self,
sub_tasks: list[ModelTask],
sub_gt_implementations: list[ModelImplementation] = 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
RDAgentLog().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
@@ -0,0 +1,145 @@
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 ModelImplementation, 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_utils import APIBackend
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,
) -> ModelImplementation:
model_information_str = target_task.get_information()
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(),
queried_former_failed_knowledge=queried_former_failed_knowledge_to_render,
)
)
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,
)
< RD_AGENT_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=True).build_messages_and_create_chat_completion(
user_prompt=user_prompt,
system_prompt=system_prompt,
json_mode=True,
),
)["code"]
# ast.parse(code)
model_implementation = ModelImplementation(
target_task,
)
model_implementation.prepare()
model_implementation.inject_code(**{"model.py": code})
return model_implementation
def evolve(
self,
*,
evo: ModelEvolvingItem,
queried_knowledge: ModelQueriedKnowledge | None = None,
**kwargs,
) -> ModelEvolvingItem:
new_evo = deepcopy(evo)
# 1.找出需要evolve的model
to_be_finished_task_index = []
for index, target_model_task in enumerate(new_evo.sub_tasks):
target_model_task_desc = target_model_task.get_information()
if target_model_task_desc in queried_knowledge.success_task_to_knowledge_dict:
new_evo.sub_implementations[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, (new_evo.sub_tasks[target_index], queried_knowledge))
for target_index in to_be_finished_task_index
],
n=MODEL_IMPL_SETTINGS.evo_multi_proc_n,
)
for index, target_index in enumerate(to_be_finished_task_index):
new_evo.sub_implementations[target_index] = result[index]
# for target_index in to_be_finished_task_index:
# new_evo.sub_implementations[target_index] = self.implement_one_model(
# new_evo.sub_tasks[target_index], queried_knowledge
# )
new_evo.corresponding_selection = to_be_finished_task_index
return new_evo
@@ -0,0 +1,167 @@
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,
EvoStep,
Knowledge,
KnowledgeBase,
QueriedKnowledge,
RAGStrategy,
)
from rdagent.core.experiment import Implementation
from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list
class ModelKnowledge(Knowledge):
def __init__(
self,
target_task: ModelTask,
implementation: Implementation,
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
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(KnowledgeBase):
def __init__(self) -> None:
self.implementation_trace: dict[str, ModelKnowledge] = dict()
self.success_task_info_set: set[str] = set()
self.task_to_embedding = dict()
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_information()
implementation = implementations.sub_implementations[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_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
@@ -22,7 +22,7 @@ class ModelImpValEval:
- If the model structure is similar, the output will change in similar way when we change the input.
Challenge:
- The key difference between it and implementing factors is that we have parameters in the layers (Factor operators often have no parameters or are given parameters).
- The key difference between it and implementing models is that we have parameters in the layers (Model operators often have no parameters or are given parameters).
- we try to initialize the model param in similar value. So only the model structure is different.
Comparing the correlation of following sequences
@@ -41,9 +41,8 @@ class ModelImpValEval:
for _ in range(round_n):
# run different model initial parameters.
for init_val in [-0.2, -0.1, 0.1, 0.2]:
data, exec_config = get_data_conf(init_val)
gt_res = gt.execute(data=data, config=exec_config)
res = gen.execute(data=data, config=exec_config)
_, gt_res = gt.execute(input_value=init_val, param_init_value=init_val)
_, res = gen.execute(input_value=init_val, param_init_value=init_val)
eval_pairs.append((res, gt_res))
# flat and concat the output
+21 -2
View File
@@ -1,13 +1,32 @@
from pathlib import Path
from typing import Union
from pydantic_settings import BaseSettings
class ModelImplSettings(BaseSettings):
workspace_path: Path = Path("./git_ignore_folder/model_imp_workspace/") # Added type annotation for work_space
class Config:
env_prefix = "MODEL_IMPL_" # Use MODEL_IMPL_ as prefix for environment variables
file_based_execution_workspace: str = str(
(Path().cwd() / "git_ignore_folder" / "model_implementation_workspace").absolute(),
)
implementation_execution_cache_location: str = str(
(Path().cwd() / "git_ignore_folder" / "model_implementation_execution_cache").absolute(),
)
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
evo_multi_proc_n: int = 1
enable_execution_cache: bool = True # whether to enable the execution cache
MODEL_IMPL_SETTINGS = ModelImplSettings()
@@ -1,59 +0,0 @@
import numpy as np
import torch
def shape_evaluator(target, prediction):
if target is None or prediction is None:
return None, 0
tar_shape = target.shape
pre_shape = prediction.shape
diff = []
for i in range(max(len(tar_shape), len(pre_shape))):
dim_tar = tar_shape[i] if i < len(tar_shape) else 0
dim_pre = pre_shape[i] if i < len(pre_shape) else 0
diff.append(abs(dim_tar - dim_pre))
metric = 1 / (np.exp(np.mean(diff)) + 1)
return diff, metric
def reshape_tensor(original_tensor, target_shape):
new_tensor = torch.zeros(target_shape)
for i, dim in enumerate(original_tensor.shape):
new_tensor = new_tensor.narrow(i, 0, dim).copy_(original_tensor)
return new_tensor
def value_evaluator(target, prediction):
if target is None or prediction is None:
return None, 0
tar_shape = target.shape
pre_shape = prediction.shape
# Determine the shape of the padded tensors
dims = [
max(s1, s2)
for s1, s2 in zip(
tar_shape + (1,) * (len(pre_shape) - len(tar_shape)),
pre_shape + (1,) * (len(tar_shape) - len(pre_shape)),
)
]
# Reshape both tensors to the determined shape
target = target.reshape(*tar_shape, *(1,) * (max(len(tar_shape), len(pre_shape)) - len(tar_shape)))
prediction = prediction.reshape(*pre_shape, *(1,) * (max(len(tar_shape), len(pre_shape)) - len(pre_shape)))
target_padded = reshape_tensor(target, dims)
prediction_padded = reshape_tensor(prediction, dims)
# Calculate the mean absolute difference
diff = torch.abs(target_padded - prediction_padded)
metric = 1 / (1 + np.exp(torch.mean(diff).item()))
return diff, metric
if __name__ == "__main__":
tar = torch.rand(4, 5, 5)
pre = torch.rand(4, 1)
print(shape_evaluator(tar, pre))
print(value_evaluator(tar, pre)[1])
+51 -137
View File
@@ -1,14 +1,16 @@
import json
import pickle
import site
import uuid
from pathlib import Path
from typing import Dict, Optional, Sequence
from typing import Dict, Optional
import torch
from rdagent.components.coder.model_coder.conf import MODEL_IMPL_SETTINGS
from rdagent.components.loader.task_loader import ModelTaskLoader
from rdagent.core.exception import CodeFormatException
from rdagent.core.experiment import Experiment, FBImplementation, ImpLoader, Task
from rdagent.core.experiment import Experiment, FBImplementation, Task
from rdagent.oai.llm_utils import md5_hash
from rdagent.utils import get_module_by_module_path
@@ -20,29 +22,20 @@ class ModelTask(Task):
variables: Dict[str, str] # map the variable name to the variable description
def __init__(
self, name: str, description: str, formulation: str, variables: Dict[str, str], key: Optional[str] = None
self, name: str, description: str, formulation: str, variables: Dict[str, str], model_type: Optional[str] = None
) -> None:
"""
Parameters
----------
key : Optional[str]
Key is a string to identify the task.
It will be used to connect to other information(e.g. ground truth).
"""
self.name = name
self.description = description
self.formulation = formulation
self.variables = variables
self.key = key
self.name: str = name
self.description: str = description
self.formulation: str = formulation
self.variables: str = variables
self.model_type: str = model_type # Tabular for tabular model, TimesSeries for time series model
def get_information(self):
return f"""name: {self.name}
description: {self.description}
formulation: {self.formulation}
variables: {self.variables}
key: {self.key}
model_type: {self.model_type}
"""
@staticmethod
@@ -75,136 +68,57 @@ class ModelImplementation(FBImplementation):
def __init__(self, target_task: Task) -> None:
super().__init__(target_task)
self.path = None
def prepare(self) -> None:
"""
Prepare for the workspace;
"""
unique_id = uuid.uuid4()
self.path = MODEL_IMPL_SETTINGS.workspace_path / f"M{unique_id}"
self.workspace_path = Path(MODEL_IMPL_SETTINGS.file_based_execution_workspace) / f"M{unique_id}"
# start with `M` so that it can be imported via python
self.path.mkdir(parents=True, exist_ok=True)
self.workspace_path.mkdir(parents=True, exist_ok=True)
def execute(self, data=None, config: dict = {}):
mod = get_module_by_module_path(str(self.path / "model.py"))
def execute(
self,
batch_size: int = 8,
num_features: int = 10,
num_timesteps: int = 4,
input_value: float = 1.0,
param_init_value: float = 1.0,
):
try:
if MODEL_IMPL_SETTINGS.enable_execution_cache:
# NOTE: cache the result for the same code
target_file_name = md5_hash(self.code_dict["model.py"])
cache_file_path = (
Path(MODEL_IMPL_SETTINGS.implementation_execution_cache_location) / f"{target_file_name}.pkl"
)
Path(MODEL_IMPL_SETTINGS.implementation_execution_cache_location).mkdir(exist_ok=True, parents=True)
if cache_file_path.exists():
return pickle.load(open(cache_file_path, "rb"))
mod = get_module_by_module_path(str(self.workspace_path / "model.py"))
model_cls = mod.model_cls
except AttributeError:
raise CodeFormatException("The model_cls is not implemented in the model.py")
# model_init =
assert isinstance(data, tuple)
node_feature, _ = data
in_channels = node_feature.size(-1)
m = model_cls(in_channels)
if self.target_task.model_type == "Tabular":
input_shape = (batch_size, num_features)
m = model_cls(num_features=input_shape[1])
elif self.target_task.model_type == "TimeSeries":
input_shape = (batch_size, num_features, num_timesteps)
m = model_cls(num_features=input_shape[1], num_timesteps=input_shape[2])
data = torch.full(input_shape, input_value)
# TODO: initialize all the parameters of `m` to `model_eval_param_init`
model_eval_param_init: float = config["model_eval_param_init"]
# initialize all parameters of `m` to `param_init_value`
for _, param in m.named_parameters():
param.data.fill_(param_init_value)
out = m(data)
execution_model_output = out.cpu().detach()
execution_feedback_str = f"Execution successful, output tensor shape: {execution_model_output.shape}"
if MODEL_IMPL_SETTINGS.enable_execution_cache:
pickle.dump((execution_feedback_str, execution_model_output), open(cache_file_path, "wb"))
return execution_feedback_str, execution_model_output
# initialize all parameters of `m` to `model_eval_param_init`
for _, param in m.named_parameters():
param.data.fill_(model_eval_param_init)
assert isinstance(data, tuple)
return m(*data)
def execute_desc(self) -> str:
return """
The the implemented code will be placed in a file like <uuid>/model.py
We'll import the model in the implementation in file `model.py` after setting the cwd into the directory
- from model import model_cls (So you must have a variable named `model_cls` in the file)
- So your implemented code could follow the following pattern
```Python
class XXXLayer(torch.nn.Module):
...
model_cls = XXXLayer
```
- initialize the model by initializing it `model_cls(input_dim=INPUT_DIM)`
- And then verify the model by comparing the output tensors by feeding specific input tensor.
"""
except Exception as e:
return f"Execution error: {e}", None
class ModelExperiment(Experiment[ModelTask, ModelImplementation]):
...
class ModelTaskLoaderJson(ModelTaskLoader):
# def __init__(self, json_uri: str, select_model: Optional[str] = None) -> None:
# super().__init__()
# self.json_uri = json_uri
# self.select_model = 'A-DGN'
# def load(self, *argT, **kwargs) -> Sequence[ModelImplTask]:
# # json is supposed to be in the format of {model_name: dict{model_data}}
# model_dict = json.load(open(self.json_uri, "r"))
# if self.select_model is not None:
# assert self.select_model in model_dict
# model_name = self.select_model
# model_data = model_dict[self.select_model]
# else:
# model_name, model_data = list(model_dict.items())[0]
# model_impl_task = ModelImplTask(
# name=model_name,
# description=model_data["description"],
# formulation=model_data["formulation"],
# variables=model_data["variables"],
# key=model_name
# )
# return [model_impl_task]
def __init__(self, json_uri: str) -> None:
super().__init__()
self.json_uri = json_uri
def load(self, *argT, **kwargs) -> Sequence[ModelTask]:
# json is supposed to be in the format of {model_name: dict{model_data}}
model_dict = json.load(open(self.json_uri, "r"))
# FIXME: the model in the json file is not right due to extraction error
# We should fix them case by case in the future
#
# formula_info = {
# "name": "Anti-Symmetric Deep Graph Network (A-DGN)",
# "description": "A framework for stable and non-dissipative DGN design. It ensures long-range information preservation between nodes and prevents gradient vanishing or explosion during training.",
# "formulation": r"\mathbf{x}^{\prime}_i = \mathbf{x}_i + \epsilon \cdot \sigma \left( (\mathbf{W}-\mathbf{W}^T-\gamma \mathbf{I}) \mathbf{x}_i + \Phi(\mathbf{X}, \mathcal{N}_i) + \mathbf{b}\right),",
# "variables": {
# r"\mathbf{x}_i": "The state of node i at previous layer",
# r"\epsilon": "The step size in the Euler discretization",
# r"\sigma": "A monotonically non-decreasing activation function",
# r"\Phi": "A graph convolutional operator",
# r"W": "An anti-symmetric weight matrix",
# r"\mathbf{x}^{\prime}_i": "The node feature matrix at layer l-1",
# r"\mathcal{N}_i": "The set of neighbors of node u",
# r"\mathbf{b}": "A bias vector",
# },
# "key": "A-DGN",
# }
model_impl_task_list = []
for model_name, model_data in model_dict.items():
model_impl_task = ModelTask(
name=model_name,
description=model_data["description"],
formulation=model_data["formulation"],
variables=model_data["variables"],
key=model_data["key"],
)
model_impl_task_list.append(model_impl_task)
return model_impl_task_list
class ModelImpLoader(ImpLoader[ModelTask, ModelImplementation]):
def __init__(self, path: Path) -> None:
self.path = Path(path)
def load(self, task: ModelTask) -> ModelImplementation:
assert task.key is not None
mti = ModelImplementation(task)
mti.prepare()
with open(self.path / f"{task.key}.py", "r") as f:
code = f.read()
mti.inject_code(**{"model.py": code})
return mti
class ModelExperiment(Experiment[ModelTask, ModelImplementation]): ...
@@ -30,7 +30,6 @@ class ModelCodeWriter(TaskGenerator[ModelExperiment]):
description=t.description,
formulation=t.formulation,
variables=t.variables,
execute_desc=mti.execute_desc(),
)
system_prompt = sys_prompt_tpl.render()
@@ -1,8 +1,8 @@
code_implement_sys: -|
You are an assistant whose job is to answer user's question."
code_implement_user: -|
code_implement_sys: |-
You are an assistant whose job is to answer user's question.
code_implement_user: |-
With the following given information, write a python code using pytorch and torch_geometric to implement the model.
This model is in the graph learning field, only have one layer.
The input will be node_feature [num_nodes, dim_feature] and edge_index [2, num_edges] (It would be the input of the forward model)
@@ -13,6 +13,15 @@ code_implement_user: -|
3. model formulation:{{formulation}}
4. model variables:{{variables}}.
You must complete the forward function as far as you can do.
\# Execution
Your implemented code will be exectued in the follow way
{{execute_desc}}
Execution Your implemented code will be executed in the follow way:
The the implemented code will be placed in a file like [uuid]/model.py
We'll import the model in the implementation in file `model.py` after setting the cwd into the directory
- from model import model_cls (So you must have a variable named `model_cls` in the file)
- So your implemented code could follow the following pattern
```Python
class XXXLayer(torch.nn.Module):
...
model_cls = XXXLayer
```
- initialize the model by initializing it `model_cls(input_dim=INPUT_DIM)`
- And then verify the model by comparing the output tensors by feeding specific input tensor.
@@ -1,8 +1,123 @@
extract_model_formulation_system: |-
offer description of the proposed model in this paper, write a latex formula with variable of the model. the format should be like " "Model Name": {
"description": "",
"formulation": "",
"variables": {
"\\hat{y}_u": "The predicted output for node u",
}"
such format content should be begin with ```json and end with ``` and the content should be in json format.
offer description of the proposed model in this paper, write a latex formula with variable of the model. the format should be like
"Model Name": {
"description": "",
"formulation": "",
"variables": {
"\\hat{y}_u": "The predicted output for node u",
}"
such format content should be begin with ```json and end with ``` and the content should be in json format.
evolving_strategy_model_coder:
system: |-
User is trying to implement some pytorch models in the following scenario:
{{ scenario }}
Your code is expected to align the scenario in any form which means The user needs to get the prediction of the model based on the input data.
To help you write the correct code, the user might provide multiple information that helps you write the correct code:
1. The user might provide you the correct code to similar models. Your should learn from these code to write the correct code.
2. The user might provide you the failed former code and the corresponding feedback to the code. The feedback contains to the execution, the code and the model output value. You should analyze the feedback and try to correct the latest code.
3. The user might provide you the suggestion to the latest fail code and some similar fail to correct pairs. Each pair contains the fail code with similar error and the corresponding corrected version code. You should learn from these suggestion to write the correct code.
Your must write your code based on your former latest attempt below which consists of your former code and code feedback, you should read the former attempt carefully and must not modify the right part of your former code.
{% if queried_former_failed_knowledge|length != 0 %}
--------------Your former latest attempt:---------------
=====Code to the former implementation=====
{{ queried_former_failed_knowledge[-1].implementation.code }}
=====Feedback to the former implementation=====
{{ queried_former_failed_knowledge[-1].feedback }}
{% endif %}
Please response the code in the following json format. Here is an example structure for the JSON output:
{
"code": "The Python code as a string."
}
user: |-
--------------Target model information:---------------
{{ model_information_str }}
{% if queried_similar_successful_knowledge|length != 0 %}
--------------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() }}
=====Code:=====
{{ similar_successful_knowledge.implementation.code }}
{% endfor %}
{% endif %}
{% if queried_former_failed_knowledge|length != 0 %}
--------------Former failed code:---------------
{% for former_failed_knowledge in queried_former_failed_knowledge %}
=====Code to implementation {{ loop.index }}=====
{{ former_failed_knowledge.implementation.code }}
=====Feedback to implementation {{ loop.index }}=====
{{ former_failed_knowledge.feedback }}
{% endfor %}
{% endif %}
evaluator_code_feedback:
system: |-
User is trying to implement some models in the following scenario:
{{ scenario }}
User will provide you the information of the model.
Your job is to check whether user's code is align with the model information and the scenario.
The user will provide the source python code and the execution error message if execution failed.
The user might provide you the ground truth code for you to provide the critic. You should not leak the ground truth code to the user in any form but you can use it to provide the critic.
User has also compared the output generated by the user's code and the ground truth code. The user will provide you some analyze result comparing two output. You may find some error in the code which caused the difference between the two output.
If the ground truth code is provided, your critic should only consider checking whether the user's code is align with the ground truth code since the ground truth is definitely correct.
If the ground truth code is not provided, your critic should consider checking whether the user's code is reasonable and correct to the description and to the scenario.
Notice that your critics are not for user to debug the code. They are sent to the coding agent to correct the code. So don't give any following items for the user to check like "Please check the code line XXX".
You should provide the suggestion to each of your critic to help the user improve the code. Please response the critic in the following format. Here is an example structure for the output:
critic 1: The critic message to critic 1
critic 2: The critic message to critic 2
user: |-
--------------Model information:---------------
{{ model_information }}
--------------Python code:---------------
{{ code }}
--------------Execution feedback:---------------
{{ model_execution_feedback }}
{% if model_value_feedback is not none %}
--------------Model value feedback:---------------
{{ model_value_feedback }}
{% endif %}
{% if gt_code is not none %}
--------------Ground truth Python code:---------------
{{ gt_code }}
{% endif %}
evaluator_final_feedback:
system: |-
User is trying to implement a model in the following scenario:
{{ scenario }}
User has finished evaluation and got some feedback from the evaluator.
The evaluator run the code and get the output and provide several feedback regarding user's code and code output. You should analyze the feedback and considering the scenario and model description to give a final decision about the evaluation result. The final decision concludes whether the model is implemented correctly and if not, detail feedback containing reason and suggestion if the final decision is False.
The implementation final decision is considered in the following logic:
1. If the value and the ground truth value are exactly the same under a small tolerance, the implementation is considered correct.
2. If no ground truth value is not provided, the implementation is considered correct if the code execution is successful and the code feedback is align with the scenario and model description.
Please response the critic in the json format. Here is an example structure for the JSON output, please strictly follow the format:
{
"final_decision": True,
"final_feedback": "The final feedback message",
}
user: |-
--------------Model information:---------------
{{ model_information }}
--------------Model Execution feedback:---------------
{{ model_execution_feedback }}
--------------Model Code feedback:---------------
{{ model_code_feedback }}
--------------Model value feedback:---------------
{{ model_value_feedback }}
@@ -28,7 +28,7 @@ def extract_model_from_doc(doc_content: str) -> dict:
Returns
-------
dict
{factor_name: dict{description, formulation, variables}}
{model_name: dict{description, formulation, variables}}
"""
session = APIBackend().build_chat_session(
session_system_prompt=document_process_prompts["extract_model_formulation_system"],