feat(kb): 接入 Dify 知识库，移除 FastGPT；AI 聊天产出+用户文档自动入库量化

- 新增 app/dify_kb.py Dify v1 客户端(连通测试/检索/文本入库) - kb_router 仅保留 Dify 外部检索分支 - knowledge_admin 去掉 FastGPT/RAGFlow，新增 dify 批量入库端点 - chat.py 注入 Dify RAG 上下文 + 助手回答 fire-and-forget 入库 - 前端 Knowledge.tsx 改为 Dify 单服务 UI(可通讯/可用验证+入库按钮) - DB: external_kb_services 删除 fastgpt，qihuai 设为 https+connected
2026-07-14 20:56:53 +08:00
1 parent d3a890eb18
commit a1669374e6
@@ -0,0 +1,248 @@
+"""
+Dify 知识库客户端
+================
+封装与 Dify 知识库(v1 API)的交互：连通性测试、检索(RAG)、文本入库。
+设计用于替换原 FastGPT 外部知识库对接。
+
+已验证的接入点（2026-07-14 实测）:
+- Base:  https://dify.qihuyun.com/v1
+         （直接 HTTP 访问会被 nginx 301 跳转到 HTTPS，必须 HTTPS 直连，
+          且内网自签证书需 verify=False）
+- Auth:  Bearer {api_key}
+- GET  /v1/datasets                               列出知识库(可据此解析 dataset id)
+- POST /v1/datasets/{id}/document/create-by-text  文本入库
+- POST /v1/datasets/{id}/document/create-by-file  文件入库(multipart)
+- POST /v1/datasets/{id}/retrieve                 检索(RAG)
+  retrieve 必需完整的 retrieval_model 字典，否则返回 400。
+"""
+import logging
+import threading
+import time
+import httpx
+
+logger = logging.getLogger("aiflow.dify_kb")
+
+# dataset id 缓存: key -> (dataset_id, ts)
+_dataset_cache: dict = {}
+_dataset_cache_lock = threading.Lock()
+_DATASET_CACHE_TTL = 3600
+
+
+def _normalize_base(api_url: str) -> str:
+    """统一 base url -> https 形式，且不带 /v1 后缀。"""
+    u = (api_url or "").strip()
+    if not u:
+        u = "https://dify.qihuyun.com/v1"
+    u = u.rstrip("/")
+    if u.endswith("/v1"):
+        u = u[: -len("/v1")]
+    if u.startswith("http://"):
+        u = "https://" + u[len("http://"):]
+    return u
+
+
+def dify_test(api_url: str, api_key: str) -> dict:
+    """验证 Dify 知识库「可通讯 + 可用」。
+
+    返回:
+        {
+          "connected": bool,
+          "usable": bool,
+          "status": "connected" | "auth_error" | "timeout" | "error",
+          "detail": str,
+          "datasets": [{"id":..., "name":...}, ...]
+        }
+    """
+    base = _normalize_base(api_url)
+    headers = {"Authorization": f"Bearer {api_key}"}
+    try:
+        with httpx.Client(base_url=base, timeout=15, verify=False) as client:
+            r = client.get("/v1/datasets", headers=headers)
+            if r.status_code == 200:
+                data = r.json()
+                datasets = data.get("data", []) or []
+                names = [d.get("name") for d in datasets]
+                # 能列出知识库即代表鉴权通过、可读取 —— 视为可用。
+                return {
+                    "connected": True,
+                    "usable": True,
+                    "status": "connected",
+                    "detail": f"Dify 连接成功，发现 {len(datasets)} 个知识库: {', '.join([n for n in names if n]) or '无'}",
+                    "datasets": [{"id": d.get("id"), "name": d.get("name")} for d in datasets],
+                }
+            elif r.status_code == 401:
+                return {
+                    "connected": True,
+                    "usable": False,
+                    "status": "auth_error",
+                    "detail": "已连通 Dify，但 API Key 鉴权失败(401)，请检查 API Key。",
+                    "datasets": [],
+                }
+            else:
+                return {
+                    "connected": False,
+                    "usable": False,
+                    "status": "error",
+                    "detail": f"Dify 返回状态码 {r.status_code}",
+                    "datasets": [],
+                }
+    except httpx.TimeoutException:
+        return {"connected": False, "usable": False, "status": "timeout", "detail": "连接超时", "datasets": []}
+    except Exception as e:
+        return {"connected": False, "usable": False, "status": "error", "detail": str(e), "datasets": []}
+
+
+def resolve_dataset_id(api_url: str, api_key: str, dataset_name: str = None) -> str:
+    """解析 dataset id（优先按名称匹配，否则取列表第一个），带缓存。"""
+    base = _normalize_base(api_url)
+    headers = {"Authorization": f"Bearer {api_key}"}
+    cache_key = f"{base}|{api_key}|{dataset_name}"
+    now = time.time()
+    with _dataset_cache_lock:
+        c = _dataset_cache.get(cache_key)
+        if c and now - c[1] < _DATASET_CACHE_TTL:
+            return c[0]
+    ds_id = None
+    try:
+        with httpx.Client(base_url=base, timeout=15, verify=False) as client:
+            r = client.get("/v1/datasets", headers=headers)
+            if r.status_code == 200:
+                datasets = r.json().get("data", []) or []
+                if dataset_name:
+                    match = [d for d in datasets if d.get("name") == dataset_name]
+                    if match:
+                        ds_id = match[0].get("id")
+                if not ds_id and datasets:
+                    ds_id = datasets[0].get("id")
+    except Exception as e:
+        logger.error(f"resolve_dataset_id failed: {e}")
+    if ds_id:
+        with _dataset_cache_lock:
+            _dataset_cache[cache_key] = (ds_id, now)
+    return ds_id
+
+
+def dify_retrieve(api_url, api_key, query, top_k=5, dataset_name=None) -> list:
+    """检索，返回 [{content, score, source, metadata}]。"""
+    ds_id = resolve_dataset_id(api_url, api_key, dataset_name)
+    if not ds_id:
+        return []
+    base = _normalize_base(api_url)
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    payload = {
+        "query": query,
+        "retrieval_model": {
+            "search_method": "semantic_search",
+            "reranking_enable": False,
+            "score_threshold_enabled": False,
+            "top_k": top_k,
+        },
+    }
+    try:
+        with httpx.Client(base_url=base, timeout=30, verify=False) as client:
+            r = client.post(f"/v1/datasets/{ds_id}/retrieve", headers=headers, json=payload)
+            if r.status_code == 200:
+                data = r.json()
+                out = []
+                for item in data.get("records", []):
+                    seg = item.get("segment", {})
+                    out.append({
+                        "content": seg.get("content", ""),
+                        "score": float(item.get("score", 0.5)),
+                        "source": f"Dify: {seg.get('document_id', '')}",
+                        "metadata": item,
+                    })
+                return out
+            else:
+                logger.warning(f"Dify retrieve returned {r.status_code}: {r.text[:200]}")
+                return []
+    except Exception as e:
+        logger.error(f"Dify retrieve error: {e}")
+        return []
+
+
+def dify_ingest_text(api_url, api_key, text, name=None, dataset_name=None) -> bool:
+    """文本入库（create-by-text）。返回是否成功。"""
+    if not text or not text.strip():
+        return False
+    ds_id = resolve_dataset_id(api_url, api_key, dataset_name)
+    if not ds_id:
+        return False
+    base = _normalize_base(api_url)
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    payload = {
+        "name": name or f"doc-{int(time.time())}",
+        "text": text,
+        "indexing_technique": "high_quality",
+    }
+    try:
+        with httpx.Client(base_url=base, timeout=60, verify=False) as client:
+            r = client.post(f"/v1/datasets/{ds_id}/document/create-by-text", headers=headers, json=payload)
+            if r.status_code in (200, 201):
+                return True
+            else:
+                logger.warning(f"Dify ingest text returned {r.status_code}: {r.text[:200]}")
+                return False
+    except Exception as e:
+        logger.error(f"Dify ingest text error: {e}")
+        return False
+
+
+async def aingest_text(api_url, api_key, text, name=None, dataset_name=None) -> bool:
+    """异步文本入库（供 chat 实时落库使用）。"""
+    if not text or not text.strip():
+        return False
+    ds_id = resolve_dataset_id(api_url, api_key, dataset_name)
+    if not ds_id:
+        return False
+    base = _normalize_base(api_url)
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    payload = {
+        "name": name or f"doc-{int(time.time())}",
+        "text": text,
+        "indexing_technique": "high_quality",
+    }
+    try:
+        async with httpx.AsyncClient(base_url=base, timeout=60, verify=False) as client:
+            r = await client.post(f"/v1/datasets/{ds_id}/document/create-by-text", headers=headers, json=payload)
+            if r.status_code in (200, 201):
+                return True
+            else:
+                logger.warning(f"Dify aingest returned {r.status_code}: {r.text[:200]}")
+                return False
+    except Exception as e:
+        logger.error(f"Dify aingest error: {e}")
+        return False
+
+
+def get_dify_kb_context(db, query: str, top_k: int = 5) -> str:
+    """读取已启用的 Dify 外部服务，检索并返回拼接后的上下文文本（供 chat 注入）。"""
+    from sqlalchemy import text as _text
+    try:
+        result = db.execute(_text(
+            "SELECT api_url, api_key, config FROM external_kb_services "
+            "WHERE service_type='dify' AND is_enabled=1 AND status='connected' LIMIT 1"
+        ))
+        row = result.fetchone()
+        if not row:
+            return ""
+        api_url, api_key, config_raw = row
+        dataset_name = None
+        if config_raw:
+            try:
+                cfg = config_raw if isinstance(config_raw, dict) else __import__("json").loads(config_raw)
+                dataset_name = cfg.get("dataset_name")
+            except Exception:
+                pass
+        hits = dify_retrieve(api_url, api_key, query, top_k=top_k, dataset_name=dataset_name)
+        if not hits:
+            return ""
+        parts = []
+        for i, h in enumerate(hits, 1):
+            content = (h.get("content") or "").strip()
+            if content:
+                parts.append(f"[{i}] {content}")
+        return "\n\n".join(parts)
+    except Exception as e:
+        logger.error(f"get_dify_kb_context error: {e}")
+        return ""
@@ -1,7 +1,7 @@
 """
 知识库路由层
 ============
-将检索请求智能路由到本地BM25、向量检索、或外部KB服务(FastGPT/Dify/RAGFlow)。
+将检索请求智能路由到本地BM25、向量检索、或外部KB服务(Dify)。
 根据知识库配置自动选择最优检索策略。

 使用方式:
@@ -9,17 +9,14 @@
    results = await kb_router_search(query, kb_id, top_k=5)
 """

