diff --git a/src/aios/__init__.py b/src/aios/__init__.py index ce866cf..6c638f0 100644 --- a/src/aios/__init__.py +++ b/src/aios/__init__.py @@ -2,7 +2,7 @@ from .proto.agent_msg import * from .proto.compute_task import * -from .agent.agent_base import AgentPrompt,CustomAIAgent +from .agent.agent_base import AgentPrompt,CustomAIAgent, AgentTodo from .agent.chatsession import AIChatSession from .agent.agent import AIAgent,AIAgentTemplete, BaseAIAgent from .agent.role import AIRole,AIRoleGroup @@ -16,11 +16,11 @@ from .frame.tunnel import AgentTunnel from .frame.contact_manager import ContactManager,Contact,FamilyMember from .frame.queue_compute_node import Queue_ComputeNode -from .environment.environment import Environment,EnvironmentEvent +from .environment.environment import BaseEnvironment,SimpleEnvironment,CompositeEnvironment from .environment.workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent,PaintEnvironment from .environment.text_to_speech_function import TextToSpeechFunction from .environment.image_2_text_function import Image2TextFunction -from .environment.workspace_env import ShellEnvironment,WorkspaceEnvironment +from .environment.workspace_env import ShellEnvironment,WorkspaceEnvironment,TodoListEnvironment,TodoListType from .storage.storage import ResourceLocation,AIStorage,UserConfig,UserConfigItem diff --git a/src/aios/agent/agent.py b/src/aios/agent/agent.py index 4614216..3048357 100644 --- a/src/aios/agent/agent.py +++ b/src/aios/agent/agent.py @@ -145,7 +145,6 @@ class AIAgent(BaseAIAgent): self.owner_promp_str = None self.contact_prompt_str = None self.history_len = 10 - self.read_report_prompt = None todo_prompts = {} @@ -161,11 +160,8 @@ class AIAgent(BaseAIAgent): } self.todo_prompts = todo_prompts - self.learn_token_limit = 4000 - self.chat_db = None self.unread_msg = Queue() # msg from other agent - self.owner_env : Environment = None self.owenr_bus = None self.enable_function_list = None @@ -187,7 +183,7 @@ class AIAgent(BaseAIAgent): logger.error("agent instance_id is None!") return False self.agent_id = config["instance_id"] - self.agent_workspace = WorkspaceEnvironment(self.agent_id) + self.agent_workspace = config["workspace"] if config.get("fullname") is None: logger.error(f"agent {self.agent_id} fullname is None!") @@ -233,9 +229,6 @@ class AIAgent(BaseAIAgent): if config.get("contact_prompt") is not None: self.contact_prompt_str = config["contact_prompt"] - if config.get("owner_env") is not None: - self.owner_env = config.get("owner_env") - if config.get("powerby") is not None: self.powerby = config["powerby"] @@ -276,16 +269,9 @@ class AIAgent(BaseAIAgent): def get_max_token_size(self) -> int: return self.max_token_size - def get_llm_learn_token_limit(self) -> int: - return self.learn_token_limit - - def get_learn_prompt(self) -> AgentPrompt: - return self.learn_prompt - def get_agent_role_prompt(self) -> AgentPrompt: return self.role_prompt - def _get_remote_user_prompt(self,remote_user:str) -> AgentPrompt: cm = ContactManager.get_instance() contact = cm.find_contact_by_name(remote_user) @@ -312,34 +298,6 @@ class AIAgent(BaseAIAgent): return None - - def _get_inner_functions(self) -> dict: - if self.owner_env is None: - return None,0 - - all_inner_function = self.owner_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() - if self.enable_function_list is not None: - if len(self.enable_function_list) > 0: - if func_name not in self.enable_function_list: - logger.debug(f"ageint {self.agent_id} ignore inner func:{func_name}") - continue - - 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 - def get_agent_prompt(self) -> AgentPrompt: return self.agent_prompt @@ -347,12 +305,9 @@ class AIAgent(BaseAIAgent): return self.agent_think_prompt def _format_msg_by_env_value(self,prompt:AgentPrompt): - if self.owner_env is None: - return - for msg in prompt.messages: old_content = msg.get("content") - msg["content"] = old_content.format_map(self.owner_env) + msg["content"] = old_content.format_map(self.agent_workspace) async def _handle_event(self,event): if event.type == "AgentThink": @@ -549,7 +504,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 = BaseAIAgent.get_inner_functions(self.owner_env) + inner_functions,function_token_len = BaseAIAgent.get_inner_functions(self.agent_workspace) system_prompt_len = self.token_len(prompt=prompt) input_len = len(msg.body) if msg.msg_type == AgentMsgType.TYPE_GROUPMSG: @@ -568,7 +523,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 = await self.do_llm_complection(prompt,msg, env=self.owner_env,inner_functions=inner_functions) + task_result = await self.do_llm_complection(prompt,msg, env=self.agent_workspace,inner_functions=inner_functions) if task_result.result_code != ComputeTaskResultCode.OK: error_resp = msg.create_error_resp(task_result.error_str) return error_resp @@ -771,6 +726,7 @@ class AIAgent(BaseAIAgent): case AgentTodoResult.TODO_RESULT_CODE_LLM_ERROR: continue case AgentTodoResult.TODO_RESULT_CODE_OK: + todo.result = do_result await todo_list.update_todo(todo.todo_id,AgentTodo.TODO_STATE_WAITING_CHECK) case AgentTodoResult.TODO_RESULT_CODE_EXEC_OP_ERROR: await todo_list.update_todo(todo.todo_id,AgentTodo.TODO_STATE_EXEC_FAILED) @@ -913,12 +869,12 @@ class AIAgent(BaseAIAgent): resp = await AIBus.get_default_bus().post_message(msg) logging.info(f"agent {self.agent_id} send msg to {msg.target} result:{resp}") - op_errors, have_error = await workspace.exec_op_list(llm_result.op_list, self.agent_id) + result_str, have_error = await workspace.exec_op_list(llm_result.op_list, self.agent_id) if have_error: result.result_code = AgentTodoResult.TODO_RESULT_CODE_EXEC_OP_ERROR #result.error_str = error_str return result - + result.result_str = result_str return result async def _llm_check_todo(self, todo: AgentTodo, prompt: AgentPrompt, workspace: WorkspaceEnvironment) -> AgentTodoResult: @@ -937,7 +893,7 @@ class AIAgent(BaseAIAgent): return result async def _llm_review_todo(self, todo:AgentTodo, prompt: AgentPrompt, workspace: WorkspaceEnvironment): - inner_functions,_ = BaseAIAgent.get_inner_functions(self.owner_env) + inner_functions,_ = BaseAIAgent.get_inner_functions(workspace) task_result:ComputeTaskResult = await self.do_llm_complection(prompt,inner_functions=inner_functions) if task_result.result_code != ComputeTaskResultCode.OK: diff --git a/src/aios/environment/environment.py b/src/aios/environment/environment.py index d4b426b..ac4559a 100644 --- a/src/aios/environment/environment.py +++ b/src/aios/environment/environment.py @@ -11,10 +11,9 @@ logger = logging.getLogger(__name__) class BaseEnvironment: - @abstractmethod - def get_id(self) -> str: + def __init__(self, workspace: str) -> None: pass - + # @abstractmethod # #TODO: how to use env? different env has different prompt # def get_env_prompt(self) -> str: @@ -50,14 +49,11 @@ class BaseEnvironment: # cls._all_env[env.get_id()] = env class SimpleEnvironment(BaseEnvironment): - def __init__(self, env_id: str) -> None: - self.env_id = env_id + def __init__(self, workspace: str) -> None: + super().__init__(workspace) self.functions: Dict[str,AIFunction] = {} self.operations: Dict[str,AIOperation] = {} - - def get_id(self) -> str: - return self.env_id - + def add_ai_function(self,func:AIFunction) -> None: self.functions[func.get_name()] = func @@ -89,17 +85,14 @@ class SimpleEnvironment(BaseEnvironment): class CompositeEnvironment(BaseEnvironment): - def __init__(self, env_id: str) -> None: - self.env_id = env_id - self.envs:Dict[str,BaseEnvironment] = {} + def __init__(self, workspace: str) -> None: + super().__init__(workspace) + self.envs:List[BaseEnvironment] = {} self.functions: Dict[str,AIFunction] = {} self.operations: Dict[str,AIOperation] = {} - - def get_id(self) -> str: - return self.env_id def add_env(self, env: BaseEnvironment) -> None: - self.envs[env.get_id()] = env + self.envs.append[env] functions = env.get_all_ai_functions() for func in functions: self.functions[func.get_name()] = func diff --git a/src/aios/environment/workspace_env.py b/src/aios/environment/workspace_env.py index 115e347..f464f0f 100644 --- a/src/aios/environment/workspace_env.py +++ b/src/aios/environment/workspace_env.py @@ -4,7 +4,9 @@ import json import logging import os import aiofiles -from typing import Any,List +import sqlite3 +import asyncio +from typing import Any,List,Dict import chardet from ..agent.agent_base import AgentMsg,AgentTodo,AgentPrompt,AgentTodoResult from ..agent.ai_function import AIFunction,SimpleAIFunction, SimpleAIOperation @@ -12,7 +14,6 @@ from ..storage.storage import AIStorage,ResourceLocation from .environment import SimpleEnvironment, CompositeEnvironment - logger = logging.getLogger(__name__) class TodoListType: @@ -20,12 +21,28 @@ class TodoListType: TO_LEARN = "learn" class TodoListEnvironment(SimpleEnvironment): - def __init__(self, root_path, list_type) -> None: - super.__init__(list_type) - self.root_path = os.path.join(root_path, list_type) + def __init__(self, workspace, list_type) -> None: + super().__init__(workspace) + self.root_path = os.path.join(workspace, list_type) if not os.path.exists(self.root_path): os.makedirs(self.root_path) - self.known_todo = {} + + self.db_path = os.path.join(self.root_path, "todo.db") + self.conn = None + try: + self.conn = sqlite3.connect(self.db_path) + except Exception as e: + logger.error("Error occurred while connecting to database: %s", e) + return None + + cursor = self.conn.cursor() + cursor.execute(''' + CREATE TABLE IF NOT EXISTS todo_list ( + id TEXT, + path TEXT + ) + ''') + self.conn.commit() async def create_todo(params): todoObj = AgentTodo.from_dict(params["todo"]) @@ -48,6 +65,23 @@ class TodoListEnvironment(SimpleEnvironment): func_handler=update_todo, )) + def _get_todo_path(self,todo_id:str) -> str: + cursor = self.conn.cursor() + cursor.execute(''' + SELECT path FROM todo_list WHERE id = ? + ''',(todo_id,)) + row = cursor.fetchone() + if row: + return row[0] + else: + return None + + def _save_todo_path(self,todo_id:str,path:str): + cursor = self.conn.cursor() + cursor.execute(''' + INSERT INTO todo_list (id,path) VALUES (?,?) + ''',(todo_id,path)) + self.conn.commit() # Task/todo system , create,update,delete,query async def get_todo_tree(self,path:str = None,deep:int = 4): @@ -142,7 +176,6 @@ class TodoListEnvironment(SimpleEnvironment): if not relative_path.startswith('/'): relative_path = '/' + relative_path result_todo.todo_path = relative_path - self.known_todo[result_todo.todo_id] = result_todo else: logger.error("get_todo_by_path:%s,parse failed!",path) @@ -150,18 +183,12 @@ class TodoListEnvironment(SimpleEnvironment): except Exception as e: logger.error("get_todo_by_path:%s,failed:%s",path,e) return None - - async def get_todo(self,id:str) -> AgentTodo: - return self.known_todo.get(id) + async def create_todo(self,parent_id:str,todo:AgentTodo) -> str: try: if parent_id: - if parent_id not in self.known_todo: - logger.error("create_todo failed: parent_id not found!") - return False - - parent_path = self.known_todo.get(parent_id).todo_path + parent_path = self._get_todo_path(parent_id) todo_path = f"{parent_path}/{todo.title}" else: todo_path = todo.title @@ -172,10 +199,10 @@ class TodoListEnvironment(SimpleEnvironment): detail_path = f"{dir_path}/detail" if todo.todo_path is None: todo.todo_path = todo_path + self._save_todo_path(todo.todo_id,todo_path) logger.info("create_todo %s",detail_path) async with aiofiles.open(detail_path, mode='w', encoding="utf-8") as f: await f.write(json.dumps(todo.to_dict())) - self.known_todo[todo.todo_id] = todo except Exception as e: logger.error("create_todo failed:%s",e) return str(e) @@ -184,7 +211,8 @@ class TodoListEnvironment(SimpleEnvironment): async def update_todo(self,todo_id:str,new_stat:str)->str: try: - todo : AgentTodo = self.known_todo.get(todo_id) + todo_path = self._get_todo_path(todo_id) + todo : AgentTodo = self.get_todo_by_fullpath(todo_path) if todo: todo.state = new_stat detail_path = f"{self.root_path}/{todo.todo_path}/detail" @@ -196,6 +224,20 @@ class TodoListEnvironment(SimpleEnvironment): except Exception as e: return str(e) + async def wait_todo_done(self,todo_id:str) -> AgentTodo: + todo_path = self._get_todo_path(todo_id) + async def check_done(): + while True: + todo : AgentTodo = self.get_todo_by_fullpath(todo_path) + if todo: + if todo.state == AgentTodo.TODO_STATE_DONE: + break + await asyncio.sleep(1) + + asyncio.create_task(check_done()) + return self.get_todo_by_fullpath(todo_path) + + async def append_worklog(self, todo:AgentTodo, result:AgentTodoResult): worklog = f"{self.root_path}/{todo.todo_path}/.worklog" @@ -213,10 +255,9 @@ class TodoListEnvironment(SimpleEnvironment): await f.write(json.dumps(json_obj)) class FilesystemEnvironment(SimpleEnvironment): - def __init__(self, root_path: str, env_id: str) -> None: - super().__init__(env_id) - self.root_path = root_path - + def __init__(self, workspace: str) -> None: + super().__init__(workspace) + self.root_path = workspace # if op["op"] == "create": # await self.create(op["path"],op["content"]) @@ -346,8 +387,8 @@ class FilesystemEnvironment(SimpleEnvironment): return None class ShellEnvironment(SimpleEnvironment): - def __init__(self, env_id: str) -> None: - super().__init__(env_id) + def __init__(self, workspace: str) -> None: + super().__init__(workspace) operator_param = { "command": "command will execute", @@ -381,20 +422,22 @@ class ShellEnvironment(SimpleEnvironment): class WorkspaceEnvironment(CompositeEnvironment): def __init__(self, env_id: str) -> None: - super().__init__(env_id) myai_path = AIStorage.get_instance().get_myai_dir() - self.root_path = f"{myai_path}/workspace/{env_id}" + root_path = f"{myai_path}/workspace/{env_id}" + super().__init__(root_path) + + self.root_path = root_path if not os.path.exists(self.root_path): os.makedirs() - self.todo_list = {} + self.todo_list: Dict[str, TodoListEnvironment] = {} self.todo_list[TodoListType.TO_WORK] = TodoListEnvironment(self.root_path,TodoListType.TO_WORK) self.todo_list[TodoListType.TO_LEARN] = TodoListEnvironment(self.root_path,TodoListType.TO_LEARN) # default environments in workspace self.add_env(self.todo_list[TodoListType.TO_WORK]) - self.add_env(ShellEnvironment("shell")) - self.add_env(FilesystemEnvironment(self.root_path, "fs")) + self.add_env(ShellEnvironment(self.root_path)) + self.add_env(FilesystemEnvironment(self.root_path)) def set_root_path(self,path:str): self.root_path = path diff --git a/src/component/agent_manager/agent_manager.py b/src/component/agent_manager/agent_manager.py index cb1ed03..0252d86 100644 --- a/src/component/agent_manager/agent_manager.py +++ b/src/component/agent_manager/agent_manager.py @@ -6,7 +6,7 @@ import sys import runpy from typing import Any, Callable, Dict, List, Optional, Union -from aios import AIAgent,AIAgentTemplete,AIStorage,Environment,BaseAIAgent,PackageEnv,PackageEnvManager,PackageMediaInfo,PackageInstallTask +from aios import AIAgent,AIAgentTemplete,AIStorage,BaseAIAgent,PackageEnv,PackageEnvManager,PackageMediaInfo,PackageInstallTask,WorkspaceEnvironment logger = logging.getLogger(__name__) @@ -28,6 +28,7 @@ class AgentManager: self.agent_templete_env : PackageEnv = None self.agent_env : PackageEnv = None self.db_path : str = None + self.environments: dict = {} self.loaded_agent_instance : Dict[str,BaseAIAgent] = None async def initial(self) -> None: @@ -49,6 +50,15 @@ class AgentManager: async def scan_all_agent(self)->None: pass + async def register_environment(self, env_id: str, init_env) -> None: + self.environments[env_id] = init_env + + async def init_environment(self, env_id: str, workspace: str): + if env_id not in self.environments: + logger.error(f"env {env_id} not found!") + return + + return self.environments[env_id] async def is_exist(self,agent_id:str) -> bool: the_aget = await self.get(agent_id) @@ -108,17 +118,28 @@ class AgentManager: config_data = await config_file.read() config = toml.loads(config_data) result_agent = AIAgent() - + + workspace = config.get("workspace", config.get("instance_id")) + workspace = WorkspaceEnvironment(workspace) + config["workspace"] = workspace + if "owner_env" in config: owner_env = config["owner_env"] - _, ext = os.path.splitext(owner_env) - if ext == ".py": - env_path = os.path.join(agent_media.full_path, owner_env) - owner_env = runpy.run_path(env_path)["init"]() - config["owner_env"] = owner_env + + def init_env(env_config: str): + _, ext = os.path.splitext(owner_env) + if ext == ".py": + env_path = os.path.join(agent_media.full_path, owner_env) + env = runpy.run_path(env_path)["init"](None, workspace.root_path) + else: + env = self.init_environment(env_config, workspace.root_path) + workspace.add_env(env) + + if isinstance(owner_env, list): + for env in owner_env: + init_env(env) else: - owner_env = Environment.get_env_by_id(config["owner_env"]) - config["owner_env"] = owner_env + init_env(owner_env) if result_agent.load_from_config(config) is False: logger.error(f"load agent from {agent_media} failed!") diff --git a/src/component/common_environment/local_document.py b/src/component/common_environment/local_document.py index c003b43..14d225d 100644 --- a/src/component/common_environment/local_document.py +++ b/src/component/common_environment/local_document.py @@ -1,16 +1,6 @@ -# import os -# import aiofiles -# import chardet -# import logging -# import string -# from knowledge import ImageObjectBuilder, DocumentObjectBuilder, KnowledgePipelineEnvironment, KnowledgePipelineJournal -# from aios_kernel.storage import AIStorage - - import os import aiofiles import chardet -import logging import string import sqlite3 import json @@ -18,45 +8,9 @@ import threading import logging from datetime import datetime from typing import Optional, List -from knowledge import ImageObjectBuilder, DocumentObjectBuilder, KnowledgePipelineEnvironment, KnowledgePipelineJournal -from aios_kernel import AIStorage, SimpleEnvironment +from aios import KnowledgePipelineEnvironment, AIStorage, SimpleEnvironment, TodoListEnvironment, TodoListType, AgentTodo, CustomAIAgent -class ScanLocalDocument: - 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 - - def path(self): - return self.config["path"] - - 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 ['.pdf', '.md', '.txt']: - logging.info(f"knowledge dir source found document file {file_path}") - yield (file_path, file_path) - yield (None, None) class MetaDatabase: def __init__(self,db_path:str): @@ -165,15 +119,16 @@ class MetaDatabase: return [row[0] for row in cursor.fetchall()] #metadata["summary"] - #metadata["catelogs"] + #metadata["catalogs"] #metadata["tags"] - def add_knowledge(self, doc_hash: str, title: str, metadata: dict,content:str = None,): + def add_knowledge(self, doc_hash: str, metadata: dict,content:str = None,): conn = self._get_conn() cursor = conn.cursor() create_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") summary = metadata.get("summary", "") catalogs = metadata.get("catalogs","") + title = metadata.get("title","") tags = ','.join(metadata.get("tags", [])) cursor.execute(''' @@ -184,7 +139,7 @@ class MetaDatabase: #llm_result["summary"] #llm_result["tags"] - #llm_result["catelog"] + #llm_result["catalog"] def set_knowledge_llm_result(self, doc_hash: str, llm_result: dict): conn = self._get_conn() cursor = conn.cursor() @@ -273,15 +228,20 @@ class MetaDatabase: return [row[0] for row in cursor.fetchall()] -class DocumentKnowledgeBase(SimpleEnvironment): +class LocalKnowledgeBase(SimpleEnvironment): + def __init__(self, workspace: str) -> None: + super().__init__(workspace) + self.root_path = f"{self.root_path}/knowledge" + self.meta_db = MetaDatabase(f"{self.root_path}/kb.db") + async def get_knowledege_catalog(self,path:str=None,only_dir =True,max_depth:int=5)->str: - if path: - full_path = f"{self.root_path}/knowledge/{path}" - else: - full_path = f"{self.root_path}/knowledge" - - catlogs,file_count = await self.get_directory_structure(full_path,max_depth,only_dir) - return catlogs + if path: + full_path = f"{self.root_path}/{path}" + else: + full_path = self.root_path + + catlogs,file_count = await self.get_directory_structure(full_path,max_depth,only_dir) + return catlogs async def get_directory_structure(self,root_dir, max_depth:int=4, only_dir=True, indent=1): file_count = 0 @@ -315,176 +275,16 @@ class DocumentKnowledgeBase(SimpleEnvironment): return structure_str, file_count # inner_function - async def get_knowledge(self,path:str) -> str: - full_path = f"{self.root_path}/knowledge/{path}" + async def get_knowledge_meta(self,path:str) -> str: + full_path = f"{self.root_path}/{path}" if os.islink(full_path): org_path = os.readlink(full_path) - hash = self.kb_db.get_hash_by_doc_path(org_path) + hash = self.meta_db.get_hash_by_doc_path(org_path) if hash: - return self.kb_db.get_knowledge(org_path) + return self.meta_db.get_knowledge(org_path) return "not found" - - -class ParseLocalDocument: - def _parse_pdf_bookmarks(self,bookmarks, parent:list): - - for item in bookmarks: - if isinstance(item,list): - self._parse_pdf_bookmarks(item,parent) - else: - if item.title: - new_item = {} - new_item["page"] = item.page.idnum - new_item["title"] = item.title - my_childs = [] - if item.childs: - if len(item.childs) > 0: - self._parse_pdf_bookmarks(item.childs, my_childs) - new_item["childs"] = my_childs - parent.append(new_item) - else: - logger.warning("parse pdf bookmarks failed: item.title is None!") - - return - - def _parse_pdf(self,doc_path:str): - metadata = {} - with open(doc_path, 'rb') as file: - reader = PyPDF2.PdfReader(file) - try: - doc_info = reader.metadata - if doc_info: - if doc_info.title: - metadata["title"] = doc_info.title - if doc_info.author: - metadata["authors"] = doc_info.author - except Exception as e: - logger.warn("parse pdf metadata failed:%s",e) - - dir_path = os.path.dirname(doc_path) - base_name = os.path.basename(doc_path) - text_content_path = f"{dir_path}/.{base_name}.txt" - full_text = "" - - for page in reader.pages: - text = page.extract_text() - full_text += text - with open(text_content_path, 'w', encoding='utf-8') as f: - f.write(full_text) - - try: - bookmarks = reader.outline - if bookmarks: - catalogs = [] - self._parse_pdf_bookmarks(bookmarks,catalogs) - metadata["catalogs"] = json.dumps(catalogs) - except Exception as e: - logger.warn("parse pdf bookmarks failed:%s",e) - - return metadata - - def _parse_txt(self,doc_path:str): - return {} - - def _parse_md(self,doc_path:str): - metadata = {} - cur_encode = "utf-8" - with open(doc_path,'rb') as f: - cur_encode = chardet.detect(f.read(1024))['encoding'] - - with open(doc_path, mode='r', encoding=cur_encode) as f: - content = f.read() - match = re.search(r'^# (.*)', content, re.MULTILINE) - if match: - metadata['title'] = match.group(1).strip() - md = Markdown(extensions=['toc']) - html_str = md.convert(content) - toc = md.toc - if toc: - metadata['catalogs'] = toc - - return metadata - - def _parse_document(self,doc_path:str): - hash_result = None - title = os.path.basename(doc_path) - meta_data = {} - - with open(doc_path, "rb") as f: - hash_md5 = hashlib.md5() - for chunk in iter(lambda: f.read(1024*1024), b""): - hash_md5.update(chunk) - hash_result = hash_md5.hexdigest() - try: - if doc_path.endswith(".md"): - meta_data = self._parse_md(doc_path) - elif doc_path.endswith(".pdf"): - meta_data = self._parse_pdf(doc_path) - except Exception as e: - logger.error("parse document %s failed:%s",doc_path,e) - traceback.print_exc() - - if meta_data.get("title"): - title = meta_data["title"] - logger.info("parse document %s!",doc_path) - return hash_result,title,meta_data - async def parse(self, file_path: str) -> str: - - - -# async def get_knowledege_catalog(self,path:str=None,only_dir =True,max_depth:int=5)->str: -# if path: -# full_path = f"{self.root_path}/knowledge/{path}" -# else: -# full_path = f"{self.root_path}/knowledge" - -# catlogs,file_count = await self.get_directory_structure(full_path,max_depth,only_dir) -# return catlogs - -# async def get_directory_structure(self,root_dir, max_depth:int=4, only_dir=True, indent=1): -# file_count = 0 -# structure_str = '' -# if os.path.isdir(root_dir): -# sub_files = [] -# with os.scandir(root_dir) as it: -# for entry in it: -# if entry.is_dir(): -# sub_structure, sub_count = await self.get_directory_structure(entry.path, max_depth, only_dir, indent + 1) -# if sub_structure: -# structure_str += sub_structure -# file_count += sub_count -# else: -# file_count += 1 -# sub_files.append(entry.name) - -# if only_dir is False: -# for file_name in sub_files: -# structure_str = structure_str + ' ' * (indent+1) + file_name + '\n' - -# dir_name = os.path.basename(root_dir) -# dir_info = f"{dir_name} " - - -# structure_str = ' ' * indent + dir_info + '\n' + structure_str - -# if indent - 1 >= max_depth: -# return None, file_count -# else: -# return structure_str, file_count - -# # inner_function -# async def get_knowledge(self,path:str) -> str: -# full_path = f"{self.root_path}/knowledge/{path}" -# if os.islink(full_path): -# org_path = os.readlink(full_path) -# hash = self.kb_db.get_hash_by_doc_path(org_path) -# if hash: -# return self.kb_db.get_knowledge(org_path) - -# return "not found" - async def load_knowledge_content(self,path:str,pos:int=0,length:int=None) -> str: if path.endswith("pdf"): logger.info("load_knowledge_content:pdf") @@ -506,14 +306,147 @@ class ParseLocalDocument: content = await f.read(length) return content - return "load content failed." - def _add_document_dir(self,path:str): - self.doc_dirs[path] = 0 +class ScanLocalDocument: + def __init__(self, env: KnowledgePipelineEnvironment, config): + self.env = env + workspace = string.Template(config["workspace"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) + path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) + self.knowledge_base = LocalKnowledgeBase(workspace) + self.path = path + + def _support_file(self,file_name:str) -> bool: + if file_name.startswith("."): + return False + + if file_name.endswith(".pdf"): + return True + if file_name.endswith(".md"): + return True + if file_name.endswith(".txt"): + return True + return False + + async def next(self): + while True: + for root, dirs, files in os.walk(self.path): + for file in files: + if self._support_file(file): + full_path = os.path.join(root, file) + full_path = os.path.normpath(full_path) + if self.knowledge_base.meta_db.is_doc_exist(full_path): + continue + yield(full_path, full_path) + else: + continue + yield(None, None) + + + +class ParseLocalDocument: + def __init__(self, env: KnowledgePipelineEnvironment, config): + self.env = env + workspace = string.Template(config["workspace"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) + self.todo_list = TodoListEnvironment(workspace, TodoListType.TO_LEARN) + self.knowledge_base = LocalKnowledgeBase(workspace) + self.token_limit = config["token_limit"] + + + async def parse(self, full_path: str) -> str: + file_stat = os.stat(full_path) + if file_stat.st_size < 1: + return full_path + hash, meta_data = self._parse_document(full_path) + await self._learn(meta_data, full_path) + self.knowledge_base.meta_db.add_doc(full_path,file_stat.st_size,file_stat.st_mtime,hash) + self.knowledge_base.meta_db.add_knowledge(hash,meta_data) + return full_path + + async def _get_meta_prompt(self,meta: dict,temp_meta = None,need_catalogs = False) -> str: + kb_tree = await self.knowledge_base.get_knowledege_catalog() + + known_obj = {} + title = meta.get("title") + if title: + known_obj["title"] = title + summary = meta.get("summary") + if summary: + known_obj["summary"] = summary + tags = meta.get("tags") + if tags: + known_obj["tags"] = tags + if need_catalogs: + catalogs = meta.get("catalogs") + if catalogs: + known_obj["catalogs"] = catalogs + + if temp_meta: + for key in temp_meta.keys(): + known_obj[key] = temp_meta[key] + + org_path = meta.get("full_path") + known_obj["orginal_path"] = org_path + return f"# Known information:\n## Current directory structure:\n{kb_tree}\n## Knowlege Metadata:\n{json.dumps(known_obj)}\n" + + async def _token_len(self, text: str) -> int: + return CustomAIAgent("", "gpt-4-1106-preview", self.token_limit).token_len(text=text) + + + async def _learn(self, meta:dict, full_path:str): + # Objectives: + # Obtain better titles, abstracts, table of contents (if necessary), tags + # Determine the appropriate place to put it (in line with the organization's goals) + # Known information: + # The reason why the target service's learn_prompt is being sorted + # Summary of the organization's work (if any) + # The current structure of the knowledge base (note the size control) gen_kb_tree_prompt (when empty, LLM should generate an appropriate initial directory structure) + # Original path, current title, abstract, table of contents + + # Sorting long files (general tricks) + # Indicate that the input is part of the content, let LLM generate intermediate results for the task + # Enter the content in sequence, when the last content block is input, LLM gets the result + + full_content = self.knowledge_base.load_knowledge_content(full_path) + full_content_len = self._token_len(full_content) + + if full_content_len < self.token_limit(): + # 短文章不用总结catalog + todo = AgentTodo() + todo.title = meta["title"] + meta_prompt = await self._get_meta_prompt(meta,None) + todo.detail = meta_prompt + full_content + self.todo_list.create_todo(None, todo) + await self.todo_list.wait_todo_done(todo.todo_id) + else: + logger.warning(f"llm_read_article: article {full_path} use LLM loop learn!") + pos = 0 + read_len = int(self.token_limit() * 1.2) + + temp_meta = {} + is_final = False + while pos < full_content_len: + _content = full_content[pos:pos+read_len] + part_cotent_len = len(_content) + if part_cotent_len < read_len: + # last chunk + is_final = True + part_content = f"<>\n{_content}" + else: + part_content = f"<>\n{_content}" + + pos = pos + read_len + todo = AgentTodo() + todo.title = meta["title"] + meta_prompt = await self._get_meta_prompt(meta,temp_meta) + todo.detail = meta_prompt + part_content + self.todo_list.create_todo(None, todo) + todo = await self.todo_list.wait_todo_done(todo.todo_id) + result_obj = json.loads(todo.result.result_str) + temp_meta = result_obj + if is_final: + break - def _parse_pdf_bookmarks(self,bookmarks, parent:list): - for item in bookmarks: if isinstance(item,list): self._parse_pdf_bookmarks(item,parent) @@ -608,64 +541,111 @@ class ParseLocalDocument: meta_data = self._parse_pdf(doc_path) except Exception as e: logger.error("parse document %s failed:%s",doc_path,e) - traceback.print_exc() + # traceback.print_exc() + + logger.info("parse document %s!",doc_path) + return hash_result, meta_data + + def _parse_pdf_bookmarks(self,bookmarks, parent:list): + for item in bookmarks: + if isinstance(item,list): + self._parse_pdf_bookmarks(item,parent) + else: + if item.title: + new_item = {} + new_item["page"] = item.page.idnum + new_item["title"] = item.title + my_childs = [] + if item.childs: + if len(item.childs) > 0: + self._parse_pdf_bookmarks(item.childs, my_childs) + new_item["childs"] = my_childs + parent.append(new_item) + else: + logger.warning("parse pdf bookmarks failed: item.title is None!") + + return + + def _parse_pdf(self,doc_path:str): + metadata = {} + with open(doc_path, 'rb') as file: + reader = PyPDF2.PdfReader(file) + try: + doc_info = reader.metadata + if doc_info: + if doc_info.title: + metadata["title"] = doc_info.title + if doc_info.author: + metadata["authors"] = doc_info.author + except Exception as e: + logger.warn("parse pdf metadata failed:%s",e) + + dir_path = os.path.dirname(doc_path) + base_name = os.path.basename(doc_path) + text_content_path = f"{dir_path}/.{base_name}.txt" + full_text = "" + + for page in reader.pages: + text = page.extract_text() + full_text += text + with open(text_content_path, 'w', encoding='utf-8') as f: + f.write(full_text) + + try: + bookmarks = reader.outline + if bookmarks: + catalogs = [] + self._parse_pdf_bookmarks(bookmarks,catalogs) + metadata["catalogs"] = json.dumps(catalogs) + except Exception as e: + logger.warn("parse pdf bookmarks failed:%s",e) + + return metadata + + def _parse_txt(self,doc_path:str): + return {} + + def _parse_md(self,doc_path:str): + metadata = {} + cur_encode = "utf-8" + with open(doc_path,'rb') as f: + cur_encode = chardet.detect(f.read(1024))['encoding'] + + with open(doc_path, mode='r', encoding=cur_encode) as f: + content = f.read() + match = re.search(r'^# (.*)', content, re.MULTILINE) + if match: + metadata['title'] = match.group(1).strip() + md = Markdown(extensions=['toc']) + html_str = md.convert(content) + toc = md.toc + if toc: + metadata['catalogs'] = toc + + return metadata + + def _parse_document(self,doc_path:str): + hash_result = None + title = os.path.basename(doc_path) + meta_data = {} + + with open(doc_path, "rb") as f: + hash_md5 = hashlib.md5() + for chunk in iter(lambda: f.read(1024*1024), b""): + hash_md5.update(chunk) + hash_result = hash_md5.hexdigest() + try: + if doc_path.endswith(".md"): + meta_data = self._parse_md(doc_path) + elif doc_path.endswith(".pdf"): + meta_data = self._parse_pdf(doc_path) + except Exception as e: + logger.error("parse document %s failed:%s",doc_path,e) + # traceback.print_exc() if meta_data.get("title"): title = meta_data["title"] logger.info("parse document %s!",doc_path) return hash_result,title,meta_data - - - def _support_file(self,file_name:str) -> bool: - if file_name.startswith("."): - return False - - if file_name.endswith(".pdf"): - return True - if file_name.endswith(".md"): - return True - if file_name.endswith(".txt"): - return True - return False - - def _scan_dir(self): - while True: - time.sleep(10) - for directory in self.doc_dirs.keys(): - now = time.time() - if now - self.doc_dirs[directory] > 60*15: - self.doc_dirs[directory] = time.time() - else: - continue - - for root, dirs, files in os.walk(directory): - for file in files: - if self._support_file(file): - full_path = os.path.join(root, file) - full_path = os.path.normpath(full_path) - if self.kb_db.is_doc_exist(full_path): - continue - - file_stat = os.stat(full_path) - if file_stat.st_size < 1: - continue - - if file_stat.st_size < 1024*1024*8: - #parse and insert - hash,title,meta_data = self._parse_document(full_path) - self.kb_db.add_doc(full_path,file_stat.st_size,file_stat.st_mtime,hash) - self.kb_db.add_knowledge(hash,title,meta_data) - - else: - self.kb_db.add_doc(full_path,file_stat.st_size,file_stat.st_mtime) - - def _scan_document(self): - while True: - time.sleep(10) - parse_queue = self.kb_db.get_docs_without_hash() - for doc_path in parse_queue: - hash,title,meta_data = self._parse_document(doc_path) - self.kb_db.set_doc_hash(doc_path,hash) - self.kb_db.add_knowledge(hash,title,meta_data) \ No newline at end of file diff --git a/src/component/common_environment/to_learn_parser.py b/src/component/common_environment/to_learn_parser.py deleted file mode 100644 index f90c8ac..0000000 --- a/src/component/common_environment/to_learn_parser.py +++ /dev/null @@ -1,214 +0,0 @@ - # 尝试自我学习，会主动获取、读取资料并进行整理 - # LLM的本质能力是处理海量知识，应该让LLM能基于知识把自己的工作处理的更好 - async def do_self_learn(self) -> None: - # 不同的workspace是否应该有不同的学习方法？ - workspace = self.get_workspace_by_msg(None) - hash_list = workspace.kb_db.get_knowledge_without_llm_title() - for hash in hash_list: - if self.agent_energy <= 0: - break - - knowledge = workspace.kb_db.get_knowledge(hash) - if knowledge is None: - continue - - full_path = knowledge.get("full_path") - if full_path is None: - continue - - if os.path.exists(full_path) is False: - logger.warning(f"do_self_learn: knowledge {full_path} is not exists!") - continue - - #TODO 可以用v-db 对不同目录的名字进行选择后，先进行一次快速的插入。有时间再慢慢用LLM整理 - result_obj = await self._llm_read_article(knowledge,full_path) - - #根据结果更新knowledge - if result_obj is not None: - workspace.kb_db.set_knowledge_llm_result(hash,result_obj) - # 在知识库中创建软链接 - path_list = result_obj.get("path") - new_title = result_obj.get("title") - if path_list: - for new_path in path_list: - full_new_path = f"/knowledge{new_path}/{new_title}" - await workspace.symlink(full_path,full_new_path) - logger.info(f"create soft link {full_path} -> {full_new_path}") - - - self.agent_energy -= 1 - - # match item.type(): - # case "book": - # self.llm_read_book(kb,item) - # learn_power -= 1 - # case "article": - # - # self.llm_read_article(kb,item) - # learn_power -= 1 - # case "video": - # self.llm_watch_video(kb,item) - # learn_power -= 1 - # case "audio": - # self.llm_listen_audio(kb,item) - # learn_power -= 1 - # case "code_project": - # self.llm_read_code_project(kb,item) - # learn_power -= 1 - # case "image": - # self.llm_view_image(kb,item) - # learn_power -= 1 - # case "other": - # self.llm_read_other(kb,item) - # learn_power -= 1 - # case _: - # self.llm_learn_any(kb,item) - # pass - - - async def do_blance_knowledge_base(selft): - # 整理自己的知识库(让分类更平衡，更由于自己以后的工作)，并尝试更新学习目标 - current_path = "/" - current_list = kb.get_list(current_path) - self_assessment_with_goal = self.get_self_assessment_with_goal() - learn_goal = {} - - - llm_blance_knowledge_base(current_path,current_list,self_assessment_with_goal,learn_goal,learn_power) - - # 主动学习 - # 方法目前只有使用搜索引擎一种？ - for goal in learn_goal.items(): - self.llm_learn_with_search_engine(kb,goal,learn_power) - if learn_power <= 0: - break - - - def parser_learn_llm_result(self,llm_result:LLMResult): - pass - - async def gen_known_info_for_knowledge_prompt(self,knowledge_item:dict,temp_meta = None,need_catalogs = False) -> AgentPrompt: - workspace =self.get_workspace_by_msg(None) - kb_tree = await workspace.get_knowledege_catalog() - - - known_obj = {} - title = knowledge_item.get("title") - if title: - known_obj["title"] = title - summary = knowledge_item.get("summary") - if summary: - known_obj["summary"] = summary - tags = knowledge_item.get("tags") - if tags: - known_obj["tags"] = tags - if need_catalogs: - catalogs = knowledge_item.get("catalogs") - if catalogs: - known_obj["catalogs"] = catalogs - - if temp_meta: - for key in temp_meta.keys(): - known_obj[key] = temp_meta[key] - - org_path = knowledge_item.get("full_path") - known_obj["orginal_path"] = org_path - know_info_str = f"# Known information:\n## Current directory structure:\n{kb_tree}\n## Knowlege Metadata:\n{json.dumps(known_obj)}\n" - return AgentPrompt(know_info_str) - - async def _llm_read_article(self,knowledge_item:dict,full_path:str) -> ComputeTaskResult: - # Objectives: - # Obtain better titles, abstracts, table of contents (if necessary), tags - # Determine the appropriate place to put it (in line with the organization's goals) - # Known information: - # The reason why the target service's learn_prompt is being sorted - # Summary of the organization's work (if any) - # The current structure of the knowledge base (note the size control) gen_kb_tree_prompt (when empty, LLM should generate an appropriate initial directory structure) - # Original path, current title, abstract, table of contents - - # Sorting long files (general tricks) - # Indicate that the input is part of the content, let LLM generate intermediate results for the task - # Enter the content in sequence, when the last content block is input, LLM gets the result - - - #full_content = item.get_article_full_content() - workspace = self.get_workspace_by_msg(None) - full_content_len = self.token_len(full_content) - - if full_content_len < self.get_llm_learn_token_limit(): - - # 短文章不用总结catelog - #path_list,summary = llm_get_summary(summary,full_content) - #prompt = self.get_agent_role_prompt() - prompt = AgentPrompt() - prompt.append(self.get_learn_prompt()) - known_info_prompt = await self.gen_known_info_for_knowledge_prompt(knowledge_item) - prompt.append(known_info_prompt) - content_prompt = AgentPrompt(full_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,is_json_resp=True) - if task_result.result_code != ComputeTaskResultCode.OK: - result_obj = {} - result_obj["error_str"] = task_result.error_str - return result_obj - - result_obj = json.loads(task_result.result_str) - return result_obj - - else: - logger.warning(f"llm_read_article: article {full_path} use LLM loop learn!") - pos = 0 - read_len = int(self.get_llm_learn_token_limit() * 1.2) - - temp_meta_data = {} - is_final = False - while pos < str_len: - _content = full_content[pos:pos+read_len] - part_cotent_len = len(_content) - if part_cotent_len < read_len: - # last chunk - is_final = True - part_content = f"<>\n{_content}" - else: - part_content = f"<>\n{_content}" - - pos = pos + read_len - prompt = AgentPrompt() - prompt.append(self.get_learn_prompt()) - known_info_prompt = await self.gen_known_info_for_knowledge_prompt(knowledge_item,temp_meta_data) - prompt.append(known_info_prompt) - 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) - if task_result.result_code != ComputeTaskResultCode.OK: - result_obj = {} - result_obj["error_str"] = task_result.error_str - return result_obj - - result_obj = json.loads(task_result.result_str) - temp_meta_data = result_obj - if is_final: - return result_obj - - return None - - - async def do_self_think(self): - session_id_list = AIChatSession.list_session(self.agent_id,self.chat_db) - for session_id in session_id_list: - if self.agent_energy <= 0: - break - used_energy = await self.think_chatsession(session_id) - self.agent_energy -= used_energy - - todo_logs = await self.get_todo_logs() - for todo_log in todo_logs: - if self.agent_energy <= 0: - break - used_energy = await self.think_todo_log(todo_log) - self.agent_energy -= used_energy - - return \ No newline at end of file