diff --git a/lpm_kernel/api/__init__.py b/lpm_kernel/api/__init__.py
index 792467bd..3c3c78f6 100644
--- a/lpm_kernel/api/__init__.py
+++ b/lpm_kernel/api/__init__.py
@@ -11,6 +11,7 @@
 from .domains.kernel2.routes_talk import talk_bp
 from .domains.user_llm_config.routes import user_llm_config_bp
 from .domains.space.space_routes import space_bp
+from .domains.backup.routes import backup_bp # Add import for backup blueprint
 
 def init_routes(app: Flask):
     """Initialize all route blueprints"""
@@ -27,6 +28,7 @@ def init_routes(app: Flask):
     app.register_blueprint(space_bp)
     app.register_blueprint(talk_bp)
     app.register_blueprint(user_llm_config_bp)
+    app.register_blueprint(backup_bp) # Register backup blueprint
     app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0
     app.config['TEMPLATES_AUTO_RELOAD'] = True
     # Disable response buffering
diff --git a/lpm_kernel/api/domains/backup/__init__.py b/lpm_kernel/api/domains/backup/__init__.py
new file mode 100644
index 00000000..021620f3
--- /dev/null
+++ b/lpm_kernel/api/domains/backup/__init__.py
@@ -0,0 +1 @@
+from .routes import *
\ No newline at end of file
diff --git a/lpm_kernel/api/domains/backup/routes.py b/lpm_kernel/api/domains/backup/routes.py
new file mode 100644
index 00000000..21e3c42d
--- /dev/null
+++ b/lpm_kernel/api/domains/backup/routes.py
@@ -0,0 +1,63 @@
+from flask import Blueprint, request, jsonify, current_app
+from lpm_kernel.backup.backup_service import BackupService
+from lpm_kernel.configs.config import Config # Import Config 
+
+backup_bp = Blueprint('backup', __name__, url_prefix='/api/backups')
+
+# Instantiate BackupService using the application config
+# Note: This assumes Config is a singleton or can be instantiated here.
+# A better approach might involve dependency injection or accessing config via current_app.
+config = Config.from_env() # Attempt to load config
+backup_service = BackupService(config=config) # Pass the loaded config
+
+@backup_bp.route('/', methods=['POST'])
+def create_backup_route():
+    """API endpoint to manually trigger a backup."""
+    try:
+        data = request.get_json() or {}
+        description = data.get('description')
+        result = backup_service.create_backup(description=description)
+        if result:
+            return jsonify(result), 201
+        else:
+            return jsonify({"status": "error", "message": "Failed to create backup"}), 500
+    except Exception as e:
+        current_app.logger.error(f"Backup creation failed: {str(e)}")
+        return jsonify({"status": "error", "message": f"Backup creation failed: {str(e)}"}), 500
+
+@backup_bp.route('/', methods=['GET'])
+def list_backups_route():
+    """API endpoint to list available backups."""
+    try:
+        backups = backup_service.list_backups()
+        return jsonify({"status": "success", "backups": backups}), 200
+    except Exception as e:
+        current_app.logger.error(f"Listing backups failed: {str(e)}")
+        return jsonify({"status": "error", "message": f"Listing backups failed: {str(e)}"}), 500
+
+@backup_bp.route('/<string:backup_id>/restore', methods=['POST'])
+def restore_backup_route(backup_id):
+    """API endpoint to restore data from a specific backup."""
+    try:
+        result = backup_service.restore_backup(backup_id)
+        if result.get('status') == 'error':
+            return jsonify(result), 404 if 'not found' in result.get('message', '').lower() else 500
+        return jsonify(result), 200
+    except Exception as e:
+        return jsonify({"status": "error", "message": f"Restore failed: {str(e)}"}), 500
+
+@backup_bp.route('/<string:backup_id>', methods=['DELETE'])
+def delete_backup_route(backup_id):
+    """API endpoint to delete a specific backup."""
+    try:
+        result = backup_service.delete_backup(backup_id)
+        if result.get('status') == 'error':
+            return jsonify(result), 404 if 'not found' in result.get('message', '').lower() else 500
+        return jsonify(result), 200
+    except Exception as e:
+        current_app.logger.error(f"Backup deletion failed: {str(e)}")
+        return jsonify({"status": "error", "message": f"Backup deletion failed: {str(e)}"}), 500
+
+# TODO: Add error handling
+# TODO: Add more robust error handling (e.g., specific error codes)
+# TODO: Consider dependency injection for BackupService
\ No newline at end of file
diff --git a/lpm_kernel/api/domains/loads/load_service.py b/lpm_kernel/api/domains/loads/load_service.py
index c3069d52..9e7e6fb8 100644
--- a/lpm_kernel/api/domains/loads/load_service.py
+++ b/lpm_kernel/api/domains/loads/load_service.py
@@ -423,28 +423,27 @@ def _clean_data_directories() -> None:
 
     @staticmethod
     def _reset_training_progress() -> None:
-        """Reset training progress objects in memory"""
+        """重置内存中的训练进度对象"""
         try:
-            # Get all possible training progress file patterns
+            from lpm_kernel.api.domains.trainprocess.trainprocess_service import TrainProcessService
+            import os
+            # 获取所有可能的训练进度文件模式
             base_dir = os.getenv('LOCAL_BASE_DIR', '.')
             progress_dir = os.path.join(base_dir, 'data', 'progress')
             if os.path.exists(progress_dir):
                 for file in os.listdir(progress_dir):
                     if file.startswith('trainprocess_progress_'):
-                        # Extract model name
+                        # 提取模型名称
                         model_name = file.replace('trainprocess_progress_', '').replace('.json', '')
-                        # Create a new service instance for each model and reset progress
+                        # 为每个模型创建新的服务实例并重置进度
                         train_service = TrainProcessService(current_model_name=model_name)
                         train_service.progress.reset_progress()
                         logger.info(f"Reset training progress for model: {model_name}")