-import os
-import json
 import logging
-import httpx
 from typing import List, Dict, Optional

 logger = logging.getLogger("aiflow.kb_router")


 async def _get_external_services(db_session, enabled_only: bool = True) -> List[Dict]:
-    """获取已配置的外部KB服务"""
+    """获取已配置的外部KB服务（仅 dify）"""
    from sqlalchemy import text
    try:
        query = "SELECT * FROM external_kb_services"
@@ -34,123 +31,34 @@ async def _get_external_services(db_session, enabled_only: bool = True) -> List[
        return []


-async def _search_fastgpt(api_url: str, api_key: str, query: str, top_k: int = 5) -> List[Dict]:
-    """搜索 FastGPT 知识库"""
+async def _search_dify(service: Dict, query: str, top_k: int = 5) -> List[Dict]:
+    """检索 Dify 知识库（走 app.dify_kb 客户端）"""
    try:
-        async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.post(
-                f"{api_url.rstrip('/')}/api/v1/dataset/search",
-                headers={
-                    "Authorization": f"Bearer {api_key}",
-                    "Content-Type": "application/json",
-                },
-                json={
-                    "query": query,
-                    "similarity": 0.5,
-                    "limit": top_k,
-                },
-            )
-            if resp.status_code == 200:
-                data = resp.json()
-                results = []
-                for item in data.get("data", {}).get("list", []):
-                    results.append({
-                        "content": item.get("q", item.get("content", "")),
-                        "score": float(item.get("score", 0.5)),
-                        "source": f"FastGPT: {item.get('collectionId', '')}",
-                        "metadata": item,
-                    })
-                return results
-            else:
-                logger.warning(f"FastGPT search returned {resp.status_code}")
-                return []
-    except Exception as e:
-        logger.error(f"FastGPT search error: {e}")
-        return []
-
-
-async def _search_dify(api_url: str, api_key: str, query: str, top_k: int = 5) -> List[Dict]:
-    """搜索 Dify 知识库"""
-    try:
-        async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.post(
-                f"{api_url.rstrip('/')}/v1/datasets/retrieval",
-                headers={
-                    "Authorization": f"Bearer {api_key}",
-                    "Content-Type": "application/json",
-                },
-                json={
-                    "query": query,
-                    "retrieval_mode": "semantic",
-                    "top_k": top_k,
-                },
-            )
-            if resp.status_code == 200:
-                data = resp.json()
-                results = []
-                for item in data.get("records", []):
-                    segment = item.get("segment", {})
-                    results.append({
-                        "content": segment.get("content", ""),
-                        "score": float(item.get("score", 0.5)),
-                        "source": f"Dify: {segment.get('id', '')}",
-                        "metadata": item,
-                    })
-                return results
-            else:
-                logger.warning(f"Dify search returned {resp.status_code}")
-                return []
+        from app.dify_kb import dify_retrieve
+        api_url = service.get("api_url", "")
+        api_key = service.get("api_key", "")
+        dataset_name = None
+        cfg = service.get("config")
+        if isinstance(cfg, str):
+            try:
+                import json
+                cfg = json.loads(cfg)
+            except Exception:
+                cfg = None
+        if isinstance(cfg, dict):
+            dataset_name = cfg.get("dataset_name")
+        hits = dify_retrieve(api_url, api_key, query, top_k=top_k, dataset_name=dataset_name)
+        return hits
    except Exception as e:
        logger.error(f"Dify search error: {e}")
        return []


-async def _search_ragflow(api_url: str, api_key: str, query: str, top_k: int = 5) -> List[Dict]:
-    """搜索 RAGFlow 知识库"""
-    try:
-        async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.post(
-                f"{api_url.rstrip('/')}/api/v1/retrieval",
-                headers={
-                    "Authorization": f"Bearer {api_key}",
-                    "Content-Type": "application/json",
-                },
-                json={
-                    "question": query,
-                    "top_k": top_k,
-                },
-            )
-            if resp.status_code == 200:
-                data = resp.json()
-                results = []
-                for item in data.get("data", {}).get("chunks", []):
-                    results.append({
-                        "content": item.get("content", ""),
-                        "score": float(item.get("similarity", 0.5)),
-                        "source": f"RAGFlow: {item.get('document_id', '')}",
-                        "metadata": item,
-                    })
-                return results
-            else:
-                logger.warning(f"RAGFlow search returned {resp.status_code}")
-                return []
-    except Exception as e:
-        logger.error(f"RAGFlow search error: {e}")
-        return []
-
-
 async def _search_external(service: Dict, query: str, top_k: int = 5) -> List[Dict]:
-    """路由到对应的外部KB服务"""
+    """路由到对应的外部KB服务（当前仅支持 dify）"""
    service_type = service.get("service_type", "")
-    api_url = service.get("api_url", "")
-    api_key = service.get("api_key", "")
-
-    if service_type == "fastgpt":
-        return await _search_fastgpt(api_url, api_key, query, top_k)
-    elif service_type == "dify":
-        return await _search_dify(api_url, api_key, query, top_k)
-    elif service_type == "ragflow":
-        return await _search_ragflow(api_url, api_key, query, top_k)
+    if service_type == "dify":
+        return await _search_dify(service, query, top_k)
    else:
        logger.warning(f"Unknown external KB type: {service_type}")
        return []
@@ -171,7 +79,7 @@ async def kb_router_search(
    检索策略:
    1. 本地 BM25 检索 (MySQL, 已有)
    2. 向量语义检索 (PostgreSQL+pgvector, 新增)
-    3. 外部KB检索 (FastGPT/Dify/RAGFlow, 按配置)
+    3. 外部KB检索 (Dify, 按配置)

    融合方式: Reciprocal Rank Fusion (RRF)

@@ -218,7 +126,7 @@ async def kb_router_search(
        except Exception as e:
            logger.warning(f"Vector search failed: {e}")

-    # 3. 外部KB检索
+    # 3. 外部KB检索 (Dify)
    if use_external and db_session:
        services = await _get_external_services(db_session)
        for service in services:
@@ -228,7 +136,7 @@ async def kb_router_search(
                    all_results.append({
                        "content": item.get("content", ""),
                        "score": item.get("score", 0.5),
-                        "source": item.get("source", f"external_{service['service_type']}"),
+                        "source": item.get("source", "external_dify"),
                        "rank": rank + 1,
                        "metadata": item.get("metadata", {}),
                    })