objected knowledge base implement for email code structure

src/aios_kernel/knowledge/__init__.py

@@ -0,0 +1,2 @@
+from .object import EmailObject, ImageObject, TextChunkObject, ObjectID
+from .knowledge_base import KnowledgeBase

src/aios_kernel/knowledge/knowledge_base.py

@@ -0,0 +1,10 @@
+# define a knowledge base class
+from src.aios_kernel.agent import AgentPrompt
+from .object import KnowledgeObject
+
+class KnowledgeBase:
+    async def insert(self, object: KnowledgeObject):
+        pass
+
+    async def query(self, prompt: AgentPrompt) -> AgentPrompt:
+        pass

src/aios_kernel/knowledge/object.py

@@ -0,0 +1,65 @@
+
+# define a object type enum
+from abc import ABC
+
+
+class ObjectType(Enum):
+    TextChunk = 1
+    Image = 2
+    Email = 101
+
+
+# define a object ID class to identify a object
+class ObjectID: # pylint: disable=too-few-public-methods
+    def __init__(self, object_type, digist):
+        self.object_type = object_type
+        self.digist = digist
+
+    def __str__(self):
+        return f"{self.object_type.name}:{self.digist}"
+    
+
+# define a object class
+class KnowledgeObject(ABC): # pylint: disable=too-few-public-methods
+    def __init__(self, object_type):
+        self.object_type = object_type
+
+    @abstractmethod
+    def get_id(self) -> ObjectID:
+        pass
+
+    # define a to binary method to convert object to binary
+    @abstractmethod
+    def to_binary(self) -> bytes:
+        pass
+
+    # define a from binary method to convert binary to object
+    @abstractmethod
+    def from_binary(self, binary: bytes):
+        pass
+
+
+# define a text chunk class
+class TextChunkObject(KnowledgeObject): # pylint: disable=too-few-public-methods
+    def __init__(self, text):
+        super().__init__(ObjectType.TextChunk)
+        self.text = text
+
+
+# define a image class
+class ImageObject(KnowledgeObject): # pylint: disable=too-few-public-methods
+    def __init__(self, meta, path):
+        super().__init__(ObjectType.Image)
+        self.meta = meta
+        self.path = path
+
+
+# define a email class
+class EmailObject(KnowledgeObject): # pylint: disable=too-few-public-methods
+    def __init__(self, meta):
+        super().__init__(ObjectType.Email)
+        self.meta = meta
+        self.text = []
+        self.images = []
+
+    

src/aios_kernel/knowledge/object_storage.py

@@ -0,0 +1,33 @@
+# import RDB LargeBinary
+from sqlalchemy import Column, String, LargeBinary, create_engine, sessionmaker, pickle
+from .object import KnowledgeObject
+
+# implement object storage with RDB
+# define object storage table
+class ObjectStorageTable(Base):
+    __tablename__ = 'object_storage'
+    id = Column(String, primary_key=True)
+    parent = Column(String, nullable=True)
+    object = Column(LargeBinary, nullable=False)
+
+    def __init__(self, id, parent, object): # pylint: disable=redefined-builtin
+        self.id = id
+        self.parent = parent
+        self.object = object
+
+# define object storage class
+class ObjectStorage:
+    async def __init__(self, db_url):
+        self.engine = create_engine(db_url)
+        self.session = sessionmaker(bind=self.engine)() # pylint: disable=not-callable
+
+    async def get(self, id) -> [KnowledgeObject, KnowledgeObject]:
+        obj = self.session.query(ObjectStorageTable).filter(ObjectStorageTable.id == id).first()
+        if obj is None:
+            return None
+        return pickle.loads(obj.object)
+    
+    # define insert method
+    async def insert(self, object, parent): # pylint: disable=redefined-builtin
+        obj = ObjectStorageTable(id, parent, pickle.dumps(object))
+        

src/aios_kernel/knowledge/vector_base.py

@@ -0,0 +1,14 @@
+# import the ObjectID class
+from .object import ObjectID
+
+# define a vector base class
+class VectorBase:
+    def __init__(self, db_url, model_name) -> None:
+        self.db_url = db_url
+        self.model_name = model_name
+
+    async def insert(self, vector: [float], id: ObjectID):
+        pass
+
+    async def query(self, vector: [float], top_k: int) -> [ObjectID]:
+        pass