Merge pull request #86 from photosssa/mvp-dev

Issue 85: Knowledge pipeline manager

doc/knowledge_pipeline.md

@@ -0,0 +1,87 @@
+# 配置knowledge pipeline
+knowledge pipeline 扫描指定的输入，把输入的内容构建为结构化的knowledge object，之后依照使用knowledge的应用场景，为object创建各种各样的索引。
+## Input
+输入定义从个人数据来源转换成结构化的knowledge object的过程，并且定义在object上调用parser的粒度。比如典型的几种Input的实现：
++ 本地目录：指定本地目录，扫描本地目录的所有文件，并且监听他的更新；对每个一个文件生成object并且写入object store；对每一个新产生的object调用parser；
++ 个人邮箱：扫描个人邮箱收件箱，并且监听新的邮件；对每一封邮件生成email object并且写入object store；对每一个新产生的email object调用parser；
++ 浏览器上下文：实现浏览器插件，对当前浏览的页面元素通过rpc传入对应的input 后端实现，生成rich text object；对每一个新产生的rich text object调用parser；
+
+## Parser
+Parser定义从input 输入的object 创建索引的过程；包括但不限于以下主要手段，以及他们的组合：
++ 向量化之后写入vector store
++ 创建各种维度的RDB，NoSQL索引
++ 向Agent send object
+
+配置Pipeline 应当包含以下几个部分：
++ Input method：包含实现input的 python module
++ Input params：inpu module的参数，比如本地路径，邮箱地址
++ Parser method：包含实现parser的 python module；如果Parser是指向Agent，这个配置是可以简化成Agent instance name；
+
+# Knowledge pipeline manager
+pipeline 管理会类似agent manager，manager管理pipeline config，从config 创建instance在后台持续运行， knowledge pipeline manager 也需要处理pipeline instance的状态管理.
+
+集成到aios shell中，加入如下命令：
++ knowledge pipelines： 返回当前运行中的pipeline实例
++ knowledge journal $pipeline [$topn]： 查询当前pipeline运行的journal日志 
++ knowledge query $object_id: 查询指定knowledge object的内容
+
+# 在aios shell中添加新的knowledge pipeline
+在$home/myai/knowledge_pipelines/, 或者开发模式下在 $source_root/rootfs/knowledge_pipelines/ 目录中，添加新的pipeline 目录, 以下以内建的pipeline Mia为例说明：
+
+## pipeline.toml
+创建pipeline.toml配置文件
++ name字段指定全局唯一的pipeline name
++ input.module字段指向相对pipeline目录的input实现
++ input.params字段定义input的输入参数，不同的input实现可以有不同的参数格式
++ parser 部分也是类似
+``` toml
+name = "Mia"
+input.module = "input.py"
+input.params.path = "${myai_dir}/data"
+parser.module = "parser.py"
+parser.params.path = "${myai_dir}/knowledge/indices/embedding"
+```
+
+## input
+input模块至少应当实现：
+```python
+async def next(self):
+```
+定义input class，实现异步迭代生成器方法next，扫描输入，对其中的每一个元素生成结构化的knowledge object；
++ 如果input中的所有元素都扫描完成了，返回None, pipeline会被标记为finish
++ 如果input可pending，等待新的输入，返回（None, None）
++ 如果要把创建的object传递到parser，返回（object_id, journal_str）,其中journal_str是产生的journal 日志中的input 部分；
+Mia中的实现就是扫描目录中的文件，对文本和图片创建object；
+```python
+def init(env: KnowledgePipelineEnvironment, params: dict)
+```
+创建input class的实例并返回
+
+## parser
+parser模块至少应当实现
+```python
+async def parse(self, object: ObjectID) -> str:
+```
+定义parser class，实现parse成员方法,对input中返回的object_id创建索引，返回journal_str.
+Mia中的实现就是对输入的object内容embedding，并且保存到chromadb中；
+```python
+def init(env: KnowledgePipelineEnvironment, params: dict)
+```
+创建parser class的实例并返回
+
+# 使用pipeline创建的索引
+pipeline定义了创建knowledge object 和索引的过程，对应的要使用pipeline创建的索引完成工作。
+还是以内建的Mia为例，不止创建名为Mia的pipeline，还在Agent中加入了查询Mia pipeline创建出来的chromadb的 Agent Mia；
+## query.py
+query 模块并不是pipeline的一部分，其逻辑是跟parser是一致的，在query中定义了一个agent可访问的query function，输入prompt，返回chromadb中embedding相近的object id；
+
+## agent.toml
+```toml
+owner_env = "../../knowledge_pipelines/Mia/query.py"
+```
+在Mia的agent template配置里，引用query模块创建的query function；并且编辑好让Mia推理调用query方法的提示词。
+
+
+
+
+

rootfs/agents/FindPhoto/agent.toml

@@ -0,0 +1,21 @@
+instance_id = "FindPhoto"
+fullname = "FindPhoto"
+llm_model_name = "gpt-4"
+max_token_size = 16000
+enable_timestamp = "false"
+owner_prompt = "我是你的主人{name}"
+contact_prompt = "我是你的朋友{name}"
+owner_env = "environment.py"
+
+[[prompt]]
+role = "system"
+content = """
+你是FindPhoto,你可以访问我的照片目录。
+
+***
+你在收到我的信息后，按如下规则处理
+1. 在第一次接受到一条信息时，优先尝试用合适的关键字查询去查询知识库。
+2. 如果信息中包含一段知识库的查询结果，尝试用查询结果处理，如果还是不能处理，尝试递增index继续查询。
+3. 如果要返回知识库结果条目，在消息开头附上他的json字符串。
+"""
+

rootfs/agents/Mia/agent.toml

@@ -1,13 +1,9 @@
 instance_id = "Mia"
 fullname = "Mia"
 #llm_model_name = "gpt-4"
-#max_token_size = 16000
-#enable_function =["add_event"]
-#enable_kb = "true"
-#enable_timestamp = "false"
 owner_prompt = "我是你的主人{name}"
 contact_prompt = "我是你的朋友{name}"
-owner_env = "knowledge"
+owner_env = "../../knowledge_pipelines/Mia/query.py"
 
 [[prompt]]
 role = "system"

rootfs/email/config.toml

@@ -1,7 +0,0 @@
-
-
-EMAIL_IMAP_SERVER = "imap.gmail.com"
-EMAIL_ADDRESS = '<>'
-EMAIL_PASSWORD = '<>'
-EMAIL_IMAP_PORT = 993
-LOCAL_DIR = 'rootfs/data'

rootfs/knowledge_pipelines/Mia/input.py

@@ -0,0 +1,68 @@
+import os
+import aiofiles
+import chardet
+import logging
+import string
+from knowledge import ImageObjectBuilder, DocumentObjectBuilder, KnowledgePipelineEnvironment, KnowledgePipelineJournal
+from aios_kernel.storage import AIStorage
+
+class KnowledgeDirSource:
+    def __init__(self, env: KnowledgePipelineEnvironment, config):
+        self.env = env
+        path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir())
+        config["path"] = path
+        self.config = config  
+
+    # @classmethod
+    # def user_config_items(cls):
+    #     return [("path", "local dir path")]
+
+    def path(self):
+        return self.config["path"]
+    
+    @staticmethod
+    async def read_txt_file(file_path:str)->str:
+        cur_encode = "utf-8"
+        async with aiofiles.open(file_path,'rb') as f:
+            cur_encode = chardet.detect(await f.read())['encoding']
+        
+        async with aiofiles.open(file_path,'r',encoding=cur_encode) as f:
+            return await f.read()
+        
+    async def next(self):
+        while True:
+            journals = self.env.journal.latest_journals(1)
+            from_time = 0
+            if len(journals) == 1:
+                latest_journal = journals[0]
+                if latest_journal.is_finish():
+                    yield None
+                    continue
+                from_time = os.path.getctime(latest_journal.get_input())
+                if os.path.getmtime(self.path()) <= from_time:
+                    yield (None, None)
+                    continue
+            
+            file_pathes = sorted(os.listdir(self.path()), key=lambda x: os.path.getctime(os.path.join(self.path(), x)))
+            for rel_path in file_pathes:
+                file_path = os.path.join(self.path(), rel_path)
+                timestamp = os.path.getctime(file_path)
+                if timestamp <= from_time:
+                    continue
+                ext = os.path.splitext(file_path)[1].lower()
+                if ext in ['.jpg', '.jpeg', '.png', '.gif', '.bmp']:
+                    logging.info(f"knowledge dir source found image file {file_path}")
+                    image = ImageObjectBuilder({}, {}, file_path).build(self.env.get_knowledge_store())
+                    await self.env.get_knowledge_store().insert_object(image)
+                    yield (image.calculate_id(), file_path)
+                if ext in ['.txt']:
+                    logging.info(f"knowledge dir source found text file {file_path}")
+                    text = await self.read_txt_file(file_path)
+                    document = DocumentObjectBuilder({}, {}, text).build(self.env.get_knowledge_store())
+                    await self.env.get_knowledge_store().insert_object(document)
+                    yield (document.calculate_id(), file_path)
+            yield (None, None)
+            
+
+def init(env: KnowledgePipelineEnvironment, params: dict) -> KnowledgeDirSource:
+    return KnowledgeDirSource(env, params)

rootfs/knowledge_pipelines/Mia/parser.py