-            
-            # Reset default training progress
+            # 重置默认训练进度
             default_train_service = TrainProcessService.get_instance()
             if default_train_service is not None:
                 default_train_service.progress.reset_progress()
-            
             logger.info("Reset default training progress")
-            
         except Exception as e:
             logger.error(f"Failed to reset training progress objects: {str(e)}")
 
diff --git a/lpm_kernel/api/domains/trainprocess/trainprocess_service.py b/lpm_kernel/api/domains/trainprocess/trainprocess_service.py
index f05850b2..99c237a2 100644
--- a/lpm_kernel/api/domains/trainprocess/trainprocess_service.py
+++ b/lpm_kernel/api/domains/trainprocess/trainprocess_service.py
@@ -1,40 +1,46 @@
 import os
 import re
 import time
+import threading
+import gc
+import subprocess
 import psutil
-from typing import Optional, Dict
+import json
+from enum import Enum
+from typing import Dict, List, Optional
+
+from lpm_kernel.configs.config import Config
+from lpm_kernel.configs.logging import get_train_process_logger, TRAIN_LOG_FILE
 from lpm_kernel.L1.utils import save_true_topics
 from lpm_kernel.L1.serializers import NotesStorage
 from lpm_kernel.kernel.note_service import NoteService
 from lpm_kernel.L2.l2_generator import L2Generator
 from lpm_kernel.L2.utils import save_hf_model
 from lpm_kernel.api.common.responses import APIResponse
+from lpm_kernel.api.common.script_executor import ScriptExecutor
 from lpm_kernel.api.domains.loads.services import LoadService
+from lpm_kernel.api.domains.trainprocess.progress_enum import Status
+from lpm_kernel.api.domains.trainprocess.train_progress import TrainProgress
+from lpm_kernel.api.domains.trainprocess.process_step import ProcessStep
+from lpm_kernel.api.domains.trainprocess.progress_holder import TrainProgressHolder
+from lpm_kernel.api.domains.kernel.routes import store_l1_data
+from lpm_kernel.train.training_params_manager import TrainingParamsManager
+from lpm_kernel.common.repository.database_session import DatabaseSession
+from lpm_kernel.backup.auto_backup import AutoBackupManager
 from lpm_kernel.kernel.chunk_service import ChunkService
 from lpm_kernel.kernel.l1.l1_manager import (
     extract_notes_from_documents,
     document_service,
     get_latest_status_bio,
     get_latest_global_bio,
+    generate_l1_from_l0,
 )
-from lpm_kernel.api.common.script_executor import ScriptExecutor
-from lpm_kernel.configs.config import Config
 from lpm_kernel.file_data.chunker import DocumentChunker
-from lpm_kernel.kernel.l1.l1_manager import generate_l1_from_l0
-import threading
-from lpm_kernel.api.domains.trainprocess.progress_enum import Status
-from lpm_kernel.api.domains.trainprocess.process_step import ProcessStep
-from lpm_kernel.api.domains.trainprocess.progress_holder import TrainProgressHolder
-from lpm_kernel.api.domains.trainprocess.training_params_manager import TrainingParamsManager
 from lpm_kernel.models.l1 import L1Bio, L1Shade
-from lpm_kernel.common.repository.database_session import DatabaseSession
-from lpm_kernel.api.domains.kernel.routes import store_l1_data
 from lpm_kernel.api.domains.trainprocess.L1_exposure_manager import output_files, query_l1_version_data, read_file_content
-import gc
-import subprocess
-from lpm_kernel.configs.logging import get_train_process_logger, TRAIN_LOG_FILE
 logger = get_train_process_logger()
 
+
 class TrainProcessService:
     """Training process service (singleton pattern)"""
     
@@ -46,9 +52,9 @@ def __new__(cls, *args, **kwargs):
             cls._instance = super().__new__(cls)
         return cls._instance
 
-    def __init__(self, current_model_name: str):
-        if current_model_name is None:
-            raise ValueError("current_model_name cannot be None")
+    def __init__(self, current_model_name: str = None):
+        if current_model_name is None and not self._initialized:
+            raise ValueError("current_model_name cannot be None when initializing")
             
         if not self._initialized:
             # Generate a unique progress file name based on model name
@@ -73,7 +79,7 @@ def __init__(self, current_model_name: str):
             self.l2_data_prepared = False
         
         # Update model name and progress instance if model name changes
-        if current_model_name != self.model_name:
+        if current_model_name is not None and current_model_name != self.model_name:
             self.model_name = current_model_name
             # Create new progress instance with updated progress file name
             self.progress = TrainProgressHolder(current_model_name)
@@ -426,6 +432,189 @@ def augment_content_retention(self) -> bool:
                 self.l2_data["config_path"]
             )
             l2_generator.merge_json_files(self.l2_data["data_output_base_dir"])
