Merge pull request #99 from photosssa/mvp-dev

Read Mail with knowledge pipeline and issue tree

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的状态，推理出一些汇总的结论：
++ 是不是有超期的事项
++ 事情是不是有在推进
++ 有哪些事情完成了

rootfs/agents/FindPhoto/agent.toml

@@ -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字符串。
-"""
-

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)

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)

rootfs/knowledge_pipelines/Mail/Issue/pipeline.toml

@@ -0,0 +1,13 @@
+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]
+summary = "巴克云公司推进中的项目"
+[[parser.params.root_issue.children]]
+summary = "去中心存储项目DMC"
+
+

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)

rootfs/knowledge_pipelines/Mail/Spider/pipeline.toml

@@ -0,0 +1,4 @@
+name = "Mail.Issue"
+input.module = "input.py"
+input.params.path = "${myai_dir}/data"
+

rootfs/knowledge_pipelines/pipelines.toml

@@ -1,3 +1,3 @@
 pipelines = [
-    "Mia"
+    "Mail/Issue"
 ]

src/aios_kernel/__init__.py

@@ -1,7 +1,7 @@
 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
+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

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
@@ -314,49 +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
 
@@ -520,7 +479,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:
@@ -539,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,inner_functions,msg)
         if task_result.result_code != ComputeTaskResultCode.OK:
             error_resp = msg.create_error_resp(task_result.error_str)
             return error_resp
@@ -702,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}")
@@ -778,9 +736,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
@@ -848,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
@@ -897,7 +855,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 +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,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 +1050,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
@@ -1149,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
@@ -1199,28 +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:

src/aios_kernel/agent_base.py

@@ -11,8 +11,10 @@ import shlex
 import json
 from typing import List
 
-from .ai_function import FunctionItem
+from .ai_function import FunctionItem, AIFunction
-from .compute_task import ComputeTaskResult
+from .compute_task import ComputeTaskResult,ComputeTaskResultCode
+from .environment import Environment
+
 
 logger = logging.getLogger(__name__)
 
@@ -565,21 +567,116 @@ class BaseAIAgent(abc.ABC):
     def get_max_token_size(self) -> int:
         pass
 
-    @abstractmethod
+    @classmethod
-    def get_llm_learn_token_limit(self) -> int:
+    def get_inner_functions(cls, env:Environment) -> (dict,int):
-        pass
+        if env is None:
+            return None,0
 
-    @abstractmethod
+        all_inner_function = env.get_all_ai_functions()
-    async def _process_msg(self,msg:AgentMsg,workspace = None) -> AgentMsg:
+        if all_inner_function is None:
-        pass
+            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
+
+    async def do_llm_complection(
+        self,
+        prompt:AgentPrompt,
+        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} ")
+        if inner_functions is None and env is not None:
+            inner_functions,_ = BaseAIAgent.get_inner_functions(env)
+        if is_json_resp:
+            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,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)
+            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(env,inner_func_call_node,call_prompt,inner_functions,org_msg)
+            
+        return task_result
+     
+    async def _execute_func(
+        self, 
+        env: Environment, 
+        inner_func_call_node: dict,
+        prompt: AgentPrompt, 
+        inner_functions: dict, 
+        org_msg:AgentMsg,
+        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:
+            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,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
+       
+        if org_msg:
+            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:
+            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(env,inner_func_call_node,prompt,inner_functions,org_msg,stack_limit-1)
+        else:
+            return task_result
 
 
 class CustomAIAgent(BaseAIAgent):
-    def __init__(self, agent_id: str, llm_model_name: str, max_token_size: int, llm_learn_token_limit: int) -> None:
+    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
-        self.llm_learn_token_limit = llm_learn_token_limit
 
     def get_id(self) -> str:
         return self.agent_id
@@ -589,6 +686,3 @@ class CustomAIAgent(BaseAIAgent):
 
     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

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:
@@ -137,7 +137,7 @@ 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):

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:
+    def get_env_prompt(self) -> str:
-    #    pass
+       pass
 
     def add_ai_function(self,func:AIFunction) -> None:
         if self.functions.get(func.get_name()) is not None:

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())
 

src/component/knowledge_manager/input/email.py

@@ -1,158 +0,0 @@
-
-class KnowledgeEmailSource:
-    def __init__(self, config:dict): 
-        self.config = config
-        self.config["type"] = "email"
-    
-    def id(self):
-        return self.config["address"]
-
-    @classmethod
-    def user_config_items(cls):
-        return [("address", "email address"),
-                ("password", "email password"),
-                ("imap_server", "imap server"),
-                ("imap_port", "imap port")
-                ]
-    
-    @classmethod
-    def local_root(cls):
-        user_data_dir = AIStorage.get_instance().get_myai_dir()
-        return os.path.abspath(f"{user_data_dir}/knowledge/email")    
-
-    async def run_once(self):
-        # read config from toml file
-        # and read from config config.local.toml if exists (config.local.toml is ignored by git)
-        logging.debug(f"knowledge email source {self.id()} run once")
-        filter = "ALL"  
-        self.client = self.email_client()
-        await self.read_emails(imap_keyword=filter)
-
-    def email_client(self) -> imaplib.IMAP4_SSL:
-        logging.info(f"read email config from {self.config.get('imap_server')}")
-        client = imaplib.IMAP4_SSL(
-            host=self.config.get('imap_server'), 
-            port=self.config.get('imap_port')
-        )
-        client.login(self.config.get('address'), self.config.get('password'))
-        return client
-
-    async def read_emails(self, folder: str = 'INBOX', imap_keyword: str = "UNSEEN"):
-        journal_client = KnowledgeJournalClient()
-        latest_journal = journal_client.latest_journal(self.id())
-        latest_uid = 0 if latest_journal is None else int(latest_journal.item_id)
-        self.client.select(folder)
-        _, data = self.client.uid('search', None, imap_keyword)
-        
-        # get email uid list
-        email_list = data[0].split()
-        logging.info(f"got {len(email_list)} emails")
-        journal_client = KnowledgeJournalClient()
-        for uid in email_list:
-            _uid = int.from_bytes(uid)
-            if _uid > latest_uid:
-                email_dir = self.check_email_saved(uid)
-                if email_dir is not None:
-                    logging.info(f"email uid {uid} already saved")
-                else:
-                    email_dir = self.read_and_save_email(uid)
-                    logging.info(f"email uid {uid} saved")
-                email_object = EmailObjectBuilder({}, email_dir).build()
-                await KnowledgeBase().insert_object(email_object)
-                journal_client.insert(KnowledgeJournal("email", self.id(), str(int.from_bytes(uid)), str(email_object.calculate_id())))
-
-
-    def read_and_save_email(self, uid: str) -> str:
-        message_parts = "(BODY.PEEK[])"
-        _, email_data = self.client.uid('fetch', uid, message_parts)
-        mail = mailparser.parse_from_bytes(email_data[0][1])
-        logging.info(f"got email subject [{mail.subject}]")
-        self.save_email(mail)
-        return self.get_local_dir_name(mail)
-
-    def get_local_dir_name(self, mail: mailparser.MailParser) -> str:
-        dir =  f"{self.local_root()}/{self.config.get('address')}"
-        name = f"{mail.subject}__{mail.date}"
-        name = hashlib.md5(name.encode('utf-8')).hexdigest()
-        return f"{dir}/{name}"
-
-    def check_email_saved(self, uid: str) -> str:
-        message_parts = "(BODY[HEADER])"
-        _, email_data = self.client.uid('fetch', uid, message_parts)
-        mail = mailparser.parse_from_bytes(email_data[0][1])
-        logging.info(f"[{uid}]check email subject [{mail.subject}]")
-        dir = self.get_local_dir_name(mail)
-        logging.info(f"check email saved {dir}")
-        file = f"{dir}/email.txt"
-        if os.path.exists(file):
-            return dir
-        return None
-
-    # save email attachment(images)
-    def save_email_attachment(self, mail: mailparser.MailParser, email_dir: str):
-        for attachment in mail.attachments:
-            if attachment['mail_content_type'] in ['image/png', 'image/jpeg', 'image/gif']:
-                print('current mail have image attachment')
-                img_dir = f"{email_dir}/image"
-                if not os.path.exists(img_dir):
-                    os.makedirs(img_dir)
-                filename = attachment['filename']
-                filefullname = f"{img_dir}/{filename}"
-                image_data = attachment['payload']
-                try:
-                    image_data = base64.b64decode(image_data)
-                except base64.binascii.Error:
-                    image_data = image_data.encode()
-                with open(filefullname, 'wb') as f:
-                    f.write(image_data)
-                    logging.info(f"save email image {filename} success")
-
-    # save email body images(html content)
-    def save_body_images(self, html_content: str, email_dir: str):
-        # get all image urls
-        soup = BeautifulSoup(html_content, 'html.parser')
-        img_tags = soup.find_all('img')
-        img_urls = [img['src'] for img in img_tags if 'src' in img.attrs]
-        logging.info(f'Found {len(img_urls)} images in email body')
-
-        name_count = 0
-        
-        if not os.path.exists(email_dir):
-            os.makedirs(email_dir)
-
-        for img_url in img_urls:
-            # keep the original image filename(last of url)
-            ext = img_url.split('/')[-1].split('.')[-1]
-            img_filename = os.path.join(email_dir, f"{name_count}.{ext}")
-            name_count += 1
-            # download image 
-            response = requests.get(img_url, stream=True)
-            if response.status_code == 200:
-                with open(img_filename, 'wb') as img_file:
-                    for chunk in response.iter_content(1024):
-                        img_file.write(chunk)
-                logging.info(f'Downloaded {img_url} to {img_filename}')
-            else:
-                logging.info(f'Failed to download {img_url}')
-
-    # save email content to local dir
-    def save_email(self, mail: mailparser.MailParser):
-        dir = f"{self.local_root()}/{self.config.get('address')}"
-        if not os.path.exists(dir):
-            os.makedirs(dir)
-        email_dir = self.get_local_dir_name(mail)
-        logging.info(f"save email to {email_dir}")
-        if not os.path.exists(email_dir):
-            os.makedirs(email_dir)
-        with open(f"{email_dir}/email.txt", "w", encoding='utf-8') as f:
-            # soup = BeautifulSoup(mail.body, 'html.parser')
-            f.write(mail.body)
-        with open(f"{email_dir}/meta.json", "w", encoding='utf-8') as f:
-            mail_dict = json.loads(mail.mail_json)
-            if 'body' in mail_dict:
-                del mail_dict['body']
-            json.dump(mail_dict, f, ensure_ascii=False, indent=4)
-            logging.info(f"save email meta info {f.name}")
-        
-        self.save_email_attachment(mail, email_dir)
-        self.save_body_images(mail.body, f"{email_dir}/body_image")

src/component/knowledge_manager/input/local_dir.py

@@ -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)

src/component/knowledge_manager/parser/embedding.py

@@ -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)

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
@@ -46,7 +42,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 +51,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)
@@ -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)

src/component/mail_environment/__init__.py

@@ -0,0 +1,3 @@
+from .issue import IssueParser
+from .local import LocalEmail
+from .spider import EmailSpider

src/component/mail_environment/issue.py

@@ -0,0 +1,314 @@
+# define a knowledge base class
+import json
+import string
+from aios_kernel import AIStorage, Environment, SimpleAIFunction, CustomAIAgent, 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
+
+    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:
+        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: 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_desc(self, desc_list:[], recursion=None):
+        desc = {
+            "id": self.id,
+            "summary": self.summary,
+            "state": self.state.name, 
+            "deadline": self.deadline,
+        }
+        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:
+        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,
+            children: child issues of 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
+            self.__flush()
+        else:
+            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.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:
+        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:
+            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, 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, issue_id: str, update: dict):
+        issue = self.get_issue_by_id(issue_id)
+        changes = {}
+        for key, value in update.items():
+            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):
+    def __init__(self, env_id: str, storage: IssueStorage) -> None:
+        super().__init__(env_id)
+        self.storage = storage
+
+        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 = {
+            "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 _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:
+    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_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"]
+
+    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(recursion=self.issue_storage)
+
+        mail_desc = Mail.prompt_desc()
+        issue_desc = Issue.prompt_desc()
+        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 CustomAIAgent("issue parser", "gpt-4-1106-preview", 4000).do_llm_complection(prompt, env=self.llm_env)
+        return "update issue"
+

src/component/mail_environment/local.py

@@ -0,0 +1,37 @@
+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))
+
+
+class LocalEmailWithFilter:
+    def __init__(self, env: KnowledgePipelineEnvironment, config:dict):
+        pass

src/component/mail_environment/mail.py

@@ -0,0 +1,264 @@
+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),
+        )
+
+    

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)
+
+    

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:
+    # def get_articl_catelog(self) -> str:
-        assert self.object_type == ObjectType.Document
+    #     assert self.object_type == ObjectType.Document
-        return self.desc.get("catelog")
+    #     return self.desc.get("catelog")
     
-    def get_article_full_content(self) -> str:
+    # def get_article_full_content(self) -> str:
-        assert self.object_type == ObjectType.Document
+    #     assert self.object_type == ObjectType.Document
-        return self.body
+    #     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
+    # @staticmethod
-    def decode(data: bytes) -> "ImageObject":
+    # def decode(data: bytes) -> "ImageObject":
-        return pickle.loads(data)
+    #     return pickle.loads(data)

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(cls, value):
+        return value + 200
 
 
 # define a object ID class to identify a object

src/knowledge/pipeline.py

@@ -1,6 +1,7 @@
 import datetime
 import sqlite3
 import os
+import logging
 from . import ObjectID, KnowledgeStore
 from enum import Enum
 
@@ -14,6 +15,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
     
@@ -66,6 +70,7 @@ 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
@@ -73,6 +78,9 @@ class KnowledgePipelineEnvironment:
     def get_knowledge_store(self) -> KnowledgeStore:
         return self.knowledge_store
     
+    def get_logger(self) -> logging.Logger:
+        return self.logger
+
 class KnowledgePipelineState(Enum):
     INIT = 0
     RUNNING = 1

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 *
@@ -45,7 +47,6 @@ shell_style = Style.from_dict({
     'error': '#8F0000 bold'
 })
 
-
 class AIOS_Shell:
     def __init__(self,username:str) -> None:
         self.username = username