@@ -0,0 +1,102 @@
+# define a knowledge base class
+import json
+import string
+from aios_kernel import ComputeKernel, AIStorage
+from knowledge import *
+
+
+class EmbeddingParser:
+    def __init__(self, env: KnowledgePipelineEnvironment, config: dict):
+        self._default_text_model = "all-MiniLM-L6-v2"
+        self._default_image_model = "clip-ViT-B-32"
+
+        path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir())
+        if not os.path.exists(path):
+            os.makedirs(path)
+        config["path"] = path
+        
+        self.env = env
+        self.config = config
+
+    def get_path(self) -> str:
+        return self.config["path"]
+
+    def __get_vector_store(self, model_name: str) -> ChromaVectorStore:
+        return ChromaVectorStore(self.get_path(), model_name)
+
+    async def __embedding_document(self, document: DocumentObject):
+        for chunk_id in document.get_chunk_list():
+            chunk = self.env.get_knowledge_store().get_chunk_reader().get_chunk(chunk_id)
+            if chunk is None:
+                raise ValueError(f"text chunk not found: {chunk_id}")
+        
+            text = chunk.read().decode("utf-8")
+            vector = await ComputeKernel.get_instance().do_text_embedding(text, self._default_text_model)
+            if vector:
+                await self.__get_vector_store(self._default_text_model).insert(vector, chunk_id)
+
+    async def __embedding_image(self, image: ImageObject):
+        # desc = {}
+        # if not not image.get_meta():
+        #     desc["meta"] = image.get_meta()
+        # if not not image.get_exif():
+        #     desc["exif"] = image.get_exif()
+        # if not not image.get_tags():
+        #     desc["tags"] = image.get_tags()
+        # vector = await self.compute_kernel.do_text_embedding(json.dumps(desc), self._default_text_model)
+        vector = await ComputeKernel.get_instance().do_image_embedding(image.calculate_id(), self._default_image_model)
+        if vector:
+            await self.__get_vector_store(self._default_image_model).insert(vector, image.calculate_id())
+
+    async def __embedding_video(self, vedio: VideoObject):
+        desc = {}
+        if not not vedio.get_meta():
+            desc["meta"] = vedio.get_meta()
+        if not not vedio.get_info():
+            desc["info"] = vedio.get_info()
+        if not not vedio.get_tags():
+            desc["tags"] = vedio.get_tags()
+        vector = await ComputeKernel.get_instance().do_text_embedding(json.dumps(desc), self._default_text_model)
+        await self.__get_vector_store(self._default_text_model).insert(vector, vedio.calculate_id())
+
+    async def __embedding_rich_text(self, rich_text: RichTextObject):
+        for document_id in rich_text.get_documents().values():
+            document = DocumentObject.decode(self.env.get_knowledge_store().get_object_store().get_object(document_id))
+            await self.__embedding_document(document)
+        for image_id in rich_text.get_images().values():
+            image = ImageObject.decode(self.env.get_knowledge_store().get_object_store().get_object(image_id))
+            await self.__embedding_image(image)
+        for video_id in rich_text.get_videos().values():
+            video = VideoObject.decode(self.env.get_knowledge_store().get_object_store().get_object(video_id))
+            await self.__embedding_video(video)
+        for rich_text_id in rich_text.get_rich_texts().values():
+            rich_text = RichTextObject.decode(self.env.get_knowledge_store().get_object_store().get_object(rich_text_id))
+            await self.__embedding_rich_text(rich_text)
+
+    async def __embedding_email(self, email: EmailObject):
+        vector = await ComputeKernel.get_instance().do_text_embedding(json.dumps(email.get_desc()), self._default_text_model)
+        await self.__get_vector_store(self._default_text_model).insert(vector, email.calculate_id())
+        await self.__embedding_rich_text(email.get_rich_text())
+
+
+    async def __do_embedding(self, object: KnowledgeObject):
+        if object.get_object_type() == ObjectType.Document:
+            await self.__embedding_document(object)
+        if object.get_object_type() == ObjectType.Image:
+            await self.__embedding_image(object)
+        if object.get_object_type() == ObjectType.Video:
+            await self.__embedding_video(object)
+        if object.get_object_type() == ObjectType.RichText:
+            await self.__embedding_rich_text(object)
+        if object.get_object_type() == ObjectType.Email:
+            await self.__embedding_email(object)
+        else:
+            pass
+    
+    async def parse(self, object: ObjectID) -> str:
+        obj = self.env.get_knowledge_store().load_object(object)
+        await self.__do_embedding(obj)
+        return "insert into vector store"
+
+def init(env: KnowledgePipelineEnvironment, params: dict) -> EmbeddingParser:
+    return EmbeddingParser(env, params)

rootfs/knowledge_pipelines/Mia/pipeline.toml

@@ -0,0 +1,6 @@
+name = "Mia"
+input.module = "input.py"
+input.params.path = "${myai_dir}/data"
+parser.module = "parser.py"
+parser.params.path = "${myai_dir}/knowledge/indices/embedding"
+

rootfs/knowledge_pipelines/Mia/query.py

@@ -0,0 +1,97 @@
+import os
+import logging
+import json
+from aios_kernel import *
+from knowledge import *
+
+class KnowledgeEnvironment(Environment):
+    def __init__(self, env_id: str) -> None:
+        super().__init__(env_id)
+        self.path = os.path.join(AIStorage.get_instance().get_myai_dir(), "knowledge/indices/embedding")
+        self._default_text_model = "all-MiniLM-L6-v2"
+        self._default_image_model = "clip-ViT-B-32"
+
+        query_param = {
+            "tokens": "key words to query", 
+            "types": "prefered knowledge types, one or more of [text, image]",
+            "index": "index of query result"
+        }
+        self.add_ai_function(SimpleAIFunction("query_knowledge", 
+                                            "vector query content from local knowledge base",
+                                            self._query, 
+                                            query_param))
+        
+    def __get_vector_store(self, model_name: str) -> ChromaVectorStore:
+        return ChromaVectorStore(self.path, model_name)
+    
+    async def query_objects(self, tokens: str, types: list[str], topk: int) -> [ObjectID]:
+        texts = []
+        if "text" in types:
+            vector = await ComputeKernel.get_instance().do_text_embedding(tokens, self._default_text_model)
+            texts = await self.__get_vector_store(self._default_text_model).query(vector, topk)
+        images = []
+        if "image" in types:
+            vector = await ComputeKernel.get_instance().do_text_embedding(tokens, self._default_image_model)
+            images = await self.__get_vector_store(self._default_image_model).query(vector, topk)
+        return texts + images
+    
+        
+    def tokens_from_objects(self, object_ids: [ObjectID]) -> list[str]:
+        results = dict()
+        for object_id in object_ids:
+            parents = KnowledgeStore().get_relation_store().get_related_root_objects(object_id)
+            # last parent is the root object
+            root_object_id = parents[0] if parents else object_id
+            logging.info(f"object_id: {str(object_id)} root_object_id: {str(root_object_id)}")
+            if str(root_object_id) in results:
+                results[str(root_object_id)].append(object_id)
+            else:
+                results[str(root_object_id)] = [root_object_id, object_id]
+        content = ""
+        result_desc = []
+        for result in results.values():
+            # first element in result is the root object
+            root_object_id = result[0]
+            if root_object_id.get_object_type() == ObjectType.Email:
+                email = KnowledgeStore().load_object(root_object_id)
+                desc = email.get_desc()
+                desc["type"] = "email"
+                desc["contents"] = []
+                result_desc.append(desc)
+                upper_list = desc["contents"]
+                result = result[1:]
+            else:
+                upper_list = result_desc
+            
+            for object_id in result:
+                if object_id.get_object_type() == ObjectType.Chunk:
+                    upper_list.append({"type": "text", "content": KnowledgeStore().get_chunk_reader().get_chunk(object_id).read().decode("utf-8")})
+                if object_id.get_object_type() == ObjectType.Image:
+                    # image = self.load_object(object_id)
+                    desc = dict()
+                    desc["id"] = str(object_id)
+                    desc["type"] = "image"
+                    upper_list.append(desc)
+                if object_id.get_object_type() == ObjectType.Video:
+                    video = KnowledgeStore().load_object(object_id)
+                    desc = video.get_desc()
+                    desc["type"] = "video"
+                    upper_list.append(desc)
+                else:
+                    pass
+        content += json.dumps(result_desc)
+        content += ".\n"  
+
+        return content
+    
+    async def _query(self, tokens: str, types: list[str] = ["text"], index: str=0):
+        index = int(index)
+        object_ids = await self.query_objects(tokens, types, 4)
+        if len(object_ids) <= index:
+            return "*** I have no more information for your reference.\n"
+        else:
+            content = "*** I have provided the following known information for your reference with json format:\n"
+            return content + self.tokens_from_objects(object_ids[index:index+1])
+
+def init() -> KnowledgeEnvironment:
+    return KnowledgeEnvironment("embedding")

rootfs/knowledge_pipelines/pipelines.toml

@@ -0,0 +1,3 @@
+pipelines = [
+    "Mia"
+]

src/aios_kernel/__init__.py

@@ -5,8 +5,6 @@ from .agent import AIAgent,AIAgentTemplete
 from .compute_kernel import ComputeKernel,ComputeTask,ComputeTaskResult,ComputeTaskState,ComputeTaskType
 from .compute_node import ComputeNode,LocalComputeNode
 from .open_ai_node import OpenAI_ComputeNode
-from .knowledge_base import KnowledgeBase, KnowledgeEnvironment
-from .knowledge_pipeline import KnowledgeEmailSource, KnowledgeDirSource, KnowledgePipline
 from .role import AIRole,AIRoleGroup
 from .workflow import Workflow
 from .bus import AIBus
@@ -25,5 +23,6 @@ from .local_stability_node import Local_Stability_ComputeNode
 from .stability_node import Stability_ComputeNode
 from .local_st_compute_node import LocalSentenceTransformer_Text_ComputeNode,LocalSentenceTransformer_Image_ComputeNode
 from .compute_node_config import ComputeNodeConfig
+from .ai_function import SimpleAIFunction
 AIOS_Version = "0.5.1, build 2023-9-28"
 

src/aios_kernel/agent.py

@@ -16,7 +16,6 @@ from .compute_task import ComputeTaskResult,ComputeTaskResultCode
 from .ai_function import AIFunction
 from .environment import Environment
 from .contact_manager import ContactManager,Contact,FamilyMember
-from .knowledge_base import KnowledgeBase
 from .compute_kernel import ComputeKernel
 from .bus import AIBus
 
@@ -123,7 +122,8 @@ class AIAgent:
             self.contact_prompt_str = config["contact_prompt"]
 
         if config.get("owner_env") is not None:
-            self.owner_env = Environment.get_env_by_id(config["owner_env"])
+            self.owner_env = config.get("owner_env")
+       
 
         if config.get("powerby") is not None:
             self.powerby = config["powerby"]
@@ -550,8 +550,7 @@ class AIAgent:
     def parser_learn_llm_result(self,llm_result:str):
         pass
 
-    async def _llm_read_article(self,kb:KnowledgeBase,item:KnowledgeObject) -> ComputeTaskResult:
+    async def _llm_read_article(self,item:KnowledgeObject) -> ComputeTaskResult:
-        #kb_env = KnowledgeBaseFileSystemEnvironment()
         full_content = item.get_article_full_content()
         full_content_len = ComputeKernel.llm_num_tokens_from_text(full_content,self.get_llm_model_name())
         if full_content_len < self.get_llm_learn_token_limit():

src/aios_kernel/knowledge_base.py

