change /knowledge commands in shell

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/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

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

rootfs/knowledge_pipelines/Mia/query.py

@@ -1,7 +1,15 @@
+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", 
@@ -12,22 +20,26 @@ class KnowledgeEnvironment(Environment):
                                             "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 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)
+            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 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)
+            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 = self.store.get_relation_store().get_related_root_objects(object_id)
+            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)}")
@@ -41,7 +53,7 @@ class KnowledgeEnvironment(Environment):
             # 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)
+                email = KnowledgeStore().load_object(root_object_id)
                 desc = email.get_desc()
                 desc["type"] = "email"
                 desc["contents"] = []
@@ -53,7 +65,7 @@ class KnowledgeEnvironment(Environment):
             
             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")})
+                    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()
@@ -61,7 +73,7 @@ class KnowledgeEnvironment(Environment):
                     desc["type"] = "image"
                     upper_list.append(desc)
                 if object_id.get_object_type() == ObjectType.Video:
-                    video = self.load_object(object_id)
+                    video = KnowledgeStore().load_object(object_id)
                     desc = video.get_desc()
                     desc["type"] = "video"
                     upper_list.append(desc)
@@ -74,9 +86,12 @@ class KnowledgeEnvironment(Environment):
     
     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)
+        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 + KnowledgeBase().tokens_from_objects(object_ids[index:index+1])
+            return content + self.tokens_from_objects(object_ids[index:index+1])
+
+def init() -> KnowledgeEnvironment:
+    return KnowledgeEnvironment("embedding")

rootfs/knowledge_pipelines/pipelines.toml

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

src/aios_kernel/__init__.py

@@ -23,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

@@ -122,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"]

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/input/local_dir.py

@@ -30,16 +30,19 @@ class KnowledgeDirSource:
             return await f.read()
         
     async def next(self):
-        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
-            from_time = os.path.getctime(latest_journal.get_input())
-            if os.path.getmtime(self.path()) <= from_time:
-                yield (None, None)
         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)

src/component/knowledge_manager/pipeline.py

@@ -4,7 +4,17 @@ 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 = {}
@@ -32,7 +42,7 @@ class KnowledgePipelineManager:
         input_module = config["input"]["module"]
         _, ext = os.path.splitext(input_module)
         if ext == ".py":
-            input_module = os.path.abspath(path, input_module)
+            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)
@@ -41,7 +51,7 @@ class KnowledgePipelineManager:
         parser_module = config["parser"]["module"]
         _, ext = os.path.splitext(parser_module)
         if ext == ".py":
-            parser_module = os.path.abspath(path, parser_module)
+            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)
@@ -54,6 +64,12 @@ class KnowledgePipelineManager:
         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"]:

src/knowledge/pipeline.py

@@ -20,6 +20,12 @@ class KnowledgePipelineJournal:
     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):
@@ -85,6 +91,12 @@ class KnowledgePipeline:
         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)
@@ -100,6 +112,8 @@ class KnowledgePipeline:
                 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:

src/service/aios_shell/aios_shell.py

@@ -27,6 +27,7 @@ sys.path.append(directory + '/../../')
 
 import proxy
 from aios_kernel import *
+from knowledge import *
 
 
 sys.path.append(directory + '/../../component/')
@@ -186,7 +187,7 @@ class AIOS_Shell:
         AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg)
         
         
-        pipelines = KnowledgePipelineManager(os.path.join(AIStorage().get_instance().get_myai_dir(), "knowledge/pipelines"))
+        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())
@@ -334,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 journal [$topn]\n"
+                              "/knowledge pipelines\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 len(args) < 2:
-                return show_text
-            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 == "pipelines":
+            pipelines = KnowledgePipelineManager.get_instance().get_pipelines()
+            print_formatted_text("\r\n".join(pipeline.get_name() for pipeline in pipelines))
         if sub_cmd == "journal":
-            topn = 10 if len(args) == 1 else int(args[1])
-            journals = [str(journal) for journal in KnowledgePipline.get_instance().get_latest_journals(topn)]
-            print_formatted_text("\r\n".join(journals))
+            name = args[1]
+            topn = 10 if len(args) == 2 else int(args[2])
+            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
@@ -381,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_data = KnowledgeBase().bytes_from_object(image)
+                image = KnowledgeStore().load_object(object_id)
+                image_data = KnowledgeStore().bytes_from_object(image)
                 image = Image.open(io.BytesIO(image_data))
                 image.show()
 
@@ -671,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
@@ -752,8 +727,8 @@ async def main():
                                '/history $num $offset',
                                '/connect $target',
                                '/contact $name',
-                               '/knowledge add email | dir',
-                               '/knowledge journal [$topn]', 
+                               '/knowledge pipelines',
+                               '/knowledge journal $pipeline [$topn]', 
                                '/knowledge query $object_id',
                                '/set_config $key',
                                '/enable $feature',