+<<<<<<< HEAD
+            
+            self.progress.mark_step_status(ProcessStep.AUGMENT_CONTENT_RETENTION, Status.COMPLETED)
+            logger.info("Content retention augmentation completed successfully")
+            return True
+            
+        except Exception as e:
+            logger.error(f"Augment content retention failed: {str(e)}")
+            self.progress.mark_step_status(ProcessStep.AUGMENT_CONTENT_RETENTION, Status.FAILED)
+            return False
+
+    def _prepare_l2_data(self):
+        """Prepare L2 data for training"""
+        if self.l2_data_prepared:
+            logger.info("L2 data already prepared, skipping preparation")
+            return
+            
+        try:
+            logger.info("Preparing L2 data...")
+            
+            # Get notes from database
+            note_service = NoteService()
+            notes = note_service.get_all_notes()
+            
+            # Get basic info
+            status_bio = get_latest_status_bio()
+            global_bio = get_latest_global_bio()
+            
+            # Combine into basic info dictionary
+            basic_info = {
+                "status_bio": status_bio,
+                "global_bio": global_bio
+            }
+            
+            # Create output directory
+            data_output_base_dir = os.path.join(os.getcwd(), "data", "l2_data")
+            os.makedirs(data_output_base_dir, exist_ok=True)
+            
+            # Set paths for L2 data
+            topics_path = os.path.join(data_output_base_dir, "topics.json")
+            entitys_path = os.path.join(data_output_base_dir, "entitys.json")
+            graph_path = os.path.join(data_output_base_dir, "graph.json")
+            config_path = os.path.join(data_output_base_dir, "config.json")
+            
+            # Store in l2_data dictionary
+            self.l2_data["notes"] = notes
+            self.l2_data["basic_info"] = basic_info
+            self.l2_data["data_output_base_dir"] = data_output_base_dir
+            self.l2_data["topics_path"] = topics_path
+            self.l2_data["entitys_path"] = entitys_path
+            self.l2_data["graph_path"] = graph_path
+            self.l2_data["config_path"] = config_path
+            
+            self.l2_data_prepared = True
+            logger.info("L2 data preparation completed successfully")
+            
+        except Exception as e:
+            logger.error(f"L2 data preparation failed: {str(e)}")
+            self._cleanup_resources()
+            raise
+
+    def _monitor_model_download(self):
+        """Monitor model download progress"""
+        try:
+            logger.info("Starting model download monitoring...")
+            
+            # Check every 5 seconds
+            while True:
+                # Sleep first to give download time to start
+                time.sleep(5)
+                
+                # Check if download is still in progress
+                if self.progress.get_step_status(ProcessStep.MODEL_DOWNLOAD) != Status.IN_PROGRESS:
+                    logger.info("Model download monitoring stopped - download completed or failed")
+                    break
+                    
+                # Log current memory usage
+                process = psutil.Process(os.getpid())
+                memory_info = process.memory_info()
+                logger.info(f"Current memory usage during download: {memory_info.rss / 1024 / 1024:.2f} MB")
+                
+        except Exception as e:
+            logger.error(f"Error in model download monitoring: {str(e)}")
+
+    def check_training_condition(self) -> bool:
+        """Check if training conditions are met"""
+        try:
+            # Check if documents exist
+            documents = self.list_documents()
+            if not documents:
+                logger.error("No documents found, training cannot proceed")
+                return False
+                
+            # Check if model name is valid
+            if not self.model_name:
+                logger.error("Model name is not set, training cannot proceed")
+                return False
+                
+            return True
+        except Exception as e:
+            logger.error(f"Error checking training conditions: {str(e)}")
+            return False
+            
+    def reset_progress(self):
+        """Reset training progress"""
+        try:
+            logger.info("Resetting training progress...")
+            self.progress.reset_progress()
+            logger.info("Training progress reset successfully")
+        except Exception as e:
+            logger.error(f"Error resetting training progress: {str(e)}")
+
+    def start_process(self):
+        """Start the training process"""
+        try:
+            logger.info("Starting training process...")
+            
+            # Reset stop flag
+            self.is_stopped = False
+            
+            # Reset progress
+            self.reset_progress()
+            
+            # Check training conditions
+            if not self.check_training_condition():
+                logger.error("Training conditions not met, aborting process")
+                return False
+                
+            # Start the process
+            self.progress.mark_status(Status.IN_PROGRESS)
+            
+            # Execute each step in sequence
+            steps = [
+                ("List documents", self.list_documents),
+                ("Generate document embeddings", self.generate_document_embeddings),
+                ("Process chunks", self.process_chunks),
+                ("Generate chunk embeddings", self.chunk_embedding),
+                ("Extract dimensional topics", self.extract_dimensional_topics),
+                ("Generate biography", self.generate_biography),
+                ("Download model", self.model_download),
+                ("Map entity network", self.map_your_entity_network),
+                ("Decode preference patterns", self.decode_preference_patterns),
+                ("Reinforce identity", self.reinforce_identity),
+                ("Augment content retention", self.augment_content_retention),
+            ]
+            
+            for step_name, step_func in steps:
+                if self.is_stopped:
+                    logger.info("Process stopped by user")
+                    self.progress.mark_status(Status.STOPPED)
+                    return False
+                    
+                logger.info(f"Executing step: {step_name}")
+                self.current_step = step_name
+                
+                # Execute step
+                success = step_func()
+                
+                if not success:
+                    logger.error(f"Step '{step_name}' failed, stopping process")
+                    self.progress.mark_status(Status.FAILED)
+                    return False
+                    
+            # All steps completed successfully
+            logger.info("All steps completed successfully")
+            self.progress.mark_status(Status.COMPLETED)
+            return True
+            
+        except Exception as e:
+            logger.error(f"Error in training process: {str(e)}")
+            self.progress.mark_status(Status.FAILED)
+            return False
+            
+    def stop_process(self):
+        """Stop the training process"""
+        try:
+            logger.info("Stopping training process...")
+            self.is_stopped = True
+            logger.info("Training process stop flag set")
+            return True
+        except Exception as e:
+            logger.error(f"Error stopping training process: {str(e)}")
+=======
             # Mark step as completed
             logger.info("Content retention augmentation completed successfully")
             self.progress.mark_step_status(ProcessStep.AUGMENT_CONTENT_RETENTION, Status.COMPLETED)
@@ -1055,6 +1244,16 @@ def start_process(self) -> bool:
             # Store the current process PID
             self.current_pid = os.getpid()  # Store the PID
             logger.info(f"Training process started with PID: {self.current_pid}")
+            
+            # Create pre-training backup
+            logger.info("Creating pre-training backup")
+            auto_backup_manager = AutoBackupManager()
+            auto_backup_manager.create_pre_training_backup(model_name=self.model_name)
+            
+            # Start periodic backup
+            logger.info("Starting periodic backup")
+            auto_backup_manager.start_periodic_backup(model_name=self.model_name)
+            
             # Get the ordered list of all steps
             ordered_steps = ProcessStep.get_ordered_steps()
 
@@ -1157,6 +1356,11 @@ def stop_process(self):
             if self.current_step == ProcessStep.TRAIN:
                 self.progress.mark_step_status(ProcessStep.TRAIN, Status.SUSPENDED)
             
+            # Stop periodic backup when training is manually stopped
+            logger.info("Stopping periodic backup due to manual stop")
+            auto_backup_manager = AutoBackupManager()
+            auto_backup_manager.stop_periodic_backup()
+            
             # First check if we have the current process PID
             if not hasattr(self, 'current_pid') or not self.current_pid:
                 logger.info("No active process PID found")
@@ -1210,4 +1414,5 @@ def stop_process(self):
                 
         except Exception as e:
             logger.error(f"Error stopping training process: {str(e)}", exc_info=True)
+>>>>>>> upstream/develop
             return False
\ No newline at end of file
diff --git a/lpm_kernel/backup/__init__.py b/lpm_kernel/backup/__init__.py
new file mode 100644
index 00000000..fb05ff58
--- /dev/null
+++ b/lpm_kernel/backup/__init__.py
@@ -0,0 +1,5 @@
+# Backup module
+from .backup_service import BackupService
+from .auto_backup import AutoBackupManager
+
+__all__ = ['BackupService', 'AutoBackupManager']
\ No newline at end of file
diff --git a/lpm_kernel/backup/auto_backup.py b/lpm_kernel/backup/auto_backup.py
new file mode 100644
index 00000000..d0af28bd
--- /dev/null
+++ b/lpm_kernel/backup/auto_backup.py
@@ -0,0 +1,189 @@
+import threading
+import time
+import random
+from datetime import datetime, timedelta
+from pathlib import Path
+import json
+import os
+
+from ..common.logging import logger
+from .backup_service import BackupService
+from ..configs.config import Config
+
+class AutoBackupManager:
+    """Manages automatic backups during training process"""
+    
+    _instance = None
+    _initialized = False
+    _lock = threading.Lock()
+    
+    def __new__(cls, *args, **kwargs):
+        if cls._instance is None:
+            with cls._lock:
+                if cls._instance is None:
+                    cls._instance = super().__new__(cls)
+        return cls._instance
+    
+    def __init__(self, config=None):
+        with self._lock:
+            if not self._initialized:
+                self.config = config or Config.from_env()
+                self.backup_service = BackupService(self.config)
+                self.backup_thread = None
+                self.stop_flag = threading.Event()
+                self.thread_lock = threading.Lock()
+                self.backup_interval = int(self.config.get("AUTO_BACKUP_INTERVAL_MINUTES", "30"))
+                self.max_auto_backups = int(self.config.get("MAX_AUTO_BACKUPS", "5"))
+                self.auto_backup_enabled = self.config.get("AUTO_BACKUP_ENABLED", "true").lower() == "true"
+                self.max_retries = int(self.config.get("BACKUP_MAX_RETRIES", "3"))
+                self.retry_delay = int(self.config.get("BACKUP_RETRY_DELAY_SECONDS", "60"))
+                self._initialized = True
+    
+    def create_pre_training_backup(self, model_name=None):
+        """Creates a backup before training starts"""
+        if not self.auto_backup_enabled:
+            logger.info("Auto backup is disabled, skipping pre-training backup")
+            return None
+            
+        description = f"Pre-training automatic backup{' for ' + model_name if model_name else ''}"
+        logger.info(f"Creating pre-training backup: {description}")
+        return self.backup_service.create_backup(description=description)
+    
+    def start_periodic_backup(self, model_name=None):
+        """Starts a thread that creates periodic backups during training"""
+        if not self.auto_backup_enabled:
+            logger.info("Auto backup is disabled, not starting periodic backups")
+            return
+            
+        # Stop any existing backup thread
+        self.stop_periodic_backup()
+        
+        # Reset stop flag
+        self.stop_flag.clear()
+        
+        # Start new backup thread
+        self.backup_thread = threading.Thread(
+            target=self._periodic_backup_worker,
+            args=(model_name,),
+            daemon=True
+        )
+        self.backup_thread.start()
+        logger.info(f"Started periodic backup thread with interval {self.backup_interval} minutes")
+    
+    def stop_periodic_backup(self):
+        """Stops the periodic backup thread if it's running"""
+        if self.backup_thread and self.backup_thread.is_alive():
+            self.stop_flag.set()
+            self.backup_thread.join(timeout=5)
+            logger.info("Stopped periodic backup thread")
+    
+    def _periodic_backup_worker(self, model_name=None):
+        """Worker function that creates periodic backups"""
+        last_backup_time = None
+        consecutive_failures = 0
+        max_consecutive_failures = 3
+        
+        while not self.stop_flag.is_set():
+            try:
+                current_time = datetime.now()
+                
+                # Check if enough time has passed since last backup
+                if last_backup_time and (current_time - last_backup_time).total_seconds() < self.backup_interval * 60:
+                    time.sleep(60)  # Check every minute
+                    continue
+                
+                # Create backup with retries
+                description = f"Training in-progress automatic backup{' for ' + model_name if model_name else ''}"
+                logger.info(f"Creating periodic backup: {description}")
+                
+                retry_count = 0
+                backup_success = False
+                base_delay = self.retry_delay
+                
+                while retry_count < self.max_retries:
+                    try:
+                        backup_result = self.backup_service.create_backup(
+                            description=description,
+                            tags=["auto", "training", model_name] if model_name else ["auto", "training"]
+                        )
+                        if backup_result:
+                            logger.info("Backup created successfully")
+                            last_backup_time = current_time
+                            consecutive_failures = 0
+                            backup_success = True
+                            break
+                        raise Exception("Backup creation failed")
+                    except Exception as backup_error:
+                        retry_count += 1
+                        error_msg = str(backup_error)
+                        if retry_count < self.max_retries:
+                            # 使用指数退避策略计算下一次重试延迟
+                            current_delay = min(base_delay * (2 ** (retry_count - 1)), 300)  # 最大延迟5分钟
+                            jitter = random.uniform(0, min(current_delay * 0.1, 30))  # 添加随机抖动
+                            retry_delay = current_delay + jitter
+                            logger.warning(f"Backup attempt {retry_count} failed: {error_msg}, retrying in {retry_delay:.1f} seconds...")
+                            time.sleep(retry_delay)
+                        else:
+                            logger.error(f"All backup attempts failed after {self.max_retries} retries: {error_msg}")
+                
+                if not backup_success:
+                    consecutive_failures += 1
+                    if consecutive_failures >= max_consecutive_failures:
+                        logger.critical(f"Stopping automatic backup after {consecutive_failures} consecutive failures")
+                        self.stop_flag.set()
+                        break
+                
+                # Clean up old auto backups if needed
+                self._cleanup_old_auto_backups()
+                
+            except Exception as e:
+                logger.error(f"Error in periodic backup: {e}", exc_info=True)
+                # Sleep a bit before continuing to next cycle
+                time.sleep(60)
+    
+    def _cleanup_old_auto_backups(self):
+        """Removes old automatic backups based on count and size limits"""
+        try:
+            # 获取所有备份
+            backups = self.backup_service.list_backups()
+            
+            # 过滤自动备份
+            auto_backups = [b for b in backups if 
+                          "automatic backup" in b.get("description", "").lower()]
+            
+            # 按时间戳排序（最新的在前）
+            auto_backups.sort(key=lambda x: x.get("timestamp", ""), reverse=True)
+            
+            # 获取配置的大小限制（默认50GB）
+            max_total_size_gb = float(self.config.get("MAX_AUTO_BACKUP_SIZE_GB", "50"))
+            max_total_size_bytes = max_total_size_gb * 1024 * 1024 * 1024
+            
+            # 跟踪已使用的总大小
+            total_size = 0
+            backups_to_keep = []
+            
+            # 首先保留最新的必需备份数量
+            min_backups_to_keep = max(1, min(self.max_auto_backups // 2, 3))  # 至少保留1个，最多保留3个最新备份
+            backups_to_keep.extend(auto_backups[:min_backups_to_keep])
+            total_size = sum(b.get("size_bytes", 0) for b in backups_to_keep)
+            
+            # 处理剩余的备份
+            for backup in auto_backups[min_backups_to_keep:]:
+                backup_size = backup.get("size_bytes", 0)
+                
+                # 如果添加此备份后仍在限制范围内，且未超过最大数量限制，则保留
+                if (total_size + backup_size <= max_total_size_bytes and 
+                    len(backups_to_keep) < self.max_auto_backups):
+                    backups_to_keep.append(backup)
+                    total_size += backup_size
+                else:
+                    # 删除不满足条件的备份
+                    backup_id = backup.get("id")
+                    if backup_id:
+                        logger.info(f"Cleaning up old auto backup: {backup_id} (size: {backup_size/(1024*1024):.2f}MB)")
+                        self.backup_service.delete_backup(backup_id)
+            
+            logger.info(f"Backup cleanup completed. Keeping {len(backups_to_keep)} backups, total size: {total_size/(1024*1024*1024):.2f}GB")
+        
+        except Exception as e:
+            logger.error(f"Error cleaning up old auto backups: {e}", exc_info=True)
\ No newline at end of file
diff --git a/lpm_kernel/backup/backup_integrity.py b/lpm_kernel/backup/backup_integrity.py
new file mode 100644
index 00000000..3ed26b0d
--- /dev/null
+++ b/lpm_kernel/backup/backup_integrity.py
@@ -0,0 +1,163 @@
+import hashlib
+import zlib
+import json
+from pathlib import Path
+from typing import Dict, List, Optional
+from cryptography.fernet import Fernet
+from ..common.logging import logger
+
+class BackupIntegrity:
+    def __init__(self, encryption_key: Optional[str] = None):
+        """Initialize backup integrity checker with optional encryption."""
+        self.encryption_key = encryption_key.encode() if encryption_key else Fernet.generate_key()
+        self.fernet = Fernet(self.encryption_key)
+    
+    def calculate_checksum(self, file_path: Path) -> str:
+        """Calculate SHA-256 checksum of a file."""
+        sha256_hash = hashlib.sha256()
+        with open(file_path, "rb") as f:
+            for byte_block in iter(lambda: f.read(4096), b""):
+                sha256_hash.update(byte_block)
+        return sha256_hash.hexdigest()
+    
+    def compress_file(self, file_path: Path, compressed_path: Optional[Path] = None, compression_level: int = 6) -> Path:
+        """Compress a file using zlib with configurable compression level (1-9)."""
+        if not compressed_path:
+            compressed_path = file_path.with_suffix('.gz')
+        
+        try:
+            from concurrent.futures import ThreadPoolExecutor
+            import threading
+            
+            chunk_size = 1024 * 1024  # 1MB chunks
+            file_lock = threading.Lock()
+            
+            def _compress_chunk(chunk):
+                compressor = zlib.compressobj(level=compression_level)
+                return compressor.compress(chunk) + compressor.flush(zlib.Z_FULL_FLUSH)
+            
+            with open(file_path, 'rb') as f_in, \
+                 open(compressed_path, 'wb') as f_out, \
+                 ThreadPoolExecutor() as executor:
+                
+                futures = []
+                while True:
+                    chunk = f_in.read(chunk_size)
+                    if not chunk:
+                        break
+                    futures.append(executor.submit(_compress_chunk, chunk))
+                
+                # 按顺序收集结果
+                for future in futures:
+                    compressed_data = future.result()
+                    with file_lock:
+                        f_out.write(compressed_data)
+                
+                # 写入最终flush
+                final_flush = zlib.compressobj(level=compression_level).flush()
+                f_out.write(final_flush)
+            
+            original_size = file_path.stat().st_size
+            compressed_size = compressed_path.stat().st_size
+            compression_ratio = (1 - compressed_size / original_size) * 100
+            logger.info(f"Compressed {file_path.name}: {compression_ratio:.1f}% reduction (Level {compression_level})")
+            
+            return compressed_path
+        except Exception as e:
+            logger.error(f"Error compressing {file_path}: {e}")
+            if compressed_path.exists():
+                compressed_path.unlink()  # 清理失败的压缩文件
+            raise
+    
+    def decompress_file(self, compressed_path: Path, output_path: Optional[Path] = None) -> Path:
+        """Decompress a zlib compressed file."""
+        if not output_path:
+            output_path = compressed_path.with_suffix('')
+        
+        try:
+            with open(compressed_path, 'rb') as f_in:
+                compressed_data = f_in.read()
+            data = zlib.decompress(compressed_data)
+            with open(output_path, 'wb') as f_out:
+                f_out.write(data)
+            return output_path
+        except Exception as e:
+            logger.error(f"Error decompressing {compressed_path}: {e}")
+            raise
+    
+    def encrypt_file(self, file_path: Path, encrypted_path: Optional[Path] = None) -> Path:
+        """Encrypt a file using Fernet symmetric encryption."""
+        if not encrypted_path:
+            encrypted_path = file_path.with_suffix('.enc')
+        
+        try:
+            with open(file_path, 'rb') as f:
+                data = f.read()
+            encrypted_data = self.fernet.encrypt(data)
+            with open(encrypted_path, 'wb') as f:
+                f.write(encrypted_data)
+            return encrypted_path
+        except Exception as e:
+            logger.error(f"Error encrypting {file_path}: {e}")
+            raise
+    
+    def decrypt_file(self, encrypted_path: Path, output_path: Optional[Path] = None) -> Path:
+        """Decrypt a Fernet encrypted file."""
+        if not output_path:
+            output_path = encrypted_path.with_suffix('')
+        
+        try:
+            with open(encrypted_path, 'rb') as f:
+                encrypted_data = f.read()
+            decrypted_data = self.fernet.decrypt(encrypted_data)
+            with open(output_path, 'wb') as f:
+                f.write(decrypted_data)
+            return output_path
+        except Exception as e:
+            logger.error(f"Error decrypting {encrypted_path}: {e}")
+            raise
+    
+    def generate_integrity_manifest(self, backup_path: Path) -> Dict:
+        """Generate integrity manifest for all files in backup."""
+        manifest = {
+            'files': [],
+            'total_files': 0,
+            'total_size': 0
+        }
+        
+        for file_path in backup_path.rglob('*'):
+            if file_path.is_file():
+                file_info = {
+                    'path': str(file_path.relative_to(backup_path)),
+                    'size': file_path.stat().st_size,
+                    'checksum': self.calculate_checksum(file_path)
+                }
+                manifest['files'].append(file_info)
+                manifest['total_files'] += 1
+                manifest['total_size'] += file_info['size']
+        
+        return manifest
+    
+    def verify_backup_integrity(self, backup_path: Path, manifest: Dict) -> bool:
+        """Verify backup integrity against manifest."""
+        try:
+            for file_info in manifest['files']:
+                file_path = backup_path / file_info['path']
+                if not file_path.exists():
+                    logger.error(f"Missing file: {file_path}")
+                    return False
+                
+                current_checksum = self.calculate_checksum(file_path)
+                if current_checksum != file_info['checksum']:
+                    logger.error(f"Checksum mismatch for {file_path}")
+                    return False
+                
+                current_size = file_path.stat().st_size
+                if current_size != file_info['size']:
+                    logger.error(f"Size mismatch for {file_path}")
+                    return False
+            
+            return True
+        except Exception as e:
+            logger.error(f"Error verifying backup integrity: {e}")
+            return False
\ No newline at end of file
diff --git a/lpm_kernel/backup/backup_service.py b/lpm_kernel/backup/backup_service.py
new file mode 100644
index 00000000..b2243e5b
--- /dev/null
+++ b/lpm_kernel/backup/backup_service.py
@@ -0,0 +1,343 @@
+import os
+import shutil
+from datetime import datetime
+import uuid
+from pathlib import Path
+import json
+from typing import Optional, Dict, List
+from tqdm import tqdm
+from ..common.logging import logger
+from .backup_integrity import BackupIntegrity
+
+# Placeholder for backup service implementation
+
+import threading
+import fcntl
+import time
+
+class BackupService:
+    def __init__(self, config):
+        self.config = config
+        # Initialize backup directory path from config, default to 'backups' relative to base_dir
+        base_dir = Path(config.get("BASE_DIR", "."))
+        self.backup_base_dir = base_dir / config.get("BACKUP_DIR", "backups")
+        self.backup_base_dir.mkdir(parents=True, exist_ok=True)
+        logger.info(f"Backup directory initialized at: {self.backup_base_dir}")
+
+        # Initialize backup integrity checker
+        encryption_key = config.get("BACKUP_ENCRYPTION_KEY")
+        self.integrity_checker = BackupIntegrity(encryption_key)
+
+        # Define source directories/files to be backed up, relative to base_dir
+        self.source_paths = [
+            Path(config.get("RESOURCES_DIR", "resources")),
+            Path(config.get("DB_FILE", "data/sqlite/lpm.db")).parent, # Backup the whole sqlite dir
+            Path(config.get("CHROMA_PERSIST_DIRECTORY", "data/chroma_db"))
+        ]
+        # Convert relative paths to absolute paths based on base_dir
+        self.absolute_source_paths = [base_dir / p for p in self.source_paths]
+        
+        # Backup settings
+        self.compress_backup = config.get("BACKUP_COMPRESS", "true").lower() == "true"
+        self.encrypt_backup = config.get("BACKUP_ENCRYPT", "false").lower() == "true"
+        
+        # Initialize locks
+        self._operation_lock = threading.Lock()
+        self._lock_file_path = self.backup_base_dir / ".backup.lock"
+        self._lock_file = None
+        
+    def _acquire_distributed_lock(self, timeout=30):
+        """Acquire distributed lock using file locking"""
+        start_time = time.time()
+        while True:
+            try:
+                self._lock_file = open(self._lock_file_path, 'w')
+                fcntl.flock(self._lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+                logger.debug("Acquired distributed lock for backup operation")
+                return True
+            except IOError:
+                if time.time() - start_time > timeout:
+                    logger.error("Timeout waiting for distributed lock")
+                    return False
+                time.sleep(1)
+                
+    def _release_distributed_lock(self):
+        """Release distributed lock"""
+        if self._lock_file:
+            try:
+                fcntl.flock(self._lock_file.fileno(), fcntl.LOCK_UN)
+                self._lock_file.close()
+                self._lock_file = None
+                logger.debug("Released distributed lock for backup operation")
+            except Exception as e:
+                logger.error(f"Error releasing distributed lock: {e}")
+
+
+    def create_backup(self, description=None, tags: Optional[List[str]] = None, name: Optional[str] = None):
+        """Creates a new backup."""
+        backup_id = str(uuid.uuid4())
+        timestamp = datetime.now()
+        backup_folder_name = timestamp.strftime("%Y%m%d_%H%M%S_") + backup_id[:8]
+        backup_path = self.backup_base_dir / backup_folder_name
+
+        try:
+            backup_path.mkdir(parents=True, exist_ok=True)
+            logger.info(f"Creating backup '{backup_id}' in {backup_path}")
+
+            copied_items = []
+            total_size = 0
+
+            # Copy source paths to backup directory
+            for src_path in self.absolute_source_paths:
+                if not src_path.exists():
+                    logger.warning(f"Source path {src_path} does not exist, skipping.")
+                    continue
+                
+                dest_path = backup_path / src_path.name
+                try:
+                    if src_path.is_dir():
+                        shutil.copytree(src_path, dest_path, dirs_exist_ok=True)
+                        logger.info(f"Copied directory {src_path} to {dest_path}")
+                    elif src_path.is_file():
+                        shutil.copy2(src_path, dest_path)
+                        logger.info(f"Copied file {src_path} to {dest_path}")
+                    else:
+                        logger.warning(f"Source path {src_path} is neither a file nor a directory, skipping.")
+                        continue
+                    
+                    # Calculate size (basic implementation, might need refinement for large dirs)
+                    current_size = sum(f.stat().st_size for f in dest_path.glob('**/*') if f.is_file())
+                    total_size += current_size
+                    copied_items.append(str(src_path.relative_to(Path(self.config.get("BASE_DIR", ".")))))
+
+                except Exception as copy_e:
+                    logger.error(f"Error copying {src_path} to {dest_path}: {copy_e}", exc_info=True)
+                    # Decide if we should continue or fail the whole backup
+                    # For now, let's log and continue
+
+            # Process backup files (compress and encrypt if enabled)
+            processed_items = []
+            with tqdm(total=len(copied_items), desc="Processing backup files") as pbar:
+                for item in copied_items:
+                    item_path = backup_path / Path(item).name
+                    if self.compress_backup:
+                        item_path = self.integrity_checker.compress_file(item_path)
+                    if self.encrypt_backup:
+                        item_path = self.integrity_checker.encrypt_file(item_path)
+                    processed_items.append(str(item_path.relative_to(backup_path)))
+                    pbar.update(1)
+
+            # Generate integrity manifest
+            integrity_manifest = self.integrity_checker.generate_integrity_manifest(backup_path)
+
+            # Create metadata file
+            metadata = {
+                "id": backup_id,
+                "name": name or backup_folder_name,
+                "timestamp": timestamp.isoformat(),
+                "description": description or "Manual backup",
+                "tags": tags or [],
+                "size_bytes": total_size,
+                "items": copied_items,
+                "processed_items": processed_items,
+                "compression_enabled": self.compress_backup,
+                "encryption_enabled": self.encrypt_backup,
+                "integrity_manifest": integrity_manifest
+            }
+            metadata_file = backup_path / "backup_metadata.json"
+            with open(metadata_file, 'w') as f:
+                json.dump(metadata, f, indent=4)
+            
+            logger.info(f"Backup '{backup_id}' created successfully.")
+            return metadata
+
+        except Exception as e:
+            logger.error(f"Failed to create backup '{backup_id}': {e}", exc_info=True)
+            # Clean up partially created backup directory if needed
+            if backup_path.exists():
+                try:
+                    shutil.rmtree(backup_path)
+                    logger.info(f"Cleaned up failed backup directory: {backup_path}")
+                except Exception as cleanup_e:
+                    logger.error(f"Error cleaning up failed backup directory {backup_path}: {cleanup_e}")
+            return None # Indicate failure
+
+    def list_backups(self):
+        """Lists available backups by reading metadata files."""
+        backups = []
+        if not self.backup_base_dir.exists():
+            logger.warning(f"Backup directory {self.backup_base_dir} does not exist.")
+            return backups
+
+        for backup_dir in self.backup_base_dir.iterdir():
+            if backup_dir.is_dir():
+                metadata_file = backup_dir / "backup_metadata.json"
+                if metadata_file.exists() and metadata_file.is_file():
+                    try:
+                        with open(metadata_file, 'r') as f:
+                            metadata = json.load(f)
+                            # Basic validation: check if 'id' and 'timestamp' exist
+                            if 'id' in metadata and 'timestamp' in metadata:
+                                backups.append(metadata)
+                            else:
+                                logger.warning(f"Invalid metadata file found in {backup_dir}, missing required fields.")
+                    except json.JSONDecodeError:
+                        logger.warning(f"Could not decode metadata file: {metadata_file}")
+                    except Exception as e:
+                        logger.error(f"Error reading metadata file {metadata_file}: {e}", exc_info=True)
+                else:
+                    logger.debug(f"Skipping directory {backup_dir}, no metadata file found.")
+        
+        # Sort backups by timestamp, newest first
+        backups.sort(key=lambda x: x.get('timestamp', ''), reverse=True)
+        logger.info(f"Found {len(backups)} backups.")
+        return backups
+
+    def restore_backup(self, backup_id, verify_integrity: bool = True):
+        """Restores data from a specific backup."""
+        if not self._acquire_distributed_lock():
+            logger.error("Failed to acquire lock for backup restoration")
+            return {"status": "error", "message": "Failed to acquire lock for backup restoration"}
+
+        total_files = 0
+        restored_files = 0
+        failed_files = []
+            
+        try:
+            logger.info(f"Attempting to restore backup: {backup_id}")
+            # Find the backup directory
+            backup_to_restore = None
+        for backup_dir in self.backup_base_dir.iterdir():
+            if backup_dir.is_dir() and backup_dir.name.endswith(backup_id[:8]): # Simple check based on folder name convention
+                metadata_file = backup_dir / "backup_metadata.json"
+                if metadata_file.exists():
+                    try:
+                        with open(metadata_file, 'r') as f:
+                            metadata = json.load(f)
+                            if metadata.get('id') == backup_id:
+                                backup_to_restore = backup_dir
+                                break
+                    except Exception as e:
+                        logger.error(f"Error reading metadata for potential restore candidate {backup_dir}: {e}")
+        
+        if not backup_to_restore:
+            logger.error(f"Backup with ID {backup_id} not found.")
+            return {"status": "error", "message": f"Backup {backup_id} not found."}
+
+        logger.info(f"Found backup directory to restore: {backup_to_restore}")
+        
+        try:
+            # Get metadata to know what was backed up
+            metadata_file = backup_to_restore / "backup_metadata.json"
+            with open(metadata_file, 'r') as f:
+                metadata = json.load(f)
+            
+            # Verify backup integrity if requested
+            if verify_integrity:
+                logger.info("Verifying backup integrity...")
+                if not self.integrity_checker.verify_backup_integrity(backup_to_restore, metadata.get('integrity_manifest', {})):
+                    return {"status": "error", "message": "Backup integrity verification failed"}
+            
+            # Get base directory from config
+            base_dir = Path(self.config.get("BASE_DIR", "."))
+            
+            # Restore each backed up item
+            restored_items = []
+            total_files = len(metadata.get('items', []))
+            
+            with tqdm(total=total_files, desc="Restoring backup") as pbar:
+                for item in metadata.get('items', []):
+                    source_path = backup_to_restore / Path(item).name
+                    target_path = base_dir / item
+                    
+                    if not source_path.exists():
+                        logger.warning(f"Source path {source_path} does not exist in backup, skipping.")
+                        failed_files.append({"path": str(source_path), "error": "Source file missing"})
+                        pbar.update(1)
+                        continue
+                    
+                    try:
+                        # Create parent directory if it doesn't exist
+                        target_path.parent.mkdir(parents=True, exist_ok=True)
+                        
+                        # Remove existing data if it exists
+                        if target_path.exists():
+                            if target_path.is_dir():
+                                shutil.rmtree(target_path)
+                                logger.info(f"Removed existing directory: {target_path}")
+                            else:
+                                target_path.unlink()
+                                logger.info(f"Removed existing file: {target_path}")
+                        
+                        # Process backup files (decompress and decrypt if needed)
+                        current_source = source_path
+                        if self.encrypt_backup:
+                            current_source = self.integrity_checker.decrypt_file(current_source)
+                        if self.compress_backup:
+                            current_source = self.integrity_checker.decompress_file(current_source)
+                        
+                        # Copy from backup to target
+                        if current_source.is_dir():
+                            shutil.copytree(current_source, target_path)
+                            logger.info(f"Restored directory from {current_source} to {target_path}")
+                        else:
+                            shutil.copy2(current_source, target_path)
+                            logger.info(f"Restored file from {current_source} to {target_path}")
+                        
+                        restored_items.append(item)
+                        restored_files += 1
+                    except Exception as e:
+                        error_msg = str(e)
+                        logger.error(f"Error restoring {item}: {error_msg}", exc_info=True)
+                        failed_files.append({"path": str(source_path), "error": error_msg})
+                    finally:
+                        pbar.update(1)
+            
+            # Return success with details
+            status = "success" if not failed_files else "partial_success"
+            return {
+                "status": status,
+                "message": f"Restored backup {backup_id}: {restored_files}/{total_files} files restored successfully",
+                "restored_items": restored_items,
+                "failed_items": failed_files,
+                "timestamp": metadata.get('timestamp'),
+                "description": metadata.get('description'),
+                "stats": {
+                    "total_files": total_files,
+                    "restored_files": restored_files,
+                    "failed_files": len(failed_files)
+                }
+            }
+            
+        except Exception as e:
+            logger.error(f"Error during restore process: {e}", exc_info=True)
+            return {"status": "error", "message": f"Failed to restore backup {backup_id}: {e}"}
+
+    def delete_backup(self, backup_id):
+        """Deletes a specific backup directory."""
+        logger.info(f"Attempting to delete backup: {backup_id}")
+        backup_to_delete = None
+        for backup_dir in self.backup_base_dir.iterdir():
+            if backup_dir.is_dir() and backup_dir.name.endswith(backup_id[:8]): # Simple check
+                metadata_file = backup_dir / "backup_metadata.json"
+                if metadata_file.exists():
+                    try:
+                        with open(metadata_file, 'r') as f:
+                            metadata = json.load(f)
+                            if metadata.get('id') == backup_id:
+                                backup_to_delete = backup_dir
+                                break
+                    except Exception as e:
+                        logger.error(f"Error reading metadata for potential delete candidate {backup_dir}: {e}")
+
+        if backup_to_delete:
+            try:
+                shutil.rmtree(backup_to_delete)
+                logger.info(f"Successfully deleted backup directory: {backup_to_delete}")
+                return {"status": "success", "message": f"Backup {backup_id} deleted."}
+            except Exception as e:
+                logger.error(f"Error deleting backup directory {backup_to_delete}: {e}", exc_info=True)
+                return {"status": "error", "message": f"Failed to delete backup {backup_id}: {e}"}
+        else:
+            logger.warning(f"Backup with ID {backup_id} not found for deletion.")
+            return {"status": "error", "message": f"Backup {backup_id} not found."}
\ No newline at end of file