From 7b5c0103e8fdebc3353f2c89f99d3896e7cf19d0 Mon Sep 17 00:00:00 2001 From: streetycat <305190374@qq.com> Date: Thu, 28 Sep 2023 00:51:52 +0000 Subject: [PATCH 1/6] local llama --- rootfs/agents/math_teacher/agent.toml | 2 +- src/aios_kernel/local_llama_compute_node.py | 117 ++++++++++---------- src/aios_kernel/queue_compute_node.py | 32 ++---- 3 files changed, 65 insertions(+), 86 deletions(-) diff --git a/rootfs/agents/math_teacher/agent.toml b/rootfs/agents/math_teacher/agent.toml index 646c8e4..f7a29a5 100644 --- a/rootfs/agents/math_teacher/agent.toml +++ b/rootfs/agents/math_teacher/agent.toml @@ -1,6 +1,6 @@ instance_id = "math_teacher" fullname = "the one" -llm_model_name = "gpt-4-0613" +llm_model_name = "LLaMA2-70B" [[prompt]] role = "system" content = "你是精通数学的老师" diff --git a/src/aios_kernel/local_llama_compute_node.py b/src/aios_kernel/local_llama_compute_node.py index 15eec79..bca0c22 100644 --- a/src/aios_kernel/local_llama_compute_node.py +++ b/src/aios_kernel/local_llama_compute_node.py @@ -1,10 +1,12 @@ +import json import logging import requests from typing import Optional, List from pydantic import BaseModel +from llama_cpp import Llama -from .compute_task import ComputeTask, ComputeTaskState, ComputeTaskType +from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType from .queue_compute_node import Queue_ComputeNode logger = logging.getLogger(__name__) @@ -14,69 +16,64 @@ This is a custom implementation, it should be redesigned. """ class LocalLlama_ComputeNode(Queue_ComputeNode): - async def execute_task(self, task: ComputeTask) -> { - "content": str, - "message": str, - "state": ComputeTaskState, - "error": { - "code": int, - "message": str, - } - }: - class GenerateResponse(BaseModel): - error: Optional[int] - msg: Optional[str] - results: Optional[List[str]] + def __init__(self, model_path: str, model_name: str): + super().__init__() + self.model_path = model_path + self.model_name = model_name + self.llm = Llama(model_path=model_path) - try: - prompt_msgs = [] - for prompt in task.params["prompts"]: - prompt_msgs.append(prompt["content"]) + async def execute_task(self, task: ComputeTask) -> ComputeTaskResult: + match task.task_type: + case ComputeTaskType.TEXT_EMBEDDING: + model_name = task.params["model_name"] + input = task.params["input"] + logger.info(f"call openai {model_name} input: {input}") + + embedding = self.llm.embed(input=input) - body = { - "prompts": prompt_msgs - } - - response = requests.post("http://aigc:7880/generate", json = body, verify=False, headers={"Content-Type": "application/json"}) - response.close() + logger.info(f"local-llama({self.model_path}) response: {resp}") - logger.info(f"LocalLlama_ComputeNode task responsed, request: {body}, status-code: {response.status_code}, headers: {response.headers}, content: {response.content}") + result = ComputeTaskResult() + result.set_from_task(task) + result.result = embedding + + return result + case ComputeTaskType.LLM_COMPLETION: + mode_name = task.params["model_name"] + prompts = task.params["prompts"] + max_token_size = task.params.get("max_token_size") + llm_inner_functions = task.params.get("inner_functions") + if max_token_size is None: + max_token_size = 4000 + + logger.info(f"local-llama({self.model_path}) prompts: {prompts}") + + resp = self.llm.create_chat_completion(model=mode_name, + messages=prompts, + functions=llm_inner_functions, # function has not support? + max_tokens=max_token_size, + temperature=0.7) # TODO: add temperature to task params? - if response.status_code != 200: - return { - "state": ComputeTaskState.ERROR, - "error": { - "code": response.status_code, - "message": "http request failed: " + str(response.status_code) - } - } - else: - resp = response.json() - if "error" in resp: - return { - "state": ComputeTaskState.ERROR, - "error": { - "code": resp["error"], - "message": "local llama failed:" + resp["msg"] - } - } - else: - return { - "state": ComputeTaskState.DONE, - "content": str(resp["results"]), - "message": str(resp["results"]) - } - except Exception as err: - import traceback - logger.error(f"{traceback.format_exc()}, error: {err}") - return { - "state": ComputeTaskState.ERROR, - "error": { - "code": -1, - "message": "unknown exception: " + str(err) - } - } + logger.info(f"local-llama({self.model_path}) response: {json.dumps(resp, indent=4)}") + + result = ComputeTaskResult() + result.set_from_task(task) + + status_code = resp["choices"][0]["finish_reason"] + match status_code: + case "function_call": + task.state = ComputeTaskState.DONE + case "stop": + task.state = ComputeTaskState.DONE + case _: + task.state = ComputeTaskState.ERROR + task.error_str = f"The status code was {status_code}." + return None + + result.result_str = resp["choices"][0]["message"]["content"] + result.result_message = resp["choices"][0]["message"] + return result async def initial(self) -> bool: return True @@ -88,7 +85,7 @@ class LocalLlama_ComputeNode(Queue_ComputeNode): pass def is_support(self, task: ComputeTask) -> bool: - return task.task_type == ComputeTaskType.LLM_COMPLETION and (not task.params["model_name"] or task.params["model_name"] == "llama") + return (task.task_type == ComputeTaskType.TEXT_EMBEDDING or task.task_type == ComputeTaskType.LLM_COMPLETION) and (not task.params["model_name"] or task.params["model_name"] == self.model_name) def is_local(self) -> bool: return True diff --git a/src/aios_kernel/queue_compute_node.py b/src/aios_kernel/queue_compute_node.py index 6d97446..6b3469b 100644 --- a/src/aios_kernel/queue_compute_node.py +++ b/src/aios_kernel/queue_compute_node.py @@ -16,15 +16,7 @@ class Queue_ComputeNode(ComputeNode): self.is_start = False @abstractmethod - async def execute_task(self, task: ComputeTask) -> { - "content": str, - "message": str, - "state": ComputeTaskState, - "error": { - "code": int, - "message": str, - } - }: + async def execute_task(self, task: ComputeTask) -> ComputeTaskResult: pass async def push_task(self, task: ComputeTask, proiority: int = 0): @@ -37,23 +29,13 @@ class Queue_ComputeNode(ComputeNode): async def _run_task(self, task: ComputeTask): task.state = ComputeTaskState.RUNNING - resp = await self.execute_task(task) - - result = ComputeTaskResult() - - result.worker_id = self.node_id - task.state = resp["state"] - - if task.state == ComputeTaskState.ERROR: - result.result_code = ComputeTaskResultCode.ERROR - task.error_str = resp["error"]["message"] + result = await self.execute_task(task) + if result is not None: + result.set_from_task(task) + result.worker_id = self.node_id else: - result.result_code = ComputeTaskResultCode.OK - result.result_str = resp["content"] - result.result_message = resp["message"] - - result.set_from_task(task) - + task.state = ComputeTaskState.ERROR + return result def start(self): From 97d69e43f76219ead5accfadceb547b7c2fe6ba2 Mon Sep 17 00:00:00 2001 From: streetycat <305190374@qq.com> Date: Thu, 28 Sep 2023 05:03:06 +0000 Subject: [PATCH 2/6] fix bugs --- src/aios_kernel/local_llama_compute_node.py | 60 +++++++++++++++------ src/aios_kernel/queue_compute_node.py | 16 +++--- 2 files changed, 51 insertions(+), 25 deletions(-) diff --git a/src/aios_kernel/local_llama_compute_node.py b/src/aios_kernel/local_llama_compute_node.py index bca0c22..a16c3a3 100644 --- a/src/aios_kernel/local_llama_compute_node.py +++ b/src/aios_kernel/local_llama_compute_node.py @@ -6,7 +6,7 @@ from typing import Optional, List from pydantic import BaseModel from llama_cpp import Llama -from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskState, ComputeTaskType +from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskResultCode, ComputeTaskState, ComputeTaskType from .queue_compute_node import Queue_ComputeNode logger = logging.getLogger(__name__) @@ -22,19 +22,26 @@ class LocalLlama_ComputeNode(Queue_ComputeNode): self.model_name = model_name self.llm = Llama(model_path=model_path) - async def execute_task(self, task: ComputeTask) -> ComputeTaskResult: + async def execute_task(self, task: ComputeTask, result: ComputeTaskResult) -> ComputeTaskResult: match task.task_type: case ComputeTaskType.TEXT_EMBEDDING: model_name = task.params["model_name"] input = task.params["input"] - logger.info(f"call openai {model_name} input: {input}") + logger.info(f"call local-llama {model_name} input: {input}") - embedding = self.llm.embed(input=input) + try: + embedding = self.llm.embed(input=input) + logger.info(f"local-llama({self.model_path}) response: {embedding}") + except Exception as e: + logger.error(f"call local-llama {model_name} run TEXT_EMBEDDING task error: {e}") + task.state = ComputeTaskState.ERROR + task.error_str = str(e) + result.error_str = str(e) + return result - logger.info(f"local-llama({self.model_path}) response: {resp}") - - result = ComputeTaskResult() - result.set_from_task(task) + logger.info(f"local-llama({self.model_path}) response: {embedding}") + task.state = ComputeTaskState.DONE + result.result_code = ComputeTaskResultCode.OK result.result = embedding return result @@ -48,19 +55,24 @@ class LocalLlama_ComputeNode(Queue_ComputeNode): logger.info(f"local-llama({self.model_path}) prompts: {prompts}") - resp = self.llm.create_chat_completion(model=mode_name, - messages=prompts, - functions=llm_inner_functions, # function has not support? - max_tokens=max_token_size, - temperature=0.7) # TODO: add temperature to task params? + try: + resp = self.llm.create_chat_completion(model=mode_name, + messages=prompts, + functions=llm_inner_functions, # function has not support? + max_tokens=max_token_size, + temperature=0.7) # TODO: add temperature to task params? + except Exception as e: + logger.error(f"local-llama({self.model_path}) run LLM_COMPLETION task error: {e}") + task.state = ComputeTaskState.ERROR + task.error_str = str(e) + result.error_str = str(e) + return result logger.info(f"local-llama({self.model_path}) response: {json.dumps(resp, indent=4)}") - result = ComputeTaskResult() - result.set_from_task(task) - status_code = resp["choices"][0]["finish_reason"] + token_usage = resp["usage"] match status_code: case "function_call": task.state = ComputeTaskState.DONE @@ -69,11 +81,25 @@ class LocalLlama_ComputeNode(Queue_ComputeNode): case _: task.state = ComputeTaskState.ERROR task.error_str = f"The status code was {status_code}." + result.error_str = f"The status code was {status_code}." + result.result_code = ComputeTaskResultCode.ERROR return None - + + result.result_code = ComputeTaskResultCode.OK result.result_str = resp["choices"][0]["message"]["content"] result.result_message = resp["choices"][0]["message"] + if token_usage: + result.result_refers["token_usage"] = token_usage + + logger.info(f"local-llama({self.model_path}) success response: {result.result_str}") + return result + case _: + task.state = ComputeTaskState.ERROR + result.result_code = ComputeTaskResultCode.ERROR + task.error_str = f"ComputeTask's TaskType : {task.task_type} not support!" + result.error_str = f"ComputeTask's TaskType : {task.task_type} not support!" + return None async def initial(self) -> bool: return True diff --git a/src/aios_kernel/queue_compute_node.py b/src/aios_kernel/queue_compute_node.py index 6b3469b..261aaa9 100644 --- a/src/aios_kernel/queue_compute_node.py +++ b/src/aios_kernel/queue_compute_node.py @@ -16,7 +16,7 @@ class Queue_ComputeNode(ComputeNode): self.is_start = False @abstractmethod - async def execute_task(self, task: ComputeTask) -> ComputeTaskResult: + async def execute_task(self, task: ComputeTask, result: ComputeTaskResult): pass async def push_task(self, task: ComputeTask, proiority: int = 0): @@ -28,13 +28,13 @@ class Queue_ComputeNode(ComputeNode): async def _run_task(self, task: ComputeTask): task.state = ComputeTaskState.RUNNING - - result = await self.execute_task(task) - if result is not None: - result.set_from_task(task) - result.worker_id = self.node_id - else: - task.state = ComputeTaskState.ERROR + + result = ComputeTaskResult() + result.result_code = ComputeTaskResultCode.ERROR + result.set_from_task(task) + result.worker_id = self.node_id + + await self.execute_task(task, result) return result From 4f6b04fe48819f8a8cbd76f2827cb923bc7ca38a Mon Sep 17 00:00:00 2001 From: streetycat <305190374@qq.com> Date: Thu, 28 Sep 2023 00:20:20 +0000 Subject: [PATCH 3/6] local llama --- src/aios_kernel/local_llama_compute_node.py | 2 +- src/aios_kernel/queue_compute_node.py | 1 + src/requirements.txt | 2 +- src/service/aios_shell/aios_shell.py | 5 +---- 4 files changed, 4 insertions(+), 6 deletions(-) diff --git a/src/aios_kernel/local_llama_compute_node.py b/src/aios_kernel/local_llama_compute_node.py index a16c3a3..7d8c87e 100644 --- a/src/aios_kernel/local_llama_compute_node.py +++ b/src/aios_kernel/local_llama_compute_node.py @@ -67,12 +67,12 @@ class LocalLlama_ComputeNode(Queue_ComputeNode): task.error_str = str(e) result.error_str = str(e) return result - logger.info(f"local-llama({self.model_path}) response: {json.dumps(resp, indent=4)}") status_code = resp["choices"][0]["finish_reason"] token_usage = resp["usage"] + match status_code: case "function_call": task.state = ComputeTaskState.DONE diff --git a/src/aios_kernel/queue_compute_node.py b/src/aios_kernel/queue_compute_node.py index 261aaa9..3c1bf3e 100644 --- a/src/aios_kernel/queue_compute_node.py +++ b/src/aios_kernel/queue_compute_node.py @@ -29,6 +29,7 @@ class Queue_ComputeNode(ComputeNode): async def _run_task(self, task: ComputeTask): task.state = ComputeTaskState.RUNNING + result = ComputeTaskResult() result.result_code = ComputeTaskResultCode.ERROR result.set_from_task(task) diff --git a/src/requirements.txt b/src/requirements.txt index c6128ee..b83e564 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -137,4 +137,4 @@ python-telegram-bot pydub stability_sdk sentence-transformers==2.2.2 -tiktoken +tiktoken \ No newline at end of file diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index a3aef85..1513317 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -164,9 +164,7 @@ class AIOS_Shell: # if await stability_api_node.initial() is not True: # logger.error("stability api node initial failed!") # ComputeKernel.get_instance().add_compute_node(stability_api_node) - - - + local_st_text_compute_node = LocalSentenceTransformer_Text_ComputeNode() if local_st_text_compute_node.initial() is not True: logger.error("local sentence transformer text embedding node initial failed!") @@ -179,7 +177,6 @@ class AIOS_Shell: else: ComputeKernel.get_instance().add_compute_node(local_st_image_compute_node) - await ComputeKernel.get_instance().start() AIBus().get_default_bus().register_unhandle_message_handler(self._handle_no_target_msg) From 193c627bdba3e9a05447f27a56b509431da7f774 Mon Sep 17 00:00:00 2001 From: streetycat <305190374@qq.com> Date: Thu, 28 Sep 2023 09:09:30 +0000 Subject: [PATCH 4/6] rebase to main --- rootfs/agents/math_teacher/agent.toml | 2 +- src/aios_kernel/compute_node_config.py | 87 +++++++++++ src/aios_kernel/local_llama_compute_node.py | 165 +++++++++++++------- src/service/aios_shell/aios_shell.py | 64 +++++++- 4 files changed, 255 insertions(+), 63 deletions(-) create mode 100644 src/aios_kernel/compute_node_config.py diff --git a/rootfs/agents/math_teacher/agent.toml b/rootfs/agents/math_teacher/agent.toml index f7a29a5..646c8e4 100644 --- a/rootfs/agents/math_teacher/agent.toml +++ b/rootfs/agents/math_teacher/agent.toml @@ -1,6 +1,6 @@ instance_id = "math_teacher" fullname = "the one" -llm_model_name = "LLaMA2-70B" +llm_model_name = "gpt-4-0613" [[prompt]] role = "system" content = "你是精通数学的老师" diff --git a/src/aios_kernel/compute_node_config.py b/src/aios_kernel/compute_node_config.py new file mode 100644 index 0000000..8e25f50 --- /dev/null +++ b/src/aios_kernel/compute_node_config.py @@ -0,0 +1,87 @@ +""" +Configuration for nodes: + +``` +├── nodes +│ └── llama +| └── 0 +| | └── url +| | └── model_name +| └── 1 +| └── url +| └── model_name +``` +""" +import logging +from typing import List + +import os +import toml + +from .local_llama_compute_node import LocalLlama_ComputeNode +from .storage import AIStorage + +# define singleton class knowledge pipline +class ComputeNodeConfig: + _instance = None + + @classmethod + def get_instance(cls): + if cls._instance is None: + cls._instance = ComputeNodeConfig() + cls._instance.__singleton_init__() + + return cls._instance + + def initial(self) -> List[LocalLlama_ComputeNode]: + config_path = self.__config_path() + logging.info(f"initial nodes from {config_path}") + + if os.path.exists(config_path): + self.config = toml.load(self.__config_path()) + if self.config is None: + return [] + + nodes = [] + llama_nodes_cfg = self.config["llama"] + if llama_nodes_cfg is not None: + for cfg in llama_nodes_cfg: + node = LocalLlama_ComputeNode(url=cfg["url"], model_name=cfg["model_name"]) + nodes.append(node) + + return nodes + + return [] + + def save(self): + with open(self.__config_path(), "w") as f: + toml.dump(self.config, f) + + def add_node(self, model_type: str, url: str, model_name: str): + if model_type == "llama": + llama_nodes_cfg = self.config.get("llama") or [] + for cfg in llama_nodes_cfg: + if url == cfg["url"] and model_name == cfg["model_name"]: + return + llama_nodes_cfg.append({"url": url, "model_name": model_name}) + self.config["llama"] = llama_nodes_cfg + + + def remove_node(self, model_type: str, url: str, model_name: str): + if model_type == "llama": + llama_nodes_cfg = self.config.get("llama") or [] + for i in range(0, len(llama_nodes_cfg)): + cfg = llama_nodes_cfg[i] + if url == cfg["url"] and model_name == cfg["model_name"]: + llama_nodes_cfg.pop(i) + + def list(self) -> str: + return toml.dumps(self.config) + + def __singleton_init__(self): + self.config = {} + + @classmethod + def __config_path(cls) -> str: + user_data_dir = AIStorage.get_instance().get_myai_dir() + return os.path.abspath(f"{user_data_dir}/etc/compute_nodes.cfg.toml") diff --git a/src/aios_kernel/local_llama_compute_node.py b/src/aios_kernel/local_llama_compute_node.py index 7d8c87e..52e6af4 100644 --- a/src/aios_kernel/local_llama_compute_node.py +++ b/src/aios_kernel/local_llama_compute_node.py @@ -4,10 +4,10 @@ import logging import requests from typing import Optional, List from pydantic import BaseModel -from llama_cpp import Llama from .compute_task import ComputeTask, ComputeTaskResult, ComputeTaskResultCode, ComputeTaskState, ComputeTaskType from .queue_compute_node import Queue_ComputeNode +from .storage import AIStorage,UserConfig logger = logging.getLogger(__name__) @@ -16,59 +16,117 @@ This is a custom implementation, it should be redesigned. """ class LocalLlama_ComputeNode(Queue_ComputeNode): - def __init__(self, model_path: str, model_name: str): + def __init__(self, url: str, model_name: str): super().__init__() - self.model_path = model_path + self.url = url self.model_name = model_name - self.llm = Llama(model_path=model_path) - async def execute_task(self, task: ComputeTask, result: ComputeTaskResult) -> ComputeTaskResult: + async def execute_task(self, task: ComputeTask, result: ComputeTaskResult): match task.task_type: case ComputeTaskType.TEXT_EMBEDDING: model_name = task.params["model_name"] input = task.params["input"] - logger.info(f"call local-llama {model_name} input: {input}") + logger.info(f"call local-llama ({self.url}, {self.model_name}) {model_name} input: {input}") - try: - embedding = self.llm.embed(input=input) - logger.info(f"local-llama({self.model_path}) response: {embedding}") - except Exception as e: - logger.error(f"call local-llama {model_name} run TEXT_EMBEDDING task error: {e}") - task.state = ComputeTaskState.ERROR - task.error_str = str(e) - result.error_str = str(e) - return result + self.embedding(input, result) - logger.info(f"local-llama({self.model_path}) response: {embedding}") - task.state = ComputeTaskState.DONE - result.result_code = ComputeTaskResultCode.OK - result.result = embedding + if result.result_code == ComputeTaskResultCode.OK: + task.state = ComputeTaskState.DONE + else: + task.state = ComputeTaskState.ERROR + task.error_str = result.error_str return result case ComputeTaskType.LLM_COMPLETION: mode_name = task.params["model_name"] prompts = task.params["prompts"] - max_token_size = task.params.get("max_token_size") - llm_inner_functions = task.params.get("inner_functions") - if max_token_size is None: - max_token_size = 4000 - logger.info(f"local-llama({self.model_path}) prompts: {prompts}") + logger.info(f"local-llama({self.url}, {self.model_name}) prompts: {prompts}") - try: - resp = self.llm.create_chat_completion(model=mode_name, - messages=prompts, - functions=llm_inner_functions, # function has not support? - max_tokens=max_token_size, - temperature=0.7) # TODO: add temperature to task params? - except Exception as e: - logger.error(f"local-llama({self.model_path}) run LLM_COMPLETION task error: {e}") + self.completion(task, result) + + if result.result_code == ComputeTaskResultCode.OK: + task.state = ComputeTaskState.DONE + else: task.state = ComputeTaskState.ERROR - task.error_str = str(e) - result.error_str = str(e) - return result + task.error_str = result.error_str + + case _: + task.state = ComputeTaskState.ERROR + result.result_code = ComputeTaskResultCode.ERROR + task.error_str = f"ComputeTask's TaskType : {task.task_type} not support!" + result.error_str = f"ComputeTask's TaskType : {task.task_type} not support!" + return None + + async def initial(self) -> bool: + return True + + def display(self) -> str: + return f"local-llama: {self.node_id}" + + def get_capacity(self): + pass + + def is_support(self, task: ComputeTask) -> bool: + return (task.task_type == ComputeTaskType.TEXT_EMBEDDING or task.task_type == ComputeTaskType.LLM_COMPLETION) and (not task.params["model_name"] or task.params["model_name"] == self.model_name) + + def is_local(self) -> bool: + return True + + def embedding(self, input: str, result: ComputeTaskResult): + body = { + "input": input + } + + try: + response = requests.post(self.url + "/v1/embeddings", json = body, verify=False, headers={"Content-Type": "application/json"}) + response.close() + + logger.info(f"local-llama({self.url}, {self.model_name}) task responsed, request: {body}, status-code: {response.status_code}, headers: {response.headers}, content: {response.content}") + + if response.status_code == 200: + resp = response.json() + result.result = resp["data"][0]["embedding"] + elif response.status_code == 422: + resp = response.json() + result.result_code = ComputeTaskResultCode.ERROR + result.error_str = "http request failed: " + str(resp["detail"][0]["msg"]) + else: + result.result_code = ComputeTaskResultCode.ERROR + result.error_str = "http request failed: " + str(response.status_code) + except Exception as e: + logger.error(f"call local-llama({self.url}, {self.model_name}) run TEXT_EMBEDDING task error: {e}") + result.result_code = ComputeTaskResultCode.ERROR + result.error_str = str(e) + return result + + def completion(self, task: ComputeTask, result: ComputeTaskResult): + mode_name = task.params["model_name"] + prompts = task.params["prompts"] + max_token_size = task.params.get("max_token_size") + llm_inner_functions = task.params.get("inner_functions") + if max_token_size is None: + max_token_size = max_token_size - logger.info(f"local-llama({self.model_path}) response: {json.dumps(resp, indent=4)}") + body = { + "messages": [], + "max_tokens": 4000 + } + + for prompt in prompts: + body["messages"].append({ + "role": prompt["role"], + "content": prompt["content"] + }) + + try: + response = requests.post(self.url + "/v1/chat/completions", json = body, verify=False, headers={"Content-Type": "application/json"}) + response.close() + + logger.info(f"local-llama({self.url}, {self.model_name}) task responsed, request: {body}, status-code: {response.status_code}, headers: {response.headers}, content: {response.content}") + + if response.status_code == 200: + resp = response.json() status_code = resp["choices"][0]["finish_reason"] token_usage = resp["usage"] @@ -91,27 +149,16 @@ class LocalLlama_ComputeNode(Queue_ComputeNode): if token_usage: result.result_refers["token_usage"] = token_usage - logger.info(f"local-llama({self.model_path}) success response: {result.result_str}") - - return result - case _: - task.state = ComputeTaskState.ERROR + logger.info(f"local-llama({self.url}, {self.model_name}) success response: {result.result_str}") + elif response.status_code == 422: + resp = response.json() result.result_code = ComputeTaskResultCode.ERROR - task.error_str = f"ComputeTask's TaskType : {task.task_type} not support!" - result.error_str = f"ComputeTask's TaskType : {task.task_type} not support!" - return None - - async def initial(self) -> bool: - return True - - def display(self) -> str: - return f"LocalLlama_ComputeNode: {self.node_id}" - - def get_capacity(self): - pass - - def is_support(self, task: ComputeTask) -> bool: - return (task.task_type == ComputeTaskType.TEXT_EMBEDDING or task.task_type == ComputeTaskType.LLM_COMPLETION) and (not task.params["model_name"] or task.params["model_name"] == self.model_name) - - def is_local(self) -> bool: - return True + result.error_str = "http request failed: " + str(resp["detail"][0]["msg"]) + else: + result.result_code = ComputeTaskResultCode.ERROR + result.error_str = "http request failed: " + str(response.status_code) + except Exception as e: + logger.error(f"call local-llama({self.url}, {self.model_name}) run LLM_COMPLETION task error: {e}") + result.result_code = ComputeTaskResultCode.ERROR + result.error_str = str(e) + return result \ No newline at end of file diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 1513317..ce447e5 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -20,14 +20,15 @@ from prompt_toolkit.auto_suggest import AutoSuggestFromHistory from prompt_toolkit.completion import WordCompleter from prompt_toolkit.styles import Style + directory = os.path.dirname(__file__) sys.path.append(directory + '/../../') - - +from aios_kernel import AIOS_Version,AgentMsgType,UserConfigItem,AIStorage,Workflow,AIAgent,AgentMsg,AgentMsgStatus,ComputeKernel,OpenAI_ComputeNode,AIBus,AIChatSession,AgentTunnel,TelegramTunnel,CalenderEnvironment,Environment,EmailTunnel,LocalLlama_ComputeNode,Local_Stability_ComputeNode,Stability_ComputeNode,PaintEnvironment +from aios_kernel import ContactManager,Contact import proxy from aios_kernel import * - +from aios_kernel.compute_node_config import ComputeNodeConfig sys.path.append(directory + '/../../component/') from agent_manager import AgentManager @@ -140,6 +141,10 @@ class AIOS_Shell: return False ComputeKernel.get_instance().add_compute_node(open_ai_node) + nodes = ComputeNodeConfig.get_instance().initial() + for node in nodes: + await node.start() + ComputeKernel.get_instance().add_compute_node(node) if await AIStorage.get_instance().is_feature_enable("llama"): llama_ai_node = LocalLlama_ComputeNode() @@ -355,6 +360,54 @@ class AIOS_Shell: journals = [str(journal) for journal in KnowledgePipline.get_instance().get_latest_journals(topn)] print_formatted_text("\r\n".join(journals)) + if sub_cmd == "query": + if len(args) < 2: + return show_text + prompt = AgentPrompt() + prompt.messages.append({"role": "user", "content":" ".join(args[1:])}) + result = await KnowledgeBase().query_prompt(prompt) + print_formatted_text(result.as_str()) + + async def handle_node_commands(self, args): + show_text = FormattedText([("class:title", "sub command not support!\n" + "/node add llama $model_name $url\n" + "/node rm llama $model_name $url\n" + "/node list\n")]) + if len(args) < 1: + return show_text + sub_cmd = args[0] + if sub_cmd == "add": + if len(args) < 2: + return show_text + if args[1] == "llama": + if len(args) < 4: + return show_text + + model_name = args[2] + url = args[3] + ComputeNodeConfig.get_instance().add_node("llama", url, model_name) + ComputeNodeConfig.get_instance().save() + node = LocalLlama_ComputeNode(url, model_name) + node.start() + ComputeKernel.get_instance().add_compute_node(node) + else: + return show_text + elif sub_cmd == "rm": + if len(args) < 2: + return show_text + if args[1] == "llama": + if len(args) < 4: + return show_text + + model_name = args[3] + url = args[4] + ComputeNodeConfig.get_instance().remove_node("llama", url, model_name) + ComputeNodeConfig.get_instance().save() + else: + return show_text + elif sub_cmd == "list": + print_formatted_text(ComputeNodeConfig.get_instance().list()) + async def call_func(self,func_name, args): match func_name: case 'send': @@ -480,6 +533,8 @@ class AIOS_Shell: format_texts.append(("",f"\n-------------------\n")) return FormattedText(format_texts) return FormattedText([("class:title", f"chatsession not found")]) + case 'node': + return await self.handle_node_commands(args) case 'exit': os._exit(0) case 'help': @@ -668,6 +723,9 @@ async def main(): '/enable $feature', '/disable $feature', '/list_config', + '/node add llama $model_name $url', + '/node rm llama $model_name $url', + '/node list', '/show', '/exit', '/help'], ignore_case=True) From 2bfa0995b1d7c14df40ece21693fcc2f6c89b3e6 Mon Sep 17 00:00:00 2001 From: streetycat <305190374@qq.com> Date: Thu, 28 Sep 2023 10:27:59 +0000 Subject: [PATCH 5/6] remove await for synchronous method --- rootfs/agents/math_teacher/agent.toml | 2 +- src/service/aios_shell/aios_shell.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/rootfs/agents/math_teacher/agent.toml b/rootfs/agents/math_teacher/agent.toml index 646c8e4..f7a29a5 100644 --- a/rootfs/agents/math_teacher/agent.toml +++ b/rootfs/agents/math_teacher/agent.toml @@ -1,6 +1,6 @@ instance_id = "math_teacher" fullname = "the one" -llm_model_name = "gpt-4-0613" +llm_model_name = "LLaMA2-70B" [[prompt]] role = "system" content = "你是精通数学的老师" diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index ce447e5..21c6e4a 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -143,13 +143,13 @@ class AIOS_Shell: nodes = ComputeNodeConfig.get_instance().initial() for node in nodes: - await node.start() + node.start() ComputeKernel.get_instance().add_compute_node(node) if await AIStorage.get_instance().is_feature_enable("llama"): llama_ai_node = LocalLlama_ComputeNode() if await llama_ai_node.initial() is True: - await llama_ai_node.start() + llama_ai_node.start() ComputeKernel.get_instance().add_compute_node(llama_ai_node) else: logger.error("llama node initial failed!") From 861096fb0c07aa8b2d8a6a0d310971dab822b735 Mon Sep 17 00:00:00 2001 From: streetycat <305190374@qq.com> Date: Thu, 28 Sep 2023 10:30:07 +0000 Subject: [PATCH 6/6] recover the for the agent --- rootfs/agents/math_teacher/agent.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rootfs/agents/math_teacher/agent.toml b/rootfs/agents/math_teacher/agent.toml index f7a29a5..646c8e4 100644 --- a/rootfs/agents/math_teacher/agent.toml +++ b/rootfs/agents/math_teacher/agent.toml @@ -1,6 +1,6 @@ instance_id = "math_teacher" fullname = "the one" -llm_model_name = "LLaMA2-70B" +llm_model_name = "gpt-4-0613" [[prompt]] role = "system" content = "你是精通数学的老师"