diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index ab12566..413b104 100644 --- a/src/aios_kernel/agent.py +++ b/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}] @@ -453,6 +533,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 diff --git a/src/aios_kernel/agent_message.py b/src/aios_kernel/agent_message.py index d597492..c5dc26d 100644 --- a/src/aios_kernel/agent_message.py +++ b/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_ACTION = 2 - TYPE_EVENT = 3 - TYPE_SYSTEM = 4 + TYPE_GROUPMSG = 1 + TYPE_INTERNAL_CALL = 10 + TYPE_ACTION = 20 + 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 @@ -111,6 +116,18 @@ class AgentMsg: resp_msg.topic = self.topic 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 diff --git a/src/aios_kernel/bus.py b/src/aios_kernel/bus.py index 8d38e53..2731702 100644 --- a/src/aios_kernel/bus.py +++ b/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: - target_id = msg.target.split(".")[0] + async def post_message(self,msg:AgentMsg,target_id = None,use_unhandle=True) -> bool: + 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): - return await self.post_message(msg,False) + if await self.unhandle_handler(self,target_id): + 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: - sender_id = msg.sender.split(".")[0] + async def send_message(self,msg:AgentMsg,target_id = None, real_sender=None) -> AgentMsg: + 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 diff --git a/src/aios_kernel/tg_tunnel.py b/src/aios_kernel/tg_tunnel.py index 47eaf91..53aed7a 100644 --- a/src/aios_kernel/tg_tunnel.py +++ b/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: - # agent_msg.body_mime = "image" - # agent_msg.body = update.message.photo[-1].get_file().download() + messag_type = message.chat.type + if messag_type == "supergroup" or messag_type == "group": + 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: - knowledge_object = KnowledgeBase().parse_object_in_message(resp_msg.body) - if knowledge_object is not None: - if knowledge_object.get_object_type() == ObjectType.Image: - image = KnowledgeBase().bytes_from_object(knowledge_object) - try: - async with aiofiles.open("tg_send_temp.png", mode='wb') as local_file: - if local_file: - await local_file.write(image) - await update.message.reply_photo("tg_send_temp.png") - except Exception as e: - logger.error(f"save image error: {e}") - return - else: - 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 + if resp_msg.body is None: + return + + if len(resp_msg.body) < 1: + await update.message.reply_text("") + return + + knowledge_object = KnowledgeBase().parse_object_in_message(resp_msg.body) + if knowledge_object is not None: + if knowledge_object.get_object_type() == ObjectType.Image: + image = KnowledgeBase().bytes_from_object(knowledge_object) + try: + async with aiofiles.open("tg_send_temp.png", mode='wb') as local_file: + if local_file: + await local_file.write(image) + await update.message.reply_photo("tg_send_temp.png") + except Exception as e: + logger.error(f"save image error: {e}") + return + else: + 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"): diff --git a/src/aios_kernel/workflow.py b/src/aios_kernel/workflow.py index 4944ce7..a25e202 100644 --- a/src/aios_kernel/workflow.py +++ b/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 diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index e494203..1e42799 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/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: - target_id = msg.target.split(".")[0] + async def _handle_no_target_msg(self,bus:AIBus,target_id:str) -> bool: agent : AIAgent = await AgentManager.get_instance().get(target_id) if agent is not None: bus.register_message_handler(target_id,agent._process_msg)