feat: skill memory (#1056)

* feat: skill memory

* feat: split task chunks for skill memories

* fix: refine the returned format from llm and parsing

* feat: add new pack oss

* feat: skill mem pipeline

* feat: fill code

* feat: modify code

* feat: modify code

* feat: async add skill memory

* feat: update ollama version

* feat: get memory return skill memory

* feat: get api add skill mem

* feat: get api add skill mem

* feat: modify env config

* feat: back set oss client

* feat: delete tmp skill code

* feat: process new package import error

* feat: modify oss config

* feat: modiy prompt and add two api

* feat: modify prompt

* feat: modify code

* feat: add logger

* feat: fix bug in memory id

* fix:skill OSS + LOCAL存 zip

* fix:skill OSS + LOCAL存 zip

* fix:skill OSS + LOCAL存 zip

* feat: new code

* fix: fix name error in polardb and related code

* fix: bug in polardb

* feat: optimize skill

* feat: local deploy

---------

Co-authored-by: yuan.wang <yuan.wang@yuanwangdebijibendiannao.local>
Co-authored-by: Wenqiang Wei <wwq38556399@163.com>
Co-authored-by: CCC <15764764+triple-c-individual@user.noreply.gitee.com>

Author: 4 people

src/memos/api/config.py

@@ -821,7 +821,12 @@ def get_product_default_config() -> dict[str, Any]:
                     "oss_config": APIConfig.get_oss_config(),
                     "skills_dir_config": {
                         "skills_oss_dir": os.getenv("SKILLS_OSS_DIR", "skill_memory/"),
-                        "skills_local_dir": os.getenv("SKILLS_LOCAL_DIR", "/tmp/skill_memory/"),
+                        "skills_local_tmp_dir": os.getenv(
+                            "SKILLS_LOCAL_TMP_DIR", "/tmp/skill_memory/"
+                        ),
+                        "skills_local_dir": os.getenv(
+                            "SKILLS_LOCAL_DIR", "/tmp/upload_skill_memory/"
+                        ),
                     },
                 },
             },

src/memos/api/server_api.py

@@ -1,13 +1,18 @@
 import logging
+import os
 
+from dotenv import load_dotenv
 from fastapi import FastAPI, HTTPException
 from fastapi.exceptions import RequestValidationError
+from starlette.staticfiles import StaticFiles
 
 from memos.api.exceptions import APIExceptionHandler
 from memos.api.middleware.request_context import RequestContextMiddleware
 from memos.api.routers.server_router import router as server_router
 
 
+load_dotenv()
+
 # Configure logging
 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
 logger = logging.getLogger(__name__)
@@ -18,6 +23,8 @@
     version="1.0.1",
 )
 
+app.mount("/download", StaticFiles(directory=os.getenv("FILE_LOCAL_PATH")), name="static_mapping")
+
 app.add_middleware(RequestContextMiddleware, source="server_api")
 # Include routers
 app.include_router(server_router)

src/memos/mem_reader/read_skill_memory/process_skill_memory.py

@@ -3,13 +3,16 @@
 import os
 import shutil
 import uuid
+import warnings
 import zipfile
 
 from concurrent.futures import as_completed
 from datetime import datetime
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
+from dotenv import load_dotenv
+
 from memos.context.context import ContextThreadPoolExecutor
 from memos.dependency import require_python_package
 from memos.embedders.base import BaseEmbedder
@@ -36,6 +39,8 @@
 from memos.types import MessageList
 
 
+load_dotenv()
+
 if TYPE_CHECKING:
     from memos.types.general_types import UserContext
 
@@ -653,42 +658,88 @@ def _rewrite_query(task_type: str, messages: MessageList, llm: BaseLLM, rewrite_
     import_name="alibabacloud_oss_v2",
     install_command="pip install alibabacloud-oss-v2",
 )
