Fix bug: Agent process message correctly in Telegeram Gorup

src/aios_kernel/agent.py

@@ -361,17 +361,97 @@ class AIAgent:
             old_content = msg.get("content")
             msg["content"] = old_content.format_map(self.owner_env)
 
+    async def _process_group_chat_msg(self,msg:AgentMsg) -> AgentMsg:
+        from .compute_kernel import ComputeKernel
+        from .bus import AIBus
+        
+        session_topic = msg.target + "#" + msg.topic
+        chatsession = AIChatSession.get_session(self.agent_id,session_topic,self.chat_db)
+        need_process = False
+        if msg.mentions is not None:
+            if self.agent_id in msg.mentions:
+                need_process = True
+                logger.info(f"agent {self.agent_id} recv a group chat message from {msg.sender},but is not mentioned,ignore!")
+
+        if need_process is not True:
+            chatsession.append(msg)
+            resp_msg = msg.create_group_resp_msg(self.agent_id,"")
+            return resp_msg
+        else:
+            msg_prompt = AgentPrompt()
+            msg_prompt.messages = [{"role":"user","content":f"{msg.sender}:{msg.body}"}]
+
+            prompt = AgentPrompt()
+            prompt.append(await self._get_agent_prompt())
+            self._format_msg_by_env_value(prompt)
+            inner_functions,function_token_len = self._get_inner_functions()
+       
+            system_prompt_len = prompt.get_prompt_token_len()
+            input_len = len(msg.body)
+
+            history_prmpt,history_token_len = await self._get_prompt_from_session_for_groupchat(chatsession,system_prompt_len + function_token_len,input_len)
+            prompt.append(history_prmpt) # chat context
+            prompt.append(msg_prompt)
+
+            logger.debug(f"Agent {self.agent_id} do llm token static system:{system_prompt_len},function:{function_token_len},history:{history_token_len},input:{input_len}, totoal prompt:{system_prompt_len + function_token_len + history_token_len} ")
+            task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions)
+            if task_result.result_code != ComputeTaskResultCode.OK:
+                logger.error(f"llm compute error:{task_result.error_str}")
+                error_resp = msg.create_error_resp(task_result.error_str)
+                return error_resp
+            
+            final_result = task_result.result_str
+
+            result_message = task_result.result.get("message")
+            if result_message:
+                inner_func_call_node = result_message.get("function_call")
+            if inner_func_call_node:
+                #TODO to save more token ,can i use msg_prompt?
+                call_prompt : AgentPrompt = copy.deepcopy(prompt)
+                final_result,error_code = await self._execute_func(inner_func_call_node,call_prompt,msg)
+                if error_code != 0:
+                    error_resp = msg.create_error_resp(final_result)
+                    return error_resp
+
+            llm_result : LLMResult = self._get_llm_result_type(final_result)
+            is_ignore = False
+            result_prompt_str = ""
+            match llm_result.state:
+                case "ignore":
+                    is_ignore = True
+                case "waiting":
+                    for sendmsg in llm_result.send_msgs:
+                        target = sendmsg.target
+                        sendmsg.topic = msg.topic
+                        sendmsg.prev_msg_id = msg.get_msg_id()
+                        send_resp = await AIBus.get_default_bus().send_message(sendmsg)
+                        if send_resp is not None:
+                            result_prompt_str += f"\n{target} response is :{send_resp.body}"
+                            agent_sesion = AIChatSession.get_session(self.agent_id,f"{sendmsg.target}#{sendmsg.topic}",self.chat_db)
+                            agent_sesion.append(sendmsg)
+                            agent_sesion.append(send_resp)
+
+                    final_result = llm_result.resp + result_prompt_str
+
+            if is_ignore is not True:
+                resp_msg = msg.create_group_resp_msg(self.agent_id,final_result)
+                chatsession.append(msg)
+                chatsession.append(resp_msg)
+
+                return resp_msg
+
+            return None
+
     async def _process_msg(self,msg:AgentMsg) -> AgentMsg:
             from .compute_kernel import ComputeKernel
             from .bus import AIBus
             
