feat: add local LLMWiki wiki vector index

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

@@ -0,0 +1,290 @@
+"""Authenticated APIs for DeerFlow-local Wiki vector search and administration."""
+
+from __future__ import annotations
+
+from typing import Any
+from uuid import uuid4
+
+from fastapi import APIRouter, HTTPException, Query, Request
+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.integrations.weknora.local_index.retrieval import is_fully_vectorized
+from deerflow.integrations.weknora.local_index.search import WikiIndexError
+
+router = APIRouter(prefix="/api/llmwiki", tags=["llmwiki-index"])
+
+
+class WikiVectorSearchRequest(BaseModel):
+    query: str = Field(..., min_length=1, max_length=4000)
+    knowledge_base_ids: list[str] = Field(default_factory=list, max_length=100)
+    top_k: int = Field(default=8, ge=1, le=50)
+    include_drafts: bool = False
+    include_content: bool = True
+
+
+class WikiIndexActionRequest(BaseModel):
+    knowledge_base_ids: list[str] = Field(default_factory=list, max_length=100)
+    force: bool = False
+    reason: str = Field(default="manual", max_length=255)
+
+
+class WikiIndexSettingsRequest(BaseModel):
+    wiki_index_enabled: bool | None = None
+    external_search_enabled: bool | None = None
+
+
+async def _actor(request: Request) -> tuple[str, bool]:
+    user = await get_optional_user_from_request(request)
+    if user is None:
+        return "default", True
+    return str(user.id), getattr(user, "system_role", None) == "admin"
+
+
+def _services(request: Request):
+    config = request.app.state.config.llmwiki.local_wiki_index
+    if not config.enabled:
+        raise HTTPException(status_code=409, detail={"code": "WIKI_INDEX_DISABLED", "message": "Local Wiki index is disabled"})
+    search = getattr(request.app.state, "llmwiki_vector_search", None)
+    index_store = getattr(request.app.state, "llmwiki_index_store", None)
+    if search is None or index_store is None:
+        raise HTTPException(status_code=503, detail={"code": "WIKI_INDEX_UNAVAILABLE", "message": "Local Wiki index is unavailable"})
+    return config, search, index_store
+
+
+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
+    if mapping_ids:
+        rows = []
+        for mapping_id in dict.fromkeys(mapping_ids):
+            row = await mapping_store.get_authorized(mapping_id, user_id, write=False, is_admin=is_admin)
+            if row is None or not row.get("wiki_index_enabled", True):
+                raise HTTPException(status_code=404, detail="Knowledge base not found")
+            rows.append(row)
+    else:
+        rows = [row for row in await mapping_store.list_visible(user_id, scope="all", is_admin=is_admin) if row.get("wiki_index_enabled", True)]
+    return rows, user_id, is_admin
+
+
+@router.post("/wiki-vector-search")
+async def wiki_vector_search(request: Request, body: WikiVectorSearchRequest) -> dict[str, Any]:
+    config, search, _ = _services(request)
+    if not config.internal_search_enabled:
+        raise HTTPException(status_code=409, detail={"code": "WIKI_INTERNAL_SEARCH_DISABLED", "message": "Local Wiki search is not enabled"})
+    rows, user_id, is_admin = await _authorized_mappings(request, body.knowledge_base_ids)
+    include_drafts = body.include_drafts and (is_admin or all(str(row.get("owner_user_id")) == user_id for row in rows))
+    try:
+        result = await search.search(body.query, rows, top_k_pages=body.top_k, include_drafts=include_drafts)
+    except WikiIndexError as exc:
+        raise HTTPException(status_code=exc.status_code, detail={"code": exc.code, "message": str(exc)}) from None
+    if not body.include_content:
+        for item in result["results"]:
+            item.pop("content", None)
+            item.pop("snippet", None)
+            for section in item["matched_sections"]:
+                section.pop("content", None)
+    return result
+
+
+@router.get("/knowledge-bases/{mapping_id}/local-wiki/pages/{slug:path}")
+async def local_wiki_page(
+    request: Request,
+    mapping_id: str,
+    slug: str,
+    include_provenance: bool = Query(default=False),
+    include_archived: bool = Query(default=False),
+) -> dict[str, Any]:
+    _, _, index_store = _services(request)
+    user_id, is_admin = await _actor(request)
+    mapping = await request.app.state.llmwiki_store.get_authorized(mapping_id, user_id, write=False, is_admin=is_admin)
+    if mapping is None:
+        raise HTTPException(status_code=404, detail="Wiki page not found")
+    page = await index_store.get_page(mapping_id, slug.strip("/"))
+    if page is None or page.get("is_remote_deleted"):
+        raise HTTPException(status_code=404, detail="Wiki page not found")
+    owner_or_admin = is_admin or str(mapping.get("owner_user_id")) == user_id
+    if page.get("status") == "archived" and not (include_archived and owner_or_admin):
+        raise HTTPException(status_code=404, detail="Wiki page not found")
+    if page.get("status") != "published" and not owner_or_admin:
+        raise HTTPException(status_code=404, detail="Wiki page not found")
+    result = {
+        "id": page["id"],
+        "knowledge_base_id": mapping_id,
+        "title": page.get("title") or "",
+        "slug": page["slug"],
+        "summary": page.get("summary") or "",
+        "content": page.get("content_md") or "",
+        "page_type": page.get("page_type") or "article",
+        "status": page.get("status"),
+        "aliases": page.get("aliases") or [],
+        "category_path": page.get("category_path") or [],
+        "parent_slug": page.get("parent_slug"),
+        "wiki_path": page.get("wiki_path"),
+        "page_version": page.get("page_version") or 0,
+        "updated_at": page.get("remote_updated_at") or page.get("updated_at"),
+        "created_at": page.get("remote_created_at") or page.get("created_at"),
+    }
+    if owner_or_admin:
+        result.update({"index_status": page.get("index_status"), "indexed_at": page.get("indexed_at")})
+        if include_provenance:
+            result.update({"source_refs": page.get("source_refs") or [], "chunk_refs": page.get("chunk_refs") or []})
+    return result
+
+
+@router.get("/wiki-index/status")
+async def wiki_index_status(request: Request) -> dict[str, Any]:
+    _, is_admin = await _actor(request)
+    if not is_admin:
+        raise HTTPException(status_code=403, detail="Administrator access required")
+    config = request.app.state.config.llmwiki.local_wiki_index
+    index_store = getattr(request.app.state, "llmwiki_index_store", None)
+    if index_store is None:
+        raise HTTPException(status_code=503, detail={"code": "WIKI_INDEX_UNAVAILABLE"})
+    mappings = await request.app.state.llmwiki_store.list_visible("system", scope="all", is_admin=True)
+    states = {row["knowledge_base_mapping_id"]: row for row in await index_store.list_index_status()}
+    items = []
+    for mapping in mappings:
+        state = states.get(mapping["id"], {"state": "not_synced", "index_revision": 0})
+        items.append(
+            {
+                **state,
+                "knowledge_base_id": mapping["id"],
+                "knowledge_base_name": mapping.get("name") or "",
+                "wiki_index_enabled": mapping.get("wiki_index_enabled", True),
+                "external_search_enabled": mapping.get("external_search_enabled", False),
+            }
+        )
+    return {
+        "enabled": config.enabled,
+        "sync_enabled": config.sync_enabled,
+        "internal_search_enabled": config.internal_search_enabled,
+        "embedding_model": config.embedding.model,
+        "embedding_fingerprint": (
+            request.app.state.llmwiki_embedding.fingerprint[:12]
+            if getattr(request.app.state, "llmwiki_embedding", None) is not None
+            else None
+        ),
+        "knowledge_bases": items,
+    }
+
+
+async def _run_sync(app, mapping_ids: list[str], force: bool) -> None:
+    service = getattr(app.state, "llmwiki_sync_service", None)
+    if service is None:
+        return
+    mappings = await app.state.llmwiki_store.list_visible("system", scope="all", is_admin=True)
+    wanted = set(mapping_ids)
+    for mapping in mappings:
+        if mapping.get("wiki_index_enabled", True) and (not wanted or mapping["id"] in wanted):
+            try:
+                await service.sync_mapping(mapping, force=force)
+            except Exception:
+                continue
+
+
+async def _run_one_sync(app, mapping: dict[str, Any], *, force: bool) -> None:
+    service = getattr(app.state, "llmwiki_sync_service", None)
+    if service is None:
+        return
+    try:
+        await service.sync_mapping(mapping, force=force)
+    except Exception:
+        return
+
+
+@router.post("/wiki-index/sync", status_code=202)
+async def sync_wiki_index(request: Request, body: WikiIndexActionRequest) -> dict[str, Any]:
+    _services(request)
+    _, is_admin = await _actor(request)
+    if not is_admin:
+        raise HTTPException(status_code=403, detail="Administrator access required")
+    job_id = str(uuid4())
+    spawn_local_wiki_index_job(request.app, _run_sync(request.app, body.knowledge_base_ids, body.force))
+    return {"job_id": job_id, "status": "accepted"}
+
+
+@router.post("/wiki-index/rebuild", status_code=202)
+async def rebuild_wiki_index(request: Request, body: WikiIndexActionRequest) -> dict[str, Any]:
+    _, _, index_store = _services(request)
+    _, is_admin = await _actor(request)
+    if not is_admin:
+        raise HTTPException(status_code=403, detail="Administrator access required")
+    mappings = await request.app.state.llmwiki_store.list_visible("system", scope="all", is_admin=True)
+    ids = body.knowledge_base_ids or [row["id"] for row in mappings if row.get("wiki_index_enabled", True)]
+    fingerprint = request.app.state.llmwiki_embedding.fingerprint
+    await index_store.mark_rebuild_required(ids, fingerprint)
+    job_id = str(uuid4())
+    spawn_local_wiki_index_job(request.app, _run_sync(request.app, ids, True))
+    return {"job_id": job_id, "status": "accepted", "reason": body.reason}
+
+
+@router.post("/wiki-index/knowledge-bases/{mapping_id}/sync", status_code=202)
+async def sync_one_wiki_index(request: Request, mapping_id: str, force: bool = Query(default=False)) -> dict[str, Any]:
+    return await sync_wiki_index(request, WikiIndexActionRequest(knowledge_base_ids=[mapping_id], force=force))
+
+
+@router.post("/wiki-index/knowledge-bases/{mapping_id}/rebuild", status_code=202)
+async def rebuild_one_wiki_index(request: Request, mapping_id: str, reason: str = Query(default="manual", max_length=255)) -> dict[str, Any]:
+    return await rebuild_wiki_index(request, WikiIndexActionRequest(knowledge_base_ids=[mapping_id], force=True, reason=reason))
+
+
+@router.get("/knowledge-bases/{mapping_id}/wiki-index-status")
+async def knowledge_base_wiki_index_status(request: Request, mapping_id: str) -> dict[str, Any]:
+    user_id, is_admin = await _actor(request)
+    mapping = await request.app.state.llmwiki_store.get_authorized(mapping_id, user_id, write=False, is_admin=is_admin)
+    if mapping is None:
+        raise HTTPException(status_code=404, detail="Knowledge base not found")
+    index_store = getattr(request.app.state, "llmwiki_index_store", None)
+    if index_store is None:
+        raise HTTPException(status_code=503, detail={"code": "WIKI_INDEX_UNAVAILABLE"})
+    states = await index_store.list_index_status([mapping_id])
+    state = states[0] if states else {"knowledge_base_mapping_id": mapping_id, "state": "not_synced", "index_revision": 0}
+    embedding = getattr(request.app.state, "llmwiki_embedding", None)
+    fingerprint = embedding.fingerprint if embedding is not None else None
+    fully_vectorized = bool(mapping.get("wiki_index_enabled", True)) and is_fully_vectorized(state, fingerprint)
+    return {
+        **state,
+        "knowledge_base_id": mapping_id,
+        "local_index_enabled": bool(request.app.state.config.llmwiki.local_wiki_index.enabled),
+        "wiki_index_enabled": mapping.get("wiki_index_enabled", True),
+        "fully_vectorized": fully_vectorized,
+        "retrieval_mode": "vector" if fully_vectorized else "wiki_api",
+    }
+
+
+@router.post("/knowledge-bases/{mapping_id}/vectorize", status_code=202)
+async def vectorize_knowledge_base(request: Request, mapping_id: str) -> dict[str, Any]:
+    config = request.app.state.config.llmwiki.local_wiki_index
+    if not config.enabled:
+        raise HTTPException(
+            status_code=409,
+            detail={"code": "WIKI_INDEX_DISABLED", "message": "本地 Wiki 向量化未在配置中启用"},
+        )
+    service = getattr(request.app.state, "llmwiki_sync_service", None)
+    if service is None:
+        raise HTTPException(status_code=503, detail={"code": "WIKI_VECTORIZATION_UNAVAILABLE"})
+    user_id, is_admin = await _actor(request)
+    mapping_store = request.app.state.llmwiki_store
+    mapping = await mapping_store.get_authorized(mapping_id, user_id, write=True, is_admin=is_admin)
+    if mapping is None:
+        raise HTTPException(status_code=404, detail="Knowledge base not found")
+    if not mapping.get("wiki_index_enabled", True):
+        mapping = await mapping_store.update_mapping(mapping_id, wiki_index_enabled=True)
+    job_id = str(uuid4())
+    spawn_local_wiki_index_job(request.app, _run_one_sync(request.app, mapping, force=True))
+    return {"job_id": job_id, "status": "accepted", "knowledge_base_id": mapping_id}
+
+
+@router.patch("/knowledge-bases/{mapping_id}/wiki-index-settings")
+async def update_wiki_index_settings(request: Request, mapping_id: str, body: WikiIndexSettingsRequest) -> dict[str, Any]:
+    user_id, is_admin = await _actor(request)
+    store = request.app.state.llmwiki_store
+    current = await store.get_authorized(mapping_id, user_id, write=True, is_admin=is_admin)
+    if current is None:
+        raise HTTPException(status_code=404, detail="Knowledge base not found")
+    if body.external_search_enabled and str(current.get("owner_user_id")) == "system" and str(current.get("name") or "").strip() == "对话沉淀":
+        raise HTTPException(status_code=403, detail="Conversation deposit can never be exposed externally")
+    updated = await store.update_mapping(mapping_id, wiki_index_enabled=body.wiki_index_enabled, external_search_enabled=body.external_search_enabled)
+    return {"knowledge_base_id": mapping_id, "wiki_index_enabled": updated.get("wiki_index_enabled", True), "external_search_enabled": updated.get("external_search_enabled", False)}