diff --git a/doc/agent memory.md b/doc/agent memory.md new file mode 100644 index 0000000..407187f --- /dev/null +++ b/doc/agent memory.md @@ -0,0 +1,34 @@ +# agent memory + +## memory的基本形式 +memory的基本形式上是 topic+内容 +topic用一个有意义的路径表示 /xxx/xxx/xxx (有点类似脑图的逻辑,可以通过逐级展开遍历浏览所有的memory) +同一个memory可以被多个路径指向 +内容则是一个json文件 + + +## Agent 使用memory的 +1. 根据当前会话的主题,尝试在known_info中加载必要的memory +2. 提供memory的 list/查询 函数, 允许agent在必要的时候 list / 查询memory +该使用逻辑的本质和kb查询逻辑很像 + +## Agent 更新/创建memory +1. 在任何llm process的过程中,agent都可以用写文件的形式创建memory +2. 更新memory通常是一个专门的 self-think过程,agent此时会用某种模式整理自己所有的logs和memory,并对memory进行更新、创建、删除 +该更新逻辑与Agent 与KB的Self-learning逻辑很像。但根据log->summary的过程基本上是 self-think独有的 + +## 实现逻辑 +基本思路: +1. 核心API是一组通用的文件操作API(有些场景可以是只读的) + 一组特化的对象查询API + 路径->Object,Object中包含ObjectId等信息 + ObjectId->Object + Object一定是一个json,里面包含可以打开原始文件的路径(fileId) +2. 通过一组文件系统描述来引导Agent操作特定文件 +3. 通过一组搜索API来引导Agent操作特定文件 + +对象查询API,基本思路是 + +ObjectId->Object + + + diff --git a/doc/package_manager.md b/doc/package_manager.md index 33d9ce7..f7d1273 100644 --- a/doc/package_manager.md +++ b/doc/package_manager.md @@ -18,6 +18,7 @@ Let's start by introducing the two important processes. Note that the dependency check during installation allows for the missing packages to be installed into the current environment. # Some Basic Concepts + - ***env***:A target environment consisting of a series of configuration files, where packages can be loaded/installed. - ***pkg***:A Package(pkg) is either a folder or a file that serves the same purpose as a folder (such as zip, iso, etc.). - ***pkg_name***:A unique string used to label a package. It's usually a readable package name, but can also include the version number or even the ContentId. diff --git a/doc/promps/readme.md b/doc/promps/readme.md new file mode 100644 index 0000000..e0cc8d0 --- /dev/null +++ b/doc/promps/readme.md @@ -0,0 +1,15 @@ +# prompts + +四个循环 +1. 立刻处理消息+深度思考整理循环 +Process <-> Self Thinking + +2. 任务迭代完整循环致力于完成所有的待完成任务(不一定是成功完成) +Task -> Todo -> Check + +3. 知识库整理 +New Knowledge -> Self-Learning +使用知识库的时机?是否有quick process和deep thing的区别? + +4. Self-Improve +根据四元组:输入,提示词,输出,上级意见 (可选),对提示词进行改进 \ No newline at end of file diff --git a/rootfs/agents/Jarvis/agent.toml b/rootfs/agents/Jarvis/agent.toml index 419eea7..132eaab 100644 --- a/rootfs/agents/Jarvis/agent.toml +++ b/rootfs/agents/Jarvis/agent.toml @@ -12,6 +12,28 @@ Your name is Jarvis, the super personal assistant to the Principal. Help the Pri Only clearly specifying the task you completed can be completed independently. """ +kb_query_desc = """ +$ introduce (have default config) +$ dir descriptions +$ dir1 +$ dir2 +$ dir3 +$ support actions(if enable) +$ support funcitons(if enable) + +现有信息以知识图谱的形式保存在存储系统中。 +1. 介绍知识图谱的结构 +2. 不同部分的规则说明(可选)创建知识图谱的指导思路(目前不允许AI自创结构) + + +# read (function) +access_knowledge_graph($op_name,$params) + +# write (action) +update_knowledge_graph($op_name,$params) + +""" + [behavior.on_message] type="AgentMessageProcess" mutil_model="gpt-4-vision-preview" @@ -45,8 +67,8 @@ known_info_tips = """ tools_tips = """ """ -llm_context.actions.enable = ["agent.workspace.create_task","agent.workspace.cancel_task"] -llm_context.functions.enable = ["agent.workspace.list_task"] +llm_context.actions.enable = ["agent.workspace.create_task","agent.workspace.cancel_task","knowledge_base.knowledge_graph_update"] +llm_context.functions.enable = ["agent.workspace.list_task","knowledge_base.knowledge_graph_read"] [behavior.triage_tasks] diff --git a/src/aios/__init__.py b/src/aios/__init__.py index 3d37b64..18f7823 100644 --- a/src/aios/__init__.py +++ b/src/aios/__init__.py @@ -29,6 +29,7 @@ from .ai_functions.image_2_text_function import Image2TextFunction from .environment.workspace_env import WorkspaceEnvironment from .storage.storage import ResourceLocation,AIStorage,UserConfig,UserConfigItem +from .storage.objfs import ObjFS from .net import * from .knowledge import * @@ -36,4 +37,4 @@ from .package_manager import * from .utils import * -AIOS_Version = "0.5.2, build 2023-12-15" +AIOS_Version = "0.5.2, build 2024-3-31" diff --git a/src/aios/agent/agent_memory.py b/src/aios/agent/agent_memory.py index c660a76..a3307e2 100644 --- a/src/aios/agent/agent_memory.py +++ b/src/aios/agent/agent_memory.py @@ -9,10 +9,11 @@ import sqlite3 import aiofiles from ..storage.storage import AIStorage +from ..knowledge.knowledge_base import BaseKnowledgeGraph,ObjFSKnowledgeGrpah from ..frame.compute_kernel import ComputeKernel from ..frame.contact_manager import ContactManager from ..frame.contact import Contact -from ..proto.ai_function import ParameterDefine, SimpleAIAction, SimpleAIFunction +from ..proto.ai_function import ParameterDefine, SimpleAIFunction from ..proto.agent_msg import AgentMsg, AgentMsgType from ..proto.agent_task import AgentWorkLog @@ -35,20 +36,34 @@ logger = logging.getLogger(__name__) class AgentMemory: - def __init__(self,agent_id:str,base_dir:str) -> None: + def __init__(self,agent_id:str,base_dir:str,enable_knowledge_graph = True) -> None: self.agent_memory_base_dir = base_dir self.agent_id:str= agent_id AIStorage.get_instance().ensure_directory_exists(self.agent_memory_base_dir) - AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/experience") - AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/contacts") - AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/relations") - AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/summary") + #AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/experience") + #AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/contacts") + #AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/relations") + #AIStorage.get_instance().ensure_directory_exists(f"{self.agent_memory_base_dir}/summary") self.memory_db:str = f"{self.agent_memory_base_dir}/memory.db" - self.model_name:str = "gp4-1106-preview" + self.threshold_hours = 72 self.last_think_time : float = 0.0 + self.enable_knowledge_graph : bool = enable_knowledge_graph + if self.enable_knowledge_graph: + kb_desc = """The Knowledgegraph is used to store important information obtained by Agent in the conversation.Use the following ways to store information: + /contacts/$name:Related information of the contact + /relations/$obj1/$obj2:The relationship between obj2 and obj1 + /summary/$topic:Based on topic summary + """ + + self.knowledge_graph = ObjFSKnowledgeGrpah(f"{self.agent_id}.memory",self.memory_db,kb_desc) + BaseKnowledgeGraph.add_kb(self.knowledge_graph) + self.simple_memory_sentences = None + else: + self.knowledge_graph = None + self.simple_memory_sentences : List[str] = [] self.load_memory_meta() @@ -84,7 +99,7 @@ class AgentMemory: return chatsession # return last record time - async def load_records(self,starttime,tokenlimit=8000)->float: + async def load_records(self,starttime,tokenlimit=8000,model_name=None)->float: # 专用思路:做聊天记录/工作经验的整理 # 通用思路:没有具体的目的,让Agent根据提示词自己工作(可能效果很差也可能很好) # 先实现通用思路 @@ -92,7 +107,7 @@ class AgentMemory: work_records = self.load_worklogs(self.agent_id,token_limit=tokenlimit) pass - async def load_chatlogs(self,msg:AgentMsg,token_limit=800): + async def load_chatlogs(self,msg:AgentMsg,token_limit=800,model_name=""): chatsession = self.get_session_from_msg(msg) # Must load n (n> = 2), and hope to load the M # The information in the # M is gradually added, knowing that it is less than 72 hours from the current time, and consumes enough tokens @@ -105,7 +120,7 @@ class AgentMemory: dt = datetime.fromtimestamp(float(msg.create_time)) formatted_time = dt.strftime('%y-%m-%d %H:%M:%S') record_str = f"{msg.sender},[{formatted_time}]\n{msg.body}\n" - token_limit -= ComputeKernel.llm_num_tokens_from_text(record_str,self.model_name) + token_limit -= ComputeKernel.llm_num_tokens_from_text(record_str) if token_limit <= 32: is_all = False break @@ -156,7 +171,7 @@ class AgentMemory: rows = c.fetchall() - return [self.from_db_row(row) for row in rows] + return [self.worklog_from_db_row(row) for row in rows] def _create_table(self,conn): c = conn.cursor() @@ -176,7 +191,7 @@ class AgentMemory: #conn.close() @classmethod - def from_db_row(self,row): + def worklog_from_db_row(self,row): log = AgentWorkLog() # 这里高度依赖表结构的顺序 log.logid, log.owner_id, log.work_type, log.timestamp, log.content, log.result, meta_str, log.operator = row @@ -202,6 +217,7 @@ class AgentMemory: def load_meta(self,Dict): self.last_think_time = Dict.get("last_think_time",0.0) + self.simple_memory_sentences = Dict.get("simple_memory_sentences",[]) def load_memory_meta(self): meta_file_path = f"{self.agent_memory_base_dir}/meta.json" @@ -230,7 +246,12 @@ class AgentMemory: self.last_think_time = last_time self.save_memory_meta() + async def get_contact_summary(self,contact_id:str) -> str: + # There is two part of contact summary + # Part 1. user defined summary (set by owner or by contac) , global , imutable + # Part 2. auto generated summary, local in agent memory , mutable + if contact_id is None: return "Contact id is None" @@ -241,107 +262,107 @@ class AgentMemory: result["relation"] = contact_info.relationship result["notes"] = contact_info.notes - summary_path = f"{self.agent_memory_base_dir}/contacts/{contact_id}.summary" - try: - async with aiofiles.open(summary_path, mode='r') as file: - result["summary"] = await file.read() + # summary_path = f"{self.agent_memory_base_dir}/contacts/{contact_id}.summary" + # try: + # async with aiofiles.open(summary_path, mode='r') as file: + # result["summary"] = await file.read() - except Exception as e: - logger.error(f"read contact summary failed: {e}") + # except Exception as e: + # logger.error(f"read contact summary failed: {e}") return json.dumps(result,ensure_ascii=False) - async def update_contact_summary(self,contact_id:str,summary:str): - summary_path = f"{self.agent_memory_base_dir}/contacts/{contact_id}.summary" - try: - async with aiofiles.open(summary_path, mode='w') as file: - await file.write(summary) - return "OK" - except Exception as e: - logger.error(f"write contact summary failed: {e}") - return "write contact summary failed: {e}" + # async def update_contact_summary(self,contact_id:str,summary:str): + # summary_path = f"{self.agent_memory_base_dir}/contacts/{contact_id}.summary" + # try: + # async with aiofiles.open(summary_path, mode='w') as file: + # await file.write(summary) + # return "OK" + # except Exception as e: + # logger.error(f"write contact summary failed: {e}") + # return "write contact summary failed: {e}" - async def get_summary(self,object_name:str) -> str: - summary_path = f"{self.agent_memory_base_dir}/{object_name}.summary" - try: - async with aiofiles.open(summary_path, mode='r') as file: - return await file.read() - except Exception as e: - logger.error(f"read summary failed: {e}") - return f"read summary failed: {e}" + # async def get_summary(self,object_name:str) -> str: + # summary_path = f"{self.agent_memory_base_dir}/{object_name}.summary" + # try: + # async with aiofiles.open(summary_path, mode='r') as file: + # return await file.read() + # except Exception as e: + # logger.error(f"read summary failed: {e}") + # return f"read summary failed: {e}" - async def update_summary(self,object_name:str,summary:str) -> str: - summary_path = f"{self.agent_memory_base_dir}/{object_name}.summary" - try: - async with aiofiles.open(summary_path, mode='w') as file: - await file.write(summary) - return "OK" - except Exception as e: - logger.error(f"write summary failed: {e}") - return f"write summary failed: {e}" + # async def update_summary(self,object_name:str,summary:str) -> str: + # summary_path = f"{self.agent_memory_base_dir}/{object_name}.summary" + # try: + # async with aiofiles.open(summary_path, mode='w') as file: + # await file.write(summary) + # return "OK" + # except Exception as e: + # logger.error(f"write summary failed: {e}") + # return f"write summary failed: {e}" - async def list_summary_object_names(self) -> List[str]: - # list dir - try: - contents = os.listdir(self.agent_memory_base_dir) - return [x for x in contents if x.endswith(".summary")] - except Exception as e: - logger.error(f"list summary object names failed: {e}") - return [] + # async def list_summary_object_names(self) -> List[str]: + # # list dir + # try: + # contents = os.listdir(self.agent_memory_base_dir) + # return [x for x in contents if x.endswith(".summary")] + # except Exception as e: + # logger.error(f"list summary object names failed: {e}") + # return [] # means object1 feel object2 is ... - async def get_relation_summary(self,object_name1:str,object_name2:str) -> str: - summary_path = f"{self.agent_memory_base_dir}/relations/{object_name1}.relation.{object_name2}.summary" - try: - async with aiofiles.open(summary_path, mode='r') as file: - await file.read() - except FileNotFoundError: - return "no summary" - except Exception as e: - logger.error(f"read relation summary failed: {e}") - return f"read relation summary failed: {e}" + # async def get_relation_summary(self,object_name1:str,object_name2:str) -> str: + # summary_path = f"{self.agent_memory_base_dir}/relations/{object_name1}.relation.{object_name2}.summary" + # try: + # async with aiofiles.open(summary_path, mode='r') as file: + # await file.read() + # except FileNotFoundError: + # return "no summary" + # except Exception as e: + # logger.error(f"read relation summary failed: {e}") + # return f"read relation summary failed: {e}" - async def update_relation_summary(self,object_name1:str,object_name2:str,summary:Dict): - summary_path = f"{self.agent_memory_base_dir}/relations/{object_name1}.relation.{object_name2}.summary" - try: - async with aiofiles.open(summary_path, mode='w') as file: - await file.write(json.dumps(summary)) - return "OK" - except Exception as e: - logger.error(f"write relation summary failed: {e}") - return "write relation summary failed: {e}" + # async def update_relation_summary(self,object_name1:str,object_name2:str,summary:Dict): + # summary_path = f"{self.agent_memory_base_dir}/relations/{object_name1}.relation.{object_name2}.summary" + # try: + # async with aiofiles.open(summary_path, mode='w') as file: + # await file.write(json.dumps(summary)) + # return "OK" + # except Exception as e: + # logger.error(f"write relation summary failed: {e}") + # return "write relation summary failed: {e}" - async def get_experience(self,topic_name:str) -> str: - experience_path = f"{self.agent_memory_base_dir}/experience/{topic_name}.experience" - try: - async with aiofiles.open(experience_path, mode='r') as file: - await file.read() - except FileNotFoundError: - return "no experience" - except Exception as e: - logger.error(f"read experience failed: {e}") - return f"read experience failed: {e}" + # async def get_experience(self,topic_name:str) -> str: + # experience_path = f"{self.agent_memory_base_dir}/experience/{topic_name}.experience" + # try: + # async with aiofiles.open(experience_path, mode='r') as file: + # await file.read() + # except FileNotFoundError: + # return "no experience" + # except Exception as e: + # logger.error(f"read experience failed: {e}") + # return f"read experience failed: {e}" - async def set_experience(self,topic_name:str,summary:str) -> str: - experience_path = f"{self.agent_memory_base_dir}/experience/{topic_name}.experience" - try: - async with aiofiles.open(experience_path, mode='w') as file: - await file.write(summary) - return "OK" - except Exception as e: - logger.error(f"write experience failed: {e}") - return "write experience failed: {e}" + # async def set_experience(self,topic_name:str,summary:str) -> str: + # experience_path = f"{self.agent_memory_base_dir}/experience/{topic_name}.experience" + # try: + # async with aiofiles.open(experience_path, mode='w') as file: + # await file.write(summary) + # return "OK" + # except Exception as e: + # logger.error(f"write experience failed: {e}") + # return "write experience failed: {e}" - async def list_experience(self) -> List[str]: - dir_path = f"{self.agent_memory_base_dir}/experience" - try: - contents = os.listdir(dir_path) - return [x for x in contents if x.endswith(".experience")] - except Exception as e: - logger.error(f"list experience failed: {e}") - return [] + # async def list_experience(self) -> List[str]: + # dir_path = f"{self.agent_memory_base_dir}/experience" + # try: + # contents = os.listdir(dir_path) + # return [x for x in contents if x.endswith(".experience")] + # except Exception as e: + # logger.error(f"list experience failed: {e}") + # return [] @staticmethod def register_ai_functions(): diff --git a/src/aios/agent/llm_process.py b/src/aios/agent/llm_process.py index df8b816..d686147 100644 --- a/src/aios/agent/llm_process.py +++ b/src/aios/agent/llm_process.py @@ -14,6 +14,7 @@ from .workspace import AgentWorkspace from .llm_context import LLMProcessContext,GlobaToolsLibrary, SimpleLLMContext from ..frame.compute_kernel import ComputeKernel +from ..knowledge.knowledge_base import BaseKnowledgeGraph from abc import ABC,abstractmethod import copy @@ -229,8 +230,7 @@ class LLMAgentBaseProcess(BaseLLMProcess): self.workspace : AgentWorkspace = None # If Workspace is not none , enable Agent Tasklist self.memory : AgentMemory = None - self.enable_kb : bool = False - self.kb = None + self.enable_kb_list : List[str] = None async def initial(self,params:Dict = None) -> bool: self.memory = params.get("memory") @@ -265,26 +265,49 @@ class LLMAgentBaseProcess(BaseLLMProcess): if config.get("context"): self.context = config.get("context") + if config.get("knowledge_grpah_introduce"): + self.knowledge_grpah_introduce = config.get("knowledge_grpah_introduce") + self.llm_context = SimpleLLMContext() if config.get("llm_context"): self.llm_context.load_from_config(config.get("llm_context")) - if config.get("enable_kb"): - self.enable_kb = config.get("enable_kb") == "true" + def prepare_knowledge_grpah_prompt(self) -> Dict: + result = {} + + result["introduce"] = BaseKnowledgeGraph.get_kb_default_desc_str() + result["knowledge_graph_list"] = {} + have_kb = False + if self.memory.enable_knowledge_graph: + result["knowledge_graph_list"][self.memory.knowledge_graph.kb_id] = self.memory.knowledge_graph.get_description() + have_kb = True + + if self.enable_kb_list: + for kb_id in self.enable_kb_list: + kb = BaseKnowledgeGraph.get_kb(kb_id) + if kb: + have_kb = True + result["knowledge_graph_list"][kb_id] = kb.get_description() + else: + logger.error(f"knowledge base {kb_id} not found") + + if have_kb is False: + return None + + return result + + def prepare_role_system_prompt(self,context_info:Dict) -> Dict: system_prompt_dict = {} - # System Prompt - ## LLM的身份说明 - system_prompt_dict["role_description"] = self.role_description - #prompt.append_system_message(self.role_description) - ## 处理信息的流程说明 + system_prompt_dict["role_description"] = self.role_description system_prompt_dict["process_rule"] = self.process_description - #prompt.append_system_message(self.process_description) - ### 回复的格式 system_prompt_dict["reply_format"] = self.reply_format - #prompt.append_system_message(self.reply_format) + + kb_prompt = self.prepare_knowledge_grpah_prompt() + if kb_prompt: + system_prompt_dict["knowledge_graph"] = kb_prompt ## Context if self.context: @@ -301,9 +324,13 @@ class LLMAgentBaseProcess(BaseLLMProcess): def get_action_desc(self) -> Dict: result = {} - actions_list = self.llm_context.get_all_ai_action() + actions_list = [] + + actions_list.extend(self.llm_context.get_all_ai_action()) + for action in actions_list: result[action.get_name()] = action.get_description() + return result async def get_inner_function_for_exec(self,func_name:str) -> AIFunction: @@ -483,10 +510,6 @@ class AgentMessageProcess(LLMAgentBaseProcess): #TODO eanble workspace functions? logger.info(f"workspace is not none,enable workspace functions") - ## 给予查询KB的权限 - if self.enable_kb: - logger.info(f"enable kb") - ### 根据Token Limit加载聊天记录 remain_token = self.get_remain_prompt_length(prompt,json.dumps(system_prompt_dict,ensure_ascii=False)) @@ -575,7 +598,7 @@ class AgentSelfThinking(LLMAgentBaseProcess): history_str = history_str + record_str - if read_history_msg >= 2: + if ComputeKernel.llm_num_tokens_from_text(history_str,self.model_name) > self.chat_summary_token_len: session_history["history"] = history_str chat_history[session_id] = session_history chatsession.summarize_pos = cur_pos diff --git a/src/aios/frame/compute_kernel.py b/src/aios/frame/compute_kernel.py index 9458e18..1ab785d 100644 --- a/src/aios/frame/compute_kernel.py +++ b/src/aios/frame/compute_kernel.py @@ -105,7 +105,7 @@ class ComputeKernel: return True @staticmethod - def llm_num_tokens_from_text(text:str,model:str) -> int: + def llm_num_tokens_from_text(text:str,model:str = None) -> int: if model is None: model = "gpt-4-turbo-preview" diff --git a/src/aios/knowledge/__init__.py b/src/aios/knowledge/__init__.py index 361a67d..e0d5dfd 100644 --- a/src/aios/knowledge/__init__.py +++ b/src/aios/knowledge/__init__.py @@ -3,4 +3,5 @@ from .vector import * from .data import * from .store import KnowledgeStore from .core_object import * -from .pipeline import * \ No newline at end of file +from .pipeline import * +from .knowledge_base import * \ No newline at end of file diff --git a/src/aios/knowledge/knowledge_base.py b/src/aios/knowledge/knowledge_base.py new file mode 100644 index 0000000..8a7993c --- /dev/null +++ b/src/aios/knowledge/knowledge_base.py @@ -0,0 +1,371 @@ +from abc import ABC, abstractmethod +import json +import os +import uuid +from typing import List + +from ..proto.ai_function import ParameterDefine, SimpleAIAction, SimpleAIFunction +from ..agent.llm_context import GlobaToolsLibrary +from ..storage.objfs import ObjFS + +import logging + +logger = logging.getLogger(__name__) + +class BaseKnowledgeGraph(ABC): + _all_knowledge_bases = {} + _default_kb = None + @classmethod + def get_kb(cls, kb_id:str): + if kb_id is None: + return cls._default_kb + + return cls._all_knowledge_bases.get(kb_id) + + @classmethod + def add_kb(cls,kb:'BaseKnowledgeGraph',is_default=False): + cls._all_knowledge_bases[kb.kb_id] = kb + if is_default: + cls._default_kb = kb + + @classmethod + def remove_kb(cls,kb_id:str): + if cls._default_kb is not None and cls._default_kb.kb_id == kb_id: + cls._default_kb = None + + if cls._all_knowledge_bases.get(kb_id) is not None: + del cls._all_knowledge_bases[kb_id] + + def __init__(self, kb_id: str,kb_desc:str=None): + self.kb_id = kb_id + if kb_desc is None: + self.kb_desc = """ + """ + else: + self.kb_desc = kb_desc + + def get_description(self)->str: + return self.kb_desc + + # 读接口: 查询,浏览 + @abstractmethod + async def serach(self, query: str,query_type:str): + pass + + @abstractmethod + async def get_obj_by_path(self,path)->str: + pass + + @abstractmethod + async def get_obj_by_id(self,obj_id)->str: + pass + + @abstractmethod + async def list_by_path(self,base_path)->List[str]: + pass + + @abstractmethod + def list_source(self) -> List[str]: + pass + + + @abstractmethod + async def add_obj(self,obj_id,obj_name,obj_content,paths) -> bool: + pass + + @abstractmethod + async def remove(self,remove_path) -> bool: + pass + + @abstractmethod + async def remove_obj(self,objid): + pass + + @abstractmethod + async def link(self,obj_id,paths) -> bool: + pass + + @abstractmethod + async def unlink(self,paths) -> bool: + pass + + @abstractmethod + async def update_obj(self,obj_id,new_content) -> bool: + pass + + @staticmethod + def get_kb_default_desc_str(): + return """The basic design of the Knowledge Graph is +1. Each object can be described in JSON, and have a unique obj_id. +2. The object can be accessed through the PATH, and multiple paths can point to the same object. +3. Carefully understand the semantics of the path, and follow the description of the knowledge graph.You can list all the sub-paths of a path through the LIST operation +All Knowledge Graph APIs return are json format string.""" + + + # 写接口:通常由KnowledgePipeline调用 + @staticmethod + def register_ai_functions(): + + async def knowledge_graph_access(parameters): + kb_id = parameters['kb_id'] + op_name = parameters['op'] + param = parameters['param'] + + + if op_name is None: + logger.error("Operation type is not specified") + return "Operation type is not specified" + if param is None: + logger.error("Operation parameters is not specified") + return "Error! Operation parameters is not specified" + param = json.loads(param) + + kb = BaseKnowledgeGraph.get_kb(kb_id) + if kb is None: + logger.error(f"Knowledge base is not found id:{kb_id}") + return "Error! Knowledge base is not found" + + if op_name == "list": + root_path = param.get("path") + if root_path is None: + logger.error("Path is not specified") + return "Error! Path is not specified" + + return json.dumps(await kb.list_by_path(root_path), ensure_ascii=False) + + if op_name == "tree": + root_path = param.get("path") + if root_path is None: + logger.error("Path is not specified") + return "Error! Path is not specified" + + depth = param.get("depth") + if depth is None: + depth = 3 + return json.dumps(await kb.tree(root_path,depth), ensure_ascii=False) + + if op_name == "read": + obj_path = param.get("path") + if obj_path is None: + logger.error("Path is not specified") + return "Error! Path is not specified" + return json.dumps(await kb.get_obj_by_path(obj_path), ensure_ascii=False) + + if op_name == "get_obj": + obj_id = param.get("obj_id") + if obj_id is None: + logger.error("Object ID is not specified") + return "Error! Object ID is not specified" + return json.dumps(await kb.get_obj_by_id(obj_id), ensure_ascii=False) + + + return "Error! Operation type is not supported" + + # search is not supported currently + func_desc = "Read knowledge graph, op_param format is as follows: list:{'path':$path}, read:{'path':$path}, get_obj:{'obj_id':$obj_id}, tree:{'path':$path,'depth':$depth}" + parameters = ParameterDefine.create_parameters({ + "kb_id": "Knowledge Base ID", + "op": "Operation Type,could be [list, read, get_obj]", + "op_param": "Operation Param, must be a json string" + }) + + knowledge_graph_access_func = SimpleAIFunction("knowledge_base.knowledge_graph_read", + func_desc, + knowledge_graph_access, + parameters) + GlobaToolsLibrary.get_instance().register_tool_function(knowledge_graph_access_func) + + async def knwoledge_graph_update(parameters): + kb_id = parameters['kb_id'] + op_name = parameters['op'] + param = parameters['param'] + result = {} + if op_name is None: + logger.error("Operation type is not specified") + result["result"] = "Error! Operation type is not specified" + return json.dumps(result, ensure_ascii=False) + if param is None: + logger.error("Operation parameters is not specified") + result["result"] = "Error! Operation parameters is not specified" + return json.dumps(result, ensure_ascii=False) + param = json.loads(param) + + kb = BaseKnowledgeGraph.get_kb(kb_id) + if kb is None: + logger.error(f"Knowledge base is not found id:{kb_id}") + result["result"] = "Error! Knowledge base is not found" + return json.dumps(result, ensure_ascii=False) + + if op_name == "write": + write_path = param.get("path") + if write_path is None: + logger.error("Path is not specified") + result["result"] = "Error! Path is not specified" + return json.dumps(result, ensure_ascii=False) + obj_content = param.get("obj_json") + if obj_content is None: + logger.error("Object content is not specified") + result["result"] = "Error! Object content is not specified" + return json.dumps(result, ensure_ascii=False) + + objid = uuid.uuid4() + objname = os.path.basename(write_path) + paths = [] + paths.append(write_path) + if await kb.add_obj(objid,objname,obj_content['content'],paths): + result["result"] = "OK" + result['obj_id'] = objid + else: + result["result"] = "Error! Add object failed" + + if op_name == "remove": + remove_path = param.get("path") + if remove_path is None: + logger.error("Path is not specified") + result["result"] = "Error! Path is not specified" + return json.dumps(result, ensure_ascii=False) + + if await kb.remove(remove_path): + result["result"] = "OK" + else: + result["result"] = "Error! Remove path failed" + + if op_name == "remove_obj": + obj_id = param.get("obj_id") + if obj_id is None: + logger.error("Object ID is not specified") + result["result"] = "Error! Object ID is not specified" + return result + + obj = await kb.get_obj_by_id(obj_id) + if obj is None: + logger.error(f"Object is not found id:{obj_id}") + result["result"] = "Error! Object is not found" + return result + + await kb.remove_obj(obj_id) + result["result"] = "OK" + + if op_name == "set_obj": + obj_id = param.get("obj_id") + if obj_id is None: + logger.error("Object ID is not specified") + result["result"] = "Error! Object ID is not specified" + return json.dumps(result, ensure_ascii=False) + + obj = await kb.get_obj_by_id(obj_id) + if obj is None: + logger.error(f"Object is not found id:{obj_id}") + result["result"] = "Error! Object is not found" + return result + + obj_content = param.get("obj_json") + if obj_content is None: + logger.error("new object is not specified") + result["result"] = "Error! new object is not specified" + return json.dumps(result, ensure_ascii=False) + + await kb.update_obj(obj_id,obj_content) + result["result"] = "OK" + + if op_name == "link": + path_from = param.get("path") + path_to = param.get("target") + if path_from is None or path_to is None: + logger.error("Path is not specified") + result["result"] = "Error! Path is not specified" + return json.dumps(result, ensure_ascii=False) + + objid = await kb.get_obj_by_path(path_to) + if objid is None: + logger.error(f"Object is not found path:{path_to}") + result["result"] = "Error!Target Object is not found" + return json.dumps(result, ensure_ascii=False) + + await kb.link(objid,[path_from]) + result["result"] = "OK" + + if op_name == "unlink": + path_will_remove = param.get("path") + if path_will_remove is None: + logger.error("Path is not specified") + result["result"] = "Error! Path is not specified" + return json.dumps(result, ensure_ascii=False) + + await kb.unlink([path_will_remove]) + result["result"] = "OK" + + return json.dumps(result, ensure_ascii=False) + + + OperationParames = """Parameters is a json string, the format is as follows: + write:{'path':$path,'obj_json':$obj_json}, + remove:{'path':$path}, + remove_obj:{'obj_id':$obj_id}, + set_obj:{'obj_id':$obj_id,'obj_json':$new_obj_json}, + link:{'path':$path,'target':$target_obj_path}, + unlink:{'path':$path} +""" + parameters = ParameterDefine.create_parameters({ + "kb_id": "Knowledge Base ID", + "op": "Operation Type,could be [write, remove, remove_obj, set_obj, link, unlink", + "param": OperationParames + }) + + knowledge_graph_update_func = SimpleAIFunction("knowledge_base.knowledge_graph_update", + "Update Knowledge Graph APIs", + knwoledge_graph_update, + parameters) + GlobaToolsLibrary.get_instance().register_tool_function(knowledge_graph_update_func) + + +class ObjFSKnowledgeGrpah(BaseKnowledgeGraph): + def __init__(self, kb_id:str,db_path:str,kb_desc:str=None): + super().__init__(kb_id,kb_desc) + self.db_path = db_path + self.obj_storage : ObjFS = ObjFS(db_path) + + async def serach(self, query: str,query_type:str): + pass + + def list_source(self): + pass + + async def get_obj_by_path(self,path)->str: + return self.obj_storage.get_obj_by_path(path) + + async def get_obj_by_id(self,obj_id)->str: + return self.obj_storage.get_obj_by_id(obj_id) + + async def list_by_path(self,base_path)->List[str]: + return self.obj_storage.list_paths(base_path) + + async def tree(self,base_path,depth:int)->str: + return self.obj_storage.tree(base_path,depth) + + async def add_obj(self,obj_id,obj_name,obj_content,paths)->bool: + self.obj_storage.add_obj(obj_id,obj_name,obj_content,paths) + + #todo 更新默认是做dict的merge + async def update_obj(self, obj_id, new_content)->bool: + return self.obj_storage.update_obj(obj_id,new_content) + + async def remove(self,remove_path)->bool: + self.obj_storage.remove_path(remove_path) + + async def remove_obj(self,objid)->bool: + self.obj_storage.remove_obj(objid) + + async def link(self,from_path,target_path)->bool: + objid = self.obj_storage.get_obj_by_path(target_path) + if objid is None: + return False + self.obj_storage.add_path(objid,from_path) + return True + + async def unlink(self,paths)->bool: + self.obj_storage.remove_path(paths) + + + + diff --git a/src/aios/storage/obj_storage.py b/src/aios/storage/obj_storage.py new file mode 100644 index 0000000..267a1a5 --- /dev/null +++ b/src/aios/storage/obj_storage.py @@ -0,0 +1,18 @@ +from typing import List + +class NamedObjectStorage: + def __init__(self, storage, name: str): + self.storage = storage + self.name = name + + async def get(self, key: str) -> bytes: + return await self.storage.get(self.name, key) + + async def put(self, key: str, data: bytes): + await self.storage.put(self.name, key, data) + + async def delete(self, key: str): + await self.storage.delete(self.name, key) + + async def list(self) -> List[str]: + return await self.storage.list(self.name) \ No newline at end of file diff --git a/src/aios/storage/objfs.py b/src/aios/storage/objfs.py new file mode 100644 index 0000000..286631f --- /dev/null +++ b/src/aios/storage/objfs.py @@ -0,0 +1,217 @@ +from abc import ABC, abstractmethod +import sqlite3 +from sqlite3 import Error +from typing import List + +import threading +import time +import uuid +import logging + +logger = logging.getLogger(__name__) + +class ObjFSReader(ABC): + @abstractmethod + def get_obj_by_path(self,path): + pass + + @abstractmethod + def get_obj_by_id(self,obj_id): + pass + + @abstractmethod + def list_paths(self,base_path): + pass + + +#ObjFS provides structured data storage similar to brain-like, as an object storage layer of Agent Friendly +class ObjFS(ObjFSReader): + def __init__(self, db_file): + """ initialize db connection """ + self.db_file = db_file + 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_file) + 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 Error as e: + logger.error("Error occurred while connecting to database: %s", e) + return None + + if conn: + self._create_table(conn) + + return conn + + def _create_table(self, conn): + try: + conn.execute('''CREATE TABLE IF NOT EXISTS objects + (id TEXT PRIMARY KEY, name TEXT, content TEXT, created_at REAL, modified_at REAL, size INTEGER)''') + + conn.execute('''CREATE TABLE IF NOT EXISTS paths + (id INTEGER PRIMARY KEY AUTOINCREMENT, path TEXT UNIQUE, obj_id TEXT, FOREIGN KEY(obj_id) REFERENCES objects(id))''') + + except Error as e: + logger.error("Error occurred while creating tables: %s", e) + + def close(self): + local = threading.local() + if not hasattr(local, 'conn'): + return + local.conn.close() + + def add_obj(self,obj_uuid, name, content, paths) -> bool: + conn = self._get_conn() + c = conn.cursor() + #obj id是guid,由外部生成 + + # 获取当前时间戳 + current_time = time.time() + + # 计算内容大小 + content_size = len(content.encode('utf-8')) + try: + # 插入对象 + c.execute("INSERT INTO objects (id, name, content, created_at, modified_at, size) VALUES (?, ?, ?, ?, ?, ?)", (obj_uuid, name, content, current_time, current_time, content_size)) + + # 插入路径 + for path in paths: + c.execute("INSERT OR IGNORE INTO paths (path, obj_id) VALUES (?, ?)", (path, obj_uuid)) + + conn.commit() + except Error as e: + logger.warning("Error occurred while adding object: %s", e) + return False + + return True + + def update_obj(self,obj_id, new_content) -> bool: + #UPDATE orders + #SET data = json_set( + # data, + # '$.items[1].price', + # 0.35 + #) + #WHERE id = 1; + + try: + conn = self._get_conn() + c = conn.cursor() + # 获取当前时间戳 + current_time = time.time() + + # 计算新内容大小 + + new_content_size = len(new_content.encode('utf-8')) + + c.execute("UPDATE objects SET content = ?, modified_at = ?, size = ? WHERE id = ?", (new_content, current_time, new_content_size, obj_id)) + conn.commit() + return True + except Error as e: + logger.warning("Error occurred while updating object: %s", e) + return False + + def add_path(self,obj_id, new_path) -> bool: + try: + conn = self._get_conn() + c = conn.cursor() + c.execute("INSERT OR IGNORE INTO paths (path, obj_id) VALUES (?, ?)", (new_path, obj_id)) + conn.commit() + return True + except Error as e: + logger.warning("Error occurred while adding path: %s", e) + return False + + def remove_path(self,path) -> bool: + try: + conn = self._get_conn() + c = conn.cursor() + #TODO + c.execute("DELETE FROM paths WHERE path = ?", (path,)) + conn.commit() + return True + except Error as e: + logger.warning("Error occurred while removing path: %s", e) + return False + + def remove_obj(self,obj_id) -> bool: + try: + conn = self._get_conn() + c = conn.cursor() + c.execute("DELETE FROM objects WHERE id = ?", (obj_id,)) + + # 删除所有与该对象相关的路径 + c.execute("DELETE FROM paths WHERE obj_id = ?", (obj_id,)) + conn.commit() + return True + except Error as e: + logger.warning("Error occurred while removing object: %s", e) + return False + + def get_obj_by_path(self,path) -> str: + try: + conn = self._get_conn() + c = conn.cursor() + c.execute("SELECT objects.id, objects.name, objects.content FROM objects JOIN paths ON objects.id = paths.obj_id WHERE paths.path = ?", (path,)) + obj_row = c.fetchone() + if obj_row: + return obj_row[2] + return None + except Error as e: + logger.warning("Error occurred while getting object by path: %s", e) + return None + + + def get_obj_by_id(self,obj_id) -> str: + try: + conn = self._get_conn() + c = conn.cursor() + c.execute("SELECT id, name, content FROM objects WHERE id = ?", (obj_id,)) + obj_row = c.fetchone() + if obj_row: + return obj_row[2] + return None + except Error as e: + logger.warning("Error occurred while getting object by id: %s", e) + return None + + def list_paths(self,base_path)->List[str]: + try: + conn = self._get_conn() + c = conn.cursor() + c.execute("SELECT path FROM paths WHERE path LIKE ? ESCAPE '/'", (base_path + "/%",)) + return [row[0] for row in c.fetchall()] + except Error as e: + logger.warning("Error occurred while listing paths: %s", e) + return None + + def tree(self, base_path,max_depth=3): + try: + conn = self._get_conn() + c = conn.cursor() + c.execute("SELECT path FROM paths WHERE path LIKE ? ESCAPE '/'", (base_path + "/%",)) + paths = [row[0] for row in c.fetchall()] + tree = {} + for path in paths: + parts = path.split("/") + node = tree + for part in parts: + if part not in node: + node[part] = {} + node = node[part] + return tree + except Error as e: + logger.warning("Error occurred while listing paths: %s", e) + return None + + diff --git a/src/component/openai_node/open_ai_node.py b/src/component/openai_node/open_ai_node.py index 345c6c4..52cb83b 100644 --- a/src/component/openai_node/open_ai_node.py +++ b/src/component/openai_node/open_ai_node.py @@ -27,7 +27,7 @@ class OpenAI_ComputeNode(ComputeNode): @classmethod def declare_user_config(cls): - if os.getenv("OPENAI_API_KEY_") is None: + if os.getenv("OPENAI_API_KEY") is None: user_config = AIStorage.get_instance().get_user_config() user_config.add_user_config("openai_api_key","openai api key",False,None) diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 2c26929..79a77cc 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -153,6 +153,7 @@ class AIOS_Shell: #AgentManager.get_instance().register_environment("knowledge", LocalKnowledgeBase) AgentWorkspace.register_ai_functions() AgentMemory.register_ai_functions() + BaseKnowledgeGraph.register_ai_functions() ShellEnvironment.register_ai_functions()