@@ -1,296 +0,0 @@
-# define a knowledge base class
-import json
-import logging
-
-from .agent_base import AgentPrompt
-from .compute_kernel import ComputeKernel 
-from .storage import AIStorage
-from .environment import Environment
-from .ai_function import SimpleAIFunction
-from knowledge import *
-
-
-class KnowledgeBase:
-    _instance = None
-
-    def __new__(cls):
-        if cls._instance is None:
-            cls._instance = super().__new__(cls)
-            cls._instance.__singleton_init__()
-
-        return cls._instance
-
-    def __singleton_init__(self) -> None:
-        self.store = KnowledgeStore()
-        self.compute_kernel = ComputeKernel.get_instance()
-        self._default_text_model = "all-MiniLM-L6-v2"
-        self._default_image_model = "clip-ViT-B-32"
-
-    async def __embedding_document(self, document: DocumentObject):
-        for chunk_id in document.get_chunk_list():
-            chunk = self.store.get_chunk_reader().get_chunk(chunk_id)
-            if chunk is None:
-                raise ValueError(f"text chunk not found: {chunk_id}")
-        
-            text = chunk.read().decode("utf-8")
-            vector = await self.compute_kernel.do_text_embedding(text, self._default_text_model)
-            if vector:
-                await self.store.get_vector_store(self._default_text_model).insert(vector, chunk_id)
-
-    async def __embedding_image(self, image: ImageObject):
-        # desc = {}
-        # if not not image.get_meta():
-        #     desc["meta"] = image.get_meta()
-        # if not not image.get_exif():
-        #     desc["exif"] = image.get_exif()
-        # if not not image.get_tags():
-        #     desc["tags"] = image.get_tags()
-        # vector = await self.compute_kernel.do_text_embedding(json.dumps(desc), self._default_text_model)
-        vector = await self.compute_kernel.do_image_embedding(image.calculate_id(), self._default_image_model)
-        if vector:
-            await self.store.get_vector_store(self._default_image_model).insert(vector, image.calculate_id())
-
-    async def __embedding_video(self, vedio: VideoObject):
-        desc = {}
-        if not not vedio.get_meta():
-            desc["meta"] = vedio.get_meta()
-        if not not vedio.get_info():
-            desc["info"] = vedio.get_info()
-        if not not vedio.get_tags():
-            desc["tags"] = vedio.get_tags()
-        vector = await self.compute_kernel.do_text_embedding(json.dumps(desc), self._default_text_model)
-        await self.store.get_vector_store(self._default_text_model).insert(vector, vedio.calculate_id())
-
-    async def __embedding_rich_text(self, rich_text: RichTextObject):
-        for document_id in rich_text.get_documents().values():
-            document = DocumentObject.decode(self.store.get_object_store().get_object(document_id))
-            await self.__embedding_document(document)
-        for image_id in rich_text.get_images().values():
-            image = ImageObject.decode(self.store.get_object_store().get_object(image_id))
-            await self.__embedding_image(image)
-        for video_id in rich_text.get_videos().values():
-            video = VideoObject.decode(self.store.get_object_store().get_object(video_id))
-            await self.__embedding_video(video)
-        for rich_text_id in rich_text.get_rich_texts().values():
-            rich_text = RichTextObject.decode(self.store.get_object_store().get_object(rich_text_id))
-            await self.__embedding_rich_text(rich_text)
-
-    async def __embedding_email(self, email: EmailObject):
-        vector = await self.compute_kernel.do_text_embedding(json.dumps(email.get_desc()), self._default_text_model)
-        await self.store.get_vector_store(self._default_text_model).insert(vector, email.calculate_id())
-        await self.__embedding_rich_text(email.get_rich_text())
-
-
-    async def __do_embedding(self, object: KnowledgeObject):
-        if object.get_object_type() == ObjectType.Document:
-            await self.__embedding_document(object)
-        if object.get_object_type() == ObjectType.Image:
-            await self.__embedding_image(object)
-        if object.get_object_type() == ObjectType.Video:
-            await self.__embedding_video(object)
-        if object.get_object_type() == ObjectType.RichText:
-            await self.__embedding_rich_text(object)
-        if object.get_object_type() == ObjectType.Email:
-            await self.__embedding_email(object)
-        else:
-            pass
-
-    # def __save_document(self, document: DocumentObject):
-    #     doc_id = document.calculate_id()
-    #     self.store.get_object_store().put_object(doc_id, document.encode())
-    #     for chunk_id in document.get_chunk_list():
-    #         self.store.get_relation_store().add_relation(chunk_id, doc_id)
-
-    # def __save_image(self, image: ImageObject):
-    #     image_id = image.calculate_id()
-    #     self.store.get_object_store().put_object(image_id, image.encode())
-
-    # def __save_video(self, video: VideoObject):
-    #     video_id = video.calculate_id()
-    #     self.store.get_object_store().put_object(video_id, video.encode())
-
-    # def __save_rich_text(self, rich_text: RichTextObject):
-    #     rich_text_id = rich_text.calculate_id()
-    #     # rich_text_enc = dict()
-    #     # rich_text_enc["desc"] = rich_text.desc
-    #     # rich_text_enc["body"] = {"documents": {}, "images": {}, "videos": {}, "rich_texts": {}}
-    #     for key, document in rich_text.get_documents().items():
-    #         self.__save_document(document)
-    #         doc_id = document.calculate_id()
-    #         self.store.get_relation_store().add_relation(doc_id, rich_text_id)
-    #         # rich_text_enc["body"]["documents"][key] = doc_id
-    #     for key, image in rich_text.get_images().items():
-    #         self.__save_image(image)
-    #         image_id = image.calculate_id()
-    #         self.store.get_relation_store().add_relation(image_id, rich_text_id)
-    #         # rich_text_enc["body"]["images"][key] = image_id
-    #     for key, video in rich_text.get_videos().items():
-    #         self.__save_video(video)
-    #         video_id = video.calculate_id()
-    #         self.store.get_relation_store().add_relation(video_id, rich_text_id)
-    #         # rich_text_enc["body"]["videos"][key] = video_id
-    #     for key, rich_text in rich_text.get_rich_texts().items():
-    #         self.__save_rich_text(rich_text)
-    #         rich_text_id = rich_text.calculate_id()
-    #         self.store.get_relation_store().add_relation(rich_text_id, rich_text_id)
-    #         # rich_text_enc["body"]["rich_texts"][key] = rich_text_id
-
-
-    #     self.store.get_object_store().put_object(rich_text_id, rich_text.encode())
-
-    # def __save_email(self, email: EmailObject):
-    #     email_id = email.calculate_id()
-    #     # email_enc = dict()
-    #     # email_enc["desc"] = email.desc
-    #     # email_enc["body"] = {"content": None}
-    #     self.__save_rich_text(email.get_rich_text())
-    #     rich_text_id = email.get_rich_text().calculate_id()
-    #     self.store.get_relation_store().add_relation(rich_text_id, email_id)
-    #     # email_enc["body"]["content"] = rich_text_id
-    #     self.store.get_object_store().put_object(email_id, email.encode())
-
-    
-    # def __save_object(self, object: KnowledgeObject):
-    #     if object.get_object_type() == ObjectType.Document:
-    #         self.__save_document(object)
-    #     if object.get_object_type() == ObjectType.Image:
-    #         self.__save_image(object)
-    #     if object.get_object_type() == ObjectType.Video:
-    #         self.__save_video(object)
-    #     if object.get_object_type() == ObjectType.RichText:
-    #         self.__save_rich_text(object)
-    #     if object.get_object_type() == ObjectType.Email:
-    #         self.__save_email(object)
-    #     else:
-    #         pass
-
-    async def insert_object(self, object: KnowledgeObject):
-        self.store.get_object_store().put_object(object.calculate_id(), object.encode())
-        await self.__do_embedding(object)
-    
-    async def query_objects(self, tokens: str, types: list[str], topk: int) -> [ObjectID]:
-        texts = []
-        if "text" in types:
-            vector = await self.compute_kernel.do_text_embedding(tokens, self._default_text_model)
-            texts = await self.store.get_vector_store(self._default_text_model).query(vector, topk)
-        images = []
-        if "image" in types:
-            vector = await self.compute_kernel.do_text_embedding(tokens, self._default_image_model)
-            images = await self.store.get_vector_store(self._default_image_model).query(vector, topk)
-        return texts + images
-
-    def load_object(self, object_id: ObjectID) -> KnowledgeObject:
-        if object_id.get_object_type() == ObjectType.Document:
-            return DocumentObject.decode(self.store.get_object_store().get_object(object_id))
-        if object_id.get_object_type() == ObjectType.Image:
-            return ImageObject.decode(self.store.get_object_store().get_object(object_id))
-        if object_id.get_object_type() == ObjectType.Video:
-            return VideoObject.decode(self.store.get_object_store().get_object(object_id))
-        if object_id.get_object_type() == ObjectType.RichText:
-            return RichTextObject.decode(self.store.get_object_store().get_object(object_id))
-        if object_id.get_object_type() == ObjectType.Email:
-            return EmailObject.decode(self.store.get_object_store().get_object(object_id))
-        else:
-            pass
-        
-
-    def tokens_from_objects(self, object_ids: [ObjectID]) -> list[str]:
-        results = dict()
-        for object_id in object_ids:
-            parents = self.store.get_relation_store().get_related_root_objects(object_id)
-            # last parent is the root object
-            root_object_id = parents[0] if parents else object_id
-            logging.info(f"object_id: {str(object_id)} root_object_id: {str(root_object_id)}")
-            if str(root_object_id) in results:
-                results[str(root_object_id)].append(object_id)
-            else:
-                results[str(root_object_id)] = [root_object_id, object_id]
-        content = ""
-        result_desc = []
-        for result in results.values():
-            # first element in result is the root object
-            root_object_id = result[0]
-            if root_object_id.get_object_type() == ObjectType.Email:
-                email = self.load_object(root_object_id)
-                desc = email.get_desc()
-                desc["type"] = "email"
-                desc["contents"] = []
-                result_desc.append(desc)
-                upper_list = desc["contents"]
-                result = result[1:]
-            else:
-                upper_list = result_desc
-            
-            for object_id in result:
-                if object_id.get_object_type() == ObjectType.Chunk:
-                    upper_list.append({"type": "text", "content": self.store.get_chunk_reader().get_chunk(object_id).read().decode("utf-8")})
-                if object_id.get_object_type() == ObjectType.Image:
-                    # image = self.load_object(object_id)
-                    desc = dict()
-                    desc["id"] = str(object_id)
-                    desc["type"] = "image"
-                    upper_list.append(desc)
-                if object_id.get_object_type() == ObjectType.Video:
-                    video = self.load_object(object_id)
-                    desc = video.get_desc()
-                    desc["type"] = "video"
-                    upper_list.append(desc)
-                else:
-                    pass
-        content += json.dumps(result_desc)
-        content += ".\n"  
-
-        return content
-    
-    def parse_object_in_message(self, message: str) -> KnowledgeObject:
-        # get message's first line 
-        logging.info(f"tg parse resp message: {message}")
-        lines = message.split("\n")
-        if len(lines) > 0:
-            message = lines[0]
-            try:
-                desc = json.loads(message)
-                if isinstance(desc, dict):
-                    object_id = desc["id"]
-                else:
-                    object_id = desc[0]["id"]
-            except Exception as e:
-                return None
-            
-            if object_id is not None:
-                return self.load_object(ObjectID.from_base58(object_id))
-            
-            
-    def bytes_from_object(self, object: KnowledgeObject) -> bytes:
-        if object.get_object_type() == ObjectType.Image:
-            image_object = object
-            return self.store.get_chunk_reader().read_chunk_list_to_single_bytes(image_object.get_chunk_list())
-           
-
-
-           
-
-
-
-class KnowledgeEnvironment(Environment):
-    def __init__(self, env_id: str) -> None:
-        super().__init__(env_id)
-
-        query_param = {
-            "tokens": "key words to query", 
-            "types": "prefered knowledge types, one or more of [text, image]",
-            "index": "index of query result"
-        }
-        self.add_ai_function(SimpleAIFunction("query_knowledge", 
-                                            "vector query content from local knowledge base",
-                                            self._query, 
-                                            query_param))
-        
-    async def _query(self, tokens: str, types: list[str] = ["text"], index: str=0):
-        index = int(index)
-        object_ids = await KnowledgeBase().query_objects(tokens, types, 4)
-        if len(object_ids) <= index:
-            return "*** I have no more information for your reference.\n"
-        else:
-            content = "*** I have provided the following known information for your reference with json format:\n"
-            return content + KnowledgeBase().tokens_from_objects(object_ids[index:index+1])

src/aios_kernel/knowledge_pipeline.py