+            if msg.msg_type == AgentMsgType.TYPE_GROUPMSG:
+                return await self._process_group_chat_msg(msg)
+
             session_topic = msg.get_sender() + "#" + msg.topic
             chatsession = AIChatSession.get_session(self.agent_id,session_topic,self.chat_db)
-            if msg.mentions is not None:
+
-                if not self.agent_id in msg.mentions:
-                    chatsession.append(msg)
-                    logger.info(f"agent {self.agent_id} recv a group chat message from {msg.sender},but is not mentioned,ignore!")
-                    return None
                 
             msg_prompt = AgentPrompt()
             msg_prompt.messages = [{"role":"user","content":msg.body}]
@@ -454,6 +534,37 @@ class AIAgent:
     def get_max_token_size(self) -> int:
         return self.max_token_size
     
+    async def _get_prompt_from_session_for_groupchat(self,chatsession:AIChatSession,system_token_len,input_token_len,is_groupchat=False):
+        history_len = (self.max_token_size * 0.7) - system_token_len - input_token_len
+        messages = chatsession.read_history(self.history_len) # read
+        result_token_len = 0
+        result_prompt = AgentPrompt()
+        read_history_msg = 0
+        for msg in reversed(messages):
+            read_history_msg += 1
+            dt = datetime.datetime.fromtimestamp(float(msg.create_time))
+            formatted_time = dt.strftime('%y-%m-%d %H:%M:%S')
+
+            if msg.sender == self.agent_id:
+                if self.enable_timestamp:
+                    result_prompt.messages.append({"role":"assistant","content":f"(create on {formatted_time}) {msg.body} "})
+                else:
+                    result_prompt.messages.append({"role":"assistant","content":msg.body})
+                
+            else:
+                if self.enable_timestamp:
+                    result_prompt.messages.append({"role":"user","content":f"(create on {formatted_time}) {msg.body} "})
+                else:
+                    result_prompt.messages.append({"role":"user","content":f"{msg.sender}:{msg.body}"})
+
+            history_len -= len(msg.body)
+            result_token_len += len(msg.body)
+            if history_len < 0:
+                logger.warning(f"_get_prompt_from_session reach limit of token,just read {read_history_msg} history message.")
+                break
+
+        return result_prompt,result_token_len
+
     async def _get_prompt_from_session(self,chatsession:AIChatSession,system_token_len,input_token_len,is_groupchat=False) -> AgentPrompt:
         # TODO: get prompt from group chat is different from single chat
         history_len = (self.max_token_size * 0.7) - system_token_len - input_token_len

src/aios_kernel/agent_message.py

@@ -8,10 +8,12 @@ from .ai_function import FunctionItem
 
 class AgentMsgType(Enum):
     TYPE_MSG = 0
-    TYPE_INTERNAL_CALL = 1
+    TYPE_GROUPMSG = 1
-    TYPE_ACTION = 2
+    TYPE_INTERNAL_CALL = 10
-    TYPE_EVENT = 3
+    TYPE_ACTION = 20
-    TYPE_SYSTEM = 4
+    TYPE_EVENT = 30
+    TYPE_SYSTEM = 40
+
 
 class AgentMsgStatus(Enum):
     RESPONSED = 0
@@ -44,6 +46,9 @@ class AgentMsg:
         self.rely_msg_id:str = None # if not none means this is a respone msg
         self.session_id:str = None
 
+        #forword info
+
+
         self.create_time = 0
         self.done_time = 0
         self.topic:str = None # topic is use to find session, not store in db
@@ -112,6 +117,18 @@ class AgentMsg:
 
         return resp_msg
     
+    def create_group_resp_msg(self,sender_id,resp_body):
+        resp_msg = AgentMsg(AgentMsgType.TYPE_GROUPMSG)
+        resp_msg.create_time = time.time()
+
+        resp_msg.rely_msg_id = self.msg_id
+        resp_msg.target = self.target
+        resp_msg.sender = sender_id
+        resp_msg.body = resp_body
+        resp_msg.topic = self.topic
+
+        return resp_msg
+
     def set(self,sender:str,target:str,body:str,topic:str=None) -> None:
         self.sender = sender
         self.target = target

