Fix workflow system prompt bugs.

rootfs/workflows/math_school/workflow.toml

@@ -20,9 +20,13 @@ agent="math_teacher"
 [[roles."小学老师".prompt]] 
 role="system"
 content="""现在时间是:{now}你在学校任职，担任小学老师。学校由 小学老师、初中老师、高中老师、教导处主任 组成。
-当你发现学生的水平不是小学生时，应使用 sendmsg(老师名称,问题) 的方法，把学生的问题转发给学校里合适的老师
+你的任何处理结果，都要用下面方式汇报给给教导处主任
-当学生发来作业时，进行批改(满分5分)，并把批改结果以 postmsg(教导处主任,学生名_作业结果) 的方法，将一次作业情况汇报给教导处主任。
+```
-你会根据教导处主任的指示，定期调整教学方法"""
+##/sendmsg 教导处主任
+处理结果 
+```
+
+"""
 
 
 [roles."初中老师"]

src/aios_kernel/agent.py

@@ -6,6 +6,7 @@ import logging
 import uuid
 import time
 import json
+import shlex
 
 from .agent_message import AgentMsg, AgentMsgStatus, AgentMsgType
 from .chatsession import AIChatSession
@@ -15,7 +16,6 @@ from .environment import Environment
 
 logger = logging.getLogger(__name__)
 
-
 class AgentPrompt:
     def __init__(self) -> None:
         self.messages = []

src/aios_kernel/agent_message.py

@@ -2,6 +2,7 @@ from enum import Enum
 import uuid
 import time 
 import re
+import shlex
 
 class AgentMsgType(Enum):
     TYPE_MSG = 0
@@ -123,19 +124,8 @@ class AgentMsg:
     
     @classmethod
     def parse_function_call(cls,func_string:str):
-        match = re.search(r'\s*(\w+)\s*\(\s*(.*)\s*\)\s*', func_string)
+        str_list = shlex.split(func_string)
-        if not match:
+        func_name = str_list[0]
-            return None
+        params = str_list[1:]
-
-        func_name = match.group(1)
-        if func_name is None:
-            return None
-        if len(func_name) < 2:
-            return None
-        
-        params_string = match.group(2).strip()    
-        params = re.split(r'\s*,\s*(?=(?:[^"]*"[^"]*")*[^"]*$)', params_string)
-        params = [param.strip('"') for param in params]
-
         return func_name, params
         

src/aios_kernel/ai_function.py

@@ -69,6 +69,21 @@ class AIFunction:
     #def load_from_config(self,config:dict) -> bool:
     #    pass
 
+class FunctionItem:
+    def __init__(self,name,args) -> None:
+        self.name = name
+        self.args = args
+        self.body = None
+
+    def append_body(self,body:str) -> None:
+        if self.body is None:
+            self.body = body
+        else:
+            self.body += body
+
+    def dumps(self) -> str:
+        pass
+
 # call chain is a combination of ai_function,group of ai_function.
 class CallChain:
     def __init__(self) -> None:

src/aios_kernel/workflow.py

@@ -4,7 +4,7 @@ import json
 import os
 import time 
 from asyncio import Queue
-from typing import Optional,Tuple
+from typing import Optional,Tuple,List
 from abc import ABC, abstractmethod
 
 from .environment import Environment,EnvironmentEvent
@@ -12,7 +12,7 @@ from .agent_message import AgentMsg,AgentMsgStatus
 from .agent import AgentPrompt,AgentMsg
 from .chatsession import AIChatSession
 from .role import AIRole,AIRoleGroup
-from .ai_function import AIFunction
+from .ai_function import AIFunction,FunctionItem
 from .compute_kernel import ComputeKernel
 from .compute_task import ComputeTask,ComputeTaskResult,ComputeTaskState
 from .bus import AIBus
@@ -42,10 +42,10 @@ class LLMResult:
     def __init__(self) -> None:
         self.state : str = "ignore"
         self.resp : str = ""
-        self.post_msgs = []
+        self.post_msgs : List[AgentMsg] = []
-        self.send_msgs = []
+        self.send_msgs : List[AgentMsg] = []
-        self.calls = []
+        self.calls : List[FunctionItem] = []
-        self.post_calls = []
+        self.post_calls : List[FunctionItem] = []
 
 
 class Workflow:
@@ -255,46 +255,59 @@ class Workflow:
         
         lines = llm_result_str.splitlines()
         is_need_wait = False
-        for line in lines:
+
-            func_call = AgentMsg.parse_function_call(line)
+        def check_args(func_item:FunctionItem):
-            if func_call:
+            match func_name:
-                func_args = func_call[1]
+                case "send_msg":# sendmsg($target_id,$msg_content)
-                match func_call[0]:
+                    if len(func_args) != 1:
-                    case "sendmsg":# sendmsg($target_id,$msg_content)
+                        logger.error(f"parse sendmsg failed! {func_call}")
-                        if len(func_args) != 2:
+                        return False
-                            logger.error(f"parse sendmsg failed! {func_call}")
+                    new_msg = AgentMsg()
-                            continue
+                    target_id = func_item.args[0]
-                        new_msg = AgentMsg()
+                    msg_content = func_item.body
-                        target_id = func_args[0]
+                    new_msg.set("_",target_id,msg_content)
-                        msg_content = func_args[1]
+
-                        new_msg.set("_",target_id,msg_content)
+                    r.send_msgs.append(new_msg)
-    
+                    is_need_wait = True
-                        r.send_msgs.append(new_msg)
+                    
-                        is_need_wait = True
+                case "post_msg":# postmsg($target_id,$msg_content)
-                        continue
+                    if len(func_args) != 1:
-                    case "postmsg":# postmsg($target_id,$msg_content)
+                        logger.error(f"parse postmsg failed! {func_call}")
-                        if len(func_args) != 2:
+                        return False
-                            logger.error(f"parse postmsg failed! {func_call}")
+                    new_msg = AgentMsg()
-                            continue
+                    target_id = func_item.args[0]
-                        new_msg = AgentMsg()
+                    msg_content = func_item.body
-                        target_id = func_args[0]
+                    new_msg.set("_",target_id,msg_content)
-                        msg_content = func_args[1]
+                    r.post_msgs.append(new_msg)
-                        new_msg.set("_",target_id,msg_content)
+                    
-                        r.post_msgs.append(new_msg)
+                case "call":# call($func_name,$args_str)
-                        continue
+                    r.calls.append(func_item)
-                    case "call":# call($func_name,$args_str)
+                    is_need_wait = True
-                        r.calls.append(func_call)
+                    return True
-                        is_need_wait = True
+                case "post_call": # post_call($func_name,$args_str)
-                        continue 
+                    r.post_calls.append(func_item)
-                    case "post_call": # post_call($func_name,$args_str)
+                    return True    
-                        r.post_calls.append(func_call)
-                        continue
                 
-                r.resp += line + "\n"
+        current_func : FunctionItem = None
+        for line in lines:
+            if line.startswith("##/"):
+                if current_func:
+                    if check_args(current_func) is False:
+                        r.resp += current_func.dumps()
+               
+                func_name,func_args = AgentMsg.parse_function_call(line[3:])
+                current_func = FunctionItem(func_name,func_args)
             else:
-                r.resp += line + "\n"
+                if current_func:
+                    current_func.append_body(line + "\n")
+                else:
+                    r.resp += line + "\n"
         
-        if is_need_wait:
+        if current_func:
+            if check_args(current_func) is False:
+                r.resp += current_func.dumps()
+                
+        if len(r.send_msgs) > 0 or len(r.calls) > 0:
             r.state = "waiting"
         else:
             r.state = "reponsed"
@@ -339,21 +352,20 @@ class Workflow:
         logger.info(f"{msg.sender} post message {msg.msg_id} to AIBus: {msg.target}")
         return await self.get_bus().send_message(msg)
 
-    async def role_call(self,call:tuple,the_role:AIRole):
+    async def role_call(self,func_item:FunctionItem,the_role:AIRole):
-        logger.info(f"{the_role.role_id} call {call[0]} with args {call[1]}")
+        logger.info(f"{the_role.role_id} call {func_item.name} ")
-        func_name = call[0]
+        arguments = func_item.args
-        arguments = call[1]
 
-        func_node : AIFunction = self.workflow_env.get_ai_function(func_name)
+        func_node : AIFunction = self.workflow_env.get_ai_function(func_item.name)
         if func_node is None:
             return "execute failed,function not found"
     
         result_str:str = await func_node.execute(**arguments)        
         return result_str
 
-    async def role_post_call(self,call:tuple,the_role:AIRole):
+    async def role_post_call(self,func_item:FunctionItem,the_role:AIRole):
-        logger.info(f"{the_role.role_id} post call {call[0]} with args {call[1]}")
+        logger.info(f"{the_role.role_id} post call {func_item.name} ")
-        return await self.role_call(call,the_role)
+        return await self.role_call(func_item,the_role)
 
     def _format_msg_by_env_value(self,prompt:AgentPrompt):
         if self.workflow_env is None:
@@ -554,3 +566,7 @@ class Workflow:
             #            the_env.attach_event_handler(k,_env_msg_handler)
             #            break
 
+
+
+
+

src/service/aios_shell/aios_shell.py

@@ -427,6 +427,7 @@ async def main():
         print_formatted_text(show_text,style=shell_style)
         #print_formatted_text(f"{shell.username}<->{shell.current_topic}@{shell.current_target} >>> {resp}",style=shell_style)
 
+
 if __name__ == "__main__":    
     asyncio.run(main())
 

test/workflow_test.py

@@ -3,4 +3,40 @@
 def test_workflow():
     
     
-    pass
+    pass
+
+
+async def _test_llm_parser():
+    test_llm_result = """
+# Foggie with AI Agent
+1.已经完成了基础系统改造，只要Foggie能安装docker image就可以实现集成
+2.安装后，用户需要提供OpenAI Token和TG Bot Token，就可以构建自己的私有AI机器人（也可以通过绑定email实现智能邮件客服）
+   我们也可以用自己的OpenAI Token给用户用，但这需要设计新的商品。OpenAI Token用起来还是挺贵的
+3.已在发布前夕，目前集成测试的主要问题是对Email和个人文件的AI分析需要比较强的性能。
+
+# DMC开源挖矿软件
+正在等待解决 Order Placement Issue
+
+
+# Foggie with AI Agent
+1. We have completed the basic system transformation. As long as Foggie can install the docker image, integration can be achieved.
+2. After installation, users need to provide an OpenAI Token and TG Bot Token to build their own private AI robot. This can also be accomplished by linking an email to implement an intelligent email customer service. We could use our own OpenAI Token for users, but this would require the design of a new product. Using the OpenAI Token can be quite costly.
+3. We are on the eve of launch. The main issue in our integrated testing currently is that AI analysis of emails and personal files requires substantial performance, and the results don't seem so smart.
+
+#DMC Open Source Mining Software
+We are waiting to resolve the Order Placement Issue.
+
+##/send_msg "xxx xxx"
+abcdcdsdf
+sfsadfasdf 
+# Foggie with AI Agent
+1. We have completed the basic system transformation. As long as Foggie can install the docker image, integration can be achieved.
+2. After installation, users need to provide an OpenAI Token and TG Bot Token to build their own private AI robot. This can also be accomplished by linking an email to implement an intelligent email customer service. We could use our own OpenAI Token for users, but this would require the design of a new product. Using the OpenAI Token can be quite costly.
+3. We are on the eve of launch. The main issue in our integrated testing cur
+
+##/call abcd "xxx xxx"
+"""
+    llm_result = Workflow.prase_llm_result(test_llm_result)
+    assert len(llm_result.calls) == 1
+    assert len(llm_result.send_msgs) == 1
+    print(llm_result)