feat: improve knowledge sync and entity alignment

offline-backend-20260512/backend/app/gateway/routers/llmwiki_index.py

@@ -2,15 +2,20 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
+import shutil
+from pathlib import Path, PurePosixPath
 from typing import Any
 from uuid import uuid4
 
-from fastapi import APIRouter, HTTPException, Query, Request
+from fastapi import APIRouter, File, Form, HTTPException, Query, Request, UploadFile
 from pydantic import BaseModel, Field
 
 from app.gateway.deps import get_optional_user_from_request
 from app.gateway.llmwiki_index_scheduler import spawn_local_wiki_index_job
+from deerflow.config.runtime_paths import resolve_path, runtime_home
+from deerflow.config.system_settings import load_system_settings, save_system_settings
 from deerflow.integrations.weknora.local_index.retrieval import is_fully_vectorized
 from deerflow.integrations.weknora.local_index.search import WikiIndexError
 
@@ -37,6 +42,10 @@ class WikiIndexSettingsRequest(BaseModel):
     external_search_enabled: bool | None = None
 
 
+class LocalEncoderConfigRequest(BaseModel):
+    model_path: str = Field(min_length=1, max_length=4096)
+
+
 async def _actor(request: Request) -> tuple[str, bool]:
     user = await get_optional_user_from_request(request)
     if user is None:
@@ -55,6 +64,149 @@ def _services(request: Request):
     return config, search, index_store
 
 
