1011 lines
38 KiB
Python
1011 lines
38 KiB
Python
"""Read-mostly global assistant knowledge API and admin import actions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
import json
|
|
import logging
|
|
from collections.abc import AsyncIterable, AsyncIterator
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from fastapi import APIRouter, File, Form, HTTPException, Query, Request, UploadFile
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.gateway.deps import get_optional_user_from_request
|
|
from deerflow.assistant_knowledge.package_io import PACKAGE_VERSION, package_path, read_package, write_jsonl_record
|
|
from deerflow.config.runtime_paths import runtime_home
|
|
from deerflow.integrations.weknora.client import WeKnoraError
|
|
from deerflow.integrations.weknora.runtime import build_weknora_client, get_resolved_llmwiki_runtime
|
|
|
|
router = APIRouter(prefix="/api/assistant-knowledge", tags=["assistant-knowledge"])
|
|
export_router = APIRouter(tags=["assistant-knowledge"])
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _package_root() -> Path:
|
|
return runtime_home() / "assistant-knowledge" / "packages"
|
|
|
|
|
|
def _job_package_url(job_id: str) -> str:
|
|
return f"/api/assistant-knowledge/import-jobs/{job_id}/package"
|
|
|
|
|
|
def _weknora_export_package_url(mapping_id: str, job_id: str) -> str:
|
|
return f"/api/llmwiki/knowledge-bases/{mapping_id}/export-package/{job_id}/download"
|
|
|
|
|
|
def _weknora_export_status_url(mapping_id: str, job_id: str) -> str:
|
|
return f"/api/llmwiki/knowledge-bases/{mapping_id}/export-package/{job_id}"
|
|
|
|
|
|
def _export_manifest_path(job_id: str) -> Path:
|
|
return package_path(_package_root(), job_id).with_suffix(".manifest.json")
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(UTC).isoformat()
|
|
|
|
|
|
def _normalize_export_manifest(payload: dict[str, Any]) -> dict[str, Any]:
|
|
job_id = str(payload.get("id") or "")
|
|
mapping_id = str(payload.get("mapping_id") or "")
|
|
path = package_path(_package_root(), job_id) if job_id else None
|
|
file_exists = bool(path and path.is_file())
|
|
status = str(payload.get("status") or "")
|
|
normalized = {
|
|
**payload,
|
|
"id": job_id,
|
|
"mapping_id": mapping_id,
|
|
"status": status,
|
|
"phase": str(payload.get("phase") or status or "unknown"),
|
|
"counts": payload.get("counts") if isinstance(payload.get("counts"), dict) else {},
|
|
"package_download_url": (
|
|
_weknora_export_package_url(mapping_id, job_id)
|
|
if mapping_id and job_id
|
|
else payload.get("package_download_url")
|
|
),
|
|
"status_url": (
|
|
_weknora_export_status_url(mapping_id, job_id)
|
|
if mapping_id and job_id
|
|
else payload.get("status_url")
|
|
),
|
|
"file_exists": file_exists,
|
|
"file_size_bytes": int(path.stat().st_size) if file_exists and path else int(payload.get("file_size_bytes") or 0),
|
|
}
|
|
return normalized
|
|
|
|
|
|
def _write_export_manifest(
|
|
*,
|
|
job_id: str,
|
|
mapping_id: str,
|
|
status: str,
|
|
phase: str,
|
|
counts: dict[str, Any] | None = None,
|
|
error: str | None = None,
|
|
) -> dict[str, Any]:
|
|
path = _export_manifest_path(job_id)
|
|
existing: dict[str, Any] = {}
|
|
if path.is_file():
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
existing = value if isinstance(value, dict) else {}
|
|
except (OSError, ValueError, TypeError):
|
|
existing = {}
|
|
now = _now_iso()
|
|
payload = {
|
|
**existing,
|
|
"id": job_id,
|
|
"mapping_id": mapping_id,
|
|
"status": status,
|
|
"phase": phase,
|
|
"counts": counts if counts is not None else existing.get("counts") or {},
|
|
"package_download_url": _weknora_export_package_url(mapping_id, job_id),
|
|
"status_url": _weknora_export_status_url(mapping_id, job_id),
|
|
"error": error,
|
|
"created_at": existing.get("created_at") or now,
|
|
"updated_at": now,
|
|
"completed_at": now if status in {"completed", "failed"} else existing.get("completed_at"),
|
|
}
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = _normalize_export_manifest(payload)
|
|
path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
|
|
return payload
|
|
|
|
|
|
def _read_export_manifest(job_id: str) -> dict[str, Any] | None:
|
|
path = _export_manifest_path(job_id)
|
|
if not path.is_file():
|
|
return None
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError, TypeError):
|
|
return None
|
|
return _normalize_export_manifest(value) if isinstance(value, dict) else None
|
|
|
|
|
|
def _list_export_manifests(mapping_id: str, *, limit: int) -> list[dict[str, Any]]:
|
|
root = _package_root()
|
|
if not root.is_dir():
|
|
return []
|
|
rows: list[dict[str, Any]] = []
|
|
for path in root.glob("*.manifest.json"):
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError, TypeError):
|
|
continue
|
|
if not isinstance(value, dict) or str(value.get("mapping_id") or "") != mapping_id:
|
|
continue
|
|
rows.append(_normalize_export_manifest(value))
|
|
rows.sort(key=lambda row: str(row.get("created_at") or row.get("updated_at") or ""), reverse=True)
|
|
return rows[:limit]
|
|
|
|
|
|
async def _actor(request: Request, *, admin: bool = False) -> str | None:
|
|
user = await get_optional_user_from_request(request)
|
|
if user is None:
|
|
return None
|
|
if admin and getattr(user, "system_role", None) != "admin":
|
|
raise HTTPException(status_code=403, detail="该操作仅管理员可用")
|
|
return str(user.id)
|
|
|
|
|
|
def _store(request: Request):
|
|
store = getattr(request.app.state, "assistant_knowledge_store", None)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="助手知识库存储不可用")
|
|
return store
|
|
|
|
|
|
class InitializeRequest(BaseModel):
|
|
name: str = Field(default="知识梳理总库", min_length=1, max_length=255)
|
|
|
|
|
|
class BaseCreateRequest(BaseModel):
|
|
name: str = Field(min_length=1, max_length=255)
|
|
description: str = Field(default="", max_length=2048)
|
|
|
|
|
|
class SkillImportRequest(BaseModel):
|
|
skill_names: list[str] = Field(min_length=1, max_length=200)
|
|
force: bool = False
|
|
|
|
|
|
class ExportToAssistantRequest(BaseModel):
|
|
assistant_base_id: str | None = Field(default=None, max_length=128)
|
|
|
|
|
|
@router.get("/base")
|
|
async def get_base(request: Request) -> dict[str, Any]:
|
|
await _actor(request)
|
|
return {"base": await _store(request).get_base()}
|
|
|
|
|
|
@router.post("/base/initialize", status_code=201)
|
|
async def initialize(request: Request, body: InitializeRequest) -> dict[str, Any]:
|
|
actor = await _actor(request, admin=True)
|
|
return {"base": await _store(request).initialize(created_by=actor, name=body.name)}
|
|
|
|
|
|
@router.get("/bases")
|
|
async def list_bases(request: Request) -> dict[str, Any]:
|
|
await _actor(request)
|
|
bases = await _store(request).list_bases()
|
|
return {"bases": bases, "total": len(bases)}
|
|
|
|
|
|
@router.post("/bases", status_code=201)
|
|
async def create_base(request: Request, body: BaseCreateRequest) -> dict[str, Any]:
|
|
actor = await _actor(request, admin=True)
|
|
return {
|
|
"base": await _store(request).create_base(
|
|
created_by=actor,
|
|
name=body.name,
|
|
description=body.description,
|
|
)
|
|
}
|
|
|
|
|
|
@router.get("/bases/{base_id}")
|
|
async def get_base_by_id(request: Request, base_id: str) -> dict[str, Any]:
|
|
await _actor(request)
|
|
base = await _store(request).get_base(base_id)
|
|
if base is None:
|
|
raise HTTPException(status_code=404, detail="助手知识库不存在")
|
|
return {"base": base}
|
|
|
|
|
|
@router.get("/wiki-pages")
|
|
async def list_wiki_pages(
|
|
request: Request,
|
|
base_id: str | None = Query(default=None),
|
|
q: str | None = Query(default=None),
|
|
page_type: str | None = Query(default=None),
|
|
limit: int = Query(default=100, ge=1, le=500),
|
|
) -> dict[str, Any]:
|
|
await _actor(request)
|
|
pages = await _store(request).list_pages(base_id=base_id, query=q, page_type=page_type, limit=limit)
|
|
return {"pages": pages, "total": len(pages)}
|
|
|
|
|
|
@router.get("/wiki-pages/{slug:path}/revisions/compare")
|
|
async def compare_revisions(request: Request, slug: str, left: str = Query(...), right: str = Query(...)) -> dict[str, Any]:
|
|
await _actor(request)
|
|
result = await _store(request).compare_revisions(left, right)
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="版本不存在或不属于同一页面")
|
|
return result
|
|
|
|
|
|
@router.get("/wiki-pages/{slug:path}/revisions")
|
|
async def list_revisions(request: Request, slug: str, base_id: str | None = Query(default=None)) -> dict[str, Any]:
|
|
await _actor(request)
|
|
return {"revisions": await _store(request).list_revisions(slug, base_id=base_id)}
|
|
|
|
|
|
@router.post("/wiki-pages/{slug:path}/revisions/{revision_id}/rollback")
|
|
async def rollback_revision(
|
|
request: Request,
|
|
slug: str,
|
|
revision_id: str,
|
|
base_id: str | None = Query(default=None),
|
|
) -> dict[str, Any]:
|
|
actor = await _actor(request, admin=True)
|
|
row = await _store(request).rollback(slug=slug, revision_id=revision_id, actor=actor, base_id=base_id)
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="页面或版本不存在")
|
|
return row
|
|
|
|
|
|
@router.get("/wiki-pages/{slug:path}")
|
|
async def get_wiki_page(request: Request, slug: str, base_id: str | None = Query(default=None)) -> dict[str, Any]:
|
|
await _actor(request)
|
|
row = await _store(request).get_page(slug, base_id=base_id)
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="Wiki 页面不存在")
|
|
return row
|
|
|
|
|
|
@router.get("/search")
|
|
async def search(
|
|
request: Request,
|
|
q: str = Query(..., min_length=1),
|
|
base_id: str | None = Query(default=None),
|
|
page_type: str | None = Query(default=None),
|
|
) -> dict[str, Any]:
|
|
await _actor(request)
|
|
rows = await _store(request).list_pages(base_id=base_id, query=q, page_type=page_type, limit=100)
|
|
return {"results": rows, "total": len(rows)}
|
|
|
|
|
|
@router.get("/sources")
|
|
async def list_sources(request: Request, base_id: str | None = Query(default=None)) -> dict[str, Any]:
|
|
await _actor(request)
|
|
return {"sources": await _store(request).list_sources(base_id=base_id)}
|
|
|
|
|
|
@router.get("/entities/{entity_id}")
|
|
async def get_entity(request: Request, entity_id: str) -> dict[str, Any]:
|
|
await _actor(request)
|
|
row = await _store(request).get_entity(entity_id)
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="实体不存在")
|
|
return row
|
|
|
|
|
|
@router.get("/relations/{relation_id}")
|
|
async def get_relation(request: Request, relation_id: str) -> dict[str, Any]:
|
|
await _actor(request)
|
|
row = await _store(request).get_relation(relation_id)
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="关系不存在")
|
|
return row
|
|
|
|
|
|
@router.delete("/sources/{source_id}", status_code=204)
|
|
async def delete_source(request: Request, source_id: str) -> None:
|
|
await _actor(request, admin=True)
|
|
if not await _store(request).delete_source(source_id):
|
|
raise HTTPException(status_code=404, detail="来源不存在")
|
|
|
|
|
|
@router.get("/import-jobs")
|
|
async def list_import_jobs(
|
|
request: Request,
|
|
base_id: str | None = Query(default=None),
|
|
limit: int = Query(default=100, ge=1, le=500),
|
|
) -> dict[str, Any]:
|
|
await _actor(request)
|
|
jobs = await _store(request).list_jobs(limit, base_id=base_id)
|
|
for job in jobs:
|
|
job["package_download_url"] = _job_package_url(str(job["id"]))
|
|
return {"jobs": jobs}
|
|
|
|
|
|
@router.get("/import-jobs/{job_id}")
|
|
async def get_import_job(request: Request, job_id: str) -> dict[str, Any]:
|
|
await _actor(request)
|
|
row = await _store(request).get_job(job_id)
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="导入任务不存在")
|
|
row["package_download_url"] = _job_package_url(job_id)
|
|
return row
|
|
|
|
|
|
@router.get("/import-jobs/{job_id}/package")
|
|
async def download_import_package(request: Request, job_id: str) -> FileResponse:
|
|
await _actor(request, admin=True)
|
|
row = await _store(request).get_job(job_id)
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="导入任务不存在")
|
|
path = package_path(_package_root(), job_id)
|
|
if not path.is_file():
|
|
raise HTTPException(status_code=404, detail="导出包尚未生成或已被清理")
|
|
return FileResponse(
|
|
path,
|
|
media_type="application/x-ndjson",
|
|
filename=f"weknora-assistant-knowledge-{job_id}.jsonl",
|
|
)
|
|
|
|
|
|
@router.post("/imports/skills", status_code=202)
|
|
async def import_skills(request: Request, _body: SkillImportRequest) -> dict[str, Any]:
|
|
await _actor(request, admin=True)
|
|
raise HTTPException(status_code=410, detail="不支持技能直接归纳到助手知识库，请先归纳到 WeKnora 普通知识库后再导入助手知识库")
|
|
|
|
|
|
@router.post("/imports/files", status_code=410)
|
|
async def import_files(request: Request) -> dict[str, Any]:
|
|
await _actor(request, admin=True)
|
|
raise HTTPException(status_code=410, detail="助手知识库不支持导入任意本地文件，请从 WeKnora 普通知识库导出后导入")
|
|
|
|
|
|
@router.post("/imports/weknora-package", status_code=202)
|
|
async def import_weknora_package_file(
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
assistant_base_id: str | None = Form(default=None),
|
|
) -> dict[str, Any]:
|
|
actor = await _actor(request, admin=True)
|
|
upload_id = str(uuid4())
|
|
filename = (file.filename or "weknora-assistant-knowledge.jsonl").strip()
|
|
job = await _store(request).create_queued_import_job(
|
|
source_type="weknora",
|
|
source_key=f"upload:{upload_id}",
|
|
source_name=filename,
|
|
trigger="manual_package_upload",
|
|
created_by=actor,
|
|
metadata={
|
|
"type": "weknora_export_package",
|
|
"filename": filename,
|
|
"upload_id": upload_id,
|
|
},
|
|
upload_batch_id=upload_id,
|
|
base_id=assistant_base_id,
|
|
phase="uploading",
|
|
)
|
|
path = package_path(_package_root(), str(job["id"]))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
size = 0
|
|
try:
|
|
with path.open("wb") as handle:
|
|
while True:
|
|
chunk = await file.read(1024 * 1024)
|
|
if not chunk:
|
|
break
|
|
size += len(chunk)
|
|
handle.write(chunk)
|
|
finally:
|
|
await file.close()
|
|
if size <= 0:
|
|
await _store(request).update_import_job(
|
|
str(job["id"]),
|
|
status="failed",
|
|
phase="failed",
|
|
error="上传文件为空",
|
|
completed=True,
|
|
)
|
|
raise HTTPException(status_code=422, detail="上传文件为空")
|
|
await _store(request).update_import_job(
|
|
str(job["id"]),
|
|
status="queued",
|
|
phase="queued",
|
|
counts={"uploaded_bytes": size},
|
|
)
|
|
asyncio.create_task(
|
|
_import_package_file_for_job(
|
|
request=request,
|
|
job_id=str(job["id"]),
|
|
path=path,
|
|
actor=actor,
|
|
)
|
|
)
|
|
job["package_download_url"] = _job_package_url(str(job["id"]))
|
|
return job
|
|
|
|
|
|
async def _write_weknora_export_package(
|
|
client: Any,
|
|
remote_id: str,
|
|
source: dict[str, Any],
|
|
path: Path,
|
|
*,
|
|
vectors: AsyncIterable[dict[str, Any]] | None = None,
|
|
) -> dict[str, int]:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp_path = path.with_suffix(path.suffix + ".tmp")
|
|
counts: dict[str, int] = {
|
|
"wiki_page": 0,
|
|
"document": 0,
|
|
"chunk": 0,
|
|
"vector": 0,
|
|
"entity": 0,
|
|
"relation": 0,
|
|
"graph": 0,
|
|
}
|
|
with tmp_path.open("w", encoding="utf-8", newline="\n") as handle:
|
|
write_jsonl_record(
|
|
handle,
|
|
"source",
|
|
{
|
|
**source,
|
|
"package_version": PACKAGE_VERSION,
|
|
"export_format": "assistant-knowledge-jsonl",
|
|
},
|
|
)
|
|
|
|
# Wiki pages first: these are the human-readable pages that the
|
|
# assistant knowledge base presents in wiki form.
|
|
page_no = 1
|
|
while True:
|
|
page = await client.list_wiki_pages(remote_id, page=page_no, page_size=100)
|
|
batch = page.get("pages") or []
|
|
for row in batch:
|
|
if isinstance(row, dict):
|
|
payload = row
|
|
slug = str(row.get("slug") or row.get("wiki_slug") or row.get("path") or "")
|
|
if slug:
|
|
try:
|
|
detail = await client.get_wiki_page(remote_id, slug)
|
|
if isinstance(detail, dict):
|
|
payload = {**row, **detail}
|
|
except WeKnoraError:
|
|
logger.warning(
|
|
"Could not fetch full WeKnora wiki page during export: kb=%s slug=%s",
|
|
remote_id,
|
|
slug,
|
|
exc_info=True,
|
|
)
|
|
write_jsonl_record(handle, "wiki_page", payload)
|
|
counts["wiki_page"] += 1
|
|
if counts["wiki_page"] >= int(page.get("total") or counts["wiki_page"]) or not batch:
|
|
break
|
|
page_no += 1
|
|
|
|
# Then export WeKnora's processed knowledge and chunks. Chunks are the
|
|
# important boundary for large RAG imports: the assistant side should
|
|
# consume the already-parsed/segmented material instead of re-reading
|
|
# arbitrary source files.
|
|
document_count = 0
|
|
page_no = 1
|
|
while True:
|
|
page = await client.list_documents(remote_id, page=page_no, page_size=100)
|
|
batch = page.get("items") or []
|
|
for document in batch:
|
|
if not isinstance(document, dict):
|
|
continue
|
|
write_jsonl_record(handle, "document", document)
|
|
counts["document"] += 1
|
|
document_count += 1
|
|
document_id = str(document.get("id") or document.get("knowledge_id") or "")
|
|
if not document_id:
|
|
continue
|
|
chunk_page = 1
|
|
while True:
|
|
values = await client.list_chunks(document_id, page=chunk_page, page_size=100)
|
|
chunk_batch = values.get("items") or []
|
|
for row in chunk_batch:
|
|
if isinstance(row, dict):
|
|
write_jsonl_record(handle, "chunk", {**row, "source_document_id": document_id})
|
|
counts["chunk"] += 1
|
|
if len(chunk_batch) < 100:
|
|
break
|
|
chunk_page += 1
|
|
if document_count >= int(page.get("total") or document_count) or not batch:
|
|
break
|
|
page_no += 1
|
|
|
|
if vectors is not None:
|
|
async for row in vectors:
|
|
if isinstance(row, dict):
|
|
write_jsonl_record(handle, "vector", row)
|
|
counts["vector"] += 1
|
|
|
|
try:
|
|
graph = await client.get_wiki_graph(remote_id, limit=10000)
|
|
except WeKnoraError:
|
|
graph = {"nodes": [], "edges": [], "capability_warning": "graph API unavailable"}
|
|
nodes = [row for row in graph.get("nodes") or [] if isinstance(row, dict)]
|
|
edges = [row for row in graph.get("edges") or [] if isinstance(row, dict)]
|
|
write_jsonl_record(handle, "graph", {**graph, "nodes": nodes, "edges": edges})
|
|
counts["graph"] += 1
|
|
for row in nodes:
|
|
if row.get("id") and (row.get("title") or row.get("name")):
|
|
write_jsonl_record(
|
|
handle,
|
|
"entity",
|
|
{
|
|
"id": row.get("id"),
|
|
"name": row.get("title") or row.get("name"),
|
|
"type": row.get("type") or "concept",
|
|
"confidence": 1.0,
|
|
},
|
|
)
|
|
counts["entity"] += 1
|
|
for row in edges:
|
|
write_jsonl_record(
|
|
handle,
|
|
"relation",
|
|
{
|
|
"source_entity_id": row.get("source"),
|
|
"target_entity_id": row.get("target"),
|
|
"predicate": row.get("predicate") or "references",
|
|
"confidence": 1.0,
|
|
"evidence": [{"source": "weknora_graph"}],
|
|
},
|
|
)
|
|
counts["relation"] += 1
|
|
tmp_path.replace(path)
|
|
return counts
|
|
|
|
|
|
async def _local_wiki_vector_context(request: Request, mapping_id: str) -> tuple[Any | None, str, str]:
|
|
store = getattr(request.app.state, "llmwiki_index_store", None)
|
|
if store is None:
|
|
return None, "", ""
|
|
embedding = getattr(request.app.state, "llmwiki_embedding", None)
|
|
fingerprint = str(getattr(embedding, "fingerprint", "") or "")
|
|
if not fingerprint:
|
|
try:
|
|
_revision, state_fingerprint, _state = await store.get_index_revision(mapping_id)
|
|
fingerprint = str(state_fingerprint or "")
|
|
except Exception: # noqa: BLE001
|
|
fingerprint = ""
|
|
if not fingerprint:
|
|
return store, "", ""
|
|
model = str(getattr(getattr(embedding, "config", None), "model", "") or "")
|
|
return store, fingerprint, model
|
|
|
|
|
|
def _local_wiki_vector_export_payload(
|
|
row: dict[str, Any],
|
|
*,
|
|
model: str,
|
|
fingerprint: str,
|
|
) -> dict[str, Any] | None:
|
|
blob = row.get("vector_blob")
|
|
if isinstance(blob, memoryview):
|
|
blob = blob.tobytes()
|
|
if not isinstance(blob, (bytes, bytearray)):
|
|
return None
|
|
page = row.get("page") if isinstance(row.get("page"), dict) else {}
|
|
return {
|
|
"id": row.get("id"),
|
|
"wiki_slug": str(page.get("slug") or "").strip("/"),
|
|
"wiki_page_id": page.get("id") or page.get("remote_page_id"),
|
|
"title": page.get("title") or "",
|
|
"section_index": int(row.get("section_index") or 0),
|
|
"heading": row.get("heading"),
|
|
"section_content": row.get("section_content") or "",
|
|
"content_hash": row.get("content_hash") or "",
|
|
"embedding_model": model,
|
|
"embedding_fingerprint": row.get("embedding_fingerprint") or fingerprint,
|
|
"embedding_dimensions": int(row.get("embedding_dimensions") or 0),
|
|
"vector_encoding": "float32-le-normalized-base64",
|
|
"vector_blob_base64": base64.b64encode(bytes(blob)).decode("ascii"),
|
|
}
|
|
|
|
|
|
async def _iter_local_wiki_vectors_for_export(
|
|
request: Request,
|
|
mapping_id: str,
|
|
*,
|
|
batch_size: int = 1000,
|
|
) -> AsyncIterator[dict[str, Any]]:
|
|
store, fingerprint, model = await _local_wiki_vector_context(request, mapping_id)
|
|
if store is None or not fingerprint:
|
|
return
|
|
try:
|
|
async for row in store.iter_vector_snapshot_rows(mapping_id, fingerprint, batch_size=batch_size):
|
|
if not isinstance(row, dict):
|
|
continue
|
|
payload = _local_wiki_vector_export_payload(row, model=model, fingerprint=fingerprint)
|
|
if payload is not None:
|
|
yield payload
|
|
except Exception: # noqa: BLE001
|
|
logger.debug("No local wiki vector snapshot available for %s", mapping_id, exc_info=True)
|
|
|
|
|
|
async def _has_local_wiki_vectors_for_export(request: Request, mapping_id: str) -> bool:
|
|
async for _row in _iter_local_wiki_vectors_for_export(request, mapping_id, batch_size=1):
|
|
return True
|
|
return False
|
|
|
|
|
|
async def _load_local_wiki_vectors_for_export(request: Request, mapping_id: str) -> list[dict[str, Any]]:
|
|
return [row async for row in _iter_local_wiki_vectors_for_export(request, mapping_id)]
|
|
|
|
|
|
async def _ensure_local_wiki_vectors_for_export(
|
|
request: Request,
|
|
mapping: dict[str, Any],
|
|
) -> None:
|
|
mapping_id = str(mapping.get("id") or mapping.get("knowledge_base_id") or "")
|
|
if not mapping_id or await _has_local_wiki_vectors_for_export(request, mapping_id):
|
|
return
|
|
|
|
service = getattr(request.app.state, "llmwiki_sync_service", None)
|
|
if service is None or not mapping_id:
|
|
return
|
|
sync_mapping = dict(mapping)
|
|
sync_mapping.setdefault("id", mapping_id)
|
|
sync_mapping.setdefault(
|
|
"weknora_id",
|
|
mapping.get("remote_knowledge_base_id") or mapping.get("weknora_id") or "",
|
|
)
|
|
if not sync_mapping.get("weknora_id"):
|
|
return
|
|
try:
|
|
logger.info("Building local wiki vectors before export: mapping=%s", mapping_id)
|
|
await service.sync_mapping(sync_mapping, force=False)
|
|
except Exception: # noqa: BLE001
|
|
logger.warning(
|
|
"Could not build local wiki vectors before export: mapping=%s",
|
|
mapping_id,
|
|
exc_info=True,
|
|
)
|
|
|
|
|
|
async def _import_package_file_for_job(
|
|
*,
|
|
request: Request,
|
|
job_id: str,
|
|
path: Path,
|
|
actor: str | None,
|
|
) -> None:
|
|
store = _store(request)
|
|
try:
|
|
await store.update_import_job(job_id, status="running", phase="loading_package")
|
|
package = await asyncio.to_thread(read_package, path)
|
|
await store.update_import_job(job_id, status="running", phase="importing")
|
|
await store.import_package_for_job(
|
|
job_id,
|
|
package=package,
|
|
created_by=actor,
|
|
mirror_to_global=True,
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.exception("Assistant knowledge package import failed: job=%s", job_id)
|
|
await store.update_import_job(
|
|
job_id,
|
|
status="failed",
|
|
phase="failed",
|
|
error=str(exc),
|
|
completed=True,
|
|
)
|
|
|
|
|
|
async def _export_weknora_then_import_for_job(
|
|
*,
|
|
request: Request,
|
|
job_id: str,
|
|
mapping: dict[str, Any],
|
|
remote_id: str,
|
|
source: dict[str, Any],
|
|
actor: str | None,
|
|
) -> None:
|
|
store = _store(request)
|
|
path = package_path(_package_root(), job_id)
|
|
try:
|
|
await store.update_import_job(job_id, status="running", phase="exporting")
|
|
runtime = get_resolved_llmwiki_runtime(request.app.state.config)
|
|
if not runtime.weknora_enabled:
|
|
raise RuntimeError("未配置普通知识库服务")
|
|
mapping_id = str(mapping.get("id") or mapping.get("knowledge_base_id") or "")
|
|
await _ensure_local_wiki_vectors_for_export(request, mapping)
|
|
counts = await _write_weknora_export_package(
|
|
build_weknora_client(runtime),
|
|
remote_id,
|
|
source,
|
|
path,
|
|
vectors=_iter_local_wiki_vectors_for_export(request, mapping_id),
|
|
)
|
|
await store.update_import_job(job_id, status="running", phase="package_ready", counts=counts)
|
|
await _import_package_file_for_job(request=request, job_id=job_id, path=path, actor=actor)
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.exception("Assistant knowledge WeKnora export failed: job=%s", job_id)
|
|
await store.update_import_job(
|
|
job_id,
|
|
status="failed",
|
|
phase="failed",
|
|
error=str(exc),
|
|
completed=True,
|
|
)
|
|
|
|
|
|
async def _export_weknora_package(client: Any, remote_id: str, source: dict[str, Any]) -> dict[str, Any]:
|
|
"""Legacy in-memory export shape, kept for tests and older callers."""
|
|
pages: list[dict[str, Any]] = []
|
|
page_no = 1
|
|
while True:
|
|
page = await client.list_wiki_pages(remote_id, page=page_no, page_size=100)
|
|
batch = page.get("pages") or []
|
|
pages.extend(batch)
|
|
if len(pages) >= int(page.get("total") or len(pages)) or not batch:
|
|
break
|
|
page_no += 1
|
|
documents: list[dict[str, Any]] = []
|
|
chunks: list[dict[str, Any]] = []
|
|
page_no = 1
|
|
while True:
|
|
page = await client.list_documents(remote_id, page=page_no, page_size=100)
|
|
batch = page.get("items") or []
|
|
documents.extend(batch)
|
|
for document in batch:
|
|
document_id = str(document.get("id") or document.get("knowledge_id") or "")
|
|
if not document_id:
|
|
continue
|
|
chunk_page = 1
|
|
while True:
|
|
values = await client.list_chunks(document_id, page=chunk_page, page_size=100)
|
|
chunk_batch = values.get("items") or []
|
|
chunks.extend([{**row, "source_document_id": document_id} for row in chunk_batch])
|
|
if len(chunk_batch) < 100:
|
|
break
|
|
chunk_page += 1
|
|
if len(documents) >= int(page.get("total") or len(documents)) or not batch:
|
|
break
|
|
page_no += 1
|
|
try:
|
|
graph = await client.get_wiki_graph(remote_id, limit=10000)
|
|
except WeKnoraError:
|
|
graph = {"nodes": [], "edges": [], "capability_warning": "graph API unavailable"}
|
|
entities = [{"id": row.get("id"), "name": row.get("title") or row.get("name"), "type": row.get("type") or "concept", "confidence": 1.0} for row in graph.get("nodes") or [] if row.get("id") and (row.get("title") or row.get("name"))]
|
|
relations = [
|
|
{"source_entity_id": row.get("source"), "target_entity_id": row.get("target"), "predicate": row.get("predicate") or "references", "confidence": 1.0, "evidence": [{"source": "weknora_graph"}]} for row in graph.get("edges") or []
|
|
]
|
|
return {"source": source, "wiki_pages": pages, "documents": documents, "chunks": chunks, "entities": entities, "relations": relations, "graph": graph}
|
|
|
|
|
|
async def _export_weknora_package_only_job(
|
|
*,
|
|
request: Request,
|
|
job_id: str,
|
|
mapping: dict[str, Any],
|
|
mapping_id: str,
|
|
remote_id: str,
|
|
source: dict[str, Any],
|
|
) -> None:
|
|
try:
|
|
_write_export_manifest(job_id=job_id, mapping_id=mapping_id, status="running", phase="exporting")
|
|
runtime = get_resolved_llmwiki_runtime(request.app.state.config)
|
|
if not runtime.weknora_enabled:
|
|
raise RuntimeError("未配置普通知识库服务")
|
|
await _ensure_local_wiki_vectors_for_export(request, mapping)
|
|
counts = await _write_weknora_export_package(
|
|
build_weknora_client(runtime),
|
|
remote_id,
|
|
source,
|
|
package_path(_package_root(), job_id),
|
|
vectors=_iter_local_wiki_vectors_for_export(request, mapping_id),
|
|
)
|
|
_write_export_manifest(
|
|
job_id=job_id,
|
|
mapping_id=mapping_id,
|
|
status="completed",
|
|
phase="package_ready",
|
|
counts=counts,
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.exception("WeKnora vectorized export package failed: mapping=%s job=%s", mapping_id, job_id)
|
|
_write_export_manifest(
|
|
job_id=job_id,
|
|
mapping_id=mapping_id,
|
|
status="failed",
|
|
phase="failed",
|
|
error=str(exc),
|
|
)
|
|
|
|
|
|
@export_router.get("/api/llmwiki/knowledge-bases/{mapping_id}/export-package")
|
|
async def list_weknora_export_packages(
|
|
request: Request,
|
|
mapping_id: str,
|
|
limit: int = Query(default=50, ge=1, le=200),
|
|
) -> dict[str, Any]:
|
|
actor = await _actor(request, admin=True)
|
|
mapping = await request.app.state.llmwiki_store.get_authorized(
|
|
mapping_id,
|
|
actor or "system",
|
|
write=False,
|
|
is_admin=True,
|
|
)
|
|
if mapping is None:
|
|
raise HTTPException(status_code=404, detail="普通知识库不存在")
|
|
rows = _list_export_manifests(mapping_id, limit=limit)
|
|
return {"jobs": rows, "total": len(rows)}
|
|
|
|
|
|
@export_router.post("/api/llmwiki/knowledge-bases/{mapping_id}/export-package", status_code=202)
|
|
async def create_weknora_export_package(request: Request, mapping_id: str) -> dict[str, Any]:
|
|
actor = await _actor(request, admin=True)
|
|
runtime = get_resolved_llmwiki_runtime(request.app.state.config)
|
|
if not runtime.weknora_enabled:
|
|
raise HTTPException(status_code=503, detail="未配置普通知识库服务")
|
|
mapping = await request.app.state.llmwiki_store.get_authorized(
|
|
mapping_id,
|
|
actor or "system",
|
|
write=False,
|
|
is_admin=True,
|
|
)
|
|
if mapping is None:
|
|
raise HTTPException(status_code=404, detail="普通知识库不存在")
|
|
job_id = str(uuid4())
|
|
source = {
|
|
"type": "weknora",
|
|
"knowledge_base_id": mapping_id,
|
|
"remote_knowledge_base_id": mapping["weknora_id"],
|
|
"knowledge_base_name": mapping["name"],
|
|
"package_kind": "weknora_vectorized_export",
|
|
}
|
|
payload = _write_export_manifest(job_id=job_id, mapping_id=mapping_id, status="queued", phase="queued")
|
|
asyncio.create_task(
|
|
_export_weknora_package_only_job(
|
|
request=request,
|
|
job_id=job_id,
|
|
mapping=mapping,
|
|
mapping_id=mapping_id,
|
|
remote_id=str(mapping["weknora_id"]),
|
|
source=source,
|
|
)
|
|
)
|
|
return payload
|
|
|
|
|
|
@export_router.get("/api/llmwiki/knowledge-bases/{mapping_id}/export-package/{job_id}")
|
|
async def get_weknora_export_package_status(request: Request, mapping_id: str, job_id: str) -> dict[str, Any]:
|
|
await _actor(request, admin=True)
|
|
payload = _read_export_manifest(job_id)
|
|
if payload is None or str(payload.get("mapping_id") or "") != mapping_id:
|
|
raise HTTPException(status_code=404, detail="导出任务不存在")
|
|
return payload
|
|
|
|
|
|
@export_router.get("/api/llmwiki/knowledge-bases/{mapping_id}/export-package/{job_id}/download")
|
|
async def download_weknora_export_package(request: Request, mapping_id: str, job_id: str) -> FileResponse:
|
|
await _actor(request, admin=True)
|
|
payload = _read_export_manifest(job_id)
|
|
if payload is None or str(payload.get("mapping_id") or "") != mapping_id:
|
|
raise HTTPException(status_code=404, detail="导出任务不存在")
|
|
if str(payload.get("status") or "") != "completed":
|
|
raise HTTPException(status_code=409, detail="导出包尚未生成完成")
|
|
path = package_path(_package_root(), job_id)
|
|
if not path.is_file():
|
|
raise HTTPException(status_code=404, detail="导出包已不存在")
|
|
return FileResponse(
|
|
path,
|
|
media_type="application/x-ndjson",
|
|
filename=f"weknora-vectorized-knowledge-{mapping_id}-{job_id}.jsonl",
|
|
)
|
|
|
|
|
|
@export_router.post("/api/llmwiki/knowledge-bases/{mapping_id}/export-to-assistant", status_code=202)
|
|
async def export_to_assistant(
|
|
request: Request,
|
|
mapping_id: str,
|
|
body: ExportToAssistantRequest | None = None,
|
|
) -> dict[str, Any]:
|
|
actor = await _actor(request, admin=True)
|
|
runtime = get_resolved_llmwiki_runtime(request.app.state.config)
|
|
if not runtime.weknora_enabled:
|
|
raise HTTPException(status_code=503, detail="未配置普通知识库服务")
|
|
mapping = await request.app.state.llmwiki_store.get_authorized(mapping_id, actor or "system", write=False, is_admin=True)
|
|
if mapping is None:
|
|
raise HTTPException(status_code=404, detail="普通知识库不存在")
|
|
source = {
|
|
"type": "weknora",
|
|
"knowledge_base_id": mapping_id,
|
|
"remote_knowledge_base_id": mapping["weknora_id"],
|
|
"knowledge_base_name": mapping["name"],
|
|
}
|
|
try:
|
|
job = await _store(request).create_queued_import_job(
|
|
source_type="weknora",
|
|
source_key=mapping_id,
|
|
source_name=str(mapping["name"]),
|
|
trigger="manual_from_weknora",
|
|
created_by=actor,
|
|
metadata=source,
|
|
mapping_id=mapping_id,
|
|
base_id=body.assistant_base_id if body else None,
|
|
phase="exporting",
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
job["package_download_url"] = _job_package_url(str(job["id"]))
|
|
asyncio.create_task(
|
|
_export_weknora_then_import_for_job(
|
|
request=request,
|
|
job_id=str(job["id"]),
|
|
mapping=mapping,
|
|
remote_id=str(mapping["weknora_id"]),
|
|
source=source,
|
|
actor=actor,
|
|
)
|
|
)
|
|
return job
|
|
|
|
|
|
@router.post("/sources/{source_id}/reimport", status_code=202)
|
|
async def reimport_source(request: Request, source_id: str) -> dict[str, Any]:
|
|
actor = await _actor(request, admin=True)
|
|
store = _store(request)
|
|
source = await store.get_source(source_id)
|
|
if source is None:
|
|
raise HTTPException(status_code=404, detail="来源不存在")
|
|
if source["source_type"] == "skill":
|
|
raise HTTPException(status_code=410, detail="不支持技能来源直接重新导入助手知识库，请重新从 WeKnora 普通知识库导入")
|
|
if source["source_type"] != "weknora":
|
|
raise HTTPException(status_code=409, detail="本地文件来源未保留原文件，请重新上传")
|
|
mapping_id = str(source.get("knowledge_base_mapping_id") or source["source_key"])
|
|
if not mapping_id or mapping_id.startswith("upload:"):
|
|
raise HTTPException(status_code=409, detail="上传包来源不能直接重新导入，请重新上传 WeKnora 导出包")
|
|
runtime = get_resolved_llmwiki_runtime(request.app.state.config)
|
|
if not runtime.weknora_enabled:
|
|
raise HTTPException(status_code=503, detail="未配置普通知识库服务")
|
|
mapping = await request.app.state.llmwiki_store.get_authorized(
|
|
mapping_id,
|
|
actor or "system",
|
|
write=False,
|
|
is_admin=True,
|
|
)
|
|
if mapping is None:
|
|
raise HTTPException(status_code=404, detail="来源普通知识库不存在")
|
|
source_payload = {
|
|
"type": "weknora",
|
|
"knowledge_base_id": mapping_id,
|
|
"remote_knowledge_base_id": mapping["weknora_id"],
|
|
"knowledge_base_name": mapping["name"],
|
|
}
|
|
try:
|
|
job = await store.create_queued_import_job(
|
|
source_type="weknora",
|
|
source_key=mapping_id,
|
|
source_name=str(mapping["name"]),
|
|
trigger="source_reimport",
|
|
created_by=actor,
|
|
metadata=source_payload,
|
|
mapping_id=mapping_id,
|
|
base_id=str(source["base_id"]),
|
|
phase="exporting",
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
job["package_download_url"] = _job_package_url(str(job["id"]))
|
|
asyncio.create_task(
|
|
_export_weknora_then_import_for_job(
|
|
request=request,
|
|
job_id=str(job["id"]),
|
|
mapping=mapping,
|
|
remote_id=str(mapping["weknora_id"]),
|
|
source=source_payload,
|
|
actor=actor,
|
|
)
|
|
)
|
|
return job
|