gyt/opendan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'MVP' into MVP

Browse Source
This commit is contained in:
Liu Zhicong
2023-09-17 18:15:24 -07:00
committed by GitHub
parent 9c1d672e3b 43c70489e7
commit 0e14cedd19
19 changed files with 501 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aios_kernel/__init__.py
+3
View File
@@ -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
src/aios_kernel/agent.py
+5 -5
View File
@@ -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
src/aios_kernel/compute_kernel.py
+5 -10
View File
@@ -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 = {}
src/aios_kernel/contact_manager.py
+34
View File
@@ -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)
src/aios_kernel/email_tunnel.py
+143
View File
@@ -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}")
src/aios_kernel/open_ai_node.py
+9 -8
View File
@@ -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()
src/aios_kernel/stability_node.py
+11 -11
View File
@@ -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...")
src/aios_kernel/tg_tunnel.py
+116
View File
@@ -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)
src/aios_kernel/tunnel.py
+63
View File
@@ -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
src/aios_kernel/workflow.py
+7 -3
View File
@@ -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}")
src/component/agent_manager/agent_manager.py
+3 -3
View File
@@ -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")
src/component/package_manager/env.py
+4 -5
View File
@@ -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]
src/component/workflow_manager/workflow_manager.py
+3 -2
View File
@@ -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
src/requirements.txt
+9 -4
View File
@@ -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
src/service/aios_shell/aios_shell.py
+30 -19
View File
@@ -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)
src/service/aios_shell/tg_aios_shell.py
-223
View File
@@ -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()