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. """