-def _upload_skills_to_oss(local_file_path: str, oss_file_path: str, client: Any) -> str:
-    import alibabacloud_oss_v2 as oss
-
-    result = client.put_object_from_file(
-        request=oss.PutObjectRequest(
-            bucket=os.getenv("OSS_BUCKET_NAME"),
-            key=oss_file_path,
-        ),
-        filepath=local_file_path,
-    )
+def _upload_skills(
+    skills_repo_backend: str,
+    skills_oss_dir: dict[str, Any] | None,
+    local_tmp_file_path: str,
+    local_save_file_path: str,
+    client: Any,
+    user_id: str,
+) -> str:
+    if skills_repo_backend == "OSS":
+        zip_filename = Path(local_tmp_file_path).name
+        oss_path = (Path(skills_oss_dir) / user_id / zip_filename).as_posix()
+
+        import alibabacloud_oss_v2 as oss
+
+        result = client.put_object_from_file(
+            request=oss.PutObjectRequest(
+                bucket=os.getenv("OSS_BUCKET_NAME"),
+                key=oss_path,
+            ),
+            filepath=local_tmp_file_path,
+        )
 
-    if result.status_code != 200:
-        logger.warning("[PROCESS_SKILLS] Failed to upload skill to OSS")
-        return ""
+        if result.status_code != 200:
+            logger.warning("[PROCESS_SKILLS] Failed to upload skill to OSS")
+            return ""
+
+        # Construct and return the URL
+        bucket_name = os.getenv("OSS_BUCKET_NAME")
+        endpoint = os.getenv("OSS_ENDPOINT").replace("https://", "").replace("http://", "")
+        url = f"https://{bucket_name}.{endpoint}/{oss_path}"
+        return url
+    else:
+        import sys
+
+        args = sys.argv
+        port = (
+            int(args[args.index("--port") + 1])
+            if "--port" in args and args.index("--port") + 1 < len(args)
+            else "8000"
+        )
 
-    # Construct and return the URL
-    bucket_name = os.getenv("OSS_BUCKET_NAME")
-    endpoint = os.getenv("OSS_ENDPOINT").replace("https://", "").replace("http://", "")
-    url = f"https://{bucket_name}.{endpoint}/{oss_file_path}"
-    return url
+        zip_path = str(local_tmp_file_path)
+        os.makedirs(local_save_file_path, exist_ok=True)
+        file_name = os.path.basename(zip_path)
+        target_full_path = os.path.join(local_save_file_path, file_name)
+        shutil.copy2(zip_path, target_full_path)
+        return f"http://localhost:{port}/download/{file_name}"
 
 
 @require_python_package(
     import_name="alibabacloud_oss_v2",
     install_command="pip install alibabacloud-oss-v2",
 )
