gavindiaz/NexQuant
1
0
Fork 0
mirror of https://github.com/NicolasBohn/NexQuant.git synced 2026-08-03 18:27:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
* 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 63f7bf23da
commit 22e1aa3330
24 changed files with 1243 additions and 291 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rdagent/components/coder/model_coder/CoSTEER/knowledge_management.py
+167
View File
@@ -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