src/aios_kernel/bus.py

@@ -1,5 +1,5 @@
 from typing import Coroutine,Dict,Any
-from .agent_message import AgentMsg,AgentMsgStatus
+from .agent_message import AgentMsg,AgentMsgStatus,AgentMsgType
 import asyncio
 from asyncio import Queue
 
@@ -23,7 +23,10 @@ class AIBusHandler:
         resp_msg = await self.handler(msg)
         if self.enable_defualt_proc:
             if resp_msg is not None:
-                await self.owner_bus.post_message(resp_msg,False)
+                if resp_msg.msg_type == AgentMsgType.TYPE_GROUPMSG:
+                    await self.owner_bus.post_message(resp_msg,resp_msg.target)
+                else:
+                    await self.owner_bus.post_message(resp_msg)
 
         return resp_msg
 
@@ -40,8 +43,11 @@ class AIBus:
         self.unhandle_handler:Coroutine = None
 
 
-    async def post_message(self,msg:AgentMsg,use_unhandle=True) -> bool:
+    async def post_message(self,msg:AgentMsg,target_id = None,use_unhandle=True) -> bool:
-        target_id = msg.target.split(".")[0]
+        if target_id is None:
+            target_id =msg.target
+
+        target_id = target_id.split(".")[0]
 
         handler = self.handlers.get(target_id)
         if handler:
@@ -55,8 +61,8 @@ class AIBus:
 
         if use_unhandle:
             if self.unhandle_handler is not None:
-                if await self.unhandle_handler(self,msg):
+                if await self.unhandle_handler(self,target_id):
-                    return await self.post_message(msg,False)
+                    return await self.post_message(msg,target_id,False)
 
         logger.warn(f"post message to {msg.target} failed!,target not found")
         return False
@@ -65,14 +71,18 @@ class AIBus:
         assert resp.rely_msg_id == org_msg_id
         return await self.post_message(resp)
 
-    async def send_message(self,msg:AgentMsg) -> AgentMsg:
+    async def send_message(self,msg:AgentMsg,target_id = None, real_sender=None) -> AgentMsg:
-        sender_id = msg.sender.split(".")[0]
+        if real_sender is None:
+            sender_id = msg.sender.split(".")[0]
+        else:
+            sender_id = real_sender.split(".")[0]
+            
         sender_handler = self.handlers.get(sender_id) # sender already register on bus
         if sender_handler is None:
             logger.warn(f"sender {sender_id} not register on AI_BUS!")
             return None
 
-        post_result = await self.post_message(msg)
+        post_result = await self.post_message(msg,target_id)
         if post_result is False:
             return None
 

src/aios_kernel/tg_tunnel.py

@@ -5,7 +5,7 @@ import uuid
 import time
 import aiofiles
 
-from telegram import Update
+from telegram import Update,Message
 from telegram import Bot
 from telegram.ext import Updater
 from telegram.error import Forbidden, NetworkError
@@ -17,7 +17,7 @@ from .knowledge_base import KnowledgeBase
 from .tunnel import AgentTunnel
 from .storage import AIStorage
 from .contact_manager import ContactManager,Contact,FamilyMember
-from .agent_message import AgentMsg
+from .agent_message import AgentMsg,AgentMsgType
 
 
 logger = logging.getLogger(__name__)
@@ -79,6 +79,7 @@ class TelegramTunnel(AgentTunnel):
         logger.info(f"tunnel {self.tunnel_id} is starting...")
 
         self.bot = Bot(self.tg_token)
+        self.bot_username = (await self.bot.get_me()).username
         self.update_queue = asyncio.Queue()
         self.bot_updater = Updater(self.bot,update_queue=self.update_queue)
 
@@ -114,21 +115,62 @@ class TelegramTunnel(AgentTunnel):
     async def _process_message(self, msg: AgentMsg) -> None:
         logger.warn(f"process message {msg.msg_id} from {msg.sender} to {msg.target}")
 
