diff --git a/backend/python/app/anysearch_client.py b/backend/python/app/anysearch_client.py index 294d262..8a6f883 100644 --- a/backend/python/app/anysearch_client.py +++ b/backend/python/app/anysearch_client.py @@ -1,118 +1,179 @@ -""" -AnySearch 联网检索客户端 -====================== -调用 AnySearch 网页/领域搜索 API (https://api.anysearch.com/v1/search)， -为 AI 聊天注入实时联网检索结果，亦可被后端端点 / Agent 工具复用。 - -约束：平台容器存在智能路由代理，对外部 API 偶发重置；本客户端使用 -trust_env=False 直连 (等价于 NO_PROXY=api.anysearch.com)，规避代理重置。 -失败一律返回空/降级结果，保证聊天链路不中断。 -""" -import logging -from typing import List, Dict, Optional - -logger = logging.getLogger("aiflow.anysearch") - -DEFAULT_API_URL = "https://api.anysearch.com/v1/search" -DEFAULT_ZONE = "cn" -REQUEST_TIMEOUT = 8.0 - - -async def anysearch_search( - api_key: str, - api_url: str = DEFAULT_API_URL, - query: str = "", - top_k: int = 5, - zone: str = DEFAULT_ZONE, - freshness: Optional[str] = None, - timeout: float = REQUEST_TIMEOUT, -) -> List[Dict]: - """调用 AnySearch 搜索，返回结构化结果列表。 - - 每个结果: - { - "content": snippet 摘要文本, - "score": 0.9..0.5 递减, - "source": "web_anysearch", - "metadata": {"url", "title", "source_domain", "zone"} - } - 失败时返回 [] (不抛异常)。 - """ - if not api_key: - return [] - try: - import httpx - except ImportError: - logger.warning("httpx not available for AnySearch") - return [] - - payload = { - "query": query, - "zone": zone, - "top_k": max(1, min(int(top_k), 20)), - } - if freshness: - payload["freshness"] = freshness - - headers = { - "Content-Type": "application/json", - "Authorization": f"Bearer {api_key}", - } - try: - # trust_env=False -> 忽略容器内 HTTP(S)_PROXY，直连 api.anysearch.com - async with httpx.AsyncClient(trust_env=False, timeout=timeout) as client: - resp = await client.post(api_url, json=payload, headers=headers) - if resp.status_code != 200: - logger.warning(f"AnySearch HTTP {resp.status_code}: {resp.text[:200]}") - return [] - data = resp.json() - if data.get("code") != 0: - logger.warning(f"AnySearch biz error: {data.get('message')}") - return [] - results = (data.get("data") or {}).get("results") or [] - hits = [] - for idx, r in enumerate(results[:top_k]): - title = r.get("title", "") or "" - url = r.get("url", "") or "" - snippet = r.get("snippet") or r.get("content") or "" - if not snippet and not title: - continue - try: - from urllib.parse import urlparse - domain = urlparse(url).netloc - except Exception: - domain = "" - hits.append({ - "content": snippet, - "score": max(0.4, 0.9 - idx * 0.08), - "source": "web_anysearch", - "metadata": { - "url": url, - "title": title, - "source_domain": domain, - "zone": zone, - }, - }) - return hits - except Exception as e: - logger.warning(f"AnySearch search failed: {e}") - return [] - - -async def anysearch_health(api_key: str, api_url: str = DEFAULT_API_URL, timeout: float = REQUEST_TIMEOUT) -> Dict: - """连通性探测：返回 {"status": "connected"/"disconnected", "detail": ...}。""" - if not api_key: - return {"status": "disconnected", "detail": "未配置 API Key", "connected": False, "usable": False} - try: - import httpx - async with httpx.AsyncClient(trust_env=False, timeout=timeout) as client: - resp = await client.post( - api_url, - json={"query": "ping", "zone": DEFAULT_ZONE, "top_k": 1}, - headers={"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"}, - ) - if resp.status_code == 200: - return {"status": "connected", "detail": "AnySearch 连接正常", "connected": True, "usable": True} - return {"status": "error", "detail": f"HTTP {resp.status_code}", "connected": False, "usable": False} - except Exception as e: - return {"status": "error", "detail": str(e)[:200], "connected": False, "usable": False} +""" +AnySearch 联网检索客户端 +====================== +调用 AnySearch 网页/领域搜索 API (https://api.anysearch.com/v1/search)， +为 AI 聊天注入实时联网检索结果，亦可被后端端点 / Agent 工具复用。 + +约束：平台容器存在智能路由代理，对外部 API 偶发重置；本客户端使用 +trust_env=False 直连 (等价于 NO_PROXY=api.anysearch.com)，规避代理重置。 +失败一律返回空/降级结果，保证聊天链路不中断。 +""" +import json +import logging +from typing import List, Dict, Optional + +logger = logging.getLogger("aiflow.anysearch") + +DEFAULT_API_URL = "https://api.anysearch.com/v1/search" +DEFAULT_ZONE = "cn" +REQUEST_TIMEOUT = 8.0 + + +async def anysearch_search( + api_key: str, + api_url: str = DEFAULT_API_URL, + query: str = "", + top_k: int = 5, + zone: str = DEFAULT_ZONE, + freshness: Optional[str] = None, + timeout: float = REQUEST_TIMEOUT, +) -> List[Dict]: + """调用 AnySearch 搜索，返回结构化结果列表。 + + 每个结果: + { + "content": snippet 摘要文本, + "score": 0.9..0.5 递减, + "source": "web_anysearch", + "metadata": {"url", "title", "source_domain", "zone"} + } + 失败时返回 [] (不抛异常)。 + """ + if not api_key: + return [] + try: + import httpx + except ImportError: + logger.warning("httpx not available for AnySearch") + return [] + + payload = { + "query": query, + "zone": zone, + "top_k": max(1, min(int(top_k), 20)), + } + if freshness: + payload["freshness"] = freshness + + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {api_key}", + } + try: + # trust_env=False -> 忽略容器内 HTTP(S)_PROXY，直连 api.anysearch.com + async with httpx.AsyncClient(trust_env=False, timeout=timeout) as client: + resp = await client.post(api_url, json=payload, headers=headers) + if resp.status_code != 200: + logger.warning(f"AnySearch HTTP {resp.status_code}: {resp.text[:200]}") + return [] + data = resp.json() + if data.get("code") != 0: + logger.warning(f"AnySearch biz error: {data.get('message')}") + return [] + results = (data.get("data") or {}).get("results") or [] + hits = [] + for idx, r in enumerate(results[:top_k]): + title = r.get("title", "") or "" + url = r.get("url", "") or "" + snippet = r.get("snippet") or r.get("content") or "" + if not snippet and not title: + continue + try: + from urllib.parse import urlparse + domain = urlparse(url).netloc + except Exception: + domain = "" + hits.append({ + "content": snippet, + "score": max(0.4, 0.9 - idx * 0.08), + "source": "web_anysearch", + "metadata": { + "url": url, + "title": title, + "source_domain": domain, + "zone": zone, + }, + }) + return hits + except Exception as e: + logger.warning(f"AnySearch search failed: {e}") + return [] + + +async def anysearch_health(api_key: str, api_url: str = DEFAULT_API_URL, timeout: float = REQUEST_TIMEOUT) -> Dict: + """连通性探测：返回 {"status": "connected"/"disconnected", "detail": ...}。""" + if not api_key: + return {"status": "disconnected", "detail": "未配置 API Key", "connected": False, "usable": False} + try: + import httpx + async with httpx.AsyncClient(trust_env=False, timeout=timeout) as client: + resp = await client.post( + api_url, + json={"query": "ping", "zone": DEFAULT_ZONE, "top_k": 1}, + headers={"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"}, + ) + if resp.status_code == 200: + return {"status": "connected", "detail": "AnySearch 连接正常", "connected": True, "usable": True} + return {"status": "error", "detail": f"HTTP {resp.status_code}", "connected": False, "usable": False} + except Exception as e: + return {"status": "error", "detail": str(e)[:200], "connected": False, "usable": False} + + +def load_anysearch_config(db_session) -> Dict: + """从 site_settings 读取全局 AnySearch 配置（失败返回空字典）。""" + if db_session is None: + return {} + try: + from app.models import SiteSetting + row = db_session.query(SiteSetting).filter(SiteSetting.key == "anysearch_config").first() + if not row or not row.value: + return {} + cfg = json.loads(row.value) + if not isinstance(cfg, dict): + return {} + return cfg + except Exception as e: + logger.warning(f"load_anysearch_config failed: {e}") + return {} + + +def save_anysearch_config( + db_session, + api_key: str, + is_enabled: bool = True, + api_url: str = DEFAULT_API_URL, + zone: str = DEFAULT_ZONE, + freshness: Optional[str] = None, + top_k: int = 5, +) -> bool: + """保存/更新全局 AnySearch 配置到 site_settings。""" + if db_session is None: + return False + try: + from app.models import SiteSetting + from datetime import datetime + cfg = { + "api_key": api_key.strip(), + "is_enabled": bool(is_enabled), + "api_url": api_url.strip() or DEFAULT_API_URL, + "zone": zone or DEFAULT_ZONE, + "freshness": freshness, + "top_k": max(1, min(int(top_k), 20)), + } + row = db_session.query(SiteSetting).filter(SiteSetting.key == "anysearch_config").first() + if row: + row.value = json.dumps(cfg, ensure_ascii=False) + row.updated_at = datetime.now() + else: + from app.models import gen_uuid + row = SiteSetting(id=gen_uuid(), key="anysearch_config", value=json.dumps(cfg, ensure_ascii=False)) + db_session.add(row) + db_session.commit() + return True + except Exception as e: + logger.warning(f"save_anysearch_config failed: {e}") + try: + db_session.rollback() + except Exception: + pass + return False diff --git a/backend/python/app/api/admin/knowledge_admin.py b/backend/python/app/api/admin/knowledge_admin.py index 99f3d9d..236057d 100644 --- a/backend/python/app/api/admin/knowledge_admin.py +++ b/backend/python/app/api/admin/knowledge_admin.py @@ -32,12 +32,17 @@ logger = logging.getLogger("aiflow") class ExternalServiceConfig(BaseModel): name: str = Field(min_length=1, max_length=200) - service_type: str = Field(..., description="dify / anysearch") + service_type: str = Field(..., description="dify") api_url: str = Field(min_length=1, max_length=500) api_key: Optional[str] = None config: Optional[Dict[str, Any]] = None +class AnySearchConfig(BaseModel): + api_key: str = Field(min_length=1, max_length=500) + is_enabled: bool = True + + class ExternalServiceUpdate(BaseModel): name: Optional[str] = None api_url: Optional[str] = None @@ -739,14 +744,11 @@ def _get_external_services_table(db: Session): async def _check_service_connection(service_type: str, api_url: str, api_key: str) -> dict: - """测试外部知识库服务连接状态（支持 Dify / AnySearch）""" + """测试外部知识库服务连接状态（支持 Dify）""" try: if service_type == "dify": from ...dify_kb import dify_test return dify_test(api_url, api_key) - if service_type == "anysearch": - from ...anysearch_client import anysearch_health - return await anysearch_health(api_key or "", api_url or "https://api.anysearch.com/v1/search") else: # 未知类型（如历史遗留的 fastgpt/ragflow），按通用方式探测可达性 headers = {} @@ -784,7 +786,6 @@ async def list_external_services( "items": services, "supported_types": [ {"type": "dify", "name": "Dify", "description": "LLM 应用开发平台，内置知识库管理（RAG）"}, - {"type": "anysearch", "name": "AnySearch 联网检索", "description": "实时联网网页/领域搜索，为 AI 聊天注入最新网络信息"}, ] } } @@ -797,8 +798,8 @@ async def create_external_service( db: Session = Depends(get_db), ): """创建新的外部知识库服务连接配置""" - # 验证服务类型（当前仅支持 Dify） - valid_types = ["dify", "anysearch"] + # 验证服务类型（当前仅支持 Dify；AnySearch 已改为全局固定配置） + valid_types = ["dify"] if req.service_type not in valid_types: raise HTTPException(status_code=400, detail=f"不支持的服务类型: {req.service_type}") @@ -881,6 +882,54 @@ async def create_external_service( } +@router.get("/anysearch/config", summary="获取 AnySearch 全局配置") +async def get_anysearch_config( + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """读取全局 AnySearch 联网检索配置（site_settings 表）。""" + from ...anysearch_client import load_anysearch_config, DEFAULT_API_URL + cfg = load_anysearch_config(db) + return { + "code": 0, + "data": { + "api_key": cfg.get("api_key", ""), + "is_enabled": cfg.get("is_enabled", False), + "api_url": cfg.get("api_url") or DEFAULT_API_URL, + "zone": cfg.get("zone", "cn"), + "freshness": cfg.get("freshness"), + "top_k": cfg.get("top_k", 5), + }, + } + + +@router.post("/anysearch/config", summary="保存 AnySearch 全局配置") +async def save_anysearch_config_endpoint( + req: AnySearchConfig, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """保存全局 AnySearch 联网检索配置，并测试连通性。""" + from ...anysearch_client import save_anysearch_config, anysearch_health, DEFAULT_API_URL + ok = save_anysearch_config(db, api_key=req.api_key, is_enabled=req.is_enabled, api_url=DEFAULT_API_URL) + if not ok: + raise HTTPException(status_code=500, detail="保存 AnySearch 配置失败") + health = await anysearch_health(req.api_key or "", DEFAULT_API_URL) + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="update_anysearch_config", + target_type="site_setting", + target_id="anysearch_config", + detail=f"更新 AnySearch 全局配置: enabled={req.is_enabled}, connected={health.get('connected')}", + ) + return { + "code": 0, + "message": "AnySearch 配置已保存", + "data": {"connection": health}, + } + + @router.put("/external/services/{service_id}", summary="更新外部知识库服务配置") async def update_external_service( service_id: str, diff --git a/backend/python/app/api/knowledge.py b/backend/python/app/api/knowledge.py index fe7625b..d708351 100644 --- a/backend/python/app/api/knowledge.py +++ b/backend/python/app/api/knowledge.py @@ -1496,27 +1496,18 @@ async def web_search_endpoint( ): """联网检索（AnySearch）：返回实时网页结果，供前端 / 智能体引用。""" try: - from sqlalchemy import text - import json as _json - from ..anysearch_client import anysearch_search - row = db.execute(text( - "SELECT api_key, api_url, config FROM external_kb_services " - "WHERE service_type='anysearch' AND is_enabled=1 LIMIT 1" - )).fetchone() - if not row: + from ..anysearch_client import anysearch_search, load_anysearch_config + cfg = load_anysearch_config(db) + if not cfg.get("is_enabled") or not cfg.get("api_key"): return {"code": 0, "data": {"results": [], "enabled": False, "message": "未配置 AnySearch 联网检索服务"}} - api_key = row[0] or "" - api_url = row[1] or "https://api.anysearch.com/v1/search" - cfg = row[2] - if isinstance(cfg, str): - try: - cfg = _json.loads(cfg) - except Exception: - cfg = {} - cfg = cfg or {} - zone = cfg.get("zone", "cn") - freshness = cfg.get("freshness") - hits = await anysearch_search(api_key, api_url, req.query, top_k=req.top_k, zone=zone, freshness=freshness) + hits = await anysearch_search( + cfg["api_key"], + cfg.get("api_url") or "https://api.anysearch.com/v1/search", + req.query, + top_k=req.top_k, + zone=cfg.get("zone", "cn"), + freshness=cfg.get("freshness"), + ) return {"code": 0, "data": {"results": hits, "enabled": True}} except Exception as e: logger.warning(f"web-search endpoint failed: {e}") diff --git a/backend/python/app/kb_router.py b/backend/python/app/kb_router.py index 8425bb0..5734857 100644 --- a/backend/python/app/kb_router.py +++ b/backend/python/app/kb_router.py @@ -1,262 +1,271 @@ -""" -知识库路由层 -============ -将检索请求智能路由到本地BM25、向量检索、或外部KB服务(Dify)。 -根据知识库配置自动选择最优检索策略。 - -使用方式: - from app.kb_router import kb_router_search - results = await kb_router_search(query, kb_id, top_k=5) -""" - -import logging -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服务（仅 dify）""" - from sqlalchemy import text - try: - query = "SELECT * FROM external_kb_services" - if enabled_only: - query += " WHERE is_enabled = 1" - query += " ORDER BY created_at DESC" - result = db_session.execute(text(query)) - columns = result.keys() - return [dict(zip(columns, row)) for row in result.fetchall()] - except Exception as e: - logger.error(f"Failed to get external services: {e}") - return [] - - -async def _search_dify(service: Dict, query: str, top_k: int = 5) -> List[Dict]: - """检索 Dify 知识库（走 app.dify_kb 客户端）""" - try: - 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_external(service: Dict, query: str, top_k: int = 5) -> List[Dict]: - """路由到对应的外部KB服务（当前仅支持 dify）""" - service_type = service.get("service_type", "") - if service_type == "dify": - return await _search_dify(service, query, top_k) - else: - logger.warning(f"Unknown external KB type: {service_type}") - return [] - -async def _get_web_services(db_session, enabled_only: bool = True) -> List[Dict]: - """获取已启用的 AnySearch 联网检索服务配置。""" - from sqlalchemy import text - try: - query = "SELECT * FROM external_kb_services WHERE service_type='anysearch'" - if enabled_only: - query += " AND is_enabled = 1" - query += " ORDER BY created_at DESC" - result = db_session.execute(text(query)) - columns = result.keys() - return [dict(zip(columns, row)) for row in result.fetchall()] - except Exception as e: - logger.error(f"Failed to get AnySearch services: {e}") - return [] - - -async def _search_anysearch(service: Dict, query: str, top_k: int = 5) -> List[Dict]: - """调用 AnySearch 联网检索（走 app.anysearch_client）。""" - try: - from app.anysearch_client import anysearch_search - api_key = service.get("api_key", "") - api_url = service.get("api_url") or "https://api.anysearch.com/v1/search" - cfg = service.get("config") - if isinstance(cfg, str): - try: - import json - cfg = json.loads(cfg) - except Exception: - cfg = None - cfg = cfg or {} - zone = cfg.get("zone", "cn") - freshness = cfg.get("freshness") - k = int(cfg.get("top_k", top_k)) - return await anysearch_search(api_key, api_url, query, top_k=k, zone=zone, freshness=freshness) - except Exception as e: - logger.error(f"AnySearch search error: {e}") - return [] - - -async def kb_router_search( - query: str, - kb_id: Optional[str] = None, - kb_ids: Optional[List[str]] = None, - top_k: int = 5, - bm25_results: Optional[List[Dict]] = None, - db_session=None, - use_vector: bool = True, - use_external: bool = True, - use_tree: bool = True, - use_web: bool = False, -) -> List[Dict]: - """ - 知识库路由层 — 混合检索 - - 检索策略: - 1. 本地 BM25 检索 (MySQL, 已有) - 2. 向量语义检索 (PostgreSQL+pgvector, 新增) - 3. 结构化分片段检索 (TreeSearch, 新增: 知识库目录树 + chunk 级精准命中) - 4. 外部KB检索 (Dify, 按配置) - 5. 联网检索 (AnySearch, 可选 use_web) - - 融合方式: Reciprocal Rank Fusion (RRF) - - Args: - query: 查询文本 - kb_id: 知识库ID - top_k: 返回结果数 - bm25_results: BM25检索结果 - db_session: 数据库会话(用于获取外部KB配置与分片段检索) - use_vector: 是否使用向量检索 - use_external: 是否使用外部KB检索 - use_tree: 是否使用 TreeSearch 结构化分片段检索 - use_web: 是否使用 AnySearch 联网检索 - - Returns: - 融合后的检索结果列表 - """ - all_results = [] - - # 1. BM25 结果 - if bm25_results: - for rank, item in enumerate(bm25_results): - all_results.append({ - "content": item.get("content", item.get("text", "")), - "score": item.get("score", 1.0 / (60 + rank + 1)), - "source": "local_bm25", - "rank": rank + 1, - "metadata": item.get("metadata", {}), - }) - - # 2. 向量检索 - if use_vector: - try: - from app.pgvector_search import vector_search - kb_targets = kb_ids if kb_ids else ([kb_id] if kb_id else [None]) - vec_results = [] - for _kid in kb_targets: - try: - vec_results.extend(await vector_search(query, _kid, top_k)) - except Exception as e: - logger.warning(f"Vector search failed for kb={_kid}: {e}") - for rank, item in enumerate(vec_results): - all_results.append({ - "content": item.get("content", ""), - "score": item.get("score", 0.5), - "source": "vector_pg", - "rank": rank + 1, - "metadata": item.get("metadata", {}), - }) - except ImportError: - logger.debug("pgvector_search not available, skipping vector search") - except Exception as e: - logger.warning(f"Vector search failed: {e}") - - # 2.5 结构化分片段检索 (TreeSearch, Level 2) - if use_tree and db_session is not None: - try: - from app.tree_search import tree_search - kb_targets = kb_ids if kb_ids else ([kb_id] if kb_id else None) - tree_hits = await tree_search( - db_session, query, kb_ids=kb_targets, top_k=top_k, mode="hybrid", - ) - for rank, item in enumerate(tree_hits): - all_results.append({ - "content": item.get("content", ""), - "score": item.get("score", 0.5), - "source": "tree_chunk", - "rank": rank + 1, - "metadata": item.get("metadata", {}), - }) - except Exception as e: - logger.warning(f"TreeSearch failed: {e}") - - # 3. 外部KB检索 (Dify) - if use_external and db_session: - services = await _get_external_services(db_session) - for service in services: - if service.get("status") == "connected": - ext_results = await _search_external(service, query, top_k) - for rank, item in enumerate(ext_results): - all_results.append({ - "content": item.get("content", ""), - "score": item.get("score", 0.5), - "source": item.get("source", "external_dify"), - "rank": rank + 1, - "metadata": item.get("metadata", {}), - }) - - # 4. 联网检索 (AnySearch, 可选) - if use_web and db_session is not None: - try: - web_services = await _get_web_services(db_session) - for svc in web_services: - web_hits = await _search_anysearch(svc, query, top_k) - for rank, item in enumerate(web_hits): - all_results.append({ - "content": item.get("content", ""), - "score": item.get("score", 0.6), - "source": item.get("source", "web_anysearch"), - "rank": rank + 1, - "metadata": item.get("metadata", {}), - }) - except Exception as e: - logger.warning(f"AnySearch web search failed: {e}") - - if not all_results: - return [] - - # RRF 融合 - rrf_k = 60 - score_map = {} - for item in all_results: - content_key = item["content"][:200] # 用内容前200字符作为去重键 - rrf_score = 1.0 / (rrf_k + item.get("rank", 1)) - - if content_key in score_map: - score_map[content_key]["score"] += rrf_score - score_map[content_key]["sources"].append(item["source"]) - else: - score_map[content_key] = { - "content": item["content"], - "score": rrf_score, - "sources": [item["source"]], - "metadata": item.get("metadata", {}), - } - - # 按融合分数排序 - merged = sorted(score_map.values(), key=lambda x: x["score"], reverse=True) - - # 标准化分数到 0-1 - if merged: - max_score = merged[0]["score"] - for item in merged: - item["score"] = item["score"] / max_score if max_score > 0 else 0 - item["sources"] = list(set(item["sources"])) - - return merged[:top_k] +""" +知识库路由层 +============ +将检索请求智能路由到本地BM25、向量检索、或外部KB服务(Dify)。 +根据知识库配置自动选择最优检索策略。 + +使用方式: + from app.kb_router import kb_router_search + results = await kb_router_search(query, kb_id, top_k=5) +""" + +import logging +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服务（仅 dify）""" + from sqlalchemy import text + try: + query = "SELECT * FROM external_kb_services" + if enabled_only: + query += " WHERE is_enabled = 1" + query += " ORDER BY created_at DESC" + result = db_session.execute(text(query)) + columns = result.keys() + return [dict(zip(columns, row)) for row in result.fetchall()] + except Exception as e: + logger.error(f"Failed to get external services: {e}") + return [] + + +async def _search_dify(service: Dict, query: str, top_k: int = 5) -> List[Dict]: + """检索 Dify 知识库（走 app.dify_kb 客户端）""" + try: + 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_external(service: Dict, query: str, top_k: int = 5) -> List[Dict]: + """路由到对应的外部KB服务（当前仅支持 dify）""" + service_type = service.get("service_type", "") + if service_type == "dify": + return await _search_dify(service, query, top_k) + else: + logger.warning(f"Unknown external KB type: {service_type}") + return [] + +async def _get_web_services(db_session, enabled_only: bool = True) -> List[Dict]: + """获取全局 AnySearch 联网检索配置（从 site_settings 读取）。""" + from app.anysearch_client import load_anysearch_config + cfg = load_anysearch_config(db_session) + if not cfg: + return [] + if enabled_only and not cfg.get("is_enabled"): + return [] + api_key = cfg.get("api_key", "") + if not api_key: + return [] + return [{ + "id": "anysearch_global", + "service_type": "anysearch", + "api_key": api_key, + "api_url": cfg.get("api_url") or "https://api.anysearch.com/v1/search", + "config": { + "zone": cfg.get("zone", "cn"), + "freshness": cfg.get("freshness"), + "top_k": cfg.get("top_k", 5), + }, + "is_enabled": cfg.get("is_enabled", False), + }] + + +async def _search_anysearch(service: Dict, query: str, top_k: int = 5) -> List[Dict]: + """调用 AnySearch 联网检索（走 app.anysearch_client）。""" + try: + from app.anysearch_client import anysearch_search + api_key = service.get("api_key", "") + api_url = service.get("api_url") or "https://api.anysearch.com/v1/search" + cfg = service.get("config") + if isinstance(cfg, str): + try: + import json + cfg = json.loads(cfg) + except Exception: + cfg = None + cfg = cfg or {} + zone = cfg.get("zone", "cn") + freshness = cfg.get("freshness") + k = int(cfg.get("top_k", top_k)) + return await anysearch_search(api_key, api_url, query, top_k=k, zone=zone, freshness=freshness) + except Exception as e: + logger.error(f"AnySearch search error: {e}") + return [] + + +async def kb_router_search( + query: str, + kb_id: Optional[str] = None, + kb_ids: Optional[List[str]] = None, + top_k: int = 5, + bm25_results: Optional[List[Dict]] = None, + db_session=None, + use_vector: bool = True, + use_external: bool = True, + use_tree: bool = True, + use_web: bool = False, +) -> List[Dict]: + """ + 知识库路由层 — 混合检索 + + 检索策略: + 1. 本地 BM25 检索 (MySQL, 已有) + 2. 向量语义检索 (PostgreSQL+pgvector, 新增) + 3. 结构化分片段检索 (TreeSearch, 新增: 知识库目录树 + chunk 级精准命中) + 4. 外部KB检索 (Dify, 按配置) + 5. 联网检索 (AnySearch, 可选 use_web) + + 融合方式: Reciprocal Rank Fusion (RRF) + + Args: + query: 查询文本 + kb_id: 知识库ID + top_k: 返回结果数 + bm25_results: BM25检索结果 + db_session: 数据库会话(用于获取外部KB配置与分片段检索) + use_vector: 是否使用向量检索 + use_external: 是否使用外部KB检索 + use_tree: 是否使用 TreeSearch 结构化分片段检索 + use_web: 是否使用 AnySearch 联网检索 + + Returns: + 融合后的检索结果列表 + """ + all_results = [] + + # 1. BM25 结果 + if bm25_results: + for rank, item in enumerate(bm25_results): + all_results.append({ + "content": item.get("content", item.get("text", "")), + "score": item.get("score", 1.0 / (60 + rank + 1)), + "source": "local_bm25", + "rank": rank + 1, + "metadata": item.get("metadata", {}), + }) + + # 2. 向量检索 + if use_vector: + try: + from app.pgvector_search import vector_search + kb_targets = kb_ids if kb_ids else ([kb_id] if kb_id else [None]) + vec_results = [] + for _kid in kb_targets: + try: + vec_results.extend(await vector_search(query, _kid, top_k)) + except Exception as e: + logger.warning(f"Vector search failed for kb={_kid}: {e}") + for rank, item in enumerate(vec_results): + all_results.append({ + "content": item.get("content", ""), + "score": item.get("score", 0.5), + "source": "vector_pg", + "rank": rank + 1, + "metadata": item.get("metadata", {}), + }) + except ImportError: + logger.debug("pgvector_search not available, skipping vector search") + except Exception as e: + logger.warning(f"Vector search failed: {e}") + + # 2.5 结构化分片段检索 (TreeSearch, Level 2) + if use_tree and db_session is not None: + try: + from app.tree_search import tree_search + kb_targets = kb_ids if kb_ids else ([kb_id] if kb_id else None) + tree_hits = await tree_search( + db_session, query, kb_ids=kb_targets, top_k=top_k, mode="hybrid", + ) + for rank, item in enumerate(tree_hits): + all_results.append({ + "content": item.get("content", ""), + "score": item.get("score", 0.5), + "source": "tree_chunk", + "rank": rank + 1, + "metadata": item.get("metadata", {}), + }) + except Exception as e: + logger.warning(f"TreeSearch failed: {e}") + + # 3. 外部KB检索 (Dify) + if use_external and db_session: + services = await _get_external_services(db_session) + for service in services: + if service.get("status") == "connected": + ext_results = await _search_external(service, query, top_k) + for rank, item in enumerate(ext_results): + all_results.append({ + "content": item.get("content", ""), + "score": item.get("score", 0.5), + "source": item.get("source", "external_dify"), + "rank": rank + 1, + "metadata": item.get("metadata", {}), + }) + + # 4. 联网检索 (AnySearch, 可选) + if use_web and db_session is not None: + try: + web_services = await _get_web_services(db_session) + for svc in web_services: + web_hits = await _search_anysearch(svc, query, top_k) + for rank, item in enumerate(web_hits): + all_results.append({ + "content": item.get("content", ""), + "score": item.get("score", 0.6), + "source": item.get("source", "web_anysearch"), + "rank": rank + 1, + "metadata": item.get("metadata", {}), + }) + except Exception as e: + logger.warning(f"AnySearch web search failed: {e}") + + if not all_results: + return [] + + # RRF 融合 + rrf_k = 60 + score_map = {} + for item in all_results: + content_key = item["content"][:200] # 用内容前200字符作为去重键 + rrf_score = 1.0 / (rrf_k + item.get("rank", 1)) + + if content_key in score_map: + score_map[content_key]["score"] += rrf_score + score_map[content_key]["sources"].append(item["source"]) + else: + score_map[content_key] = { + "content": item["content"], + "score": rrf_score, + "sources": [item["source"]], + "metadata": item.get("metadata", {}), + } + + # 按融合分数排序 + merged = sorted(score_map.values(), key=lambda x: x["score"], reverse=True) + + # 标准化分数到 0-1 + if merged: + max_score = merged[0]["score"] + for item in merged: + item["score"] = item["score"] / max_score if max_score > 0 else 0 + item["sources"] = list(set(item["sources"])) + + return merged[:top_k] diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index 1131d0d..8c64c6f 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -959,6 +959,10 @@ export const adminAPI = { checkAdminKnowledgeExtStatus: (id: string) => get(`/admin/knowledge/external/services/${id}/status`), + getAdminAnySearchConfig: () => get('/admin/knowledge/anysearch/config'), + + saveAdminAnySearchConfig: (data: { api_key: string; is_enabled: boolean }) => post('/admin/knowledge/anysearch/config', data), + syncAdminKnowledge: (kbId: string, data: any) => post(`/admin/knowledge/external/sync/${kbId}`, data), batchDeleteAdminKnowledge: (ids: string[]) => post('/admin/knowledge/batch/delete', ids), diff --git a/frontend/src/pages/admin/Knowledge.tsx b/frontend/src/pages/admin/Knowledge.tsx index 4e41635..f51b564 100644 --- a/frontend/src/pages/admin/Knowledge.tsx +++ b/frontend/src/pages/admin/Knowledge.tsx @@ -71,6 +71,11 @@ const Knowledge: React.FC = () => { const [ingesting, setIngesting] = useState(false); const [ingestResult, setIngestResult] = useState(null); + // AnySearch global config + const [anysearchForm, setAnysearchForm] = useState<{ api_key: string; is_enabled: boolean }>({ api_key: '', is_enabled: false }); + const [anysearchLoading, setAnysearchLoading] = useState(false); + const [anysearchMsg, setAnysearchMsg] = useState(''); + // KB Modal const [showModal, setShowModal] = useState(false); const [editingKB, setEditingKB] = useState(null); @@ -123,6 +128,33 @@ const Knowledge: React.FC = () => { try { const res = await adminAPI.getAdminKnowledgeExtServices(); if (res.code === 0) setExtServices(res.data.items || []); } catch (e) {} }; + const fetchAnySearchConfig = async () => { + try { + const res = await adminAPI.getAdminAnySearchConfig(); + if (res.code === 0) { + const d = res.data || {}; + setAnysearchForm({ api_key: d.api_key || '', is_enabled: !!d.is_enabled }); + } + } catch (e) {} + }; + + const handleSaveAnySearch = async () => { + setAnysearchLoading(true); setAnysearchMsg(''); + try { + const res = await adminAPI.saveAdminAnySearchConfig(anysearchForm); + if (res.code === 0) { + const conn = res.data?.connection || {}; + setAnysearchMsg(conn.connected ? '✅ 已保存并连接成功' : `⚠️ 已保存但连接失败：${conn.detail || '未知错误'}`); + } else { + setAnysearchMsg(`❌ 保存失败：${res.message || '未知错误'}`); + } + } catch (e: any) { + setAnysearchMsg(`❌ 保存失败：${e?.message || '网络错误'}`); + } finally { + setAnysearchLoading(false); + } + }; + const handleCreateKB = async () => { try { const res = await adminAPI.createAdminKnowledge(kbForm); @@ -334,7 +366,7 @@ const Knowledge: React.FC = () => { - + {/* Table */} @@ -523,54 +555,62 @@ const Knowledge: React.FC = () => { {/* External Service Modal */} {showExtModal && (
setShowExtModal(false)}> -
e.stopPropagation()}> -

