From 05d2e4a9cf8ad0df59c0a4cc96dd35ba3605c90e Mon Sep 17 00:00:00 2001 From: Liu Zhicong Date: Wed, 30 Aug 2023 12:30:41 -0700 Subject: [PATCH] 1.workflow has reached a state of basic functionality. 2. implemented AI message bus infrastructure, --- .gitignore | 1 + doc/mvp/process model.drawio | 136 +++++++ doc/mvp/workflow.drawio | 4 +- rootfs/agents/agent_1/agent.toml | 2 +- rootfs/agents/email_reader/agent.toml | 5 + rootfs/agents/manager/agent.toml | 6 + rootfs/workflows/plan_maker/workflow.toml | 74 ++++ rootfs/workflows/workflows.cfg | 2 + src/aios_kernel/__init__.py | 7 +- src/aios_kernel/agent.py | 137 ++----- src/aios_kernel/agent_message.py | 39 ++ src/aios_kernel/bus.py | 129 +++++++ src/aios_kernel/chatsession.py | 253 +++++++++++++ src/aios_kernel/compute_kernel.py | 5 +- src/aios_kernel/open_ai_node.py | 4 +- src/aios_kernel/role.py | 51 ++- src/aios_kernel/workflow.py | 350 ++++++++++++------ src/component/agent_manager/agent_manager.py | 14 +- .../workflow_manager/workflow_manager.py | 73 +++- src/service/aios_shell/aios_shell.py | 76 ++-- test/chatsession_test.py | 75 ++++ 21 files changed, 1146 insertions(+), 297 deletions(-) create mode 100644 doc/mvp/process model.drawio create mode 100644 rootfs/agents/email_reader/agent.toml create mode 100644 rootfs/agents/manager/agent.toml create mode 100644 rootfs/workflows/plan_maker/workflow.toml create mode 100644 rootfs/workflows/workflows.cfg create mode 100644 src/aios_kernel/agent_message.py create mode 100644 src/aios_kernel/bus.py create mode 100644 src/aios_kernel/chatsession.py create mode 100644 test/chatsession_test.py diff --git a/.gitignore b/.gitignore index 1d74e21..4ef429f 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ .vscode/ +*.pyc diff --git a/doc/mvp/process model.drawio b/doc/mvp/process model.drawio new file mode 100644 index 0000000..86d56e7 --- /dev/null +++ b/doc/mvp/process model.drawio @@ -0,0 +1,136 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/doc/mvp/workflow.drawio b/doc/mvp/workflow.drawio index e49374a..2b3e239 100644 --- a/doc/mvp/workflow.drawio +++ b/doc/mvp/workflow.drawio @@ -1,6 +1,6 @@ - + @@ -220,7 +220,7 @@ - + diff --git a/rootfs/agents/agent_1/agent.toml b/rootfs/agents/agent_1/agent.toml index 9b13b5f..81966ab 100644 --- a/rootfs/agents/agent_1/agent.toml +++ b/rootfs/agents/agent_1/agent.toml @@ -1,4 +1,4 @@ -instance_id = "agent:xxxxxxabcde" +instance_id = "agent_1" fullname = "tracy wang" [[prompt]] role = "system" diff --git a/rootfs/agents/email_reader/agent.toml b/rootfs/agents/email_reader/agent.toml new file mode 100644 index 0000000..a6baba9 --- /dev/null +++ b/rootfs/agents/email_reader/agent.toml @@ -0,0 +1,5 @@ +instance_id = "agent:xxxxxxabcde" +fullname = "jackma" +[[prompt]] +role = "system" +content = "你擅长处理商务电子邮件" diff --git a/rootfs/agents/manager/agent.toml b/rootfs/agents/manager/agent.toml new file mode 100644 index 0000000..5bb2ba6 --- /dev/null +++ b/rootfs/agents/manager/agent.toml @@ -0,0 +1,6 @@ +instance_id = "agent:xxxxxxabcde" +fullname = "musk" +[[prompt]] +role = "system" +content = "你有丰富的管理技能，擅长将复杂工作拆解成简单的任务，让团队成员高效协作。" + diff --git a/rootfs/workflows/plan_maker/workflow.toml b/rootfs/workflows/plan_maker/workflow.toml new file mode 100644 index 0000000..e5b150f --- /dev/null +++ b/rootfs/workflows/plan_maker/workflow.toml @@ -0,0 +1,74 @@ +# workflow实例场景 +# 能展示sub workflow +# 能展示env的整合 +# 能展示filter的使用 +# 能展示function的调用 +# 能展示基于工作目标/KPI的sub workflow迭代流程 +# 多人场景安排 + +# 例子：举办一个团队活动 +# 方案讨论（通过交互引导的方式收集主人的需求）与确定 + +# 活动前： +# 通讯员，对接管理参加活动的人的情况 (email spider) +# 酒店预订 简单：搜索（酒店评价） 处理异常 +# 行程（票务）预订 ：搜索，处理异常 +# 餐饮预订：给出方案，确定细节，预订 + +# 活动中： +# 进行统计和分析，调整设备 +# 安保，空调，音乐，拍照，录像 +# 响应紧急情况 + +# 活动结束后： +# 整理照片，视频，进行必要的二次创作，发送给相关人员 +# 对活动进行总结，提出改件意见（指导下一次活动） + +# 1. 人员 +# 主管，负责和客户沟通，并对每个环境的结果进行总结 +# 嘉宾对接 +# 酒店组 +# 行程组 +# 财务组 +# 多媒体组 +name = "plan_maker" + +[filter] +"*" = "manager" + +[roles.manager] +name = "manager" +fullname = "经理" +agent="manager" +[[roles.manager.prompt]] +role="system" +content="""你是一个活动策划公司的经理,与客户对接并向团队下达指令。你的团队分为下面几个小组:嘉宾对接组,酒店预定组,行程预订组,财务组,活动摄像组。活动策划分为四个阶段：方案讨论，活动前，活动中，活动后。你会根据客户的需求，对团队进行分工，分别完成各个阶段的工作。你的基本工作模式是： +1. 收到客户的明确的指令后,基于客户的已有信息和客户商量活动方案，和活动策划公司无关的业务你会回答‘与我无关’。当和客户完成活动方案的确认后,你会将拆解后的任务分配给各个小组 +2. 根据目前已经确认的活动方案，你要根据时间适时的检查不同小组的工作情况。当收到小组的工作情况反馈后，你会站在全局的角度判断是否需要调整活动方案，如果需要调整，你会和客户商量重新确定方案，然后再将调整后的方案分配给各个小组。 +3. 有时工作小组会主动与你沟通，反馈一些问题。你会站在全局的角度给与指导，适当的调整工作小组的工作目标。如果反馈的问题需要你和客户沟通，你会和客户沟通后重新确定方案。再将调整后的方案分配给受到影响各个小组。 +4. 当你决定要和工作小组通信时，请使用`sendmsg({小组名称},{内容}`）的形式。""" + + + + +[sub_workflows] +[sub_workflows."嘉宾对接组"] +name="嘉宾对接组" +filter = {"*" = "leader"} + +# 展现读取email和发送email与嘉宾沟通的能力 +[sub_workflows."嘉宾对接组".environments.email] +new_mail = "收到来自{event.data.from},标题为{event.data.subject}的邮件，内容为{event.data.content}的电子邮件" # 这里将new_mail事件转换为了一个来自环境的message + +[sub_workflows."嘉宾对接组".roles.leader] +name = "leader" +fullname = "嘉宾对接组组长" +agent="email_reader" +[[sub_workflows."嘉宾对接组".roles.leader.prompt]] +role="system" +content="""你是一家活动策划公司的嘉宾对接组的组长，你的工作是基于已知信息,当前活动信息、公司经理的指令与嘉宾沟通，收集嘉宾的信息，然后将信息反馈给经理。在你看来，参加活动的多少有成员都是嘉宾，你可以通过你知道的信息给不同的成员进行分级。你的基本工作模式是： +1. 处理收到的邮件，如果邮件来自嘉宾，你会尝试从邮件的表态和内容中分享嘉宾的需要，并结合你对当前活动方案的理解判断是否需要和经理沟通，如果需要和经理沟通，你会将嘉宾的需求总结和告诉经理。不需要沟通的事项可以直接回复嘉宾。 +2. 你总是通过`call_function(get_env,'parent.topic'`的形式查询当前的活动方案。等待函数返回后，你会根据函数的返回结果继续处理上一个对话。 +3. 当你决定要和经理通信时，请使用`send_msg(manager,{内容}`）的形式，内容的长度不超过200字。 +4. 当你决定要回复嘉宾时，请使用`call_function(sendmail,{嘉宾邮件地址},{标题},{内容}）的形式，内容的长度不超过500字。""" +# 这里是孤立工作模式，组长只和经理沟通，也可以赋予其和其它组沟通的能力 diff --git a/rootfs/workflows/workflows.cfg b/rootfs/workflows/workflows.cfg new file mode 100644 index 0000000..f8a61a0 --- /dev/null +++ b/rootfs/workflows/workflows.cfg @@ -0,0 +1,2 @@ +main = "./" +cache = "./.workflows" \ No newline at end of file diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index 04d4cc5..bd1434b 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -1,7 +1,10 @@ from .environment import Environment,EnvironmentEvent -from .agent import AgentMsg,AIAgent,AIAgentTemplete,AgentMsgState,AgentPrompt,AgentMsgState +from .agent_message import AgentMsg,AgentMsgState +from .chatsession import AIChatSession +from .agent import AIAgent,AIAgentTemplete,AgentPrompt from .compute_kernel import ComputeKernel,ComputeTask from .compute_node import ComputeNode,LocalComputeNode from .open_ai_node import OpenAI_ComputeNode from .role import AIRole,AIRoleGroup -from .workflow import Workflow \ No newline at end of file +from .workflow import Workflow +from .bus import AIBus \ No newline at end of file diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 8346649..2680695 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -1,51 +1,17 @@ from typing import Optional -from enum import Enum + from asyncio import Queue import asyncio import logging import uuid import time +from .agent_message import AgentMsg +from .chatsession import AIChatSession logger = logging.getLogger(__name__) -class AgentMsgState(Enum): - RESPONSED = 0 - INIT = 1 - SENDING = 2 - PROCESSING = 3 - ERROR = 4 - -class AgentMsg: - def __init__(self) -> None: - self.create_time = 0 - self.sender:str = None - self.target:str = None - self.body:str = None - self.state = AgentMsgState.INIT - - self.resp_msg = None - - def set(self,sender:str,target:str,body:str) -> None: - self.sender = sender - self.target = target - self.body = body - self.create_time = time.time() - - def get_msg_id(self) -> str: - pass - - def get_sender(self) -> str: - return self.sender - - def get_target(self) -> str: - return self.target - - # return workflow_name, role_name, session_id - def parser_target(self,target:str) -> None: - pass - class AgentPrompt: def __init__(self) -> None: self.messages = [] @@ -59,6 +25,9 @@ class AgentPrompt: return result_str def append(self,prompt): + if prompt is None: + return + self.messages.extend(prompt.messages) def load_from_config(self,config:list) -> bool: @@ -69,28 +38,6 @@ class AgentPrompt: self.messages = config return True -# chat session store the chat history between owner and agent -# chat session might be large, so can read / write at stream mode. -class AIChatSession: - def __init__(self,owner_id) -> None: - self.owner_id = owner_id - - def get_owner_id(self) -> str: - return self.owner_id - - def append_post(self,msg:AgentMsg) -> None: - """append msg to session, msg is post from session (owner => msg.target)""" - pass - - def append_recv(self,msg:AgentMsg) -> None: - """append msg to session, msg is recv from msg'sender (msg.sender => owner)""" - pass - - def attach_event_handler(self,handler) -> None: - """chat session changed event handler""" - pass - - #TODO : add iterator interface for read chat history class AIAgentTemplete: def __init__(self) -> None: @@ -121,14 +68,14 @@ class AIAgent: def __init__(self) -> None: self.prompt:AgentPrompt = None self.llm_model_name:str = None - self.max_token_size:int = 0 + self.max_token_size:int = 3600 self.instance_id:str = None self.template_id:str = None self.fullname:str = None self.powerby = None self.enable = True - self.chat_sessions = {} + self.chat_db = None self.unread_msg = Queue() # msg from other agent @classmethod @@ -170,27 +117,6 @@ class AIAgent: return True - def post_msg(self,msg:AgentMsg) -> None: - # TODO: drop same msg already processed - msg.state = AgentMsgState.SENDING - self.unread_msg.put_nowait(msg) - - def start(self) -> None: - async def _process_msg_loop(): - while True: - msg = await self.unread_msg.get() - if msg is None: - continue - msg.state = AgentMsgState.PROCESSING - resp_msg = await self._process_msg(msg) - if resp_msg is None: - msg.state = AgentMsgState.ERROR - continue - else: - msg.state = AgentMsgState.RESPONSED - msg.resp_msg = resp_msg - - asyncio.create_task(_process_msg_loop()) def _get_llm_result_type(self,result:str) -> str: if result == "ignore": @@ -200,16 +126,19 @@ class AIAgent: async def _process_msg(self,msg:AgentMsg) -> AgentMsg: from .compute_kernel import ComputeKernel - + session_topic = msg.get_sender() + "#" + msg.topic + chatsession = AIChatSession.get_session(self.instance_id,session_topic,self.chat_db) prompt = AgentPrompt() prompt.append(self.prompt) - msg_prompt = AgentPrompt() - msg_prompt.messages = [{"role":msg.sender,"content":msg.body}] - prompt.append(msg_prompt) + # prompt.append(self._get_function_prompt(the_role.get_name())) # prompt.append(self._get_knowlege_prompt(the_role.get_name())) - # prompt.append(await self._get_prompt_from_session(chatsession,the_role.get_name())) # chat context - + prompt.append(await self._get_prompt_from_session(chatsession)) # chat context + + msg_prompt = AgentPrompt() + msg_prompt.messages = [{"role":"user","content":msg.body}] + prompt.append(msg_prompt) + result = await ComputeKernel().do_llm_completion(prompt,self.llm_model_name,self.max_token_size) final_result = result result_type : str = self._get_llm_result_type(result) @@ -257,12 +186,13 @@ class AIAgent: if is_ignore is not True: # TODO : how to get inner chat session? - chatsession = self.get_chat_session(msg.sender) resp_msg = AgentMsg() resp_msg.set(self.instance_id,msg.sender,final_result) + resp_msg.topic = msg.topic + if chatsession is not None: chatsession.append_recv(msg) - chatsession.append_post(final_result) + chatsession.append_post(resp_msg) return resp_msg @@ -277,25 +207,20 @@ class AIAgent: def get_template_id(self) -> str: return self.template_id - def get_chat_session_for_msg(self,msg:AgentMsg) -> AIChatSession: - pass - - def get_chat_session(self,remote:str,topic_name:str=None) -> AIChatSession: - if topic_name is None: - topic_name = "_" - - result_session = self.chat_sessions.get(topic_name + "@" + remote) - if result_session is not None: - return result_session - - result_session = AIChatSession(self) - self.chat_sessions[topic_name + "@" + remote] = result_session - return result_session - - def get_llm_model_name(self) -> str: return self.llm_model_name def get_max_token_size(self) -> int: return self.max_token_size + + async def _get_prompt_from_session(self,chatsession:AIChatSession) -> AgentPrompt: + messages = chatsession.read_history() # read last 10 message + result_prompt = AgentPrompt() + for msg in reversed(messages): + if msg.target == chatsession.owner_id: + result_prompt.messages.append({"role":"user","content":f"{msg.sender}:{msg.body}"}) + if msg.sender == chatsession.owner_id: + result_prompt.messages.append({"role":"assistant","content":msg.body}) + + return result_prompt diff --git a/src/aios_kernel/agent_message.py b/src/aios_kernel/agent_message.py new file mode 100644 index 0000000..3aa05c1 --- /dev/null +++ b/src/aios_kernel/agent_message.py @@ -0,0 +1,39 @@ +from enum import Enum +import uuid +import time + +class AgentMsgState(Enum): + RESPONSED = 0 + INIT = 1 + SENDING = 2 + PROCESSING = 3 + ERROR = 4 + +class AgentMsg: + def __init__(self) -> None: + self.create_time = 0 + self.sender:str = None + self.target:str = None + self.body:str = None + self.topic:str = "T#" + uuid.uuid4().hex + #self.msg_type = 0 + self.state = AgentMsgState.INIT + self.resp_msg = None + + def set(self,sender:str,target:str,body:str,topic:str=None) -> None: + self.id = "msg#" + uuid.uuid4().hex + self.sender = sender + self.target = target + self.body = body + self.create_time = time.time() + if topic: + self.topic = topic + + def get_msg_id(self) -> str: + return self.id + + def get_sender(self) -> str: + return self.sender + + def get_target(self) -> str: + return self.target diff --git a/src/aios_kernel/bus.py b/src/aios_kernel/bus.py new file mode 100644 index 0000000..6c92a8a --- /dev/null +++ b/src/aios_kernel/bus.py @@ -0,0 +1,129 @@ +from typing import Any +from .agent_message import AgentMsg,AgentMsgState +import asyncio +from asyncio import Queue + +import logging + +logger = logging.getLogger(__name__) + +class AIBusHandler: + def __init__(self,handler:Any) -> None: + self.handler = handler + self.working_task = None + self.results = {} + self.queue:Queue = Queue() + + async def handle_message(self,msg:AgentMsg) -> Any: + if self.handler is None: + return None + + return await self.handler(msg) + + +class AIBus: + _instance = None + @classmethod + def get_default_bus(cls): + if cls._instance is None: + cls._instance = AIBus() + return cls._instance + + def __init__(self) -> None: + self.handlers = {} + self.unhandle_handler = None + + async def post_message(self,target_id,msg:AgentMsg,use_unhandle=True) -> bool: + handler = self.handlers.get(target_id) + if handler: + handler.queue.put_nowait(msg) + self.start_process(target_id) + return True + + if use_unhandle: + if self.unhandle_handler is not None: + if await self.unhandle_handler(self,msg): + return await self.post_message(target_id,msg,False) + + logger.warn(f"post message to {msg.target} failed!,target not found") + return False + + def resp_message(self,my_id:str,org_msg_id:str,resp:AgentMsg) -> None: + handler = self.handlers.get(my_id) + if handler is None: + return None + handler.results[org_msg_id] = resp + + async def get_message_resp(self,name:str,msg_id:str) -> AgentMsg: + handler = self.handlers.get(name) + if handler is None: + return None + + return handler.results.get(msg_id) + + async def send_message(self,target_id:str,msg:AgentMsg) -> AgentMsg: + post_result = await self.post_message(target_id,msg) + if post_result is False: + return None + + handler = self.handlers.get(target_id) + if handler is None: + return None + + retry_times = 0 + while True: + resp = handler.results.get(msg.id) + if resp is not None: + msg.resp_msg = resp + msg.state = AgentMsgState.RESPONSED + return resp + + await asyncio.sleep(0.2) + retry_times += 1 + if retry_times > 100: + msg.state = AgentMsgState.ERROR + return None + + return None + + def register_unhandle_message_handler(self,handler:Any) -> Queue: + self.unhandle_handler = handler + + # means sub + def register_message_handler(self,handler_name:str,handler:Any) -> Queue: + handler_node = AIBusHandler(handler) + self.handlers[handler_name] = handler_node + return handler_node.queue + + async def process_queue(self, handler:AIBusHandler): + while True: + # Wait for a message + message = await handler.queue.get() + + try: + # Try to handle the message + result = await handler.handle_message(message) + handler.results[message.id] = result + except Exception as e: + # If an error occurs, put the message back into the queue + logger.error(f"handle message {message.id} failed! {e}") + #self.queues[name].put_nowait(message) + + return + + def start_process(self,target_name): + handler = self.handlers.get(target_name) + if handler is None: + logger.error(f"handler {target_name} not found!") + return + + if handler.handler is None: + return + + if handler.working_task is not None: + logger.warn(f"handler {target_name} is already working!") + return + + handler.working_task = asyncio.create_task(self.process_queue(handler)) + + \ No newline at end of file diff --git a/src/aios_kernel/chatsession.py b/src/aios_kernel/chatsession.py new file mode 100644 index 0000000..90890ed --- /dev/null +++ b/src/aios_kernel/chatsession.py @@ -0,0 +1,253 @@ + +import sqlite3 # Because sqlite3 IO operation is small, so we can use sqlite3 directly.(so we don't need to use async sqlite3 now) +from sqlite3 import Error +import logging +import threading +import datetime +import uuid + +from .agent_message import AgentMsg + +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 + + 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: + logging.error("Error occurred while connecting to database: %s", e) + return None + + if conn: + self._create_table(conn) + + return conn + + def close(self): + if not hasattr(self.local, 'conn'): + return + self.local.conn.close() + + def _create_table(self, conn): + """ create table """ + try: + # create sessions table + conn.execute(""" + CREATE TABLE IF NOT EXISTS ChatSessions ( + SessionID TEXT PRIMARY KEY, + SessionOwner TEXT, + SessionTopic TEXT, + StartTime TEXT + ); + """) + + # create messages table + conn.execute(""" + CREATE TABLE IF NOT EXISTS Messages ( + MessageID TEXT PRIMARY KEY, + SessionID TEXT, + SenderID TEXT, + ReceiverID TEXT, + Timestamp TEXT, + Content TEXT, + Status INTEGER + ); + """) + conn.commit() + except Error as e: + logging.error("Error occurred while creating tables: %s", e) + + def insert_chatsession(self, session_id, session_owner,session_topic, start_time): + """ insert a new session into the ChatSessions table """ + try: + conn = self._get_conn() + conn.execute(""" + INSERT INTO ChatSessions (SessionID, SessionOwner,SessionTopic, StartTime) + VALUES (?,?, ?, ?) + """, (session_id, session_owner,session_topic, start_time)) + conn.commit() + return 0 # return 0 if successful + except Error as e: + logging.error("Error occurred while inserting session: %s", e) + return -1 # return -1 if an error occurs + + def insert_message(self, message_id, session_id, sender_id, receiver_id, timestamp, content, status): + """ insert a new message into the Messages table """ + try: + conn = self._get_conn() + conn.execute(""" + INSERT INTO Messages (MessageID, SessionID, SenderID, ReceiverID, Timestamp, Content, Status) + VALUES (?, ?, ?, ?, ?, ?, ?) + """, (message_id, session_id, sender_id, receiver_id, timestamp, content, status)) + conn.commit() + return 0 # return 0 if successful + except Error as e: + logging.error("Error occurred while inserting message: %s", e) + return -1 # return -1 if an error occurs + + def get_chatsession_by_id(self, session_id): + """Get a message by its ID""" + conn = self._get_conn() + c = conn.cursor() + c.execute("SELECT * FROM ChatSessions WHERE SessionID = ?", (session_id,)) + chatsession = c.fetchone() + return chatsession + + def get_chatsession_by_owner_topic(self, owner_id, topic): + """Get a chatsession by its owner and topic""" + conn = self._get_conn() + c = conn.cursor() + c.execute("SELECT * FROM ChatSessions WHERE SessionOwner = ? AND SessionTopic = ?", (owner_id,topic)) + chatsession = c.fetchone() + return chatsession + + def get_chatsessions(self, limit, offset): + """ retrieve sessions with pagination """ + try: + conn = self._get_conn() + cursor = conn.cursor() + cursor.execute(""" + SELECT * FROM ChatSessions + ORDER BY StartTime DESC + LIMIT ? OFFSET ? + """, (limit, offset)) + results = cursor.fetchall() + #self.close() + return results # return 0 and the result if successful + except Error as e: + logging.error("Error occurred while getting sessions: %s", e) + return -1, None # return -1 and None if an error occurs + + def get_message_by_id(self, message_id): + """Get a message by its ID""" + conn =self._get_conn() + c = conn.cursor() + c.execute("SELECT MessageID,SessionID,SenderID,ReceiverID,Timestamp,Content,Status FROM Messages WHERE MessageID = ?", (message_id,)) + message = c.fetchone() + return message + + def get_messages(self, session_id, limit, offset): + """ retrieve messages of a session with pagination """ + try: + conn = self._get_conn() + cursor = conn.cursor() + cursor.execute(""" + SELECT MessageID,SessionID,SenderID,ReceiverID,Timestamp,Content,Status FROM Messages + WHERE SessionID = ? + ORDER BY Timestamp DESC + LIMIT ? OFFSET ? + """, (session_id, limit, offset)) + results = cursor.fetchall() + #self.close() + return results # return 0 and the result if successful + except Error as e: + logging.error("Error occurred while getting messages: %s", e) + return -1, None # return -1 and None if an error occurs + + def update_message_status(self, message_id, status): + """ update the status of a message """ + try: + conn = self._get_conn() + conn.execute(""" + UPDATE Messages + SET Status = ? + WHERE MessageID = ? + """, (status, message_id)) + conn.commit() + return 0 # return 0 if successful + except Error as e: + logging.error("Error occurred while updating message status: %s", e) + return -1 # return -1 if an error occurs + + +# chat session store the chat history between owner and agent +# chat session might be large, so can read / write at stream mode. +class AIChatSession: + _dbs = {} + #@classmethod + #async def get_session_by_id(cls,session_id:str,db_path:str): + # db = cls._dbs.get(db_path) + # if db is None: + # db = ChatSessionDB(db_path) + # cls._dbs[db_path] = db + # db.get_chatsession_by_id(session_id) + # #result = AIChatSession() + + @classmethod + def get_session(cls,owner_id:str,session_topic:str,db_path:str,auto_create = True) -> str: + db = cls._dbs.get(db_path) + if db is None: + db = ChatSessionDB(db_path) + cls._dbs[db_path] = db + + result = None + session = db.get_chatsession_by_owner_topic(owner_id,session_topic) + if session is None: + if auto_create: + session_id = "CS#" + uuid.uuid4().hex + db.insert_chatsession(session_id,owner_id,session_topic,datetime.datetime.now()) + result = AIChatSession(owner_id,session_id,db) + else: + result = AIChatSession(owner_id,session[0],db) + result.topic = session_topic + + return result + + + def __init__(self,owner_id:str, session_id:str, db:ChatSessionDB) -> None: + self.owner_id :str = owner_id + self.session_id : str = session_id + self.db : ChatSessionDB = db + + self.topic : str = None + self.start_time : str = None + + def get_owner_id(self) -> str: + return self.owner_id + + def read_history(self, number:int=10,offset=0) -> [AgentMsg]: + msgs = self.db.get_messages(self.session_id, number, offset) + result = [] + for msg in msgs: + agent_msg = AgentMsg() + agent_msg.id = msg[0] + agent_msg.sender = msg[2] + agent_msg.target = msg[3] + agent_msg.create_time = msg[4] + agent_msg.body = msg[5] + # agent_msg.state = msg[6] + + result.append(agent_msg) + return result + + def append(self,msg:AgentMsg) -> None: + self.db.insert_message(msg.id,self.session_id,msg.sender,msg.target,msg.create_time,msg.body,0) + + def append_post(self,msg:AgentMsg) -> None: + """append msg to session, msg is post from session (owner => msg.target)""" + assert msg.sender == self.owner_id,"post message means msg.sender == self.owner_id" + self.append(msg) + + + def append_recv(self,msg:AgentMsg) -> None: + """append msg to session, msg is recv from msg'sender (msg.sender => owner)""" + assert msg.target == self.owner_id,"recv message means msg.target == self.owner_id" + self.append(msg) + + #def attach_event_handler(self,handler) -> None: + # """chat session changed event handler""" + # pass + + #TODO : add iterator interface for read chat history diff --git a/src/aios_kernel/compute_kernel.py b/src/aios_kernel/compute_kernel.py index 84351a7..1d05afd 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/src/aios_kernel/compute_kernel.py @@ -19,16 +19,13 @@ class ComputeKernel: if cls._instance is None: cls._instance = super().__new__(cls) cls._instance.is_start = False - else: - print("ComputeKernel is already created!") + return cls._instance def __init__(self) -> None: if self.is_start is True: - print("ComputeKernel is already start!") return - print("init ComputeKernel!!!") self.is_start = True self.task_queue = Queue() self.is_start = False diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index cf88815..7f5c8e6 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -54,7 +54,7 @@ class OpenAI_ComputeNode(ComputeNode): logger.info(f"call openai {mode_name} prompts: {prompts}") resp = openai.ChatCompletion.create(model=mode_name, messages=prompts, - max_tokens=2000, + max_tokens=4000, temperature=1.2) logger.info(f"openai response: {resp}") @@ -69,7 +69,7 @@ class OpenAI_ComputeNode(ComputeNode): result.worker_id = self.node_id result.result_str = resp["choices"][0]["message"]["content"] result.result = resp["choices"][0]["message"] - + return result def start(self): diff --git a/src/aios_kernel/role.py b/src/aios_kernel/role.py index 0b22a95..c4d97ab 100644 --- a/src/aios_kernel/role.py +++ b/src/aios_kernel/role.py @@ -1,25 +1,66 @@ -from .agent import AIAgent +import logging + +from .agent import AIAgent,AgentPrompt class AIRole: def __init__(self) -> None: self.agent_instance_id : str = None self.role_name : str = None - self.agent : AIAgent = None + self.fullname : str = None + self.agent_name : str = None + self.prompt : AgentPrompt = None self.introduce : str = None + self.agent = None def load_from_config(self,config:dict) -> bool: - pass + name_node = config.get("name") + if name_node is None: + logging.error("role name is not found!") + return False + self.role_name = name_node + agent_id_node = config.get("agent") + if agent_id_node is None: + logging.error("agent id is not found!") + return False + self.agent_name = agent_id_node + + prompt_node = config.get("prompt") + if prompt_node: + self.prompt = AgentPrompt() + if self.prompt.load_from_config(prompt_node) is False: + logging.error("load prompt failed!") + return False + + intro_node = config.get("intro") + if intro_node is not None: + self.introduce = intro_node + def get_intro(self) -> str: return self.introduce def get_name(self) -> str: return self.role_name + def get_prompt(self) -> AgentPrompt: + return self.prompt + class AIRoleGroup: def __init__(self) -> None: - self.roles : dict[str,str]= None - pass + self.roles : dict[str,AIRole] = {} + + def load_from_config(self,config:dict) -> bool: + for k,v in config.items(): + role = AIRole() + if role.load_from_config(v) is False: + logging.error(f"load role {k} failed!") + return False + + self.roles[k] = role + + return True + def get(self,role_name:str) -> AIRole: + return self.roles.get(role_name) \ No newline at end of file diff --git a/src/aios_kernel/workflow.py b/src/aios_kernel/workflow.py index 1c842b8..b3c3127 100644 --- a/src/aios_kernel/workflow.py +++ b/src/aios_kernel/workflow.py @@ -6,146 +6,217 @@ from typing import Optional,Tuple from abc import ABC, abstractmethod from .environment import Environment,EnvironmentEvent -from .agent import AgentPrompt,AgentMsg,AIChatSession -from .role import AIRole +from .agent_message import AgentMsg,AgentMsgState +from .agent import AgentPrompt,AgentMsg +from .chatsession import AIChatSession +from .role import AIRole,AIRoleGroup from .ai_function import CallChain from .compute_kernel import ComputeKernel +from .bus import AIBus logger = logging.getLogger(__name__) class MessageFilter: def __init__(self) -> None: - pass - def select(self,msg:AgentMsg) -> AIRole: - pass + self.filters = {} + + def select(self,msg:AgentMsg) -> str: + star_target = self.filters.get("*") + if star_target is not None: + return star_target + + # TODO: add more filter + return None + + def load_from_config(self,config:dict) -> bool: + self.filters = config + return True class Workflow: def __init__(self) -> None: + self.workflow_name : str = None self.rule_prompt : AgentPrompt = None self.workflow_config = None - self.role_group = None + self.role_group : dict = None self.input_filter : MessageFilter= None - self.msg_queue = Queue() self.connected_environment = {} + self.sub_workflows = {} + self.owner_workflow = None + self.db_file = None + + self.is_start = False + self.msg_queue = Queue() + + def get_bus(self) -> AIBus: + return AIBus.get_default_bus() + + def set_owner(self,owner): + self.owner_workflow = owner + + def load_from_config(self,config:dict) -> bool: + if config is None: + return False - def load_from_disk(self,config_path:str,context_dir_path) -> int: - pass + if config.get("name") is None: + logger.error("workflow config must have name") + return False + self.workflow_name = config.get("name") - #workflow is asynchronous. - # When processing one message, it can process another message at the same time. - # chatsession is synchronous, it has to wait for the previous message to finish processing before it can process the next message. - # Therefore, post a message needs to specify the session_id explicitly, if not specified it will be automatically created by workflow. - def post_msg(self,msg:AgentMsg) -> None: - self.msg_queue.put_nowait(msg) - return - - async def send_msg(self,msg:AgentMsg) -> str: - pass - - async def run(self): - # TODO add tracking design of msg processing - while True: - the_msg = await self._pop_msg() - chatsession:AIChatSession = self._get_chat_session_for_msg(the_msg) - if chatsession is None: - logger.error(f"get_chat_session_for_msg return None for :{the_msg}") - continue + #if config.get("rule_prompt") is None: + # logger.error("workflow config must have rule_prompt") + # return False + #self.rule_prompt = AgentPrompt() + #if self.rule_prompt.load_from_config(config.get("rule_prompt")) is False: + # logger.error("Workflow load rule_prompt failed") + # return False + if config.get("roles") is None: + logger.error("workflow config must have roles") + return False + self.role_group = AIRoleGroup() + if self.role_group.load_from_config(config.get("roles")) is False: + logger.error("Workflow load role_group failed") + return False - chatsession.append_recv(the_msg) + if config.get("input_filter") is not None: + self.input_filter = MessageFilter() + if self.input_filter.load_from_config(config.get("input_filter")) is False: + logger.error("Workflow load input_filter failed") + return False + + sub_workflows = config.get("sub_workflows") + if sub_workflows is not None: + if self._load_sub_workflows(sub_workflows) is False: + logger.error("Workflow load sub workflows failed") + return False + + #TODO: load env - async def _process_msg(msg:AgentMsg,the_role) -> None: - # prompt generat progress is most important part of workflow(app) develope - prompt = AgentPrompt() - prompt.append(the_role.get_prompt()) - prompt.append(self.get_workflow_rule_prompt()) - prompt.append(self._get_function_prompt(the_role.get_name())) - prompt.append(self._get_knowlege_prompt(the_role.get_name())) - prompt.append(await self._get_prompt_from_session(chatsession,the_role.get_name())) # chat context + return True - result = await ComputeKernel().do_llm_completion(prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size()) + def _load_sub_workflows(self,config:dict) -> bool: + for k,v in config.items(): + sub_workflow = Workflow() + sub_workflow.set_owner(self) + if sub_workflow.load_from_config(v) is False: + logger.error(f"load sub workflow {k} failed!") + return False + self.sub_workflows[k] = sub_workflow + return True + + async def _process_msg(self,msg:AgentMsg): + final_result = None + chatsession = None + if self.input_filter is not None: + select_role_id = self.input_filter.select(msg) + if select_role_id is not None: + select_role = self.role_group.get(select_role_id) + if select_role is None: + logger.error(f"input_filter return invalid role id:{select_role_id}, role not found in role_group") + return None + + result = await self._role_process_msg(msg,select_role) + if result is None: + logger.error(f"_process_msg return None for :{msg}") + return + if chatsession is not None: + chatsession.append_post(result) final_result = result - result_type : str = self._get_llm_result_type(result) - is_ignore = False - match result_type: - case "function": - callchain:CallChain = self._parse_function_call_chain(result) - resp = await callchain.exec() - if callchain.have_result(): - # generator proc resp prompt with WAITING state - proc_resp_prompt:AgentPrompt = self._get_resp_prompt(resp,msg,the_role,prompt,chatsession) - final_result = await ComputeKernel().do_llm_completion(proc_resp_prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size()) - return final_result - - - case "send_message": - # send message to other / sub workflow - next_msg:AgentMsg = self._parse_to_msg(result) - if next_msg is not None: - # TODO: Next Target can be another role in workflow - next_workflow:Workflow = self.get_workflow(next_msg.get_target()) - inner_chat_session = the_role.agent.get_chat_session(next_msg.get_target(),next_msg.get_session_id()) + else: + logger.error(f"input_filter return None for :{msg}") + return + + else: + results = {} + for this_role in self.role_group.roles.values(): + # TODO : we would do this in parallel + a_result = await self._role_process_msg(msg,this_role) + results[this_role.get_name()] = a_result - inner_chat_session.append_post(next_msg) - resp = await next_workflow.send_msg(next_msg) - inner_chat_session.append_recv(resp) - # generator proc resp prompt with WAITING state - proc_resp_prompt:AgentPrompt = self._get_resp_prompt(resp,msg,the_role,prompt,chatsession) - final_result = await ComputeKernel().do_llm_completion(proc_resp_prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size()) - - return final_result - - case "post_message": - # post message to other / sub workflow - next_msg:AgentMsg = self._parse_to_msg(result) - if next_msg is not None: - next_workflow:Workflow = self.get_workflow(next_msg.get_target()) - inner_chat_session = the_role.agent.get_chat_session(next_msg.get_target(),next_msg.get_session_id()) - inner_chat_session.append_post(next_msg) - next_workflow.post_msg(next_msg) + # merge result from all roles + # TODO: one input msg can have multiple result msg, at this while ,we only support one result msg + final_result:AgentMsg = self._merge_msg_result(results) + if chatsession is not None: + chatsession.append_post(final_result) + + return final_result - case "ignore": - is_ignore = True - - if is_ignore is not True: - # TODO : how to get inner chat session? - inner_chat_session = the_role.agent.get_chat_session_for_msg(msg) - if inner_chat_session is not None: - inner_chat_session.append_input(msg) - inner_chat_session.append_result(final_result) - return result - - async def _workflow_process_msg(msg:AgentMsg) -> None: - final_result = None - if self.input_filter is not None: - select_role = self.input_filter.select(msg) - if select_role is not None: - result = await _process_msg(msg,select_role) - if result is None: - logger.error(f"_process_msg return None for :{msg}") - return - if chatsession is not None: - chatsession.append_post(result) - final_result = result - - else: - results = {} - for this_role in self.role_group.roles: - a_result = asyncio.create_task(_process_msg(msg,this_role)) - results[this_role.get_name()] = a_result - # merge result from all roles - # TODO: one input msg can have multiple result msg, at this while ,we only support one result msg - final_result:AgentMsg = self._merge_msg_result(results) - if chatsession is not None: - chatsession.append_post(final_result) - if final_result is not None: - # TODO post message to source - pass - - asyncio.create_task(_workflow_process_msg(the_msg)) + async def _role_process_msg(self,msg:AgentMsg,the_role:AIRole) -> None: + # TODO : we just record role's chatsession, but in future, we would record workflow's chatsession(like a groupo chat) + session_topic = f"{the_role.get_name()}#{msg.sender}#{msg.topic}" + chatsession = AIChatSession.get_session(self.workflow_name,session_topic,self.db_file) + if chatsession is None: + logger.error(f"get session {session_topic}@{self.workflow_name} failed!") + return None + + # prompt generat progress is most important part of workflow(app) develope + prompt = AgentPrompt() + prompt.append(the_role.agent.prompt) + prompt.append(the_role.get_prompt()) + + # prompt.append(self.get_workflow_rule_prompt()) + # prompt.append(self._get_function_prompt(the_role.get_name())) + # prompt.append(self._get_knowlege_prompt(the_role.get_name())) + + prompt.append(await self._get_prompt_from_session(chatsession)) + #prompt.append(await self._get_prompt_from_session(chatsession,the_role.get_name())) # chat context + + msg_prompt = AgentPrompt() + msg_prompt.messages = [{"role":"user","content":msg.body}] + prompt.append(msg_prompt) + + result = await ComputeKernel().do_llm_completion(prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size()) + chatsession.append_recv(msg) + final_result = result + + result_type : str = self._get_llm_result_type(result) + is_ignore = False + match result_type: + case "function": + callchain:CallChain = self._parse_function_call_chain(result) + resp = await callchain.exec() + if callchain.have_result(): + # generator proc resp prompt with WAITING state + proc_resp_prompt:AgentPrompt = self._get_resp_prompt(resp,msg,the_role,prompt,chatsession) + final_result = await ComputeKernel().do_llm_completion(proc_resp_prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size()) + return final_result + + + case "send_message": + # send message to other / sub workflow + next_msg:AgentMsg = self._parse_to_msg(result) + if next_msg is not None: + next_msg.sender = self.workflow_name + logger.info(f"W#{self.workflow_name} send message to {next_msg.get_target()}") + resp_msg = await self.get_bus().send_message(next_msg.get_target(),next_msg) + if resp_msg is not None: + msg_prompt = AgentPrompt() + msg_prompt.messages = [{"role":"assistant","content":result},{"role":"user","content":f"{next_msg.get_target()}:{resp_msg.body}"}] + + final_result = await ComputeKernel().do_llm_completion(proc_resp_prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size()) + + + case "post_message": + # post message to other / sub workflow + next_msg:AgentMsg = self._parse_to_msg(result) + if next_msg is not None: + next_msg.sender = self.workflow_name + logger.info(f"W#{self.workflow_name} post message to {next_msg.get_target()}") + self.get_bus().post_message(next_msg.get_target(),next_msg) + + case "ignore": + is_ignore = True + + if is_ignore: + return None + + resp_msg = AgentMsg() + resp_msg.set(self.workflow_name,msg.sender,final_result) + chatsession.append_post(resp_msg) + return resp_msg async def _pop_msg(self) -> AgentMsg: pass @@ -153,14 +224,25 @@ class Workflow: def _get_chat_session_for_msg(self,msg:AgentMsg) -> AIChatSession: pass - async def _get_prompt_from_session(self,chatsession:AIChatSession,role_name:str) -> AgentPrompt: - pass + async def _get_prompt_from_session(self,chatsession:AIChatSession) -> AgentPrompt: + messages = chatsession.read_history() # read last 10 message + result_prompt = AgentPrompt() + for msg in reversed(messages): + if msg.target == chatsession.owner_id: + result_prompt.messages.append({"role":"user","content":f"{msg.sender}:{msg.body}"}) + if msg.sender == chatsession.owner_id: + result_prompt.messages.append({"role":"assistant","content":msg.body}) + + return result_prompt def _get_msg_queue(self,session_id:str): pass def _merge_msg_result(self,results:dict) -> AgentMsg: - pass + # TODO: one input msg can have multiple result msg, at this while ,we only support one result msg + for k,v in results.items(): + if v is not None: + return v def _get_function_prompt(self,role_name:str) -> AgentPrompt: pass @@ -168,20 +250,48 @@ class Workflow: def _get_knowlege_prompt(self,role_name:str) -> AgentPrompt: pass - def _get_resp_prompt(self,resp:str,msg:AgentMsg,role:AIRole,prompt:AgentPrompt,chatsession:AIChatSession) -> AgentPrompt: + def _get_resp_prompt(self,resp:str,msg:AgentMsg,role:AIRole,prompt:AgentPrompt) -> AgentPrompt: pass def get_workflow_rule_prompt(self) -> AgentPrompt: return self.rule_prompt def _get_llm_result_type(self,llm_resp_str:str) -> str: - pass + if llm_resp_str == "ignore": + return "ignore" + + if llm_resp_str.find("sendmsg(") != -1: + return "send_message" + + if llm_resp_str.find("postmsg(") != -1: + return "post_message" + + if llm_resp_str.find("call(") != -1: + return "function" + + return "text" def _parse_function_call_chain(self,llm_resp_str) -> CallChain: pass def _parse_to_msg(self,llm_resp_str) -> AgentMsg: - pass + lines = llm_resp_str.splitlines() + for line in lines: + if line.startswith("sendmsg("): + line = line[8:] + _index = line.find(",") + msg = AgentMsg() + msg.set("",line[:_index],line[_index+1:]) + return msg + + if line.startswith("postmsg("): + line = line[8:] + _index = line.find(",") + msg = AgentMsg() + msg.set("",line[:_index],line[_index+1:]) + return msg + + return None def get_workflow(self,workflow_name:str): """get workflow from known workflow list or sub workflow list""" diff --git a/src/component/agent_manager/agent_manager.py b/src/component/agent_manager/agent_manager.py index 3d1284f..9b05b62 100644 --- a/src/component/agent_manager/agent_manager.py +++ b/src/component/agent_manager/agent_manager.py @@ -10,10 +10,6 @@ logger = logging.getLogger(__name__) class AgentManager: _instance = None - - def __init__(self) -> None: - self.loaded_agent_instance = {} - pass def __new__(cls): if cls._instance is None: @@ -22,8 +18,10 @@ class AgentManager: def initial(self,root_dir:str) -> None: - self.agent_templete_env : PackageEnv = PackageEnvManager().get_env(f"{root_dir}templetes/templetes.cfg") - self.agent_env : PackageEnv = PackageEnvManager().get_env(f"{root_dir}agents/agents.cfg") + self.agent_templete_env : PackageEnv = PackageEnvManager().get_env(f"{root_dir}/templetes/templetes.cfg") + self.agent_env : PackageEnv = PackageEnvManager().get_env(f"{root_dir}/agents/agents.cfg") + self.db_path = f"{root_dir}/agents_chat.db" + self.loaded_agent_instance = {} if self.agent_templete_env is None: raise Exception("agent_manager initial failed") @@ -41,8 +39,8 @@ class AgentManager: the_agent : AIAgent = await self._load_agent_from_media(agent_media_info) if the_agent is None: logger.warn(f"load agent {agent_id} from media failed!") - else: - the_agent.start() + + the_agent.chat_db = self.db_path return the_agent def remove(self,agent_id:str)->int: diff --git a/src/component/workflow_manager/workflow_manager.py b/src/component/workflow_manager/workflow_manager.py index 2a6a68d..28479dd 100644 --- a/src/component/workflow_manager/workflow_manager.py +++ b/src/component/workflow_manager/workflow_manager.py @@ -1,11 +1,74 @@ +import logging +import toml from aios_kernel import Workflow +from package_manager import PackageEnv,PackageEnvManager,PackageMediaInfo,PackageInstallTask +from agent_manager import AgentManager +logger = logging.getLogger(__name__) +import os class WorkflowManager: - def __init__(self) -> None: - pass - + _instance = None + + def __new__(cls): + if cls._instance is None: + cls._instance = super(WorkflowManager, cls).__new__(cls) + return cls._instance + + def initial(self,root_dir:str) -> None: + self.loaded_workflow = {} + self.workflow_env = PackageEnvManager().get_env(f"{root_dir}/workflows.cfg") + self.db_file = os.path.abspath(f"{root_dir}/workflows.db") + if self.workflow_env is None: + raise Exception("WorkflowManager initial failed") + + async def get_agent_default_workflow(self,agent_id:str) -> Workflow: pass - def get_workflow(self,workflow_id:str) -> Workflow: - pass \ No newline at end of file + async def get_workflow(self,workflow_id:str) -> Workflow: + the_workflow : Workflow = self.loaded_workflow.get(workflow_id) + if the_workflow: + return the_workflow + + # try load from disk + workflow_media_info = self.workflow_env.load(workflow_id) + if workflow_media_info is None: + return None + + the_workflow = await self._load_workflow_from_media(workflow_media_info) + if the_workflow is None: + logger.warn(f"load workflow {workflow_id} from media failed!") + + for v in the_workflow.role_group.roles.values(): + agent = await AgentManager().get(v.agent_name) + if agent is None: + logger.error(f"load agent {v.agent_name} failed!") + return None + v.agent = agent + + + return the_workflow + + async def _load_workflow_from_media(self,workflow_media:PackageMediaInfo) -> Workflow: + reader = self.workflow_env._create_media_loader(workflow_media) + if reader is None: + logger.error(f"create media loader for {workflow_media} failed!") + return None + + try: + config_file = await reader.read("workflow.toml","r") + if config_file is None: + logger.error(f"read workflow config from {workflow_media} failed!") + return None + + config_data = await config_file.read() + config = toml.loads(config_data) + result_workflow = Workflow() + if result_workflow.load_from_config(config) is False: + logger.error(f"load workflow from {workflow_media} failed!") + return None + result_workflow.db_file = self.db_file + return result_workflow + except Exception as e: + logger.error(f"read workflow.toml cfg from {workflow_media} failed! unexpected error occurred: {str(e)}") + return None \ 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 7d989ca..fa48bdd 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -17,23 +17,42 @@ from prompt_toolkit.completion import WordCompleter directory = os.path.dirname(__file__) sys.path.append(directory + '/../../') -from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgState,ComputeKernel,OpenAI_ComputeNode +from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgState,ComputeKernel,OpenAI_ComputeNode,AIBus sys.path.append(directory + '/../../component/') from agent_manager import AgentManager from workflow_manager import WorkflowManager + class AIOS_Shell: def __init__(self,username:str) -> None: self.username = username + self.user_chatsession = {} + async def _handle_no_target_msg(self,bus:AIBus,msg:AgentMsg) -> bool: + agent : AIAgent = await AgentManager().get(msg.target) + if agent is not None: + bus.register_message_handler(msg.target,agent._process_msg) + return True + + a_workflow = await WorkflowManager().get_workflow(msg.target) + if a_workflow is not None: + bus.register_message_handler(msg.target,a_workflow._process_msg) + for subflow in a_workflow.sub_workflows.values(): + bus.register_message_handler(subflow.workflow_name,subflow._process_msg) + return True + + return False + + async def initial(self) -> bool: - AgentManager().initial(directory + "/../../../rootfs/") - WorkflowManager().initial(directory + "/../../../../rootfs/workflows/workflows.cfg") + AgentManager().initial(os.path.abspath( directory + "/../../../rootfs/")) + WorkflowManager().initial(os.path.abspath(directory + "/../../../rootfs/workflows/")) open_ai_node = OpenAI_ComputeNode() open_ai_node.start() ComputeKernel().add_compute_node(open_ai_node) + AIBus().get_default_bus().register_unhandle_message_handler(self._handle_no_target_msg) return True @@ -43,38 +62,12 @@ class AIOS_Shell: async def send_msg(self,msg:str,target_id:str,sender:str = None) -> str: agent_msg = AgentMsg() agent_msg.set(sender,target_id,msg) - - agent : AIAgent = await AgentManager().get(target_id) - if agent is not None: - agent.post_msg(agent_msg) - - a_workflow = WorkflowManager().get_workflow(target_id) - if a_workflow is not None: - a_workflow.post_msg(agent_msg) - - - async def check_timer(): - check_times = 0 - while True: - if agent_msg.state == AgentMsgState.RESPONSED: - break - - if agent_msg.state == AgentMsgState.ERROR: - break - - if check_times >= 20: - agent_msg.state = AgentMsgState.ERROR - break - - await asyncio.sleep(0.5) - check_times += 1 - - await asyncio.create_task(check_timer()) - if agent_msg.state == AgentMsgState.RESPONSED: - return agent_msg.resp_msg.body - - return "error!" - + agent_msg.topic = "default" + resp = await AIBus().get_default_bus().send_message(target_id,agent_msg) + if resp is not None: + return resp.body + else: + return "error!" async def install_workflow(self,workflow_id:Workflow) -> None: pass @@ -87,21 +80,20 @@ def proc_input_by_agent(): def show_help(): print("this is help") +history = FileHistory('history.txt') +session = PromptSession(history=history) async def main(): print("aios shell prepareing...") - logging.basicConfig(level=logging.INFO, - format='%(asctime)s %(name)s %(levelname)s %(message)s') - + logging.basicConfig(level=logging.INFO,format='[%(asctime)s]%(name)s[%(levelname)s]: %(message)s') shell = AIOS_Shell("user") await shell.initial() print(f"aios shell {shell.get_version()} ready.") - completer = WordCompleter(['list agent', 'list workflow', 'exit', 'help'], ignore_case=True) #history = FileHistory('history.txt') while True: - user_input = await PromptSession().prompt_async('>>> ',completer=completer) + user_input = await session.prompt_async('>>> ',completer=completer) match user_input: case "list agent": print(AgentManager().list_agent()) @@ -121,8 +113,8 @@ async def main(): target_id = args[1] msg_content = args[2] resp = await shell.send_msg(msg_content,target_id,shell.username) - print(f"<<< {resp}") + print(f"<<<{target_id} : {resp}") -if __name__ == "__main__": +if __name__ == "__main__": asyncio.run(main()) diff --git a/test/chatsession_test.py b/test/chatsession_test.py new file mode 100644 index 0000000..c528556 --- /dev/null +++ b/test/chatsession_test.py @@ -0,0 +1,75 @@ +import unittest +import sys +import os +import sqlite3 +directory = os.path.dirname(__file__) +sys.path.append(directory + '/../src') + +from aios_kernel import ChatSessionDB + + +class TestChatDatabase(unittest.TestCase): + + def setUp(self): + """Function to setup the test case""" + self.db_file = 'test_chat.db' + self.chat_db = ChatSessionDB(self.db_file) + + def tearDown(self): + """Function to cleanup after the test case""" + self.chat_db.close() + os.remove(self.db_file) + + def test_database_creation(self): + """Test if the database is created""" + self.assertTrue(os.path.exists(self.db_file)) + + def test_table_creation(self): + """Test if the tables are created in the database""" + conn = sqlite3.connect(self.db_file) + cursor = conn.cursor() + + # Check if ChatSessions table exists + cursor.execute(""" + SELECT count(name) FROM sqlite_master WHERE type='table' AND name='ChatSessions' + """) + self.assertEqual(cursor.fetchone()[0], 1) + + # Check if Messages table exists + cursor.execute(""" + SELECT count(name) FROM sqlite_master WHERE type='table' AND name='Messages' + """) + self.assertEqual(cursor.fetchone()[0], 1) + + conn.close() + + def test_insert_and_get_session(self): + """Test if we can insert and retrieve a session""" + session_id = "session1" + session_owner = "user1" + session_topic = "topic1" + start_time = "2023-08-28 12:00:00" + + self.chat_db.insert_chatsession(session_id,session_owner, session_topic, start_time) + session = self.chat_db.get_chatsession_by_id(session_id) + + self.assertEqual(session, (session_id,session_owner,session_topic, start_time)) + + def test_insert_and_get_message(self): + """Test if we can insert and retrieve a message""" + message_id = "message1" + session_id = "session1" + sender_id = "user1" + receiver_id = "user2" + timestamp = "2023-08-28 12:30:00" + content = "Hello, world!" + status = 0 + + self.chat_db.insert_message(message_id, session_id, sender_id, receiver_id, timestamp, content, status) + message = self.chat_db.get_message_by_id(message_id) + + self.assertEqual(message, (message_id, session_id, sender_id, receiver_id, timestamp, content, status)) + + +if __name__ == '__main__': + unittest.main() \ No newline at end of file