@@ -1,411 +0,0 @@
-"""
-Capture your email locally, and parse out the pictures in the email body and the pictures, videos and other files in the attachment. Subsequently, it supports vectorized analysis of your personal data and serves as a knowledge base to enable large language model answers. Better results.
-
-An example of a local file is as follows:
-├── data
-│ └── alex0072@gmail.com
-│   └── 5de3e52f3a6b90cabe6cbdd4ae3a5c5b
-│     ├── email.txt
-│     ├── meta.json
-│     ├── image
-│     │   ├── 0648B869@99C03070.DB94B354.jpg
-│     └── body_image
-│         ├── 11044884873.jpg
-│         ├── 282985198265470.gif
-│         └── dd-login-service-min.png
-
-"""
-import asyncio
-import datetime
-import sqlite3
-import imaplib
-import logging
-import mailparser
-import hashlib
-import json
-import base64
-import chardet
-import aiofiles
-
-from bs4 import BeautifulSoup
-import requests
-import os
-import toml
-from .storage import AIStorage, UserConfigItem
-from .knowledge_base import KnowledgeBase, ImageObjectBuilder, ObjectID, ObjectType, DocumentObjectBuilder, EmailObjectBuilder, EmailObject
-
-class KnowledgeJournal:
-    def __init__(self, source_type: str, source_id: str, item_id: str, object_id: str, timestamp=None):
-        # define a timestamp variable
-        self.timestamp = datetime.datetime.now() if timestamp is None else timestamp
-        self.object_id = object_id
-        self.source_type = source_type
-        self.source_id = source_id
-        self.item_id = item_id
-
-    def __str__(self) -> str:
-        if self.source_type == "dir":
-            object_id = ObjectID.from_base58(self.object_id)
-            object_type = None
-            if object_id.get_object_type() == ObjectType.Image:
-                object_type = "image"
-            else:
-                pass
-            return f"Add {object_type} from {os.path.join(self.source_id, self.item_id)}"
-        if self.source_type == "email":
-            object_id = ObjectID.from_base58(self.object_id)
-            email = EmailObject.decode(KnowledgeBase().store.get_object_store().get_object(object_id))
-            meta = email.get_meta()
-            return f'Add email from {os.path.join(self.source_id)} subject {meta["subject"]}'
-
-
-# init sqlite3 client
-class KnowledgeJournalClient:
-    def __init__(self):
-        knowledge_dir = os.path.join(AIStorage.get_instance().get_myai_dir(), "knowledge")
-        if not os.path.exists(knowledge_dir):
-            os.makedirs(knowledge_dir)
-        self.journal_path = os.path.join(knowledge_dir, "journal.db")
-    
-        conn = sqlite3.connect(self.journal_path)
-        conn.execute(
-            '''CREATE TABLE IF NOT EXISTS journal (
-                id INTEGER PRIMARY KEY AUTOINCREMENT, 
-                time DATETIME DEFAULT CURRENT_TIMESTAMP, 
-                source_type TEXT,
-                source_id TEXT,
-                item_id TEXT, 
-                object_id TEXT)'''
-        )
-        conn.commit()
-
-    def insert(self, journal: KnowledgeJournal):
-        conn = sqlite3.connect(self.journal_path)
-        conn.execute(
-            "INSERT INTO journal (time, source_type, source_id, item_id, object_id) VALUES (?, ?, ?, ?, ?)",
-            (journal.timestamp, journal.source_type, journal.source_id, journal.item_id, journal.object_id),
-        )
-        conn.commit()
-
-    def latest_journal(self, source_id: str) -> KnowledgeJournal:
-        conn = sqlite3.connect(self.journal_path)
-        cursor = conn.cursor()
-        cursor.execute("SELECT * FROM journal WHERE source_id = ? ORDER BY id DESC LIMIT 1", (source_id,))
-        result = cursor.fetchone()
-        if result is None:
-            return None
-        else:
-            (_, timestamp, source_type, sorce_id, item_id, object_id) = result
-            return KnowledgeJournal(source_type, sorce_id, item_id, object_id, timestamp)
-        
-    def latest_journals(self, topn) -> [KnowledgeJournal]:
-        conn = sqlite3.connect(self.journal_path)
-        cursor = conn.cursor()
-        cursor.execute("SELECT * FROM journal ORDER BY id DESC LIMIT ?", (topn,))
-        return [KnowledgeJournal(source_type, sorce_id, item_id, object_id, timestamp) for (_, timestamp, source_type, sorce_id, item_id, object_id) in cursor.fetchall()]
-
-
-class KnowledgeEmailSource:
-    def __init__(self, config:dict): 
-        self.config = config
-        self.config["type"] = "email"
-    
-    def id(self):
-        return self.config["address"]
-
-    @classmethod
-    def user_config_items(cls):
-        return [("address", "email address"),
-                ("password", "email password"),
-                ("imap_server", "imap server"),
-                ("imap_port", "imap port")
-                ]
-    
-    @classmethod
-    def local_root(cls):
-        user_data_dir = AIStorage.get_instance().get_myai_dir()
-        return os.path.abspath(f"{user_data_dir}/knowledge/email")    
-
-    async def run_once(self):
-        # read config from toml file
-        # and read from config config.local.toml if exists (config.local.toml is ignored by git)
-        logging.debug(f"knowledge email source {self.id()} run once")
-        filter = "ALL"  
-        self.client = self.email_client()
-        await self.read_emails(imap_keyword=filter)
-
-    def email_client(self) -> imaplib.IMAP4_SSL:
-        logging.info(f"read email config from {self.config.get('imap_server')}")
-        client = imaplib.IMAP4_SSL(
-            host=self.config.get('imap_server'), 
-            port=self.config.get('imap_port')
-        )
-        client.login(self.config.get('address'), self.config.get('password'))
-        return client
-
-    async def read_emails(self, folder: str = 'INBOX', imap_keyword: str = "UNSEEN"):
-        journal_client = KnowledgeJournalClient()
-        latest_journal = journal_client.latest_journal(self.id())
-        latest_uid = 0 if latest_journal is None else int(latest_journal.item_id)
-        self.client.select(folder)
-        _, data = self.client.uid('search', None, imap_keyword)
-        
-        # get email uid list
-        email_list = data[0].split()
-        logging.info(f"got {len(email_list)} emails")
-        journal_client = KnowledgeJournalClient()
-        for uid in email_list:
-            _uid = int.from_bytes(uid)
-            if _uid > latest_uid:
-                email_dir = self.check_email_saved(uid)
-                if email_dir is not None:
-                    logging.info(f"email uid {uid} already saved")
-                else:
-                    email_dir = self.read_and_save_email(uid)
-                    logging.info(f"email uid {uid} saved")
-                email_object = EmailObjectBuilder({}, email_dir).build()
-                await KnowledgeBase().insert_object(email_object)
-                journal_client.insert(KnowledgeJournal("email", self.id(), str(int.from_bytes(uid)), str(email_object.calculate_id())))
-
-
-    def read_and_save_email(self, uid: str) -> str:
-        message_parts = "(BODY.PEEK[])"
-        _, email_data = self.client.uid('fetch', uid, message_parts)
-        mail = mailparser.parse_from_bytes(email_data[0][1])
-        logging.info(f"got email subject [{mail.subject}]")
-        self.save_email(mail)
-        return self.get_local_dir_name(mail)
-
-    def get_local_dir_name(self, mail: mailparser.MailParser) -> str:
-        dir =  f"{self.local_root()}/{self.config.get('address')}"
-        name = f"{mail.subject}__{mail.date}"
-        name = hashlib.md5(name.encode('utf-8')).hexdigest()
-        return f"{dir}/{name}"
-
-    def check_email_saved(self, uid: str) -> str:
-        message_parts = "(BODY[HEADER])"
-        _, email_data = self.client.uid('fetch', uid, message_parts)
-        mail = mailparser.parse_from_bytes(email_data[0][1])
-        logging.info(f"[{uid}]check email subject [{mail.subject}]")
-        dir = self.get_local_dir_name(mail)
-        logging.info(f"check email saved {dir}")
-        file = f"{dir}/email.txt"
-        if os.path.exists(file):
-            return dir
-        return None
-
-    # save email attachment(images)
-    def save_email_attachment(self, mail: mailparser.MailParser, email_dir: str):
-        for attachment in mail.attachments:
-            if attachment['mail_content_type'] in ['image/png', 'image/jpeg', 'image/gif']:
-                print('current mail have image attachment')
-                img_dir = f"{email_dir}/image"
-                if not os.path.exists(img_dir):
-                    os.makedirs(img_dir)
-                filename = attachment['filename']
-                filefullname = f"{img_dir}/{filename}"
-                image_data = attachment['payload']
-                try:
-                    image_data = base64.b64decode(image_data)
-                except base64.binascii.Error:
-                    image_data = image_data.encode()
-                with open(filefullname, 'wb') as f:
-                    f.write(image_data)
-                    logging.info(f"save email image {filename} success")
-
-    # save email body images(html content)
-    def save_body_images(self, html_content: str, email_dir: str):
-        # get all image urls
-        soup = BeautifulSoup(html_content, 'html.parser')
-        img_tags = soup.find_all('img')
-        img_urls = [img['src'] for img in img_tags if 'src' in img.attrs]
-        logging.info(f'Found {len(img_urls)} images in email body')
-
-        name_count = 0
-        
-        if not os.path.exists(email_dir):
-            os.makedirs(email_dir)
-
-        for img_url in img_urls:
-            # keep the original image filename(last of url)
-            ext = img_url.split('/')[-1].split('.')[-1]
-            img_filename = os.path.join(email_dir, f"{name_count}.{ext}")
-            name_count += 1
-            # download image 
-            response = requests.get(img_url, stream=True)
-            if response.status_code == 200:
-                with open(img_filename, 'wb') as img_file:
-                    for chunk in response.iter_content(1024):
-                        img_file.write(chunk)
-                logging.info(f'Downloaded {img_url} to {img_filename}')
-            else:
-                logging.info(f'Failed to download {img_url}')
-
-    # save email content to local dir
-    def save_email(self, mail: mailparser.MailParser):
-        dir = f"{self.local_root()}/{self.config.get('address')}"
-        if not os.path.exists(dir):
-            os.makedirs(dir)
-        email_dir = self.get_local_dir_name(mail)
-        logging.info(f"save email to {email_dir}")
-        if not os.path.exists(email_dir):
-            os.makedirs(email_dir)
-        with open(f"{email_dir}/email.txt", "w", encoding='utf-8') as f:
-            # soup = BeautifulSoup(mail.body, 'html.parser')
-            f.write(mail.body)
-        with open(f"{email_dir}/meta.json", "w", encoding='utf-8') as f:
-            mail_dict = json.loads(mail.mail_json)
-            if 'body' in mail_dict:
-                del mail_dict['body']
-            json.dump(mail_dict, f, ensure_ascii=False, indent=4)
-            logging.info(f"save email meta info {f.name}")
-        
-        self.save_email_attachment(mail, email_dir)
-        self.save_body_images(mail.body, f"{email_dir}/body_image")
-
-
-class KnowledgeDirSource:
-    def __init__(self, config):
-        self.config = config
-        config["path"] = os.path.abspath(config["path"])
-        self.config["type"] = "dir"
-
-    @classmethod
-    def user_config_items(cls):
-        return [("path", "local dir path")]
-    
-    def id(self):
-        return self.config["path"]
-
-    def path(self):
-        return self.config["path"]
-    
-    @staticmethod
-    async def read_txt_file(file_path:str)->str:
-        cur_encode = "utf-8"
-        async with aiofiles.open(file_path,'rb') as f:
-            cur_encode = chardet.detect(await f.read())['encoding']
-        
-        async with aiofiles.open(file_path,'r',encoding=cur_encode) as f:
-            return await f.read()
-        
-    async def run_once(self):
-        logging.debug(f"knowledge dir source {self.id()} run once")
-        journal_client = KnowledgeJournalClient()
-        latest_journal = journal_client.latest_journal(self.id())
-        if latest_journal is not None:
-            if os.path.getmtime(self.path()) <= latest_journal.timestamp:
-                logging.debug(f"knowledge dir source {self.id()} ingnored for no update")
-                return
-        file_pathes = sorted(os.listdir(self.path()), key=lambda x: os.path.getctime(os.path.join(self.path(), x)))
-        for rel_path in file_pathes:
-            file_path = os.path.join(self.path(), rel_path)
-            timestamp = os.path.getctime(file_path)
-            if latest_journal is not None:
-                if timestamp <= latest_journal.timestamp:
-                    continue
-            ext = os.path.splitext(file_path)[1].lower()
-            if ext in ['.jpg', '.jpeg', '.png', '.gif', '.bmp']:
-                logging.info(f"knowledge dir source {self.id()} found image file {file_path}")
-                image = ImageObjectBuilder({}, {}, file_path).build()
-                await KnowledgeBase().insert_object(image)
-                journal_client.insert(KnowledgeJournal("dir", self.id(), rel_path, str(image.calculate_id()), timestamp))
-            if ext in ['.txt']:
-                logging.info(f"knowledge dir source {self.id()} found text file {file_path}")
-                text = await self.read_txt_file(file_path)
-    
-                document = DocumentObjectBuilder({}, {}, text).build()
-                await KnowledgeBase().insert_object(document)
-                journal_client.insert(KnowledgeJournal("dir", self.id(), rel_path, str(document.calculate_id()), timestamp))
-            
-
-
-
-# define singleton class knowledge pipline
-class KnowledgePipline:
-    _instance = None
-
-    @classmethod
-    def get_instance(cls):
-        if cls._instance is None:
-            cls._instance = KnowledgePipline()
-            cls._instance.__singleton_init__()
-
-        return cls._instance
-    
-    def initial(self):
-        config_path = self.__config_path()
-        logging.info(f"initial knowledge pipline from {config_path}")
-        if os.path.exists(config_path):
-            config = toml.load(self.__config_path())
-            for source_config in config["sources"]:
-                if source_config['type'] == 'email':
-                    self.add_email_source(KnowledgeEmailSource(source_config))
-                if source_config['type'] == 'dir':
-                    self.add_dir_source(KnowledgeDirSource(source_config))
-        user_data_dir = AIStorage.get_instance().get_myai_dir()
-        default_dir = os.path.abspath(f"{user_data_dir}/data")
-        if not os.path.exists(default_dir):
-            os.makedirs(default_dir)
-        self.add_dir_source(KnowledgeDirSource({"path": default_dir}))
-
-        return True
-
-    def __singleton_init__(self):
-        self.knowledge_base = KnowledgeBase()
-        self.email_sources = dict()
-        self.dir_sources = dict()
-        self.source_queue = list()
-        self.run_lock = asyncio.Lock()
-        asyncio.create_task(self.run_loop())
-        
-
-    def save_config(self):
-        config = dict()
-        config["sources"] = [source.config for source in self.source_queue]
-        with open(self.__config_path(), "w") as f:
-            toml.dump(config, f)
-        
-
-    @classmethod
-    def __config_path(cls) -> str:
-        user_data_dir = AIStorage.get_instance().get_myai_dir()
-        return os.path.abspath(f"{user_data_dir}/etc/knowledge.cfg.toml")
-
-
-    def add_email_source(self, source: KnowledgeEmailSource):
-        if self.email_sources.get(source.id()) is not None: 
-            return "already exists"
-        self.email_sources[source.id()] = source
-        self.source_queue.append(source)
-        return None
-
-    def add_dir_source(self, source: KnowledgeDirSource):
-        if self.dir_sources.get(source.id()) is not None:
-            logging.info(f"knowledge add source {source.id()} failed for already exists") 
-            return "already exists"
-        logging.info(f"knowledge added source {source.id()}") 
-        self.dir_sources[source.id()] = source
-        self.source_queue.append(source)
-        return None
-    
-    def get_latest_journals(self, topn) -> [KnowledgeJournal]:
-        return KnowledgeJournalClient().latest_journals(topn)
-    
-    async def run_loop(self):
-        while True:
-            await self.run_once()
-            await asyncio.sleep(5)
-
-    async def run_once(self):
-        logging.info(f"knowledge pipeline started") 
-        # sources = list()
-        # async with self.run_lock:
-        #     for source in self.source_queue:
-        #         sources.append(source)
-        # for source in sources: 
-        #     await source.run_once()
-        for source in self.source_queue:
-            await source.run_once()
-        
-        logging.info(f"knowledge pipeline finished") 

