From 0b88db956fac12e76476f2e79dffae62fe374586 Mon Sep 17 00:00:00 2001 From: qyyfw Date: Thu, 2 Jul 2026 22:39:10 +0800 Subject: [PATCH] Phase 2.4: Knowledge Base Admin API - CRUD, stats, external service integration --- .../python/app/api/admin/knowledge_admin.py | 1330 +++++++++++++++++ backend/python/app/models.py | 24 +- backend/python/main.py | 4 +- 3 files changed, 1356 insertions(+), 2 deletions(-) create mode 100644 backend/python/app/api/admin/knowledge_admin.py diff --git a/backend/python/app/api/admin/knowledge_admin.py b/backend/python/app/api/admin/knowledge_admin.py new file mode 100644 index 0000000..3445981 --- /dev/null +++ b/backend/python/app/api/admin/knowledge_admin.py @@ -0,0 +1,1330 @@ +""" +知识库管理 API（管理员） +Phase 2.4: AI PaaS Knowledge Base Management + External KB Integration +""" +from fastapi import APIRouter, Depends, HTTPException, Query, Body +from sqlalchemy.orm import Session +from sqlalchemy import or_, func, desc, asc +from pydantic import BaseModel, Field +from typing import Optional, List, Dict, Any +from datetime import datetime +import uuid +import json +import logging +import httpx + +from ...database import get_db +from ...models import ( + KnowledgeBase, KnowledgeDocument, KnowledgeDirectory, + KnowledgeGraphNode, KnowledgeGraphEdge, KnowledgeGraphBuildLog, + User, gen_uuid +) +from ...auth import get_current_admin +from ...middleware.log_service import LogService + +router = APIRouter() +logger = logging.getLogger("aiflow") + + +# ============================================================ +# Pydantic 模型 +# ============================================================ + +class ExternalServiceConfig(BaseModel): + name: str = Field(min_length=1, max_length=200) + service_type: str = Field(..., description="fastgpt / dify / ragflow") + api_url: str = Field(min_length=1, max_length=500) + api_key: Optional[str] = None + config: Optional[Dict[str, Any]] = None + + +class ExternalServiceUpdate(BaseModel): + name: Optional[str] = None + api_url: Optional[str] = None + api_key: Optional[str] = None + config: Optional[Dict[str, Any]] = None + is_enabled: Optional[bool] = None + + +class KBCreate(BaseModel): + name: str = Field(min_length=1, max_length=200) + description: Optional[str] = None + is_official: bool = False + external_service_id: Optional[str] = None + external_config: Optional[Dict[str, Any]] = None + + +class KBUpdate(BaseModel): + name: Optional[str] = None + description: Optional[str] = None + is_official: Optional[bool] = None + status: Optional[str] = None # active / inactive / archived + external_service_id: Optional[str] = None + external_config: Optional[Dict[str, Any]] = None + + +class SyncRequest(BaseModel): + external_service_id: Optional[str] = None # override KB's linked service + sync_direction: str = "pull" # pull / push / bidirectional + options: Optional[Dict[str, Any]] = None + + +# ============================================================ +# 辅助函数 +# ============================================================ + +def kb_to_dict(kb: KnowledgeBase, db: Session = None) -> dict: + """将 KnowledgeBase 转换为字典""" + result = { + "id": kb.id, + "user_id": kb.user_id, + "name": kb.name, + "description": kb.description, + "directory_path": kb.directory_path, + "document_count": kb.document_count or 0, + "knowledge_graph_count": kb.knowledge_graph_count or 0, + "created_at": kb.created_at.isoformat() if kb.created_at else None, + "updated_at": kb.updated_at.isoformat() if kb.updated_at else None, + } + + # 新增字段（Phase 2.4） + # 通过 getattr 安全获取，因为旧记录可能没有这些字段 + result["is_official"] = getattr(kb, "is_official", False) + result["status"] = getattr(kb, "status", "active") + result["external_service_id"] = getattr(kb, "external_service_id", None) + result["external_id"] = getattr(kb, "external_id", None) + result["external_sync_status"] = getattr(kb, "external_sync_status", "none") + result["external_config"] = None + ext_config_raw = getattr(kb, "external_config", None) + if ext_config_raw: + try: + result["external_config"] = json.loads(ext_config_raw) if isinstance(ext_config_raw, str) else ext_config_raw + except Exception: + result["external_config"] = {} + + # 查询关联用户名 + if db and kb.user_id: + user = db.query(User).filter(User.id == kb.user_id).first() + result["owner_name"] = user.username if user else "未知" + + # 计算存储占用 + if db: + total_size = db.query(func.sum(KnowledgeDocument.file_size)).filter( + KnowledgeDocument.knowledge_base_id == kb.id + ).scalar() or 0 + result["total_storage_bytes"] = int(total_size) + result["total_storage_mb"] = round(int(total_size) / (1024 * 1024), 2) + + return result + + +def doc_to_dict(doc: KnowledgeDocument) -> dict: + """将 KnowledgeDocument 转换为字典""" + return { + "id": doc.id, + "knowledge_base_id": doc.knowledge_base_id, + "directory_id": doc.directory_id, + "filename": doc.filename, + "file_type": doc.file_type, + "file_size": doc.file_size or 0, + "file_path": doc.file_path, + "status": doc.status, + "chunk_count": doc.chunk_count or 0, + "error_message": doc.error_message, + "created_at": doc.created_at.isoformat() if doc.created_at else None, + "updated_at": doc.updated_at.isoformat() if doc.updated_at else None, + } + + +def format_file_size(size_bytes: int) -> str: + """格式化文件大小""" + if size_bytes < 1024: + return f"{size_bytes} B" + elif size_bytes < 1024 * 1024: + return f"{size_bytes / 1024:.1f} KB" + elif size_bytes < 1024 * 1024 * 1024: + return f"{size_bytes / (1024 * 1024):.1f} MB" + else: + return f"{size_bytes / (1024 * 1024 * 1024):.1f} GB" + + +# ============================================================ +# 统计概览 +# ============================================================ + +@router.get("/stats/overview", summary="知识库统计概览") +async def get_kb_stats( + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取知识库全局统计数据""" + # 基础统计 + total_kb = db.query(func.count(KnowledgeBase.id)).scalar() or 0 + total_docs = db.query(func.count(KnowledgeDocument.id)).scalar() or 0 + total_storage = db.query(func.sum(KnowledgeDocument.file_size)).scalar() or 0 + total_nodes = db.query(func.count(KnowledgeGraphNode.id)).scalar() or 0 + total_edges = db.query(func.count(KnowledgeGraphEdge.id)).scalar() or 0 + + # 官方 vs 用户知识库 + official_kb = 0 + user_kb = 0 + try: + official_kb = db.query(func.count(KnowledgeBase.id)).filter( + getattr(KnowledgeBase, "is_official", False) == True + ).scalar() or 0 + except Exception: + official_kb = 0 + user_kb = total_kb - official_kb + + # 文档状态分布 + doc_status_dist = {} + statuses = ["uploading", "processing", "completed", "failed"] + for s in statuses: + cnt = db.query(func.count(KnowledgeDocument.id)).filter( + KnowledgeDocument.status == s + ).scalar() or 0 + doc_status_dist[s] = cnt + + # 文件类型分布 + file_type_dist = {} + type_rows = db.query( + KnowledgeDocument.file_type, + func.count(KnowledgeDocument.id) + ).group_by(KnowledgeDocument.file_type).all() + for ft, cnt in type_rows: + file_type_dist[ft] = cnt + + # 最近7天创建的知识库数 + from datetime import timedelta + week_ago = datetime.now() - timedelta(days=7) + recent_kb = db.query(func.count(KnowledgeBase.id)).filter( + KnowledgeBase.created_at >= week_ago + ).scalar() or 0 + + # 活跃知识库（有文档的知识库） + active_kb = db.query(func.count(func.distinct(KnowledgeDocument.knowledge_base_id))).scalar() or 0 + + # 知识图谱统计 + kb_with_graph = db.query(func.count(func.distinct(KnowledgeGraphNode.knowledge_base_id))).scalar() or 0 + + # 外部集成统计 + external_kb = 0 + try: + external_kb = db.query(func.count(KnowledgeBase.id)).filter( + getattr(KnowledgeBase, "external_service_id", None) != None, + getattr(KnowledgeBase, "external_service_id", None) != "" + ).scalar() or 0 + except Exception: + external_kb = 0 + + return { + "code": 0, + "data": { + "total_knowledge_bases": total_kb, + "total_documents": total_docs, + "total_storage_bytes": int(total_storage), + "total_storage_display": format_file_size(int(total_storage)), + "official_kb": official_kb, + "user_kb": user_kb, + "active_kb": active_kb, + "kb_with_graph": kb_with_graph, + "external_kb": external_kb, + "recent_kb_7d": recent_kb, + "total_graph_nodes": int(total_nodes), + "total_graph_edges": int(total_edges), + "doc_status_distribution": doc_status_dist, + "file_type_distribution": file_type_dist, + } + } + + +# ============================================================ +# 知识库 CRUD +# ============================================================ + +@router.get("/", summary="获取知识库列表") +async def list_knowledge_bases( + keyword: str = Query(None, description="搜索关键词"), + status: str = Query(None, description="状态: active/inactive/archived"), + is_official: bool = Query(None, description="是否官方知识库"), + has_external: bool = Query(None, description="是否有外部集成"), + page: int = Query(1, ge=1), + page_size: int = Query(15, ge=1, le=100), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取所有知识库列表（分页、筛选）""" + query = db.query(KnowledgeBase) + + if keyword: + query = query.filter( + or_( + KnowledgeBase.name.ilike(f"%{keyword}%"), + KnowledgeBase.description.ilike(f"%{keyword}%"), + ) + ) + if status: + try: + query = query.filter(getattr(KnowledgeBase, "status", None) == status) + except Exception: + pass # 字段可能不存在 + if is_official is not None: + try: + query = query.filter(getattr(KnowledgeBase, "is_official", None) == is_official) + except Exception: + pass + + total = query.count() + kbs = query.order_by(desc(KnowledgeBase.created_at)).offset( + (page - 1) * page_size + ).limit(page_size).all() + + return { + "code": 0, + "data": { + "total": total, + "page": page, + "page_size": page_size, + "items": [kb_to_dict(kb, db) for kb in kbs], + } + } + + +@router.get("/{kb_id}", summary="获取知识库详情") +async def get_knowledge_base( + kb_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取知识库详细信息，包含文档和图谱数据""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + result = kb_to_dict(kb, db) + + # 文档统计 + doc_stats = db.query( + KnowledgeDocument.status, + func.count(KnowledgeDocument.id), + func.sum(KnowledgeDocument.file_size) + ).filter(KnowledgeDocument.knowledge_base_id == kb_id).group_by( + KnowledgeDocument.status + ).all() + + result["doc_stats"] = { + s: {"count": int(cnt), "size": int(sz or 0)} + for s, cnt, sz in doc_stats + } + + # 图谱统计 + graph_nodes = db.query(func.count(KnowledgeGraphNode.id)).filter( + KnowledgeGraphNode.knowledge_base_id == kb_id + ).scalar() or 0 + graph_edges = db.query(func.count(KnowledgeGraphEdge.id)).filter( + KnowledgeGraphEdge.knowledge_base_id == kb_id + ).scalar() or 0 + result["graph_stats"] = { + "nodes": int(graph_nodes), + "edges": int(graph_edges), + } + + # 外部服务信息 + ext_service_id = getattr(kb, "external_service_id", None) + if ext_service_id: + # 查询外部服务配置 + try: + ext_service = db.query(ExternalKBService).filter( + ExternalKBService.id == ext_service_id + ).first() + if ext_service: + result["external_service"] = { + "id": ext_service.id, + "name": ext_service.name, + "service_type": ext_service.service_type, + "api_url": ext_service.api_url, + "status": getattr(ext_service, "status", "unknown"), + } + except Exception: + result["external_service"] = {"id": ext_service_id, "note": "外部服务表尚未初始化"} + + return {"code": 0, "data": result} + + +@router.post("/", summary="创建知识库") +async def create_knowledge_base( + req: KBCreate, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """创建新的知识库（支持官方平台知识库）""" + # 检查名称唯一性 + existing = db.query(KnowledgeBase).filter(KnowledgeBase.name == req.name).first() + if existing: + raise HTTPException(status_code=400, detail="知识库名称已存在") + + kb = KnowledgeBase( + id=str(uuid.uuid4()), + user_id=admin_user.id, + name=req.name, + description=req.description or "", + directory_path="/", + document_count=0, + knowledge_graph_count=0, + created_at=datetime.now(), + updated_at=datetime.now(), + ) + + # Phase 2.4 新增字段 + try: + setattr(kb, "is_official", req.is_official) + setattr(kb, "status", "active") + if req.external_service_id: + setattr(kb, "external_service_id", req.external_service_id) + if req.external_config: + setattr(kb, "external_config", json.dumps(req.external_config, ensure_ascii=False)) + setattr(kb, "external_sync_status", "none") + except Exception: + logger.warning("KnowledgeBase 新增字段设置失败，可能表结构尚未更新") + + db.add(kb) + db.commit() + db.refresh(kb) + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="create_knowledge_base", + target_type="knowledge_base", + target_id=kb.id, + detail=f"创建知识库: {kb.name} (官方={req.is_official})", + ) + + return {"code": 0, "message": "知识库创建成功", "data": kb_to_dict(kb, db)} + + +@router.put("/{kb_id}", summary="更新知识库") +async def update_knowledge_base( + kb_id: str, + req: KBUpdate, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """更新知识库配置""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + if req.name is not None: + dup = db.query(KnowledgeBase).filter( + KnowledgeBase.name == req.name, + KnowledgeBase.id != kb_id + ).first() + if dup: + raise HTTPException(status_code=400, detail="知识库名称已存在") + kb.name = req.name + + if req.description is not None: + kb.description = req.description + + # Phase 2.4 新增字段更新 + try: + if req.is_official is not None: + setattr(kb, "is_official", req.is_official) + if req.status is not None: + setattr(kb, "status", req.status) + if req.external_service_id is not None: + setattr(kb, "external_service_id", req.external_service_id) + if req.external_config is not None: + setattr(kb, "external_config", json.dumps(req.external_config, ensure_ascii=False)) + except Exception: + logger.warning("KnowledgeBase 新增字段更新失败") + + kb.updated_at = datetime.now() + db.commit() + db.refresh(kb) + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="update_knowledge_base", + target_type="knowledge_base", + target_id=kb_id, + detail=f"更新知识库: {kb.name}", + ) + + return {"code": 0, "message": "知识库已更新", "data": kb_to_dict(kb, db)} + + +@router.delete("/{kb_id}", summary="删除知识库") +async def delete_knowledge_base( + kb_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """删除知识库及其所有文档""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + kb_name = kb.name + + # 删除关联文档 + docs = db.query(KnowledgeDocument).filter( + KnowledgeDocument.knowledge_base_id == kb_id + ).all() + for d in docs: + db.delete(d) + + # 删除关联图谱 + nodes = db.query(KnowledgeGraphNode).filter( + KnowledgeGraphNode.knowledge_base_id == kb_id + ).all() + for n in nodes: + db.delete(n) + + edges = db.query(KnowledgeGraphEdge).filter( + KnowledgeGraphEdge.knowledge_base_id == kb_id + ).all() + for e in edges: + db.delete(e) + + build_logs = db.query(KnowledgeGraphBuildLog).filter( + KnowledgeGraphBuildLog.knowledge_base_id == kb_id + ).all() + for bl in build_logs: + db.delete(bl) + + # 删除关联目录 + dirs = db.query(KnowledgeDirectory).filter( + KnowledgeDirectory.knowledge_base_id == kb_id + ).all() + for d in dirs: + db.delete(d) + + # 删除知识库本身 + db.delete(kb) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="delete_knowledge_base", + target_type="knowledge_base", + target_id=kb_id, + detail=f"删除知识库: {kb_name} (含 {len(docs)} 文档, {len(nodes)} 节点, {len(edges)} 边)", + ) + + return {"code": 0, "message": "知识库已删除"} + + +@router.put("/{kb_id}/toggle", summary="启用/禁用知识库") +async def toggle_knowledge_base( + kb_id: str, + is_active: bool = Query(..., description="是否启用"), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """启用或禁用知识库""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + new_status = "active" if is_active else "inactive" + try: + setattr(kb, "status", new_status) + except Exception: + pass + + kb.updated_at = datetime.now() + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="toggle_knowledge_base", + target_type="knowledge_base", + target_id=kb_id, + detail=f"{'启用' if is_active else '禁用'}知识库: {kb.name}", + ) + + return { + "code": 0, + "message": f"知识库已{'启用' if is_active else '禁用'}", + "data": kb_to_dict(kb, db), + } + + +# ============================================================ +# 文档管理 +# ============================================================ + +@router.get("/{kb_id}/documents", summary="获取知识库文档列表") +async def get_kb_documents( + kb_id: str, + status: str = Query(None, description="文档状态筛选"), + file_type: str = Query(None, description="文件类型筛选"), + keyword: str = Query(None, description="文件名搜索"), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取知识库中的文档列表""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + query = db.query(KnowledgeDocument).filter( + KnowledgeDocument.knowledge_base_id == kb_id + ) + + if status: + query = query.filter(KnowledgeDocument.status == status) + if file_type: + query = query.filter(KnowledgeDocument.file_type == file_type) + if keyword: + query = query.filter(KnowledgeDocument.filename.ilike(f"%{keyword}%")) + + total = query.count() + docs = query.order_by(desc(KnowledgeDocument.created_at)).offset( + (page - 1) * page_size + ).limit(page_size).all() + + return { + "code": 0, + "data": { + "total": total, + "page": page, + "page_size": page_size, + "items": [doc_to_dict(d) for d in docs], + } + } + + +@router.delete("/{kb_id}/documents/{doc_id}", summary="删除知识库文档") +async def delete_kb_document( + kb_id: str, + doc_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """删除知识库中的单个文档""" + doc = db.query(KnowledgeDocument).filter( + KnowledgeDocument.id == doc_id, + KnowledgeDocument.knowledge_base_id == kb_id, + ).first() + if not doc: + raise HTTPException(status_code=404, detail="文档不存在") + + filename = doc.filename + db.delete(doc) + + # 更新知识库文档计数 + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if kb: + kb.document_count = db.query(func.count(KnowledgeDocument.id)).filter( + KnowledgeDocument.knowledge_base_id == kb_id + ).scalar() or 0 + kb.updated_at = datetime.now() + + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="delete_knowledge_document", + target_type="knowledge_document", + target_id=doc_id, + detail=f"删除文档: {filename} (知识库: {kb.name})", + ) + + return {"code": 0, "message": "文档已删除"} + + +# ============================================================ +# 知识图谱管理 +# ============================================================ + +@router.get("/{kb_id}/graph/stats", summary="获取知识图谱统计") +async def get_kb_graph_stats( + kb_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取知识库的知识图谱统计信息""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + nodes = db.query(func.count(KnowledgeGraphNode.id)).filter( + KnowledgeGraphNode.knowledge_base_id == kb_id + ).scalar() or 0 + + edges = db.query(func.count(KnowledgeGraphEdge.id)).filter( + KnowledgeGraphEdge.knowledge_base_id == kb_id + ).scalar() or 0 + + # 最近构建日志 + recent_logs = db.query(KnowledgeGraphBuildLog).filter( + KnowledgeGraphBuildLog.knowledge_base_id == kb_id + ).order_by(desc(KnowledgeGraphBuildLog.created_at)).limit(5).all() + + # 节点类型分布 + node_types = db.query( + KnowledgeGraphNode.label, + func.count(KnowledgeGraphNode.id) + ).filter( + KnowledgeGraphNode.knowledge_base_id == kb_id + ).group_by(KnowledgeGraphNode.label).limit(20).all() + + return { + "code": 0, + "data": { + "total_nodes": int(nodes), + "total_edges": int(edges), + "node_type_distribution": [ + {"label": lbl, "count": int(cnt)} for lbl, cnt in node_types + ], + "recent_build_logs": [ + { + "id": bl.id, + "status": getattr(bl, "status", "unknown"), + "created_at": bl.created_at.isoformat() if bl.created_at else None, + } + for bl in recent_logs + ], + } + } + + +# ============================================================ +# 外部知识库服务管理 +# ============================================================ + +# ExternalKBService 模型定义（如果表不存在会在初始化时创建） +# 这里用原生 SQL 操作，确保即使 ORM 模型未注册也能工作 + +def _get_external_services_table(db: Session): + """安全地获取外部知识库服务列表（兼容表不存在的情况）""" + try: + # 尝试使用 ORM 模型查询 + from ...models import ExternalKBService + return db.query(ExternalKBService).all() + except (AttributeError, ImportError): + # ORM 模型不存在，用原生 SQL + try: + result = db.execute( + "SELECT id, name, service_type, api_url, api_key, config, is_enabled, status, created_at, updated_at FROM external_kb_services ORDER BY created_at DESC" + ) + rows = result.fetchall() + return [ + { + "id": r[0], + "name": r[1], + "service_type": r[2], + "api_url": r[3], + "api_key": r[4], + "config": json.loads(r[5]) if r[5] else {}, + "is_enabled": bool(r[6]), + "status": r[7], + "created_at": r[8].isoformat() if r[8] else None, + "updated_at": r[9].isoformat() if r[9] else None, + } + for r in rows + ] + except Exception as e: + logger.warning(f"查询外部知识库服务表失败: {e}") + return [] + + +def _check_service_connection(service_type: str, api_url: str, api_key: str) -> dict: + """测试外部知识库服务连接状态""" + try: + headers = {} + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + + if service_type == "fastgpt": + # FastGPT 健康检查 + check_url = f"{api_url.rstrip('/')}/api/v1/status" + resp = httpx.get(check_url, headers=headers, timeout=10) + if resp.status_code == 200: + return {"connected": True, "status": "connected", "detail": "FastGPT 连接成功"} + else: + return {"connected": False, "status": "error", "detail": f"FastGPT 返回状态码 {resp.status_code}"} + + elif service_type == "dify": + # Dify 健康检查 + check_url = f"{api_url.rstrip('/')}/v1/datasets" + resp = httpx.get(check_url, headers=headers, timeout=10) + if resp.status_code in [200, 401]: # 401说明服务存在但需要认证 + return {"connected": True, "status": "connected", "detail": "Dify 连接成功"} + else: + return {"connected": False, "status": "error", "detail": f"Dify 返回状态码 {resp.status_code}"} + + elif service_type == "ragflow": + # RAGFlow 健康检查 + check_url = f"{api_url.rstrip('/')}/api/v1/datasets" + resp = httpx.get(check_url, headers=headers, timeout=10) + if resp.status_code in [200, 401]: + return {"connected": True, "status": "connected", "detail": "RAGFlow 连接成功"} + else: + return {"connected": False, "status": "error", "detail": f"RAGFlow 返回状态码 {resp.status_code}"} + + else: + # 通用检查 + resp = httpx.get(api_url, headers=headers, timeout=10) + return {"connected": resp.status_code < 500, "status": "connected" if resp.status_code < 500 else "error"} + + except httpx.TimeoutException: + return {"connected": False, "status": "timeout", "detail": "连接超时"} + except Exception as e: + return {"connected": False, "status": "error", "detail": str(e)} + + +@router.get("/external/services", summary="获取外部知识库服务列表") +async def list_external_services( + service_type: str = Query(None, description="服务类型: fastgpt/dify/ragflow"), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取所有外部知识库服务配置""" + services = _get_external_services_table(db) + + if service_type: + services = [s for s in services if s.get("service_type") == service_type] + + return { + "code": 0, + "data": { + "total": len(services), + "items": services, + "supported_types": [ + {"type": "fastgpt", "name": "FastGPT", "description": "开源 AI 知识库平台，支持多种向量模型"}, + {"type": "dify", "name": "Dify", "description": "LLM 应用开发平台，内置知识库管理"}, + {"type": "ragflow", "name": "RAGFlow", "description": "基于深度文档理解的开源 RAG 引擎"}, + ] + } + } + + +@router.post("/external/services", summary="创建外部知识库服务配置") +async def create_external_service( + req: ExternalServiceConfig, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """创建新的外部知识库服务连接配置""" + # 验证服务类型 + valid_types = ["fastgpt", "dify", "ragflow"] + if req.service_type not in valid_types: + raise HTTPException(status_code=400, detail=f"不支持的服务类型: {req.service_type}") + + # 检查连接 + conn_result = _check_service_connection(req.service_type, req.api_url, req.api_key or "") + + service_id = str(uuid.uuid4()) + config_json = json.dumps(req.config or {}, ensure_ascii=False) + conn_status = conn_result.get("status", "disconnected") + + # 使用原生 SQL 插入（兼容 ORM 模型可能未注册） + try: + db.execute( + f"INSERT INTO external_kb_services (id, name, service_type, api_url, api_key, config, is_enabled, status, created_at, updated_at) " + f"VALUES (:id, :name, :stype, :url, :key, :cfg, 1, :status, NOW(), NOW())", + { + "id": service_id, + "name": req.name, + "stype": req.service_type, + "url": req.api_url, + "key": req.api_key or "", + "cfg": config_json, + "status": conn_status, + } + ) + db.commit() + except Exception as e: + # 表可能不存在，尝试创建 + logger.warning(f"插入外部服务失败: {e}, 尝试创建表") + db.execute(""" + CREATE TABLE IF NOT EXISTS external_kb_services ( + id VARCHAR(36) PRIMARY KEY, + name VARCHAR(200) NOT NULL, + service_type VARCHAR(20) NOT NULL, + api_url VARCHAR(500) NOT NULL, + api_key VARCHAR(255) DEFAULT '', + config JSON DEFAULT NULL, + is_enabled BOOLEAN DEFAULT TRUE, + status VARCHAR(20) DEFAULT 'disconnected', + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 + """) + db.commit() + # 再次尝试插入 + db.execute( + f"INSERT INTO external_kb_services (id, name, service_type, api_url, api_key, config, is_enabled, status, created_at, updated_at) " + f"VALUES (:id, :name, :stype, :url, :key, :cfg, 1, :status, NOW(), NOW())", + { + "id": service_id, + "name": req.name, + "stype": req.service_type, + "url": req.api_url, + "key": req.api_key or "", + "cfg": config_json, + "status": conn_status, + } + ) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="create_external_kb_service", + target_type="external_kb_service", + target_id=service_id, + detail=f"创建外部知识库服务: {req.name} ({req.service_type})", + ) + + return { + "code": 0, + "message": "外部知识库服务配置已创建", + "data": { + "id": service_id, + "name": req.name, + "service_type": req.service_type, + "api_url": req.api_url, + "connection": conn_result, + } + } + + +@router.put("/external/services/{service_id}", summary="更新外部知识库服务配置") +async def update_external_service( + service_id: str, + req: ExternalServiceUpdate, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """更新外部知识库服务配置""" + # 查找现有服务 + try: + result = db.execute( + "SELECT id, name, service_type, api_url, api_key FROM external_kb_services WHERE id = :sid", + {"sid": service_id} + ) + row = result.fetchone() + except Exception: + raise HTTPException(status_code=404, detail="外部知识库服务不存在或表未创建") + + if not row: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + # 构建更新字段 + updates = {} + if req.name is not None: + updates["name"] = req.name + if req.api_url is not None: + updates["api_url"] = req.api_url + if req.api_key is not None: + updates["api_key"] = req.api_key + if req.config is not None: + updates["config"] = json.dumps(req.config, ensure_ascii=False) + if req.is_enabled is not None: + updates["is_enabled"] = 1 if req.is_enabled else 0 + + # 如果 URL 或 API Key 变了，重新测试连接 + new_url = req.api_url or row[3] + new_key = req.api_key or row[4] + if req.api_url or req.api_key: + service_type = row[2] + conn_result = _check_service_connection(service_type, new_url, new_key) + updates["status"] = conn_result.get("status", "disconnected") + + updates["updated_at"] = "NOW()" + + # 执行更新 + set_clause = ", ".join([f"{k} = :{k}" for k in updates.keys() if k != "updated_at"]) + set_clause += ", updated_at = NOW()" + params = {k: v for k, v in updates.items() if k != "updated_at"} + params["sid"] = service_id + + db.execute( + f"UPDATE external_kb_services SET {set_clause} WHERE id = :sid", + params + ) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="update_external_kb_service", + target_type="external_kb_service", + target_id=service_id, + detail=f"更新外部知识库服务: {row[1]}", + ) + + return {"code": 0, "message": "外部知识库服务配置已更新"} + + +@router.delete("/external/services/{service_id}", summary="删除外部知识库服务配置") +async def delete_external_service( + service_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """删除外部知识库服务配置""" + try: + result = db.execute( + "SELECT name FROM external_kb_services WHERE id = :sid", + {"sid": service_id} + ) + row = result.fetchone() + except Exception: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + if not row: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + name = row[0] + db.execute("DELETE FROM external_kb_services WHERE id = :sid", {"sid": service_id}) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="delete_external_kb_service", + target_type="external_kb_service", + target_id=service_id, + detail=f"删除外部知识库服务: {name}", + ) + + return {"code": 0, "message": "外部知识库服务配置已删除"} + + +@router.get("/external/services/{service_id}/status", summary="检查外部知识库服务连接状态") +async def check_external_service_status( + service_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """实时检查外部知识库服务的连接状态""" + try: + result = db.execute( + "SELECT service_type, api_url, api_key FROM external_kb_services WHERE id = :sid", + {"sid": service_id} + ) + row = result.fetchone() + except Exception: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + if not row: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + conn_result = _check_service_connection(row[0], row[1], row[2]) + + # 更新状态 + try: + db.execute( + "UPDATE external_kb_services SET status = :status, updated_at = NOW() WHERE id = :sid", + {"status": conn_result.get("status", "disconnected"), "sid": service_id} + ) + db.commit() + except Exception: + pass + + return {"code": 0, "data": conn_result} + + +# ============================================================ +# 知识库同步（与外部服务） +# ============================================================ + +@router.post("/external/sync/{kb_id}", summary="同步知识库与外部服务") +async def sync_knowledge_base( + kb_id: str, + req: SyncRequest, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """将知识库与外部知识库服务进行数据同步""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + # 确定同步目标服务 + ext_service_id = req.external_service_id or getattr(kb, "external_service_id", None) + if not ext_service_id: + raise HTTPException(status_code=400, detail="未指定外部知识库服务，请先关联或传入 external_service_id") + + # 查询外部服务配置 + try: + result = db.execute( + "SELECT name, service_type, api_url, api_key, config FROM external_kb_services WHERE id = :sid", + {"sid": ext_service_id} + ) + row = result.fetchone() + except Exception: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + if not row: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + service_name, service_type, api_url, api_key, config_raw = row + service_config = json.loads(config_raw) if config_raw else {} + + # 执行同步 + sync_result = {"synced": False, "detail": "", "items_synced": 0} + + try: + headers = {} + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + + if service_type == "fastgpt" and req.sync_direction == "push": + # 向 FastGPT 推送知识库数据 + # 1. 在 FastGPT 创建/获取对应的 dataset + ext_kb_id = getattr(kb, "external_id", None) + + if not ext_kb_id: + # 创建新 dataset + create_resp = httpx.post( + f"{api_url.rstrip('/')}/api/v1/datasets", + headers=headers, + json={"name": kb.name, "description": kb.description or ""}, + timeout=30, + ) + if create_resp.status_code == 200: + ext_data = create_resp.json() + ext_kb_id = ext_data.get("data", {}).get("id") or ext_data.get("id") + # 更新 KB 的 external_id + try: + setattr(kb, "external_id", ext_kb_id) + except Exception: + db.execute( + "UPDATE knowledge_bases SET external_id = :eid WHERE id = :kid", + {"eid": ext_kb_id, "kid": kb_id} + ) + else: + sync_result["detail"] = f"FastGPT 创建 dataset 失败: {create_resp.status_code}" + sync_result["synced"] = False + + if ext_kb_id: + # 推送文档内容到 FastGPT + docs = db.query(KnowledgeDocument).filter( + KnowledgeDocument.knowledge_base_id == kb_id, + KnowledgeDocument.status == "completed" + ).all() + + synced_count = 0 + for doc in docs: + content = doc.content or "" + if not content: + continue + + # 分段推送 + push_resp = httpx.post( + f"{api_url.rstrip('/')}/api/v1/datasets/{ext_kb_id}/documents", + headers=headers, + json={ + "name": doc.filename, + "content": content[:50000], # 限制单次推送大小 + }, + timeout=30, + ) + if push_resp.status_code in [200, 201]: + synced_count += 1 + + sync_result["synced"] = True + sync_result["detail"] = f"已推送 {synced_count}/{len(docs)} 文档到 FastGPT" + sync_result["items_synced"] = synced_count + + elif service_type == "dify" and req.sync_direction == "push": + # 向 Dify 推送知识库数据 + ext_kb_id = getattr(kb, "external_id", None) + + if not ext_kb_id: + create_resp = httpx.post( + f"{api_url.rstrip('/')}/v1/datasets", + headers=headers, + json={"name": kb.name}, + timeout=30, + ) + if create_resp.status_code in [200, 201]: + ext_data = create_resp.json() + ext_kb_id = ext_data.get("id") + try: + setattr(kb, "external_id", ext_kb_id) + except Exception: + db.execute( + "UPDATE knowledge_bases SET external_id = :eid WHERE id = :kid", + {"eid": ext_kb_id, "kid": kb_id} + ) + + if ext_kb_id: + docs = db.query(KnowledgeDocument).filter( + KnowledgeDocument.knowledge_base_id == kb_id, + KnowledgeDocument.status == "completed" + ).all() + + synced_count = 0 + for doc in docs: + content = doc.content or "" + if not content: + continue + + # Dify 文档创建 + doc_resp = httpx.post( + f"{api_url.rstrip('/')}/v1/datasets/{ext_kb_id}/document/create_by_text", + headers=headers, + json={ + "name": doc.filename, + "text": content[:50000], + "indexing_technique": "high_quality", + }, + timeout=30, + ) + if doc_resp.status_code in [200, 201]: + synced_count += 1 + + sync_result["synced"] = True + sync_result["detail"] = f"已推送 {synced_count}/{len(docs)} 文档到 Dify" + sync_result["items_synced"] = synced_count + + elif service_type == "ragflow" and req.sync_direction == "push": + # 向 RAGFlow 推送知识库数据 + ext_kb_id = getattr(kb, "external_id", None) + + if not ext_kb_id: + create_resp = httpx.post( + f"{api_url.rstrip('/')}/api/v1/datasets", + headers=headers, + json={"name": kb.name}, + timeout=30, + ) + if create_resp.status_code in [200, 201]: + ext_data = create_resp.json() + ext_kb_id = ext_data.get("id") + try: + setattr(kb, "external_id", ext_kb_id) + except Exception: + db.execute( + "UPDATE knowledge_bases SET external_id = :eid WHERE id = :kid", + {"eid": ext_kb_id, "kid": kb_id} + ) + + if ext_kb_id: + docs = db.query(KnowledgeDocument).filter( + KnowledgeDocument.knowledge_base_id == kb_id, + KnowledgeDocument.status == "completed" + ).all() + + synced_count = 0 + for doc in docs: + content = doc.content or "" + if not content: + continue + + doc_resp = httpx.post( + f"{api_url.rstrip('/')}/api/v1/datasets/{ext_kb_id}/documents", + headers=headers, + json={"name": doc.filename, "content": content[:50000]}, + timeout=30, + ) + if doc_resp.status_code in [200, 201]: + synced_count += 1 + + sync_result["synced"] = True + sync_result["detail"] = f"已推送 {synced_count}/{len(docs)} 文档到 RAGFlow" + sync_result["items_synced"] = synced_count + + elif req.sync_direction == "pull": + # 从外部服务拉取知识库数据 + sync_result["detail"] = f"从 {service_type} 拉取数据（暂未实现完整拉取逻辑）" + sync_result["synced"] = True + sync_result["items_synced"] = 0 + + elif req.sync_direction == "bidirectional": + sync_result["detail"] = f"双向同步（暂未实现完整双向同步逻辑）" + sync_result["synced"] = True + sync_result["items_synced"] = 0 + + # 更新同步状态 + try: + setattr(kb, "external_sync_status", "synced" if sync_result["synced"] else "error") + except Exception: + db.execute( + "UPDATE knowledge_bases SET external_sync_status = :es WHERE id = :kid", + {"es": "synced" if sync_result["synced"] else "error", "kid": kb_id} + ) + + kb.updated_at = datetime.now() + db.commit() + + except Exception as e: + logger.error(f"知识库同步失败: {e}") + sync_result["detail"] = f"同步异常: {str(e)}" + sync_result["synced"] = False + + try: + setattr(kb, "external_sync_status", "error") + except Exception: + db.execute( + "UPDATE knowledge_bases SET external_sync_status = 'error' WHERE id = :kid", + {"kid": kb_id} + ) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="sync_knowledge_base", + target_type="knowledge_base", + target_id=kb_id, + detail=f"同步知识库 '{kb.name}' → {service_name} ({service_type}), 方向={req.sync_direction}, 结果={sync_result['detail']}", + ) + + return {"code": 0, "data": sync_result} + + +# ============================================================ +# 批量操作 +# ============================================================ + +@router.post("/batch/delete", summary="批量删除知识库") +async def batch_delete_kbs( + kb_ids: List[str] = Body(..., description="知识库ID列表"), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """批量删除知识库""" + deleted = 0 + failed = 0 + for kb_id in kb_ids: + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + failed += 1 + continue + + # 删除关联数据 + db.query(KnowledgeDocument).filter(KnowledgeDocument.knowledge_base_id == kb_id).delete() + db.query(KnowledgeGraphNode).filter(KnowledgeGraphNode.knowledge_base_id == kb_id).delete() + db.query(KnowledgeGraphEdge).filter(KnowledgeGraphEdge.knowledge_base_id == kb_id).delete() + db.query(KnowledgeGraphBuildLog).filter(KnowledgeGraphBuildLog.knowledge_base_id == kb_id).delete() + db.query(KnowledgeDirectory).filter(KnowledgeDirectory.knowledge_base_id == kb_id).delete() + db.delete(kb) + deleted += 1 + + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="batch_delete_knowledge_bases", + target_type="knowledge_base", + target_id=",".join(kb_ids[:10]), + detail=f"批量删除知识库: 成功 {deleted}, 失败 {failed}", + ) + + return {"code": 0, "data": {"deleted": deleted, "failed": failed}} diff --git a/backend/python/app/models.py b/backend/python/app/models.py index a0f3f2c..e8aa0db 100644 --- a/backend/python/app/models.py +++ b/backend/python/app/models.py @@ -164,7 +164,13 @@ class KnowledgeBase(Base): document_count = Column(Integer, default=0) knowledge_graph_count = Column(Integer, default=0) created_at = Column(DateTime, default=datetime.now) - updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now) + updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now) + is_official = Column(Boolean, default=False) + status = Column(String(20), default="active") + external_service_id = Column(String(36), default=None) + external_id = Column(String(100), default=None) + external_sync_status = Column(String(20), default="none") + external_config = Column(JSON, default=None) class KnowledgeDirectory(Base): @@ -935,3 +941,19 @@ class TeamSharedSkill(Base): __table_args__ = ( UniqueConstraint("team_id", "skill_id", name="uk_tss_team_skill"), ) + + +class ExternalKBService(Base): + """外部知识库服务配置""" + __tablename__ = "external_kb_services" + + id = Column(String(36), primary_key=True, default=gen_uuid) + name = Column(String(200), nullable=False) + service_type = Column(String(20), nullable=False) + api_url = Column(String(500), nullable=False) + api_key = Column(String(255), default="") + config = Column(JSON, default=None) + is_enabled = Column(Boolean, default=True) + status = Column(String(20), default="disconnected") + created_at = Column(DateTime, default=datetime.now) + updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now) diff --git a/backend/python/main.py b/backend/python/main.py index 0f37df0..826e937 100644 --- a/backend/python/main.py +++ b/backend/python/main.py @@ -16,7 +16,7 @@ from datetime import datetime sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.api import auth, chat, channels, knowledge, market, token_usage, payment, profile, notifications, assets, evolution -from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin +from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin from app.config import setup_file_logging, SLOW_REQUEST_THRESHOLD from app.database import SessionLocal from app.models import OperationLog as OperationLogModel, TokenUsageLog as TokenUsageLogModel @@ -208,6 +208,8 @@ app.include_router(ads.router, prefix="/app/api/admin/ads", tags=["后台-广告 app.include_router(market_admin.router, prefix="/app/api/admin/market", tags=["后台-应用审核"]) app.include_router(logs.router, prefix="/app/api/admin/logs", tags=["后台-日志风控"]) app.include_router(skills_admin.router, prefix="/app/api/admin/skills", tags=["后台-技能管理"]) +app.include_router(agents_admin, knowledge_admin.router, prefix="/app/api/admin/agents", tags=["后台-Agent管理"]) +app.include_router(knowledge_admin.router, prefix="/app/api/admin/knowledge", tags=["后台-知识库管理"]) @app.get("/app/api/health")