+def _local_encoder(request: Request):
+    config = request.app.state.config.llmwiki.local_wiki_index.embedding
+    embedding = getattr(request.app.state, "llmwiki_embedding", None)
+    if config.provider != "local":
+        raise HTTPException(status_code=409, detail={"code": "EMBEDDING_NOT_LOCAL", "message": "当前 embedding provider 不是 local"})
+    if embedding is None:
+        raise HTTPException(status_code=503, detail={"code": "LOCAL_ENCODER_UNAVAILABLE", "message": "本地编码器未配置或未启用"})
+    return embedding
+
+
+def _validated_encoder_path(raw_path: str) -> Path:
+    path = resolve_path(raw_path.strip())
+    if not path.is_dir():
+        raise HTTPException(status_code=422, detail="编码器目录不存在")
+    if not ((path / "onnx" / "model.onnx").is_file() or (path / "model.onnx").is_file()):
+        raise HTTPException(status_code=422, detail="编码器目录缺少 onnx/model.onnx 或 model.onnx")
+    return path
+
+
+def _safe_relative_upload_path(value: str) -> Path:
+    normalized = str(value or "").replace("\\", "/").strip("/")
+    candidate = PurePosixPath(normalized)
+    if not normalized or candidate.is_absolute() or ".." in candidate.parts:
+        raise HTTPException(status_code=422, detail=f"非法模型文件路径：{value}")
+    if len(normalized) > 512:
+        raise HTTPException(status_code=422, detail="模型文件路径过长")
+    return Path(*candidate.parts)
+
+
+def _copy_encoder_uploads(files: list[UploadFile], paths: list[str], staging: Path) -> int:
+    total = 0
+    for upload, raw_path in zip(files, paths, strict=True):
+        relative = _safe_relative_upload_path(raw_path)
+        destination = staging / relative
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        upload.file.seek(0)
+        with destination.open("wb") as output:
+            while chunk := upload.file.read(4 * 1024 * 1024):
+                output.write(chunk)
+                total += len(chunk)
+    return total
+
+
+def _activate_uploaded_encoder(staging: Path) -> Path:
+    model_root = (runtime_home() / "models").resolve()
+    model_root.mkdir(parents=True, exist_ok=True)
+    active = (model_root / "bge-m3").resolve()
+    if active.parent != model_root:
+        raise RuntimeError("编码器目标目录越界")
+    backup = (model_root / f".bge-m3-backup-{uuid4().hex}").resolve()
+    if active.exists():
+        active.replace(backup)
+    try:
+        staging.replace(active)
+    except Exception:
+        if backup.exists() and not active.exists():
+            backup.replace(active)
+        raise
+    if backup.exists():
+        shutil.rmtree(backup)
+    return active
+
+
+async def _save_encoder_path(request: Request, path: Path) -> dict[str, object]:
+    embedding = _local_encoder(request)
+    current = load_system_settings()
+    encoder_settings = current.local_wiki_encoder.model_copy(update={"model_path": str(path)})
+    save_system_settings(current.model_copy(update={"local_wiki_encoder": encoder_settings}))
+    await embedding.reset_local_encoder(model_path=str(path))
+    return embedding.local_encoder_status()
+
+
+@router.get("/local-encoder/status")
+async def local_encoder_status(request: Request) -> dict[str, object]:
+    _, is_admin = await _actor(request)
+    if not is_admin:
+        raise HTTPException(status_code=403, detail="Administrator access required")
+    return _local_encoder(request).local_encoder_status()
+
+
+@router.post("/local-encoder/start", status_code=202)
+async def start_local_encoder(request: Request) -> dict[str, object]:
+    _, is_admin = await _actor(request)
+    if not is_admin:
+        raise HTTPException(status_code=403, detail="Administrator access required")
+    try:
+        return _local_encoder(request).start_local_encoder()
+    except ValueError as exc:
+        raise HTTPException(status_code=409, detail={"code": "LOCAL_ENCODER_START_REJECTED", "message": str(exc)}) from None
+
+
+@router.put("/local-encoder/config")
+async def configure_local_encoder(request: Request, body: LocalEncoderConfigRequest) -> dict[str, object]:
+    _, is_admin = await _actor(request)
+    if not is_admin:
+        raise HTTPException(status_code=403, detail="Administrator access required")
+    return await _save_encoder_path(request, _validated_encoder_path(body.model_path))
+
+
+@router.post("/local-encoder/upload")
+async def upload_local_encoder(
+    request: Request,
+    files: list[UploadFile] = File(...),
+    paths: list[str] = Form(...),
+) -> dict[str, object]:
+    """Stream an offline model directory into runtime storage and activate it.
+
+    Starlette spools multipart files to disk, and the copy loop is chunked, so
+    multi-gigabyte ONNX models do not need to be loaded into Gateway memory.
+    """
+
+    _, is_admin = await _actor(request)
+    if not is_admin:
+        raise HTTPException(status_code=403, detail="Administrator access required")
+    if not files or len(files) != len(paths):
+        raise HTTPException(status_code=422, detail="模型文件与相对路径不匹配")
+    if len(files) > 500:
+        raise HTTPException(status_code=422, detail="模型目录文件数不能超过 500")
+
+    staging_root = (runtime_home() / "model-uploads").resolve()
+    staging = (staging_root / uuid4().hex).resolve()
+    if staging.parent != staging_root:
+        raise HTTPException(status_code=500, detail="模型暂存目录越界")
+    staging.mkdir(parents=True, exist_ok=False)
+    try:
+        uploaded_bytes = await asyncio.to_thread(_copy_encoder_uploads, files, paths, staging)
+        _validated_encoder_path(str(staging))
+        await _local_encoder(request).reset_local_encoder()
+        active = await asyncio.to_thread(_activate_uploaded_encoder, staging)
+        status = await _save_encoder_path(request, active)
+        return {**status, "uploaded_files": len(files), "uploaded_bytes": uploaded_bytes}
+    except HTTPException:
+        raise
+    except OSError as exc:
+        logger.exception("Failed to store uploaded local encoder")
+        raise HTTPException(status_code=507, detail=f"编码器文件保存失败：{exc}") from None
+    finally:
+        for upload in files:
+            await upload.close()
+        if staging.exists():
+            await asyncio.to_thread(shutil.rmtree, staging, True)
+
+
 async def _authorized_mappings(request: Request, mapping_ids: list[str]) -> tuple[list[dict[str, Any]], str, bool]:
     user_id, is_admin = await _actor(request)
     mapping_store = request.app.state.llmwiki_store