src/aios_kernel/tg_tunnel.py

@@ -10,9 +10,7 @@ from telegram import Bot
 from telegram.ext import Updater
 from telegram.error import Forbidden, NetworkError
 
-from knowledge.object.object_id import ObjectType
+from knowledge import ObjectType, KnowledgeStore
-
-from .knowledge_base import KnowledgeBase
 
 from .tunnel import AgentTunnel
 from .storage import AIStorage
@@ -236,10 +234,10 @@ class TelegramTunnel(AgentTunnel):
                     await update.message.reply_text("")
                     return
                 
-                knowledge_object = KnowledgeBase().parse_object_in_message(resp_msg.body)
+                knowledge_object = KnowledgeStore().parse_object_in_message(resp_msg.body)
                 if knowledge_object is not None:
                     if knowledge_object.get_object_type() == ObjectType.Image:
-                        image = KnowledgeBase().bytes_from_object(knowledge_object)
+                        image = KnowledgeStore().bytes_from_object(knowledge_object)
                         try:
                             async with aiofiles.open("tg_send_temp.png", mode='wb') as local_file:
                                 if local_file:

src/component/agent_manager/agent_manager.py

@@ -1,9 +1,11 @@
 
 import logging
 import toml
+import os
+import runpy
 from typing import Any, Callable, Dict, List, Optional, Union
 
-from aios_kernel import AIAgent,AIAgentTemplete,AIStorage
+from aios_kernel import AIAgent,AIAgentTemplete,AIStorage,Environment
 from package_manager import PackageEnv,PackageEnvManager,PackageMediaInfo,PackageInstallTask
 
 logger = logging.getLogger(__name__)
@@ -105,6 +107,18 @@ class AgentManager:
             config_data = await config_file.read()
             config = toml.loads(config_data)
             result_agent = AIAgent()
+
+            if "owner_env" in config:
+                owner_env = config["owner_env"]
+                _, ext = os.path.splitext(owner_env)
+                if ext == ".py":
+                    env_path = os.path.join(agent_media.full_path, owner_env)
+                    owner_env = runpy.run_path(env_path)["init"]()
+                    config["owner_env"] = owner_env
+                else:
+                    owner_env = Environment.get_env_by_id(config["owner_env"])
+                config["owner_env"] = owner_env
+
             if result_agent.load_from_config(config) is False:
                 logger.error(f"load agent from {agent_media} failed!")
                 return None

src/component/knowledge_manager/__init__.py

@@ -0,0 +1 @@
+from .pipeline import KnowledgePipelineManager

src/component/knowledge_manager/input/email.py

@@ -1,103 +1,92 @@
-"""
-Capture your email locally, and parse out the pictures in the email body and the pictures, videos and other files in the attachment. Subsequently, it supports vectorized analysis of your personal data and serves as a knowledge base to enable large language model answers. Better results.
 
-An example of a local file is as follows:
+class KnowledgeEmailSource:
-├── data
+    def __init__(self, config:dict): 
-│ └── alex0072@gmail.com
+        self.config = config
-│   └── 5de3e52f3a6b90cabe6cbdd4ae3a5c5b
+        self.config["type"] = "email"
-│     ├── email.txt
-│     ├── meta.json
-│     ├── image
-│     │   ├── 0648B869@99C03070.DB94B354.jpg
-│     └── body_image
-│         ├── 11044884873.jpg
-│         ├── 282985198265470.gif
-│         └── dd-login-service-min.png
     
-"""
+    def id(self):
+        return self.config["address"]
 
-import imaplib
+    @classmethod
-import os
+    def user_config_items(cls):
-import toml
+        return [("address", "email address"),
-import logging
+                ("password", "email password"),
-import mailparser
+                ("imap_server", "imap server"),
-import hashlib
+                ("imap_port", "imap port")
-import json
+                ]
-import base64
-from bs4 import BeautifulSoup
-import requests
     
-class EmailSpider:
+    @classmethod
-    def __init__(self):
+    def local_root(cls):
-        # logger config
+        user_data_dir = AIStorage.get_instance().get_myai_dir()
-        self.logger = logging.getLogger('email spider')
+        return os.path.abspath(f"{user_data_dir}/knowledge/email")    
-        self.logger.setLevel(logging.DEBUG)
-        ch = logging.StreamHandler()
-        formatter = logging.Formatter('%(asctime)s [%(name)s] [%(levelname)s] %(message)s')
-        ch.setFormatter(formatter)
-        self.logger.addHandler(ch)
 
+    async def run_once(self):
         # read config from toml file
         # and read from config config.local.toml if exists (config.local.toml is ignored by git)
-        self.config = toml.load('./rootfs/email/config.toml')
+        logging.debug(f"knowledge email source {self.id()} run once")
-        if os.path.exists('./rootfs/email/config.local.toml'):
+        filter = "ALL"  
-            self.config = toml.load('./rootfs/email/config.local.toml')
-
         self.client = self.email_client()
+        await self.read_emails(imap_keyword=filter)
 
     def email_client(self) -> imaplib.IMAP4_SSL:
-        self.logger.info(f"read email config from {self.config.get('EMAIL_IMAP_SERVER')}")
+        logging.info(f"read email config from {self.config.get('imap_server')}")
         client = imaplib.IMAP4_SSL(
-            host=self.config.get('EMAIL_IMAP_SERVER'), 
+            host=self.config.get('imap_server'), 
-            port=self.config.get('EMAIL_IMAP_PORT')
+            port=self.config.get('imap_port')
         )
-        client.login(self.config.get('EMAIL_ADDRESS'), self.config.get('EMAIL_PASSWORD'))
+        client.login(self.config.get('address'), self.config.get('password'))
         return client
 
-    def list_box(self):
+    async def read_emails(self, folder: str = 'INBOX', imap_keyword: str = "UNSEEN"):
-        _, mailbox_list = self.client.list()
+        journal_client = KnowledgeJournalClient()
-        for mailbox in mailbox_list:
+        latest_journal = journal_client.latest_journal(self.id())
-            print(mailbox.decode())
+        latest_uid = 0 if latest_journal is None else int(latest_journal.item_id)
-
-    def read_emails(self, folder: str = 'INBOX', imap_keyword: str = "UNSEEN"):
         self.client.select(folder)
         _, data = self.client.uid('search', None, imap_keyword)
         
         # get email uid list
         email_list = data[0].split()
-        self.logger.info(f"got {len(email_list)} emails")
+        logging.info(f"got {len(email_list)} emails")
-        email_list.reverse()
+        journal_client = KnowledgeJournalClient()
         for uid in email_list:
-            if self.check_email_saved(uid):
+            _uid = int.from_bytes(uid)
-                self.logger.info(f"email uid {uid} already saved")
+            if _uid > latest_uid:
-            else:
+                email_dir = self.check_email_saved(uid)
-                self.read_and_save_email(uid)
+                if email_dir is not None:
-                self.logger.info(f"email uid {uid} saved")
+                    logging.info(f"email uid {uid} already saved")
+                else:
+                    email_dir = self.read_and_save_email(uid)
+                    logging.info(f"email uid {uid} saved")
+                email_object = EmailObjectBuilder({}, email_dir).build()
+                await KnowledgeBase().insert_object(email_object)
+                journal_client.insert(KnowledgeJournal("email", self.id(), str(int.from_bytes(uid)), str(email_object.calculate_id())))
 