-    async def conver_tg_msg_to_agent_msg(self,update:Update) -> AgentMsg:
+
+    async def conver_tg_msg_to_agent_msg(self,message:Message) -> AgentMsg:
         agent_msg = AgentMsg()
         agent_msg.topic = "_telegram"
-        agent_msg.msg_id = "tg_msg#" + str(update.message.message_id) + "#" + uuid.uuid4().hex
+        agent_msg.msg_id = "tg_msg#" + str(message.message_id) + "#" + uuid.uuid4().hex
         agent_msg.target = self.target_id
-        agent_msg.body = update.message.text
+        agent_msg.body = message.text
         agent_msg.create_time = time.time()
-        #if update.message.photo is not None:
+        messag_type = message.chat.type
-        #    agent_msg.body_mime = "image"
+        if messag_type == "supergroup" or messag_type == "group":
-        #    agent_msg.body = update.message.photo[-1].get_file().download()
+            agent_msg.target = f"tg_group{message.chat_id}"
+            agent_msg.msg_type = AgentMsgType.TYPE_GROUPMSG
+            agent_msg.mentions = []
+        else:
+            agent_msg.msg_type = AgentMsgType.TYPE_MSG
+
+        if message.entities:
+            for entity in message.entities:
+                if entity.type == 'mention':
+                    mention = message.text[entity.offset:entity.offset+entity.length]
+                    if mention == '@' + self.bot_username:
+                        agent_msg.mentions.append(self.target_id)
+                    else:
+                        agent_msg.mentions.append(mention)
+                    
+        if message.caption_entities:
+            for entity in message.caption_entities:
+                if entity.type == 'mention':
+                    mention = message.caption[entity.offset:entity.offset+entity.length]
+                    if mention == '@' + self.bot_username:
+                        agent_msg.mentions.append(self.target_id)
+                    else:
+                        agent_msg.mentions.append(mention)
+
         return agent_msg
 
+    def is_bot_mentioned(self,message:Message):
+        if message.entities:
+            for entity in message.entities:
+                if entity.type == 'mention':
+                    mention = message.text[entity.offset:entity.offset+entity.length]
+                    if mention == '@' + self.bot_username:
+                        return True
 
+        if message.caption_entities:
+            for entity in message.caption_entities:
+                if entity.type == 'mention':
+                    mention = message.caption[entity.offset:entity.offset+entity.length]
+                    if mention == '@' + self.bot_username:
+                        return True
+
+        return False
 
     async def on_message(self, bot:Bot, update: Update) -> None:
+        message = update.message
+        logger.info(f"on_message: {message.message_id} from {message.from_user.username} ({update.effective_user.username}) to {message.chat.title}({message.chat.id})")
         if update.effective_user.is_bot:
             logger.warning(f"ignore message from telegram bot {update.effective_user.id}")
             return None
@@ -139,12 +181,9 @@ class TelegramTunnel(AgentTunnel):
         
         self.in_process_tg_msg[update.message.message_id] = True
 
-
+        agent_msg = await self.conver_tg_msg_to_agent_msg(message)
         cm : ContactManager = ContactManager.get_instance()
         reomte_user_name = f"{update.effective_user.id}@telegram"
-        #owner_tg_username = AIStorage.get_instance().get_user_config().get_value("telegram")
-        #owner_name = AIStorage.get_instance().get_user_config().get_value("username")
-
 
         contact : Contact = cm.find_contact_by_telegram(update.effective_user.username)
         if contact is None:
@@ -173,37 +212,50 @@ class TelegramTunnel(AgentTunnel):
                 cm.add_contact(contact.name, contact)
                 reomte_user_name = contact.name
              
-        agent_msg = await self.conver_tg_msg_to_agent_msg(update)
+
         agent_msg.sender = reomte_user_name
