diff --git a/rootfs/agents/Tracy Wang/agent.toml b/rootfs/agents/Tracy Wang/agent.toml index bb427a7..c815314 100644 --- a/rootfs/agents/Tracy Wang/agent.toml +++ b/rootfs/agents/Tracy Wang/agent.toml @@ -3,4 +3,11 @@ fullname = "Tracy Wang" [[prompt]] role = "system" -content = "你是我的私人英文老师，和我用地道的美式英语进行交流。你会在和我交流的同时，调整我的输入成为更地道的美式句子，并根据你对我英文水平的预测，对可能发错英的单词标上音标。如果我给你发中文，说明我不知道这句话用美式英语怎么说，你依旧按上述规则回应我。" +content = """ +Your name is Tracy Wang, and you are my advanced private English tutor. +## You will assess my English proficiency based on all available information, using a 5-point scale. +## While interacting with me normally, you will adjust my input into more idiomatic American sentences. +## Depending on my level of English, you will annotate potentially incorrect words with phonetic symbols or provide expanded explanations for certain words and phrases. +## If I send you something that is not in English, it means I don't know how to say it in American English. You will first translate what I've sent into English and then respond according to the above rules. +## You will chat with me like a friend, rather than just teaching me lessons. +""" \ No newline at end of file diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 52dd2df..333fcb4 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -239,13 +239,12 @@ class AIAgent: async def _get_prompt_from_session(self,chatsession:AIChatSession,is_groupchat=False) -> AgentPrompt: # TODO: get prompt from group chat is different from single chat - messages = chatsession.read_history() # read last 10 message + messages = chatsession.read_history() # read result_prompt = AgentPrompt() for msg in reversed(messages): - if msg.target == chatsession.owner_id: - result_prompt.messages.append({"role":"user","content":msg.body}) - elif msg.sender == chatsession.owner_id: + if msg.sender == self.agent_id: result_prompt.messages.append({"role":"assistant","content":msg.body}) - + else: + result_prompt.messages.append({"role":"user","content":msg.body}) return result_prompt diff --git a/src/aios_kernel/agent_message.py b/src/aios_kernel/agent_message.py index 0640319..8f7ab04 100644 --- a/src/aios_kernel/agent_message.py +++ b/src/aios_kernel/agent_message.py @@ -67,6 +67,7 @@ class AgentMsg: @classmethod def create_internal_call_msg(self,func_name:str,args:dict,prev_msg_id:str,caller:str): msg = AgentMsg(AgentMsgType.TYPE_INTERNAL_CALL) + msg.create_time = time.time() msg.func_name = func_name msg.args = args msg.prev_msg_id = prev_msg_id @@ -75,6 +76,7 @@ class AgentMsg: def create_action_msg(self,action_name:str,args:dict,caller:str): msg = AgentMsg(AgentMsgType.TYPE_ACTION) + msg.create_time = time.time() msg.func_name = action_name msg.args = args msg.prev_msg_id = self.msg_id @@ -85,7 +87,7 @@ class AgentMsg: def create_resp_msg(self,resp_body): resp_msg = AgentMsg() resp_msg.msg_id = "msg#" + uuid.uuid4().hex - self.create_time = time.time() + resp_msg.create_time = time.time() resp_msg.rely_msg_id = self.msg_id resp_msg.sender = self.target diff --git a/src/aios_kernel/bus.py b/src/aios_kernel/bus.py index 22bf209..1e97733 100644 --- a/src/aios_kernel/bus.py +++ b/src/aios_kernel/bus.py @@ -90,7 +90,7 @@ class AIBus: await asyncio.sleep(0.2) retry_times += 1 - if retry_times > 100: + if retry_times > 5*120: # default timeout is 120 sec msg.status = AgentMsgStatus.ERROR return None diff --git a/src/aios_kernel/compute_kernel.py b/src/aios_kernel/compute_kernel.py index 525c266..18ddeaf 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/src/aios_kernel/compute_kernel.py @@ -30,8 +30,6 @@ class ComputeKernel: self.is_start = False self.compute_nodes = {} - self.start() - def run(self, task: ComputeTask) -> None: # check there is compute node can support this task if self.is_task_support(task) is False: @@ -41,7 +39,7 @@ class ComputeKernel: # add task to working_queue self.task_queue.put_nowait(task) - def start(self): + async def start(self): if self.is_start is True: logger.warn("compute_kernel is already start") return diff --git a/src/aios_kernel/queue_compute_node.py b/src/aios_kernel/queue_compute_node.py index e975cd8..1475672 100644 --- a/src/aios_kernel/queue_compute_node.py +++ b/src/aios_kernel/queue_compute_node.py @@ -53,7 +53,7 @@ class Queue_ComputeNode(ComputeNode): return result - def start(self): + async def start(self): async def _run_task_loop(): while True: task = await self.task_queue.get() diff --git a/src/aios_kernel/tg_tunnel.py b/src/aios_kernel/tg_tunnel.py index 36c991f..55337a6 100644 --- a/src/aios_kernel/tg_tunnel.py +++ b/src/aios_kernel/tg_tunnel.py @@ -2,12 +2,17 @@ import logging import threading import asyncio import uuid +import time from typing import Callable from telegram import ForceReply, Update from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters +from telegram import Bot +from telegram.ext import Updater +from telegram.error import Forbidden, NetworkError + from .tunnel import AgentTunnel from .contact_manager import ContactManager from .agent_message import AgentMsg @@ -42,8 +47,27 @@ class TelegramTunnel(AgentTunnel): super().__init__() self.is_start = False self.tg_token = tg_token + #self.bot:Bot = None + #self.update_queue = None #self.tunnel_id = "tg_tunnel#" + self.app.bot.id + async def _do_process_raw_message(self,bot: Bot, update_id: int) -> int: + """Echo the message the user sent.""" + # Request updates after the last update_id + updates = await bot.get_updates(offset=update_id, timeout=10, allowed_updates=Update.ALL_TYPES) + for update in updates: + next_update_id = update.update_id + 1 + + # your bot can receive updates without messages + # and not all messages contain text + if update.message and update.message.text: + # Reply to the message + #logger.info("Found message %s!", update.message.text) + await self.on_message(update) + return next_update_id + + return update_id + async def start(self) -> bool: if self.is_start: logger.warning(f"tunnel {self.tunnel_id} is already started") @@ -51,16 +75,38 @@ class TelegramTunnel(AgentTunnel): self.is_start = True logger.warning(f"tunnel {self.tunnel_id} is starting...") - self.app:Application = Application.builder().token(self.tg_token).build() - self.app.add_handler(MessageHandler(filters.TEXT, self.on_message)) + self.bot = Bot(self.tg_token) + self.update_queue = asyncio.Queue() + self.bot_updater = Updater(self.bot,update_queue=self.update_queue) + + #self.app:Application = Application.builder().token(self.tg_token).build() + + #self.app.add_handler(MessageHandler(filters.TEXT, self.on_message)) - def _run_app(): - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - self.app.run_polling(allowed_updates=Update.ALL_TYPES) + async def _run_app(): + #loop = asyncio.new_event_loop() + #asyncio.set_event_loop(loop) + #self.app.run_polling(allowed_updates=Update.ALL_TYPES) + + try: + update_id = (await self.bot.get_updates())[0].update_id + except IndexError: + update_id = None - self.poll_thread = threading.Thread(target=_run_app) - self.poll_thread.start() + #logger.info("listening for new messages...") + while True: + try: + update_id = await self._do_process_raw_message(self.bot, update_id) + except NetworkError: + await asyncio.sleep(1) + except Forbidden: + # The user has removed or blocked the bot. + update_id += 1 + + #self.poll_thread = threading.Thread(target=_run_app) + #self.poll_thread.start() + asyncio.create_task(_run_app()) + logger.warning(f"tunnel {self.tunnel_id} started.") return True async def close(self) -> None: @@ -75,7 +121,7 @@ class TelegramTunnel(AgentTunnel): agent_msg.msg_id = "tg_msg#" + str(update.message.message_id) + "#" + uuid.uuid4().hex agent_msg.target = self.target_id agent_msg.body = update.message.text - agent_msg.create_time = update.message.date.timestamp() + 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() @@ -83,7 +129,7 @@ class TelegramTunnel(AgentTunnel): - async def on_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + async def on_message(self, update: Update) -> None: cm = ContactManager.get_instance() reomte_user_name = f"{update.effective_user.id}@telegram" #contact = cm.get_by_name(update.effective_user.username) @@ -97,7 +143,9 @@ class TelegramTunnel(AgentTunnel): 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 context.bot.send_chat_action(chat_id=update.effective_chat.id, action="thinking") 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 resp_msg is None: await update.message.reply_text(f"{self.target_id} process message error") else: diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 01cbc7b..e5f2642 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -88,12 +88,13 @@ class AIOS_Shell: if await open_ai_node.initial() is not True: logger.error("openai node initial failed!") return False - ComputeKernel.get_instance().add_compute_node(open_ai_node) llama_ai_node = LocalLlama_ComputeNode() - llama_ai_node.start() - ComputeKernel().add_compute_node(llama_ai_node) + await llama_ai_node.start() + ComputeKernel.get_instance().add_compute_node(llama_ai_node) + + await ComputeKernel.get_instance().start() AIBus().get_default_bus().register_unhandle_message_handler(self._handle_no_target_msg) AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg) @@ -318,7 +319,7 @@ async def main(): '/show', '/exit', '/help'], ignore_case=True) - + await asyncio.sleep(0.2) while True: user_input = await session.prompt_async(f"{shell.username}<->{shell.current_topic}@{shell.current_target}$",completer=completer,style=shell_style) if len(user_input) <= 1: