From 97b18e9f6694208c9554e4ec446e7f7cab6cb349 Mon Sep 17 00:00:00 2001 From: tsukasa Date: Thu, 2 Nov 2023 10:40:37 +0800 Subject: [PATCH 1/4] email parser document --- doc/tob email.md | 62 +++++++ .../knowledge_pipelines/Bucky.Mail/input.py | 156 ++++++++++++++++++ .../knowledge_manager/input/email.py | 7 +- src/component/knowledge_manager/pipeline.py | 4 +- src/knowledge/pipeline.py | 5 + 5 files changed, 226 insertions(+), 8 deletions(-) create mode 100644 doc/tob email.md create mode 100644 rootfs/knowledge_pipelines/Bucky.Mail/input.py diff --git a/doc/tob email.md b/doc/tob email.md new file mode 100644 index 0000000..f59124e --- /dev/null +++ b/doc/tob email.md @@ -0,0 +1,62 @@ +# issue tree +最核心的机制是树状的issue管理，一个issue应当包含以下属性： ++ 谁提出来的 ++ 分配给谁的，如果有的话 ++ 起始日期 ++ deadline，如果有的话 ++ 在哪个邮件里面提出的，引用某个email的原始链接 ++ 这个issue的summary，有几种情况， + + 一个新的任务，要达成什么目标 + + 提出了一个问题，需求答案 + + 解决了某个issue，完成了task或者解答了一个问题 ++ 推断出来的 issue的状态，进行中，关闭，超时，完成了 ++ parent issue + +knowledge维护一个issue tree，从一个root issue出发（root可以是抽象的，比如一个组织的存在，并不是具体的）；knowledge env 提供对这个issue tree的维护接口： ++ 新增issue ++ 更新issue + +# parse email +假定从从某个起始日期开始，以每天为单位，扫描当天新增的email，对每封email： +1. 输入email 和 从knowledge base获取 issue tree +2. llm提示词应当包括：issue tree， email正文， knowledge env， llm完成如下推理： ++ email正文提出了一个新的issue，在knowledge env新增issue ++ email正文改变了一个issue的状态 + + 通报完成了一个task + + 回答了一个问题 + + 明确改变一个issue的状态：认为完成，要延期，认为要取消 ++ 根据推理结果正确产生knowledge env 的调用，更新issue tree的状态 + +## 推理部分可能的out of token： +1. 裁剪掉已经关闭，超时的 issue +2. 根据标题特征，是不是对某个email的回复，定位到某个issue， 裁剪出 sub tree +2. 很长的邮件正文： + 1. 第一种方法：先llm推理email的summary，再把summary当正文输入推理issue + 2. 第二种方法（我觉得更好）：分片迭代输入email正文，单次llm推理的提示词就变成：issue tree， 当前email summary， 当段email正文，knowledge env： + + env里面新增一个method，更新当前email summary + + +# build issue tree +## 第一种结构：基于knowledge pipeline +1. pipeline input： 判定当前时间晚于 起始时间并且早于下一个自然天，开始爬正确范围内的邮件输入 +2. pipeline parser：包含准备user prompt 的计算部分，和几个agent ++ 计算部分： 裁剪issue tree，[可选的：调用llm推理生成summary] ++ agent 部分： + + agent提示词：从输入的结构化issue tree， 和邮件正文，回复对issue tree knowledge env的调用 + + 输入提示词： email 正文或者summary，裁剪后的issue tree ++ parser的流程： + 对每一个输入的email，查询（裁剪）当前issue tree，把email 和 issue tree 当作user prompt发送给agent，等待agent返回 + + +## 第二种结构：基于agent workspace（待定） +1. schedule task：在每一天产生一个build issue tree task +2. build issue tree agent: 响应build issue tree task（可不可以以计算为入口，还是只能agent入口） ++ agent调用email env，读出一封邮件 ++ agent调用knowledge env，返回issue tree ++ agent从邮件内容和issue tree推理，回复对issue tree knowledge env 的调用 + +# query issue tree +主动的或者被动的根据当前issue tree的状态，推理出一些汇总的结论： ++ 是不是有超期的事项 ++ 事情是不是有在推进 ++ 有哪些事情完成了 \ No newline at end of file diff --git a/rootfs/knowledge_pipelines/Bucky.Mail/input.py b/rootfs/knowledge_pipelines/Bucky.Mail/input.py new file mode 100644 index 0000000..23ae40a --- /dev/null +++ b/rootfs/knowledge_pipelines/Bucky.Mail/input.py @@ -0,0 +1,156 @@ +import os +import aiofiles +import chardet +import logging +import string +from knowledge import ImageObjectBuilder, DocumentObjectBuilder, KnowledgePipelineEnvironment, KnowledgePipelineJournal +from aios_kernel.storage import AIStorage + +class KnowledgeEmailSource: + def __init__(self, env: KnowledgePipelineEnvironment, config:dict): + self.config = config + + # @classmethod + # def user_config_items(cls): + # return [("address", "email address"), + # ("password", "email password"), + # ("imap_server", "imap server"), + # ("imap_port", "imap port") + # ] + + 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/component/knowledge_manager/input/email.py b/src/component/knowledge_manager/input/email.py index a8eb353..3bcc6b3 100644 --- a/src/component/knowledge_manager/input/email.py +++ b/src/component/knowledge_manager/input/email.py @@ -1,6 +1,6 @@ class KnowledgeEmailSource: - def __init__(self, config:dict): + def __init__(self, env: KnowledgePipelineEnvironment, config:dict): self.config = config self.config["type"] = "email" @@ -15,11 +15,6 @@ class KnowledgeEmailSource: ("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) diff --git a/src/component/knowledge_manager/pipeline.py b/src/component/knowledge_manager/pipeline.py index 55a5105..1701f4f 100644 --- a/src/component/knowledge_manager/pipeline.py +++ b/src/component/knowledge_manager/pipeline.py @@ -46,7 +46,7 @@ class KnowledgePipelineManager: input_init = runpy.run_path(input_module)["init"] else: input_init = self.input_modules.get(input_module) - input_params = config["input"]["params"] + input_params = config["input"].get("params") parser_module = config["parser"]["module"] _, ext = os.path.splitext(parser_module) @@ -55,7 +55,7 @@ class KnowledgePipelineManager: parser_init = runpy.run_path(parser_module)["init"] else: parser_init = self.parser_modules.get(parser_module) - parser_params = config["parser"]["params"] + parser_params = config["parser"].get("params") data_path = os.path.join(self.root_dir, name) diff --git a/src/knowledge/pipeline.py b/src/knowledge/pipeline.py index af3b665..62f246a 100644 --- a/src/knowledge/pipeline.py +++ b/src/knowledge/pipeline.py @@ -1,6 +1,7 @@ import datetime import sqlite3 import os +import logging from . import ObjectID, KnowledgeStore from enum import Enum @@ -66,12 +67,16 @@ class KnowledgePipelineEnvironment: os.makedirs(pipeline_path) self.pipeline_path = pipeline_path self.journal = KnowledgePipelineJournalClient(pipeline_path) + self.logger = logging.getLogger() def get_journal(self) -> KnowledgePipelineJournalClient: return self.journal def get_knowledge_store(self) -> KnowledgeStore: return self.knowledge_store + + def get_logger(self) -> logging.Logger: + return self.logger class KnowledgePipelineState(Enum): INIT = 0 From 9c00187041136561c943b5ded1a4011994bf8635 Mon Sep 17 00:00:00 2001 From: tsukasa Date: Tue, 14 Nov 2023 18:12:26 +0800 Subject: [PATCH 2/4] a issue parser of email --- rootfs/agents/FindPhoto/agent.toml | 21 -- rootfs/agents/FindPhoto/enviroment.py | 0 .../knowledge_pipelines/Bucky.Mail/input.py | 156 ----------- .../knowledge_pipelines/Mail/Issue/local.py | 9 + .../knowledge_pipelines/Mail/Issue/parser.py | 10 + .../Mail/Issue/pipeline.toml | 9 + .../knowledge_pipelines/Mail/Spider/input.py | 10 + .../Mail/Spider/pipeline.toml | 4 + rootfs/knowledge_pipelines/pipelines.toml | 2 +- src/aios_kernel/__init__.py | 2 +- src/aios_kernel/agent_base.py | 117 +++++++- src/aios_kernel/environment.py | 4 +- .../knowledge_manager/input/email.py | 153 ---------- .../knowledge_manager/input/local_dir.py | 68 ----- .../knowledge_manager/parser/embedding.py | 102 ------- src/component/knowledge_manager/pipeline.py | 6 +- src/component/mail_environment/1/mail.json | 6 + src/component/mail_environment/1/mail.txt | 51 ++++ src/component/mail_environment/__init__.py | 3 + src/component/mail_environment/issue.py | 200 +++++++++++++ src/component/mail_environment/local.py | 34 +++ src/component/mail_environment/mail.py | 265 ++++++++++++++++++ src/component/mail_environment/spider.py | 53 ++++ src/knowledge/object/object.py | 18 +- src/knowledge/object/object_id.py | 11 + src/knowledge/pipeline.py | 3 + src/service/aios_shell/aios_shell.py | 1 - 27 files changed, 797 insertions(+), 521 deletions(-) delete mode 100644 rootfs/agents/FindPhoto/agent.toml delete mode 100644 rootfs/agents/FindPhoto/enviroment.py delete mode 100644 rootfs/knowledge_pipelines/Bucky.Mail/input.py create mode 100644 rootfs/knowledge_pipelines/Mail/Issue/local.py create mode 100644 rootfs/knowledge_pipelines/Mail/Issue/parser.py create mode 100644 rootfs/knowledge_pipelines/Mail/Issue/pipeline.toml create mode 100644 rootfs/knowledge_pipelines/Mail/Spider/input.py create mode 100644 rootfs/knowledge_pipelines/Mail/Spider/pipeline.toml delete mode 100644 src/component/knowledge_manager/input/email.py delete mode 100644 src/component/knowledge_manager/input/local_dir.py delete mode 100644 src/component/knowledge_manager/parser/embedding.py create mode 100644 src/component/mail_environment/1/mail.json create mode 100644 src/component/mail_environment/1/mail.txt create mode 100644 src/component/mail_environment/__init__.py create mode 100644 src/component/mail_environment/issue.py create mode 100644 src/component/mail_environment/local.py create mode 100644 src/component/mail_environment/mail.py create mode 100644 src/component/mail_environment/spider.py diff --git a/rootfs/agents/FindPhoto/agent.toml b/rootfs/agents/FindPhoto/agent.toml deleted file mode 100644 index ddb3f3f..0000000 --- a/rootfs/agents/FindPhoto/agent.toml +++ /dev/null @@ -1,21 +0,0 @@ -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 deleted file mode 100644 index e69de29..0000000 diff --git a/rootfs/knowledge_pipelines/Bucky.Mail/input.py b/rootfs/knowledge_pipelines/Bucky.Mail/input.py deleted file mode 100644 index 23ae40a..0000000 --- a/rootfs/knowledge_pipelines/Bucky.Mail/input.py +++ /dev/null @@ -1,156 +0,0 @@ -import os -import aiofiles -import chardet -import logging -import string -from knowledge import ImageObjectBuilder, DocumentObjectBuilder, KnowledgePipelineEnvironment, KnowledgePipelineJournal -from aios_kernel.storage import AIStorage - -class KnowledgeEmailSource: - def __init__(self, env: KnowledgePipelineEnvironment, config:dict): - self.config = config - - # @classmethod - # def user_config_items(cls): - # return [("address", "email address"), - # ("password", "email password"), - # ("imap_server", "imap server"), - # ("imap_port", "imap port") - # ] - - 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/rootfs/knowledge_pipelines/Mail/Issue/local.py b/rootfs/knowledge_pipelines/Mail/Issue/local.py new file mode 100644 index 0000000..1edfff8 --- /dev/null +++ b/rootfs/knowledge_pipelines/Mail/Issue/local.py @@ -0,0 +1,9 @@ +import sys +import os +from knowledge import KnowledgePipelineEnvironment +directory = os.path.dirname(__file__) +sys.path.append(directory + '/../../../../src/component/') + +from mail_environment import LocalEmail +def init(env: KnowledgePipelineEnvironment, params: dict): + return LocalEmail(env, params) \ No newline at end of file diff --git a/rootfs/knowledge_pipelines/Mail/Issue/parser.py b/rootfs/knowledge_pipelines/Mail/Issue/parser.py new file mode 100644 index 0000000..a5f819b --- /dev/null +++ b/rootfs/knowledge_pipelines/Mail/Issue/parser.py @@ -0,0 +1,10 @@ +import sys +import os +from knowledge import * +directory = os.path.dirname(__file__) + +sys.path.append(directory + '/../../../../src/component/') +from mail_environment import IssueParser + +def init(env: KnowledgePipelineEnvironment, params: dict): + return IssueParser(env, params) \ No newline at end of file diff --git a/rootfs/knowledge_pipelines/Mail/Issue/pipeline.toml b/rootfs/knowledge_pipelines/Mail/Issue/pipeline.toml new file mode 100644 index 0000000..d909ba9 --- /dev/null +++ b/rootfs/knowledge_pipelines/Mail/Issue/pipeline.toml @@ -0,0 +1,9 @@ +name = "Mail.Issue" +input.module = "local.py" +input.params.path = "${myai_dir}/mail" +input.params.watch = true +parser.module = "parser.py" +parser.params.mail_path = "${myai_dir}/mail" +parser.params.issue_path = "${myai_dir}/mail/issue.json" +parser.params.root_issue = "巴克云公司推进中的项目" + diff --git a/rootfs/knowledge_pipelines/Mail/Spider/input.py b/rootfs/knowledge_pipelines/Mail/Spider/input.py new file mode 100644 index 0000000..d5e6958 --- /dev/null +++ b/rootfs/knowledge_pipelines/Mail/Spider/input.py @@ -0,0 +1,10 @@ +import sys +import os +from knowledge import KnowledgePipelineEnvironment +directory = os.path.dirname(__file__) +sys.path.append(directory + '/../../../../src/component/') + +from mail_environment import EmailSpider + +def init(env: KnowledgePipelineEnvironment, params: dict): + return EmailSpider(env, params) \ No newline at end of file diff --git a/rootfs/knowledge_pipelines/Mail/Spider/pipeline.toml b/rootfs/knowledge_pipelines/Mail/Spider/pipeline.toml new file mode 100644 index 0000000..df9f306 --- /dev/null +++ b/rootfs/knowledge_pipelines/Mail/Spider/pipeline.toml @@ -0,0 +1,4 @@ +name = "Mail.Issue" +input.module = "input.py" +input.params.path = "${myai_dir}/data" + diff --git a/rootfs/knowledge_pipelines/pipelines.toml b/rootfs/knowledge_pipelines/pipelines.toml index 3c78084..7e69436 100644 --- a/rootfs/knowledge_pipelines/pipelines.toml +++ b/rootfs/knowledge_pipelines/pipelines.toml @@ -1,3 +1,3 @@ pipelines = [ - "Mia" + "Mail/Issue" ] \ No newline at end of file diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index 9093b07..a1ab85f 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -1,7 +1,7 @@ from .environment import Environment,EnvironmentEvent from .agent_base import AgentMsg,AgentMsgStatus,AgentMsgType,AgentPrompt from .chatsession import AIChatSession -from .agent import AIAgent,AIAgentTemplete +from .agent import AIAgent,AIAgentTemplete, BaseAIAgent from .compute_kernel import ComputeKernel,ComputeTask,ComputeTaskResult,ComputeTaskState,ComputeTaskType from .compute_node import ComputeNode,LocalComputeNode from .open_ai_node import OpenAI_ComputeNode diff --git a/src/aios_kernel/agent_base.py b/src/aios_kernel/agent_base.py index 95b5e9e..60c5fe6 100644 --- a/src/aios_kernel/agent_base.py +++ b/src/aios_kernel/agent_base.py @@ -11,8 +11,10 @@ import shlex import json from typing import List -from .ai_function import FunctionItem -from .compute_task import ComputeTaskResult +from .ai_function import FunctionItem, AIFunction +from .compute_task import ComputeTaskResult,ComputeTaskResultCode +from .environment import Environment + logger = logging.getLogger(__name__) @@ -592,3 +594,114 @@ class CustomAIAgent(BaseAIAgent): def get_llm_learn_token_limit(self) -> int: return self.llm_learn_token_limit + +class BaseAIAgent: + def __init__(self) -> None: + pass + + @classmethod + def _get_inner_functions(cls, env:Environment) -> (dict,int): + if env is None: + return None,0 + + all_inner_function = env.get_all_ai_functions() + if all_inner_function is None: + return None,0 + + result_func = [] + result_len = 0 + for inner_func in all_inner_function: + func_name = inner_func.get_name() + this_func = {} + this_func["name"] = func_name + this_func["description"] = inner_func.get_description() + this_func["parameters"] = inner_func.get_parameters() + result_len += len(json.dumps(this_func)) / 4 + result_func.append(this_func) + + return result_func,result_len + + @classmethod + async def do_llm_complection( + cls, + env:Environment, + prompt:AgentPrompt, + org_msg:AgentMsg, + llm_model_name:str, + max_token_size:int + ) -> ComputeTaskResult: + from .compute_kernel import ComputeKernel + #logger.debug(f"Agent {self.agent_id} do llm token static system:{system_prompt_len},function:{function_token_len},history:{history_token_len},input:{input_len}, totoal prompt:{system_prompt_len + function_token_len + history_token_len} ") + inner_functions,inner_functions_len = cls._get_inner_functions(env) + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,llm_model_name,max_token_size,inner_functions) + if task_result.result_code != ComputeTaskResultCode.OK: + logger.error(f"llm compute error:{task_result.error_str}") + #error_resp = msg.create_error_resp(task_result.error_str) + return task_result + + result_message = task_result.result.get("message") + inner_func_call_node = None + if result_message: + inner_func_call_node = result_message.get("function_call") + + if inner_func_call_node: + call_prompt : AgentPrompt = copy.deepcopy(prompt) + task_result = await cls._execute_func(env,inner_func_call_node,call_prompt,inner_functions,org_msg,llm_model_name,max_token_size) + + return task_result + + @classmethod + async def _execute_func( + cls, + env: Environment, + inner_func_call_node: dict, + prompt: AgentPrompt, + inner_functions: dict, + org_msg:AgentMsg, + llm_model_name:str, + max_token_size:int, + stack_limit = 5 + ) -> ComputeTaskResult: + from .compute_kernel import ComputeKernel + func_name = inner_func_call_node.get("name") + arguments = json.loads(inner_func_call_node.get("arguments")) + logger.info(f"llm execute inner func:{func_name} ({json.dumps(arguments)})") + + func_node : AIFunction = env.get_ai_function(func_name) + if func_node is None: + result_str = f"execute {func_name} error,function not found" + else: + if org_msg: + ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target) + + try: + result_str:str = await func_node.execute(**arguments) + except Exception as e: + result_str = f"execute {func_name} error:{str(e)}" + logger.error(f"llm execute inner func:{func_name} error:{e}") + + + logger.info("llm execute inner func result:" + result_str) + + prompt.messages.append({"role":"function","content":result_str,"name":func_name}) + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,llm_model_name,max_token_size,inner_functions) + if task_result.result_code != ComputeTaskResultCode.OK: + logger.error(f"llm compute error:{task_result.error_str}") + return task_result + + ineternal_call_record.result_str = task_result.result_str + ineternal_call_record.done_time = time.time() + if org_msg: + org_msg.inner_call_chain.append(ineternal_call_record) + + inner_func_call_node = None + if stack_limit > 0: + result_message : dict = task_result.result.get("message") + if result_message: + inner_func_call_node = result_message.get("function_call") + + if inner_func_call_node: + return await cls._execute_func(env,inner_func_call_node,prompt,inner_functions,org_msg,llm_model_name,max_token_size,stack_limit-1) + else: + return task_result +>>>>>>> 2f9cee9 (a issue parser of email) diff --git a/src/aios_kernel/environment.py b/src/aios_kernel/environment.py index cf5e9b3..ab384cd 100644 --- a/src/aios_kernel/environment.py +++ b/src/aios_kernel/environment.py @@ -45,8 +45,8 @@ class Environment: #@abstractmethod #TODO: how to use env? different env has different prompt - #def get_env_prompt(self) -> str: - # pass + def get_env_prompt(self) -> str: + pass def add_ai_function(self,func:AIFunction) -> None: if self.functions.get(func.get_name()) is not None: diff --git a/src/component/knowledge_manager/input/email.py b/src/component/knowledge_manager/input/email.py deleted file mode 100644 index 3bcc6b3..0000000 --- a/src/component/knowledge_manager/input/email.py +++ /dev/null @@ -1,153 +0,0 @@ - -class KnowledgeEmailSource: - def __init__(self, env: KnowledgePipelineEnvironment, 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") - ] - - 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/component/knowledge_manager/input/local_dir.py b/src/component/knowledge_manager/input/local_dir.py deleted file mode 100644 index e9a54f4..0000000 --- a/src/component/knowledge_manager/input/local_dir.py +++ /dev/null @@ -1,68 +0,0 @@ -import os -import aiofiles -import chardet -import logging -import string -from knowledge import ImageObjectBuilder, DocumentObjectBuilder, KnowledgePipelineEnvironment, KnowledgePipelineJournal -from aios_kernel.storage import AIStorage - -class KnowledgeDirSource: - def __init__(self, env: KnowledgePipelineEnvironment, config): - self.env = env - path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) - config["path"] = path - self.config = config - - # @classmethod - # def user_config_items(cls): - # return [("path", "local dir path")] - - def path(self): - return self.config["path"] - - @staticmethod - async def read_txt_file(file_path:str)->str: - cur_encode = "utf-8" - async with aiofiles.open(file_path,'rb') as f: - cur_encode = chardet.detect(await f.read())['encoding'] - - async with aiofiles.open(file_path,'r',encoding=cur_encode) as f: - return await f.read() - - async def next(self): - while True: - journals = self.env.journal.latest_journals(1) - from_time = 0 - if len(journals) == 1: - latest_journal = journals[0] - if latest_journal.is_finish(): - yield None - continue - from_time = os.path.getctime(latest_journal.get_input()) - if os.path.getmtime(self.path()) <= from_time: - yield (None, None) - continue - - file_pathes = sorted(os.listdir(self.path()), key=lambda x: os.path.getctime(os.path.join(self.path(), x))) - for rel_path in file_pathes: - file_path = os.path.join(self.path(), rel_path) - timestamp = os.path.getctime(file_path) - if timestamp <= from_time: - continue - ext = os.path.splitext(file_path)[1].lower() - if ext in ['.jpg', '.jpeg', '.png', '.gif', '.bmp']: - logging.info(f"knowledge dir source found image file {file_path}") - image = ImageObjectBuilder({}, {}, file_path).build(self.env.get_knowledge_store()) - await self.env.get_knowledge_store().insert_object(image) - yield (image.calculate_id(), file_path) - if ext in ['.txt']: - logging.info(f"knowledge dir source found text file {file_path}") - text = await self.read_txt_file(file_path) - document = DocumentObjectBuilder({}, {}, text).build(self.env.get_knowledge_store()) - await self.env.get_knowledge_store().insert_object(document) - yield (document.calculate_id(), file_path) - yield (None, None) - - -def init(env: KnowledgePipelineEnvironment, params: dict) -> KnowledgeDirSource: - return KnowledgeDirSource(env, params) \ No newline at end of file diff --git a/src/component/knowledge_manager/parser/embedding.py b/src/component/knowledge_manager/parser/embedding.py deleted file mode 100644 index cb8b5f8..0000000 --- a/src/component/knowledge_manager/parser/embedding.py +++ /dev/null @@ -1,102 +0,0 @@ -# define a knowledge base class -import json -import string -from aios_kernel import ComputeKernel, AIStorage -from knowledge import * - - -class EmbeddingParser: - def __init__(self, env: KnowledgePipelineEnvironment, config: dict): - self._default_text_model = "all-MiniLM-L6-v2" - self._default_image_model = "clip-ViT-B-32" - - path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) - if not os.path.exists(path): - os.makedirs(path) - config["path"] = path - - self.env = env - self.config = config - - def get_path(self) -> str: - return self.config["path"] - - def __get_vector_store(self, model_name: str) -> ChromaVectorStore: - return ChromaVectorStore(self.get_path(), model_name) - - async def __embedding_document(self, document: DocumentObject): - for chunk_id in document.get_chunk_list(): - chunk = self.env.get_knowledge_store().get_chunk_reader().get_chunk(chunk_id) - if chunk is None: - raise ValueError(f"text chunk not found: {chunk_id}") - - text = chunk.read().decode("utf-8") - vector = await ComputeKernel.get_instance().do_text_embedding(text, self._default_text_model) - if vector: - await self.__get_vector_store(self._default_text_model).insert(vector, chunk_id) - - async def __embedding_image(self, image: ImageObject): - # desc = {} - # if not not image.get_meta(): - # desc["meta"] = image.get_meta() - # if not not image.get_exif(): - # desc["exif"] = image.get_exif() - # if not not image.get_tags(): - # desc["tags"] = image.get_tags() - # vector = await self.compute_kernel.do_text_embedding(json.dumps(desc), self._default_text_model) - vector = await ComputeKernel.get_instance().do_image_embedding(image.calculate_id(), self._default_image_model) - if vector: - await self.__get_vector_store(self._default_image_model).insert(vector, image.calculate_id()) - - async def __embedding_video(self, vedio: VideoObject): - desc = {} - if not not vedio.get_meta(): - desc["meta"] = vedio.get_meta() - if not not vedio.get_info(): - desc["info"] = vedio.get_info() - if not not vedio.get_tags(): - desc["tags"] = vedio.get_tags() - vector = await ComputeKernel.get_instance().do_text_embedding(json.dumps(desc), self._default_text_model) - await self.__get_vector_store(self._default_text_model).insert(vector, vedio.calculate_id()) - - async def __embedding_rich_text(self, rich_text: RichTextObject): - for document_id in rich_text.get_documents().values(): - document = DocumentObject.decode(self.env.get_knowledge_store().get_object_store().get_object(document_id)) - await self.__embedding_document(document) - for image_id in rich_text.get_images().values(): - image = ImageObject.decode(self.env.get_knowledge_store().get_object_store().get_object(image_id)) - await self.__embedding_image(image) - for video_id in rich_text.get_videos().values(): - video = VideoObject.decode(self.env.get_knowledge_store().get_object_store().get_object(video_id)) - await self.__embedding_video(video) - for rich_text_id in rich_text.get_rich_texts().values(): - rich_text = RichTextObject.decode(self.env.get_knowledge_store().get_object_store().get_object(rich_text_id)) - await self.__embedding_rich_text(rich_text) - - async def __embedding_email(self, email: EmailObject): - vector = await ComputeKernel.get_instance().do_text_embedding(json.dumps(email.get_desc()), self._default_text_model) - await self.__get_vector_store(self._default_text_model).insert(vector, email.calculate_id()) - await self.__embedding_rich_text(email.get_rich_text()) - - - async def __do_embedding(self, object: KnowledgeObject): - if object.get_object_type() == ObjectType.Document: - await self.__embedding_document(object) - if object.get_object_type() == ObjectType.Image: - await self.__embedding_image(object) - if object.get_object_type() == ObjectType.Video: - await self.__embedding_video(object) - if object.get_object_type() == ObjectType.RichText: - await self.__embedding_rich_text(object) - if object.get_object_type() == ObjectType.Email: - await self.__embedding_email(object) - else: - pass - - async def parse(self, object: ObjectID) -> str: - obj = self.env.get_knowledge_store().load_object(object) - await self.__do_embedding(obj) - return "insert into vector store" - -def init(env: KnowledgePipelineEnvironment, params: dict) -> EmbeddingParser: - return EmbeddingParser(env, params) \ No newline at end of file diff --git a/src/component/knowledge_manager/pipeline.py b/src/component/knowledge_manager/pipeline.py index 1701f4f..efe0936 100644 --- a/src/component/knowledge_manager/pipeline.py +++ b/src/component/knowledge_manager/pipeline.py @@ -23,10 +23,6 @@ class KnowledgePipelineManager: "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 @@ -84,6 +80,6 @@ class KnowledgePipelineManager: 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: + with open(os.path.join(pipeline_path, "pipeline.toml"), 'r', encoding='utf-8') as f: pipeline_config = toml.load(f) self.add_pipeline(pipeline_config, pipeline_path) diff --git a/src/component/mail_environment/1/mail.json b/src/component/mail_environment/1/mail.json new file mode 100644 index 0000000..8a30f00 --- /dev/null +++ b/src/component/mail_environment/1/mail.json @@ -0,0 +1,6 @@ +{ + "subject": "开发dmc开源客户端", + "from_addr": "sichangjun@buckyos.com", + "to_addr": ["liuzhicong@buckyos.com"], + "date": "2023-4-10 21:00" +} \ No newline at end of file diff --git a/src/component/mail_environment/1/mail.txt b/src/component/mail_environment/1/mail.txt new file mode 100644 index 0000000..1abf50b --- /dev/null +++ b/src/component/mail_environment/1/mail.txt @@ -0,0 +1,51 @@ +最小功能集：4.15准备好oktc合约和客户端工具；4.21之前完成一些矿场节点部署； +Oktc合约和rust接口 （秋总） +实现以DMC结算的bill 和 order （done） +实现链上挑战—— merkle联通证明；用户提交低深度半路径；矿工提供叶子原文和高深度半路径；（doing） +实现按照价格和质押率索引的spv节点 —— 支持用户按参数匹配下单；（doing） +实现对eth发起的http请求 auth 头认证 —— 支持https实现 链下的 write/restore/challenge 身份认证；（doing） +实现oktc client event listener的block number本地持久化；（doing） +用户端工具：面向普通存储用户，一键备份和恢复； +源数据管理服务： +注册本地路径，生成链下挑战密码本 ——随机偏移和长度QA（done） + 生成链上挑战密码本 —— merkle根和半深度茎节点（doing）； +账号服务： +添加本地路径，选择bill id发起order（done） +按参数匹配order ——依赖按参数索引的spv 服务（doing） +生成order提交到交付服务；（done） +交付服务 +注册order和关联的源数据（done） +提交merkle根（done） +监听链事件，同步order状态，向miner写入源数据；（done） +使用密码本持续发起链下挑战（done） +实现链上挑战（doing） +恢复数据到本地目录（done） +关键日志服务 +关键状态改变日志写入（doing） +关键状态日志事件监听接口（doing） +轻量命令行客户端 +服务部署脚本（doing） +传入本地文件 和 order参数一键完成备份 （doing） +一键恢复（doing） +链下挑战失效时，自动发起链上挑战 ——依赖关键日志服务（doing） +mysql实现移植到sqlite（看时间和问题多不多；undo） +矿场端工具：本版本不是发布重点；保证能在自由的节点上稳定运行即可；结构上保证了一定的伸缩性； +存储扇区管理服务：包括gateway 和 node；保证简单部署和扩展，可靠性暂时不需实现； +扇区gateway服务：注册node，分发扇区读写请求到node；（doing） +扇区node服务：注册本地目录到node，注册为可用扇区；（测试中单node可以直接接入，done） +账号服务： +添加扇区和挂单参数，生成bill（done）； +监听链事件，响应新的order转入交付服务；（done） +监听链事件，响应链上挑战转入交付服务；（doing） +交付服务： +注册order和关联的扇区；（done） +对user提供交付相关http接口： +查询miner端order状态（done） +写入源数据（done） +完成写入后准备生成merkle root准备证明（done） +恢复源数据（done） +接收链下挑战返回证明（done） +自动提交链上挑战（doing） +关键日志服务 +关键状态改变日志写入（可推迟实现；undo） +自有节点部署 \ No newline at end of file diff --git a/src/component/mail_environment/__init__.py b/src/component/mail_environment/__init__.py new file mode 100644 index 0000000..402cf90 --- /dev/null +++ b/src/component/mail_environment/__init__.py @@ -0,0 +1,3 @@ +from .issue import IssueParser +from .local import LocalEmail +from .spider import EmailSpider \ No newline at end of file diff --git a/src/component/mail_environment/issue.py b/src/component/mail_environment/issue.py new file mode 100644 index 0000000..53ed3c6 --- /dev/null +++ b/src/component/mail_environment/issue.py @@ -0,0 +1,200 @@ +# define a knowledge base class +import json +import string +from aios_kernel import ComputeKernel, AIStorage, Environment, SimpleAIFunction, BaseAIAgent, AgentPrompt, AgentMsg +from knowledge import * +from .mail import MailStorage, Mail + +class IssueState(Enum): + Open = 1 + InProgress = 2 + Closed = 3 + +class IssueUpdateHistory: + def __init__(self, source: str, changes: dict) -> None: + self.source = source + self.changes = changes + + +class Issue: + def __init__(self) -> None: + self.id = None + self.summary = "" + self.state = IssueState.Open + self.source: str = None + self.create_time: datetime = None + self.deadline: datetime = None + self.update_history = [] + self.children = [] + self.parent: ObjectID = None + + @classmethod + def object_type(cls) -> ObjectType: + return ObjectType.from_user_def_type_code(0) + + def to_prompt(self) -> str: + prompt = { + "id": self.id, + "summary": self.summary, + "state": self.state.name, + "deadline": self.deadline + } + return json.dumps(prompt) + + @classmethod + def prompt_desc(cls) -> str: + return '''a issue contains following fileds: { + id: a guid string to identify a issue + summary: summary of this issue + state: state of this issue, will be one of [Open, InProgress, Closed], + deadline: if issue is not closed, deadline is the time to close this issue + } + ''' + + def calculate_id(self) -> str: + desc = { + "summary": self.summary, + "source": self.source, + "create_time": self.create_time, + "deadline": self.deadline, + "parent": self.parent, + } + id = str(KnowledgeObject(Issue.object_type(), desc).calculate_id()) + self.id = id + return id + + +class IssueStorage: + def __init__(self, path: str, root: Issue=None) -> None: + self.path = path + if not os.path.exists(path): + self.root = root + else: + self.root = json.load(open(path, "r")) + + def __flush(self): + json.dump(self.root, open(self.path, "w")) + + def get_issue_by_id(self, id: str) -> Issue: + self.root() + + def __get_issue_by_mail_in_subtree(self, root_issue: Issue, mail_id: str): + if root_issue.source == mail_id: + return root_issue + if root_issue.children is None or len(root_issue.children) == 0: + return None + for child_issue in root_issue.children: + this_issue = self.__get_issue_by_mail_in_subtree(child_issue, mail_id) + if this_issue is not None: + return this_issue + return None + + def get_issue_by_mail(self, mail_storage: MailStorage, mail: Mail) -> Issue: + if mail.reply_to is None: + return self.root + this_mail = mail_storage.get_mail_by_id(mail.reply_to) + while True: + issue = self.__get_issue_by_mail_in_subtree(self.root, this_mail.id) + if issue is not None: + return issue + if this_mail.replay_to is None: + return self.root + this_mail = mail_storage.get_mail_by_id(this_mail.reply_to) + + + def add_issue(self, source_id: str, issue: dict): + parent_id = issue.get("parent") + parent_issue = self.get_issue(parent_id) + issue: Issue = issue + issue["source"] = source_id + issue.calculate_id() + parent_issue.children.append(issue) + self.__flush() + + def update_issue(self, source_id: str, update: dict): + issue = self.get_issue(update["id"]) + source = update["source"] + changes = {} + for key, value in update.items(): + if key != "id" and key is not "source": + changes[key] = { + "old": issue[key], + "new": value, + } + issue[key] = value + issue.update_history.append(IssueUpdateHistory(source, changes)) + + self.__flush() + + +class IssueParserEnvironment(Environment): + def __init__(self, env_id: str, storage: IssueStorage) -> None: + super().__init__(env_id) + self.storage = storage + + update_description = '''update issue with email object''' + + update_param = { + "source_id": "update issue with which email object id", + "update_content": '''issue fileds to update, json format; + if id field exists, update the issue with the id; + if id filed not exists, create a new issue with the content; + other fileds in update_content will be updated to the issue; + ''', + } + self.add_ai_function(SimpleAIFunction("update_issue", + update_description, + self._update, + update_param)) + + async def _update(self, source_id: str, update_content: str): + update_issue = json.loads(update_content) + issue_id = update_issue.get("id") + if issue_id: + self.storage.update_issue(source_id, update_issue) + else: + self.storage.add_issue(source_id, update_issue) + + + +class IssueParser: + def __init__(self, env: KnowledgePipelineEnvironment, config: dict): + mail_path = string.Template(config["mail_path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) + issue_path = string.Template(config["issue_path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) + config["path"] = issue_path + + self.env = env + self.config = config + self.mail_storage = MailStorage(mail_path) + + root_issue = None + if "root_issue" in config: + root_issue = Issue() + root_issue.summary = config["root_issue"] + self.issue_storage = IssueStorage(issue_path, root_issue) + self.llm_env = IssueParserEnvironment("issue_parser", self.issue_storage) + + def get_path(self) -> str: + return self.config["path"] + + async def parse(self, mail_id: ObjectID) -> str: + mail_id = str(mail_id) + mail = self.mail_storage.get_mail_by_id(mail_id) + issue = self.issue_storage.get_issue_by_mail(self.mail_storage, mail) + mail_str = mail.to_prompt() + issue_str = issue.to_prompt() + + mail_desc = Mail.prompt_desc() + issue_desc = Issue.prompt_desc() + prompt = f'''I'll give a mail in json format, {mail_desc}; + and a issue in json format, {issue_desc}; + you should read this mail {mail_str}, see if this mail associated with this issue {issue_str}; + if this mail is about a new child issue of this issue, create a new issue with this mail, fill param update_content's summary field will mail content, set parent field with id of this issue; + if this mail will update this issue, set id filed to this issue, fill update_content param with new summary and new state with this mail content; + then you should call update_issue function with source_id set to this mail id, and update_content in json format; + if this mail is not associated with issue, you should ignore this mail without an function call; + ''' + + llm_result = await BaseAIAgent.do_llm_complection(self.llm_env, AgentPrompt(prompt), AgentMsg(), "gpt-4", 16000) + return "update issue" + diff --git a/src/component/mail_environment/local.py b/src/component/mail_environment/local.py new file mode 100644 index 0000000..07b038d --- /dev/null +++ b/src/component/mail_environment/local.py @@ -0,0 +1,34 @@ +import os +import logging +import json +import string +from knowledge import * +from aios_kernel.storage import AIStorage +from .mail import Mail, MailStorage + + +class LocalEmail: + def __init__(self, env: KnowledgePipelineEnvironment, config:dict): + self.config = config + self.env = env + path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) + self.mail_storage = MailStorage(path, config.get("watch")) + + async def next(self): + while True: + parsed = None + journals = self.env.journal.latest_journals(1) + if len(journals) == 1: + latest_journal = journals[0] + if latest_journal.is_finish(): + yield None + continue + parsed = str(latest_journal.get_object_id()) + + mail_id = self.mail_storage.next_mail_id(parsed) + if mail_id is None: + yield (None, None) + else: + yield (mail_id, str(mail_id)) + + \ No newline at end of file diff --git a/src/component/mail_environment/mail.py b/src/component/mail_environment/mail.py new file mode 100644 index 0000000..ff5ef15 --- /dev/null +++ b/src/component/mail_environment/mail.py @@ -0,0 +1,265 @@ +import asyncio +import json +import mailparser +import base64 +import requests +import datetime +from bs4 import BeautifulSoup +import sqlite3 +import html2text +from knowledge import * + +class Mail: + def __init__(self, **kwargs) -> None: + self.from_addr = kwargs.get("From") + self.to_addr = kwargs.get("To") + self.subject = kwargs.get("Subject") + self.date = kwargs.get("Date") + self.bcc = kwargs.get("BCC") + self.cc = kwargs.get("CC") + self.reply_to = None + self.id: str = None + self.content: str = None + + def to_prompt(self) -> str: + prompt = { + "id": self.id, + "subject": self.subject, + "from": self.from_addr, + "date": self.date, + "content": self.content + } + return json.dumps(prompt) + + @classmethod + def prompt_desc(cls) -> dict: + return '''a mail contains following fileds: { + id: a guid string to identify a mail + subject: subject of this mail + from: sender address of this mail + date: date of this mail + content: content of this mail + } + ''' + + def get_date(self) -> datetime.datetime: + datetime.datetime.strptime(self.date, "%Y-%m-%d %H:%M") + + def calculate_id(self) -> str: + desc = { + "from_addr": self.from_addr, + "to_addr": self.to_addr, + "subject": self.subject, + "date": self.date, + "content": self.content, + "reply_to": self.reply_to + } + id = str(KnowledgeObject(ObjectType.Email, desc).calculate_id()) + self.id = id + return id + +class MailStorage: + def __init__(self, root, watch=False): + self.root = root + if not os.path.exists(root): + os.makedirs(root) + db_file = os.path.join(root, "mail.db") + + self.conn = sqlite3.connect(db_file) + cursor = self.conn.cursor() + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS mails ( + uid INTEGER PRIMARY KEY, + object_id TEXT, + date DATETIME, + from_addr TEXT + ) + """ + ) + + if watch: + asyncio.create_task(self.watch_root()) + + def object_id_to_uid(self, object_id): + cursor = self.conn.cursor() + cursor.execute( + """ + SELECT uid FROM mails WHERE object_id = ? + """, + (object_id,), + ) + row = cursor.fetchone() + if row: + return row[0] + return None + + def uid_to_object_id(self, uid): + cursor = self.conn.cursor() + cursor.execute( + """ + SELECT object_id FROM mails WHERE uid = ? + """, + (uid,), + ) + row = cursor.fetchone() + if row: + return row[0] + return None + + def lastest_uid(self): + cursor = self.conn.cursor() + cursor.execute( + """ + SELECT uid FROM mails ORDER BY uid DESC LIMIT 1 + """ + ) + row = cursor.fetchone() + if row: + return row[0] + return None + + def lastest_mail_id(self): + cursor = self.conn.cursor() + cursor.execute( + """ + SELECT object_id FROM mails ORDER BY uid DESC LIMIT 1 + """ + ) + row = cursor.fetchone() + if row: + return row[0] + return None + + def next_mail_id(self, id): + uid = 0 if id is None else self.object_id_to_uid(id) + cursor = self.conn.cursor() + cursor.execute( + """ + SELECT object_id FROM mails WHERE uid > ? ORDER BY uid ASC LIMIT 1 + """, + (uid,), + ) + row = cursor.fetchone() + if row: + return row[0] + return None + + + + def get_mail_by_id(self, id): + uid = self.object_id_to_uid(id) + mail = Mail() + mail.id = id + mail_dir = self.mail_dir(uid) + mail_json = json.load(open(f"{mail_dir}/mail.json", "r", encoding='utf-8')) + mail.__dict__.update(mail_json) + with open(f"{mail_dir}/mail.txt", "r", encoding='utf-8') as f: + mail_content = f.read() + mail.content = mail_content + return mail + + def mail_dir(self, uid): + return os.path.join(self.root, str(uid)) + + # for debug + async def watch_root(self): + while True: + latest_uid = self.lastest_uid() + for uid in os.listdir(self.root): + mail_dir = os.path.join(self.root, uid) + if uid.isdigit() and os.path.isdir(mail_dir): + uid = int(uid) + if uid <= latest_uid: + continue + mail = Mail() + mail_json = json.load(open(f"{mail_dir}/mail.json", "r", encoding='utf-8')) + + mail.__dict__.update(mail_json) + # mail content + with open(f"{mail_dir}/mail.txt", "r", encoding='utf-8') as f: + mail_content = f.read() + mail.content = mail_content + mail.calculate_id() + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT INTO mails (uid, object_id, date, from_addr) + VALUES (?, ?, ?, ?) + """, + (uid, mail.id, mail.get_date(), mail.from_addr), + ) + self.conn.commit() + await asyncio.sleep(10) + + def download(self, uid, mail: mailparser.MailParser): + mail_dir = self.mail_dir(uid) + os.makedirs(dir) + + meta = json.loads(mail.mail_json) + mail = Mail(**meta) + reply_to = meta.get("In-Reply-To") + if reply_to: + mail.reply_to = self.uid_to_object_id(reply_to) + + h = html2text.HTML2Text() + h.ignore_links = True + h.ignore_images = True + mail_content = h.handle(mail.body) + mail.content = mail_content + + mail.calculate_id() + del mail.content + json.dump(mail.__dict__, open(f"{mail_dir}/mail.json", "w", encoding='utf-8')) + + # save mail content + with open(f"{mail_dir}/mail.txt", "w", encoding='utf-8') as f: + f.write(mail_content) + + for attachment in mail.attachments: + if attachment['mail_content_type'] in ['image/png', 'image/jpeg', 'image/gif']: + filename = attachment['filename'] + filefullname = f"{mail_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") + + # get all image urls + soup = BeautifulSoup(mail.body, '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 + + 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(mail_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}') + + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT INTO mails (uid, object_id, date, from_addr) + VALUES (?, ?, ?, ?) + """, + (uid, mail.id, mail.date, mail.from_addr), + ) + + + \ No newline at end of file diff --git a/src/component/mail_environment/spider.py b/src/component/mail_environment/spider.py new file mode 100644 index 0000000..2502468 --- /dev/null +++ b/src/component/mail_environment/spider.py @@ -0,0 +1,53 @@ +import os +import logging +import json +import imaplib +import mailparser +from knowledge import * +from aios_kernel.storage import AIStorage + + +class EmailSpider: + def __init__(self, env: KnowledgePipelineEnvironment, config:dict): + self.config = config + self.env = env + self.env.get_logger().info(f"read email config from {self.config.get('imap_server')}") + self.client = imaplib.IMAP4_SSL( + host=self.config.get('imap_server'), + port=self.config.get('imap_port') + ) + self.client.login(self.config.get('address'), self.config.get('password')) + self.mail_local_root = os.path.join(self.env.pipeline_path, self.config.get("address")) + os.makedirs(self.mail_local_root) + + async def next(self): + while True: + _, data = self.client.uid('search', None, "ALL") + uid_list = data[0].split() + if uid_list.len() == 0: + yield (None, None) + continue + + journals = self.env.journal.latest_journals(1) + from_uid = 0 + if len(journals) == 1: + latest_journal = journals[0] + if latest_journal.is_finish(): + yield None + continue + from_uid = int(latest_journal.get_input()) + if int.from_bytes(uid_list[-1]) <= from_uid: + yield (None, None) + continue + + for uid in uid_list: + _uid = int.from_bytes(uid) + if _uid > from_uid: + message_parts = "(BODY.PEEK[])" + _, email_data = self.client.uid('fetch', uid, message_parts) + mail = mailparser.parse_from_bytes(email_data[0][1]) + self.save_email(_uid, mail) + + yield (None, None) + + \ No newline at end of file diff --git a/src/knowledge/object/object.py b/src/knowledge/object/object.py index 75852c3..f34da2d 100644 --- a/src/knowledge/object/object.py +++ b/src/knowledge/object/object.py @@ -51,13 +51,13 @@ class KnowledgeObject(ABC): def get_summary(self) -> str: return self.desc.get("summary") - def get_articl_catelog(self) -> str: - assert self.object_type == ObjectType.Document - return self.desc.get("catelog") + # def get_articl_catelog(self) -> str: + # assert self.object_type == ObjectType.Document + # return self.desc.get("catelog") - def get_article_full_content(self) -> str: - assert self.object_type == ObjectType.Document - return self.body + # def get_article_full_content(self) -> str: + # assert self.object_type == ObjectType.Document + # return self.body def calculate_id(self): # Convert the object_type and desc to string and compute the SHA256 hash @@ -73,6 +73,6 @@ class KnowledgeObject(ABC): def encode(self) -> bytes: return pickle.dumps(self) - @staticmethod - def decode(data: bytes) -> "ImageObject": - return pickle.loads(data) + # @staticmethod + # def decode(data: bytes) -> "ImageObject": + # return pickle.loads(data) diff --git a/src/knowledge/object/object_id.py b/src/knowledge/object/object_id.py index 46955f4..3a07b2b 100644 --- a/src/knowledge/object/object_id.py +++ b/src/knowledge/object/object_id.py @@ -13,6 +13,17 @@ class ObjectType(IntEnum): Document = 103 RichText = 104 Email = 105 + UserDef = 200 + + def is_user_def(self) -> bool: + return self.value >= 200 + + def get_user_def_type_code(self): + return (self.value - 200) if self.is_user_def() else None + + @classmethod + def from_user_def_type_code(value): + return value + 200 # define a object ID class to identify a object diff --git a/src/knowledge/pipeline.py b/src/knowledge/pipeline.py index 62f246a..b902225 100644 --- a/src/knowledge/pipeline.py +++ b/src/knowledge/pipeline.py @@ -14,6 +14,9 @@ class KnowledgePipelineJournal: def is_finish(self) -> bool: return self.object_id is None + + def get_object_id(self) -> ObjectID: + return self.object_id def get_input(self) -> str: return self.input diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 11c5036..ccbce5a 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -45,7 +45,6 @@ shell_style = Style.from_dict({ 'error': '#8F0000 bold' }) - class AIOS_Shell: def __init__(self,username:str) -> None: self.username = username From a63e9b6745227b1eab0f6996be7b5bd22cdb4683 Mon Sep 17 00:00:00 2001 From: tsukasa Date: Mon, 20 Nov 2023 22:01:18 +0800 Subject: [PATCH 3/4] read mail with issue tree pipeline works --- .../Mail/Issue/pipeline.toml | 6 +- src/aios_kernel/agent.py | 20 +- src/aios_kernel/agent_base.py | 22 +- src/aios_kernel/compute_kernel.py | 10 +- src/aios_kernel/open_ai_node.py | 6 +- src/component/mail_environment/1/mail.json | 6 - src/component/mail_environment/1/mail.txt | 51 ---- src/component/mail_environment/issue.py | 222 +++++++++++++----- src/component/mail_environment/local.py | 5 +- src/component/mail_environment/mail.py | 1 - src/knowledge/object/object_id.py | 2 +- src/service/aios_shell/aios_shell.py | 4 +- 12 files changed, 215 insertions(+), 140 deletions(-) delete mode 100644 src/component/mail_environment/1/mail.json delete mode 100644 src/component/mail_environment/1/mail.txt diff --git a/rootfs/knowledge_pipelines/Mail/Issue/pipeline.toml b/rootfs/knowledge_pipelines/Mail/Issue/pipeline.toml index d909ba9..4cb377d 100644 --- a/rootfs/knowledge_pipelines/Mail/Issue/pipeline.toml +++ b/rootfs/knowledge_pipelines/Mail/Issue/pipeline.toml @@ -5,5 +5,9 @@ input.params.watch = true parser.module = "parser.py" parser.params.mail_path = "${myai_dir}/mail" parser.params.issue_path = "${myai_dir}/mail/issue.json" -parser.params.root_issue = "巴克云公司推进中的项目" +[parser.params.root_issue] +summary = "巴克云公司推进中的项目" +[[parser.params.root_issue.children]] +summary = "去中心存储项目DMC" + diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 82c9eb0..095410d 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -14,6 +14,7 @@ import sys from .agent_base import AgentMsg, AgentMsgStatus, AgentMsgType, FunctionItem, LLMResult, AgentPrompt, AgentReport, \ AgentTodo, AgentTodoResult, AgentWorkLog, BaseAIAgent + from .chatsession import AIChatSession from .compute_task import ComputeTaskResult,ComputeTaskResultCode from .ai_function import AIFunction @@ -287,6 +288,7 @@ class AIAgent(BaseAIAgent): return None + def _get_inner_functions(self) -> dict: if self.owner_env is None: return None,0 @@ -357,6 +359,7 @@ class AIAgent(BaseAIAgent): else: return task_result + def get_agent_prompt(self) -> AgentPrompt: return self.agent_prompt @@ -520,7 +523,7 @@ class AIAgent(BaseAIAgent): if todo_count > 0: have_known_info = True known_info_str += f"## todo\n{todos_str}\n" - inner_functions,function_token_len = self._get_inner_functions() + inner_functions,function_token_len = BaseAIAgent.get_inner_functions(self.owner_env) system_prompt_len = prompt.get_prompt_token_len() input_len = len(msg.body) if msg.msg_type == AgentMsgType.TYPE_GROUPMSG: @@ -540,7 +543,7 @@ class AIAgent(BaseAIAgent): logger.debug(f"Agent {self.agent_id} do llm token static system:{system_prompt_len},function:{function_token_len},history:{history_token_len},input:{input_len}, totoal prompt:{system_prompt_len + function_token_len + history_token_len} ") #task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) - task_result = await self._do_llm_complection(prompt,inner_functions,msg) + task_result = await self._do_llm_complection(prompt,msg,inner_functions=inner_functions) if task_result.result_code != ComputeTaskResultCode.OK: error_resp = msg.create_error_resp(task_result.error_str) return error_resp @@ -778,9 +781,9 @@ class AIAgent(BaseAIAgent): todo_tree = workspace.get_todo_tree("/") prompt.append(AgentPrompt(todo_tree)) - inner_functions,function_token_len = self._get_inner_functions() + inner_functions,_ = BaseAIAgent.get_inner_functions(self.owner_env) - task_result:ComputeTaskResult = await self._do_llm_complection(prompt,inner_functions) + task_result:ComputeTaskResult = await self._do_llm_complection(prompt,inner_functions=inner_functions) if task_result.result_code != ComputeTaskResultCode.OK: logger.error(f"_llm_review_todos compute error:{task_result.error_str}") return @@ -897,7 +900,8 @@ class AIAgent(BaseAIAgent): prompt.append(todo.detail) prompt.append(todo.result) - task_result:ComputeTaskResult = await self._do_llm_complection(prompt,workspace.get_inner_functions(),None,True) + inner_functions,_ = BaseAIAgent.get_inner_functions(workspace) + task_result:ComputeTaskResult = await self._do_llm_complection(prompt,inner_functions=inner_functions,is_json_resp=True) if task_result.result_code != ComputeTaskResultCode.OK: logger.error(f"_llm_check_todo compute error:{task_result.error_str}") @@ -1058,7 +1062,7 @@ class AIAgent(BaseAIAgent): prompt.append(content_prompt) env_functions = None #env_functions,function_len = workspace.get_knowledge_base_ai_functions() - task_result:ComputeTaskResult = await self._do_llm_complection(prompt,env_functions,None,True) + task_result:ComputeTaskResult = await self._do_llm_complection(prompt,is_json_resp=True) if task_result.result_code != ComputeTaskResultCode.OK: result_obj = {} result_obj["error_str"] = task_result.error_str @@ -1091,9 +1095,8 @@ class AIAgent(BaseAIAgent): prompt.append(known_info_prompt) content_prompt = AgentPrompt(part_content) prompt.append(content_prompt) - env_functions = None #env_functions,function_len = workspace.get_knowledge_base_ai_functions() - task_result:ComputeTaskResult = await self._do_llm_complection(prompt,env_functions,None,True) + task_result:ComputeTaskResult = await self._do_llm_complection(prompt,is_json_resp=True) if task_result.result_code != ComputeTaskResultCode.OK: result_obj = {} result_obj["error_str"] = task_result.error_str @@ -1222,6 +1225,7 @@ class AIAgent(BaseAIAgent): return task_result + def need_work(self) -> bool: if self.do_prompt is not None: return True diff --git a/src/aios_kernel/agent_base.py b/src/aios_kernel/agent_base.py index 60c5fe6..75d06b2 100644 --- a/src/aios_kernel/agent_base.py +++ b/src/aios_kernel/agent_base.py @@ -600,7 +600,7 @@ class BaseAIAgent: pass @classmethod - def _get_inner_functions(cls, env:Environment) -> (dict,int): + def get_inner_functions(cls, env:Environment) -> (dict,int): if env is None: return None,0 @@ -624,18 +624,24 @@ class BaseAIAgent: @classmethod async def do_llm_complection( cls, - env:Environment, prompt:AgentPrompt, - org_msg:AgentMsg, llm_model_name:str, - max_token_size:int + max_token_size:int, + org_msg:AgentMsg=None, + env:Environment=None, + inner_functions=None, + is_json_resp=False, ) -> ComputeTaskResult: from .compute_kernel import ComputeKernel #logger.debug(f"Agent {self.agent_id} do llm token static system:{system_prompt_len},function:{function_token_len},history:{history_token_len},input:{input_len}, totoal prompt:{system_prompt_len + function_token_len + history_token_len} ") - inner_functions,inner_functions_len = cls._get_inner_functions(env) - task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,llm_model_name,max_token_size,inner_functions) + if inner_functions is None and env is not None: + inner_functions,_ = cls.get_inner_functions(env) + if is_json_resp: + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,"json",llm_model_name,max_token_size,inner_functions,timeout=None) + else: + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,"text",llm_model_name,max_token_size,inner_functions,timeout=None) if task_result.result_code != ComputeTaskResultCode.OK: - logger.error(f"llm compute error:{task_result.error_str}") + logger.error(f"_do_llm_complection llm compute error:{task_result.error_str}") #error_resp = msg.create_error_resp(task_result.error_str) return task_result @@ -649,7 +655,7 @@ class BaseAIAgent: task_result = await cls._execute_func(env,inner_func_call_node,call_prompt,inner_functions,org_msg,llm_model_name,max_token_size) return task_result - + @classmethod async def _execute_func( cls, diff --git a/src/aios_kernel/compute_kernel.py b/src/aios_kernel/compute_kernel.py index 080d455..6174fa8 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/src/aios_kernel/compute_kernel.py @@ -127,7 +127,7 @@ class ComputeKernel: self.run(task_req) return task_req - async def _wait_task(self,task_req:ComputeTask)->ComputeTaskResult: + async def _wait_task(self,task_req:ComputeTask, timeout=60)->ComputeTaskResult: async def check_timer(): check_times = 0 while True: @@ -136,8 +136,8 @@ class ComputeKernel: if task_req.state == ComputeTaskState.ERROR: break - - if check_times >= 120: + + if timeout is not None and check_times >= timeout*2: task_req.state = ComputeTaskState.ERROR break @@ -155,9 +155,9 @@ class ComputeKernel: return time_out_result - async def do_llm_completion(self, prompt: AgentPrompt,resp_mode:str="text", mode_name: Optional[str] = None, max_token: int = 0, inner_functions = None) -> str: + async def do_llm_completion(self, prompt: AgentPrompt,resp_mode:str="text", mode_name: Optional[str]=None, max_token:int=0, inner_functions=None, timeout=60) -> str: task_req = self.llm_completion(prompt, resp_mode,mode_name, max_token,inner_functions) - return await self._wait_task(task_req) + return await self._wait_task(task_req, timeout) def text_embedding(self,input:str,model_name:Optional[str] = None): diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index f35fda7..2999a27 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -200,7 +200,7 @@ class OpenAI_ComputeNode(ComputeNode): max_token_size = 4000 result_token = max_token_size - client = AsyncOpenAI() + client = AsyncOpenAI(api_key=self.openai_api_key) try: if llm_inner_functions is None: logger.info(f"call openai {mode_name} prompts: {prompts}") @@ -215,7 +215,7 @@ class OpenAI_ComputeNode(ComputeNode): messages=prompts, response_format = response_format, functions=llm_inner_functions, - #max_tokens=result_token, + max_tokens=result_token, ) # TODO: add temperature to task params? except Exception as e: logger.error(f"openai run LLM_COMPLETION task error: {e}") @@ -267,8 +267,8 @@ class OpenAI_ComputeNode(ComputeNode): logger.info(f"openai_node get task: {task.display()}") result = await self._run_task(task) if result is not None: - task.state = ComputeTaskState.DONE task.result = result + task.state = ComputeTaskState.DONE asyncio.create_task(_run_task_loop()) diff --git a/src/component/mail_environment/1/mail.json b/src/component/mail_environment/1/mail.json deleted file mode 100644 index 8a30f00..0000000 --- a/src/component/mail_environment/1/mail.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "subject": "开发dmc开源客户端", - "from_addr": "sichangjun@buckyos.com", - "to_addr": ["liuzhicong@buckyos.com"], - "date": "2023-4-10 21:00" -} \ No newline at end of file diff --git a/src/component/mail_environment/1/mail.txt b/src/component/mail_environment/1/mail.txt deleted file mode 100644 index 1abf50b..0000000 --- a/src/component/mail_environment/1/mail.txt +++ /dev/null @@ -1,51 +0,0 @@ -最小功能集：4.15准备好oktc合约和客户端工具；4.21之前完成一些矿场节点部署； -Oktc合约和rust接口 （秋总） -实现以DMC结算的bill 和 order （done） -实现链上挑战—— merkle联通证明；用户提交低深度半路径；矿工提供叶子原文和高深度半路径；（doing） -实现按照价格和质押率索引的spv节点 —— 支持用户按参数匹配下单；（doing） -实现对eth发起的http请求 auth 头认证 —— 支持https实现 链下的 write/restore/challenge 身份认证；（doing） -实现oktc client event listener的block number本地持久化；（doing） -用户端工具：面向普通存储用户，一键备份和恢复； -源数据管理服务： -注册本地路径，生成链下挑战密码本 ——随机偏移和长度QA（done） - 生成链上挑战密码本 —— merkle根和半深度茎节点（doing）； -账号服务： -添加本地路径，选择bill id发起order（done） -按参数匹配order ——依赖按参数索引的spv 服务（doing） -生成order提交到交付服务；（done） -交付服务 -注册order和关联的源数据（done） -提交merkle根（done） -监听链事件，同步order状态，向miner写入源数据；（done） -使用密码本持续发起链下挑战（done） -实现链上挑战（doing） -恢复数据到本地目录（done） -关键日志服务 -关键状态改变日志写入（doing） -关键状态日志事件监听接口（doing） -轻量命令行客户端 -服务部署脚本（doing） -传入本地文件 和 order参数一键完成备份 （doing） -一键恢复（doing） -链下挑战失效时，自动发起链上挑战 ——依赖关键日志服务（doing） -mysql实现移植到sqlite（看时间和问题多不多；undo） -矿场端工具：本版本不是发布重点；保证能在自由的节点上稳定运行即可；结构上保证了一定的伸缩性； -存储扇区管理服务：包括gateway 和 node；保证简单部署和扩展，可靠性暂时不需实现； -扇区gateway服务：注册node，分发扇区读写请求到node；（doing） -扇区node服务：注册本地目录到node，注册为可用扇区；（测试中单node可以直接接入，done） -账号服务： -添加扇区和挂单参数，生成bill（done）； -监听链事件，响应新的order转入交付服务；（done） -监听链事件，响应链上挑战转入交付服务；（doing） -交付服务： -注册order和关联的扇区；（done） -对user提供交付相关http接口： -查询miner端order状态（done） -写入源数据（done） -完成写入后准备生成merkle root准备证明（done） -恢复源数据（done） -接收链下挑战返回证明（done） -自动提交链上挑战（doing） -关键日志服务 -关键状态改变日志写入（可推迟实现；undo） -自有节点部署 \ No newline at end of file diff --git a/src/component/mail_environment/issue.py b/src/component/mail_environment/issue.py index 53ed3c6..bc505ca 100644 --- a/src/component/mail_environment/issue.py +++ b/src/component/mail_environment/issue.py @@ -14,7 +14,16 @@ class IssueUpdateHistory: def __init__(self, source: str, changes: dict) -> None: self.source = source self.changes = changes - + + def to_json_dict(self) -> dict: + return { + "source": self.source, + "changes": self.changes, + } + + @classmethod + def from_json_dict(cls, json_dict: dict) -> "IssueUpdateHistory": + return IssueUpdateHistory(json_dict["source"], json_dict["changes"]) class Issue: def __init__(self) -> None: @@ -26,20 +35,80 @@ class Issue: self.deadline: datetime = None self.update_history = [] self.children = [] - self.parent: ObjectID = None + self.parent: str = None + + def to_json_dict(self) -> dict: + json_dict = { + "id": self.id, + "summary": self.summary, + "state": self.state.name, + "create_time": self.create_time, + "deadline": self.deadline, + "source": self.source, + "parent": self.parent, + } + if self.children is not None and len(self.children) > 0: + json_dict["children"] = [] + for child in self.children: + json_dict["children"].append(child.to_json_dict()) + if self.update_history is not None and len(self.update_history) > 0: + json_dict["update_history"] = [] + for history in self.update_history: + json_dict["update_history"].append(history.to_json_dict()) + + return json_dict + + @classmethod + def from_json_dict(cls, json_dict: dict) -> "Issue": + issue = Issue() + issue.id = json_dict["id"] + issue.summary = json_dict["summary"] + issue.state = IssueState[json_dict["state"]] + issue.create_time = json_dict["create_time"] + issue.deadline = json_dict["deadline"] + issue.source = json_dict["source"] + issue.parent = json_dict["parent"] + if "children" in json_dict: + issue.children = [] + for child_json_dict in json_dict["children"]: + child = Issue.from_json_dict(child_json_dict) + issue.children.append(child) + if "update_history" in json_dict: + issue.update_history = [] + for history_json_dict in json_dict["update_history"]: + history = IssueUpdateHistory.from_json_dict(history_json_dict) + issue.update_history.append(history) + return issue + @classmethod def object_type(cls) -> ObjectType: return ObjectType.from_user_def_type_code(0) - - def to_prompt(self) -> str: - prompt = { + + def __to_desc(self, desc_list:[], recursion=None): + desc = { "id": self.id, "summary": self.summary, "state": self.state.name, - "deadline": self.deadline + "deadline": self.deadline, } - return json.dumps(prompt) + desc_list.append(desc) + if not recursion or not self.parent: + return + else: + parent = recursion.get_issue_by_id(self.parent) + parent.__to_desc(desc_list, recursion) + + def to_prompt(self, recursion=None) -> str: + desc_list = [] + self.__to_desc(desc_list, recursion) + root = desc_list.pop() + while len(desc_list) > 0: + child = desc_list.pop() + root["child"] = child + root = child + return json.dumps(root) + @classmethod def prompt_desc(cls) -> str: @@ -47,7 +116,8 @@ class Issue: id: a guid string to identify a issue summary: summary of this issue state: state of this issue, will be one of [Open, InProgress, Closed], - deadline: if issue is not closed, deadline is the time to close this issue + deadline: if issue is not closed, deadline is the time to close this issue, + children: child issues of this issue } ''' @@ -69,14 +139,27 @@ class IssueStorage: self.path = path if not os.path.exists(path): self.root = root + self.__flush() else: - self.root = json.load(open(path, "r")) + root_dict = json.load(open(path, "r", encoding="utf-8")) + self.root = Issue.from_json_dict(root_dict) def __flush(self): - json.dump(self.root, open(self.path, "w")) + json.dump(self.root.to_json_dict(), open(self.path, "w", encoding="utf-8"), ensure_ascii=False, indent=4) + + def __get_issue_by_id_in_subtree(self, root_issue: Issue, id: str): + if root_issue.id == id: + return root_issue + if root_issue.children is None or len(root_issue.children) == 0: + return None + for child_issue in root_issue.children: + this_issue = self.__get_issue_by_id_in_subtree(child_issue, id) + if this_issue is not None: + return this_issue + return None def get_issue_by_id(self, id: str) -> Issue: - self.root() + return self.__get_issue_by_id_in_subtree(self.root, id) def __get_issue_by_mail_in_subtree(self, root_issue: Issue, mail_id: str): if root_issue.source == mail_id: @@ -102,29 +185,30 @@ class IssueStorage: this_mail = mail_storage.get_mail_by_id(this_mail.reply_to) - def add_issue(self, source_id: str, issue: dict): - parent_id = issue.get("parent") - parent_issue = self.get_issue(parent_id) - issue: Issue = issue - issue["source"] = source_id + def add_issue(self, source_id: str, parent_id: str, summary: str): + parent_issue = self.get_issue_by_id(parent_id) + issue = Issue() + issue.summary = summary + issue.source = source_id + issue.parent = parent_id issue.calculate_id() parent_issue.children.append(issue) self.__flush() + return issue - def update_issue(self, source_id: str, update: dict): - issue = self.get_issue(update["id"]) - source = update["source"] + def update_issue(self, source_id: str, issue_id: str, update: dict): + issue = self.get_issue_by_id(issue_id) changes = {} for key, value in update.items(): - if key != "id" and key is not "source": - changes[key] = { - "old": issue[key], - "new": value, - } - issue[key] = value - issue.update_history.append(IssueUpdateHistory(source, changes)) + changes[key] = { + "old": issue[key], + "new": value, + } + issue.__dict__[key] = value + issue.update_history.append(IssueUpdateHistory(source_id, changes)) self.__flush() + return issue class IssueParserEnvironment(Environment): @@ -132,29 +216,37 @@ class IssueParserEnvironment(Environment): super().__init__(env_id) self.storage = storage - update_description = '''update issue with email object''' - + create_description = '''create a new issue''' + create_param = { + "mail_id": "new issue with which email object id", + "issue_id": '''new issue's parent issue id''', + "summary": '''new issue's summary''', + } + self.add_ai_function(SimpleAIFunction("create_issue", + create_description, + self._create, + create_param)) + + update_description = '''update an existing issue''' update_param = { - "source_id": "update issue with which email object id", - "update_content": '''issue fileds to update, json format; - if id field exists, update the issue with the id; - if id filed not exists, create a new issue with the content; - other fileds in update_content will be updated to the issue; - ''', + "mail_id": "update issue with which email object id", + "issue_id": '''update issue's id''', + "summary": '''issue's new summary''', } self.add_ai_function(SimpleAIFunction("update_issue", update_description, self._update, update_param)) - async def _update(self, source_id: str, update_content: str): - update_issue = json.loads(update_content) - issue_id = update_issue.get("id") - if issue_id: - self.storage.update_issue(source_id, update_issue) - else: - self.storage.add_issue(source_id, update_issue) - + async def _create(self, mail_id: str, issue_id: str, summary: str): + issue = self.storage.add_issue(mail_id, issue_id, summary) + return issue.id + + async def _update(self, mail_id: str, issue_id: str, summary: str): + update = {} + update["summary"] = summary + issue = self.storage.update_issue(mail_id, issue_id, update) + return issue.id class IssueParser: @@ -169,10 +261,30 @@ class IssueParser: root_issue = None if "root_issue" in config: - root_issue = Issue() - root_issue.summary = config["root_issue"] + root_config = config["root_issue"] + root_issue = IssueParser.__load_issue_config(root_config) + IssueParser.__calac_issue_id(root_issue) + self.issue_storage = IssueStorage(issue_path, root_issue) self.llm_env = IssueParserEnvironment("issue_parser", self.issue_storage) + + @classmethod + def __load_issue_config(cls, issue_config: dict) -> Issue: + issue = Issue() + issue.summary = issue_config["summary"] + if "children" in issue_config: + for child_config in issue_config["children"]: + child_issue = cls.__load_issue_config(child_config) + issue.children.append(child_issue) + return issue + + @classmethod + def __calac_issue_id(cls, issue: Issue): + issue_id = issue.calculate_id() + for child in issue.children: + child.parent = issue_id + cls.__calac_issue_id(child) + def get_path(self) -> str: return self.config["path"] @@ -182,19 +294,21 @@ class IssueParser: mail = self.mail_storage.get_mail_by_id(mail_id) issue = self.issue_storage.get_issue_by_mail(self.mail_storage, mail) mail_str = mail.to_prompt() - issue_str = issue.to_prompt() + issue_str = issue.to_prompt(recursion=self.issue_storage) mail_desc = Mail.prompt_desc() issue_desc = Issue.prompt_desc() - prompt = f'''I'll give a mail in json format, {mail_desc}; - and a issue in json format, {issue_desc}; - you should read this mail {mail_str}, see if this mail associated with this issue {issue_str}; - if this mail is about a new child issue of this issue, create a new issue with this mail, fill param update_content's summary field will mail content, set parent field with id of this issue; - if this mail will update this issue, set id filed to this issue, fill update_content param with new summary and new state with this mail content; - then you should call update_issue function with source_id set to this mail id, and update_content in json format; - if this mail is not associated with issue, you should ignore this mail without an function call; - ''' + prompt = AgentPrompt() + prompt.system_message = {"role": "system", "content": f''' + I'm a CEO of a company named 巴克云; You'ar my assistant, and you should help me to manage my issues. Issues is a concept in software development of this company, but I use it to manage my work. + I'll give you a mail in json format, {mail_desc}; + and a issue in json format, {issue_desc}. Read mail's fileds and issue's fileds, and decide if you should update the issue or create a new issue with this mail. + Then call the function create_issue or update_issue. + if this mail is not associated with issue, you should ignore this mail.'''} + + prompt.append(AgentPrompt(f'''Mail is {mail_str}, issue is {issue_str}. Answer me the function's return value or None if igonred. + ''')) - llm_result = await BaseAIAgent.do_llm_complection(self.llm_env, AgentPrompt(prompt), AgentMsg(), "gpt-4", 16000) + llm_result = await BaseAIAgent.do_llm_complection(prompt, "gpt-4-1106-preview", 4000, env=self.llm_env) return "update issue" diff --git a/src/component/mail_environment/local.py b/src/component/mail_environment/local.py index 07b038d..6ddbfba 100644 --- a/src/component/mail_environment/local.py +++ b/src/component/mail_environment/local.py @@ -31,4 +31,7 @@ class LocalEmail: else: yield (mail_id, str(mail_id)) - \ No newline at end of file + +class LocalEmailWithFilter: + def __init__(self, env: KnowledgePipelineEnvironment, config:dict): + pass \ No newline at end of file diff --git a/src/component/mail_environment/mail.py b/src/component/mail_environment/mail.py index ff5ef15..9f76c0b 100644 --- a/src/component/mail_environment/mail.py +++ b/src/component/mail_environment/mail.py @@ -261,5 +261,4 @@ class MailStorage: (uid, mail.id, mail.date, mail.from_addr), ) - \ No newline at end of file diff --git a/src/knowledge/object/object_id.py b/src/knowledge/object/object_id.py index 3a07b2b..6cef161 100644 --- a/src/knowledge/object/object_id.py +++ b/src/knowledge/object/object_id.py @@ -22,7 +22,7 @@ class ObjectType(IntEnum): return (self.value - 200) if self.is_user_def() else None @classmethod - def from_user_def_type_code(value): + def from_user_def_type_code(cls, value): return value + 200 diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index ccbce5a..6ecf52e 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -23,7 +23,9 @@ from prompt_toolkit.styles import Style directory = os.path.dirname(__file__) sys.path.append(directory + '/../../') - +# import os +# os.environ['HTTP_PROXY'] = '127.0.0.1:10809' +# os.environ['HTTPS_PROXY'] = '127.0.0.1:10809' import proxy from aios_kernel import * From d7cdf2576e265d342c2e6ed3da536815b8fe3b7b Mon Sep 17 00:00:00 2001 From: tsukasa Date: Mon, 20 Nov 2023 23:36:28 +0800 Subject: [PATCH 4/4] read mail with issue tree pipeline works --- src/aios_kernel/__init__.py | 2 +- src/aios_kernel/agent.py | 84 +++---------------------- src/aios_kernel/agent_base.py | 83 +++++++++--------------- src/component/mail_environment/issue.py | 4 +- 4 files changed, 40 insertions(+), 133 deletions(-) diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index a1ab85f..187ee0b 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -1,5 +1,5 @@ from .environment import Environment,EnvironmentEvent -from .agent_base import AgentMsg,AgentMsgStatus,AgentMsgType,AgentPrompt +from .agent_base import AgentMsg,AgentMsgStatus,AgentMsgType,AgentPrompt,CustomAIAgent from .chatsession import AIChatSession from .agent import AIAgent,AIAgentTemplete, BaseAIAgent from .compute_kernel import ComputeKernel,ComputeTask,ComputeTaskResult,ComputeTaskState,ComputeTaskType diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 095410d..478449f 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -316,50 +316,6 @@ class AIAgent(BaseAIAgent): return result_func,result_len - async def _execute_func(self,inner_func_call_node:dict,prompt:AgentPrompt,inner_functions,org_msg:AgentMsg=None,stack_limit = 5) -> ComputeTaskResult: - func_name = inner_func_call_node.get("name") - arguments = json.loads(inner_func_call_node.get("arguments")) - logger.info(f"llm execute inner func:{func_name} ({json.dumps(arguments)})") - - func_node : AIFunction = self.owner_env.get_ai_function(func_name) - if func_node is None: - result_str = f"execute {func_name} error,function not found" - else: - if org_msg: - ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target) - - try: - result_str:str = await func_node.execute(**arguments) - except Exception as e: - result_str = f"execute {func_name} error:{str(e)}" - logger.error(f"llm execute inner func:{func_name} error:{e}") - - - logger.info("llm execute inner func result:" + result_str) - - prompt.messages.append({"role":"function","content":result_str,"name":func_name}) - task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) - if task_result.result_code != ComputeTaskResultCode.OK: - logger.error(f"_execute_func llm compute error:{task_result.error_str}") - return task_result - - ineternal_call_record.result_str = task_result.result_str - ineternal_call_record.done_time = time.time() - if org_msg: - org_msg.inner_call_chain.append(ineternal_call_record) - - inner_func_call_node = None - if stack_limit > 0: - result_message : dict = task_result.result.get("message") - if result_message: - inner_func_call_node = result_message.get("function_call") - - if inner_func_call_node: - return await self._execute_func(inner_func_call_node,prompt,org_msg,stack_limit-1) - else: - return task_result - - def get_agent_prompt(self) -> AgentPrompt: return self.agent_prompt @@ -542,8 +498,7 @@ class AIAgent(BaseAIAgent): logger.debug(f"Agent {self.agent_id} do llm token static system:{system_prompt_len},function:{function_token_len},history:{history_token_len},input:{input_len}, totoal prompt:{system_prompt_len + function_token_len + history_token_len} ") - #task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) - task_result = await self._do_llm_complection(prompt,msg,inner_functions=inner_functions) + task_result = await self.do_llm_complection(prompt,msg,inner_functions=inner_functions) if task_result.result_code != ComputeTaskResultCode.OK: error_resp = msg.create_error_resp(task_result.error_str) return error_resp @@ -705,7 +660,7 @@ class AIAgent(BaseAIAgent): prompt.append(AgentPrompt(work_summary)) prompt.append(AgentPrompt(report.content)) - task_result:ComputeTaskResult = await self._do_llm_complection(prompt) + task_result:ComputeTaskResult = await self.do_llm_complection(prompt) if task_result.error_str is not None: logger.error(f"_llm_read_report compute error:{task_result.error_str}") @@ -783,7 +738,7 @@ class AIAgent(BaseAIAgent): prompt.append(AgentPrompt(todo_tree)) inner_functions,_ = BaseAIAgent.get_inner_functions(self.owner_env) - task_result:ComputeTaskResult = await self._do_llm_complection(prompt,inner_functions=inner_functions) + task_result:ComputeTaskResult = await self.do_llm_complection(prompt,inner_functions=inner_functions) if task_result.result_code != ComputeTaskResultCode.OK: logger.error(f"_llm_review_todos compute error:{task_result.error_str}") return @@ -851,7 +806,7 @@ class AIAgent(BaseAIAgent): #prompt.append(work_log_prompt) prompt.append(self.get_prompt_from_todo(todo)) - task_result:ComputeTaskResult = await self._do_llm_complection(prompt) + task_result:ComputeTaskResult = await self.do_llm_complection(prompt) if task_result.error_str is not None: logger.error(f"_llm_do compute error:{task_result.error_str}") result.result_code = AgentTodoResult.TODO_RESULT_CODE_LLM_ERROR @@ -901,7 +856,7 @@ class AIAgent(BaseAIAgent): prompt.append(todo.result) inner_functions,_ = BaseAIAgent.get_inner_functions(workspace) - task_result:ComputeTaskResult = await self._do_llm_complection(prompt,inner_functions=inner_functions,is_json_resp=True) + task_result:ComputeTaskResult = await self.do_llm_complection(prompt,inner_functions=inner_functions,is_json_resp=True) if task_result.result_code != ComputeTaskResultCode.OK: logger.error(f"_llm_check_todo compute error:{task_result.error_str}") @@ -1062,7 +1017,7 @@ class AIAgent(BaseAIAgent): prompt.append(content_prompt) env_functions = None #env_functions,function_len = workspace.get_knowledge_base_ai_functions() - task_result:ComputeTaskResult = await self._do_llm_complection(prompt,is_json_resp=True) + task_result:ComputeTaskResult = await self.do_llm_complection(prompt,is_json_resp=True) if task_result.result_code != ComputeTaskResultCode.OK: result_obj = {} result_obj["error_str"] = task_result.error_str @@ -1096,7 +1051,7 @@ class AIAgent(BaseAIAgent): content_prompt = AgentPrompt(part_content) prompt.append(content_prompt) #env_functions,function_len = workspace.get_knowledge_base_ai_functions() - task_result:ComputeTaskResult = await self._do_llm_complection(prompt,is_json_resp=True) + task_result:ComputeTaskResult = await self.do_llm_complection(prompt,is_json_resp=True) if task_result.result_code != ComputeTaskResultCode.OK: result_obj = {} result_obj["error_str"] = task_result.error_str @@ -1152,7 +1107,7 @@ class AIAgent(BaseAIAgent): logger.info(f"agent {self.agent_id} think session {session_id} is finished!,no more history") break #3) llm summarize chat history - task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,None) + task_result:ComputeTaskResult = await self.do_llm_complection(prompt) if task_result.result_code != ComputeTaskResultCode.OK: logger.error(f"think_chatsession llm compute error:{task_result.error_str}") break @@ -1202,29 +1157,6 @@ class AIAgent(BaseAIAgent): return known_info,result_token_len return None,0 - async def _do_llm_complection(self,prompt:AgentPrompt,inner_functions:dict=None,org_msg:AgentMsg=None,is_json_resp = False) -> ComputeTaskResult: - from .compute_kernel import ComputeKernel - #logger.debug(f"Agent {self.agent_id} do llm token static system:{system_prompt_len},function:{function_token_len},history:{history_token_len},input:{input_len}, totoal prompt:{system_prompt_len + function_token_len + history_token_len} ") - if is_json_resp: - task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,"json",self.llm_model_name,self.max_token_size,inner_functions) - else: - task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,"text",self.llm_model_name,self.max_token_size,inner_functions) - if task_result.result_code != ComputeTaskResultCode.OK: - logger.error(f"_do_llm_complection llm compute error:{task_result.error_str}") - #error_resp = msg.create_error_resp(task_result.error_str) - return task_result - - result_message = task_result.result.get("message") - inner_func_call_node = None - if result_message: - inner_func_call_node = result_message.get("function_call") - - if inner_func_call_node: - call_prompt : AgentPrompt = copy.deepcopy(prompt) - task_result = await self._execute_func(inner_func_call_node,call_prompt,inner_functions,org_msg) - - return task_result - def need_work(self) -> bool: if self.do_prompt is not None: diff --git a/src/aios_kernel/agent_base.py b/src/aios_kernel/agent_base.py index 75d06b2..f75132c 100644 --- a/src/aios_kernel/agent_base.py +++ b/src/aios_kernel/agent_base.py @@ -567,38 +567,6 @@ class BaseAIAgent(abc.ABC): def get_max_token_size(self) -> int: pass - @abstractmethod - def get_llm_learn_token_limit(self) -> int: - pass - - @abstractmethod - async def _process_msg(self,msg:AgentMsg,workspace = None) -> AgentMsg: - pass - - -class CustomAIAgent(BaseAIAgent): - def __init__(self, agent_id: str, llm_model_name: str, max_token_size: int, llm_learn_token_limit: int) -> None: - self.agent_id = agent_id - self.llm_model_name = llm_model_name - self.max_token_size = max_token_size - self.llm_learn_token_limit = llm_learn_token_limit - - def get_id(self) -> str: - return self.agent_id - - def get_llm_model_name(self) -> str: - return self.llm_model_name - - def get_max_token_size(self) -> int: - return self.max_token_size - - def get_llm_learn_token_limit(self) -> int: - return self.llm_learn_token_limit - -class BaseAIAgent: - def __init__(self) -> None: - pass - @classmethod def get_inner_functions(cls, env:Environment) -> (dict,int): if env is None: @@ -621,12 +589,9 @@ class BaseAIAgent: return result_func,result_len - @classmethod async def do_llm_complection( - cls, + self, prompt:AgentPrompt, - llm_model_name:str, - max_token_size:int, org_msg:AgentMsg=None, env:Environment=None, inner_functions=None, @@ -635,11 +600,11 @@ class BaseAIAgent: from .compute_kernel import ComputeKernel #logger.debug(f"Agent {self.agent_id} do llm token static system:{system_prompt_len},function:{function_token_len},history:{history_token_len},input:{input_len}, totoal prompt:{system_prompt_len + function_token_len + history_token_len} ") if inner_functions is None and env is not None: - inner_functions,_ = cls.get_inner_functions(env) + inner_functions,_ = BaseAIAgent.get_inner_functions(env) if is_json_resp: - task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,"json",llm_model_name,max_token_size,inner_functions,timeout=None) + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,resp_mode="json",mode_name=self.get_llm_model_name(),max_token=self.get_max_token_size(),inner_functions=inner_functions,timeout=None) else: - task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,"text",llm_model_name,max_token_size,inner_functions,timeout=None) + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,resp_mode="text",mode_name=self.get_llm_model_name(),max_token=self.get_max_token_size(),inner_functions=inner_functions,timeout=None) if task_result.result_code != ComputeTaskResultCode.OK: logger.error(f"_do_llm_complection llm compute error:{task_result.error_str}") #error_resp = msg.create_error_resp(task_result.error_str) @@ -652,20 +617,17 @@ class BaseAIAgent: if inner_func_call_node: call_prompt : AgentPrompt = copy.deepcopy(prompt) - task_result = await cls._execute_func(env,inner_func_call_node,call_prompt,inner_functions,org_msg,llm_model_name,max_token_size) + task_result = await self._execute_func(env,inner_func_call_node,call_prompt,inner_functions,org_msg) return task_result - @classmethod async def _execute_func( - cls, + self, env: Environment, inner_func_call_node: dict, prompt: AgentPrompt, inner_functions: dict, org_msg:AgentMsg, - llm_model_name:str, - max_token_size:int, stack_limit = 5 ) -> ComputeTaskResult: from .compute_kernel import ComputeKernel @@ -677,9 +639,6 @@ class BaseAIAgent: if func_node is None: result_str = f"execute {func_name} error,function not found" else: - if org_msg: - ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target) - try: result_str:str = await func_node.execute(**arguments) except Exception as e: @@ -690,15 +649,16 @@ class BaseAIAgent: logger.info("llm execute inner func result:" + result_str) prompt.messages.append({"role":"function","content":result_str,"name":func_name}) - task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,llm_model_name,max_token_size,inner_functions) + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,mode_name=self.get_llm_model_name(),max_token=self.get_max_token_size(),inner_functions=inner_functions) if task_result.result_code != ComputeTaskResultCode.OK: logger.error(f"llm compute error:{task_result.error_str}") return task_result - - ineternal_call_record.result_str = task_result.result_str - ineternal_call_record.done_time = time.time() + if org_msg: - org_msg.inner_call_chain.append(ineternal_call_record) + internal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target) + internal_call_record.result_str = task_result.result_str + internal_call_record.done_time = time.time() + org_msg.inner_call_chain.append(internal_call_record) inner_func_call_node = None if stack_limit > 0: @@ -707,7 +667,22 @@ class BaseAIAgent: inner_func_call_node = result_message.get("function_call") if inner_func_call_node: - return await cls._execute_func(env,inner_func_call_node,prompt,inner_functions,org_msg,llm_model_name,max_token_size,stack_limit-1) + return await self._execute_func(env,inner_func_call_node,prompt,inner_functions,org_msg,stack_limit-1) else: return task_result ->>>>>>> 2f9cee9 (a issue parser of email) + + +class CustomAIAgent(BaseAIAgent): + def __init__(self, agent_id: str, llm_model_name: str, max_token_size: int) -> None: + self.agent_id = agent_id + self.llm_model_name = llm_model_name + self.max_token_size = max_token_size + + def get_id(self) -> str: + return self.agent_id + + def get_llm_model_name(self) -> str: + return self.llm_model_name + + def get_max_token_size(self) -> int: + return self.max_token_size \ No newline at end of file diff --git a/src/component/mail_environment/issue.py b/src/component/mail_environment/issue.py index bc505ca..246171e 100644 --- a/src/component/mail_environment/issue.py +++ b/src/component/mail_environment/issue.py @@ -1,7 +1,7 @@ # define a knowledge base class import json import string -from aios_kernel import ComputeKernel, AIStorage, Environment, SimpleAIFunction, BaseAIAgent, AgentPrompt, AgentMsg +from aios_kernel import AIStorage, Environment, SimpleAIFunction, CustomAIAgent, AgentPrompt, AgentMsg from knowledge import * from .mail import MailStorage, Mail @@ -309,6 +309,6 @@ class IssueParser: prompt.append(AgentPrompt(f'''Mail is {mail_str}, issue is {issue_str}. Answer me the function's return value or None if igonred. ''')) - llm_result = await BaseAIAgent.do_llm_complection(prompt, "gpt-4-1106-preview", 4000, env=self.llm_env) + llm_result = await CustomAIAgent("issue parser", "gpt-4-1106-preview", 4000).do_llm_complection(prompt, env=self.llm_env) return "update issue"