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