From 4e45130140a3908406dff3575887bf67ba874d2e Mon Sep 17 00:00:00 2001 From: wugren Date: Thu, 21 Sep 2023 15:52:56 +0800 Subject: [PATCH 1/5] story maker --- rootfs/agents/fairy_tale_writer/agent.toml | 7 ++ rootfs/agents/studio_director/agent.toml | 7 ++ rootfs/workflows/story_maker/workflow.toml | 37 +++++++ src/aios_kernel/__init__.py | 1 + src/aios_kernel/chatsession.py | 27 ++--- src/aios_kernel/compute_kernel.py | 50 +++++++-- src/aios_kernel/google_text_to_speech_node.py | 81 ++++++++++++-- src/aios_kernel/text_to_speech_function.py | 101 ++++++++++++++++++ src/aios_kernel/workflow_env.py | 45 ++++---- src/service/aios_shell/aios_shell.py | 73 +++++++------ 10 files changed, 341 insertions(+), 88 deletions(-) create mode 100644 rootfs/agents/fairy_tale_writer/agent.toml create mode 100644 rootfs/agents/studio_director/agent.toml create mode 100644 rootfs/workflows/story_maker/workflow.toml create mode 100644 src/aios_kernel/text_to_speech_function.py diff --git a/rootfs/agents/fairy_tale_writer/agent.toml b/rootfs/agents/fairy_tale_writer/agent.toml new file mode 100644 index 0000000..1f2256a --- /dev/null +++ b/rootfs/agents/fairy_tale_writer/agent.toml @@ -0,0 +1,7 @@ +instance_id = "fairy_tale_writer" +fullname = "tracy wang" +llm_model_name = "gpt-3.5-turbo-16k-0613" + +[[prompt]] +role = "system" +content = "你是一个童话做作家，能够写出各种有趣的童话。" diff --git a/rootfs/agents/studio_director/agent.toml b/rootfs/agents/studio_director/agent.toml new file mode 100644 index 0000000..eb61bca --- /dev/null +++ b/rootfs/agents/studio_director/agent.toml @@ -0,0 +1,7 @@ +instance_id = "studio_director" +fullname = "tracy wang" +llm_model_name = "gpt-3.5-turbo-16k-0613" + +[[prompt]] +role = "system" +content = "你是一个演播导演，请将下面故事改编成朗读剧本，提取旁白和角色台词，每个角色需要有性别、年龄、以及每句台词的语气。并调用text_to_speech function生成音频数据。" diff --git a/rootfs/workflows/story_maker/workflow.toml b/rootfs/workflows/story_maker/workflow.toml new file mode 100644 index 0000000..30a26d6 --- /dev/null +++ b/rootfs/workflows/story_maker/workflow.toml @@ -0,0 +1,37 @@ +name = "story_maker" + + +[filter] +"*" = "manager" + +[roles.manager] +name = "manager" +fullname = "总导演" +agent="manager" +[[roles.manager.prompt]] +role="system" +content=""" +你是一个语音故事制作总导演,与客户对接并向团队下达指令。你的团队分为下面几个成员:writer,studio_director。一个故事制作分成两个阶段：让writer写出故事，再交由studio_director演播故事。你的基本工作模式是： +1. 收到客户的明确的指令后,让writer写出故事 +2. 将writer写出的故事交给studio_director演播 +3. 当你决定要和成员通信时，请使用下面形式输出需要通信的消息 +``` +##/send_msg 成员名称 +内容 +``` +""" + +[roles.writer] +name = "writer" +agent = "fairy_tale_writer" +fullname = "作家" +[[roles.writer.prompt]] +role="system" +content="" + +[roles.studio_director] +name = "studio_director" +agent = "studio_director" +[[roles.studio_director.prompt]] +role="system" +content="" diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index 79d4456..9eba338 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -19,6 +19,7 @@ from .tg_tunnel import TelegramTunnel from .email_tunnel import EmailTunnel from .storage import ResourceLocation,AIStorage,UserConfig,UserConfigItem from .contact_manager import ContactManager,Contact,FamilyMember +from .text_to_speech_function import TextToSpeechFunction AIOS_Version = "0.5.1, build 2023-9-17" diff --git a/src/aios_kernel/chatsession.py b/src/aios_kernel/chatsession.py index 5893683..ceb9821 100644 --- a/src/aios_kernel/chatsession.py +++ b/src/aios_kernel/chatsession.py @@ -35,10 +35,10 @@ class ChatSessionDB: self._create_table(conn) return conn - + def close(self): if not hasattr(self.local, 'conn'): - return + return self.local.conn.close() def _create_table(self, conn): @@ -56,7 +56,7 @@ class ChatSessionDB: # create messages table # reciver_id could be None - + conn.execute(""" CREATE TABLE IF NOT EXISTS Messages ( MessageID TEXT PRIMARY KEY, @@ -142,7 +142,7 @@ class ChatSessionDB: except Error as e: logging.error("Error occurred while inserting message: %s", e) return -1 # return -1 if an error occurs - + def get_chatsession_by_id(self, session_id): """Get a message by its ID""" conn = self._get_conn() @@ -150,7 +150,7 @@ class ChatSessionDB: c.execute("SELECT * FROM ChatSessions WHERE SessionID = ?", (session_id,)) chatsession = c.fetchone() return chatsession - + def get_chatsession_by_owner_topic(self, owner_id, topic): """Get a chatsession by its owner and topic""" conn = self._get_conn() @@ -175,7 +175,7 @@ class ChatSessionDB: except Error as e: logging.error("Error occurred while getting sessions: %s", e) return -1, None # return -1 and None if an error occurs - + def get_message_by_id(self, message_id): """Get a message by its ID""" conn =self._get_conn() @@ -216,7 +216,7 @@ class ChatSessionDB: except Error as e: logging.error("Error occurred while updating message status: %s", e) return -1 # return -1 if an error occurs - + # chat session store the chat history between owner and agent # chat session might be large, so can read / write at stream mode. @@ -230,7 +230,7 @@ class AIChatSession: # cls._dbs[db_path] = db # db.get_chatsession_by_id(session_id) # #result = AIChatSession() - + @classmethod def get_session(cls,owner_id:str,session_topic:str,db_path:str,auto_create = True) -> str: db = cls._dbs.get(db_path) @@ -249,21 +249,22 @@ class AIChatSession: result = AIChatSession(owner_id,session[0],db) result.topic = session_topic - return result - + return result + def __init__(self,owner_id:str, session_id:str, db:ChatSessionDB) -> None: self.owner_id :str = owner_id self.session_id : str = session_id self.db : ChatSessionDB = db - + self.topic : str = None self.start_time : str = None def get_owner_id(self) -> str: return self.owner_id - + def read_history(self, number:int=10,offset=0) -> [AgentMsg]: + return [] msgs = self.db.get_messages(self.session_id, number, offset) result = [] for msg in msgs: @@ -298,4 +299,4 @@ class AIChatSession: # """chat session changed event handler""" # pass - #TODO : add iterator interface for read chat history + #TODO : add iterator interface for read chat history diff --git a/src/aios_kernel/compute_kernel.py b/src/aios_kernel/compute_kernel.py index 97f9b9b..5859efe 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/src/aios_kernel/compute_kernel.py @@ -7,7 +7,7 @@ from asyncio import Queue from .agent import AgentPrompt from .compute_node import ComputeNode -from .compute_task import ComputeTask, ComputeTaskState, ComputeTaskResult +from .compute_task import ComputeTask, ComputeTaskState, ComputeTaskResult, ComputeTaskType logger = logging.getLogger(__name__) @@ -23,7 +23,7 @@ class ComputeKernel: if cls._instance is None: cls._instance = ComputeKernel() return cls._instance - + def __init__(self) -> None: self.is_start = False self.task_queue = Queue() @@ -73,7 +73,7 @@ class ComputeKernel: hit_pos = random.randint(0, total_weights - 1) for i in range(min(len(support_nodes) - 1, hit_pos), -1, -1): if support_nodes[i]["pos"] <= hit_pos: - return node + return support_nodes[i]["node"] logger.warning( f"task {task.display()} is not support by any compute node") @@ -137,7 +137,7 @@ class ComputeKernel: task_req.set_text_embedding_params(input,model_name) self.run(task_req) return task_req - + async def do_text_embedding(self,input:str,model_name:Optional[str] = None) -> [float]: task_req = self.text_embedding(input,model_name) async def check_timer(): @@ -155,11 +155,45 @@ class ComputeKernel: await asyncio.sleep(0.5) check_times += 1 - + await asyncio.create_task(check_timer()) if task_req.state == ComputeTaskState.DONE: return task_req.result.result - - return "error!" - + + return "error!" + + async def do_text_to_speech(self, + input:str, + language_code:Optional[str] = None, + gender: Optional[str] = None, + age: Optional[str] = None, + voice_name: Optional[str] = None, + tone: Optional[str] = None): + task_req = ComputeTask() + task_req.params["text"] = input + task_req.params["language_code"] = language_code + task_req.params["gender"] = gender + task_req.params["age"] = age + task_req.params["voice_name"] = voice_name + task_req.params["tone"] = tone + task_req.task_type = ComputeTaskType.TEXT_2_VOICE + self.run(task_req) + + check_times = 0 + while True: + if task_req.state == ComputeTaskState.DONE: + break + if task_req.state == ComputeTaskState.ERROR: + break + if check_times >= 60: + task_req.state = ComputeTaskState.ERROR + break + await asyncio.sleep(0.5) + check_times += 1 + + if task_req.state == ComputeTaskState.DONE: + return task_req.result.result + else: + raise Exception("do_text_to_speech failed!") + diff --git a/src/aios_kernel/google_text_to_speech_node.py b/src/aios_kernel/google_text_to_speech_node.py index 7709eef..b05f276 100644 --- a/src/aios_kernel/google_text_to_speech_node.py +++ b/src/aios_kernel/google_text_to_speech_node.py @@ -21,24 +21,69 @@ see:https://cloud.google.com/text-to-speech/docs/before-you-begin class GoogleTextToSpeechNode(ComputeNode): _instance = None - def __new__(cls, *args, **kwargs): + @classmethod + def get_instance(cls): if cls._instance is None: - cls._instance = super(GoogleTextToSpeechNode, cls).__new__(cls) - cls._instance.is_start = False + cls._instance = cls() return cls._instance def __init__(self): super().__init__() - if self.is_start is True: - logger.warn("GoogleTextToSpeechNode is already start") - return - - self.is_start = True self.node_id = "google_text_to_speech_node" self.task_queue = Queue() self.client = texttospeech.TextToSpeechClient() + self.language_list = { + "cnm-CN": { + "female": ["cmn-CN-Standard-A", + "cmn-CN-Standard-D", + "cmn-CN-Wavenet-A", + "cmn-CN-Wavenet-D", + "cmn-TW-Standard-A", + "cmn-TW-Wavenet-A"], + "man": ["cmn-CN-Standard-B", + "cmn-CN-Standard-C", + "cmn-CN-Wavenet-B", + "cmn-CN-Wavenet-C", + "cmn-TW-Standard-B", + "cmn-TW-Standard-C", + "cmn-TW-Wavenet-B", + "cmn-TW-Wavenet-C"] + }, + "en-US": { + "female": ["en-US-Neural2-C", + "en-US-Neural2-E", + "en-US-Neural2-F", + "en-US-Neural2-G", + "en-US-Neural2-H", + "en-US-News-K", + "en-US-News-L", + "en-US-Standard-C", + "en-US-Standard-E", + "en-US-Standard-F", + "en-US-Standard-G", + "en-US-Standard-H", + "en-US-Studio-O", + "en-US-Wavenet-C", + "en-US-Wavenet-E", + "en-US-Wavenet-F", + "en-US-Wavenet-G", + "en-US-Wavenet-H"], + "man": ["en-US-Polyglot-1", + "en-US-Standard-A", + "en-US-Standard-B", + "en-US-Standard-D", + "en-US-Standard-I", + "en-US-Standard-J", + "en-US-Studio-M", + "en-US-Wavenet-A", + "en-US-Wavenet-B", + "en-US-Wavenet-D", + "en-US-Wavenet-I", + "en-US-Wavenet-J"] + } + } self.start() def start(self): @@ -64,10 +109,24 @@ class GoogleTextToSpeechNode(ComputeNode): task.state = ComputeTaskState.RUNNING language_code = task.params["language_code"] text = task.params["text"] + voice_name = task.params["voice_name"] + gender = task.params["gender"] + age = task.params["age"] + + if language_code == "zh": + language_code = "cnm-CN" + elif language_code == "en": + language_code = "en-US" + else: + raise Exception(f"language_code {language_code} not support") + + lang_list = self.language_list[language_code][gender] + voice = lang_list[hash(voice_name) % len(lang_list)] synthesis_input = texttospeech.SynthesisInput(text=text) voice = texttospeech.VoiceSelectionParams(language_code=language_code, - ssml_gender=texttospeech.SsmlVoiceGender.NEUTRAL) + ssml_gender=texttospeech.SsmlVoiceGender.NEUTRAL, + name=voice) audio_config = texttospeech.AudioConfig(audio_encoding=texttospeech.AudioEncoding.MP3) @@ -95,8 +154,8 @@ class GoogleTextToSpeechNode(ComputeNode): def get_capacity(self): return 0 - def is_support(self, task_type: ComputeTaskType) -> bool: - if task_type == ComputeTaskType.TEXT_2_VOICE: + def is_support(self, task: ComputeTask) -> bool: + if task.task_type == ComputeTaskType.TEXT_2_VOICE: return True return False diff --git a/src/aios_kernel/text_to_speech_function.py b/src/aios_kernel/text_to_speech_function.py new file mode 100644 index 0000000..3237974 --- /dev/null +++ b/src/aios_kernel/text_to_speech_function.py @@ -0,0 +1,101 @@ +import io +import logging +import os +import random +from typing import Dict + +from aios_kernel import ComputeKernel +from aios_kernel.ai_function import AIFunction + +from pydub import AudioSegment + +logger = logging.getLogger(__name__) + +class TextToSpeechFunction(AIFunction): + def __init__(self): + self.func_id = "text_to_speech" + self.description = "根据输入的剧本生成音频数据" + + def get_name(self) -> str: + return self.func_id + + def get_description(self) -> str: + return self.description + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "language": {"type": "string", "description": "演播语言", "enum": ["zh", "en"]}, + "roles": {"type": "array", "items": { + "type": "object", + "properties": { + "name": {"type": "string", "description": "角色名字"}, + "gender": {"type": "string", "description": "角色性别", "enum": ["man", "female"]}, + "age": {"type": "string", "description": "年龄", "enum": ["child", "adult"]}, + }}}, + "lines": {"type": "array", "items": { + "type": "object", + "properties": { + "name": {"type": "string", "description": "角色名字"}, + "tone": {"type": "string", "description": "演播情感", + "enum": ["happy", "sad", "angry", "fear", "disgust", "surprise", "neutral"]}, + "text": {"type": "string", "description": "台词"}, + } + }} + } + } + + async def execute(self, **kwargs) -> str: + logger.info(f"execute text_to_speech function: {kwargs}") + + language = kwargs.get("language") + if language is None: + language = "zh" + roles = kwargs.get("roles") + lines = kwargs.get("lines") + + audio = None + for line in lines: + name = line.get("name") + tone = line.get("tone") + text = line.get("text") + gender = None + age = None + for role in roles: + role_name = role.get("name") + if role_name == name: + gender = role.get("gender") + age = role.get("age") + break + i = 0 + while i < 3: + try: + data = await ComputeKernel.get_instance().do_text_to_speech(text, language, gender, age, name, tone) + if audio is None: + audio = AudioSegment.from_mp3(io.BytesIO(data)) + else: + audio = audio + AudioSegment.from_mp3(io.BytesIO(data)) + break + except Exception as e: + logger.error(f"do_text_to_speech failed: {e}") + i += 1 + continue + + if audio is not None: + path = os.path.join(os.curdir, "{}.mp3".format(random.sample('zyxwvutsrqponmlkjihgfedcba', 10))) + audio.export(path, format="mp3") + return "complete.file path:{}".format(path) + else: + return "failed" + + def is_local(self) -> bool: + return True + + def is_in_zone(self) -> bool: + return True + + def is_ready_only(self) -> bool: + return False + + diff --git a/src/aios_kernel/workflow_env.py b/src/aios_kernel/workflow_env.py index 120b3fb..b7211ef 100644 --- a/src/aios_kernel/workflow_env.py +++ b/src/aios_kernel/workflow_env.py @@ -7,6 +7,8 @@ from sqlite3 import Error import threading import logging from typing import Optional + +from .text_to_speech_function import TextToSpeechFunction from .environment import Environment,EnvironmentEvent from .ai_function import SimpleAIFunction from .storage import AIStorage @@ -23,8 +25,8 @@ class CalenderEvent(EnvironmentEvent): self.data = data def display(self) -> str: - return f"#event timer:{self.data}" - + return f"#event timer:{self.data}" + # AI Calender GOAL: Let user use "create notify after 2 days" to create a timer event class CalenderEnvironment(Environment): def __init__(self, env_id: str) -> None: @@ -39,7 +41,7 @@ class CalenderEnvironment(Environment): #self.add_ai_function(SimpleAIFunction("serach_events", # "search events in calender", # self._search_events)) - + get_param = { "start_time": "start time (UTC) of event", "end_time": "end time (UTC) of event" @@ -59,14 +61,14 @@ class CalenderEnvironment(Environment): self.add_ai_function(SimpleAIFunction("add_event", "add event to calender", self._add_event,add_param)) - + delete_param = { "event_id": "id of event" } self.add_ai_function(SimpleAIFunction("delete_event", "delete event from calender", self._delete_event,delete_param)) - + update_param = { "event_id": "id of event", "new_title": "new title of event", @@ -79,7 +81,7 @@ class CalenderEnvironment(Environment): self.add_ai_function(SimpleAIFunction("update_event", "update event in calender", self._update_event,update_param)) - + #self.add_ai_function(SimpleAIFunction("user_confirm", # "user confirm", # self._user_confirm)) @@ -98,7 +100,7 @@ class CalenderEnvironment(Environment): ); """) await db.commit() - + async def _add_event(self,title, start_time, end_time, participants=None, location=None, details=None): async with aiosqlite.connect(self.db_file) as db: await db.execute(""" @@ -127,7 +129,7 @@ class CalenderEnvironment(Environment): _event["details"] = row[6] result[row[0]] = _event return json.dumps(result, indent=4, sort_keys=True) - + async def _get_events_by_time_range(self,start_time, end_time): async with aiosqlite.connect(self.db_file) as db: cursor = await db.execute(""" @@ -153,7 +155,7 @@ class CalenderEnvironment(Environment): return "No event." return json.dumps(result, indent=4, sort_keys=True) - + async def _update_event(self,event_id, new_title=None, new_participants=None, new_location=None, new_details=None ,start_time=None, end_time=None): fields_to_update = [] values = [] @@ -191,8 +193,8 @@ class CalenderEnvironment(Environment): WHERE id = ?; """ - values.append(event_id) - + values.append(event_id) + async with aiosqlite.connect(self.db_file) as db: await db.execute(sql_update_query, values) await db.commit() @@ -212,16 +214,16 @@ class CalenderEnvironment(Environment): async def start(self) -> None: if self.is_run: - return + return self.is_run = True await self.init_db() - + self.register_get_handler("now",self.get_now) async def timer_loop(): while True: if self.is_run == False: break - + await asyncio.sleep(1.0) now = datetime.now() formatted_time = now.strftime('%Y-%m-%d %H:%M:%S') @@ -238,13 +240,13 @@ class CalenderEnvironment(Environment): def get_now(self)->str: now = datetime.now() formatted_time = now.strftime('%Y-%m-%d %H:%M:%S') - return formatted_time + return formatted_time async def _get_now(self) -> str: now = datetime.now() formatted_time = now.strftime('%Y-%m-%d %H:%M:%S') return formatted_time - + # Default Workflow Environment(Context) class WorkflowEnvironment(Environment): def __init__(self, env_id: str,db_file:str) -> None: @@ -252,6 +254,7 @@ class WorkflowEnvironment(Environment): self.db_file = db_file self.local = threading.local() self.table_name = "WorkflowEnv_" + env_id + self.add_ai_function(TextToSpeechFunction()) def _get_conn(self): @@ -259,7 +262,7 @@ class WorkflowEnvironment(Environment): if not hasattr(self.local, 'conn'): self.local.conn = self._create_connection() return self.local.conn - + def _create_connection(self): """ create a database connection to a SQLite database """ conn = None @@ -273,10 +276,10 @@ class WorkflowEnvironment(Environment): self._create_table(conn) return conn - + def close(self): if not hasattr(self.local, 'conn'): - return + return self.local.conn.close() def _create_table(self, conn): @@ -293,7 +296,7 @@ class WorkflowEnvironment(Environment): conn.commit() except Error as e: logging.error("Error occurred while creating tables: %s", e) - + def _do_get_value(self, key: str) -> str | None: try: conn = self._get_conn() @@ -311,7 +314,7 @@ class WorkflowEnvironment(Environment): super().set_value(key,str_value) if is_storage is False: return - + try: conn = self._get_conn() conn.execute(""" diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 984bb08..f3df06a 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -64,17 +64,17 @@ class AIOS_Shell: target_id = msg.target.split(".")[0] agent : AIAgent = await AgentManager.get_instance().get(target_id) if agent is not None: - agent.owner_env = Environment.get_env_by_id("calender") + 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_instance().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_instance().get(target_id) if agent is not None: @@ -96,7 +96,10 @@ class AIOS_Shell: logger.error("openai node initial failed!") return False ComputeKernel.get_instance().add_compute_node(open_ai_node) - + + google_text_to_speech_node = GoogleTextToSpeechNode.get_instance() + ComputeKernel.get_instance().add_compute_node(google_text_to_speech_node) + llama_ai_node = LocalLlama_ComputeNode() await llama_ai_node.start() # ComputeKernel.get_instance().add_compute_node(llama_ai_node) @@ -116,7 +119,7 @@ class AIOS_Shell: tunnels_config_path = os.path.abspath(f"{user_data_dir}/etc/tunnels.cfg.toml") tunnel_config = None - try: + try: tunnel_config = toml.load(tunnels_config_path) if tunnel_config is not None: await AgentTunnel.load_all_tunnels_from_config(tunnel_config) @@ -156,11 +159,11 @@ class AIOS_Shell: case "email": tunnel_config["type"] = "EmailTunnel" case _: - error_text = FormattedText([("class:error", f"tunnel type {tunnel_type}not support!")]) + error_text = FormattedText([("class:error", f"tunnel type {tunnel_type}not support!")]) print_formatted_text(error_text,style=shell_style) return None - intro_text = FormattedText([("class:prompt", tunnel_introduce)]) + intro_text = FormattedText([("class:prompt", tunnel_introduce)]) print_formatted_text(intro_text,style=shell_style) for key,item in intpu_table.items(): user_input = await try_get_input(f"{key} : {item.desc}") @@ -174,7 +177,7 @@ class AIOS_Shell: async def append_tunnel_config(self,tunnel_config): user_data_dir = AIStorage.get_instance().get_myai_dir() tunnels_config_path = os.path.abspath(f"{user_data_dir}/etc/tunnels.cfg.toml") - try: + try: all_tunnels = toml.load(tunnels_config_path) if all_tunnels is not None: all_tunnels[tunnel_config["tunnel_id"]] = tunnel_config @@ -255,7 +258,7 @@ class AIOS_Shell: AIStorage.get_instance().get_user_config().set_value(key,value) await AIStorage.get_instance().get_user_config().save_to_user_config() show_text = FormattedText([("class:title", f"set {key} to {value} success!")]) - + return show_text case 'connect': show_text = FormattedText([("class:title", "args error, /connect $target telegram | email")]) @@ -293,8 +296,8 @@ class AIOS_Shell: if len(args) >= 1: self.username = args[0] AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg) - - return self.username + " login success!" + + return self.username + " login success!" case 'history': num = 10 offset = 0 @@ -324,9 +327,9 @@ class AIOS_Shell: return FormattedText([("class:title", f"help~~~")]) -########################################################################################################################## +########################################################################################################################## history = FileHistory('aios_shell_history.txt') -session = PromptSession(history=history) +session = PromptSession(history=history) def parse_function_call(func_string): if len(func_string) > 2: @@ -337,7 +340,7 @@ def parse_function_call(func_string): return func_name, params else: return None - + async def try_get_input(desc:str,check_func:callable = None) -> str: user_input = await session.prompt_async(f"{desc} \nType /exit to abort. \nPlease input:",style=shell_style) err_str = "" @@ -347,13 +350,13 @@ async def try_get_input(desc:str,check_func:callable = None) -> str: return user_input else: return None - + else: is_ok,err_str = check_func(user_input) if is_ok: return user_input - - error_text = FormattedText([("class:error", err_str)]) + + error_text = FormattedText([("class:error", err_str)]) print_formatted_text(error_text,style=shell_style) return await try_get_input(desc,check_func) @@ -373,7 +376,7 @@ async def main_daemon_loop(shell:AIOS_Shell): return 0 def print_welcome_screen(): - print("\033[1;31m") + print("\033[1;31m") logo = """ \t _______ ____________________ __ \t __ __ \______________________ __ \__ |__ | / / @@ -384,9 +387,9 @@ def print_welcome_screen(): """ print(logo) - print("\033[0m") + print("\033[0m") - print("\033[1;32m \t\tWelcome to OpenDAN - Your Personal AI OS\033[0m\n") + print("\033[1;32m \t\tWelcome to OpenDAN - Your Personal AI OS\033[0m\n") introduce = """ \tThe core goal of version 0.5.1 is to turn the concept of AIOS into code and get it up and running as quickly as possible. @@ -396,12 +399,12 @@ def print_welcome_screen(): \twe intend to strengthen some components. This document will explain these changes and provide an update \ton the current development progress of MVP(0.5.1,0.5.2) -""" +""" print(introduce) - print(f"\033[1;34m \t\tVersion: {AIOS_Version}\n\033") - print("\033[1;33m \tOpenDAN is an open-source project, let's define the future of Humans and AI together.\033[0m") - print("\033[1;33m \tGithub\t: https://github.com/fiatrete/OpenDAN-Personal-AI-OS\033[0m") + print(f"\033[1;34m \t\tVersion: {AIOS_Version}\n\033") + print("\033[1;33m \tOpenDAN is an open-source project, let's define the future of Humans and AI together.\033[0m") + print("\033[1;33m \tGithub\t: https://github.com/fiatrete/OpenDAN-Personal-AI-OS\033[0m") print("\033[1;33m \tWebsite\t: https://www.opendan.ai\033[0m") print("\n\n") @@ -412,19 +415,19 @@ async def main(): logging.basicConfig(filename="aios_shell.log",filemode="w",encoding='utf-8',force=True, level=logging.INFO, format='[%(asctime)s]%(name)s[%(levelname)s]: %(message)s') - + if os.path.isdir(f"{directory}/../../../rootfs"): AIStorage.get_instance().is_dev_mode = True else: - AIStorage.get_instance().is_dev_mode = False + AIStorage.get_instance().is_dev_mode = False is_daemon = False if os.name != 'nt': if os.getppid() == 1: is_daemon = True - shell = AIOS_Shell("user") - shell.declare_all_user_config() + shell = AIOS_Shell("user") + shell.declare_all_user_config() await AIStorage.get_instance().initial() check_result = AIStorage.get_instance().get_user_config().check_config() if check_result is not None: @@ -435,7 +438,7 @@ async def main(): #Remind users to enter necessary configurations. if await get_user_config_from_input(check_result) is False: return 1 - + init_result = await shell.initial() if init_result is False: if is_daemon: @@ -451,8 +454,8 @@ async def main(): proxy.apply_storage() #TODO: read last input config - completer = WordCompleter(['/send $target $msg $topic', - '/open $target $topic', + completer = WordCompleter(['/send $target $msg $topic', + '/open $target $topic', '/history $num $offset', '/login $username', '/connect $target', @@ -462,7 +465,7 @@ async def main(): '/set_config $key', '/list_config', '/show', - '/exit', + '/exit', '/help'], ignore_case=True) current = AIStorage.get_instance().get_user_config().get_value("shell.current") @@ -470,7 +473,7 @@ async def main(): shell.current_target = current[1] shell.current_topic = current[0] - await asyncio.sleep(0.2) + 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: @@ -491,6 +494,6 @@ async def main(): #print_formatted_text(f"{shell.username}<->{shell.current_topic}@{shell.current_target} >>> {resp}",style=shell_style) -if __name__ == "__main__": +if __name__ == "__main__": asyncio.run(main()) From 5c2dd13ab2e8735e5673d05957be74ed7f241343 Mon Sep 17 00:00:00 2001 From: wugren Date: Fri, 22 Sep 2023 00:09:21 +0800 Subject: [PATCH 2/5] Fix story maker bug --- rootfs/agents/fairy_tale_writer/agent.toml | 1 + rootfs/agents/manager/agent.toml | 2 + rootfs/agents/studio_director/agent.toml | 4 +- rootfs/workflows/story_maker/workflow.toml | 6 +- src/aios_kernel/agent.py | 78 +++++++------- src/aios_kernel/bus.py | 44 ++++---- src/aios_kernel/compute_kernel.py | 4 +- src/aios_kernel/open_ai_node.py | 23 +++-- src/aios_kernel/text_to_speech_function.py | 4 +- src/aios_kernel/workflow.py | 114 +++++++++++---------- 10 files changed, 147 insertions(+), 133 deletions(-) diff --git a/rootfs/agents/fairy_tale_writer/agent.toml b/rootfs/agents/fairy_tale_writer/agent.toml index 1f2256a..7955df6 100644 --- a/rootfs/agents/fairy_tale_writer/agent.toml +++ b/rootfs/agents/fairy_tale_writer/agent.toml @@ -1,6 +1,7 @@ instance_id = "fairy_tale_writer" fullname = "tracy wang" llm_model_name = "gpt-3.5-turbo-16k-0613" +enable_function = [] [[prompt]] role = "system" diff --git a/rootfs/agents/manager/agent.toml b/rootfs/agents/manager/agent.toml index 5bb2ba6..e4bba03 100644 --- a/rootfs/agents/manager/agent.toml +++ b/rootfs/agents/manager/agent.toml @@ -1,5 +1,7 @@ instance_id = "agent:xxxxxxabcde" fullname = "musk" +enable_function = [] + [[prompt]] role = "system" content = "你有丰富的管理技能，擅长将复杂工作拆解成简单的任务，让团队成员高效协作。" diff --git a/rootfs/agents/studio_director/agent.toml b/rootfs/agents/studio_director/agent.toml index eb61bca..65a7c07 100644 --- a/rootfs/agents/studio_director/agent.toml +++ b/rootfs/agents/studio_director/agent.toml @@ -1,7 +1,7 @@ instance_id = "studio_director" fullname = "tracy wang" -llm_model_name = "gpt-3.5-turbo-16k-0613" +enable_function = ["text_to_speech"] [[prompt]] role = "system" -content = "你是一个演播导演，请将下面故事改编成朗读剧本，提取旁白和角色台词，每个角色需要有性别、年龄、以及每句台词的语气。并调用text_to_speech function生成音频数据。" +content = "你是一个故事播音员，请将下面故事改编成播音剧本，提取旁白和角色台词，以及每个角色需要有性别、年龄、以及每句台词的语气，最后生成音频文件。" diff --git a/rootfs/workflows/story_maker/workflow.toml b/rootfs/workflows/story_maker/workflow.toml index 30a26d6..540665f 100644 --- a/rootfs/workflows/story_maker/workflow.toml +++ b/rootfs/workflows/story_maker/workflow.toml @@ -8,10 +8,12 @@ name = "story_maker" name = "manager" fullname = "总导演" agent="manager" +enable_function = [] + [[roles.manager.prompt]] role="system" content=""" -你是一个语音故事制作总导演,与客户对接并向团队下达指令。你的团队分为下面几个成员:writer,studio_director。一个故事制作分成两个阶段：让writer写出故事，再交由studio_director演播故事。你的基本工作模式是： +你是一个语音故事制作总导演,与客户对接并向团队下达指令。你的团队分为下面几个成员:writer,studio_director。一个故事制作分成两个阶段：让writer写出故事，再交由studio_director演播故事生成音频文件。你的基本工作模式是： 1. 收到客户的明确的指令后,让writer写出故事 2. 将writer写出的故事交给studio_director演播 3. 当你决定要和成员通信时，请使用下面形式输出需要通信的消息 @@ -25,6 +27,7 @@ content=""" name = "writer" agent = "fairy_tale_writer" fullname = "作家" +enable_function = [] [[roles.writer.prompt]] role="system" content="" @@ -32,6 +35,7 @@ content="" [roles.studio_director] name = "studio_director" agent = "studio_director" +enable_function = ["text_to_speech"] [[roles.studio_director.prompt]] role="system" content="" diff --git a/src/aios_kernel/agent.py b/src/aios_kernel/agent.py index cc79637..653ac0a 100644 --- a/src/aios_kernel/agent.py +++ b/src/aios_kernel/agent.py @@ -26,7 +26,7 @@ class AgentPrompt: self.system_message = None def as_str(self)->str: - result_str = "" + result_str = "" if self.system_message: result_str += self.system_message.get("role") + ":" + self.system_message.get("content") + "\n" if self.messages: @@ -34,18 +34,18 @@ class AgentPrompt: result_str += msg.get("role") + ":" + msg.get("content") + "\n" return result_str - + def to_message_list(self): result = [] if self.system_message: result.append(self.system_message) result.extend(self.messages) return result - + def append(self,prompt): if prompt is None: return - + if prompt.system_message is not None: if self.system_message is None: self.system_message = prompt.system_message @@ -99,9 +99,9 @@ class AIAgentTemplete: logger.error("load prompt from config failed!") return False - + return True - + class AIAgent: def __init__(self) -> None: @@ -111,7 +111,7 @@ class AIAgent: self.agent_id:str = None self.template_id:str = None self.fullname:str = None - self.powerby = None + self.powerby = None self.enable = True self.enable_kb = False self.enable_timestamp = False @@ -124,7 +124,7 @@ class AIAgent: self.owner_env : Environment = None self.owenr_bus = None self.enable_function_list = [] - + @classmethod def create_from_templete(cls,templete:AIAgentTemplete, fullname:str): # Agent just inherit from templete on craete,if template changed,agent will not change @@ -137,7 +137,7 @@ class AIAgent: result_agent.powerby = templete.author result_agent.prompt = templete.prompt return result_agent - + def load_from_config(self,config:dict) -> bool: if config.get("instance_id") is None: logger.error("agent instance_id is None!") @@ -188,7 +188,7 @@ class AIAgent: if llm_result_str == "ignore": r.state = "ignore" return r - + lines = llm_result_str.splitlines() is_need_wait = False @@ -205,7 +205,7 @@ class AIAgent: r.send_msgs.append(new_msg) is_need_wait = True - + case "post_msg":# postmsg($target_id,$msg_content) if len(func_args) != 1: logger.error(f"parse postmsg failed! {func_call}") @@ -215,22 +215,22 @@ class AIAgent: msg_content = func_item.body new_msg.set(self.agent_id,target_id,msg_content) r.post_msgs.append(new_msg) - + case "call":# call($func_name,$args_str) r.calls.append(func_item) is_need_wait = True return True case "post_call": # post_call($func_name,$args_str) r.post_calls.append(func_item) - return True - + return True + current_func : FunctionItem = None for line in lines: if line.startswith("##/"): if current_func: if check_args(current_func) is False: r.resp += current_func.dumps() - + func_name,func_args = AgentMsg.parse_function_call(line[3:]) current_func = FunctionItem(func_name,func_args) else: @@ -238,11 +238,11 @@ class AIAgent: current_func.append_body(line + "\n") else: r.resp += line + "\n" - + if current_func: if check_args(current_func) is False: r.resp += current_func.dumps() - + if len(r.send_msgs) > 0 or len(r.calls) > 0: r.state = "waiting" else: @@ -279,21 +279,22 @@ class AIAgent: def _get_inner_functions(self) -> dict: if self.owner_env is None: return None - + all_inner_function = self.owner_env.get_all_ai_functions() if all_inner_function is None: return None - + result_func = [] result_len = 0 for inner_func in all_inner_function: func_name = inner_func.get_name() - if self.enable_function_list: + if self.enable_function_list is not None: if len(self.enable_function_list) > 0: if func_name not in self.enable_function_list: logger.debug(f"ageint {self.agent_id} ignore inner func:{func_name}") continue - + else: + continue this_func = {} this_func["name"] = func_name this_func["description"] = inner_func.get_description() @@ -313,28 +314,29 @@ class AIAgent: func_node : AIFunction = self.owner_env.get_ai_function(func_name) if func_node is None: return "execute failed,function not found" - + ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target) try: result_str:str = await func_node.execute(**arguments) except Exception as e: - result_str = "call error:" + str(e) + result_str = "call error:" + str(e) logger.error(f"llm execute inner func:{func_name} error:{e}") - + + logger.info("llm execute inner func result:" + result_str) inner_functions,inner_function_len = self._get_inner_functions() prompt.messages.append({"role":"function","content":result_str,"name":func_name}) 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() org_msg.inner_call_chain.append(ineternal_call_record) if stack_limit > 0: inner_func_call_node = task_result.result_message.get("function_call") - + if inner_func_call_node: - return await self._execute_func(inner_func_call_node,prompt,org_msg,stack_limit-1) + return await self._execute_func(inner_func_call_node,prompt,org_msg,stack_limit-1) else: return task_result.result_str @@ -344,7 +346,7 @@ class AIAgent: def _format_msg_by_env_value(self,prompt:AgentPrompt): if self.owner_env is None: return - + for msg in prompt.messages: old_content = msg.get("content") msg["content"] = old_content.format_map(self.owner_env) @@ -380,7 +382,7 @@ class AIAgent: system_prompt_len = prompt.get_prompt_token_len() input_len = len(msg.body) - + history_prmpt,history_token_len = await self._get_prompt_from_session(chatsession,system_prompt_len + function_token_len,input_len) prompt.append(history_prmpt) # chat context @@ -397,7 +399,7 @@ class AIAgent: if inner_func_call_node: #TODO to save more token ,can i use msg_prompt? final_result = await self._execute_func(inner_func_call_node,prompt,msg) - + llm_result : LLMResult = self._get_llm_result_type(final_result) is_ignore = False result_prompt_str = "" @@ -415,21 +417,21 @@ class AIAgent: 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_resp_msg(final_result) chatsession.append(msg) chatsession.append(resp_msg) - + return resp_msg - + return None - + def get_id(self) -> str: return self.agent_id - + def get_fullname(self) -> str: return self.fullname @@ -438,14 +440,14 @@ class AIAgent: def get_llm_model_name(self) -> str: return self.llm_model_name - + def get_max_token_size(self) -> int: return self.max_token_size - + 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 - messages = chatsession.read_history() # read + messages = chatsession.read_history() # read result_token_len = 0 result_prompt = AgentPrompt() read_history_msg = 0 diff --git a/src/aios_kernel/bus.py b/src/aios_kernel/bus.py index 29fc186..16ede81 100644 --- a/src/aios_kernel/bus.py +++ b/src/aios_kernel/bus.py @@ -19,14 +19,14 @@ class AIBusHandler: async def handle_message(self,msg:AgentMsg) -> Any: if self.handler is None: return None - + 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) return resp_msg - + class AIBus: _instance = None @classmethod @@ -48,16 +48,16 @@ class AIBus: if msg.rely_msg_id is not None: handler.results[msg.rely_msg_id] = msg return None - + handler.queue.put_nowait(msg) self.start_process(target_id) return True - + if use_unhandle: if self.unhandle_handler is not None: if await self.unhandle_handler(self,msg): return await self.post_message(msg,False) - + logger.warn(f"post message to {msg.target} failed!,target not found") return False @@ -71,7 +71,7 @@ class AIBus: 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) if post_result is False: return None @@ -84,37 +84,39 @@ class AIBus: msg.status = AgentMsgStatus.RESPONSED del sender_handler.results[msg.msg_id] return resp - + await asyncio.sleep(0.2) retry_times += 1 - if retry_times > 5*120: # default timeout is 120 sec + if retry_times > 5*240: # default timeout is 240 sec msg.status = AgentMsgStatus.ERROR return None - + return None - + def register_unhandle_message_handler(self,handler:Any) -> Queue: self.unhandle_handler = handler # means sub - def register_message_handler(self,handler_name:str,handler:Any) -> Queue: - handler_node = AIBusHandler(handler,self) + def register_message_handler(self,handler_name:str,handler:Any) -> Queue: + handler_node = AIBusHandler(handler,self) self.handlers[handler_name] = handler_node return handler_node.queue - + async def process_queue(self, handler:AIBusHandler): while True: # Wait for a message message = await handler.queue.get() - #try: + try: # Try to handle the message - await handler.handle_message(message) - #except Exception as e: + await handler.handle_message(message) + except Exception as e: # If an error occurs, put the message back into the queue - # logger.error(f"handle message {message.msg_id} failed! {e}") + logger.error(f"handle message {message.msg_id} failed! {e}") + logger.exception(e) + raise e #self.queues[name].put_nowait(message) - + return def start_process(self,target_name): @@ -122,12 +124,12 @@ class AIBus: if handler is None: logger.error(f"handler {target_name} not found!") return - + if handler.handler is None: return if handler.working_task is not None: logger.warn(f"handler {target_name} is already working!") return - - handler.working_task = asyncio.create_task(self.process_queue(handler)) \ No newline at end of file + + handler.working_task = asyncio.create_task(self.process_queue(handler)) diff --git a/src/aios_kernel/compute_kernel.py b/src/aios_kernel/compute_kernel.py index 5859efe..fffc8ac 100644 --- a/src/aios_kernel/compute_kernel.py +++ b/src/aios_kernel/compute_kernel.py @@ -118,7 +118,7 @@ class ComputeKernel: if task_req.state == ComputeTaskState.ERROR: break - if check_times >= 20: + if check_times >= 120: task_req.state = ComputeTaskState.ERROR break @@ -129,7 +129,7 @@ class ComputeKernel: if task_req.state == ComputeTaskState.DONE: return task_req.result - return "error!" + raise Exception("error!") def text_embedding(self,input:str,model_name:Optional[str] = None): diff --git a/src/aios_kernel/open_ai_node.py b/src/aios_kernel/open_ai_node.py index 475fa22..e610be7 100644 --- a/src/aios_kernel/open_ai_node.py +++ b/src/aios_kernel/open_ai_node.py @@ -20,7 +20,7 @@ class OpenAI_ComputeNode(ComputeNode): if cls._instance is None: cls._instance = OpenAI_ComputeNode() return cls._instance - + @classmethod def declare_user_config(cls): if os.getenv("OPENAI_API_KEY_") is None: @@ -46,7 +46,7 @@ class OpenAI_ComputeNode(ComputeNode): if self.openai_api_key is None: logger.error("openai_api_key is None!") return False - + openai.api_key = self.openai_api_key self.start() return True @@ -68,7 +68,7 @@ class OpenAI_ComputeNode(ComputeNode): resp = openai.Embedding.create(model=model_name, input=input) - + # resp = { # "object": "list", # "data": [ @@ -86,7 +86,7 @@ class OpenAI_ComputeNode(ComputeNode): logger.info(f"openai response: {resp}") - result = ComputeTaskResult() + result = ComputeTaskResult() result.set_from_task(task) result.worker_id = self.node_id result.result = resp["data"][0]["embedding"] @@ -100,23 +100,23 @@ class OpenAI_ComputeNode(ComputeNode): if max_token_size is None: max_token_size = 4000 - result_token = int(max_token_size * 0.4) - - logger.info(f"call openai {mode_name} prompts: {prompts}") + result_token = max_token_size if llm_inner_functions is None: + logger.info(f"call openai {mode_name} prompts: {prompts}") resp = openai.ChatCompletion.create(model=mode_name, messages=prompts, max_tokens=result_token, temperature=0.7) else: + logger.info(f"call openai {mode_name} prompts: {prompts} functions: {json.dumps(llm_inner_functions)}") resp = openai.ChatCompletion.create(model=mode_name, messages=prompts, functions=llm_inner_functions, max_tokens=result_token, temperature=0.7) # TODO: add temperature to task params? - + logger.info(f"openai response: {json.dumps(resp, indent=4)}") result = ComputeTaskResult() @@ -139,6 +139,7 @@ class OpenAI_ComputeNode(ComputeNode): result.result_message = resp["choices"][0]["message"] if token_usage: result.result_refers["token_usage"] = token_usage + logger.info(f"openai success response: {result.result_str}") return result case _: task.state = ComputeTaskState.ERROR @@ -148,7 +149,7 @@ class OpenAI_ComputeNode(ComputeNode): if self.is_start is True: return self.is_start = True - + async def _run_task_loop(): while True: task = await self.task_queue.get() @@ -171,13 +172,13 @@ class OpenAI_ComputeNode(ComputeNode): def is_support(self, task: ComputeTask) -> bool: - if task.task_type == ComputeTaskType.LLM_COMPLETION: + if task.task_type == ComputeTaskType.LLM_COMPLETION: if not task.params["model_name"]: return True model_name : str = task.params["model_name"] if model_name.startswith("gpt-"): return True - + if task.task_type == ComputeTaskType.TEXT_EMBEDDING: if task.params["model_name"] == "text-embedding-ada-002": return True diff --git a/src/aios_kernel/text_to_speech_function.py b/src/aios_kernel/text_to_speech_function.py index 3237974..2f7bff0 100644 --- a/src/aios_kernel/text_to_speech_function.py +++ b/src/aios_kernel/text_to_speech_function.py @@ -83,9 +83,9 @@ class TextToSpeechFunction(AIFunction): continue if audio is not None: - path = os.path.join(os.curdir, "{}.mp3".format(random.sample('zyxwvutsrqponmlkjihgfedcba', 10))) + path = os.path.join(os.path.realpath(os.curdir), "{}.mp3".format(''.join(random.sample('zyxwvutsrqponmlkjihgfedcba', 10)))) audio.export(path, format="mp3") - return "complete.file path:{}".format(path) + return "已经生成音频文件, 文件路径为{}".format(path) else: return "failed" diff --git a/src/aios_kernel/workflow.py b/src/aios_kernel/workflow.py index 97893b6..00bb255 100644 --- a/src/aios_kernel/workflow.py +++ b/src/aios_kernel/workflow.py @@ -2,7 +2,7 @@ import logging import asyncio import json import os -import time +import time from asyncio import Queue from typing import Optional,Tuple,List from abc import ABC, abstractmethod @@ -32,7 +32,7 @@ class MessageFilter: # TODO: add more filter return None - + def load_from_config(self,config:dict) -> bool: self.filters = config return True @@ -68,8 +68,8 @@ class Workflow: def load_from_config(self,config:dict) -> bool: if config is None: return False - - if config.get("name") is None: + + if config.get("name") is None: logger.error("workflow config must have name") return False self.workflow_name = config.get("name") @@ -84,7 +84,7 @@ class Workflow: if self.rule_prompt.load_from_config(config.get("prompt")) is False: logger.error("Workflow load prompt failed") return False - + if config.get("roles") is None: logger.error("workflow config must have roles") return False @@ -106,13 +106,13 @@ class Workflow: self.env_db_file = self.owner_workflow.env_db_file self.workflow_env = WorkflowEnvironment(self.workflow_id,self.env_db_file) - env_ndoe = config.get("enviroment") + env_ndoe = config.get("enviroment") if env_ndoe is not None: if self._load_env_from_config(env_ndoe) is False: logger.error("Workflow load env failed") return False - connected_env_ndoe = config.get("connected_env") + connected_env_ndoe = config.get("connected_env") if connected_env_ndoe is not None: for _node in connected_env_ndoe: env_id = _node.get("env_id") @@ -124,13 +124,13 @@ class Workflow: logger.error(f"Workflow load connected_env failed, env {env_id} not found!") return False self.connect_to_environment(remote_env,_node.get("event2msg")) - + sub_workflows = config.get("sub_workflows") if sub_workflows is not None: if self._load_sub_workflows(sub_workflows) is False: logger.error("Workflow load sub workflows failed") return False - + return True def _load_env_from_config(self,config:dict) -> bool: @@ -147,7 +147,7 @@ class Workflow: return False self.sub_workflows[k] = sub_workflow return True - + def _parse_msg_target(self,s:str)->list[str]: return s.split(".") @@ -170,12 +170,12 @@ class Workflow: if current_workflow is None: logger.error(f"sub workflow {inner_obj_id[i]} not found!") return None - + i += 1 - + logger.error(f"{msg.target} not found! forword message failed!") return None - + def get_workflow_id_from_target(self,target:str) -> str: target_list = target.split(".") if len(target_list) == 0: @@ -203,11 +203,11 @@ class Workflow: #1. workflow start process message final_result = None - + # this is workflow's group_chat session session_topic = msg.sender + "#" + msg.topic chatsesssion = AIChatSession.get_session(self.workflow_id,session_topic,self.db_file) - + #2. find role by msg.mentions or workflow's selector logic if msg.mentions is not None: if not self.workflow_id in msg.mentions: @@ -219,20 +219,20 @@ class Workflow: this_role = self.role_group.get(mention) if this_role is not None: return await self.role_process_msg(msg,this_role,chatsesssion) - + if self.input_filter is not None: select_role_id = self.input_filter.select(msg) - if select_role_id is not None: + if select_role_id is not None: select_role = self.role_group.get(select_role_id) if select_role is None: logger.error(f"input_filter return invalid role id:{select_role_id}, role not found in role_group") return None - + return await self.role_process_msg(msg,select_role,chatsesssion) else: logger.error(f"input_filter return None for :{msg.body}") return None - + logger.error(f"{self.workflow_id}:no role can process this msg:{msg.body}") return final_result @@ -245,7 +245,7 @@ class Workflow: if llm_result_str == "ignore": r.state = "ignore" return r - + lines = llm_result_str.splitlines() is_need_wait = False @@ -262,7 +262,7 @@ class Workflow: r.send_msgs.append(new_msg) is_need_wait = True - + case "post_msg":# postmsg($target_id,$msg_content) if len(func_args) != 1: logger.error(f"parse postmsg failed! {func_call}") @@ -272,22 +272,22 @@ class Workflow: msg_content = func_item.body new_msg.set("_",target_id,msg_content) r.post_msgs.append(new_msg) - + case "call":# call($func_name,$args_str) r.calls.append(func_item) is_need_wait = True return True case "post_call": # post_call($func_name,$args_str) r.post_calls.append(func_item) - return True - + return True + current_func : FunctionItem = None for line in lines: if line.startswith("##/"): if current_func: if check_args(current_func) is False: r.resp += current_func.dumps() - + func_name,func_args = AgentMsg.parse_function_call(line[3:]) current_func = FunctionItem(func_name,func_args) else: @@ -295,7 +295,7 @@ class Workflow: current_func.append_body(line + "\n") else: r.resp += line + "\n" - + if current_func: if check_args(current_func) is False: r.resp += current_func.dumps() @@ -309,14 +309,14 @@ class Workflow: async def role_post_msg(self,msg:AgentMsg,the_role:AIRole,workflow_chat_session:AIChatSession): msg.sender = the_role.get_role_id() - + target_role = self.role_group.get(msg.target) if target_role: msg.target = target_role.get_role_id() logger.info(f"{msg.sender} post message {msg.msg_id} to inner role: {msg.target}") asyncio.create_task(self.role_process_msg(msg,target_role,workflow_chat_session)) return - + target_workflow = self.sub_workflows.get(msg.target) if target_workflow: msg.target = target_workflow.workflow_id @@ -341,7 +341,7 @@ class Workflow: # msg.target = target_workflow.workflow_id logger.info(f"{msg.sender} send message {msg.msg_id} to sub workflow: {msg.target}") return await target_workflow._process_msg(msg) - + logger.info(f"{msg.sender} post message {msg.msg_id} to AIBus: {msg.target}") return await self.get_bus().send_message(msg) @@ -352,8 +352,8 @@ class Workflow: func_node : AIFunction = self.workflow_env.get_ai_function(func_item.name) if func_node is None: return "execute failed,function not found" - - result_str:str = await func_node.execute(**arguments) + + result_str:str = await func_node.execute(**arguments) return result_str async def role_post_call(self,func_item:FunctionItem,the_role:AIRole): @@ -363,7 +363,7 @@ class Workflow: def _format_msg_by_env_value(self,prompt:AgentPrompt): if self.workflow_env is None: return - + for msg in prompt.messages: old_content = msg.get("content") msg["content"] = old_content.format_map(self.workflow_env) @@ -372,15 +372,17 @@ class Workflow: all_inner_function = self.workflow_env.get_all_ai_functions() if all_inner_function is None: return None - + result_func = [] for inner_func in all_inner_function: func_name = inner_func.get_name() - if the_role.enable_function_list: + if the_role.enable_function_list is not None: if len(the_role.enable_function_list) > 0: if func_name not in the_role.enable_function_list: - logger.debug(f"ageint {self.agent_id} ignore inner func:{func_name}") + logger.debug(f"agent {self.agent_id} ignore inner func:{func_name}") continue + else: + continue this_func = {} this_func["name"] = func_name this_func["description"] = inner_func.get_description() @@ -399,17 +401,17 @@ class Workflow: func_node : AIFunction = self.workflow_env.get_ai_function(func_name) if func_node is None: return "execute failed,function not found" - + ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target) result_str:str = await func_node.execute(**arguments) - - inner_functions = self._get_inner_functions() + + inner_functions = self._get_inner_functions(the_role) prompt.messages.append({"role":"function","content":result_str,"name":func_name}) 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) - + ineternal_call_record.result_str = task_result.result_str ineternal_call_record.done_time = time.time() org_msg.inner_call_chain.append(ineternal_call_record) @@ -419,13 +421,13 @@ class Workflow: return await self._role_execute_func(the_role,inner_func_call_node,prompt,org_msg,stack_limit-1) else: return task_result.result_str - + def _is_in_same_workflow(self,msg) -> bool: pass - async def role_process_msg(self,msg:AgentMsg,the_role:AIRole,workflow_chat_session:AIChatSession): + async def role_process_msg(self,msg:AgentMsg,the_role:AIRole,workflow_chat_session:AIChatSession): msg.target = the_role.get_role_id() - + prompt = AgentPrompt() prompt.append(the_role.agent.prompt) @@ -433,7 +435,7 @@ class Workflow: prompt.append(the_role.get_prompt()) # prompt.append(self._get_function_prompt(the_role.get_name())) # prompt.append(self._get_knowlege_prompt(the_role.get_name())) - + #support group chat, user content include sender name! prompt.append(await self._get_prompt_from_session(workflow_chat_session)) @@ -443,15 +445,15 @@ class Workflow: self._format_msg_by_env_value(prompt) inner_functions = self._get_inner_functions(the_role) - + async def _do_process_msg(): #TODO: send msg to agent might be better? 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}") - + inner_func_call_node = task_result.result_message.get("function_call") - + if inner_func_call_node: #TODO to save more token ,can i use msg_prompt? result_str = await self._role_execute_func(the_role,inner_func_call_node,prompt,msg) @@ -461,7 +463,7 @@ class Workflow: postmsg.prev_msg_id = msg.get_msg_id() # might be craete a new msg.topic for this postmsg postmsg.topic = msg.topic - + await self.role_post_msg(postmsg,the_role,workflow_chat_session) if not self._is_in_same_workflow(postmsg): role_sesion = AIChatSession.get_session(the_role.get_role_id(),f"{postmsg.target}#{msg.topic}",self.db_file) @@ -469,14 +471,14 @@ class Workflow: else: # message will be saved in role.process_message pass - - + + for post_call in result.post_calls: action_msg = msg.create_action_msg(post_call[0],post_call[1],the_role.get_role_id()) workflow_chat_session.append(action_msg) await self.role_post_call(post_call,the_role) #save post_call - + result_prompt_str = "" match result.state: case "ignore": @@ -506,11 +508,11 @@ class Workflow: else: # message will be saved in role.process_message pass - + for call in result.calls: action_msg = msg.create_action_msg(call[0],call[1],call_result,the_role.get_role_id) call_result = await self.role_call(call,the_role) - + if call_result is not None: result_prompt_str += f"\ncall {call[0]} result is :{call_result}" #save action @@ -522,7 +524,7 @@ class Workflow: prompt.append(result_prompt) r = await _do_process_msg() return r - + return await _do_process_msg() async def _get_prompt_from_session(self,chatsession:AIChatSession) -> AgentPrompt: @@ -533,9 +535,9 @@ class Workflow: result_prompt.messages.append({"role":"assistant","content":msg.body}) else: result_prompt.messages.append({"role":"user","content":f"{msg.body}"}) - + return result_prompt - + def _get_knowlege_prompt(self,role_name:str) -> AgentPrompt: pass @@ -557,7 +559,7 @@ class Workflow: # if k == "role": # continue # else: - # + # # def _env_msg_handler(env_event:EnvironmentEvent) -> None: # the_msg:AgentMsg= self._env_event_to_msg(env_event) # self.role_post_msg From b8345eccaf8e06b0d13f715feea3d0fe4c1260c8 Mon Sep 17 00:00:00 2001 From: wugren Date: Fri, 22 Sep 2023 12:22:36 +0800 Subject: [PATCH 3/5] Modify story_maker prompt word --- rootfs/agents/studio_director/agent.toml | 3 ++- rootfs/workflows/story_maker/workflow.toml | 3 ++- src/aios_kernel/text_to_speech_function.py | 6 +++--- 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/rootfs/agents/studio_director/agent.toml b/rootfs/agents/studio_director/agent.toml index 65a7c07..5a5d297 100644 --- a/rootfs/agents/studio_director/agent.toml +++ b/rootfs/agents/studio_director/agent.toml @@ -1,7 +1,8 @@ instance_id = "studio_director" fullname = "tracy wang" +llm_model_name = "gpt-3.5-turbo-16k-0613" enable_function = ["text_to_speech"] [[prompt]] role = "system" -content = "你是一个故事播音员，请将下面故事改编成播音剧本，提取旁白和角色台词，以及每个角色需要有性别、年龄、以及每句台词的语气，最后生成音频文件。" +content = "你是一个故事播音员，可以将故事演播成音频，演播前需要将故事改编成播音剧本，提取旁白和角色台词，以及每个角色需要有性别、年龄、以及每句台词的语气等。如果生成了音频文件则告知你的用户。" diff --git a/rootfs/workflows/story_maker/workflow.toml b/rootfs/workflows/story_maker/workflow.toml index 540665f..6ae5e55 100644 --- a/rootfs/workflows/story_maker/workflow.toml +++ b/rootfs/workflows/story_maker/workflow.toml @@ -16,7 +16,8 @@ content=""" 你是一个语音故事制作总导演,与客户对接并向团队下达指令。你的团队分为下面几个成员:writer,studio_director。一个故事制作分成两个阶段：让writer写出故事，再交由studio_director演播故事生成音频文件。你的基本工作模式是： 1. 收到客户的明确的指令后,让writer写出故事 2. 将writer写出的故事交给studio_director演播 -3. 当你决定要和成员通信时，请使用下面形式输出需要通信的消息 +3. 获得音频文件之后，将音频文件的存放路径交给客户 +4. 当你决定要和成员通信时，请使用下面形式输出需要通信的消息 ``` ##/send_msg 成员名称 内容 diff --git a/src/aios_kernel/text_to_speech_function.py b/src/aios_kernel/text_to_speech_function.py index 2f7bff0..90dfd49 100644 --- a/src/aios_kernel/text_to_speech_function.py +++ b/src/aios_kernel/text_to_speech_function.py @@ -14,7 +14,7 @@ logger = logging.getLogger(__name__) class TextToSpeechFunction(AIFunction): def __init__(self): self.func_id = "text_to_speech" - self.description = "根据输入的剧本生成音频数据" + self.description = "根据输入的剧本生成音频文件，成功时会返回音频文件路径" def get_name(self) -> str: return self.func_id @@ -85,9 +85,9 @@ class TextToSpeechFunction(AIFunction): if audio is not None: path = os.path.join(os.path.realpath(os.curdir), "{}.mp3".format(''.join(random.sample('zyxwvutsrqponmlkjihgfedcba', 10)))) audio.export(path, format="mp3") - return "已经生成音频文件, 文件路径为{}".format(path) + return "exec text_to_speech OK，speech file store at {}".format(path) else: - return "failed" + return "exec text_to_speech failed" def is_local(self) -> bool: return True From b195e6a6a454f8eb0e005026d5960004282f9693 Mon Sep 17 00:00:00 2001 From: wugren Date: Fri, 22 Sep 2023 13:38:44 +0800 Subject: [PATCH 4/5] add google application credentials --- src/aios_kernel/google_text_to_speech_node.py | 21 ++++++++++-- src/service/aios_shell/aios_shell.py | 32 ++++++++++++------- 2 files changed, 39 insertions(+), 14 deletions(-) diff --git a/src/aios_kernel/google_text_to_speech_node.py b/src/aios_kernel/google_text_to_speech_node.py index b05f276..9bee115 100644 --- a/src/aios_kernel/google_text_to_speech_node.py +++ b/src/aios_kernel/google_text_to_speech_node.py @@ -3,9 +3,11 @@ import os import asyncio from asyncio import Queue import logging +from typing import Optional from google.cloud import texttospeech +from .storage import AIStorage from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType from .compute_node import ComputeNode @@ -31,8 +33,7 @@ class GoogleTextToSpeechNode(ComputeNode): super().__init__() self.node_id = "google_text_to_speech_node" self.task_queue = Queue() - - self.client = texttospeech.TextToSpeechClient() + self.client: Optional[texttospeech.TextToSpeechClient] = None self.language_list = { "cnm-CN": { @@ -86,6 +87,14 @@ class GoogleTextToSpeechNode(ComputeNode): } self.start() + def init(self): + user_config = AIStorage.get_instance().get_user_config() + google_application_credentials = user_config.get_value("google_application_credentials") + if google_application_credentials is None: + raise Exception("google_application_credentials is None!") + os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = google_application_credentials + self.client = texttospeech.TextToSpeechClient() + def start(self): async def _run_task_loop(): while True: @@ -161,3 +170,11 @@ class GoogleTextToSpeechNode(ComputeNode): def is_local(self) -> bool: return False + + def declare_user_config(self): + if os.getenv("GOOGLE_APPLICATION_CREDENTIALS") is None: + user_config = AIStorage.get_instance().get_user_config() + user_config.add_user_config("google_application_credentials", + "google application credentials, please visit:https://cloud.google.com/text-to-speech/docs/before-you-begin", + False, + None) diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index f3df06a..7e0a5ca 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -59,6 +59,9 @@ class AIOS_Shell: user_config.add_user_config("shell.current","last opened target and topic",True,"default@Jarvis") proxy.declare_user_config() + google_text_to_speech = GoogleTextToSpeechNode.get_instance() + google_text_to_speech.declare_user_config() + async def _handle_no_target_msg(self,bus:AIBus,msg:AgentMsg) -> bool: target_id = msg.target.split(".")[0] @@ -97,8 +100,13 @@ class AIOS_Shell: return False ComputeKernel.get_instance().add_compute_node(open_ai_node) - google_text_to_speech_node = GoogleTextToSpeechNode.get_instance() - ComputeKernel.get_instance().add_compute_node(google_text_to_speech_node) + try: + google_text_to_speech_node = GoogleTextToSpeechNode.get_instance() + google_text_to_speech_node.init() + ComputeKernel.get_instance().add_compute_node(google_text_to_speech_node) + except Exception as e: + logger.error(f"google text to speech node initial failed! {e}") + return False llama_ai_node = LocalLlama_ComputeNode() await llama_ai_node.start() @@ -116,7 +124,7 @@ class AIOS_Shell: contact_config_path =os.path.abspath(f"{user_data_dir}/contacts.toml") cm = ContactManager.get_instance(contact_config_path) cm.load_data() - + tunnels_config_path = os.path.abspath(f"{user_data_dir}/etc/tunnels.cfg.toml") tunnel_config = None try: @@ -125,10 +133,10 @@ class AIOS_Shell: await AgentTunnel.load_all_tunnels_from_config(tunnel_config) except Exception as e: logger.warning(f"load tunnels config from {tunnels_config_path} failed!") - + KnowledgePipline.get_instance().initial() - return True - + return True + def get_version(self) -> str: return "0.5.1" @@ -169,10 +177,10 @@ class AIOS_Shell: user_input = await try_get_input(f"{key} : {item.desc}") if user_input is None: return None - - tunnel_config[key] = user_input - return tunnel_config + tunnel_config[key] = user_input + + return tunnel_config async def append_tunnel_config(self,tunnel_config): user_data_dir = AIStorage.get_instance().get_myai_dir() @@ -235,7 +243,7 @@ class AIOS_Shell: prompt.messages.append({"role": "user", "content":" ".join(args[1:])}) result = await KnowledgeBase().query_prompt(prompt) print_formatted_text(result.as_str()) - + async def call_func(self,func_name, args): match func_name: case 'send': @@ -252,7 +260,7 @@ class AIOS_Shell: key = args[0] config_item = AIStorage.get_instance().get_user_config().get_config_item(key) old_value = AIStorage.get_instance().get_user_config().get_value(key) - + if config_item is not None: value = await session.prompt_async(f"{key} : {config_item.desc} \nCurrent : {old_value}\nPlease input new value:",style=shell_style) AIStorage.get_instance().get_user_config().set_value(key,value) @@ -459,7 +467,7 @@ async def main(): '/history $num $offset', '/login $username', '/connect $target', - '/knowledge add email | dir', + '/knowledge add email | dir', '/knowledge journal [$topn]', '/knowledge query $query' '/set_config $key', From 001482ea23499a0dfaea6807eecc7fc602466fc4 Mon Sep 17 00:00:00 2001 From: wugren Date: Fri, 22 Sep 2023 13:50:43 +0800 Subject: [PATCH 5/5] Modify agent name --- .../agents/{studio_director => speecher}/agent.toml | 2 +- rootfs/workflows/story_maker/workflow.toml | 12 ++++++------ 2 files changed, 7 insertions(+), 7 deletions(-) rename rootfs/agents/{studio_director => speecher}/agent.toml (92%) diff --git a/rootfs/agents/studio_director/agent.toml b/rootfs/agents/speecher/agent.toml similarity index 92% rename from rootfs/agents/studio_director/agent.toml rename to rootfs/agents/speecher/agent.toml index 5a5d297..dea7f70 100644 --- a/rootfs/agents/studio_director/agent.toml +++ b/rootfs/agents/speecher/agent.toml @@ -1,4 +1,4 @@ -instance_id = "studio_director" +instance_id = "speecher" fullname = "tracy wang" llm_model_name = "gpt-3.5-turbo-16k-0613" enable_function = ["text_to_speech"] diff --git a/rootfs/workflows/story_maker/workflow.toml b/rootfs/workflows/story_maker/workflow.toml index 6ae5e55..fd3876a 100644 --- a/rootfs/workflows/story_maker/workflow.toml +++ b/rootfs/workflows/story_maker/workflow.toml @@ -13,9 +13,9 @@ enable_function = [] [[roles.manager.prompt]] role="system" content=""" -你是一个语音故事制作总导演,与客户对接并向团队下达指令。你的团队分为下面几个成员:writer,studio_director。一个故事制作分成两个阶段：让writer写出故事，再交由studio_director演播故事生成音频文件。你的基本工作模式是： +你是一个语音故事制作总导演,与客户对接并向团队下达指令。你的团队分为下面几个成员:writer,speecher。一个故事制作分成两个阶段：让writer写出故事，再交由speecher演播故事生成音频文件。你的基本工作模式是： 1. 收到客户的明确的指令后,让writer写出故事 -2. 将writer写出的故事交给studio_director演播 +2. 将writer写出的故事交给speecher演播 3. 获得音频文件之后，将音频文件的存放路径交给客户 4. 当你决定要和成员通信时，请使用下面形式输出需要通信的消息 ``` @@ -33,10 +33,10 @@ enable_function = [] role="system" content="" -[roles.studio_director] -name = "studio_director" -agent = "studio_director" +[roles.speecher] +name = "speecher" +agent = "speecher" enable_function = ["text_to_speech"] -[[roles.studio_director.prompt]] +[[roles.speecher.prompt]] role="system" content=""