diff --git a/rootfs/agents/Thinker/agent.toml b/rootfs/agents/Thinker/agent.toml index 05332f6..b79cc9c 100644 --- a/rootfs/agents/Thinker/agent.toml +++ b/rootfs/agents/Thinker/agent.toml @@ -16,5 +16,5 @@ You mainly use the following methods to generate summary: 4. Try to understand the attitude of different people on different topics or events 5. For the key information or TODO in the information, such as the time, place, amount and other information of the certainty, it must be stored in the summary. -You have a summary of simplicity and profound nonsense, and you don't need to have any polite words to me.Just give me a summary. +Just give me a summary without any other word. """ diff --git a/rootfs/agents/Tracy/agent.toml b/rootfs/agents/Tracy/agent.toml index 8d44476..6338d06 100644 --- a/rootfs/agents/Tracy/agent.toml +++ b/rootfs/agents/Tracy/agent.toml @@ -1,5 +1,8 @@ instance_id = "Tracy" fullname = "Tracy" +llm_model_name = "Llama-2-13b-chat" +max_token_size = 2000 + [[prompt]] role = "system" content = """ diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index 2b1b8ef..a9a5302 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -24,5 +24,6 @@ from .stability_node import Stability_ComputeNode from .local_st_compute_node import LocalSentenceTransformer_Text_ComputeNode,LocalSentenceTransformer_Image_ComputeNode from .compute_node_config import ComputeNodeConfig from .ai_function import SimpleAIFunction +from .workspace_env import WorkspaceEnvironment AIOS_Version = "0.5.2, build 2023-11-1" diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 9e2ca15..13339cb 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -485,12 +485,12 @@ class AIAgent: summary = self.llm_select_session_summary(msg,chatsession) prompt.append(AgentPrompt(summary)) - known_info_str = "# 已知信息\n" + known_info_str = "# Known information\n" have_known_info = False todos_str,todo_count = await workspace.get_todo_tree() if todo_count > 0: have_known_info = True - known_info_str += f"## 已有todo\n{todos_str}\n" + known_info_str += f"## todo\n{todos_str}\n" inner_functions,function_token_len = self._get_inner_functions() system_prompt_len = prompt.get_prompt_token_len() input_len = len(msg.body) @@ -879,40 +879,63 @@ class AIAgent: # 尝试自我学习，会主动获取、读取资料并进行整理 # LLM的本质能力是处理海量知识，应该让LLM能基于知识把自己的工作处理的更好 - def do_self_learn(self) -> None: + async def do_self_learn(self) -> None: # 不同的workspace是否应该有不同的学习方法？ - learn_power = self.get_learn_power() - kb = self.get_knowledge_base() - for item in kb.un_learn_items(): - if learn_power <= 0: + 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 - match item.type(): - case "book": - self.llm_read_book(kb,item) - learn_power -= 1 - case "article": - # 可以用vdb 对不同目录的名字进行选择后，先进行一次快速的插入。有时间再慢慢用LLM整理 - 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 - + + knowledge = workspace.kb_db.get_knowledge_by_hash(hash) + if knowledge is None: + continue + + if os.path.exists(knowledge.path) is False: + logger.warning(f"do_self_learn: knowledge {knowledge.path} is not exists!") + continue + + #TODO 可以用v-db 对不同目录的名字进行选择后，先进行一次快速的插入。有时间再慢慢用LLM整理 + llm_result = await self._llm_read_article(knowledge) + + #根据结果更新knowledge + if llm_result is not None: + workspace.kb_db.update_knowledge_by_hash(hash,llm_result) + # 在知识库中创建软链接 + + + + 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) @@ -933,48 +956,86 @@ class AIAgent: def parser_learn_llm_result(self,llm_result:LLMResult): pass - async def _llm_read_article(self,item:KnowledgeObject) -> ComputeTaskResult: - full_content = item.get_article_full_content() - full_content_len = ComputeKernel.llm_num_tokens_from_text(full_content,self.get_llm_model_name()) + async def gen_known_info_for_knowledge_prompt(self,knowledge_item:dict,need_catalogs = False) -> AgentPrompt: + #已知信息： + # 组织的工作总结（如有）待完成 + # 现在知识库的结构（注意大小控制）gen_kb_tree_prompt (当为空的时候应该让LLM生成一个合适的初始目录结构) + # 原始路径，现在标题，摘要，目录 + 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 + + org_path = knowledge_item.get("path") + known_obj["orginal_path"] = org_path + know_info_str = f"# Known information\n{json.dumps(known_obj)}\n" + return AgentPrompt(know_info_str) + + + + + + + async def _llm_read_article(self,knowledge_item:dict) -> ComputeTaskResult: + #目标： + # 得到更好的标题，摘要，目录 （如有必要）,tags + # 应放的合适的位置 （结合组织的目标） + #已知信息： + # 整理是为什么目标服务的 learn_prompt + # 组织的工作总结（如有） + # 现在知识库的结构（注意大小控制）gen_kb_tree_prompt (当为空的时候应该让LLM生成一个合适的初始目录结构) + # 原始路径，现在标题，摘要，目录 + + + # 整理长文件（通用技巧） + # 告诉输入的是部分内容，让LLM为任务产生中间结果 + # 依次输入内容，在最后一个内容块输入时，LLM得到结果 + + #full_content = item.get_article_full_content() + workspace = self.get_workspace_by_msg(None) + full_content = await workspace.load_knowledge_content(knowledge_item["hash"]) + if full_content is None: + return + 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() - learn_prompt = self.get_learn_prompt() - cotent_prompt = AgentPrompt(full_content) - prompt.append(learn_prompt) - prompt.append(cotent_prompt) - - env_functions = self._get_inner_functions() - + #prompt = self.get_agent_role_prompt() + prompt = 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 = workspace.get_knowledge_base_ai_functions() task_result:ComputeTaskResult = await self._do_llm_complection(prompt,env_functions) if task_result.result_code != ComputeTaskResultCode.OK: - return task_result - llm_result = LLMResult.from_str(task_result.result_str) - path_list,summary = self.parser_learn_llm_result(llm_result) + 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: - # 用传统方法对文章进行一些处理，目的是尽可能减少LLM调用的次数 - catelog = item.get_articl_catelog() - chunk_content = full_content.read(self.get_llm_learn_token_limit()) - summary = kb.try_get_summary(catelog,full_content) - - while chunk_content is not None: - #path_list,summarycatelog = llm_get_summary(summary,chunk_content) - #learn_prompt = self.get_learn_prompt_with_summary() - - prompt = AgentPrompt("summary") - learn_prompt.append(prompt) - prompt = AgentPrompt(chunk_content) - learn_prompt.append(prompt) - - #llm_result = self.do_llm_competion(learn_prompt) - #path_list,summary,catelog = parser_learn_llm_result(llm_result) - - #chunk_content = full_content.read(self.get_llm_learn_token_limit()) + logger.warning(f"llm_read_article: article {knowledge_item['path']} is too long,just read summary!") + result_obj = {} + result_obj["error_str"] = f"llm_read_article: article {knowledge_item['path']} is too long,just read summary!" + return result_obj - kb.insert_item(path_list,item,catelog,summary) async def do_self_think(self): session_id_list = AIChatSession.list_session(self.agent_id,self.chat_db) @@ -1043,7 +1104,7 @@ class AIAgent: known_info = "" if chatsession.summary is not None: if len(chatsession.summary) > 1: - known_info += f"## 最近交流的总结 \n {chatsession.summary}\n" + known_info += f"## Recent conversation summary \n {chatsession.summary}\n" result_token_len -= len(chatsession.summary) have_known_info = True @@ -1062,7 +1123,7 @@ class AIAgent: logger.warning(f"_get_prompt_from_session reach limit of token,just read {read_history_msg} history message.") break - known_info += f"## 最近的沟通记录 \n {histroy_str}\n" + known_info += f"## Recent conversation history \n {histroy_str}\n" if have_known_info: return known_info,result_token_len @@ -1103,6 +1164,8 @@ class AIAgent: return False def need_self_learn(self) -> bool: + if self.learn_prompt is not None: + return True return False def wake_up(self) -> None: @@ -1145,6 +1208,9 @@ class AIAgent: logger.error(f"agent {self.agent_id} on timer error:{e},{tb_str}") continue + def token_len(self,text:str) -> int: + return ComputeKernel.llm_num_tokens_from_text(text,self.get_llm_model_name()) + diff --git a/src/aios_kernel/chatsession.py b/src/aios_kernel/chatsession.py index 9d704c8..0302e12 100644 --- a/src/aios_kernel/chatsession.py +++ b/src/aios_kernel/chatsession.py @@ -12,15 +12,15 @@ from .agent_base import AgentMsgType, AgentMsg, AgentMsgStatus class ChatSessionDB: def __init__(self, db_file): """ initialize db connection """ - self.local = threading.local() self.db_file = db_file self._get_conn() def _get_conn(self): """ get db connection """ - if not hasattr(self.local, 'conn'): - self.local.conn = self._create_connection(self.db_file) - return self.local.conn + local = threading.local() + if not hasattr(local, 'conn'): + local.conn = self._create_connection(self.db_file) + return local.conn def _create_connection(self, db_file): """ create a database connection to a SQLite database """ diff --git a/src/aios_kernel/contact.py b/src/aios_kernel/contact.py index ba80df5..361c699 100644 --- a/src/aios_kernel/contact.py +++ b/src/aios_kernel/contact.py @@ -43,7 +43,8 @@ class Contact: self.active_tunnels[msg.sender] = tunnel await tunnel.post_message(msg) return None - + + logger.warn(f"contact {self.name} cann't get tunnel,post message failed!") def get_active_tunnel(self,agent_id) -> AgentTunnel: diff --git a/src/aios_kernel/simple_kb_db.py b/src/aios_kernel/simple_kb_db.py new file mode 100644 index 0000000..29480bd --- /dev/null +++ b/src/aios_kernel/simple_kb_db.py @@ -0,0 +1,225 @@ + +import sqlite3 +import json +import threading +import logging +from datetime import datetime + +from typing import Optional, List + +logger = logging.getLogger(__name__) + +class SimpleKnowledgeDB: + def __init__(self,db_path:str): + self.db_path = db_path + self._get_conn() + + def _get_conn(self): + """ get db connection """ + local = threading.local() + if not hasattr(local, 'conn'): + local.conn = self._create_connection(self.db_path) + return local.conn + + + def _create_connection(self, db_file): + """ create a database connection to a SQLite database """ + conn = None + try: + conn = sqlite3.connect(db_file) + except Exception as e: + logger.error("Error occurred while connecting to database: %s", e) + return None + + if conn: + self._create_tables(conn) + + return conn + + def _create_tables(self,conn): + cursor = conn.cursor() + cursor.execute(''' + CREATE TABLE IF NOT EXISTS documents ( + doc_path TEXT PRIMARY KEY, + length INTEGER, + last_modify TEXT, + doc_hash TEXT, + create_time TEXT + ) + ''') + cursor.execute(''' + CREATE TABLE IF NOT EXISTS knowledge ( + doc_hash TEXT PRIMARY KEY, + title TEXT, + summary TEXT, + content TEXT, + catalogs TEXT, + tags TEXT, + llm_title TEXT, + llm_summary TEXT, + create_time TEXT + ) + ''') + + cursor.execute(''' + CREATE INDEX IF NOT EXISTS idx_documents_doc_hash + ON documents (doc_hash) + ''') + + cursor.execute(''' + CREATE INDEX IF NOT EXISTS idx_knowledge_tags + ON knowledge (tags) + ''') + + conn.commit() + + def add_doc(self, doc_path: str, length: int, last_modify: str, doc_hash: Optional[str] = None): + conn = self._get_conn() + cursor = conn.cursor() + create_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + cursor.execute(''' + INSERT INTO documents (doc_path, length, last_modify, doc_hash,create_time) + VALUES (?, ?, ?, ?,?) + ''', (doc_path, length, last_modify, doc_hash,create_time)) + conn.commit() + + def is_doc_exist(self, doc_path: str) -> bool: + conn = self._get_conn() + cursor = conn.cursor() + cursor.execute(''' + SELECT doc_path + FROM documents + WHERE doc_path = ? + ''', (doc_path,)) + return len(cursor.fetchall()) > 0 + + def set_doc_hash(self, doc_path: str, doc_hash: str): + conn = self._get_conn() + cursor = conn.cursor() + cursor.execute(''' + UPDATE documents + SET doc_hash = ? + WHERE doc_path = ? + ''', (doc_hash, doc_path)) + conn.commit() + + def get_docs_without_hash(self,limit:int=1024) -> List[str]: + conn = self._get_conn() + cursor = conn.cursor() + cursor.execute(''' + SELECT doc_path + FROM documents + WHERE doc_hash IS NULL OR doc_hash = '' + ORDER BY create_time DESC + LIMIT ? + ''',(limit,)) + return [row[0] for row in cursor.fetchall()] + + #metadata["summary"] + #metadata["catelogs"] + #metadata["tags"] + def add_knowledge(self, doc_hash: str, title: 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","") + tags = ','.join(metadata.get("tags", [])) + + cursor.execute(''' + INSERT INTO knowledge (doc_hash, title , summary , catalogs , tags,create_time) + VALUES (?, ?, ?, ?, ?,?) + ''', (doc_hash, title, summary, catalogs, tags,create_time)) + conn.commit() + + #llm_result["summary"] + #llm_result["tags"] + #llm_result["catelog"] + def set_knowledge_llm_result(self, doc_hash: str, llm_result: dict): + conn = self._get_conn() + cursor = conn.cursor() + + title = llm_result.get("title", "") + summary = llm_result.get("summary", "") + catalogs = json.dumps(llm_result.get("catalogs", [])) + tags = ','.join(llm_result.get("tags", [])) + + cursor.execute(''' + UPDATE knowledge + SET llm_title = ?,llm_summary = ?, catalogs = ?, tags = ? + WHERE doc_hash = ? + ''', (title,summary, catalogs, tags, doc_hash)) + conn.commit() + + def get_hash_by_doc_path(self, doc_path: str) -> Optional[str]: + conn = self._get_conn() + cursor = conn.cursor() + cursor.execute(''' + SELECT doc_hash + FROM documents + WHERE doc_path = ? + ''', (doc_path,)) + row = cursor.fetchone() + if row is None: + return None + return row[0] + + def get_knowledge(self, doc_hash: str) -> Optional[dict]: + conn = self._get_conn() + cursor = conn.cursor() + cursor.execute(''' + SELECT title, summary, catalogs, tags, llm_title, llm_summary + FROM knowledge + WHERE doc_hash = ? + ''', (doc_hash,)) + row = cursor.fetchone() + if row is None: + return None + + # get doc path + cursor.execute(''' + SELECT doc_path + FROM documents + WHERE doc_hash = ? + ''', (doc_hash,)) + row2 = cursor.fetchone() + if row2 is None: + return None + doc_path = row2[0] + + return { + "full_path": doc_path, + "title": row[0], + "summary": row[1], + "catalogs": json.loads(row[2]), + "tags": row[3].split(","), + } + + def get_knowledge_without_llm_title(self,limit:int=16) -> List[str]: + conn = self._get_conn() + cursor = conn.cursor() + cursor.execute(''' + SELECT doc_hash + FROM knowledge + WHERE llm_title IS NULL OR llm_title = '' + ORDER BY create_time DESC + LIMIT ? + ''',(limit,)) + return [row[0] for row in cursor.fetchall()] + + def query_docs_by_tag(self, tag: str) -> List[str]: + conn = self._get_conn() + cursor = conn.cursor() + tag_json = json.dumps(tag) # 将标签转换为 JSON 字符串 + cursor.execute(''' + SELECT documents.doc_path + FROM documents + JOIN knowledge ON documents.doc_hash = knowledge.doc_hash + WHERE json_extract(knowledge.tags, '$') LIKE ? + ''', (tag)) + return [row[0] for row in cursor.fetchall()] + + def query(self,sql:str): + pass + #cursor = self.conn.cursor() diff --git a/src/aios_kernel/tg_tunnel.py b/src/aios_kernel/tg_tunnel.py index f454d41..eefc5fe 100644 --- a/src/aios_kernel/tg_tunnel.py +++ b/src/aios_kernel/tg_tunnel.py @@ -164,6 +164,7 @@ class TelegramTunnel(AgentTunnel): agent_msg.mentions = [] else: agent_msg.msg_type = AgentMsgType.TYPE_MSG + agent_msg.mentions = [] if message.entities: for entity in message.entities: diff --git a/src/aios_kernel/workflow_env.py b/src/aios_kernel/workflow_env.py index 9e0e85c..94cac90 100644 --- a/src/aios_kernel/workflow_env.py +++ b/src/aios_kernel/workflow_env.py @@ -266,9 +266,6 @@ class CalenderEnvironment(Environment): return f"Execute set_contact OK , contact {name} updated!" - - - async def start(self) -> None: if self.is_run: return diff --git a/src/aios_kernel/workspace_env.py b/src/aios_kernel/workspace_env.py index 7aedfb5..3912d37 100644 --- a/src/aios_kernel/workspace_env.py +++ b/src/aios_kernel/workspace_env.py @@ -1,5 +1,6 @@ # this env is designed for workflow owner filesystem, support file/directory operations +import hashlib import json import subprocess import logging @@ -17,10 +18,13 @@ from typing import Any,List import os import chardet +from markdown import Markdown +import PyPDF2 from .agent_base import AgentMsg,AgentTodo,AgentPrompt,AgentTodoResult from .environment import Environment,EnvironmentEvent from .ai_function import AIFunction,SimpleAIFunction from .storage import AIStorage,ResourceLocation +from .simple_kb_db import SimpleKnowledgeDB logger = logging.getLogger(__name__) @@ -33,6 +37,10 @@ class WorkspaceEnvironment(Environment): os.makedirs(self.root_path+"/todos") self.known_todo = {} + self.kb_db = SimpleKnowledgeDB(f"{self.root_path}/kb.db") + self.doc_dirs = {} + self._scan_thread = None + self._scan_dirthread = None def set_root_path(self,path:str): @@ -44,9 +52,10 @@ class WorkspaceEnvironment(Environment): def get_role_prompt(self,role_id:str) -> AgentPrompt: return None - def get_knowledge_base(self) -> str: + def get_knowledge_base(self,root_dir=None,indent=0) -> str: pass + def get_do_prompt(self,todo:AgentTodo=None)->AgentPrompt: return None @@ -94,7 +103,9 @@ class WorkspaceEnvironment(Environment): return result_str,have_error - + + # file system operation: list,read,write,delete,move,stat + # inner_function async def list(self,path:str,only_dir:bool=False) -> str: directory_path = self.root_path + path items = [] @@ -108,7 +119,8 @@ class WorkspaceEnvironment(Environment): items.append({"name": entry.name, "type": item_type}) return json.dumps(items) - + + # inner_function async def read(self,path:str) -> str: file_path = self.root_path + path cur_encode = "utf-8" @@ -119,10 +131,8 @@ class WorkspaceEnvironment(Environment): content = await f.read(2048) return content - # use diff to update large file content - async def write_diff(self,path:str,diff): - pass + # operation or inner_function (MOST IMPORTANT FUNCTION) async def write(self,path:str,content:str,is_append:bool=False) -> str: file_path = self.root_path + path try: @@ -130,24 +140,24 @@ class WorkspaceEnvironment(Environment): async with aiofiles.open(file_path, mode='a', encoding="utf-8") as f: await f.write(content) else: - async with aiofiles.open(file_path, mode='w', encoding="utf-8") as f: - await f.write(content) + if content is None: + # create dir + dir_path = self.root_path + path + os.makedirs(dir_path) + return True + else: + file_path = self.root_path + path + os.makedirs(os.path.dirname(file_path),exist_ok=True) + async with aiofiles.open(file_path, mode='w', encoding="utf-8") as f: + await f.write(content) + return True + except Exception as e: return str(e) return None - - async def create(self,path:str,content:str=None) -> bool: - if content is None: - # create dir - dir_path = self.root_path + path - os.makedirs(dir_path) - return True - else: - file_path = self.root_path + path - async with aiofiles.open(file_path, mode='w', encoding="utf-8") as f: - await f.write(content) - return True + + # operation or inner_function async def delete(self,path:str) -> str: try: file_path = self.root_path + path @@ -157,21 +167,107 @@ class WorkspaceEnvironment(Environment): return None - async def mkdir(self,path:str) -> bool: - dir_path = self.root_path + path - os.makedirs(dir_path) - return True - - async def rename(self,path:str,new_name:str) -> str: + # operation or inner_function + async def move(self,path:str,new_path:str) -> str: try: file_path = self.root_path + path - new_path = self.root_path + new_name + new_path = self.root_path + new_path os.rename(file_path,new_path) except Exception as e: return str(e) return None + + # inner_function + async def stat(self,path:str) -> str: + try: + file_path = self.root_path + path + stat = os.stat(file_path) + return json.dumps(stat) + except Exception as e: + return str(e) + # operation or inner_function + async def symlink(self,path:str,target:str) -> str: + try: + file_path = self.root_path + path + target_path = self.root_path + target + os.symlink(file_path,target_path) + except Exception as e: + return str(e) + + return None + + # TODO use diff to update large file content + async def update_by_diff(self,path:str,diff): + + pass + + # doc system （read_only,agent cann't modify doc） + + # inner_function + async def list_db(self) -> str: + pass + # inner_function + async def get_db_desc(self,db_name:str) -> str: + pass + # inner_function + async def query(self,db_name:str,sql:str) -> str: + pass + + # search (web) + # inner_function + async def google_search(self,keyword:str,opt=None) -> str: + pass + + # inner_function + async def local_search(self,keyword:str,root_path=None ,opt=None) -> str: + pass + + # inner_function, might be return a image is better + async def web_get(self,url:str) -> str: + pass + + # inner_function + async def blockchain_get(self,chainid:str,query:dict) -> str: + pass + + # code interpreter + # inner_function or operation + async def eval_code(self,pycode:str) -> str: + pass + + # operation or inner_function + async def improve_code(self,path:str): + pass + + # operation or inner_function + async def run(self,file_path:str)->str: + pass + + # operation or inner_function + async def pub_service(self,project_path:str): + pass + + # operation or inner_function + async def exec_tx(self,chain_id:str,tx:dict) -> str: + pass + + # social ability + # operation or inner_function + async def post_message(self,target:str,msg:AgentMsg,wait_time) -> AgentMsg: + pass + + # operation or inner_function + async def add_contact(self,name:str,contact_info) -> str: + pass + + # inner_function , include contact realtime info + async def get_contact(self,name_list:List[str],opt:dict) -> List: + pass + + + # Task/todo system , create,update,delete,query async def get_todo_tree(self,path:str = None,deep:int = 4): if path: directory_path = self.root_path + "/todos/" + path @@ -334,164 +430,244 @@ class WorkspaceEnvironment(Environment): json_obj["logs"] = logs await f.write(json.dumps(json_obj)) -class CodeInterpreter: - def __init__(self, language, debug_mode): - self.language = language - self.proc = None - self.active_line = None - self.debug_mode = debug_mode + async def set_wakeup_timer(self,todo_id:str,timestamp:int) -> str: + pass - def start_process(self): - start_cmd = sys.executable + " -i -q -u" - self.proc = subprocess.Popen(start_cmd.split(), - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - bufsize=0) + # knowledge base system + def get_knowledge_base_ai_functions(self): + func_result = {} - # Start watching ^ its `stdout` and `stderr` streams - threading.Thread(target=self.save_and_display_stream, - args=(self.proc.stdout, False), # Passes False to is_error_stream - daemon=True).start() - threading.Thread(target=self.save_and_display_stream, - args=(self.proc.stderr, True), # Passes True to is_error_stream - daemon=True).start() + func_result["get_knowledge_catalog"] = SimpleAIFunction("get_knowledge_catalog","get knowledge catalog in tree format", + self.get_knowledege_catalog, + {"path":f"catalog path,none is /","depth":"max depth of catalog tree,default is 4"}) + func_result["get_knowledge"] = SimpleAIFunction("get_knowledge","get knowledge metadata", + self.get_knowledge, + {"path":f"knowledge path"}) + func_result["load_knowledge_content"] = SimpleAIFunction("load_knowledge_content","load knowledge content", + self.load_knowledge_content, + {"path":f"knowledge path","pos":"start position of content","length":"length of content"}) + return func_result - def warp_code(self,pycode:str)->str: - # Add import traceback - code = "import traceback\n" + pycode - # Parse the input code into an AST - parsed_code = ast.parse(code) - # Wrap the entire code's AST in a single try-except block - try_except = ast.Try( - body=parsed_code.body, - handlers=[ - ast.ExceptHandler( - type=ast.Name(id="Exception", ctx=ast.Load()), - name=None, - body=[ - ast.Expr( - value=ast.Call( - func=ast.Attribute(value=ast.Name(id="traceback", ctx=ast.Load()), attr="print_exc", ctx=ast.Load()), - args=[], - keywords=[] - ) - ), - ] - ) - ], - orelse=[], - finalbody=[] - ) - - parsed_code.body = [try_except] - return ast.unparse(parsed_code) - - def run(self,py_code:str): - """ - Executes code. - """ - # Get code to execute - self.code = py_code - - # Start the subprocess if it hasn't been started - if not self.proc: - try: - self.start_process() - except Exception as e: - # Sometimes start_process will fail! - # Like if they don't have `node` installed or something. - - traceback_string = traceback.format_exc() - self.output = traceback_string - # Before you return, wait for the display to catch up? - # (I'm not sure why this works) - time.sleep(0.1) - - return self.output - - self.output = "" - - self.print_cmd = 'print("{}")' - code = self.warp_code(py_code) - - if self.debug_mode: - print("Running code:") - print(code) - print("---") - - self.done = threading.Event() - self.done.clear() - - # Write code to stdin of the process - try: - self.proc.stdin.write(code + "\n") - self.proc.stdin.flush() - except BrokenPipeError: - return - self.done.wait() - time.sleep(0.1) - return self.output - - def save_and_display_stream(self, stream, is_error_stream): - - for line in iter(stream.readline, ''): - if self.debug_mode: - print("Recieved output line:") - print(line) - print("---") - - line = line.strip() - if is_error_stream and "KeyboardInterrupt" in line: - raise KeyboardInterrupt - elif "END_OF_EXECUTION" in line: - self.done.set() - self.active_line = None - else: - self.output += "\n" + line - self.output = self.output.strip() - - - -class ShellEnvironment(Environment): - def __init__(self, env_id: str) -> None: - super().__init__(env_id) - - operator_param = { - "command": "command will execute", - } - self.add_ai_function(SimpleAIFunction("shell_exec", - "execute shell command in linux bash", - self.shell_exec,operator_param)) - - #run_code_param = { - # "pycode": "python code will execute", - #} - #self.add_ai_function(SimpleAIFunction("run_code", - # "execute python code", - # self.run_code,run_code_param)) - - - async def shell_exec(self,command:str) -> str: - import asyncio.subprocess - process = await asyncio.create_subprocess_shell( - command, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE - ) - stdout, stderr = await process.communicate() - returncode = process.returncode - if returncode == 0: - return f"Execute success! stdout is:\n{stdout}\n" + 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: - return f"Execute failed! stderr is:\n{stderr}\n" + 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) - async def run_code(self,pycode:str) -> str: - interpreter = CodeInterpreter("python",True) - return interpreter.run(pycode) + 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=0) -> str: + full_path = f"{self.root_path}/knowledge/{path}" + if os.islink(full_path): + org_path = os.readlink(full_path) + if full_path.endswith("pdf"): + logger.info("load_knowledge_content:pdf") + return "pdf is not support now!" + else: + async with aiofiles.open(full_path,'rb') as f: + cur_encode = chardet.detect(f.read(1024))['encoding'] + + async with aiofiles.open(full_path, mode='r', encoding=cur_encode) as f: + f.seek(pos) + content = await f.read(length) + return content + + return "load content failed." + + def _add_document_dir(self,path:str): + self.doc_dirs[path] = 0 + + def _start_scan_document(self): + if self._scan_thread is None: + self._scan_thread = threading.Thread(target=self._scan_document) + self._scan_thread.start() + if self._scan_dirthread is None: + self._scan_dirthread = threading.Thread(target=self._scan_dir) + self._scan_dirthread.start() + + 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) + 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 + + bookmarks = reader.outline + if bookmarks: + catalogs = [] + self._parse_pdf_bookmarks(bookmarks,catalogs) + metadata["catalogs"] = json.dumps(catalogs) + + 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"] + + return hash_result,title,meta_data + + + def _support_file(self,file_name:str) -> bool: + 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) + + # merge to standard workspace env, **ABANDON this!** class KnowledgeBaseFileSystemEnvironment(Environment): def __init__(self, env_id: str) -> None: @@ -537,3 +713,37 @@ class KnowledgeBaseFileSystemEnvironment(Environment): content = await f.read(2048) return content + +class ShellEnvironment(Environment): + def __init__(self, env_id: str) -> None: + super().__init__(env_id) + + operator_param = { + "command": "command will execute", + } + self.add_ai_function(SimpleAIFunction("shell_exec", + "execute shell command in linux bash", + self.shell_exec,operator_param)) + + #run_code_param = { + # "pycode": "python code will execute", + #} + #self.add_ai_function(SimpleAIFunction("run_code", + # "execute python code", + # self.run_code,run_code_param)) + + + async def shell_exec(self,command:str) -> str: + import asyncio.subprocess + process = await asyncio.create_subprocess_shell( + command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await process.communicate() + returncode = process.returncode + if returncode == 0: + return f"Execute success! stdout is:\n{stdout}\n" + else: + return f"Execute failed! stderr is:\n{stderr}\n" + diff --git a/src/requirements.txt b/src/requirements.txt index 0c452e5..20eebbf 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -138,3 +138,5 @@ pydub stability_sdk sentence-transformers==2.2.2 tiktoken +markdown +PyPDF2 \ No newline at end of file diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 696b8c6..11c5036 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -217,7 +217,9 @@ class AIOS_Shell: return "0.5.1" async def send_msg(self,msg:str,target_id:str,topic:str,sender:str = None) -> str: - #AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg) + if sender == self.username: + AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg) + agent_msg = AgentMsg() agent_msg.set(sender,target_id,msg) agent_msg.topic = topic