-    def read_and_save_email(self, uid: str):
+
+    def read_and_save_email(self, uid: str) -> str:
         message_parts = "(BODY.PEEK[])"
         _, email_data = self.client.uid('fetch', uid, message_parts)
         mail = mailparser.parse_from_bytes(email_data[0][1])
-        self.logger.info(f"got email subject [{mail.subject}]")
+        logging.info(f"got email subject [{mail.subject}]")
         self.save_email(mail)
+        return self.get_local_dir_name(mail)
 
     def get_local_dir_name(self, mail: mailparser.MailParser) -> str:
-        dir =  f"{self.config.get('LOCAL_DIR')}/{self.config.get('EMAIL_ADDRESS')}"
+        dir =  f"{self.local_root()}/{self.config.get('address')}"
         name = f"{mail.subject}__{mail.date}"
         name = hashlib.md5(name.encode('utf-8')).hexdigest()
         return f"{dir}/{name}"
 
-    def check_email_saved(self, uid: str):
+    def check_email_saved(self, uid: str) -> str:
         message_parts = "(BODY[HEADER])"
         _, email_data = self.client.uid('fetch', uid, message_parts)
         mail = mailparser.parse_from_bytes(email_data[0][1])
-        self.logger.info(f"[{uid}]check email subject [{mail.subject}]")
+        logging.info(f"[{uid}]check email subject [{mail.subject}]")
         dir = self.get_local_dir_name(mail)
-        self.logger.info(f"check email saved {dir}")
+        logging.info(f"check email saved {dir}")
         file = f"{dir}/email.txt"
         if os.path.exists(file):
-            return False
+            return dir
-        return False
+        return None
 
     # save email attachment(images)
     def save_email_attachment(self, mail: mailparser.MailParser, email_dir: str):
@@ -116,7 +105,7 @@ class EmailSpider:
                     image_data = image_data.encode()
                 with open(filefullname, 'wb') as f:
                     f.write(image_data)
-                    self.logger.info(f"save email image {filename} success")
+                    logging.info(f"save email image {filename} success")
 
     # save email body images(html content)
     def save_body_images(self, html_content: str, email_dir: str):
@@ -124,48 +113,46 @@ class EmailSpider:
         soup = BeautifulSoup(html_content, 'html.parser')
         img_tags = soup.find_all('img')
         img_urls = [img['src'] for img in img_tags if 'src' in img.attrs]
-        self.logger.info(f'Found {len(img_urls)} images in email body')
+        logging.info(f'Found {len(img_urls)} images in email body')
+
+        name_count = 0
         
         if not os.path.exists(email_dir):
             os.makedirs(email_dir)
 
         for img_url in img_urls:
             # keep the original image filename(last of url)
-            img_filename = os.path.join(email_dir, img_url.split('/')[-1])
+            ext = img_url.split('/')[-1].split('.')[-1]
+            img_filename = os.path.join(email_dir, f"{name_count}.{ext}")
+            name_count += 1
             # download image 
             response = requests.get(img_url, stream=True)
             if response.status_code == 200:
                 with open(img_filename, 'wb') as img_file:
                     for chunk in response.iter_content(1024):
                         img_file.write(chunk)
-                self.logger.info(f'Downloaded {img_url} to {img_filename}')
+                logging.info(f'Downloaded {img_url} to {img_filename}')
             else:
-                self.logger.info(f'Failed to download {img_url}')
+                logging.info(f'Failed to download {img_url}')
 
     # save email content to local dir
     def save_email(self, mail: mailparser.MailParser):
-        dir = f"{self.config.get('LOCAL_DIR')}/{self.config.get('EMAIL_ADDRESS')}"
+        dir = f"{self.local_root()}/{self.config.get('address')}"
         if not os.path.exists(dir):
             os.makedirs(dir)
         email_dir = self.get_local_dir_name(mail)
-        self.logger.info(f"save email to {email_dir}")
+        logging.info(f"save email to {email_dir}")
         if not os.path.exists(email_dir):
             os.makedirs(email_dir)
-        with open(f"{email_dir}/email.txt", "w") as f:
+        with open(f"{email_dir}/email.txt", "w", encoding='utf-8') as f:
+            # soup = BeautifulSoup(mail.body, 'html.parser')
             f.write(mail.body)
         with open(f"{email_dir}/meta.json", "w", encoding='utf-8') as f:
             mail_dict = json.loads(mail.mail_json)
             if 'body' in mail_dict:
                 del mail_dict['body']
             json.dump(mail_dict, f, ensure_ascii=False, indent=4)
-            self.logger.info(f"save email meta info {f.name}")
+            logging.info(f"save email meta info {f.name}")
         
         self.save_email_attachment(mail, email_dir)
         self.save_body_images(mail.body, f"{email_dir}/body_image")
-
-
-if __name__ == "__main__":
-    spider = EmailSpider()
-    folder = 'INBOX'
-    imap_keyword = "ALL"
-    spider.read_emails(folder, imap_keyword)

src/component/knowledge_manager/input/local_dir.py

@@ -0,0 +1,68 @@
+import os
+import aiofiles
+import chardet
+import logging
+import string
+from knowledge import ImageObjectBuilder, DocumentObjectBuilder, KnowledgePipelineEnvironment, KnowledgePipelineJournal
+from aios_kernel.storage import AIStorage
+
+class KnowledgeDirSource:
+    def __init__(self, env: KnowledgePipelineEnvironment, config):
+        self.env = env
+        path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir())
+        config["path"] = path
+        self.config = config  
+
+    # @classmethod
+    # def user_config_items(cls):
+    #     return [("path", "local dir path")]
+
+    def path(self):
+        return self.config["path"]
+    
+    @staticmethod
+    async def read_txt_file(file_path:str)->str:
+        cur_encode = "utf-8"
+        async with aiofiles.open(file_path,'rb') as f:
+            cur_encode = chardet.detect(await f.read())['encoding']
+        
+        async with aiofiles.open(file_path,'r',encoding=cur_encode) as f:
+            return await f.read()
+        
+    async def next(self):
+        while True:
+            journals = self.env.journal.latest_journals(1)
+            from_time = 0
+            if len(journals) == 1:
+                latest_journal = journals[0]
+                if latest_journal.is_finish():
+                    yield None
+                    continue
+                from_time = os.path.getctime(latest_journal.get_input())
+                if os.path.getmtime(self.path()) <= from_time:
+                    yield (None, None)
+                    continue
+            
+            file_pathes = sorted(os.listdir(self.path()), key=lambda x: os.path.getctime(os.path.join(self.path(), x)))
+            for rel_path in file_pathes:
+                file_path = os.path.join(self.path(), rel_path)
+                timestamp = os.path.getctime(file_path)
+                if timestamp <= from_time:
+                    continue
+                ext = os.path.splitext(file_path)[1].lower()
+                if ext in ['.jpg', '.jpeg', '.png', '.gif', '.bmp']:
+                    logging.info(f"knowledge dir source found image file {file_path}")
+                    image = ImageObjectBuilder({}, {}, file_path).build(self.env.get_knowledge_store())
+                    await self.env.get_knowledge_store().insert_object(image)
+                    yield (image.calculate_id(), file_path)
+                if ext in ['.txt']:
+                    logging.info(f"knowledge dir source found text file {file_path}")
+                    text = await self.read_txt_file(file_path)
+                    document = DocumentObjectBuilder({}, {}, text).build(self.env.get_knowledge_store())
+                    await self.env.get_knowledge_store().insert_object(document)
+                    yield (document.calculate_id(), file_path)
+            yield (None, None)
+            
+
+def init(env: KnowledgePipelineEnvironment, params: dict) -> KnowledgeDirSource:
+    return KnowledgeDirSource(env, params)

src/component/knowledge_manager/parser/embedding.py

@@ -0,0 +1,102 @@
+# define a knowledge base class
+import json
+import string
+from aios_kernel import ComputeKernel, AIStorage
+from knowledge import *
+
+
+class EmbeddingParser:
+    def __init__(self, env: KnowledgePipelineEnvironment, config: dict):
+        self._default_text_model = "all-MiniLM-L6-v2"
+        self._default_image_model = "clip-ViT-B-32"
+
+        path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir())
+        if not os.path.exists(path):
+            os.makedirs(path)
+        config["path"] = path
+        
+        self.env = env
+        self.config = config
+
+    def get_path(self) -> str:
+        return self.config["path"]
+
+    def __get_vector_store(self, model_name: str) -> ChromaVectorStore:
+        return ChromaVectorStore(self.get_path(), model_name)
+
+    async def __embedding_document(self, document: DocumentObject):
+        for chunk_id in document.get_chunk_list():
+            chunk = self.env.get_knowledge_store().get_chunk_reader().get_chunk(chunk_id)
+            if chunk is None:
+                raise ValueError(f"text chunk not found: {chunk_id}")
+        
+            text = chunk.read().decode("utf-8")
+            vector = await ComputeKernel.get_instance().do_text_embedding(text, self._default_text_model)
+            if vector:
+                await self.__get_vector_store(self._default_text_model).insert(vector, chunk_id)
+
+    async def __embedding_image(self, image: ImageObject):
+        # desc = {}
+        # if not not image.get_meta():
+        #     desc["meta"] = image.get_meta()
+        # if not not image.get_exif():
+        #     desc["exif"] = image.get_exif()
+        # if not not image.get_tags():
+        #     desc["tags"] = image.get_tags()
+        # vector = await self.compute_kernel.do_text_embedding(json.dumps(desc), self._default_text_model)
+        vector = await ComputeKernel.get_instance().do_image_embedding(image.calculate_id(), self._default_image_model)
+        if vector:
+            await self.__get_vector_store(self._default_image_model).insert(vector, image.calculate_id())
+
+    async def __embedding_video(self, vedio: VideoObject):
+        desc = {}
+        if not not vedio.get_meta():
+            desc["meta"] = vedio.get_meta()
+        if not not vedio.get_info():
+            desc["info"] = vedio.get_info()
+        if not not vedio.get_tags():
+            desc["tags"] = vedio.get_tags()
+        vector = await ComputeKernel.get_instance().do_text_embedding(json.dumps(desc), self._default_text_model)
+        await self.__get_vector_store(self._default_text_model).insert(vector, vedio.calculate_id())
+
+    async def __embedding_rich_text(self, rich_text: RichTextObject):
+        for document_id in rich_text.get_documents().values():
+            document = DocumentObject.decode(self.env.get_knowledge_store().get_object_store().get_object(document_id))
+            await self.__embedding_document(document)
+        for image_id in rich_text.get_images().values():
+            image = ImageObject.decode(self.env.get_knowledge_store().get_object_store().get_object(image_id))
+            await self.__embedding_image(image)
+        for video_id in rich_text.get_videos().values():
+            video = VideoObject.decode(self.env.get_knowledge_store().get_object_store().get_object(video_id))
+            await self.__embedding_video(video)
+        for rich_text_id in rich_text.get_rich_texts().values():
+            rich_text = RichTextObject.decode(self.env.get_knowledge_store().get_object_store().get_object(rich_text_id))
+            await self.__embedding_rich_text(rich_text)
+
+    async def __embedding_email(self, email: EmailObject):
+        vector = await ComputeKernel.get_instance().do_text_embedding(json.dumps(email.get_desc()), self._default_text_model)
+        await self.__get_vector_store(self._default_text_model).insert(vector, email.calculate_id())
+        await self.__embedding_rich_text(email.get_rich_text())
+
+
+    async def __do_embedding(self, object: KnowledgeObject):
+        if object.get_object_type() == ObjectType.Document:
+            await self.__embedding_document(object)
+        if object.get_object_type() == ObjectType.Image:
+            await self.__embedding_image(object)
+        if object.get_object_type() == ObjectType.Video:
+            await self.__embedding_video(object)
+        if object.get_object_type() == ObjectType.RichText:
+            await self.__embedding_rich_text(object)
+        if object.get_object_type() == ObjectType.Email:
+            await self.__embedding_email(object)
+        else:
+            pass
+    
+    async def parse(self, object: ObjectID) -> str:
+        obj = self.env.get_knowledge_store().load_object(object)
+        await self.__do_embedding(obj)
+        return "insert into vector store"
+
+def init(env: KnowledgePipelineEnvironment, params: dict) -> EmbeddingParser:
+    return EmbeddingParser(env, params)