-        self.ai_bus.register_message_handler(reomte_user_name, self._process_message)
-        #await bot.send_chat_action(chat_id=update.effective_chat.id, action="typing")
-        resp_msg = await self.ai_bus.send_message(agent_msg)
         logger.info(f"process message {agent_msg.msg_id} from {agent_msg.sender} to {agent_msg.target}")
+        if agent_msg.msg_type == AgentMsgType.TYPE_GROUPMSG:
+            self.ai_bus.register_message_handler(agent_msg.target, self._process_message)
+            resp_msg = await self.ai_bus.send_message(agent_msg,self.target_id,agent_msg.target)
+        else:
+            self.ai_bus.register_message_handler(reomte_user_name, self._process_message)
+            resp_msg = await self.ai_bus.send_message(agent_msg)
+        #await bot.send_chat_action(chat_id=update.effective_chat.id, action="typing")
+
+
+
         if resp_msg is None:
             await update.message.reply_text(f"System Error: Timeout,{self.target_id}  no resopnse! Please check logs/aios.log for more details!")
         else:
             if resp_msg.body_mime is None:
-                if resp_msg.body is not None:
+                if resp_msg.body is None:
-                    knowledge_object = KnowledgeBase().parse_object_in_message(resp_msg.body)
+                    return
-                    if knowledge_object is not None:
+                
-                        if knowledge_object.get_object_type() == ObjectType.Image:
+                if len(resp_msg.body) < 1:
-                            image = KnowledgeBase().bytes_from_object(knowledge_object)
+                    await update.message.reply_text("")
-                            try:
+                    return
-                                async with aiofiles.open("tg_send_temp.png", mode='wb') as local_file:
+                
-                                    if local_file:
+                knowledge_object = KnowledgeBase().parse_object_in_message(resp_msg.body)
-                                        await local_file.write(image)
+                if knowledge_object is not None:
-                                        await update.message.reply_photo("tg_send_temp.png")
+                    if knowledge_object.get_object_type() == ObjectType.Image:
-                            except Exception as e:
+                        image = KnowledgeBase().bytes_from_object(knowledge_object)
-                                logger.error(f"save image error: {e}")
+                        try:
-                            return
+                            async with aiofiles.open("tg_send_temp.png", mode='wb') as local_file:
-                    else:
+                                if local_file:
-                        pos = resp_msg.body.find("audio file")
+                                    await local_file.write(image)
-                        if pos != -1:
+                                    await update.message.reply_photo("tg_send_temp.png")
-                            audio_file = resp_msg.body[pos+11:].strip()
+                        except Exception as e:
-                            if audio_file.startswith("\""):
+                            logger.error(f"save image error: {e}")
-                                audio_file = audio_file[1:-1]
+                        return
-                            await update.message.reply_voice(audio_file)
+                else:
-                            return
+                    pos = resp_msg.body.find("audio file")
+                    if pos != -1:
+                        audio_file = resp_msg.body[pos+11:].strip()
+                        if audio_file.startswith("\""):
+                            audio_file = audio_file[1:-1]
+                        await update.message.reply_voice(audio_file)
+                        return
                 await update.message.reply_text(resp_msg.body)
             else:
                 if resp_msg.body_mime.startswith("image"):

src/aios_kernel/workflow.py

@@ -326,7 +326,7 @@ class Workflow:
             asyncio.create_task(target_workflow._process_msg(msg))
 
         logger.info(f"{msg.sender} post message {msg.msg_id} to AIBus: {msg.target}")
-        await self.get_bus().post_message(msg.target,msg)
+        await self.get_bus().post_message(msg,msg.target)
         return
 
 

src/service/aios_shell/aios_shell.py

@@ -80,8 +80,7 @@ class AIOS_Shell:
 
 
 
-    async def _handle_no_target_msg(self,bus:AIBus,msg:AgentMsg) -> bool:
+    async def _handle_no_target_msg(self,bus:AIBus,target_id:str) -> bool:
-        target_id = msg.target.split(".")[0]
         agent : AIAgent = await AgentManager.get_instance().get(target_id)
         if agent is not None:
             bus.register_message_handler(target_id,agent._process_msg)