diff --git a/doc/knowledge_pipeline.md b/doc/knowledge_pipeline.md new file mode 100644 index 0000000..fedb5aa --- /dev/null +++ b/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方法的提示词。 + + + + + diff --git a/rootfs/agents/FindPhoto/agent.toml b/rootfs/agents/FindPhoto/agent.toml new file mode 100644 index 0000000..ddb3f3f --- /dev/null +++ b/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字符串。 +""" + diff --git a/rootfs/agents/FindPhoto/enviroment.py b/rootfs/agents/FindPhoto/enviroment.py new file mode 100644 index 0000000..e69de29 diff --git a/rootfs/agents/Mia/agent.toml b/rootfs/agents/Mia/agent.toml index 770c07e..9dce4b0 100644 --- a/rootfs/agents/Mia/agent.toml +++ b/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" diff --git a/rootfs/email/config.toml b/rootfs/email/config.toml deleted file mode 100644 index 1bf4bca..0000000 --- a/rootfs/email/config.toml +++ /dev/null @@ -1,7 +0,0 @@ - - -EMAIL_IMAP_SERVER = "imap.gmail.com" -EMAIL_ADDRESS = '<>' -EMAIL_PASSWORD = '<>' -EMAIL_IMAP_PORT = 993 -LOCAL_DIR = 'rootfs/data' \ No newline at end of file diff --git a/rootfs/knowledge_pipelines/Mia/input.py b/rootfs/knowledge_pipelines/Mia/input.py new file mode 100644 index 0000000..e9a54f4 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/rootfs/knowledge_pipelines/Mia/parser.py b/rootfs/knowledge_pipelines/Mia/parser.py new file mode 100644 index 0000000..cb8b5f8 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/rootfs/knowledge_pipelines/Mia/pipeline.toml b/rootfs/knowledge_pipelines/Mia/pipeline.toml new file mode 100644 index 0000000..9b9e062 --- /dev/null +++ b/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" + diff --git a/rootfs/knowledge_pipelines/Mia/query.py b/rootfs/knowledge_pipelines/Mia/query.py new file mode 100644 index 0000000..cc4b557 --- /dev/null +++ b/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") \ No newline at end of file diff --git a/rootfs/knowledge_pipelines/pipelines.toml b/rootfs/knowledge_pipelines/pipelines.toml new file mode 100644 index 0000000..3c78084 --- /dev/null +++ b/rootfs/knowledge_pipelines/pipelines.toml @@ -0,0 +1,3 @@ +pipelines = [ + "Mia" +] \ No newline at end of file diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index e19ba35..f632e03 100644 --- a/src/aios_kernel/__init__.py +++ b/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" diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 5d9c756..34d8967 100644 --- a/src/aios_kernel/agent.py +++ b/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: - #kb_env = KnowledgeBaseFileSystemEnvironment() + async def _llm_read_article(self,item:KnowledgeObject) -> ComputeTaskResult: 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(): diff --git a/src/aios_kernel/knowledge_base.py b/src/aios_kernel/knowledge_base.py deleted file mode 100644 index ac94459..0000000 --- a/src/aios_kernel/knowledge_base.py +++ /dev/null @@ -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]) \ No newline at end of file diff --git a/src/aios_kernel/knowledge_pipeline.py b/src/aios_kernel/knowledge_pipeline.py deleted file mode 100644 index 759793a..0000000 --- a/src/aios_kernel/knowledge_pipeline.py +++ /dev/null @@ -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") \ No newline at end of file diff --git a/src/aios_kernel/tg_tunnel.py b/src/aios_kernel/tg_tunnel.py index dcc7abd..923fb14 100644 --- a/src/aios_kernel/tg_tunnel.py +++ b/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_base import KnowledgeBase +from knowledge import ObjectType, KnowledgeStore 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: diff --git a/src/component/agent_manager/agent_manager.py b/src/component/agent_manager/agent_manager.py index 06ed62c..ef18253 100644 --- a/src/component/agent_manager/agent_manager.py +++ b/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 diff --git a/src/component/knowledge_manager/__init__.py b/src/component/knowledge_manager/__init__.py new file mode 100644 index 0000000..8f6fd1f --- /dev/null +++ b/src/component/knowledge_manager/__init__.py @@ -0,0 +1 @@ +from .pipeline import KnowledgePipelineManager \ No newline at end of file diff --git a/src/service/spider/email_spider.py b/src/component/knowledge_manager/input/email.py similarity index 50% rename from src/service/spider/email_spider.py rename to src/component/knowledge_manager/input/email.py index cd44097..a8eb353 100644 --- a/src/service/spider/email_spider.py +++ b/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: -├── 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 +class KnowledgeEmailSource: + def __init__(self, config:dict): + self.config = config + self.config["type"] = "email" + + def id(self): + return self.config["address"] -""" - -import imaplib -import os -import toml -import logging -import mailparser -import hashlib -import json -import base64 -from bs4 import BeautifulSoup -import requests - -class EmailSpider: - def __init__(self): - # logger config - self.logger = logging.getLogger('email spider') - 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) + @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) - self.config = toml.load('./rootfs/email/config.toml') - if os.path.exists('./rootfs/email/config.local.toml'): - self.config = toml.load('./rootfs/email/config.local.toml') - + 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: - 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'), - port=self.config.get('EMAIL_IMAP_PORT') + host=self.config.get('imap_server'), + 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): - _, mailbox_list = self.client.list() - for mailbox in mailbox_list: - print(mailbox.decode()) - - def read_emails(self, folder: str = 'INBOX', imap_keyword: str = "UNSEEN"): + 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() - self.logger.info(f"got {len(email_list)} emails") - email_list.reverse() + logging.info(f"got {len(email_list)} emails") + journal_client = KnowledgeJournalClient() for uid in email_list: - if self.check_email_saved(uid): - self.logger.info(f"email uid {uid} already saved") - else: - self.read_and_save_email(uid) - self.logger.info(f"email uid {uid} saved") + _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): + + 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 False + return dir + 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) \ No newline at end of file + self.save_body_images(mail.body, f"{email_dir}/body_image") \ No newline at end of file diff --git a/src/component/knowledge_manager/input/local_dir.py b/src/component/knowledge_manager/input/local_dir.py new file mode 100644 index 0000000..e9a54f4 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/src/component/knowledge_manager/parser/embedding.py b/src/component/knowledge_manager/parser/embedding.py new file mode 100644 index 0000000..cb8b5f8 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/src/component/knowledge_manager/pipeline.py b/src/component/knowledge_manager/pipeline.py new file mode 100644 index 0000000..55a5105 --- /dev/null +++ b/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) diff --git a/src/knowledge/__init__.py b/src/knowledge/__init__.py index 680c632..361a67d 100644 --- a/src/knowledge/__init__.py +++ b/src/knowledge/__init__.py @@ -2,4 +2,5 @@ from .object import * from .vector import * from .data import * from .store import KnowledgeStore -from .core_object import * \ No newline at end of file +from .core_object import * +from .pipeline import * \ No newline at end of file diff --git a/src/knowledge/core_object/document_object.py b/src/knowledge/core_object/document_object.py index b69f59f..43e8b9e 100644 --- a/src/knowledge/core_object/document_object.py +++ b/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: - chunk_list = KnowledgeStore().get_chunk_list_writer().create_chunk_list_from_text(self.text) + def build(self, store) -> DocumentObject: + 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 diff --git a/src/knowledge/core_object/email_object.py b/src/knowledge/core_object/email_object.py index 108c9cc..64d78b3 100644 --- a/src/knowledge/core_object/email_object.py +++ b/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() - relation = KnowledgeStore().get_relation_store() + store = store.get_object_store() + relation = store.get_relation_store() # Read meta.json meta = {} diff --git a/src/knowledge/core_object/image_object.py b/src/knowledge/core_object/image_object.py index 340bcc2..1033bf7 100644 --- a/src/knowledge/core_object/image_object.py +++ b/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) diff --git a/src/knowledge/core_object/video_object.py b/src/knowledge/core_object/video_object.py index 56e9829..1ac9338 100644 --- a/src/knowledge/core_object/video_object.py +++ b/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: - chunk_list = KnowledgeStore().get_chunk_list_writer().create_chunk_list_from_file( + def build(self, store) -> VideoObject: + 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) diff --git a/src/knowledge/pipeline.py b/src/knowledge/pipeline.py new file mode 100644 index 0000000..af3b665 --- /dev/null +++ b/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 + + + \ No newline at end of file diff --git a/src/knowledge/store.py b/src/knowledge/store.py index abcf496..a2ebd77 100644 --- a/src/knowledge/store.py +++ b/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,10 +43,8 @@ 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: return self.relation_store @@ -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: - if model_name not in self.vector_store: - self.vector_store[model_name] = ChromaVectorStore(self.root, model_name) - return self.vector_store[model_name] + async def insert_object(self, object: KnowledgeObject): + self.object_store.put_object(object.calculate_id(), object.encode()) + + 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()) \ No newline at end of file diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 43441b4..b8ce7a6 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/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() @@ -330,46 +334,21 @@ class AIOS_Shell: cm.add_contact(contact_name,contact) 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" + show_text = FormattedText([("class:title", "sub command not support!\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 @@ -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_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() @@ -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 journal [$topn]', + '/knowledge pipelines', + '/knowledge journal $pipeline [$topn]', '/knowledge query $object_id', '/set_config $key', '/enable $feature', diff --git a/src/service/email_spider/converter.py b/src/service/email_spider/converter.py deleted file mode 100644 index 193b220..0000000 --- a/src/service/email_spider/converter.py +++ /dev/null @@ -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 - - \ No newline at end of file diff --git a/src/service/email_spider/main.py b/src/service/email_spider/main.py deleted file mode 100644 index cb75abd..0000000 --- a/src/service/email_spider/main.py +++ /dev/null @@ -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()) - - diff --git a/src/service/email_spider/spider.py b/src/service/email_spider/spider.py deleted file mode 100644 index 0b5a212..0000000 --- a/src/service/email_spider/spider.py +++ /dev/null @@ -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 \ No newline at end of file