From ce033c2381ac46b5720714a76d3dbd04cfacc48b Mon Sep 17 00:00:00 2001 From: tsukasa Date: Thu, 19 Oct 2023 10:05:08 +0800 Subject: [PATCH 1/5] add knowledge pipeline manager --- rootfs/agents/FindPhoto/agent.toml | 21 + rootfs/agents/FindPhoto/enviroment.py | 0 rootfs/email/config.toml | 7 - .../local_photos_embedding/pipeline.toml | 6 + rootfs/knowledge_pipelines/pipelines.toml | 2 + src/aios_kernel/knowledge_base.py | 296 ------------- src/aios_kernel/knowledge_pipeline.py | 411 ------------------ src/knowledge/__init__.py | 3 +- src/knowledge/input/__init__.py | 1 + src/knowledge/input/email.py | 158 +++++++ src/knowledge/input/local_dir.py | 61 +++ src/knowledge/parser/__init__.py | 0 src/knowledge/parser/embedding.py | 93 ++++ src/knowledge/pipeline.py | 165 +++++++ src/knowledge/query/embedding.py | 82 ++++ src/knowledge/store.py | 53 ++- src/service/aios_shell/aios_shell.py | 10 +- 17 files changed, 641 insertions(+), 728 deletions(-) create mode 100644 rootfs/agents/FindPhoto/agent.toml create mode 100644 rootfs/agents/FindPhoto/enviroment.py delete mode 100644 rootfs/email/config.toml create mode 100644 rootfs/knowledge_pipelines/local_photos_embedding/pipeline.toml create mode 100644 rootfs/knowledge_pipelines/pipelines.toml delete mode 100644 src/aios_kernel/knowledge_base.py delete mode 100644 src/aios_kernel/knowledge_pipeline.py create mode 100644 src/knowledge/input/__init__.py create mode 100644 src/knowledge/input/email.py create mode 100644 src/knowledge/input/local_dir.py create mode 100644 src/knowledge/parser/__init__.py create mode 100644 src/knowledge/parser/embedding.py create mode 100644 src/knowledge/pipeline.py create mode 100644 src/knowledge/query/embedding.py 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/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/local_photos_embedding/pipeline.toml b/rootfs/knowledge_pipelines/local_photos_embedding/pipeline.toml new file mode 100644 index 0000000..e26e7a8 --- /dev/null +++ b/rootfs/knowledge_pipelines/local_photos_embedding/pipeline.toml @@ -0,0 +1,6 @@ +name = "LocalPhotoEmbedding" +input.module = "local_dir" +input.params.path = "~/myai/photos" +parser.module = "embedding" +parser.params.path = "~/myai/knowledge/indices/photo_embedding" + diff --git a/rootfs/knowledge_pipelines/pipelines.toml b/rootfs/knowledge_pipelines/pipelines.toml new file mode 100644 index 0000000..99d5b59 --- /dev/null +++ b/rootfs/knowledge_pipelines/pipelines.toml @@ -0,0 +1,2 @@ +[[pipelines]] +"local_photos_embedding" \ No newline at end of file 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/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/input/__init__.py b/src/knowledge/input/__init__.py new file mode 100644 index 0000000..0e9e2a5 --- /dev/null +++ b/src/knowledge/input/__init__.py @@ -0,0 +1 @@ +import local_dir \ No newline at end of file diff --git a/src/knowledge/input/email.py b/src/knowledge/input/email.py new file mode 100644 index 0000000..a8eb353 --- /dev/null +++ b/src/knowledge/input/email.py @@ -0,0 +1,158 @@ + +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") \ No newline at end of file diff --git a/src/knowledge/input/local_dir.py b/src/knowledge/input/local_dir.py new file mode 100644 index 0000000..2bbafe1 --- /dev/null +++ b/src/knowledge/input/local_dir.py @@ -0,0 +1,61 @@ +import os +import aiofiles +import chardet +from knowledge import ImageObjectBuilder, DocumentObjectBuilder, KnowledgeBase + +class KnowledgeDirSource: + def __init__(self, config): + self.config = config + config["path"] = os.path.abspath(config["path"]) + + # @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): + # 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 + while True: + 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) + yield image.calculate_id() + # 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) + yield document.calculate_id() + # journal_client.insert(KnowledgeJournal("dir", self.id(), rel_path, str(document.calculate_id()), timestamp)) + yield 0 + + +def init(params: dict) -> KnowledgeDirSource: + return KnowledgeDirSource(params) \ No newline at end of file diff --git a/src/knowledge/parser/__init__.py b/src/knowledge/parser/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/knowledge/parser/embedding.py b/src/knowledge/parser/embedding.py new file mode 100644 index 0000000..e512eea --- /dev/null +++ b/src/knowledge/parser/embedding.py @@ -0,0 +1,93 @@ +# define a knowledge base class +import json +from aios_kernel import ComputeKernel, AIStorage +from knowledge import * + + +class EmbeddingParser: + def __init__(self, params: dict): + self.store = KnowledgeStore() + self.compute_kernel = ComputeKernel.get_instance() + self.knowledge_base = KnowledgeBase() + self._default_text_model = "all-MiniLM-L6-v2" + self._default_image_model = "clip-ViT-B-32" + self.vector_store = ChromaVectorStore(AIStorage().get_myai_dir() / "knowledge", self._default_text_model) + + def __get_vector_store(self, model_name: str) -> ChromaVectorStore: + return ChromaVectorStore(AIStorage().get_myai_dir() / "knowledge", self._default_text_model) + + async def __embedding_document(self, document: DocumentObject): + for chunk_id in document.get_chunk_list(): + chunk = self.knowledge_base.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.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.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.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.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): + obj = self.knowledge_base.load_object(object) + await self.__do_embedding(obj) + +def init(params: dict) -> EmbeddingParser: + return EmbeddingParser(params) \ No newline at end of file diff --git a/src/knowledge/pipeline.py b/src/knowledge/pipeline.py new file mode 100644 index 0000000..9c7e349 --- /dev/null +++ b/src/knowledge/pipeline.py @@ -0,0 +1,165 @@ + +# class KnowledgePipelineTemplate +import runpy +import toml +import datetime +import sqlite3 +import os +from . import ObjectID, KnowledgeStore +import asyncio + +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: + self.object_id is None + + + +# 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_base = KnowledgeStore() + self.pipeline_path = pipeline_path + self.journal = KnowledgePipelineJournalClient(pipeline_path) + +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 + + 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: + for input in await 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 None: + parser_journal = await self.parser.parse(object_id) + self.env.journal.insert(object_id, input_journal, parser_journal) + if self.state == KnowledgePipelineState.STOPPED: + return + if self.state == KnowledgePipelineState.FINISHED: + return + +class KnowledgePipelineManager: + 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) + + 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 = self.input_modules[config["input"]["module"]] + _, ext = os.path.splitext(input_module) + if ext == ".py": + input_module = os.path.abspath(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 = self.parser_modules[config["parser"]["module"]] + _, ext = os.path.splitext(parser_module) + if ext == ".py": + parser_module = os.path.abspath(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 = 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) + + 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") + 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) + + + \ No newline at end of file diff --git a/src/knowledge/query/embedding.py b/src/knowledge/query/embedding.py new file mode 100644 index 0000000..f29bb43 --- /dev/null +++ b/src/knowledge/query/embedding.py @@ -0,0 +1,82 @@ + +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_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 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 + + 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/knowledge/store.py b/src/knowledge/store.py index abcf496..ebfda27 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, DocumentObject, ImageObject, VideoObject, RichTextObject, EmailObject +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.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 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..191e0d1 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -112,9 +112,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 +183,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(AIStorage().get_instance().get_myai_dir() / "knowledge" / "pipelines") + pipelines.load_dir(AIStorage().get_instance().get_system_app_dir() / "knowledge_pipelines") + pipelines.load_dir(AIStorage().get_instance().get_myai_dir() / "knowledge_pipelines") + asyncio.create_task(pipelines.run()) TelegramTunnel.register_to_loader() EmailTunnel.register_to_loader() From f80ec04272248ca25f04d8951bac76d2ca349897 Mon Sep 17 00:00:00 2001 From: tsukasa Date: Thu, 19 Oct 2023 10:47:45 +0800 Subject: [PATCH 2/5] add knowledge pipeline manager --- src/aios_kernel/__init__.py | 2 -- src/aios_kernel/agent.py | 4 +--- src/aios_kernel/tg_tunnel.py | 8 +++----- src/knowledge/core_object/document_object.py | 7 +++---- src/knowledge/core_object/email_object.py | 7 +++---- src/knowledge/core_object/image_object.py | 5 ++--- src/knowledge/core_object/video_object.py | 5 ++--- src/knowledge/pipeline.py | 1 + src/knowledge/store.py | 4 ++-- 9 files changed, 17 insertions(+), 26 deletions(-) diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index e19ba35..fc11ad0 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 diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 5d9c756..bacafe5 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 @@ -550,8 +549,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/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/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 index 9c7e349..e051925 100644 --- a/src/knowledge/pipeline.py +++ b/src/knowledge/pipeline.py @@ -7,6 +7,7 @@ import sqlite3 import os from . import ObjectID, KnowledgeStore import asyncio +from enum import Enum class KnowledgePipelineJournal: def __init__(self, time: datetime.datetime, object_id: str, input: str, parser: str): diff --git a/src/knowledge/store.py b/src/knowledge/store.py index ebfda27..803c3b6 100644 --- a/src/knowledge/store.py +++ b/src/knowledge/store.py @@ -1,7 +1,7 @@ import os -from .object import ObjectStore, ObjectRelationStore, ObjectID, ObjectType, KnowledgeObject, DocumentObject, ImageObject, VideoObject, RichTextObject, EmailObject -from core_object import DocumentObject, ImageObject, VideoObject, RichTextObject, EmailObject +from .object import ObjectStore, ObjectRelationStore, ObjectID, ObjectType, KnowledgeObject +from .core_object import DocumentObject, ImageObject, VideoObject, RichTextObject, EmailObject from .data import ChunkStore, ChunkTracker, ChunkListWriter, ChunkReader import json import logging From 030077e4d3e44ab9d34f3381cad29c9962d4b70a Mon Sep 17 00:00:00 2001 From: tsukasa Date: Thu, 19 Oct 2023 17:09:27 +0800 Subject: [PATCH 3/5] knowledge pipeline manager init --- .../local_embedding/pipeline.toml | 6 + .../local_photos_embedding/pipeline.toml | 6 - rootfs/knowledge_pipelines/pipelines.toml | 5 +- src/component/knowledge_manager/__init__.py | 1 + .../knowledge_manager}/input/email.py | 0 .../knowledge_manager/input/local_dir.py | 65 +++++++ .../knowledge_manager}/parser/embedding.py | 55 +++--- src/component/knowledge_manager/pipeline.py | 73 ++++++++ .../knowledge_manager}/query/embedding.py | 0 src/knowledge/input/__init__.py | 1 - src/knowledge/input/local_dir.py | 61 ------- src/knowledge/parser/__init__.py | 0 src/knowledge/pipeline.py | 91 ++-------- src/knowledge/store.py | 10 +- src/service/aios_shell/aios_shell.py | 7 +- src/service/email_spider/converter.py | 24 --- src/service/email_spider/main.py | 12 -- src/service/email_spider/spider.py | 17 -- src/service/spider/email_spider.py | 171 ------------------ 19 files changed, 206 insertions(+), 399 deletions(-) create mode 100644 rootfs/knowledge_pipelines/local_embedding/pipeline.toml delete mode 100644 rootfs/knowledge_pipelines/local_photos_embedding/pipeline.toml create mode 100644 src/component/knowledge_manager/__init__.py rename src/{knowledge => component/knowledge_manager}/input/email.py (100%) create mode 100644 src/component/knowledge_manager/input/local_dir.py rename src/{knowledge => component/knowledge_manager}/parser/embedding.py (55%) create mode 100644 src/component/knowledge_manager/pipeline.py rename src/{knowledge => component/knowledge_manager}/query/embedding.py (100%) delete mode 100644 src/knowledge/input/__init__.py delete mode 100644 src/knowledge/input/local_dir.py delete mode 100644 src/knowledge/parser/__init__.py delete mode 100644 src/service/email_spider/converter.py delete mode 100644 src/service/email_spider/main.py delete mode 100644 src/service/email_spider/spider.py delete mode 100644 src/service/spider/email_spider.py diff --git a/rootfs/knowledge_pipelines/local_embedding/pipeline.toml b/rootfs/knowledge_pipelines/local_embedding/pipeline.toml new file mode 100644 index 0000000..ee8c5b0 --- /dev/null +++ b/rootfs/knowledge_pipelines/local_embedding/pipeline.toml @@ -0,0 +1,6 @@ +name = "LocalEmbedding" +input.module = "local_dir" +input.params.path = "${myai_dir}/data" +parser.module = "embedding" +parser.params.path = "${myai_dir}/knowledge/indices/embedding" + diff --git a/rootfs/knowledge_pipelines/local_photos_embedding/pipeline.toml b/rootfs/knowledge_pipelines/local_photos_embedding/pipeline.toml deleted file mode 100644 index e26e7a8..0000000 --- a/rootfs/knowledge_pipelines/local_photos_embedding/pipeline.toml +++ /dev/null @@ -1,6 +0,0 @@ -name = "LocalPhotoEmbedding" -input.module = "local_dir" -input.params.path = "~/myai/photos" -parser.module = "embedding" -parser.params.path = "~/myai/knowledge/indices/photo_embedding" - diff --git a/rootfs/knowledge_pipelines/pipelines.toml b/rootfs/knowledge_pipelines/pipelines.toml index 99d5b59..e6bf006 100644 --- a/rootfs/knowledge_pipelines/pipelines.toml +++ b/rootfs/knowledge_pipelines/pipelines.toml @@ -1,2 +1,3 @@ -[[pipelines]] -"local_photos_embedding" \ No newline at end of file +pipelines = [ + "local_embedding" +] \ No newline at end of file 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/knowledge/input/email.py b/src/component/knowledge_manager/input/email.py similarity index 100% rename from src/knowledge/input/email.py rename to src/component/knowledge_manager/input/email.py 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..76c1137 --- /dev/null +++ b/src/component/knowledge_manager/input/local_dir.py @@ -0,0 +1,65 @@ +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): + 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: + 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/knowledge/parser/embedding.py b/src/component/knowledge_manager/parser/embedding.py similarity index 55% rename from src/knowledge/parser/embedding.py rename to src/component/knowledge_manager/parser/embedding.py index e512eea..cb8b5f8 100644 --- a/src/knowledge/parser/embedding.py +++ b/src/component/knowledge_manager/parser/embedding.py @@ -1,31 +1,39 @@ # define a knowledge base class import json +import string from aios_kernel import ComputeKernel, AIStorage from knowledge import * class EmbeddingParser: - def __init__(self, params: dict): - self.store = KnowledgeStore() - self.compute_kernel = ComputeKernel.get_instance() - self.knowledge_base = KnowledgeBase() + def __init__(self, env: KnowledgePipelineEnvironment, config: dict): self._default_text_model = "all-MiniLM-L6-v2" self._default_image_model = "clip-ViT-B-32" - self.vector_store = ChromaVectorStore(AIStorage().get_myai_dir() / "knowledge", self._default_text_model) + + 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(AIStorage().get_myai_dir() / "knowledge", self._default_text_model) + return ChromaVectorStore(self.get_path(), model_name) async def __embedding_document(self, document: DocumentObject): for chunk_id in document.get_chunk_list(): - chunk = self.knowledge_base.store.get_chunk_reader().get_chunk(chunk_id) + 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 self.compute_kernel.do_text_embedding(text, self._default_text_model) + 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) + await self.__get_vector_store(self._default_text_model).insert(vector, chunk_id) async def __embedding_image(self, image: ImageObject): # desc = {} @@ -36,9 +44,9 @@ class EmbeddingParser: # 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) + 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()) + await self.__get_vector_store(self._default_image_model).insert(vector, image.calculate_id()) async def __embedding_video(self, vedio: VideoObject): desc = {} @@ -48,26 +56,26 @@ class EmbeddingParser: 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.get_vector_store(self._default_text_model).insert(vector, vedio.calculate_id()) + 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.store.get_object_store().get_object(document_id)) + 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.store.get_object_store().get_object(image_id)) + 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.store.get_object_store().get_object(video_id)) + 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.store.get_object_store().get_object(rich_text_id)) + 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 self.compute_kernel.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()) + 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()) @@ -85,9 +93,10 @@ class EmbeddingParser: else: pass - async def parse(self, object: ObjectID): - obj = self.knowledge_base.load_object(object) + 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(params: dict) -> EmbeddingParser: - return EmbeddingParser(params) \ No newline at end of file +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..653ea9c --- /dev/null +++ b/src/component/knowledge_manager/pipeline.py @@ -0,0 +1,73 @@ +import os +import runpy +import toml +import asyncio +from knowledge import KnowledgePipelineEnvironment, KnowledgePipeline + +class KnowledgePipelineManager: + 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.abspath(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.abspath(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) + + 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/query/embedding.py b/src/component/knowledge_manager/query/embedding.py similarity index 100% rename from src/knowledge/query/embedding.py rename to src/component/knowledge_manager/query/embedding.py diff --git a/src/knowledge/input/__init__.py b/src/knowledge/input/__init__.py deleted file mode 100644 index 0e9e2a5..0000000 --- a/src/knowledge/input/__init__.py +++ /dev/null @@ -1 +0,0 @@ -import local_dir \ No newline at end of file diff --git a/src/knowledge/input/local_dir.py b/src/knowledge/input/local_dir.py deleted file mode 100644 index 2bbafe1..0000000 --- a/src/knowledge/input/local_dir.py +++ /dev/null @@ -1,61 +0,0 @@ -import os -import aiofiles -import chardet -from knowledge import ImageObjectBuilder, DocumentObjectBuilder, KnowledgeBase - -class KnowledgeDirSource: - def __init__(self, config): - self.config = config - config["path"] = os.path.abspath(config["path"]) - - # @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): - # 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 - while True: - 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) - yield image.calculate_id() - # 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) - yield document.calculate_id() - # journal_client.insert(KnowledgeJournal("dir", self.id(), rel_path, str(document.calculate_id()), timestamp)) - yield 0 - - -def init(params: dict) -> KnowledgeDirSource: - return KnowledgeDirSource(params) \ No newline at end of file diff --git a/src/knowledge/parser/__init__.py b/src/knowledge/parser/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/knowledge/pipeline.py b/src/knowledge/pipeline.py index e051925..e637fb7 100644 --- a/src/knowledge/pipeline.py +++ b/src/knowledge/pipeline.py @@ -1,12 +1,7 @@ - -# class KnowledgePipelineTemplate -import runpy -import toml import datetime import sqlite3 import os from . import ObjectID, KnowledgeStore -import asyncio from enum import Enum class KnowledgePipelineJournal: @@ -17,9 +12,13 @@ class KnowledgePipelineJournal: self.parser = parser def is_finish(self) -> bool: - self.object_id is None + return self.object_id is None + def get_input(self) -> str: + return self.input + def get_parser(self) -> str: + return self.parser # init sqlite3 client class KnowledgePipelineJournalClient: @@ -43,7 +42,7 @@ class KnowledgePipelineJournalClient: 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 (?, ?, ?)", + "INSERT INTO journal (time, object_id, input, parser) VALUES (?, ?, ?, ?)", (timestamp, str(object_id), input, parser), ) conn.commit() @@ -56,10 +55,18 @@ class KnowledgePipelineJournalClient: class KnowledgePipelineEnvironment: def __init__(self, pipeline_path: str): - self.knowledge_base = KnowledgeStore() + 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 @@ -84,13 +91,13 @@ class KnowledgePipeline: self.parser = self.parser_init(self.env, self.parser_params) self.state = KnowledgePipelineState.RUNNING if self.state == KnowledgePipelineState.RUNNING: - for input in await self.input.next(): + 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 None: + if object_id is not None: parser_journal = await self.parser.parse(object_id) self.env.journal.insert(object_id, input_journal, parser_journal) if self.state == KnowledgePipelineState.STOPPED: @@ -98,69 +105,5 @@ class KnowledgePipeline: if self.state == KnowledgePipelineState.FINISHED: return -class KnowledgePipelineManager: - 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) - - 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 = self.input_modules[config["input"]["module"]] - _, ext = os.path.splitext(input_module) - if ext == ".py": - input_module = os.path.abspath(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 = self.parser_modules[config["parser"]["module"]] - _, ext = os.path.splitext(parser_module) - if ext == ".py": - parser_module = os.path.abspath(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 = 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) - - 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") - 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) - \ No newline at end of file diff --git a/src/knowledge/store.py b/src/knowledge/store.py index 803c3b6..a2ebd77 100644 --- a/src/knowledge/store.py +++ b/src/knowledge/store.py @@ -68,15 +68,15 @@ class KnowledgeStore: 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)) + return DocumentObject.decode(self.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)) + return ImageObject.decode(self.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)) + return VideoObject.decode(self.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)) + return RichTextObject.decode(self.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)) + return EmailObject.decode(self.object_store.get_object(object_id)) else: pass diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 191e0d1..0dc47ea 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -32,6 +32,7 @@ from aios_kernel 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__) @@ -185,9 +186,9 @@ class AIOS_Shell: AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg) - pipelines = KnowledgePipelineManager(AIStorage().get_instance().get_myai_dir() / "knowledge" / "pipelines") - pipelines.load_dir(AIStorage().get_instance().get_system_app_dir() / "knowledge_pipelines") - pipelines.load_dir(AIStorage().get_instance().get_myai_dir() / "knowledge_pipelines") + pipelines = KnowledgePipelineManager(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() 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 diff --git a/src/service/spider/email_spider.py b/src/service/spider/email_spider.py deleted file mode 100644 index cd44097..0000000 --- a/src/service/spider/email_spider.py +++ /dev/null @@ -1,171 +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 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) - - # 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') - - self.client = self.email_client() - - def email_client(self) -> imaplib.IMAP4_SSL: - self.logger.info(f"read email config from {self.config.get('EMAIL_IMAP_SERVER')}") - client = imaplib.IMAP4_SSL( - host=self.config.get('EMAIL_IMAP_SERVER'), - port=self.config.get('EMAIL_IMAP_PORT') - ) - client.login(self.config.get('EMAIL_ADDRESS'), self.config.get('EMAIL_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"): - 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() - 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") - - def read_and_save_email(self, uid: 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}]") - self.save_email(mail) - - def get_local_dir_name(self, mail: mailparser.MailParser) -> str: - dir = f"{self.config.get('LOCAL_DIR')}/{self.config.get('EMAIL_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): - 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}]") - dir = self.get_local_dir_name(mail) - self.logger.info(f"check email saved {dir}") - file = f"{dir}/email.txt" - if os.path.exists(file): - return False - return False - - # 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) - self.logger.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] - self.logger.info(f'Found {len(img_urls)} images in email body') - - 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]) - # 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}') - else: - self.logger.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')}" - 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}") - if not os.path.exists(email_dir): - os.makedirs(email_dir) - with open(f"{email_dir}/email.txt", "w") as f: - 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}") - - 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 From d0b74545ebddd7db3724fc9791fda2fdfc195c89 Mon Sep 17 00:00:00 2001 From: tsukasa Date: Fri, 20 Oct 2023 14:19:41 +0800 Subject: [PATCH 4/5] change /knowledge commands in shell --- rootfs/agents/Mia/agent.toml | 6 +- rootfs/knowledge_pipelines/Mia/input.py | 68 ++++++++++++ rootfs/knowledge_pipelines/Mia/parser.py | 102 ++++++++++++++++++ .../{local_embedding => Mia}/pipeline.toml | 6 +- .../knowledge_pipelines/Mia/query.py | 35 ++++-- rootfs/knowledge_pipelines/pipelines.toml | 2 +- src/aios_kernel/__init__.py | 1 + src/aios_kernel/agent.py | 3 +- src/component/agent_manager/agent_manager.py | 16 ++- .../knowledge_manager/input/local_dir.py | 21 ++-- src/component/knowledge_manager/pipeline.py | 20 +++- src/knowledge/pipeline.py | 14 +++ src/service/aios_shell/aios_shell.py | 59 +++------- 13 files changed, 279 insertions(+), 74 deletions(-) create mode 100644 rootfs/knowledge_pipelines/Mia/input.py create mode 100644 rootfs/knowledge_pipelines/Mia/parser.py rename rootfs/knowledge_pipelines/{local_embedding => Mia}/pipeline.toml (56%) rename src/component/knowledge_manager/query/embedding.py => rootfs/knowledge_pipelines/Mia/query.py (68%) 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/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/local_embedding/pipeline.toml b/rootfs/knowledge_pipelines/Mia/pipeline.toml similarity index 56% rename from rootfs/knowledge_pipelines/local_embedding/pipeline.toml rename to rootfs/knowledge_pipelines/Mia/pipeline.toml index ee8c5b0..9b9e062 100644 --- a/rootfs/knowledge_pipelines/local_embedding/pipeline.toml +++ b/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" diff --git a/src/component/knowledge_manager/query/embedding.py b/rootfs/knowledge_pipelines/Mia/query.py similarity index 68% rename from src/component/knowledge_manager/query/embedding.py rename to rootfs/knowledge_pipelines/Mia/query.py index f29bb43..cc4b557 100644 --- a/src/component/knowledge_manager/query/embedding.py +++ b/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]) \ No newline at end of file + 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 index e6bf006..3c78084 100644 --- a/rootfs/knowledge_pipelines/pipelines.toml +++ b/rootfs/knowledge_pipelines/pipelines.toml @@ -1,3 +1,3 @@ pipelines = [ - "local_embedding" + "Mia" ] \ No newline at end of file diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index fc11ad0..f632e03 100644 --- a/src/aios_kernel/__init__.py +++ b/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" diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index bacafe5..34d8967 100644 --- a/src/aios_kernel/agent.py +++ b/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"] 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/input/local_dir.py b/src/component/knowledge_manager/input/local_dir.py index 76c1137..e9a54f4 100644 --- a/src/component/knowledge_manager/input/local_dir.py +++ b/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) diff --git a/src/component/knowledge_manager/pipeline.py b/src/component/knowledge_manager/pipeline.py index 653ea9c..55a5105 100644 --- a/src/component/knowledge_manager/pipeline.py +++ b/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"]: diff --git a/src/knowledge/pipeline.py b/src/knowledge/pipeline.py index e637fb7..af3b665 100644 --- a/src/knowledge/pipeline.py +++ b/src/knowledge/pipeline.py @@ -19,6 +19,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: @@ -84,6 +90,12 @@ class KnowledgePipeline: 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: @@ -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: diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 0dc47ea..b8ce7a6 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/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()) @@ -333,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 @@ -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', From 8bc63c912a255798ee2de17acbd9da0c5d8ad569 Mon Sep 17 00:00:00 2001 From: tsukasa Date: Fri, 20 Oct 2023 15:08:01 +0800 Subject: [PATCH 5/5] add pipeline issue --- doc/knowledge_pipeline.md | 87 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 87 insertions(+) create mode 100644 doc/knowledge_pipeline.md 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方法的提示词。 + + + + +