1.workflow has reached a state of basic functionality.

2. implemented AI message bus infrastructure,
2023-08-30 12:30:41 -07:00
parent ccbef2104b
commit 05d2e4a9cf
21 changed files with 1146 additions and 297 deletions
@@ -1 +1,2 @@
 .vscode/
+*.pyc
@@ -0,0 +1,136 @@
+<mxfile host="65bd71144e">
+    <diagram id="EAoW1hfBXHFT-SqL8OPS" name="Page-1">
+        <mxGraphModel dx="1563" dy="697" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="1100" pageHeight="850" math="0" shadow="0">
+            <root>
+                <mxCell id="0"/>
+                <mxCell id="1" parent="0"/>
+                <mxCell id="33" value="" style="edgeStyle=none;html=1;" edge="1" parent="1" source="2" target="3">
+                    <mxGeometry relative="1" as="geometry"/>
+                </mxCell>
+                <mxCell id="34" style="edgeStyle=none;html=1;exitX=0.5;exitY=1;exitDx=0;exitDy=0;entryX=0;entryY=0.5;entryDx=0;entryDy=0;" edge="1" parent="1" source="2" target="4">
+                    <mxGeometry relative="1" as="geometry"/>
+                </mxCell>
+                <mxCell id="2" value="User Msg" style="ellipse;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="20" y="230" width="90" height="60" as="geometry"/>
+                </mxCell>
+                <mxCell id="36" value="event/msg" style="edgeStyle=none;html=1;exitX=1;exitY=0.5;exitDx=0;exitDy=0;entryX=0;entryY=0.5;entryDx=0;entryDy=0;" edge="1" parent="1" source="3" target="5">
+                    <mxGeometry relative="1" as="geometry"/>
+                </mxCell>
+                <mxCell id="3" value="Gateway" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="180" y="235" width="100" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="35" style="edgeStyle=none;html=1;exitX=0.5;exitY=0;exitDx=0;exitDy=0;entryX=0.5;entryY=1;entryDx=0;entryDy=0;" edge="1" parent="1" source="4" target="3">
+                    <mxGeometry relative="1" as="geometry"/>
+                </mxCell>
+                <mxCell id="4" value="telegram roboter" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="180" y="340" width="100" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="37" style="edgeStyle=none;html=1;exitX=1;exitY=0.5;exitDx=0;exitDy=0;entryX=0;entryY=0.5;entryDx=0;entryDy=0;" edge="1" parent="1" source="5" target="7">
+                    <mxGeometry relative="1" as="geometry">
+                        <Array as="points">
+                            <mxPoint x="520" y="350"/>
+                        </Array>
+                    </mxGeometry>
+                </mxCell>
+                <mxCell id="5" value="Workflow&lt;br&gt;@docker" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="390" y="230" width="100" height="60" as="geometry"/>
+                </mxCell>
+                <mxCell id="6" value="LLM&lt;br&gt;ComputeNode&lt;br&gt;Run in VM" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="485" y="40" width="100" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="7" value="Agent@docker" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="560" y="330" width="100" height="70" as="geometry"/>
+                </mxCell>
+                <mxCell id="8" value="LLM ComputeNode&lt;br&gt;Run in docker" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="355" y="40" width="100" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="10" value="Node Client" style="shape=process;whiteSpace=wrap;html=1;backgroundOutline=1;" vertex="1" parent="1">
+                    <mxGeometry x="405" y="200" width="70" height="30" as="geometry"/>
+                </mxCell>
+                <mxCell id="11" value="Node Client" style="shape=process;whiteSpace=wrap;html=1;backgroundOutline=1;" vertex="1" parent="1">
+                    <mxGeometry x="575" y="300" width="70" height="30" as="geometry"/>
+                </mxCell>
+                <mxCell id="13" value="FileSystem" style="shape=cylinder3;whiteSpace=wrap;html=1;boundedLbl=1;backgroundOutline=1;size=15;" vertex="1" parent="1">
+                    <mxGeometry x="195" y="700" width="120" height="80" as="geometry"/>
+                </mxCell>
+                <mxCell id="14" value="Object Store" style="shape=cylinder3;whiteSpace=wrap;html=1;boundedLbl=1;backgroundOutline=1;size=15;" vertex="1" parent="1">
+                    <mxGeometry x="345" y="700" width="120" height="80" as="geometry"/>
+                </mxCell>
+                <mxCell id="15" value="RDB" style="shape=cylinder3;whiteSpace=wrap;html=1;boundedLbl=1;backgroundOutline=1;size=15;" vertex="1" parent="1">
+                    <mxGeometry x="495" y="700" width="120" height="80" as="geometry"/>
+                </mxCell>
+                <mxCell id="16" value="VDB" style="shape=cylinder3;whiteSpace=wrap;html=1;boundedLbl=1;backgroundOutline=1;size=15;" vertex="1" parent="1">
+                    <mxGeometry x="645" y="700" width="120" height="80" as="geometry"/>
+                </mxCell>
+                <mxCell id="17" value="Zone States:" style="text;html=1;strokeColor=none;fillColor=none;align=center;verticalAlign=middle;whiteSpace=wrap;rounded=0;" vertex="1" parent="1">
+                    <mxGeometry x="110" y="710" width="70" height="80" as="geometry"/>
+                </mxCell>
+                <mxCell id="32" value="" style="edgeStyle=none;html=1;" edge="1" parent="1" source="18" target="3">
+                    <mxGeometry relative="1" as="geometry"/>
+                </mxCell>
+                <mxCell id="18" value="Environment" style="shape=parallelogram;perimeter=parallelogramPerimeter;whiteSpace=wrap;html=1;fixedSize=1;" vertex="1" parent="1">
+                    <mxGeometry x="167.5" y="80" width="125" height="60" as="geometry"/>
+                </mxCell>
+                <mxCell id="41" value="" style="edgeStyle=none;html=1;" edge="1" parent="1" source="20" target="40">
+                    <mxGeometry relative="1" as="geometry"/>
+                </mxCell>
+                <mxCell id="20" value="Functions" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="575" y="400" width="70" height="40" as="geometry"/>
+                </mxCell>
+                <mxCell id="21" value="Functions" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="405" y="290" width="70" height="40" as="geometry"/>
+                </mxCell>
+                <mxCell id="23" value="email-spieder" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="220" y="600" width="100" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="25" value="embading pipline" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="357.5" y="600" width="155" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="27" value="Workflow,Agent,Env这3个概念都是主机无关的，只要能访问到其状态，任意Zone主机运行都可以" style="text;html=1;strokeColor=none;fillColor=none;align=center;verticalAlign=middle;whiteSpace=wrap;rounded=0;" vertex="1" parent="1">
+                    <mxGeometry x="558" y="180" width="212" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="38" value="Workflow's&lt;br&gt;state" style="shape=note;whiteSpace=wrap;html=1;backgroundOutline=1;darkOpacity=0.05;size=21;" vertex="1" parent="1">
+                    <mxGeometry x="415" y="345" width="60" height="80" as="geometry"/>
+                </mxCell>
+                <mxCell id="39" value="Agent's&lt;br&gt;state" style="shape=note;whiteSpace=wrap;html=1;backgroundOutline=1;darkOpacity=0.05;size=21;" vertex="1" parent="1">
+                    <mxGeometry x="585" y="450" width="60" height="80" as="geometry"/>
+                </mxCell>
+                <mxCell id="40" value="ComputeNode&lt;br&gt;Run in docker" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="740" y="395" width="100" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="42" value="LLM&lt;br&gt;ComputeNode&lt;br&gt;Run on Cloud" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="615" y="40" width="100" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="46" style="edgeStyle=none;html=1;exitX=1;exitY=0.5;exitDx=0;exitDy=0;entryX=0;entryY=0.25;entryDx=0;entryDy=0;" edge="1" parent="1" source="43" target="18">
+                    <mxGeometry relative="1" as="geometry"/>
+                </mxCell>
+                <mxCell id="43" value="Device" style="ellipse;whiteSpace=wrap;html=1;aspect=fixed;" vertex="1" parent="1">
+                    <mxGeometry x="70" y="70" width="50" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="45" style="edgeStyle=none;html=1;exitX=1;exitY=1;exitDx=0;exitDy=0;entryX=0.25;entryY=0;entryDx=0;entryDy=0;" edge="1" parent="1" source="44" target="18">
+                    <mxGeometry relative="1" as="geometry"/>
+                </mxCell>
+                <mxCell id="44" value="Device" style="ellipse;whiteSpace=wrap;html=1;aspect=fixed;" vertex="1" parent="1">
+                    <mxGeometry x="125" y="10" width="50" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="47" value="Device" style="ellipse;whiteSpace=wrap;html=1;aspect=fixed;" vertex="1" parent="1">
+                    <mxGeometry x="295" y="140" width="50" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="48" value="Driver Sync Service" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="545" y="600" width="90" height="50" as="geometry"/>
+                </mxCell>
+                <mxCell id="49" value="Zone backup service" style="rounded=1;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+                    <mxGeometry x="665" y="600" width="90" height="50" as="geometry"/>
+                </mxCell>
+            </root>
+        </mxGraphModel>
+    </diagram>
+    <diagram id="gXHOratDfWrfJ3-cl-bj" name="Page-2">
+        <mxGraphModel dx="2266" dy="1010" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="1100" pageHeight="850" math="0" shadow="0">
+            <root>
+                <mxCell id="0"/>
+                <mxCell id="1" parent="0"/>
+            </root>
+        </mxGraphModel>
+    </diagram>
+</mxfile>
@@ -1,6 +1,6 @@
 <mxfile host="65bd71144e" pages="3">
    <diagram id="C5RBs43oDa-KdzZeNtuy" name="Page-1">
