diff --git a/rootfs/agents/AgentAssistant/environment.py b/rootfs/agents/AgentAssistant/environment.py index 35fa35b..e5ce961 100644 --- a/rootfs/agents/AgentAssistant/environment.py +++ b/rootfs/agents/AgentAssistant/environment.py @@ -3,9 +3,10 @@ from typing import Optional import toml -from aios_kernel import Environment, SimpleAIFunction import os +from aios.agent.ai_function import SimpleAIFunction +from aios.environment.environment import Environment local_path = os.path.split(os.path.realpath(__file__))[0] diff --git a/rootfs/agents/DBQueryer/environment.py b/rootfs/agents/DBQueryer/environment.py index e733c45..8c8ec02 100644 --- a/rootfs/agents/DBQueryer/environment.py +++ b/rootfs/agents/DBQueryer/environment.py @@ -1,7 +1,7 @@ from typing import Optional -from aios_kernel import Environment -from aios_kernel.sql_database_function import GetTableInfosFunction, ExecuteSqlFunction +from aios.environment.environment import Environment +from aios.environment.sql_database_function import GetTableInfosFunction, ExecuteSqlFunction class DBQuerierEnvironment(Environment): diff --git a/rootfs/agents/TextSummary/agent.py b/rootfs/agents/TextSummary/agent.py new file mode 100644 index 0000000..537f520 --- /dev/null +++ b/rootfs/agents/TextSummary/agent.py @@ -0,0 +1,49 @@ +import copy + +from aios.agent.agent_base import CustomAIAgent, AgentPrompt +from aios.knowledge.data.writer import split_text +from aios.proto.agent_msg import AgentMsg, AgentMsgType +from aios.proto.compute_task import ComputeTaskResultCode + + +class TextSummaryAgent(CustomAIAgent): + def __init__(self): + super().__init__("TextSummary", "Text Summary", 128000) + + async def _process_msg(self, msg: AgentMsg, workspace=None) -> AgentMsg: + if msg.msg_type is not AgentMsgType.TYPE_MSG: + return AgentMsg.create_error_resp(msg, "only support msg type") + + if msg.body_mime is not None and msg.body_mime != "text/plain": + return AgentMsg.create_error_resp(msg, "only support text/plain mime type") + + chunks = split_text(msg.body, separators=["\n\n", "\n"], chunk_size=4000, chunk_overlap=200, length_function=len) + + prompt = AgentPrompt() + prompt.system_message = "Your job is to generate a summary based on the input." + if len(chunks) == 1: + prompt.append(AgentPrompt(chunks[0])) + resp = await self.do_llm_complection(prompt) + if resp.result_code != ComputeTaskResultCode.OK: + return msg.create_error_resp(resp.error_str) + return msg.create_resp_msg(resp.result_str) + + segments = [] + for i, chunk in enumerate(chunks): + seg_prompt = copy.deepcopy(prompt) + seg_prompt.append(AgentPrompt(chunk)) + resp = await self.do_llm_complection(seg_prompt) + if resp.result_code != ComputeTaskResultCode.OK: + return msg.create_error_resp(resp.error_str) + segments.append(resp.result_str) + + segments_str = "\n".join(segments) + prompt.append(AgentPrompt(f"以下文本分段之后的各段摘要，请合并生成一个完整摘要：\n{segments_str}")) + resp = await self.do_llm_complection(prompt) + if resp.result_code != ComputeTaskResultCode.OK: + return msg.create_error_resp(resp.error_str) + return msg.create_resp_msg(resp.result_str) + + +def init(): + return TextSummaryAgent() diff --git a/rootfs/agents/Vision/agent.toml b/rootfs/agents/Vision/agent.toml new file mode 100644 index 0000000..672c219 --- /dev/null +++ b/rootfs/agents/Vision/agent.toml @@ -0,0 +1,8 @@ +instance_id = "Vision" +fullname = "Vision" +llm_model_name = "gpt-4-1106-preview" + +[[prompt]] +role = "system" +content = """Your job is to analyze user input images and videos and respond based on user intent. +If the user requests a video and you receive key frames of the video, please reply to the user's question based on the key frame content.""" diff --git a/src/aios/__init__.py b/src/aios/__init__.py index 1a0c512..ce866cf 100644 --- a/src/aios/__init__.py +++ b/src/aios/__init__.py @@ -27,6 +27,7 @@ from .storage.storage import ResourceLocation,AIStorage,UserConfig,UserConfigIte from .net import * from .knowledge import * from .package_manager import * +from .utils import * -AIOS_Version = "0.5.2, build 2023-11-30" \ No newline at end of file +AIOS_Version = "0.5.2, build 2023-11-30" diff --git a/src/aios/agent/agent.py b/src/aios/agent/agent.py index ad8dcf8..2b97326 100644 --- a/src/aios/agent/agent.py +++ b/src/aios/agent/agent.py @@ -27,6 +27,7 @@ from ..environment.workspace_env import WorkspaceEnvironment from ..storage.storage import AIStorage from ..knowledge import * +from ..utils import video_utils, image_utils logger = logging.getLogger(__name__) @@ -423,11 +424,39 @@ class AIAgent(BaseAIAgent): async def _create_openai_thread(self) -> str: return None + def check_and_to_base64(self, image_path: str) -> str: + if image_utils.is_file(image_path): + return image_utils.to_base64(image_path, (1024, 1024)) + else: + return image_path + async def _process_msg(self,msg:AgentMsg,workspace = None) -> AgentMsg: msg_prompt = AgentPrompt() if msg.msg_type == AgentMsgType.TYPE_GROUPMSG: need_process = False - msg_prompt.messages = [{"role":"user","content":f"{msg.sender}:{msg.body}"}] + if msg.is_image_msg(): + image_prompt, images = msg.get_image_body() + if image_prompt is None: + content = [[{"type": "text", "text": f"{msg.sender}'s message"}]] + content.extend([{"type": "image_url", "image_url": {"url": self.check_and_to_base64(image)}} for image in images]) + msg_prompt.messages = [{"role": "user", "content": content}] + else: + content = [{"type": "text", "text": f"{msg.sender}:{image_prompt}"}] + content.extend([{"type": "image_url", "image_url": {"url": self.check_and_to_base64(image)}} for image in images]) + msg_prompt.messages = [{"role": "user", "content": content}] + elif msg.is_video_msg(): + video_prompt, video = msg.get_video_body() + frames = video_utils.extract_frames(video, (1024, 1024)) + if video_prompt is None: + content = [{"type": "text", "text": f"{msg.sender}'s message"}] + content.extend([{"type": "image_url", "image_url": {"url": frame}} for frame in frames]) + msg_prompt.messages = [{"role": "user", "content": content}] + else: + content = [{"type": "text", "text": f"{msg.sender}:{video_prompt}"}] + content.extend([{"type": "image_url", "image_url": {"url": frame}} for frame in frames]) + msg_prompt.messages = [{"role": "user", "content": content}] + else: + msg_prompt.messages = [{"role":"user","content":f"{msg.sender}:{msg.body}"}] session_topic = msg.target + "#" + msg.topic chatsession = AIChatSession.get_session(self.agent_id,session_topic,self.chat_db) @@ -441,7 +470,25 @@ class AIAgent(BaseAIAgent): resp_msg = msg.create_group_resp_msg(self.agent_id,"") return resp_msg else: - msg_prompt.messages = [{"role":"user","content":msg.body}] + if msg.is_image_msg(): + image_prompt, images = msg.get_image_body() + if image_prompt is None: + msg_prompt.messages = [{"role": "user", "content": [{"type": "image_url", "image_url": {"url": self.check_and_to_base64(image)}} for image in images]}] + else: + content = [{"type": "text", "text": image_prompt}] + content.extend([{"type": "image_url", "image_url": {"url": self.check_and_to_base64(image)}} for image in images]) + msg_prompt.messages = [{"role": "user", "content": content}] + elif msg.is_video_msg(): + video_prompt, video = msg.get_video_body() + frames = video_utils.extract_frames(video, (1024, 1024)) + if video_prompt is None: + msg_prompt.messages = [{"role": "user", "content": [{"type": "image_url", "image_url": {"url": frame}} for frame in frames]}] + else: + content = [{"type": "text", "text": video_prompt}] + content.extend([{"type": "image_url", "image_url": {"url": frame}} for frame in frames]) + msg_prompt.messages = [{"role": "user", "content": content}] + else: + msg_prompt.messages = [{"role":"user","content":msg.body}] session_topic = msg.get_sender() + "#" + msg.topic chatsession = AIChatSession.get_session(self.agent_id,session_topic,self.chat_db) if self.enable_thread: diff --git a/src/aios/agent/agent_base.py b/src/aios/agent/agent_base.py index 15c738e..4094c36 100644 --- a/src/aios/agent/agent_base.py +++ b/src/aios/agent/agent_base.py @@ -9,7 +9,7 @@ import time import re import shlex import json -from typing import List +from typing import List, Tuple from .ai_function import FunctionItem, AIFunction from ..proto.agent_msg import AgentMsg, AgentMsgType @@ -410,6 +410,10 @@ class BaseAIAgent(abc.ABC): def get_max_token_size(self) -> int: pass + @abstractmethod + async def _process_msg(self,msg:AgentMsg,workspace = None) -> AgentMsg: + pass + @classmethod def get_inner_functions(cls, env:Environment) -> (dict,int): if env is None: @@ -445,10 +449,29 @@ class BaseAIAgent(abc.ABC): #logger.debug(f"Agent {self.agent_id} do llm token static system:{system_prompt_len},function:{function_token_len},history:{history_token_len},input:{input_len}, totoal prompt:{system_prompt_len + function_token_len + history_token_len} ") if inner_functions is None and env is not None: inner_functions,_ = BaseAIAgent.get_inner_functions(env) + + model_name = self.get_llm_model_name() + if org_msg.is_video_msg() or org_msg.is_image_msg(): + if model_name.startswith("gpt-4"): + model_name = "gpt-4-vision-preview" if is_json_resp: - task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,resp_mode="json",mode_name=self.get_llm_model_name(),max_token=self.get_max_token_size(),inner_functions=inner_functions,timeout=None) + task_result: ComputeTaskResult = await (ComputeKernel.get_instance() + .do_llm_completion( + prompt, + resp_mode="json", + mode_name=model_name, + max_token=self.get_max_token_size(), + inner_functions=inner_functions, + timeout=None)) else: - task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,resp_mode="text",mode_name=self.get_llm_model_name(),max_token=self.get_max_token_size(),inner_functions=inner_functions,timeout=None) + task_result: ComputeTaskResult = await (ComputeKernel.get_instance() + .do_llm_completion( + prompt, + resp_mode="text", + mode_name=model_name, + max_token=self.get_max_token_size(), + inner_functions=inner_functions, + timeout=None)) if task_result.result_code != ComputeTaskResultCode.OK: logger.error(f"_do_llm_complection llm compute error:{task_result.error_str}") #error_resp = msg.create_error_resp(task_result.error_str) @@ -478,7 +501,6 @@ class BaseAIAgent(abc.ABC): stack_limit = 5 ) -> ComputeTaskResult: from ..frame.compute_kernel import ComputeKernel - arguments = None try: func_name = inner_func_call_node.get("name") diff --git a/src/aios/knowledge/data/writer.py b/src/aios/knowledge/data/writer.py index 0381bc6..6ebf3ac 100644 --- a/src/aios/knowledge/data/writer.py +++ b/src/aios/knowledge/data/writer.py @@ -19,10 +19,10 @@ def _join_docs(docs: List[str], separator: str) -> Optional[str]: return text def _merge_splits( - splits: Iterable[str], - separator: str, - chunk_size: int, - chunk_overlap: int, + splits: Iterable[str], + separator: str, + chunk_size: int, + chunk_overlap: int, length_function: Callable[[str], int] ) -> List[str]: # We now want to combine these smaller pieces into medium size @@ -86,11 +86,11 @@ def _split_text_with_regex( return [s for s in splits if s != ""] -def _split_text( - text: str, - separators: List[str], - chunk_size: int, - chunk_overlap: int, +def split_text( + text: str, + separators: List[str], + chunk_size: int, + chunk_overlap: int, length_function: Callable[[str], int] ) -> List[str]: @@ -127,7 +127,7 @@ def _split_text( if not new_separators: final_chunks.append(s) else: - other_info = _split_text(s, new_separators, chunk_size, chunk_overlap, length_function) + other_info = split_text(s, new_separators, chunk_size, chunk_overlap, length_function) final_chunks.extend(other_info) if _good_splits: merged_text = _merge_splits(_good_splits, _separator, chunk_size, chunk_overlap, length_function) @@ -153,7 +153,7 @@ class ChunkListWriter: chunk = file.read(chunk_size) if not chunk: break - + chunk_len = len(chunk) chunk_id = ChunkID.hash_data(chunk) chunk_list.append(chunk_id) @@ -176,14 +176,14 @@ class ChunkListWriter: file_hash = HashValue(hash_obj.digest()) # print(f"calc file hash: {file_path}, {file_hash}") - + return ChunkList(chunk_list, file_hash) def create_chunk_list_from_text( - self, - text: str, - chunk_size: int = 4000, - chunk_overlap: int = 200, + self, + text: str, + chunk_size: int = 4000, + chunk_overlap: int = 200, separators: str = ["\n\n", "\n", " ", ""] ) -> ChunkList: enc = tiktoken.encoding_for_model("gpt-3.5-turbo") @@ -196,8 +196,8 @@ class ChunkListWriter: disallowed_special="all", ) ) - - text_list = _split_text(text, separators, chunk_size, chunk_overlap, length_function) + + text_list = split_text(text, separators, chunk_size, chunk_overlap, length_function) chunk_list = [] hash_obj = hashlib.sha256() @@ -211,4 +211,4 @@ class ChunkListWriter: self.chunk_store.put_chunk(chunk_id, chunk_bytes) hash = HashValue(hash_obj.digest()) - return ChunkList(chunk_list, hash) \ No newline at end of file + return ChunkList(chunk_list, hash) diff --git a/src/aios/proto/agent_msg.py b/src/aios/proto/agent_msg.py index 3c1e2a0..c7209bf 100644 --- a/src/aios/proto/agent_msg.py +++ b/src/aios/proto/agent_msg.py @@ -1,7 +1,10 @@ +import json import logging +import shlex import uuid from enum import Enum import time +from typing import Tuple, List logger = logging.getLogger(__name__) @@ -35,8 +38,6 @@ class AgentMsgStatus(Enum): # 逻辑上的同一个Message在同一个session中看到的msgid相同 # 在不同的session中看到的msgid不同 - - class AgentMsg: def __init__(self,msg_type=AgentMsgType.TYPE_MSG) -> None: self.msg_id = "msg#" + uuid.uuid4().hex @@ -136,14 +137,79 @@ class AgentMsg: return resp_msg - def set(self,sender:str,target:str,body:str,topic:str=None) -> None: + def set(self,sender:str,target:str,body:str,topic:str=None,body_mime:str=None) -> None: self.sender = sender self.target = target self.body = body + self.body_mime = body_mime self.create_time = time.time() if topic: self.topic = topic + @staticmethod + def create_image_body(images: [str], prompt: str = None): + return json.dumps({"images": images, "prompt": prompt}) + + @staticmethod + def parse_image_body(image_body: str) -> Tuple[str, List[str]]: + body = json.loads(image_body) + return body.get("prompt"), body.get("images") + + @staticmethod + def create_video_body(video: str, prompt: str = None): + return json.dumps({"video": video, "prompt": prompt}) + + @staticmethod + def parse_video_body(video_body: str) -> Tuple[str, str]: + body = json.loads(video_body) + return body.get("prompt"), body.get("video") + + def set_image(self, sender: str, target: str, image_format: str, images: [str], prompt: str = None, topic: str = None): + self.sender = sender + self.target = target + self.create_time = time.time() + self.body_mime = f"image/{image_format}" + self.body = self.create_image_body(images, prompt) + if topic: + self.topic = topic + + def is_image_msg(self) -> bool: + if self.body_mime is None: + return False + if self.body_mime.startswith("image/"): + return True + return False + + def get_image_body(self) -> Tuple[str, List[str]]: + if self.body_mime is None: + return None + if self.body_mime.startswith("image/"): + return self.parse_image_body(self.body) + return None + + def set_video(self, sender: str, target: str, video_format: str, video: str, prompt: str = None, topic: str = None): + self.sender = sender + self.target = target + self.create_time = time.time() + self.body_mime = f"video/{video_format}" + self.body = self.create_video_body(video, prompt) + if topic: + self.topic = topic + + def get_video_body(self) -> Tuple[str, str]: + if self.body_mime is None: + return None + if self.body_mime.startswith("video/"): + return self.parse_video_body(self.body) + return None + + def is_video_msg(self) -> bool: + if self.body_mime is None: + return False + if self.body_mime.startswith("video/"): + return True + return False + def get_msg_id(self) -> str: return self.msg_id @@ -164,4 +230,4 @@ class AgentMsg: str_list = shlex.split(func_string) func_name = str_list[0] params = str_list[1:] - return func_name, params \ No newline at end of file + return func_name, params diff --git a/src/aios/utils/__init__.py b/src/aios/utils/__init__.py new file mode 100644 index 0000000..dbaab83 --- /dev/null +++ b/src/aios/utils/__init__.py @@ -0,0 +1,2 @@ +from . import image_utils +from . import video_utils diff --git a/src/aios/utils/image_utils.py b/src/aios/utils/image_utils.py new file mode 100644 index 0000000..3d07b13 --- /dev/null +++ b/src/aios/utils/image_utils.py @@ -0,0 +1,40 @@ +import base64 +import os.path +from typing import Tuple + +import cv2 + + +def to_base64(image_path: str, resize: Tuple[int, int] = None) -> str: + """Convert image to base64.""" + ext = os.path.splitext(image_path)[1][1:] + if resize is None: + with open(image_path, "rb") as image_file: + base64_image = base64.b64encode(image_file.read()).decode("utf-8") + return f"data:image/{ext};base64,{base64_image}" + else: + dest_width, dest_height = resize + img = cv2.imread(image_path) + width, height = img.shape[:2] + if width > dest_width or height > dest_height: + width_rate = dest_width / width + height_rate = dest_height / height + rate = min(width_rate, height_rate) + dest_width = int(width * rate) + dest_height = int(height * rate) + img = cv2.resize(img, (dest_width, dest_height), interpolation=cv2.INTER_AREA) + _, buf = cv2.imencode(f".{ext}", img) + base64_image = base64.b64encode(buf).decode("utf-8") + return f"data:image/{ext};base64,{base64_image}" + + +def is_file(image_path: str) -> bool: + return os.path.isfile(image_path) + + +def is_base64(image_path: str) -> bool: + return image_path.startswith("data:image/") + + +def is_url(image_path: str) -> bool: + return image_path.startswith("http://") or image_path.startswith("https://") diff --git a/src/aios/utils/video_utils.py b/src/aios/utils/video_utils.py new file mode 100644 index 0000000..dd687cf --- /dev/null +++ b/src/aios/utils/video_utils.py @@ -0,0 +1,122 @@ +import base64 +from typing import List, Tuple + +import cv2 +import numpy as np + + +def precess_image(image): + ''' + Graying and GaussianBlur + :param image: The image matrix,np.array + :return: The processed image matrix,np.array + ''' + gray_image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) + gray_image = cv2.GaussianBlur(gray_image, (3, 3), 0) + return gray_image + + +def abs_diff(pre_image, curr_image): + ''' + Calculate absolute difference between pre_image and curr_image + :param pre_image:The image in past frame,np.array + :param curr_image:The image in current frame,np.array + :return: + ''' + gray_pre_image = precess_image(pre_image) + gray_curr_image = precess_image(curr_image) + diff = cv2.absdiff(gray_pre_image, gray_curr_image) + res, diff = cv2.threshold(diff, 0, 255, cv2.THRESH_BINARY+cv2.THRESH_OTSU) + cnt_diff = np.sum(np.sum(diff)) + return cnt_diff + + +def exponential_smoothing(alpha, s): + ''' + Primary exponential smoothing + :param alpha: Smoothing factor,num + :param s: List of data,list + :return: List of data after smoothing,list + ''' + s_temp = [s[0]] + print(s_temp) + for i in range(1, len(s), 1): + s_temp.append(alpha * s[i - 1] + (1 - alpha) * s_temp[i - 1]) + return s_temp + + +def extract_frames(video_path: str, resize: Tuple[int, int] = None, smooth=False, alpha=0.07, window=25) -> List[str]: + """Extract frames from video.""" + frames = [] + vidcap = cv2.VideoCapture(video_path) + diff = [] + frm = 0 + pre_image = np.array([]) + cur_image = np.array([]) + + while True: + frm = frm + 1 + success, image = vidcap.read() + if not success: + break + + if frm == 1: + pre_image = image + cur_image = image + else: + pre_image = cur_image + cur_image = image + + diff.append(abs_diff(pre_image, cur_image)) + + if smooth: + diff = exponential_smoothing(alpha, diff) + + diff = np.array(diff) + mean = np.mean(diff) + dev = np.std(diff) + diff = (diff - mean) / dev + + idx = [] + for i, d in enumerate(diff): + ub = len(diff) - 1 + lb = 0 + if not i - window // 2 < lb: + lb = i - window // 2 + if not i + window // 2 > ub: + ub = i + window // 2 + + comp_window = diff[lb: ub] + if d >= max(comp_window): + idx.append(i) + + tmp = np.array(idx) + tmp = tmp + 1 + idx = set(tmp.tolist()) + vidcap.release() + + vidcap = cv2.VideoCapture(video_path) + i = 0 + frm = 0 + while vidcap.isOpened() and i < 10: + frm = frm + 1 + success, image = vidcap.read() + if not success: + break + if frm not in idx: + continue + if resize is not None: + dest_width, dest_height = resize + width, height = image.shape[:2] + if width > dest_width or height > dest_height: + width_rate = dest_width / width + height_rate = dest_height / height + rate = min(width_rate, height_rate) + dest_width = int(width * rate) + dest_height = int(height * rate) + image = cv2.resize(image, (dest_width, dest_height), interpolation=cv2.INTER_AREA) + _, buffer = cv2.imencode(".jpg", image) + frames.append(f"data:image/jpg;base64,{base64.b64encode(buffer).decode('utf-8')}") + i += 1 + vidcap.release() + return frames diff --git a/src/component/agent_manager/agent_manager.py b/src/component/agent_manager/agent_manager.py index aeff7b4..cb1ed03 100644 --- a/src/component/agent_manager/agent_manager.py +++ b/src/component/agent_manager/agent_manager.py @@ -130,14 +130,11 @@ class AgentManager: logger.error(f"read agent.toml cfg from {agent_media} failed! unexpected error occurred: {str(e)}") return None - agent_name = os.path.split(agent_media.full_path)[1] - spec = importlib.util.spec_from_file_location(agent_name, custom_agent) - the_api = importlib.util.module_from_spec(spec) - spec.loader.exec_module(the_api) - if not hasattr(the_api,"Agent"): + agent = runpy.run_path(custom_agent) + if "init" not in agent: logger.error(f"read agent.toml cfg from {agent_media} failed! unexpected error occurred: {str(e)}") return None - return the_api.Agent() + return agent["init"]() diff --git a/src/component/mail_environment/issue.py b/src/component/mail_environment/issue.py index 0c55e7e..075d12b 100644 --- a/src/component/mail_environment/issue.py +++ b/src/component/mail_environment/issue.py @@ -19,7 +19,7 @@ class IssueUpdateHistory: "source": self.source, "changes": self.changes, } - + @classmethod def from_json_dict(cls, json_dict: dict) -> "IssueUpdateHistory": return IssueUpdateHistory(json_dict["source"], json_dict["changes"]) @@ -40,7 +40,7 @@ class Issue: json_dict = { "id": self.id, "summary": self.summary, - "state": self.state.name, + "state": self.state.name, "create_time": self.create_time, "deadline": self.deadline, "source": self.source, @@ -54,7 +54,7 @@ class Issue: json_dict["update_history"] = [] for history in self.update_history: json_dict["update_history"].append(history.to_json_dict()) - + return json_dict @classmethod @@ -78,26 +78,26 @@ class Issue: history = IssueUpdateHistory.from_json_dict(history_json_dict) issue.update_history.append(history) return issue - + @classmethod def object_type(cls) -> ObjectType: return ObjectType.from_user_def_type_code(0) - + def __to_desc(self, desc_list:[], recursion=None): desc = { "id": self.id, "summary": self.summary, - "state": self.state.name, + "state": self.state.name, "deadline": self.deadline, } desc_list.append(desc) if not recursion or not self.parent: - return + return else: parent = recursion.get_issue_by_id(self.parent) parent.__to_desc(desc_list, recursion) - + def to_prompt(self, recursion=None) -> str: desc_list = [] self.__to_desc(desc_list, recursion) @@ -107,8 +107,8 @@ class Issue: root["child"] = child root = child return json.dumps(root) - - + + @classmethod def prompt_desc(cls) -> str: return '''a issue contains following fileds: { @@ -119,7 +119,7 @@ class Issue: children: child issues of this issue } ''' - + def calculate_id(self) -> str: desc = { "summary": self.summary, @@ -183,7 +183,7 @@ class IssueStorage: return self.root this_mail = mail_storage.get_mail_by_id(this_mail.reply_to) - + def add_issue(self, source_id: str, parent_id: str, summary: str): parent_issue = self.get_issue_by_id(parent_id) issue = Issue() @@ -204,11 +204,19 @@ class IssueStorage: "new": value, } issue.__dict__[key] = value - issue.update_history.append(IssueUpdateHistory(source_id, changes)) + issue.update_history.append(IssueUpdateHistory(source_id, changes)) self.__flush() return issue - + + +class IssueAgent(CustomAIAgent): + async def _process_msg(self, msg: AgentMsg, workspace=None) -> AgentMsg: + pass + + def __init__(self, agent_id: str, llm_model_name: str, max_token_size: int) -> None: + super().__init__(agent_id, llm_model_name, max_token_size) + class IssueParserEnvironment(Environment): def __init__(self, env_id: str, storage: IssueStorage) -> None: @@ -217,30 +225,30 @@ class IssueParserEnvironment(Environment): create_description = '''create a new issue''' create_param = { - "mail_id": "new issue with which email object id", + "mail_id": "new issue with which email object id", "issue_id": '''new issue's parent issue id''', "summary": '''new issue's summary''', } - self.add_ai_function(SimpleAIFunction("create_issue", + self.add_ai_function(SimpleAIFunction("create_issue", create_description, - self._create, + self._create, create_param)) - + update_description = '''update an existing issue''' update_param = { - "mail_id": "update issue with which email object id", + "mail_id": "update issue with which email object id", "issue_id": '''update issue's id''', "summary": '''issue's new summary''', } - self.add_ai_function(SimpleAIFunction("update_issue", + self.add_ai_function(SimpleAIFunction("update_issue", update_description, - self._update, + self._update, update_param)) - + async def _create(self, mail_id: str, issue_id: str, summary: str): issue = self.storage.add_issue(mail_id, issue_id, summary) return issue.id - + async def _update(self, mail_id: str, issue_id: str, summary: str): update = {} update["summary"] = summary @@ -253,7 +261,7 @@ class IssueParser: mail_path = string.Template(config["mail_path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) issue_path = string.Template(config["issue_path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) config["path"] = issue_path - + self.env = env self.config = config self.mail_storage = MailStorage(mail_path) @@ -268,7 +276,7 @@ class IssueParser: self.llm_env = IssueParserEnvironment("issue_parser", self.issue_storage) @classmethod - def __load_issue_config(cls, issue_config: dict) -> Issue: + def __load_issue_config(cls, issue_config: dict) -> Issue: issue = Issue() issue.summary = issue_config["summary"] if "children" in issue_config: @@ -276,15 +284,15 @@ class IssueParser: child_issue = cls.__load_issue_config(child_config) issue.children.append(child_issue) return issue - + @classmethod def __calac_issue_id(cls, issue: Issue): issue_id = issue.calculate_id() for child in issue.children: child.parent = issue_id cls.__calac_issue_id(child) - - + + def get_path(self) -> str: return self.config["path"] @@ -304,8 +312,8 @@ class IssueParser: and a issue in json format, {issue_desc}. Read mail's fileds and issue's fileds, and decide if you should update the issue or create a new issue with this mail. Then call the function create_issue or update_issue. if this mail is not associated with issue, you should ignore this mail.'''} - - prompt.append(AgentPrompt(f'''Mail is {mail_str}, issue is {issue_str}. Answer me the function's return value or None if igonred. + + prompt.append(IssueAgent(f'''Mail is {mail_str}, issue is {issue_str}. Answer me the function's return value or None if igonred. ''')) llm_result = await CustomAIAgent("issue parser", "gpt-4-1106-preview", 4000).do_llm_complection(prompt, env=self.llm_env) diff --git a/src/component/openai_node/open_ai_node.py b/src/component/openai_node/open_ai_node.py index 1e595fd..e45dd95 100644 --- a/src/component/openai_node/open_ai_node.py +++ b/src/component/openai_node/open_ai_node.py @@ -8,8 +8,10 @@ import json import aiohttp import base64 import requests +from openai._types import NOT_GIVEN from aios import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType,ComputeTaskResultCode,ComputeNode,AIStorage,UserConfig +from aios import image_utils logger = logging.getLogger(__name__) @@ -92,15 +94,19 @@ class OpenAI_ComputeNode(ComputeNode): def _image_2_text(self, task: ComputeTask): logger.info('openai image_2_text') # 本地图片处理 - def encode_image(image_path): - with open(image_path, "rb") as image_file: - return base64.b64encode(image_file.read()).decode('utf-8') + headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.openai_api_key }" } model_name = task.params["model_name"] - base64_image = encode_image(task.params["image_path"]) + image_path = task.params["image_path"] + + if image_utils.is_file(image_path): + url = image_utils.to_base64(image_path, (1024, 1024)) + else: + url = image_path + payload = { "model": model_name, "messages": [ @@ -114,7 +120,7 @@ class OpenAI_ComputeNode(ComputeNode): { "type": "image_url", "image_url": { - "url": f"data:image/jpeg;base64,{base64_image}" + "url": url } } ] @@ -196,7 +202,16 @@ class OpenAI_ComputeNode(ComputeNode): if max_token_size is None: max_token_size = 4000 - result_token = max_token_size + if mode_name == "gpt-4-vision-preview": + response_format = NOT_GIVEN + llm_inner_functions = None + if max_token_size > 4096: + result_token = 4096 + else: + result_token = max_token_size + else: + result_token = NOT_GIVEN + client = AsyncOpenAI(api_key=self.openai_api_key) try: if llm_inner_functions is None: @@ -204,7 +219,7 @@ class OpenAI_ComputeNode(ComputeNode): resp = await client.chat.completions.create(model=mode_name, messages=prompts, response_format = response_format, - #max_tokens=result_token, + max_tokens=result_token, ) else: logger.info(f"call openai {mode_name} prompts: \n\t {prompts} \nfunctions: \n\t{json.dumps(llm_inner_functions)}") @@ -212,7 +227,7 @@ class OpenAI_ComputeNode(ComputeNode): messages=prompts, response_format = response_format, functions=llm_inner_functions, - # max_tokens=result_token, + max_tokens=result_token, ) # TODO: add temperature to task params? except Exception as e: logger.error(f"openai run LLM_COMPLETION task error: {e}") @@ -222,7 +237,12 @@ class OpenAI_ComputeNode(ComputeNode): return result logger.info(f"openai response: {resp}") - status_code = resp.choices[0].finish_reason + if mode_name == "gpt-4-vision-preview": + status_code = resp.choices[0].finish_reason + if status_code is None: + status_code = resp.choices[0].finish_details['type'] + else: + status_code = resp.choices[0].finish_reason token_usage = resp.usage match status_code: diff --git a/src/component/tg_tunnel.py b/src/component/tg_tunnel.py index c0941ba..fa0e5e5 100644 --- a/src/component/tg_tunnel.py +++ b/src/component/tg_tunnel.py @@ -1,4 +1,6 @@ +import datetime import logging +import os.path import threading import asyncio import uuid @@ -51,6 +53,9 @@ class TelegramTunnel(AgentTunnel): self.allow_group = "contact" self.in_process_tg_msg = {} self.chatid_record = {} + self.telegram_cache = os.path.join(AIStorage.get_instance().get_myai_dir(), "telegram") + if not os.path.exists(self.telegram_cache): + os.makedirs(self.telegram_cache) async def _do_process_raw_message(self,bot: Bot, update_id: int) -> int: # Request updates after the last update_id @@ -58,7 +63,7 @@ class TelegramTunnel(AgentTunnel): for update in updates: next_update_id = update.update_id + 1 - if update.message and update.message.text: + if update.message and (update.message.text or (update.message.photo and len(update.message.photo) > 0) or update.message.video): await self.on_message(bot,update) return next_update_id @@ -96,9 +101,10 @@ class TelegramTunnel(AgentTunnel): update_id += 1 except Exception as e: logger.error(f"tg_tunnel error:{e}") + logger.exception(e) await asyncio.sleep(1) - + asyncio.create_task(_run_app()) logger.info(f"tunnel {self.tunnel_id} started.") @@ -120,7 +126,7 @@ class TelegramTunnel(AgentTunnel): # if chatid is None: # logger.warning(f"tg_tunnel process message {msg.msg_id} from agent {msg.sender} to human {msg.target} failed! chatid not found!") # return None - + # if bot is None: # logger.warning(f"tg_tunnel process message {msg.msg_id} from agent {msg.sender} to human {msg.target} failed! bot not found!") # return None @@ -130,7 +136,7 @@ class TelegramTunnel(AgentTunnel): # await bot.send_message(chat_id=chatid,text=msg.body) # logging.info(f"tg_tunnel send message {msg.msg_id} from agent {msg.sender} to human {msg.target} @ chatid:{chatid}success!") # return None - + # logger.warning(f"tg_tunnel process message {msg.msg_id} from agent {msg.sender} to human {msg.target} failed! contact not found!") # return None @@ -143,13 +149,37 @@ class TelegramTunnel(AgentTunnel): else: logger.warning(f"tg_tunnel process message {msg.msg_id} from agent {msg.sender} to human {msg.target} failed! chatid not found!") + def get_cache_path(self) -> str: + today = datetime.datetime.today() + path = os.path.join(self.telegram_cache, str(today.year), str(today.month)) + if not os.path.exists(path): + os.makedirs(path) + return path async def conver_tg_msg_to_agent_msg(self,message:Message) -> AgentMsg: agent_msg = AgentMsg() agent_msg.topic = "_telegram" agent_msg.msg_id = "tg_msg#" + str(message.message_id) + "#" + uuid.uuid4().hex agent_msg.target = self.target_id - agent_msg.body = message.text + if message.text is not None: + agent_msg.body = message.text + elif message.photo is not None and len(message.photo) > 0: + photo_files = [] + photo_file = await message.photo[-1].get_file() + ext = photo_file.file_path.rsplit(".")[-1] + file_path = os.path.join(self.get_cache_path(), photo_file.file_id + f".{ext}") + await photo_file.download_to_drive(file_path) + photo_files.append(file_path) + agent_msg.body = agent_msg.create_image_body(photo_files, message.caption) + agent_msg.body_mime = f"image/{ext}" + elif message.video is not None: + video_file = await message.video.get_file() + ext = video_file.file_path.rsplit(".")[-1] + file_path = os.path.join(self.get_cache_path(), video_file.file_id + f".{ext}") + await video_file.download_to_drive(file_path) + agent_msg.body = agent_msg.create_video_body(file_path, message.caption) + agent_msg.body_mime = f"video/{ext}" + agent_msg.create_time = time.time() messag_type = message.chat.type if messag_type == "supergroup" or messag_type == "group": @@ -168,7 +198,7 @@ class TelegramTunnel(AgentTunnel): agent_msg.mentions.append(self.target_id) else: agent_msg.mentions.append(mention) - + if message.caption_entities: for entity in message.caption_entities: if entity.type == 'mention': @@ -203,11 +233,11 @@ class TelegramTunnel(AgentTunnel): if update.effective_user.is_bot: logger.warning(f"ignore message from telegram bot {update.effective_user.id}") return None - + if self.in_process_tg_msg.get(update.message.message_id) is not None: logger.warning(f"ignore message from telegram bot {update.effective_user.id}") return None - + self.in_process_tg_msg[update.message.message_id] = True agent_msg = await self.conver_tg_msg_to_agent_msg(message) @@ -226,7 +256,7 @@ class TelegramTunnel(AgentTunnel): if self.allow_group != "contact" and self.allow_group !="guest": await update.message.reply_text(f"You're not supposed to talk to me! Please contact my father~") return - + else: if self.allow_group != "guest": await update.message.reply_text(f"The current Telegram account is not in the contact list. If you want to receive a reply, you can add the configuration in the contacts.toml file or switch tunnel to guest mode.") @@ -246,7 +276,7 @@ class TelegramTunnel(AgentTunnel): if contact is not None: contact.set_active_tunnel(self.target_id,self) self.chatid_record[reomte_user_name] = update.effective_chat.id - self.ai_bus.register_message_handler(reomte_user_name,contact._process_msg) + self.ai_bus.register_message_handler(reomte_user_name,contact._process_msg) agent_msg.sender = reomte_user_name logger.info(f"process message {agent_msg.msg_id} from {agent_msg.sender} to {agent_msg.target}") @@ -266,11 +296,11 @@ class TelegramTunnel(AgentTunnel): if resp_msg.body_mime is None: if resp_msg.body is None: return - + if len(resp_msg.body) < 1: await update.message.reply_text("") return - + knowledge_object = KnowledgeStore().parse_object_in_message(resp_msg.body) if knowledge_object is not None: if knowledge_object.get_object_type() == ObjectType.Image: diff --git a/src/requirements.txt b/src/requirements.txt index 1a41e3d..11f207f 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -151,3 +151,5 @@ psycopg2-binary pyodbc oracledb html2text +docx2txt +opencv-python diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 1a13476..884acef 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -33,7 +33,7 @@ from component.llama_node.local_llama_compute_node import LocalLlama_ComputeNode sys.path.append(directory + '/../../component/') -from google_node import * +from google_node import * from llama_node import * from openai_node import * from sd_node import * @@ -240,12 +240,12 @@ class AIOS_Shell: def get_version(self) -> str: return "0.5.1" - async def send_msg(self,msg:str,target_id:str,topic:str,sender:str = None) -> str: + async def send_msg(self,msg:str,target_id:str,topic:str,sender:str = None, msg_mime:str=None) -> str: if sender == self.username: AIBus().get_default_bus().register_message_handler(self.username,self._user_process_msg) agent_msg = AgentMsg() - agent_msg.set(sender,target_id,msg) + agent_msg.set(sender,target_id,msg,body_mime=msg_mime) agent_msg.topic = topic resp = await AIBus.get_default_bus().send_message(agent_msg) if resp is not None: @@ -455,6 +455,54 @@ class AIOS_Shell: show_text = FormattedText([("class:title", f"{self.current_topic}@{self.current_target} >>> "), ("class:content", resp)]) return show_text + case 'send_img': + sender = None + if len(args) == 4: + target_id = args[0] + msg_content = args[1] + image_path = args[2] + topic = args[3] + sender = self.username + elif len(args) == 5: + target_id = args[0] + msg_content = args[1] + image_path = args[2] + topic = args[3] + sender = args[4] + + ext = os.path.splitext(image_path)[1][1:] + resp = await self.send_msg(AgentMsg.create_image_body([image_path], msg_content), + target_id, + topic, + sender, + f"image/{ext}") + show_text = FormattedText([("class:title", f"{self.current_topic}@{self.current_target} >>> "), + ("class:content", resp)]) + return show_text + case 'send_video': + sender = None + if len(args) == 4: + target_id = args[0] + msg_content = args[1] + video_path = args[2] + topic = args[3] + sender = self.username + elif len(args) == 5: + target_id = args[0] + msg_content = args[1] + video_path = args[2] + topic = args[3] + sender = args[4] + + ext = os.path.splitext(video_path)[1][1:] + resp = await self.send_msg(AgentMsg.create_video_body(video_path, msg_content), + target_id, + topic, + sender, + f"video/{ext}") + show_text = FormattedText([("class:title", f"{self.current_topic}@{self.current_target} >>> "), + ("class:content", resp)]) + return show_text case 'set_config': show_text = FormattedText([("class:error", f"set config args error,/set_config $config_item! ")]) if len(args) == 1: @@ -770,6 +818,8 @@ async def main(): return await main_daemon_loop(shell) completer = WordCompleter(['/send $target $msg $topic', + '/send_img $target $msg $img_path $topic', + '/send_video $target &msg &video_path $topic', '/open $target $topic', '/history $num $offset', '/connect $target',