diff --git a/memory_scope/handler/global_context.py b/memory_scope/chat/global_context.py similarity index 100% rename from memory_scope/handler/global_context.py rename to memory_scope/chat/global_context.py diff --git a/memory_scope/chat/memory_chat.py b/memory_scope/chat/memory_chat.py index 4fe6efbb..69311966 100644 --- a/memory_scope/chat/memory_chat.py +++ b/memory_scope/chat/memory_chat.py @@ -2,8 +2,8 @@ import datetime from typing import List from memory_scope.chat.base_memory_chat import BaseMemoryChat +from memory_scope.chat.global_context import GLOBAL_CONTEXT from memory_scope.enumeration.message_role_enum import MessageRoleEnum -from memory_scope.handler.global_context import GLOBAL_CONTEXT from memory_scope.models.base_model import BaseModel from memory_scope.node.message import Message from memory_scope.prompts.prompt_cn import SYSTEM_PROMPT, MEMORY_PROMPT diff --git a/memory_scope/constants/common_constants.py b/memory_scope/constants/common_constants.py index 10768e24..3706e39e 100644 --- a/memory_scope/constants/common_constants.py +++ b/memory_scope/constants/common_constants.py @@ -1,6 +1,3 @@ -from enumeration.dash_api_enum import DashApiEnum -from enumeration.env_type import EnvType - APP_ENV = "APP_ENV" PIPELINE = "pipeline" @@ -82,37 +79,3 @@ MAX_WORKERS = "max_workers" TIME_MATCHED = "time_matched" QUERY_KEYWORDS = "query_keywords" - -DASH_ENV_URL_DICT = { - EnvType.DAILY: "https://dashscope.aliyuncs.com", - # EnvType.PRE: "https://dashscope.aliyuncs.com", - EnvType.PRE: "http://nlb-a3gi6od2xpdx16ezde.cn-beijing.nlb.aliyuncs.com", - EnvType.PROD: "http://ep-2zei3b9a7e2e447bd259.epsrv-2zexnj17q1p8mtjwe3dx.cn-beijing.privatelink.aliyuncs.com", -} - -DASH_API_URL_DICT = { - DashApiEnum.GENERATION: "/api/v1/services/aigc/text-generation/generation", - DashApiEnum.EMBEDDING: "/api/v1/services/embeddings/text-embedding/text-embedding", - DashApiEnum.RERANK: "/api/v1/services/rerank/text-rerank/text-rerank", -} - -ES_ENV_URL_DICT = { - EnvType.DAILY: "http://es-cn-lr53pmrna0002pffb.public.elasticsearch.aliyuncs.com:9200", - EnvType.PRE: "http://ep-bp1i04ae830e377a26a4.epsrv-bp15vzbd1o3umr1girls.cn-hangzhou.privatelink.aliyuncs.com:9200", - EnvType.PROD: "http://ep-2zeibdbbe2904414e741.epsrv-2zet33kwqg8bphgmm36f.cn-beijing.privatelink.aliyuncs.com:9200", -} - -WEEKDAYS = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"] - -DATATIME_WORD_LIST = ["天", "周", "月", "年", "星期", "点", "分钟", "小时", "秒", "上午", "下午", "早上", "早晨", - "晚上", "中午", "日", "夜", "清晨", "傍晚", "凌晨", "岁"] - -TIME_FORMAT_V1 = "{year}年{month}月{day}日{weekday}{hour}点" - -DATATIME_KEY_MAP = { - "年": "year", - "月": "month", - "日": "day", - "周": "week", - "星期几": "weekday", -} diff --git a/memory_scope/handler/init_handler.py b/memory_scope/handler/init_handler.py deleted file mode 100644 index fa1979de..00000000 --- a/memory_scope/handler/init_handler.py +++ /dev/null @@ -1,90 +0,0 @@ -import json -import os -from concurrent.futures import ThreadPoolExecutor -from typing import Dict, Any - -from memory_scope.db.base_db_client import BaseDBClient -from memory_scope.models.base_model import BaseModel -from memory_scope.monitor.base_monitor import BaseMonitor -from memory_scope.utils.tool_functions import init_instance_by_config_v2 -from memory_scope.worker.base_worker import BaseWorker - - -class InitHandler(object): - - def __init__(self, path: str): - self.path: str = path - self.config_name: str = os.path.basename(path) - self.config_base_dir: str = os.path.dirname(path) - - self.config: dict = {} - self.global_configs: Dict[str, Any] = {} - self.worker_dict: Dict[str, BaseWorker] = {} - self.model_dict: Dict[str, BaseModel] = {} - self.db_client: BaseDBClient | None = None - self.monitor: BaseMonitor | None = None - self.thread_pool: ThreadPoolExecutor | None = None - - self.worker_base_dir: str = "" - self.model_base_dir: str = "" - self.db_base_dir: str = "" - self.minitor_base_dir: str = "" - - self.retrieve_pipeline: str = "" - self.summary_short_pipeline: str = "" - self.summary_long_pipeline: str = "" - - def init(self): - with open(self.path) as f: - self.config = json.load(f) - - self.retrieve_pipeline = self.config["pipelines"]["retrieve"] - self.summary_short_pipeline = self.config["pipelines"]["summary_short"] - self.summary_long_pipeline = self.config["pipelines"]["summary_long"] - self.global_configs = self.config["global"] - self.set_global_config() - - self.init_workers(self.config["workers"]) - self.init_db(self.config["db"]) - self.init_chat_model(self.config["chat_model"]) - self.init_monitor(self.config["monitor"]) - - def set_global_config(self): - """set global_configs & set apikey into env - """ - self.worker_base_dir = self.global_configs["worker_base_dir"] - self.model_base_dir = self.global_configs["model_base_dir"] - self.db_base_dir = self.global_configs["db_base_dir"] - self.minitor_base_dir = self.global_configs["minitor_base_dir"] - self.thread_pool = ThreadPoolExecutor(max_workers=int(self.global_configs["max_workers"])) - # TODO @ sen - - def init_workers(self, worker_config_name: str): - """ load worker config & init workers - """ - with open(os.path.join(self.config_base_dir, worker_config_name)) as f: - worker_config_dict = json.load(f) - - for worker_name, worker_config in worker_config_dict.items(): - if worker_name in self.worker_dict: - raise RuntimeError(f"worker_name={worker_name} is repeated!") - - self.worker_dict[worker_name] = init_instance_by_config_v2(worker_config, - default_clazz_path=self.worker_base_dir, - suffix_name="worker", - **self.global_configs) - - self.init_model(worker_config.get("embedding_model")) - self.init_model(worker_config.get("generation_model")) - self.init_model(worker_config.get("rank_model")) - - - def init_db(self, db_config: dict): - self.db_client = init_instance_by_config_v2(db_config, default_clazz_path=self.db_base_dir) - - def init_chat_model(self, chat_model_config: dict): - chat_model_name = chat_model_config["name"] - self.init_model(chat_model_name) - - def init_monitor(self, monitor_config: dict): - self.monitor = init_instance_by_config_v2(monitor_config, default_clazz_path=self.db_base_dir) diff --git a/memory_scope/pipeline/__init__.py b/memory_scope/pipeline/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/memory_scope/pipeline/memory.py b/memory_scope/pipeline/memory.py deleted file mode 100644 index 062a9786..00000000 --- a/memory_scope/pipeline/memory.py +++ /dev/null @@ -1,18 +0,0 @@ -from typing import List, Dict - -from pydantic import Field, BaseModel - -from node.message import Message -from node.user_attribute import UserAttribute - -class MemoryServiceRequestModel(BaseModel): - user: UserConfig = None - - messages: List[Message] = Field(..., - description="summary: 多轮对话的list,默认按照时间正序，最后一条是最新的; retrieve: 最后一条是query") - - user_profile: List[UserAttribute] = Field([], description="user_profile") - - ext_info: Dict[str, str] = Field({}, description="extra information") - - extra_user_attrs: List = [] \ No newline at end of file diff --git a/memory_scope/pipeline/memory_service.py b/memory_scope/pipeline/memory_service.py deleted file mode 100644 index 3a505a02..00000000 --- a/memory_scope/pipeline/memory_service.py +++ /dev/null @@ -1,151 +0,0 @@ -import json -import re -import time -from concurrent.futures import ThreadPoolExecutor, as_completed -from importlib import import_module -from itertools import zip_longest -from typing import Dict, Any - -from worker.base_worker import BaseWorker -from utils.context_handler import ContextHandler -from utils.logger import Logger -from utils.timer import timer, Timer -from common.tool_functions import under_line_to_hump -from constants import common_constants -from constants.common_constants import RESPONSE_EXT_INFO, MAX_WORKERS, PIPELINE -from enumeration.memory_method_enum import MemoryMethodEnum -from pipeline.memory import MemoryServiceRequestModel -from cli.cli_config import C -from utils.tool_functions import init_instance_by_config - - -class MemoryService(object): - def __init__(self, method: MemoryMethodEnum): - self.method = method - self.context_handler = ContextHandler() - - # 线程池 - self.thread_pool = ThreadPoolExecutor(max_workers=C.thread_pool_max_count) - - # 全部初始化的worker - self.worker_dict: Dict[str, BaseWorker] = {} - - # 日志 - self.logger: Logger = Logger.get_memory_logger() - - # 初始化pipeline - self.pipeline_list = self.get_pipeline() - self.print_and_init_worker(self.pipeline_list) - - def get_worker(self, worker_name: str, is_multi_thread: bool = False) -> BaseWorker: - return init_instance_by_config( - config = C.worker.get(worker_name), - try_kwargs={ - "is_multi_thread": is_multi_thread, - "thread_pool": self.thread_pool - } - ) - - def worker_run(self, worker_list: list[str]) -> bool: - for worker_name in worker_list: - worker = self.worker_dict[worker_name] - # 执行子类实现的_run函数 - worker.run() - # 保存worker的运行信息 - self.run_infos.append(worker.run_info_dict) - # 结束pipeline - if not worker.continue_run: - return False - return True - - @timer - def print_and_init_worker(self, pipeline_list: list[list]): - self.logger.info("----- Pipeline Begin -----") - i: int = 0 - for pipeline_part in pipeline_list: - if len(pipeline_part) == 1: - for w in pipeline_part[0]: - self.logger.info(f"stage{i}: {w}") - self.worker_dict[w] = self.get_worker(w) - i += 1 - else: - for w_zip in zip_longest(*pipeline_part, fillvalue="-"): - self.logger.info(f"stage{i}: {' | '.join(w_zip)}") - i += 1 - for w in w_zip: - if w == "-": - continue - self.worker_dict[w] = self.get_worker(w, is_multi_thread=True) - self.logger.info("----- Pipeline End -----") - - def get_context(self, key: str, default=None) -> Any: - return self.context_handler.get_context(key, default) - - def flush(self, request: MemoryServiceRequestModel): - # 全局上下文，worker之间交换参数和变量 - self.context_handler.flush() - - # 运行信息 - self.run_infos = [] - self.context_handler.set_context(common_constants.REQUEST, request) - for pipeline_part in self.pipeline_list: - pipeline_part.flush(self.context_handler) - - @timer - def get_pipeline(self) -> list[list]: - pipeline_str = C.pipeline.get(self.method) - self.logger.info(f"pipeline={pipeline_str}") - - # re-match e.g., [a|b],c,[d,e,f|g,h],j - pattern = r'(\[[^\]]*\]|[^,]+)' - pipeline_split = re.findall(pattern, pipeline_str) - - pipeline_list = [] - for pipeline_part in pipeline_split: - # e.g., [d,e,f|g,h] - pipeline_part = pipeline_part.strip() - if '[' in pipeline_part or ']' in pipeline_part: - pipeline_part = pipeline_part.replace('[', '').replace(']', '') - - # e.g., ["d,e,f", "g,h"] - line_split = [x.strip() for x in pipeline_part.split("|") if x] - if len(line_split) <= 0: - continue - - # e.g., ["d","e","f"] - pipeline_list.append([x.split(",") for x in line_split]) - - return pipeline_list - - def run(self): - # run workers in multi threads - with self.thread_pool, Timer("ALL_PIPELINE"): - for pipeline_part in self.pipeline_list: - if len(pipeline_part) == 1: - if not self.worker_run(pipeline_part[0]): - break - elif self.max_workers == 1: - for worker_list in pipeline_part: - self.worker_run(worker_list) - else: - t_list = [] - for worker_list in pipeline_part: - time.sleep(0.001) - t_list.append(self.thread_pool.submit(self.worker_run, worker_list)) - - flag = True - for future in as_completed(t_list): - if not future.result(): - flag = False - break - if not flag: - break - - # 获取ext_info - ext_info = self.get_context(RESPONSE_EXT_INFO) - if ext_info is None: - ext_info = {} - self.context_handler.set_context(RESPONSE_EXT_INFO, ext_info) - - # 保存 run_info_list - ext_info["run_infos"] = json.dumps(self.run_infos, ensure_ascii=False) diff --git a/memory_scope/pipeline/operator.py b/memory_scope/pipeline/operator.py deleted file mode 100644 index ac1b317b..00000000 --- a/memory_scope/pipeline/operator.py +++ /dev/null @@ -1,17 +0,0 @@ -# -*- coding: utf-8 -*- -"""A common base class for Pipeline""" -from abc import ABC -from abc import abstractmethod - - -class Operator(ABC): - """ - Abstract base class `Operator` defines a protocol for classes that - implement callable behavior. - The class is designed to be subclassed with an overridden `__call__` - method that specifies the execution logic for the operator. - """ - - @abstractmethod - def __call__(self) -> None: - """Calling function"""