-        <mxGraphModel dx="1800" dy="674" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="827" pageHeight="1169" math="0" shadow="0">
+        <mxGraphModel dx="1800" dy="877" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="827" pageHeight="1169" math="0" shadow="0">
            <root>
                <mxCell id="WIyWlLk6GJQsqaUBKTNV-0"/>
                <mxCell id="WIyWlLk6GJQsqaUBKTNV-1" parent="WIyWlLk6GJQsqaUBKTNV-0"/>
@@ -220,7 +220,7 @@
        </mxGraphModel>
    </diagram>
    <diagram id="7NYJTgo0U9cdVshLy85U" name="Page-3">
-        <mxGraphModel dx="1800" dy="674" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="850" pageHeight="1100" math="0" shadow="0">
+        <mxGraphModel dx="1800" dy="877" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="850" pageHeight="1100" math="0" shadow="0">
            <root>
                <mxCell id="0"/>
                <mxCell id="1" parent="0"/>
@@ -1,4 +1,4 @@
-instance_id = "agent:xxxxxxabcde"
+instance_id = "agent_1"
 fullname = "tracy wang"
 [[prompt]]
 role = "system"
@@ -0,0 +1,5 @@
+instance_id = "agent:xxxxxxabcde"
+fullname = "jackma"
+[[prompt]]
+role = "system"
+content = "你擅长处理商务电子邮件"
@@ -0,0 +1,6 @@
+instance_id = "agent:xxxxxxabcde"
+fullname = "musk"
+[[prompt]]
+role = "system"
+content = "你有丰富的管理技能，擅长将复杂工作拆解成简单的任务，让团队成员高效协作。"
+
@@ -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字。"""
+# 这里是孤立工作模式，组长只和经理沟通，也可以赋予其和其它组沟通的能力
@@ -0,0 +1,2 @@
+main = "./"
+cache = "./.workflows"
@@ -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
+from .bus import AIBus
@@ -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,15 +126,18 @@ 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
@@ -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
+
@@ -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
@@ -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))
+
+    
@@ -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 
@@ -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
@@ -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}")
        
@@ -1,14 +1,40 @@
-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
@@ -16,10 +42,25 @@ class AIRole:
    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)

    
@@ -6,65 +6,172 @@ 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

-    def load_from_disk(self,config_path:str,context_dir_path) -> int:
-        pass
+        self.is_start = False
+        self.msg_queue = Queue()

-    #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)
+    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
+        
+        if config.get("name") is None: 
+            logger.error("workflow config must have name")
+            return False
+        self.workflow_name = config.get("name")
+
+        #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
+
+        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
+        
+        return True
+
+    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
+            else:
+                logger.error(f"input_filter return None for :{msg}")
                return
                
-    async def send_msg(self,msg:AgentMsg) -> str:
-        pass
+        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

-    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)
+            # 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
+
+
+
+
+    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_chat_session_for_msg return None for :{the_msg}")
-                continue
+            logger.error(f"get session {session_topic}@{self.workflow_name} failed!")
+            return None
        
-            chatsession.append_recv(the_msg)
-        
-            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.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,the_role.get_name())) # chat context
+
+        # 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:
@@ -82,70 +189,34 @@ class Workflow:
                # 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())
+                    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}"}]
                                               
-                            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)
+                    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 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)
+        if is_ignore:
+            return None
        
-                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))
+        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"""
@@ -11,10 +11,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:
            cls._instance = super(AgentManager, cls).__new__(cls)
@@ -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:
@@ -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
+    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
@@ -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,39 +62,13 @@ 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
-        
+        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,7 +113,7 @@ 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__":    
    asyncio.run(main())
@@ -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()