外部知识库服务配置

- {extServices.length > 0 && ( -
- - - - {extServices.map(svc => ( - - - - - - - - ))} - -
名称类型URL状态操作
{svc.name}Dify{svc.api_url}{statusBadge(svc.status)} - - -
-
- )} -
-

配置 Dify 知识库

-
+
e.stopPropagation()}> +

外部知识库服务配置

+ + {/* Section 1: Dify 外部服务列表 */} +
+

已接入外部服务

+ {extServices.length > 0 ? ( +
+ + + + {extServices.map(svc => ( + + + + + + + + ))} + +
名称类型URL状态操作
{svc.name}Dify{svc.api_url}{statusBadge(svc.status)} + + +
+
+ ) : ( +
暂无外部服务，可在下方添加 Dify 知识库。
+ )} +
+ + {/* Section 2: Dify 配置表单 */} +
+

配置 Dify 知识库

+
setExtForm({ ...extForm, name: e.target.value })} placeholder="如: 生产环境 Dify" style={fullInput} />
setExtForm({ ...extForm, dataset_name: e.target.value })} placeholder="如: qihuai" style={fullInput} />
-
setExtForm({ ...extForm, api_url: e.target.value })} placeholder="https://dify.qihuyun.com/v1" style={fullInput} />
-
setExtForm({ ...extForm, api_key: e.target.value })} placeholder="dataset-xxxx" style={fullInput} />
+
setExtForm({ ...extForm, api_url: e.target.value })} placeholder="https://dify.qihuyun.com/v1" style={fullInput} />
+
setExtForm({ ...extForm, api_key: e.target.value })} placeholder="dataset-xxxx" style={fullInput} />
{extMsg && ( -
+
{extMsg}
)} -
+
-
{ingestResult && ( -
+
入库结果
AI 聊天回答：{ingestResult.chat_answers} 条
用户上传文档：{ingestResult.user_docs} 条
@@ -578,6 +618,45 @@ const Knowledge: React.FC = () => {
)}
+ + {/* Section 3: AnySearch 全局固定配置 */} +
+

AnySearch 联网检索（全局配置）

+
+
+ + setAnysearchForm({ ...anysearchForm, api_key: e.target.value })} + placeholder="as_sk_xxxx" + style={fullInput} + /> +
+ +
+ + {anysearchMsg && ( +
+ {anysearchMsg} +
+ )} + +
+ + 配置对平台所有用户生效，AI 聊天将自动注入实时联网结果。 +
+
+ + {/* Footer */} +
+ +
)}