-def _delete_skills_from_oss(oss_file_path: str, client: Any) -> Any:
-    import alibabacloud_oss_v2 as oss
-
-    result = client.delete_object(
-        oss.DeleteObjectRequest(
-            bucket=os.getenv("OSS_BUCKET_NAME"),
-            key=oss_file_path,
+def _delete_skills(
+    skills_repo_backend: str,
+    zip_filename: str,
+    client: Any,
+    skills_oss_dir: dict[str, Any] | None,
+    local_save_file_path: str,
+    user_id: str,
+) -> Any:
+    if skills_repo_backend == "OSS":
+        old_path = (Path(skills_oss_dir) / user_id / zip_filename).as_posix()
+        import alibabacloud_oss_v2 as oss
+
+        return client.delete_object(
+            oss.DeleteObjectRequest(
+                bucket=os.getenv("OSS_BUCKET_NAME"),
+                key=old_path,
+            )
         )
-    )
-    return result
+    else:
+        target_full_path = os.path.join(local_save_file_path, zip_filename)
+        target_path = Path(target_full_path)
+        try:
+            if target_path.is_file():
+                target_path.unlink()
+                logger.info(f"本地文件 {target_path} 已成功删除")
+            else:
+                print(f"本地文件 {target_path} 不存在，无需删除")
+        except Exception as e:
+            print(f"删除本地文件时出错：{e}")
 
 
 def _write_skills_to_file(
@@ -698,7 +749,7 @@ def _write_skills_to_file(
     skill_name = skill_memory.get("name", "unnamed_skill").replace(" ", "_").lower()
 
     # Create tmp directory for user if it doesn't exist
-    tmp_dir = Path(skills_dir_config["skills_local_dir"]) / user_id
+    tmp_dir = Path(skills_dir_config["skills_local_tmp_dir"]) / user_id
     tmp_dir.mkdir(parents=True, exist_ok=True)
 
     # Create skill directory directly in tmp_dir
@@ -889,6 +940,54 @@ def create_skill_memory_item(
     return TextualMemoryItem(id=item_id, memory=memory_content, metadata=metadata)
 
 
+def _skill_init(skills_repo_backend, oss_config, skills_dir_config):
+    if skills_repo_backend == "OSS":
+        # Validate required configurations
+        if not oss_config:
+            logger.warning(
+                "[PROCESS_SKILLS] OSS configuration is required for skill memory processing"
+            )
+            return None, None, False
+
+        if not skills_dir_config:
+            logger.warning(
+                "[PROCESS_SKILLS] Skills directory configuration is required for skill memory processing"
+            )
+            return None, None, False
+
+        # Validate skills_dir has required keys
+        required_keys = ["skills_local_tmp_dir", "skills_local_dir", "skills_oss_dir"]
+        missing_keys = [key for key in required_keys if key not in skills_dir_config]
+        if missing_keys:
+            logger.warning(
+                f"[PROCESS_SKILLS] Skills directory configuration missing required keys: {', '.join(missing_keys)}"
+            )
+            return None, None, False
+
+        oss_client = create_oss_client(oss_config)
+        if not oss_client:
+            logger.warning("[PROCESS_SKILLS] Failed to create OSS client")
+            return None, None, False
+        return oss_client, missing_keys, True
+    else:
+        return None, None, True
+
+
+def _get_skill_file_storage_location() -> str:
+    # SKILLS_REPO_BACKEND: Skill 文件保存地址 OSS/LOCAL
+    allowed_backends = {"OSS", "LOCAL"}
+    raw_backend = os.getenv("SKILLS_REPO_BACKEND")
+    if raw_backend in allowed_backends:
+        return raw_backend
+    else:
+        warnings.warn(
+            "环境变量【SKILLS_REPO_BACKEND】赋值错误，本次使用 LOCAL 存储 skill",
+            UserWarning,
+            stacklevel=1,
+        )
+        return "LOCAL"
+
+
 def process_skill_memory_fine(
     fast_memory_items: list[TextualMemoryItem],
     info: dict[str, Any],
@@ -902,36 +1001,16 @@ def process_skill_memory_fine(
     complete_skill_memory: bool = True,
     **kwargs,
 ) -> list[TextualMemoryItem]:
-    # Validate required configurations
-    if not oss_config:
-        logger.warning("[PROCESS_SKILLS] OSS configuration is required for skill memory processing")
-        return []
-
-    if not skills_dir_config:
-        logger.warning(
-            "[PROCESS_SKILLS] Skills directory configuration is required for skill memory processing"
-        )
+    skills_repo_backend = _get_skill_file_storage_location()
+    oss_client, missing_keys, flag = _skill_init(skills_repo_backend, oss_config, skills_dir_config)
+    if not flag:
         return []
 
     chat_history = kwargs.get("chat_history")
     if not chat_history or not isinstance(chat_history, list):
         chat_history = []
         logger.warning("[PROCESS_SKILLS] History is None in Skills")
 
-    # Validate skills_dir has required keys
-    required_keys = ["skills_local_dir", "skills_oss_dir"]
-    missing_keys = [key for key in required_keys if key not in skills_dir_config]
-    if missing_keys:
-        logger.warning(
-            f"[PROCESS_SKILLS] Skills directory configuration missing required keys: {', '.join(missing_keys)}"
-        )
-        return []
-
-    oss_client = create_oss_client(oss_config)
-    if not oss_client:
-        logger.warning("[PROCESS_SKILLS] Failed to create OSS client")
-        return []
-
     messages = _reconstruct_messages_from_memory_items(fast_memory_items)
 
     chat_history, messages = _preprocess_extract_messages(chat_history, messages)
@@ -1060,23 +1139,27 @@ def _full_extract():
                 old_memory = old_memories_map.get(old_memory_id)
 
                 if old_memory:
-                    # Get old OSS path from the old memory's metadata
-                    old_oss_path = getattr(old_memory.metadata, "url", None)
+                    # Get old path from the old memory's metadata
+                    old_path = getattr(old_memory.metadata, "url", None)
 
-                    if old_oss_path:
+                    if old_path:
                         try:
                             # delete old skill from OSS
-                            zip_filename = Path(old_oss_path).name
-                            old_oss_path = (
-                                Path(skills_dir_config["skills_oss_dir"]) / user_id / zip_filename
-                            ).as_posix()
-                            _delete_skills_from_oss(old_oss_path, oss_client)
+                            zip_filename = Path(old_path).name
+                            _delete_skills(
+                                skills_repo_backend=skills_repo_backend,
+                                zip_filename=zip_filename,
+                                client=oss_client,
+                                skills_oss_dir=skills_dir_config["skills_oss_dir"],
+                                local_save_file_path=skills_dir_config["skills_local_dir"],
+                                user_id=user_id,
+                            )
                             logger.info(
-                                f"[PROCESS_SKILLS] Deleted old skill from OSS: {old_oss_path}"
+                                f"[PROCESS_SKILLS] Deleted old skill from {skills_repo_backend}: {old_path}"
                             )
                         except Exception as e:
                             logger.warning(
-                                f"[PROCESS_SKILLS] Failed to delete old skill from OSS: {e}"
+                                f"[PROCESS_SKILLS] Failed to delete old skill from {skills_repo_backend}: {e}"
                             )
 
                     # delete old skill from graph db
@@ -1086,24 +1169,23 @@ def _full_extract():
                             f"[PROCESS_SKILLS] Deleted old skill from graph db: {old_memory_id}"
                         )
 
-            # Upload new skill to OSS
+            # Upload new skill
             # Use the same filename as the local zip file
-            zip_filename = Path(zip_path).name
-            oss_path = (
-                Path(skills_dir_config["skills_oss_dir"]) / user_id / zip_filename
-            ).as_posix()
-
-            # _upload_skills_to_oss returns the URL
-            url = _upload_skills_to_oss(
-                local_file_path=str(zip_path), oss_file_path=oss_path, client=oss_client
+            url = _upload_skills(
+                skills_repo_backend=skills_repo_backend,
+                skills_oss_dir=skills_dir_config["skills_oss_dir"],
+                local_tmp_file_path=zip_path,
+                local_save_file_path=skills_dir_config["skills_local_dir"],
+                client=oss_client,
+                user_id=user_id,
             )
 
             # Set URL directly to skill_memory
             skill_memory["url"] = url
 
-            logger.info(f"[PROCESS_SKILLS] Uploaded skill to OSS: {url}")
+            logger.info(f"[PROCESS_SKILLS] Uploaded skill to {skills_repo_backend}: {url}")
         except Exception as e:
-            logger.warning(f"[PROCESS_SKILLS] Error uploading skill to OSS: {e}")
+            logger.warning(f"[PROCESS_SKILLS] Error uploading skill to {skills_repo_backend}: {e}")
             skill_memory["url"] = ""  # Set to empty string if upload fails
         finally:
             # Clean up local files after upload