diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..58b3427 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,27 @@ +**/__pycache__ +**/*venv +**/.classpath +**/.dockerignore +**/.env +**/.git +**/.gitignore +**/.project +**/.settings +**/.toolstarget +**/.vs +**/.vscode +**/*.*proj.user +**/*.dbmdl +**/*.jfm +**/bin +**/charts +**/docker-compose* +**/compose* +**/Dockerfile* +**/node_modules +**/npm-debug.log +**/obj +**/secrets.dev.yaml +**/values.dev.yaml +**/.db +**/.python-version diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..65296e4 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,10 @@ +FROM python:3.11 +WORKDIR /opt/aios +COPY ./src /opt/aios +COPY ./rootfs /var/aios +RUN pip install --no-cache-dir -r /opt/opendan/requirements.txt + +ENV PYTHONDONTWRITEBYTECODE 1 +ENV PYTHONUNBUFFERED 1 + +CMD ["python3","./service/aios_shell/aios_shell.py"] \ No newline at end of file diff --git a/rootfs/tunnels.cfg.toml b/rootfs/tunnels.cfg.toml new file mode 100644 index 0000000..07682fc --- /dev/null +++ b/rootfs/tunnels.cfg.toml @@ -0,0 +1,19 @@ +#[[tunnels]] +#tunnel_id = "MyRoobot" +#type="TelegramTunnel" +#target="agent_1" + +#token="" + +#[[tunnels]] +#tunnel_id="MyEmailRobot" +#type="EmailTunnel" +#target="agent_1" + +#email="your_email@msn.com" +#imap="outlook.office365.com:993" +#smtp="outlook.office365.com:587" +#user="" +#password="" +#folder="inbox" +#interval=10 \ No newline at end of file diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index 8133573..6daa1b1 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -11,3 +11,6 @@ from .bus import AIBus from .workflow_env import WorkflowEnvironment,CalenderEnvironment,CalenderEvent from .whisper_node import WhisperComputeNode from .google_text_to_speech_node import GoogleTextToSpeechNode +from .tunnel import AgentTunnel +from .tg_tunnel import TelegramTunnel +from .email_tunnel import EmailTunnel diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index 0c3e106..52dd2df 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -82,7 +82,7 @@ class AIAgent: self.chat_db = None self.unread_msg = Queue() # msg from other agent self.owner_env : Environment = None - self.owenr_bus : AIBus = None + self.owenr_bus = None @classmethod def create_from_templete(cls,templete:AIAgentTemplete, fullname:str): @@ -164,7 +164,7 @@ class AIAgent: inner_functions = self._get_inner_functions() prompt.messages.append({"role":"function","content":result_str,"name":func_name}) - task_result:ComputeTaskResult = await ComputeKernel().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) ineternal_call_record.result_str = task_result.result_str ineternal_call_record.done_time = time.time() @@ -198,7 +198,7 @@ class AIAgent: inner_functions = self._get_inner_functions() - task_result:ComputeTaskResult = await ComputeKernel().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,self.llm_model_name,self.max_token_size,inner_functions) final_result = task_result.result_str inner_func_call_node = task_result.result_message.get("function_call") @@ -243,8 +243,8 @@ class AIAgent: 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":"user","content":msg.body}) + elif msg.sender == chatsession.owner_id: result_prompt.messages.append({"role":"assistant","content":msg.body}) return result_prompt diff --git a/src/aios_kernel/compute_kernel.py b/src/aios_kernel/compute_kernel.py index 5fea958..b69b879 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/src/aios_kernel/compute_kernel.py @@ -17,19 +17,14 @@ logger = logging.getLogger(__name__) class ComputeKernel: _instance = None - - def __new__(cls): + @classmethod + def get_instance(cls): if cls._instance is None: - cls._instance = super().__new__(cls) - cls._instance.is_start = False - + cls._instance = ComputeKernel() return cls._instance - + def __init__(self) -> None: - if self.is_start is True: - return - - self.is_start = True + self.is_start = False self.task_queue = Queue() self.is_start = False self.compute_nodes = {} diff --git a/src/aios_kernel/contact_manager.py b/src/aios_kernel/contact_manager.py new file mode 100644 index 0000000..cdc050a --- /dev/null +++ b/src/aios_kernel/contact_manager.py @@ -0,0 +1,34 @@ +from typing import List + +class Contact: + def __init__(self,name:str) -> None: + self.name = name + self.tags = [] + + def is_zone_owner(self,zone_id=None) -> bool: + return True + + def get_tags(self)->List[str]: + return self.tags + + def get_name(self)->str: + return self.name + + +class ContactManager: + _instance = None + @classmethod + def get_instance(cls): + if cls._instance is None: + cls._instance = ContactManager() + return cls._instance + + def __init__(self) -> None: + self.contacts = {} + self.contacts["liuzhicong"] = Contact("liuzhicong") + + #def get_by_addr(self,addr:str) -> Contact: + # pass + + def get_by_name(self,name:str) -> Contact: + return self.contacts.get(name) \ No newline at end of file diff --git a/src/aios_kernel/email_tunnel.py b/src/aios_kernel/email_tunnel.py new file mode 100644 index 0000000..c4f75b4 --- /dev/null +++ b/src/aios_kernel/email_tunnel.py @@ -0,0 +1,143 @@ +import asyncio +import aiosmtplib +import aioimaplib +import email +from email.header import decode_header +import mailparser +import logging +import time +import datetime +from .tunnel import AgentTunnel +from .agent_message import AgentMsg + +from email.message import EmailMessage + +logger = logging.getLogger(__name__) + +class EmailTunnel(AgentTunnel): + @classmethod + def register_to_loader(cls): + async def load_email_tunnel(config:dict) -> AgentTunnel: + result_tunnel = EmailTunnel() + if await result_tunnel.load_from_config(config): + return result_tunnel + else: + return None + + AgentTunnel.register_loader("EmailTunnel",load_email_tunnel) + + async def load_from_config(self,config:dict)->bool: + self.target_id = config["target"] + self.tunnel_id = config["tunnel_id"] + + self.type = "TelegramTunnel" + self.email = config["email"] + self.imap_server = config["imap"] + s = self.imap_server.split(":") + if len(s) == 2: + self.imap_server = s[0] + self.imap_port = int(s[1]) + + self.smtp_server = config["smtp"] + s = self.smtp_server.split(":") + if len(s) == 2: + self.smtp_server = s[0] + self.smtp_port = int(s[1]) + + self.login_user = config["user"] + self.login_password = config["password"] + self.folder = config["folder"] + self.check_interval = config["interval"] + + return True + + def __init__(self) -> None: + super().__init__() + self.is_start = False + self.read_email = {} + + async def on_new_email(self,mail:mailparser.MailParser) -> None: + remote_email_addr = mail.from_[0][1] + remote_user_name = remote_email_addr.split("@")[0] + agent_msg = self.conver_mail_to_agent_msg(mail) + agent_msg.sender = remote_user_name + agent_msg.target = self.target_id + self.ai_bus.register_message_handler(remote_user_name, self._process_message) + + resp_msg = await self.ai_bus.send_message(agent_msg) + if resp_msg is None: + await self.reply_email(remote_email_addr,"Sorry, I can't understand your message","") + else: + if resp_msg.body_mime is None: + await self.reply_email(remote_email_addr,"result",resp_msg.body) + + async def reply_email(self,target_email:str,title:str,msg:str) -> None: + email_msg = EmailMessage() + email_msg['Subject'] = f"Reply: {title}" + email_msg['From'] = self.email + email_msg['To'] = target_email + email_msg.set_content(msg) + + await aiosmtplib.send( + email_msg, + hostname = self.smtp_server, + port=self.smtp_port, + username=self.login_user, + password=self.login_password, + ) + + + + def conver_mail_to_agent_msg(self,mail:mailparser.MailParser) -> AgentMsg: + msg = AgentMsg() + msg.set("",self.target_id,mail.text_plain[0]) + msg.topic = "email" + return msg + + async def check_email(self) -> None: + self.last_check_num = 0 + self.last_check_time = datetime.datetime.now() + while True: + if self.is_start == False: + return + + await asyncio.sleep(self.check_interval) + imap_client = aioimaplib.IMAP4_SSL(host=self.imap_server,port=self.imap_port) + await imap_client.wait_hello_from_server() + await imap_client.login(self.login_user, self.login_password) + + date_since = self.last_check_time.strftime("%d-%b-%Y") + + await imap_client.select(self.folder) + status, messages = await imap_client.search('UNSEEN',charset='US-ASCII') + self.last_check_time = datetime.datetime.now() + if status == "OK": + message_numbers = messages[0].split() + for num in message_numbers: + num = int(num) + if self.read_email.get(num) is not None: + continue + + status, email_data = await imap_client.fetch(str(num), "(RFC822)") + if status == "OK": + #r = email.message_from_bytes(email_data[1]) + mail = mailparser.parse_from_bytes(email_data[1]) + self.read_email[num] = mail + await self.on_new_email(mail) + + await imap_client.logout() + + async def start(self) -> bool: + if self.is_start: + logger.warning(f"tunnel {self.tunnel_id} is already started") + return False + self.is_start = True + + asyncio.create_task(self.check_email()) + return True + + async def close(self) -> None: + self.is_start = False + + async def _process_message(self, msg: AgentMsg) -> None: + logger.warn(f"process message {msg.msg_id} from {msg.sender} to {msg.target}") diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index 3a665a3..2e87710 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -13,20 +13,17 @@ logger = logging.getLogger(__name__) class OpenAI_ComputeNode(ComputeNode): _instance = None - - def __new__(cls): + @classmethod + def get_instance(cls): if cls._instance is None: - cls._instance = super(OpenAI_ComputeNode, cls).__new__(cls) - cls._instance.is_start = False + cls._instance = OpenAI_ComputeNode() return cls._instance + def __init__(self) -> None: super().__init__() - if self.is_start is True: - logger.warn("OpenAI_ComputeNode is already start") - return - self.is_start = True + self.is_start = False # openai.organization = "org-AoKrOtF2myemvfiFfnsSU8rF" #buckycloud self.openai_api_key = "" self.node_id = "openai_node" @@ -89,6 +86,10 @@ class OpenAI_ComputeNode(ComputeNode): return result def start(self): + if self.is_start is True: + return + self.is_start = True + async def _run_task_loop(): while True: task = await self.task_queue.get() diff --git a/src/aios_kernel/stability_node.py b/src/aios_kernel/stability_node.py index 8d40878..1efc7c8 100644 --- a/src/aios_kernel/stability_node.py +++ b/src/aios_kernel/stability_node.py @@ -16,20 +16,17 @@ logger = logging.getLogger(__name__) class Stability_ComputeNode(ComputeNode): _instanace = None + @classmethod + def get_instance(cls): + if cls._instance is None: + cls._instance = Stability_ComputeNode() + return cls._instance - def __new__(cls): - if cls._instanace is None: - cls._instanace = super(Stability_ComputeNode, cls).__new__(cls) - cls._instanace.is_start = False - return cls._instanace def __init__(self) -> None: super().__init__() - if self.is_start is True: - logger.warn("Stability_ComputeNode is already start") - return - self.is_start = True + self.is_start = False self.node_id = "stability_node" self.api_key = "" self.engine = "stable-diffusion-512-v2-1" @@ -88,8 +85,8 @@ class Stability_ComputeNode(ComputeNode): for resp in answers: for artifact in resp.artifacts: - logger.info("artifact:", artifact.id, - artifact.type, artifact.finish_reason) + logger.info(f"artifact:{artifact.id},{artifact.type},{artifact.finish_reason}") + if artifact.finish_reason == generation.FILTER: logging.warn("request activated the API's safety filters") if artifact.type == generation.ARTIFACT_IMAGE: @@ -108,6 +105,9 @@ class Stability_ComputeNode(ComputeNode): return None def start(self): + if self.is_start: + return + self.is_start = True async def _run_task_loop(): while True: logger.info("stability_node is waiting for task...") diff --git a/src/aios_kernel/tg_tunnel.py b/src/aios_kernel/tg_tunnel.py new file mode 100644 index 0000000..efa23fc --- /dev/null +++ b/src/aios_kernel/tg_tunnel.py @@ -0,0 +1,116 @@ +import logging +import threading +import asyncio +import uuid + +from typing import Callable + +from telegram import ForceReply, Update +from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters + +from .tunnel import AgentTunnel +from .contact_manager import ContactManager +from .agent_message import AgentMsg + +logger = logging.getLogger(__name__) + +class TelegramTunnel(AgentTunnel): + + @classmethod + def register_to_loader(cls): + async def load_tg_tunnel(config:dict) -> AgentTunnel: + result_tunnel = TelegramTunnel("") + if await result_tunnel.load_from_config(config): + return result_tunnel + else: + return None + + AgentTunnel.register_loader("TelegramTunnel",load_tg_tunnel) + + + async def load_from_config(self,config:dict)->bool: + self.tg_token = config["token"] + self.target_id = config["target"] + self.tunnel_id = config["tunnel_id"] + self.type = "TelegramTunnel" + return True + + def dump_to_config(self) -> dict: + pass + + def __init__(self,tg_token:str) -> None: + super().__init__() + self.is_start = False + self.tg_token = tg_token + #self.tunnel_id = "tg_tunnel#" + self.app.bot.id + + async def start(self) -> bool: + if self.is_start: + logger.warning(f"tunnel {self.tunnel_id} is already started") + return False + self.is_start = True + + 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) + + self.poll_thread = threading.Thread(target=_run_app) + self.poll_thread.start() + return True + + async def close(self) -> None: + pass + + 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: + agent_msg = AgentMsg() + agent_msg.topic = "_telegram" + 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() + #if update.message.photo is not None: + # agent_msg.body_mime = "image" + # agent_msg.body = update.message.photo[-1].get_file().download() + return agent_msg + + + + async def on_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + cm = ContactManager.get_instance() + reomte_user_name = f"{update.effective_user.id}@telegram" + #contact = cm.get_by_name(update.effective_user.username) + #if contact is not None: + # reomte_user_name = contact.get_name() + #if contact is None: + # update.message.reply_text(f"{self.target_id} process message error, unknown user!") + #if not contact.is_zone_owner(): + # update.message.reply_text(f"{self.target_id} process message error, you are not my owner!") + + 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) + resp_msg = await self.ai_bus.send_message(agent_msg) + if resp_msg is None: + await update.message.reply_text(f"{self.target_id} process message error") + else: + if resp_msg.body_mime is None: + await update.message.reply_text(resp_msg.body) + else: + if resp_msg.body_mime.startswith("image"): + photo_file = open(resp_msg.body,"rb") + if photo_file: + await update.message.reply_photo(resp_msg.body) + else: + await update.message.reply_text(resp_msg.body) + else: + await update.message.reply_text(resp_msg.body) + + + \ No newline at end of file diff --git a/src/aios_kernel/tunnel.py b/src/aios_kernel/tunnel.py new file mode 100644 index 0000000..bbd18d3 --- /dev/null +++ b/src/aios_kernel/tunnel.py @@ -0,0 +1,63 @@ +from abc import ABC, abstractmethod +import logging +from typing import Coroutine +from .agent_message import AgentMsg +from .bus import AIBus + +logger = logging.getLogger(__name__) + +class AgentTunnel(ABC): + _all_loader = {} + _all_tunnels = {} + @classmethod + def register_loader(cls,tunnel_type:str,loader:Coroutine) -> None: + cls._all_loader[tunnel_type] = loader + + @classmethod + async def load_all_tunnels_from_config(cls,config:dict) -> None: + for tunnel_config in config: + loader = cls._all_loader.get(tunnel_config["type"]) + if loader is not None: + tunnel = await loader(tunnel_config) + if tunnel is not None: + cls._all_tunnels[tunnel.tunnel_id] = tunnel + tunnel.connect_to(AIBus.get_default_bus(),tunnel.target_id) + await tunnel.start() + else: + logger.error(f"load tunnel {tunnel_config['tunnel_id']} failed") + else: + logger.error(f"load tunnel {tunnel_config['type']} failed,loader not found") + + def __init__(self) -> None: + super().__init__() + self.tunnel_id = None + self.target_id = None + self.target_type = None + self.ai_bus = None + self.is_connected = False + + def connect_to(self, ai_bus:AIBus,target_id: str) -> None: + """ + Connect to the agent with the given id + """ + if self.is_connected: + logger.warning(f"tunnel {self.tunnel_id} is already connected to {self.target_id}") + return + self.target_id = target_id + self.target_type = "agent" + self.ai_bus = ai_bus + self.is_connected = True + + + + @abstractmethod + async def start(self) -> bool: + pass + + @abstractmethod + async def close(self) -> None: + pass + + @abstractmethod + async def _process_message(self, msg: AgentMsg) -> None: + pass diff --git a/src/aios_kernel/workflow.py b/src/aios_kernel/workflow.py index 58d0538..46b7507 100644 --- a/src/aios_kernel/workflow.py +++ b/src/aios_kernel/workflow.py @@ -122,7 +122,11 @@ class Workflow: connected_env_ndoe = config.get("connected_env") if connected_env_ndoe is not None: for _node in connected_env_ndoe: - remote_env = Environment.get_env_by_id(_node.get("env_id")) + env_id = _node.get("env_id") + if env_id is None: + continue + + remote_env = Environment.get_env_by_id(_node.get(env_id)) if remote_env is None: logger.error(f"Workflow load connected_env failed, env {env_id} not found!") return False @@ -391,7 +395,7 @@ class Workflow: inner_functions = self._get_inner_functions() prompt.messages.append({"role":"function","content":result_str,"name":func_name}) - task_result:ComputeTaskResult = await ComputeKernel().do_llm_completion(prompt, + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt, the_role.agent.llm_model_name,the_role.agent.max_token_size, inner_functions) @@ -431,7 +435,7 @@ class Workflow: async def _do_process_msg(): #TODO: send msg to agent might be better? - task_result:ComputeTaskResult = await ComputeKernel().do_llm_completion(prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size(),inner_functions) + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,the_role.agent.get_llm_model_name(),the_role.agent.get_max_token_size(),inner_functions) result_str = task_result.result_str logger.info(f"{the_role.role_id} process {msg.sender}:{msg.body},llm str is :{result_str}") diff --git a/src/component/agent_manager/agent_manager.py b/src/component/agent_manager/agent_manager.py index 9b05b62..7c27215 100644 --- a/src/component/agent_manager/agent_manager.py +++ b/src/component/agent_manager/agent_manager.py @@ -11,12 +11,12 @@ logger = logging.getLogger(__name__) class AgentManager: _instance = None - def __new__(cls): + @classmethod + def get_instance(cls): if cls._instance is None: - cls._instance = super(AgentManager, cls).__new__(cls) + cls._instance = AgentManager() return cls._instance - 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") diff --git a/src/component/package_manager/env.py b/src/component/package_manager/env.py index 0ef9cfe..4e1b68a 100644 --- a/src/component/package_manager/env.py +++ b/src/component/package_manager/env.py @@ -134,16 +134,15 @@ class PackageEnv: class PackageEnvManager: _instance = None - def __new__(cls): + @classmethod + def get_instance(cls): if cls._instance is None: - cls._instance = super(PackageEnvManager, cls).__new__(cls) + cls._instance = PackageEnvManager() return cls._instance def __init__(self) -> None: self._pkg_envs = {} - - pass - + def get_env(self,cfg_path:str) -> PackageEnv: if cfg_path in self._pkg_envs: return self._pkg_envs[cfg_path] diff --git a/src/component/workflow_manager/workflow_manager.py b/src/component/workflow_manager/workflow_manager.py index 5603745..daeee13 100644 --- a/src/component/workflow_manager/workflow_manager.py +++ b/src/component/workflow_manager/workflow_manager.py @@ -9,9 +9,10 @@ import os class WorkflowManager: _instance = None - def __new__(cls): + @classmethod + def get_instance(cls): if cls._instance is None: - cls._instance = super(WorkflowManager, cls).__new__(cls) + cls._instance = WorkflowManager() return cls._instance diff --git a/src/requirements.txt b/src/requirements.txt index 7de369e..ac6723c 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -1,9 +1,14 @@ aiofiles==23.2.1 -aiohttp==3.8.5 -openai==0.28.0 -Pillow==10.0.0 -Pillow==10.0.0 +aiohttp==3.7.0 +aioimaplib==1.0.1 +aiosmtplib==2.0.2 +beautifulsoup4==4.12.2 +mail_parser==3.15.0 +openai==0.27.10 +Pillow==10.0.1 prompt_toolkit==3.0.39 +python-telegram-bot==20.5 +Requests==2.31.0 stability_sdk==0.8.4 toml==0.10.2 google-cloud-texttospeech==2.14.1 diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 73d8c93..b182e48 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -4,6 +4,7 @@ import sys import os import logging import re +import toml from typing import Any, Optional, TypeVar, Tuple, Sequence import argparse @@ -17,6 +18,18 @@ from prompt_toolkit.auto_suggest import AutoSuggestFromHistory from prompt_toolkit.completion import WordCompleter from prompt_toolkit.styles import Style + + + +directory = os.path.dirname(__file__) +sys.path.append(directory + '/../../') +from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession,AgentTunnel,TelegramTunnel,CalenderEnvironment,Environment,EmailTunnel + +sys.path.append(directory + '/../../component/') +from agent_manager import AgentManager +from workflow_manager import WorkflowManager + + shell_style = Style.from_dict({ 'title': '#87d7ff bold', #RGB 'content': '#007f00 bold', @@ -24,16 +37,6 @@ shell_style = Style.from_dict({ }) -directory = os.path.dirname(__file__) -sys.path.append(directory + '/../../') -from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession - -sys.path.append(directory + '/../../component/') -from agent_manager import AgentManager -from workflow_manager import WorkflowManager -from aios_kernel import CalenderEnvironment,Environment - - class AIOS_Shell: def __init__(self,username:str) -> None: self.username = username @@ -42,13 +45,13 @@ class AIOS_Shell: async def _handle_no_target_msg(self,bus:AIBus,msg:AgentMsg) -> bool: target_id = msg.target.split(".")[0] - agent : AIAgent = await AgentManager().get(target_id) + agent : AIAgent = await AgentManager.get_instance().get(target_id) if agent is not None: agent.owner_env = Environment.get_env_by_id("calender") bus.register_message_handler(target_id,agent._process_msg) return True - a_workflow = await WorkflowManager().get_workflow(target_id) + a_workflow = await WorkflowManager.get_instance().get_workflow(target_id) if a_workflow is not None: bus.register_message_handler(target_id,a_workflow._process_msg) return True @@ -56,7 +59,7 @@ class AIOS_Shell: return False async def is_agent(self,target_id:str) -> bool: - agent : AIAgent = await AgentManager().get(target_id) + agent : AIAgent = await AgentManager.get_instance().get(target_id) if agent is not None: return True else: @@ -67,13 +70,21 @@ class AIOS_Shell: cal_env.start() Environment.set_env_by_id("calender",cal_env) - AgentManager().initial(os.path.abspath(directory + "/../../../rootfs/")) - WorkflowManager().initial(os.path.abspath(directory + "/../../../rootfs/workflows/")) - open_ai_node = OpenAI_ComputeNode() + AgentManager.get_instance().initial(os.path.abspath(directory + "/../../../rootfs/")) + WorkflowManager.get_instance().initial(os.path.abspath(directory + "/../../../rootfs/workflows/")) + open_ai_node = OpenAI_ComputeNode.get_instance() open_ai_node.start() - ComputeKernel().add_compute_node(open_ai_node) + ComputeKernel.get_instance().add_compute_node(open_ai_node) 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) + + TelegramTunnel.register_to_loader() + EmailTunnel.register_to_loader() + + tunnels_config_path = os.path.abspath(directory + "/../../../rootfs/tunnels.cfg.toml") + tunnel_config = toml.load(tunnels_config_path) + await AgentTunnel.load_all_tunnels_from_config(tunnel_config["tunnels"]) + return True @@ -132,9 +143,9 @@ class AIOS_Shell: db_path = "" if await self.is_agent(self.current_target): - db_path = AgentManager().db_path + db_path = AgentManager.get_instance().db_path else: - db_path = WorkflowManager().db_file + db_path = WorkflowManager.get_instance().db_file chatsession:AIChatSession = AIChatSession.get_session(self.current_target,f"{self.username}#{self.current_topic}",db_path,False) if chatsession is not None: msgs = chatsession.read_history(num,offset) diff --git a/src/service/aios_shell/tg_aios_shell.py b/src/service/aios_shell/tg_aios_shell.py deleted file mode 100644 index 6160ec8..0000000 --- a/src/service/aios_shell/tg_aios_shell.py +++ /dev/null @@ -1,223 +0,0 @@ - -import asyncio -import sys -import os -import logging -import re - -from typing import Any, Optional, TypeVar, Tuple, Sequence -import argparse - -from telegram import ForceReply, Update -from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters - - -directory = os.path.dirname(__file__) -sys.path.append(directory + '/../../') -from aios_kernel import Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession - -sys.path.append(directory + '/../../component/') -from agent_manager import AgentManager -from workflow_manager import WorkflowManager -from aios_kernel import CalenderEnvironment,Environment - -import logging - -# Enable logging -logging.basicConfig( - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO -) -# set higher logging level for httpx to avoid all GET and POST requests being logged -logging.getLogger("httpx").setLevel(logging.WARNING) - -logger = logging.getLogger(__name__) - -class AIOS_Shell: - def __init__(self,username:str) -> None: - self.username = username - self.current_target = "_" - self.current_topic = "default" - - async def _handle_no_target_msg(self,bus:AIBus,msg:AgentMsg) -> bool: - target_id = msg.target.split(".")[0] - agent : AIAgent = await AgentManager().get(target_id) - if agent is not None: - bus.register_message_handler(target_id,agent._process_msg) - return True - - a_workflow = await WorkflowManager().get_workflow(target_id) - if a_workflow is not None: - bus.register_message_handler(target_id,a_workflow._process_msg) - return True - - return False - - async def is_agent(self,target_id:str) -> bool: - agent : AIAgent = await AgentManager().get(target_id) - if agent is not None: - return True - else: - return False - - async def initial(self) -> bool: - cal_env = CalenderEnvironment("calender") - cal_env.start() - Environment.set_env_by_id("calender",cal_env) - - 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 - - - def get_version(self) -> str: - return "0.0.1" - - async def send_msg(self,msg:str,target_id:str,topic:str,sender:str = None) -> str: - agent_msg = AgentMsg() - agent_msg.set(sender,target_id,msg) - agent_msg.topic = topic - 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 - - async def call_func(self,func_name, args): - match func_name: - case 'send': - target_id = args[0] - msg_content = args[1] - topic = args[2] - resp = await self.send_msg(msg_content,target_id,topic,self.username) - - - return resp - case 'open': - if len(args) >= 1: - target_id = args[0] - if len(args) >= 2: - topic = args[1] - - self.current_target = target_id - self.current_topic = topic - show_text = f"current session switch to {topic}@{target_id}" - return show_text - - - case 'history': - num = 10 - offset = 0 - if args is not None: - if len(args) >= 1: - num = args[0] - if len(args) >= 2: - offset = args[1] - - db_path = "" - if await self.is_agent(self.current_target): - db_path = AgentManager().db_path - else: - db_path = WorkflowManager().db_file - chatsession:AIChatSession = AIChatSession.get_session(self.current_target,f"{self.username}#{self.current_topic}",db_path,False) - if chatsession is not None: - msgs = chatsession.read_history(num,offset) - result_str = "" - for msg in reversed(msgs): - result_str += f"{msg.sender} >>> {msg.body}" - result_str += f"\n-------------------\n" - return result_str - return f"chatsession not found" - case 'help': - return "help info " - -shell:AIOS_Shell = None - -async def _init_shell(): - global shell - if shell is None: - shell = AIOS_Shell("user") - await shell.initial() - return - -def parse_function_call(func_string): - match = re.search(r'\s*(\w+)\s*\(\s*(.*)\s*\)\s*', func_string) - if not match: - return None - - func_name = match.group(1) - params_string = match.group(2).strip() - params = re.split(r'\s*,\s*(?=(?:[^"]*"[^"]*")*[^"]*$)', params_string) - params = [param.strip('"') for param in params] - if len(params[0]) == 0: - params = None - - return func_name, params - -async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - """Send a message when the command /start is issued.""" - await _init_shell() - user = update.effective_user - await update.message.reply_html( - rf"Hi {user.mention_html()}!", - reply_markup=ForceReply(selective=True), - ) - - -async def on_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - await _init_shell() - cmd_msg = update.message.text[1:] - logger.info("command:" + cmd_msg) - func_call = parse_function_call(cmd_msg) - if func_call: - shell.username = update.effective_user - show_text = await shell.call_func(func_call[0], func_call[1]) - await update.message.reply_text(show_text) - else: - await update.message.reply_text("command not found!") - -async def on_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - await _init_shell() - shell.username = update.effective_user - resp = await shell.send_msg(update.message.text,shell.current_target,shell.current_topic,shell.username) - await update.message.reply_text(resp) - - - - -def main() -> None: - """Start the bot.""" - # Create the Application and pass it your bot's token. - tg_robot_token = os.getenv("TG_ROBOT_TOKEN") - if tg_robot_token is None: - print("TG_ROBOT_TOKEN not found!") - tg_robot_token = input("input your telegram robot token:") - - - if tg_robot_token is None: - return - - application = Application.builder().token(tg_robot_token).build() - - # on different commands - answer in Telegram - application.add_handler(CommandHandler("start", start)) - application.add_handler(CommandHandler("help", on_command)) - application.add_handler(CommandHandler("send", on_command)) - application.add_handler(CommandHandler("open", on_command)) - application.add_handler(CommandHandler("history", on_command)) - - # on non command i.e message - echo the message on Telegram - application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, on_message)) - - # Run the bot until the user presses Ctrl-C - application.run_polling(allowed_updates=Update.ALL_TYPES) - - -if __name__ == "__main__": - main() \ No newline at end of file