diff --git a/rootfs/agents/TextSummary/agent.py b/rootfs/agents/TextSummary/agent.py new file mode 100644 index 0000000..ba6f505 --- /dev/null +++ b/rootfs/agents/TextSummary/agent.py @@ -0,0 +1,48 @@ +import copy + +from aios_kernel import CustomAIAgent, AgentMsg, AgentMsgType, AgentPrompt +from aios_kernel.compute_task import ComputeTaskResultCode +from knowledge.data.writer import split_text + + +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..f3bf29a --- /dev/null +++ b/rootfs/agents/Vision/agent.toml @@ -0,0 +1,7 @@ +instance_id = "Vision" +fullname = "Vision" +llm_model_name = "gpt-4-vision-preview" + +[[prompt]] +role = "system" +content = """你的工作对用户输入的图片和视频做分析，并根据用户的意图做出回应。""" diff --git a/src/aios/agent/agent.py b/src/aios/agent/agent.py index ad8dcf8..f9e728d 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 . 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.image_to_base64(image_path) + 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", "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", "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) + if video_prompt is None: + content = [{"type": "text", "text": f"{msg.sender}'s message"}] + content.extend([{"type": "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", "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", "url": image} for image in images]}] + else: + content = [{"type": "text", "text": image_prompt}] + content.extend([{"type": "image_url", "url": 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) + if video_prompt is None: + msg_prompt.messages = [{"role": "user", "content": [{"type": "image_url", "url": frame} for frame in frames]}] + else: + content = [{"type": "text", "text": video_prompt}] + content.extend([{"type": "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..78e431d 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("gpt4"): + 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) 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_kernel/image_utils.py b/src/aios_kernel/image_utils.py new file mode 100644 index 0000000..dd706bc --- /dev/null +++ b/src/aios_kernel/image_utils.py @@ -0,0 +1,22 @@ +import base64 +import os.path + + +def to_base64(image_path: str) -> str: + """Convert image to base64.""" + ext = os.path.splitext(image_path)[1] + 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}" + + +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_kernel/video_utils.py b/src/aios_kernel/video_utils.py new file mode 100644 index 0000000..444fab8 --- /dev/null +++ b/src/aios_kernel/video_utils.py @@ -0,0 +1,18 @@ +import base64 +from typing import List + +import cv2 + + +def extract_frames(video_path: str) -> List[str]: + """Extract frames from video.""" + frames = [] + vidcap = cv2.VideoCapture(video_path) + while vidcap.isOpened(): + success, image = vidcap.read() + if not success: + break + _, buffer = cv2.imencode(".jpg", image) + frames.append(base64.b64encode(buffer).decode("utf-8")) + 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/openai_node/open_ai_node.py b/src/component/openai_node/open_ai_node.py index 1e595fd..1918b2d 100644 --- a/src/component/openai_node/open_ai_node.py +++ b/src/component/openai_node/open_ai_node.py @@ -11,6 +11,7 @@ import requests from aios import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType,ComputeTaskResultCode,ComputeNode,AIStorage,UserConfig + logger = logging.getLogger(__name__) @@ -92,15 +93,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) + else: + url = image_path + payload = { "model": model_name, "messages": [ @@ -114,7 +119,7 @@ class OpenAI_ComputeNode(ComputeNode): { "type": "image_url", "image_url": { - "url": f"data:image/jpeg;base64,{base64_image}" + "url": url } } ] 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 0ac7a81..16de836 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -238,12 +238,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: