diff --git a/backend/package/yuxi/channel/utils/__init__.py b/backend/package/yuxi/channel/utils/__init__.py
new file mode 100644
index 00000000..13964a71
--- /dev/null
+++ b/backend/package/yuxi/channel/utils/__init__.py
@@ -0,0 +1,35 @@
+from yuxi.channel.utils.files import (
+    compute_file_hash,
+    delete_file,
+    ensure_directory,
+    list_files,
+    read_json_file,
+    safe_write_file,
+    write_json_file,
+)
+from yuxi.channel.utils.lazy import (
+    LRULazyModule,
+    Lazy,
+    LazyClass,
+    LazyModule,
+    lazy_property,
+    run_sync,
+    run_sync_partial,
+)
+
+__all__ = [
+    "Lazy",
+    "lazy_property",
+    "LazyModule",
+    "LazyClass",
+    "run_sync",
+    "run_sync_partial",
+    "LRULazyModule",
+    "read_json_file",
+    "write_json_file",
+    "compute_file_hash",
+    "ensure_directory",
+    "safe_write_file",
+    "list_files",
+    "delete_file",
+]
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/utils/file_service.py b/backend/package/yuxi/channel/utils/file_service.py
new file mode 100644
index 00000000..1e4b36c0
--- /dev/null
+++ b/backend/package/yuxi/channel/utils/file_service.py
@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import hashlib
+import os
+import re
+from pathlib import Path
+
+IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".svg"}
+VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".flv", ".wmv"}
+ALLOWED_EXTENSIONS = IMAGE_EXTENSIONS | VIDEO_EXTENSIONS | {".pdf", ".txt", ".json", ".zip", ".rar"}
+
+_DEFAULT_MAX_FILE_BYTES = 50 * 1024 * 1024
+
+
+class FileService:
+    def __init__(self, storage_dir: str | Path, max_file_bytes: int = _DEFAULT_MAX_FILE_BYTES):
+        self.storage_dir = Path(storage_dir)
+        self.storage_dir.mkdir(parents=True, exist_ok=True)
+        self.max_file_bytes = max_file_bytes
+
+    def classify(self, filename: str) -> str:
+        ext = os.path.splitext(filename)[1].lower()
+        if ext in IMAGE_EXTENSIONS:
+            return "image"
+        if ext in VIDEO_EXTENSIONS:
+            return "video"
+        return "file"
+
+    def generate_safe_name(self, original_name: str) -> str:
+        name = self._sanitize_filename(original_name)
+        h = hashlib.sha256(name.encode()).hexdigest()[:12]
+        ext = os.path.splitext(original_name)[1].lower()
+        return f"{h}_{name}"
+
+    async def save_upload(self, content: bytes, safe_name: str) -> Path:
+        dest = self.storage_dir / safe_name
+        dest.write_bytes(content)
+        return dest
+
+    @staticmethod
+    def _sanitize_filename(filename: str) -> str:
+        return re.sub(r"[^\w.\-]", "_", filename)
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/utils/files.py b/backend/package/yuxi/channel/utils/files.py
new file mode 100644
index 00000000..d3d9fdd1
--- /dev/null
+++ b/backend/package/yuxi/channel/utils/files.py
@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+def read_json_file(file_path: str | Path) -> dict[str, Any]:
+    """读取 JSON 文件并返回字典。"""
+    file_path = Path(file_path)
+    if not file_path.exists():
+        raise FileNotFoundError(f"File not found: {file_path}")
+    return json.loads(file_path.read_text(encoding="utf-8"))
+
+
+def write_json_file(file_path: str | Path, data: dict[str, Any], *, indent: int = 2) -> None:
+    """将字典写入 JSON 文件。"""
+    file_path = Path(file_path)
+    file_path.parent.mkdir(parents=True, exist_ok=True)
+    file_path.write_text(json.dumps(data, indent=indent, ensure_ascii=False), encoding="utf-8")
+
+
+def compute_file_hash(file_path: str | Path, algorithm: str = "sha256") -> str:
+    """计算文件的哈希值。"""
+    file_path = Path(file_path)
+    if not file_path.exists():
+        raise FileNotFoundError(f"File not found: {file_path}")
+
+    hasher = hashlib.new(algorithm)
+    hasher.update(file_path.read_bytes())
+    return hasher.hexdigest()
+
+
+def ensure_directory(dir_path: str | Path) -> Path:
+    """确保目录存在，如果不存在则创建。"""
+    path = Path(dir_path)
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def safe_write_file(file_path: str | Path, content: str, *, encoding: str = "utf-8") -> None:
+    """安全写入文件：先写入临时文件，再原子替换。"""
+    file_path = Path(file_path)
+    file_path.parent.mkdir(parents=True, exist_ok=True)
+    temp_path = file_path.with_suffix(f"{file_path.suffix}.tmp")
+    temp_path.write_text(content, encoding=encoding)
+    temp_path.replace(file_path)
+
+
+def list_files(
+    directory: str | Path,
+    pattern: str = "*",
+    *,
+    recursive: bool = False,
+) -> list[Path]:
+    """列出目录中的文件。"""
+    directory = Path(directory)
+    if not directory.exists():
+        return []
+
+    if recursive:
+        return list(directory.rglob(pattern))
+    return list(directory.glob(pattern))
+
+
+def delete_file(file_path: str | Path, *, missing_ok: bool = True) -> bool:
+    """删除文件。"""
+    file_path = Path(file_path)
+    try:
+        file_path.unlink(missing_ok=missing_ok)
+        return True
+    except OSError:
+        logger.warning("Failed to delete file: %s", file_path)
+        return False
diff --git a/backend/package/yuxi/channel/utils/lazy.py b/backend/package/yuxi/channel/utils/lazy.py
new file mode 100644
index 00000000..a7f76bf7
--- /dev/null
+++ b/backend/package/yuxi/channel/utils/lazy.py
@@ -0,0 +1,195 @@
+from __future__ import annotations
+
+import asyncio
+import functools
+import importlib
+import logging
+import threading
+from collections import OrderedDict
+from typing import Any, Callable, Generic, TypeVar
+
+T = TypeVar("T")
+logger = logging.getLogger(__name__)
+
+
+class Lazy(Generic[T]):
+    """延迟初始化包装器 — 线程安全，支持自定义工厂函数。"""
+
+    def __init__(self, factory: Callable[[], T]):
+        self._factory = factory
+        self._value: T | None = None
+        self._lock = threading.Lock()
+        self._initialized = False
+
+    @property
+    def value(self) -> T:
+        if not self._initialized:
+            with self._lock:
+                if not self._initialized:
+                    self._value = self._factory()
+                    self._initialized = True
+        return self._value
+
+    def reset(self) -> None:
+        with self._lock:
+            self._value = None
+            self._initialized = False
+
+    def is_initialized(self) -> bool:
+        return self._initialized
+
+
+def lazy_property(factory: Callable[[], T]) -> property:
+    """装饰器：将方法转换为延迟加载属性。"""
+    _lazy = Lazy(factory)
+
+    @property
+    def prop(self) -> T:
+        return _lazy.value
+
+    return prop
+
+
+class LazyModule:
+    """延迟导入模块 — 避免循环导入与启动时性能开销。"""
+
+    def __init__(self, module_name: str):
+        self._module_name = module_name
+        self._module: Any = None
+        self._lock = threading.Lock()
+
+    def __getattr__(self, name: str) -> Any:
+        if self._module is None:
+            with self._lock:
+                if self._module is None:
+                    self._module = importlib.import_module(self._module_name)
+        return getattr(self._module, name)
+
+    def __call__(self, *args: Any, **kwargs: Any) -> Any:
+        return self.__getattr__("__call__")(*args, **kwargs)
+
+    def __dir__(self) -> list[str]:
+        return dir(self._module if self._module is not None else importlib.import_module(self._module_name))
+
+    def __repr__(self) -> str:
+        status = "loaded" if self._module is not None else "not loaded"
+        return f"LazyModule({self._module_name!r}, {status})"
+
+
+class LazyClass:
+    """延迟导入类 — 按需解析模块中的类，支持实例化和属性访问。"""
+
+    def __init__(self, module_name: str, class_name: str):
+        self._module_name = module_name
+        self._class_name = class_name
+        self._cls: type | None = None
+        self._lock = threading.Lock()
+
+    def _resolve(self) -> type:
+        if self._cls is None:
+            with self._lock:
+                if self._cls is None:
+                    module = importlib.import_module(self._module_name)
+                    self._cls = getattr(module, self._class_name)
+        return self._cls
+
+    def __call__(self, *args: Any, **kwargs: Any) -> Any:
+        return self._resolve()(*args, **kwargs)
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._resolve(), name)
+
+    def __repr__(self) -> str:
+        status = "resolved" if self._cls is not None else "not resolved"
+        return f"LazyClass({self._module_name!r}, {self._class_name!r}, {status})"
+
+
+async def run_sync(func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
+    """在默认线程池中运行同步函数，返回协程。"""
+    return await asyncio.get_running_loop().run_in_executor(
+        None, functools.partial(func, *args, **kwargs)
+    )
+
+
+async def run_sync_partial(func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
+    """带偏应用的 run_sync — 将 *args/**kwargs 与 func 绑定后在线程池执行。"""
+    bound = functools.partial(func, *args, **kwargs)
+    return await asyncio.get_running_loop().run_in_executor(None, bound)
+
+
+class LRULazyModule:
+    """LRU 缓存的延迟模块加载器 — 线程安全，支持异步 get/invalidate/clear。"""
+
+    def __init__(self, maxsize: int):
+        if not isinstance(maxsize, int) or maxsize <= 0:
+            raise ValueError("maxsize must be a positive integer")
+        self._maxsize = maxsize
+        self._cache: OrderedDict[str, Any] = OrderedDict()
+        self._lock = threading.Lock()
+        self._pending: dict[str, asyncio.Event] = {}
+
+    @property
+    def size(self) -> int:
+        with self._lock:
+            return len(self._cache)
+
+    def __len__(self) -> int:
+        return self.size
+
+    def __contains__(self, key: str) -> bool:
+        with self._lock:
+            return key in self._cache
+
+    def __getitem__(self, key: str) -> Any:
+        with self._lock:
+            if key not in self._cache:
+                raise KeyError(key)
+            self._cache.move_to_end(key)
+            return self._cache[key]
+
+    async def get(self, module_name: str) -> Any:
+        with self._lock:
+            if module_name in self._cache:
+                self._cache.move_to_end(module_name)
+                return self._cache[module_name]
+
+            if module_name in self._pending:
+                event = self._pending[module_name]
+                is_loader = False
+            else:
+                event = asyncio.Event()
+                self._pending[module_name] = event
+                is_loader = True
+
+        if not is_loader:
+            await event.wait()
+            with self._lock:
+                return self._cache[module_name]
+
+        try:
+            module = importlib.import_module(module_name)
+        except Exception:
+            with self._lock:
+                self._pending.pop(module_name, None)
+                event.set()
+            raise
+
+        with self._lock:
+            self._pending.pop(module_name, None)
+            self._cache[module_name] = module
+            self._cache.move_to_end(module_name)
+            if len(self._cache) > self._maxsize:
+                self._cache.popitem(last=False)
+            event.set()
+        return module
+
+    async def invalidate(self, module_name: str) -> None:
+        with self._lock:
+            self._cache.pop(module_name, None)
+
+    async def clear(self) -> None:
+        with self._lock:
+            self._cache.clear()
+
+    def __repr__(self) -> str:
+        return f"LRULazyModule(maxsize={self._maxsize}, size={self.size})"