Files
NexQuant/rdagent/core/evolving_framework.py
T
Xu Yang a126c84c92 Refine all the implementation code to higher quality for release (#29)
* refine CI script

* refine all the code to higher quality

* refine the script to factor extraction and implementation

* add task loader interface

* add a task loader interface && move pdf analysis to pdf task loader

* change the name to global variables

---------

Co-authored-by: xuyang1 <xuyang1@microsoft.com>
2024-06-18 11:50:03 +08:00

102 lines
2.3 KiB
Python

from __future__ import annotations
import copy
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any
class Feedback:
pass
class Knowledge:
pass
class QueriedKnowledge:
pass
class KnowledgeBase(ABC):
@abstractmethod
def query(
self,
) -> QueriedKnowledge | None:
raise NotImplementedError
class EvolvableSubjects:
"""The target object to be evolved"""
def clone(self) -> EvolvableSubjects:
return copy.deepcopy(self)
class QlibEvolvableSubjects(EvolvableSubjects): ...
@dataclass
class EvoStep:
"""At a specific step,
based on
- previous trace
- newly RAG kownledge `QueriedKnowledge`
the EvolvableSubjects is evolved to a new one `EvolvableSubjects`.
(optional) After evaluation, we get feedback `feedback`.
"""
evolvable_subjects: EvolvableSubjects
queried_knowledge: QueriedKnowledge | None = None
feedback: Feedback | None = None
class EvolvingStrategy(ABC):
@abstractmethod
def evolve(
self,
*evo: EvolvableSubjects,
evolving_trace: list[EvoStep] | None = None,
queried_knowledge: QueriedKnowledge | None = None,
**kwargs: Any,
) -> EvolvableSubjects:
"""The evolving trace is a list of (evolvable_subjects, feedback) ordered
according to the time.
The reason why the parameter is important for the evolving.
- evolving_trace: the historical feedback is important.
- queried_knowledge: queried knowledge
"""
class RAGStrategy(ABC):
"""Retrival Augmentation Generation Strategy"""
def __init__(self, knowledgebase: KnowledgeBase) -> None:
self.knowledgebase = knowledgebase
@abstractmethod
def query(
self,
evo: EvolvableSubjects,
evolving_trace: list[EvoStep],
**kwargs: Any,
) -> QueriedKnowledge | None:
pass
@abstractmethod
def generate_knowledge(
self,
evolving_trace: list[EvoStep],
*,
return_knowledge: bool = False,
**kwargs: Any,
) -> Knowledge | None:
"""Generating new knowledge based on the evolving trace.
- It is encouraged to query related knowledge before generating new knowledge.
RAGStrategy should maintain the new knowledge all by itself.
"""