src/component/knowledge_manager/pipeline.py

@@ -0,0 +1,89 @@
+import os
+import runpy
+import toml
+import asyncio
+from knowledge import KnowledgePipelineEnvironment, KnowledgePipeline
+
+
+class KnowledgePipelineManager:
+    @classmethod
+    def initial(cls, root_dir: str):
+        cls._instance = KnowledgePipelineManager(root_dir)
+        return cls._instance
+
+    @classmethod
+    def get_instance(cls):
+        return cls._instance    
+    
+    def __init__(self, root_dir: str):
+        self.root_dir = root_dir
+        self.input_modules = {}
+        self.parser_modules = {}
+        self.pipelines = {
+            "names": {},
+            "running": []
+        }
+        from .input import local_dir
+        self.register_input("local_dir", local_dir.init)
+        from .parser import embedding
+        self.register_parser("embedding", embedding.init)
+
+    def register_input(self, name: str, init_method):
+        self.input_modules[name] = init_method
+        
+    def register_parser(self, name: str, parser_method):
+        self.parser_modules[name] = parser_method
+
+    def add_pipeline(self, config: dict, path: str):
+        name = config["name"]
+        if name in self.pipelines["names"]:
+            return
+        
+        input_module = config["input"]["module"]
+        _, ext = os.path.splitext(input_module)
+        if ext == ".py":
+            input_module = os.path.join(path, input_module)
+            input_init = runpy.run_path(input_module)["init"]
+        else:
+            input_init = self.input_modules.get(input_module)
+        input_params = config["input"]["params"]
+
+        parser_module = config["parser"]["module"]
+        _, ext = os.path.splitext(parser_module)
+        if ext == ".py":
+            parser_module = os.path.join(path, parser_module)
+            parser_init = runpy.run_path(parser_module)["init"]
+        else:
+            parser_init = self.parser_modules.get(parser_module)
+        parser_params = config["parser"]["params"]
+
+
+        data_path = os.path.join(self.root_dir, name)
+        env = KnowledgePipelineEnvironment(data_path)
+        pipeline = KnowledgePipeline(name, env, input_init, input_params, parser_init, parser_params)
+        self.pipelines["names"][name] = pipeline
+        self.pipelines["running"].append(pipeline)
+
+    def get_pipelines(self) -> [KnowledgePipeline]:
+        return self.pipelines["running"]
+    
+    def get_pipeline(self, name: str) -> KnowledgePipeline:
+        return self.pipelines["names"].get(name)
+
+    async def run(self):
+        while True:
+            for pipeline in self.pipelines["running"]:
+                await pipeline.run()
+            await asyncio.sleep(5)
+
+    def load_dir(self, root: str):
+        config_path = os.path.join(root, "pipelines.toml")
+        if not os.path.exists(config_path):
+            return 
+        with open(config_path, "r") as f:
+            config = toml.load(f)
+        for path in config["pipelines"]:
+            pipeline_path = os.path.join(root, path)
+            with open(os.path.join(pipeline_path, "pipeline.toml")) as f:
+                pipeline_config = toml.load(f)
+                self.add_pipeline(pipeline_config, pipeline_path)

src/knowledge/__init__.py

@@ -3,3 +3,4 @@ from .vector import *
 from .data import *
 from .store import KnowledgeStore
 from .core_object import *
+from .pipeline import *

src/knowledge/core_object/document_object.py

@@ -1,7 +1,6 @@
 from ..object import KnowledgeObject, ObjectRelationStore
 from ..data import ChunkList, ChunkListWriter
 from ..object import ObjectType
-from .. import KnowledgeStore
 
 # desc
 #   meta
@@ -49,13 +48,13 @@ class DocumentObjectBuilder:
         self.text = text
         return self
 
-    def build(self) -> DocumentObject:
+    def build(self, store) -> DocumentObject:
-        chunk_list = KnowledgeStore().get_chunk_list_writer().create_chunk_list_from_text(self.text)
+        chunk_list = store.get_chunk_list_writer().create_chunk_list_from_text(self.text)
         doc = DocumentObject(self.meta, self.tags, chunk_list)
         doc_id = doc.calculate_id()
         
         # Add relation to store
         for chunk_id in chunk_list.chunk_list:
-            KnowledgeStore().get_relation_store().add_relation(chunk_id, doc_id)
+            store.get_relation_store().add_relation(chunk_id, doc_id)
             
         return doc

src/knowledge/core_object/email_object.py

@@ -1,4 +1,3 @@
-from .. import KnowledgeStore
 from .rich_text_object import RichTextObject, RichTextObjectBuilder
 from ..object import ObjectID, ObjectType, KnowledgeObject
 from .document_object import DocumentObjectBuilder
@@ -68,11 +67,11 @@ class EmailObjectBuilder:
         self.folder = folder
         return self
 
-    def build(self) -> EmailObject:
+    def build(self, store) -> EmailObject:
         
         # Just get the object store and relation store from global KnowledgeStore
-        store = KnowledgeStore().get_object_store()
+        store = store.get_object_store()
-        relation = KnowledgeStore().get_relation_store()
+        relation = store.get_relation_store()
         
         # Read meta.json
         meta = {}

src/knowledge/core_object/image_object.py

@@ -1,7 +1,6 @@
 from ..object import KnowledgeObject
 from ..data import ChunkList, ChunkListWriter
 from ..object import ObjectType
-from .. import KnowledgeStore
 import os
 
 # desc
@@ -86,10 +85,10 @@ class ImageObjectBuilder:
         self.restore_file = restore_file
         return self
 
-    def build(self) -> ImageObject:
+    def build(self, store) -> ImageObject:
         
         file_size = os.path.getsize(self.image_file)
-        chunk_list = KnowledgeStore().get_chunk_list_writer().create_chunk_list_from_file(
+        chunk_list = store.get_chunk_list_writer().create_chunk_list_from_file(
             self.image_file, 1024 * 1024 * 4, self.restore_file
         )
         exif = get_exif_data(self.image_file)

src/knowledge/core_object/video_object.py

@@ -1,7 +1,6 @@
 from ..object import KnowledgeObject
 from ..data import ChunkList, ChunkListWriter
 from ..object import ObjectType
-from .. import KnowledgeStore
 
 # desc
 #   meta
@@ -76,8 +75,8 @@ class VideoObjectBuilder:
         self.restore_file = restore_file
         return self
 
-    def build(self) -> VideoObject:
+    def build(self, store) -> VideoObject:
-        chunk_list = KnowledgeStore().get_chunk_list_writer().create_chunk_list_from_file(
+        chunk_list = store.get_chunk_list_writer().create_chunk_list_from_file(
             self.video_file, 1024 * 1024 * 4, self.restore_file
         )
         info = get_video_info(self.video_file)

src/knowledge/pipeline.py

@@ -0,0 +1,123 @@
+import datetime
+import sqlite3
+import os
+from . import ObjectID, KnowledgeStore
+from enum import Enum
+
+class KnowledgePipelineJournal:
+    def __init__(self, time: datetime.datetime, object_id: str, input: str, parser: str):
+        self.time = time
+        self.object_id = None if object_id is None else ObjectID.from_base58(object_id)
+        self.input = input
+        self.parser = parser
+    
+    def is_finish(self) -> bool:
+        return self.object_id is None
+
+    def get_input(self) -> str:
+        return self.input
+    
+    def get_parser(self) -> str:
+        return self.parser
+    
+    def __str__(self) -> str:
+        if self.is_finish():
+            return f"{self.time}: finished)"
+        else:
+            return f"{self.time}: object:{self.object_id} input:{self.input}, parser:{self.parser})"
+
+# init sqlite3 client
+class KnowledgePipelineJournalClient:
+    def __init__(self, pipeline_path: str = None):
+        if not os.path.exists(pipeline_path):
+            os.makedirs(pipeline_path)
+        self.journal_path = os.path.join(pipeline_path, "journal.db")
+    
+        conn = sqlite3.connect(self.journal_path)
+        conn.execute(
+            '''CREATE TABLE IF NOT EXISTS journal (
+                id INTEGER PRIMARY KEY AUTOINCREMENT, 
+                time DATETIME DEFAULT CURRENT_TIMESTAMP,  
+                object_id TEXT, 
+                input TEXT, 
+                parser TEXT)'''
+        )
+        conn.commit()
+
+    def insert(self, object_id: ObjectID, input: str, parser: str, timestamp: datetime.datetime = None):
+        timestamp = datetime.datetime.now() if timestamp is None else timestamp
+        conn = sqlite3.connect(self.journal_path)
+        conn.execute(
+            "INSERT INTO journal (time, object_id, input, parser) VALUES (?, ?, ?, ?)",
+            (timestamp, str(object_id), input, parser),
+        )
+        conn.commit()
+          
+    def latest_journals(self, topn) -> [KnowledgePipelineJournal]:
+        conn = sqlite3.connect(self.journal_path)
+        cursor = conn.cursor()
+        cursor.execute("SELECT * FROM journal ORDER BY id DESC LIMIT ?", (topn,))
+        return [KnowledgePipelineJournal(time, object_id, input, parser) for (_, time, object_id, input, parser) in cursor.fetchall()]
+
+class KnowledgePipelineEnvironment:
+    def __init__(self, pipeline_path: str):
+        self.knowledge_store = KnowledgeStore()
+        if not os.path.exists(pipeline_path):
+            os.makedirs(pipeline_path)
+        self.pipeline_path = pipeline_path
+        self.journal = KnowledgePipelineJournalClient(pipeline_path)
+
+    def get_journal(self) -> KnowledgePipelineJournalClient:
+        return self.journal
+    
+    def get_knowledge_store(self) -> KnowledgeStore:
+        return self.knowledge_store
+
+class KnowledgePipelineState(Enum):
+    INIT = 0
+    RUNNING = 1
+    STOPPED = 2
+    FINISHED = 3
+
+class KnowledgePipeline:
+    def __init__(self, name: str, env: KnowledgePipelineEnvironment, input_init, input_params, parser_init, parser_params):
+        self.name = name
+        self.state = KnowledgePipelineState.INIT
+        self.input_init = input_init
+        self.input_params = input_params
+        self.parser_init = parser_init
+        self.parser_params = parser_params
+        self.env = env
+        self.input = None
+        self.parser = None
+
+    def get_name(self):
+        return self.name
+
+    def get_journal(self) -> KnowledgePipelineJournalClient:
+        return self.env.journal
+        
+    async def run(self):
+        if self.state == KnowledgePipelineState.INIT:
+            self.input = self.input_init(self.env, self.input_params)
+            self.parser = self.parser_init(self.env, self.parser_params)
+            self.state = KnowledgePipelineState.RUNNING
+        if self.state == KnowledgePipelineState.RUNNING:
+            async for input in self.input.next():
+                if input is None:
+                    self.state = KnowledgePipelineState.FINISHED
+                    self.env.journal.insert(None, "finished", "finished")
+                    return 
+                (object_id, input_journal) = input
+                if object_id is not None:
+                    parser_journal = await self.parser.parse(object_id)
+                    self.env.journal.insert(object_id, input_journal, parser_journal)
+                else:
+                    return
+        if self.state == KnowledgePipelineState.STOPPED:
+            return 
+        if self.state == KnowledgePipelineState.FINISHED:
+            return 
+
+
+    

src/knowledge/store.py

@@ -1,8 +1,9 @@
 import os
 
-from .object import ObjectStore, ObjectRelationStore
+from .object import ObjectStore, ObjectRelationStore, ObjectID, ObjectType, KnowledgeObject
+from .core_object import DocumentObject, ImageObject, VideoObject, RichTextObject, EmailObject
 from .data import ChunkStore, ChunkTracker, ChunkListWriter, ChunkReader
-from .vector import ChromaVectorStore, VectorBase
+import json
 import logging
 
 
@@ -17,7 +18,7 @@ class KnowledgeStore:
             cls._instance = super().__new__(cls)
             
             import aios_kernel
-            knowledge_dir = aios_kernel.storage.AIStorage().get_myai_dir() / "knowledge"
+            knowledge_dir = aios_kernel.storage.AIStorage().get_myai_dir() / "knowledge" / "objects"
 
             if not os.path.exists(knowledge_dir):
                 os.makedirs(knowledge_dir)
@@ -42,8 +43,6 @@ class KnowledgeStore:
         self.chunk_tracker = ChunkTracker(chunk_store_dir)
         self.chunk_list_writer = ChunkListWriter(self.chunk_store, self.chunk_tracker)
         self.chunk_reader = ChunkReader(self.chunk_store, self.chunk_tracker)
-        self.vector_store = {}
-
 
     
     def get_relation_store(self) -> ObjectRelationStore:
@@ -64,7 +63,43 @@ class KnowledgeStore:
     def get_chunk_reader(self) -> ChunkReader:
         return self.chunk_reader
     
-    def get_vector_store(self, model_name: str) -> VectorBase:
+    async def insert_object(self, object: KnowledgeObject):
-        if model_name not in self.vector_store:
+        self.object_store.put_object(object.calculate_id(), object.encode())
-            self.vector_store[model_name] = ChromaVectorStore(self.root, model_name)
+    
-        return self.vector_store[model_name]
+    def load_object(self, object_id: ObjectID) -> KnowledgeObject:
+        if object_id.get_object_type() == ObjectType.Document:
+            return DocumentObject.decode(self.object_store.get_object(object_id))
+        if object_id.get_object_type() == ObjectType.Image:
+            return ImageObject.decode(self.object_store.get_object(object_id))
+        if object_id.get_object_type() == ObjectType.Video:
+            return VideoObject.decode(self.object_store.get_object(object_id))
+        if object_id.get_object_type() == ObjectType.RichText:
+            return RichTextObject.decode(self.object_store.get_object(object_id))
+        if object_id.get_object_type() == ObjectType.Email:
+            return EmailObject.decode(self.object_store.get_object(object_id))
+        else:
+            pass
+    
+    def parse_object_in_message(self, message: str) -> KnowledgeObject:
+        # get message's first line 
+        logging.info(f"tg parse resp message: {message}")
+        lines = message.split("\n")
+        if len(lines) > 0:
+            message = lines[0]
+            try:
+                desc = json.loads(message)
+                if isinstance(desc, dict):
+                    object_id = desc["id"]
+                else:
+                    object_id = desc[0]["id"]
+            except Exception as e:
+                return None
+            
+            if object_id is not None:
+                return self.load_object(ObjectID.from_base58(object_id))
+            
+            
+    def bytes_from_object(self, object: KnowledgeObject) -> bytes:
+        if object.get_object_type() == ObjectType.Image:
+            image_object = object
+            return self.get_chunk_reader().read_chunk_list_to_single_bytes(image_object.get_chunk_list())

src/service/aios_shell/aios_shell.py

@@ -27,11 +27,13 @@ sys.path.append(directory + '/../../')
 
 import proxy
 from aios_kernel import *
+from knowledge import *
 
 
 sys.path.append(directory + '/../../component/')
 from agent_manager import AgentManager
 from workflow_manager import WorkflowManager
+from knowledge_manager import KnowledgePipelineManager
 
 
 logger = logging.getLogger(__name__)
@@ -112,9 +114,6 @@ class AIOS_Shell:
 
             cm.add_family_member(self.username,owenr)
 
-        knowledge_env = KnowledgeEnvironment("knowledge")
-        Environment.set_env_by_id("knowledge",knowledge_env)
-
         cal_env = CalenderEnvironment("calender")
         await cal_env.start()
         Environment.set_env_by_id("calender",cal_env)
@@ -186,7 +185,12 @@ class AIOS_Shell:
 
         AIBus().get_default_bus().register_unhandle_message_handler(self._handle_no_target_msg)
         AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg)
-        KnowledgePipline.get_instance().initial()
+        
+        
+        pipelines = KnowledgePipelineManager.initial(os.path.join(AIStorage().get_instance().get_myai_dir(), "knowledge/pipelines"))
+        pipelines.load_dir(os.path.join(AIStorage().get_instance().get_system_app_dir(), "knowledge_pipelines"))
+        pipelines.load_dir(os.path.join(AIStorage().get_instance().get_myai_dir(), "knowledge_pipelines"))
+        asyncio.create_task(pipelines.run())
 
         TelegramTunnel.register_to_loader()
         EmailTunnel.register_to_loader()
@@ -331,45 +335,20 @@ class AIOS_Shell:
     
     async def handle_knowledge_commands(self, args):
         show_text = FormattedText([("class:title", "sub command not support!\n"
-                              "/knowledge add email | dir\n"
+                              "/knowledge pipelines\n" 
-                              "/knowledge journal [$topn]\n"
+                              "/knowledge journal $pipeline [$topn]\n"
                               "/knowledge query $object_id\n")])
         if len(args) < 1:
             return show_text
         sub_cmd = args[0]
-        if sub_cmd == "add":
+        if sub_cmd == "pipelines":
-            if len(args) < 2:
+            pipelines = KnowledgePipelineManager.get_instance().get_pipelines()
-                return show_text
+            print_formatted_text("\r\n".join(pipeline.get_name() for pipeline in pipelines))
-            if args[1] == "email":
-                config = dict()
-                for key, item in KnowledgeEmailSource.user_config_items():
-                    user_input = await try_get_input(f"{key} : {item}")
-                    if user_input is None:
-                        return show_text
-                    config[key] = user_input
-                error = KnowledgePipline.get_instance().add_email_source(KnowledgeEmailSource(config))
-                if error is not None:
-                    return FormattedText([("class:title", f"/knowledge add email failed {error}\n")])
-                else:
-                    KnowledgePipline.get_instance().save_cosnfig()
-            if args[1] == "dir":
-                config = dict()
-                for key, item in KnowledgeDirSource.user_config_items():
-                    user_input = await try_get_input(f"{key} : {item}")
-                    if user_input is None:
-                        return show_text
-                    config[key] = user_input
-                error = KnowledgePipline.get_instance().add_dir_source(KnowledgeDirSource(config))
-                if error is not None:
-                    return FormattedText([("class:title", f"/knowledge add dir failed {error}\n")])
-                else:
-                    KnowledgePipline.get_instance().save_config()
-            else:
-                return show_text
         if sub_cmd == "journal":
-            topn = 10 if len(args) == 1 else int(args[1])
+            name = args[1]
-            journals = [str(journal) for journal in KnowledgePipline.get_instance().get_latest_journals(topn)]
+            topn = 10 if len(args) == 2 else int(args[2])
-            print_formatted_text("\r\n".join(journals))
+            journals = [str(journal) for journal in KnowledgePipelineManager.get_instance().get_pipeline(name).get_journal().latest_journals(topn)]
+            print_formatted_text("\r\n".join(str(journal) for journal in journals))
         if sub_cmd == "query":
             if len(args) < 2:
                 return show_text
@@ -378,8 +357,8 @@ class AIOS_Shell:
             if object_id.get_object_type() == ObjectType.Image:
                 from PIL import Image
                 import io
-                image = KnowledgeBase().load_object(object_id)
+                image = KnowledgeStore().load_object(object_id)
-                image_data = KnowledgeBase().bytes_from_object(image)
+                image_data = KnowledgeStore().bytes_from_object(image)
                 image = Image.open(io.BytesIO(image_data))
                 image.show()
 
@@ -668,9 +647,8 @@ def print_welcome_screen():
 \033[1;94m\tGive your Agent a Telegram account :\033[0m /connect $agent_name
 \033[1;94m\tAdd personal files to the AI Knowledge Base. \033[0m
 \t\t1) Copy your file to ~/myai/data 
-\t\t2) /knowlege add dir
 \033[1;94m\tSearch your knowledge base :\033[0m /open Mia
-\033[1;94m\tCheck the progress of AI reading personal data :\033[0m /knowledge journal
+\033[1;94m\tCheck the progress of AI reading personal data :\033[0m /knowledge $pipeline journal
 \033[1;94m\tQuery object with ID in knowledge base :\033[0m /knowledge query $object_id
 \033[1;94m\tOpen AI Bash (For Developer Only):\033[0m /open ai_bash
 \033[1;94m\tEnable AIGC Feature :\033[0m /enable aigc
@@ -749,8 +727,8 @@ async def main():
                                '/history $num $offset',
                                '/connect $target',
                                '/contact $name',
-                               '/knowledge add email | dir',
+                               '/knowledge pipelines',
-                               '/knowledge journal [$topn]', 
+                               '/knowledge journal $pipeline [$topn]', 
                                '/knowledge query $object_id',
                                '/set_config $key',
                                '/enable $feature',

src/service/email_spider/converter.py

@@ -1,24 +0,0 @@
-from aios_kernel.knowledge import KnowledgeBase, EmailObject
-
-# define a email converter class
-
-class EmailConverter:
-    # define init method
-    def __init__(self, local_dir, knowledge_base: KnowledgeBase) -> None:
-        pass
-
-    async def run(self):
-        # convert the email to knowledge object
-        for email_dir in self._next():
-            # convert the email to knowledge object
-            knowledge_object = self._convert(email_dir)
-            # insert the knowledge object to knowledge base
-            await self.knowledge_base.insert(knowledge_object)
-
-    def _next(self) -> str:
-        pass
-
-    def _convert(self, email_dir) -> EmailObject:
-        pass
-
-    

src/service/email_spider/main.py

@@ -1,12 +0,0 @@
-import asyncio
-from .spider import EmailSpider, EmailConverter
-
-
-if __name__ == "__main__":
-    spider = EmailSpider("smtp.163.com","user","pwd","./email")
-    asyncio.run(spider.run())
-
-    converter = EmailConverter("./email",KnowledgeBase())
-    asyncio.run(converter.run())
-
-    

src/service/email_spider/spider.py

@@ -1,17 +0,0 @@
-# define a email spider class
-
-class EmailSpider:
-    def __init__(self, address, account, pwd, local_dir) -> None:
-        pass
-
-    async def run(self): 
-        # spide the email from the email server
-        for email_link in self._next():
-            # save the email to local directory
-            self._save(email_link)
-
-    def _next(self):
-        pass
-
-    def _save(self, email_link) -> str:
-        pass