Implement scoped real knowledge retrieval

ai_knowledge_base_v2/apps/backend/app/api/admin_settings.py

@@ -69,7 +69,7 @@ def debug_agent(
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="模型不存在")
 
     scopes = _agent_knowledge_scopes(db, payload.knowledgeIds)
-    chunks = FeishuKnowledgeService.retrieve(payload.question, scopes)
+    chunks = FeishuKnowledgeService.retrieve(payload.question, scopes, db)
     prompt = _build_agent_debug_prompt(payload.promptContent, payload.question, chunks)
     rag_result = RagResult(question=payload.question, knowledge_scopes=scopes, chunks=chunks, prompt=prompt)
     result = ModelClientService.debug_model(

ai_knowledge_base_v2/apps/backend/app/core/config.py

@@ -34,9 +34,10 @@ class Settings(BaseSettings):
     mock_sms_enabled: bool = True
     mock_sms_code: str = "123456"
     sms_code_expire_minutes: int = 5
-    mock_rag_enabled: bool = True
+    mock_rag_enabled: bool = False
     mock_model_enabled: bool = True
-    feishu_mock_enabled: bool = True
+    feishu_mock_enabled: bool = False
+    feishu_search_url: str = ""
     feishu_app_id: str = ""
     feishu_app_secret: str = ""
     feishu_timeout_seconds: int = 20

ai_knowledge_base_v2/apps/backend/app/services/feishu_service.py

@@ -1,11 +1,15 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
 import time
 from typing import TYPE_CHECKING, Any
 
 import httpx
+from sqlalchemy import select
+from sqlalchemy.orm import Session
 
 from app.core.config import get_settings
+from app.models.ai_config import SystemConfig
 from app.services.external_errors import ExternalServiceError
 from app.services.knowledge_service import KnowledgeScope
 
@@ -19,6 +23,13 @@ FEISHU_OPEN_API = "https://open.feishu.cn"
 _token_cache: dict[str, Any] = {"token": "", "expires_at": 0.0}
 
 
+@dataclass(frozen=True)
+class FeishuRetrievalConfig:
+    search_url: str
+    timeout_seconds: int
+    retry_count: int
+
+
 def _get_tenant_token() -> str:
     """获取飞书 tenant_access_token，自动缓存"""
     settings = get_settings()
@@ -236,69 +247,51 @@ def _get_space_node_tree(space_id: str) -> list[dict]:
 
 
 class FeishuKnowledgeService:
-    """飞书知识库检索服务 - 直接调用飞书官方 API"""
+    """飞书知识库检索服务：优先搜索接口，失败时飞书直读。"""
 
     # 文档内容缓存: node_token -> (content, timestamp)
     _content_cache: dict[str, tuple[str, float]] = {}
     _CACHE_TTL = 300  # 缓存 5 分钟
 
     @classmethod
-    def retrieve(cls, question: str, scopes: list[KnowledgeScope]) -> list[RetrievedChunk]:
+    def retrieve(
+        cls,
+        question: str,
+        scopes: list[KnowledgeScope],
+        db: Session | None = None,
+    ) -> list[RetrievedChunk]:
         if not scopes:
             return []
 
-        settings = get_settings()
-        if settings.feishu_mock_enabled:
-            return cls._retrieve_mock(question, scopes)
+        config = _feishu_retrieval_config(db)
+        if config.search_url:
+            try:
+                return cls._retrieve_from_search(question, scopes, config)
+            except ExternalServiceError:
+                pass
         return cls._retrieve_from_feishu(question, scopes)
 
     @classmethod
-    def _retrieve_mock(cls, question: str, scopes: list[KnowledgeScope]) -> list[RetrievedChunk]:
+    def _retrieve_from_search(
+        cls,
+        question: str,
+        scopes: list[KnowledgeScope],
+        config: FeishuRetrievalConfig,
+    ) -> list[RetrievedChunk]:
         from app.services.rag_service import RetrievedChunk
 
-        mock_documents = [
-            {
-                "title": "一期产品目标",
-                "keywords": {"一期", "目标", "问答", "登录", "知识库", "飞书", "用户端", "后台"},
-                "content": "一期要交付企业飞书知识库 AI 问答系统，核心包括用户登录、AI 问答、基于已开放知识库回答和后台管理能力。",
-            },
-            {
-                "title": "知识库开放过滤规则",
-                "keywords": {"权限", "开放", "用户", "知识库", "禁用", "可见"},
-                "content": "RAG 检索前必须先过滤知识库开放状态，只允许已开放、未禁用的知识库参与回答，已开放知识库默认所有用户可见。",
-            },
-            {
-                "title": "无命中兜底规则",
-                "keywords": {"无命中", "兜底", "编造", "检索", "答案", "命中"},
-                "content": "当知识库没有命中相关内容时，系统必须固定返回无命中兜底文案。",
-            },
-            {
-                "title": "技术实现边界",
-                "keywords": {"模型", "prompt", "大模型", "日志", "sse", "流式", "追溯"},
-                "content": "后端需要组装系统 Prompt、检索片段、历史上下文和当前问题，通过 SSE 流式输出，并记录 AI 请求日志。",
-            },
-        ]
+        payload = _build_search_payload(question, scopes)
+        last_error: Exception | None = None
+        for _ in range(max(1, config.retry_count + 1)):
+            try:
+                response = httpx.post(config.search_url, json=payload, timeout=config.timeout_seconds)
+                response.raise_for_status()
+                body = response.json()
+                return _parse_search_chunks(body, scopes, RetrievedChunk)
+            except (ValueError, httpx.HTTPError) as exc:
+                last_error = exc
 
-        normalized_question = question.lower()
-        matched_documents = []
-        for document in mock_documents:
-            if any(keyword in normalized_question for keyword in document["keywords"]):
-                matched_documents.append(document)
-
-        if not matched_documents:
-            return []
-
-        primary_scope = scopes[0]
-        return [
-            RetrievedChunk(
-                knowledge_id=primary_scope.id,
-                knowledge_name=primary_scope.name,
-                title=document["title"],
-                content=document["content"],
-                source_url=None,
-            )
-            for document in matched_documents[:3]
-        ]
+        raise ExternalServiceError(f"飞书搜索接口调用失败：{last_error}", provider="feishu-search")
 
     @classmethod
     def _retrieve_from_feishu(cls, question: str, scopes: list[KnowledgeScope]) -> list[RetrievedChunk]:
@@ -493,3 +486,139 @@ class FeishuKnowledgeService:
         scored.sort(key=lambda x: x[4], reverse=True)
 
         return [(s, t, c, u) for s, t, c, u, _ in scored]
+
+
+def _feishu_retrieval_config(db: Session | None) -> FeishuRetrievalConfig:
+    settings = get_settings()
+    return FeishuRetrievalConfig(
+        search_url=_config_text(db, "feishu_search_url", settings.feishu_search_url),
+        timeout_seconds=_config_int(db, "feishu_timeout_seconds", settings.feishu_timeout_seconds, minimum=1, maximum=120),
+        retry_count=_config_int(db, "feishu_retry_count", settings.feishu_retry_count, minimum=0, maximum=10),
+    )
+
+
+def _config_text(db: Session | None, key: str, default: str) -> str:
+    value = _config_value(db, key)
+    return value.strip() if value is not None else default
+
+
+def _config_int(db: Session | None, key: str, default: int, *, minimum: int, maximum: int) -> int:
+    value = _config_value(db, key)
+    if value is not None:
+        try:
+            return max(minimum, min(maximum, int(float(value.strip()))))
+        except ValueError:
+            pass
+    return default
+
+
+def _config_value(db: Session | None, key: str) -> str | None:
+    if db is None:
+        return None
+    config = db.scalar(select(SystemConfig).where(SystemConfig.config_key == key))
+    if config is None or not config.config_value.strip():
+        return None
+    return config.config_value
+
+
+def _build_search_payload(question: str, scopes: list[KnowledgeScope]) -> dict:
+    scope_payload = [
+        {
+            "id": scope.id,
+            "name": scope.name,
+            "spaceId": scope.feishu_space_id,
+            "nodeId": scope.feishu_node_id,
+        }
+        for scope in scopes
+    ]
+    return {
+        "query": question.strip(),
+        "question": question.strip(),
+        "limit": 5,
+        "knowledgeScopes": scope_payload,
+        "scopes": scope_payload,
+        "spaceIds": [scope.feishu_space_id for scope in scopes if scope.feishu_space_id],
+        "nodeIds": [scope.feishu_node_id for scope in scopes if scope.feishu_node_id],
+    }
+
+
+def _parse_search_chunks(body: Any, scopes: list[KnowledgeScope], chunk_cls: type) -> list:
+    if isinstance(body, dict) and body.get("code") not in (None, 0, "0"):
+        raise ExternalServiceError(
+            f"飞书搜索接口返回错误：{body.get('message') or body.get('msg') or body.get('error') or '未知错误'}",
+            provider="feishu-search",
+        )
+
+    items = _extract_search_items(body)
+    chunks = []
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        scope = _match_search_scope(item, scopes)
+        if scope is None:
+            continue
+        content = _first_text(item, ("content", "text", "snippet", "summary", "answer"))
+        if not content:
+            continue
+        chunks.append(
+            chunk_cls(
+                knowledge_id=scope.id,
+                knowledge_name=scope.name,
+                title=_first_text(item, ("title", "name", "documentTitle", "docTitle")) or scope.name,
+                content=content,
+                source_url=_first_text(item, ("sourceUrl", "source_url", "url", "link")),
+            )
+        )
+        if len(chunks) >= 5:
+            break
+    return chunks
+
+
+def _extract_search_items(body: Any) -> list:
+    data = body.get("data", body) if isinstance(body, dict) else body
+    if isinstance(data, list):
+        return data
+    if not isinstance(data, dict):
+        return []
+    for key in ("chunks", "results", "items", "documents", "records"):
+        value = data.get(key)
+        if isinstance(value, list):
+            return value
+    return []
+
+
+def _match_search_scope(item: dict, scopes: list[KnowledgeScope]) -> KnowledgeScope | None:
+    knowledge_id = _first_text(item, ("knowledgeId", "knowledge_id", "kbId", "kb_id"))
+    if knowledge_id:
+        for scope in scopes:
+            if str(scope.id) == knowledge_id:
+                return scope
+
+    node_id = _first_text(item, ("nodeId", "node_id", "feishuNodeId", "feishu_node_id"))
+    if node_id:
+        for scope in scopes:
+            if scope.feishu_node_id == node_id:
+                return scope
+        return None
+
+    space_id = _first_text(item, ("spaceId", "space_id", "feishuSpaceId", "feishu_space_id"))
+    if space_id:
+        for scope in scopes:
+            if scope.feishu_space_id == space_id:
+                return scope
+        return None
+
+    # 搜索适配服务收到的请求已经被限定到 knowledgeScopes；
+    # 如果返回结果不带范围字段，只能按 scoped response 处理。
+    return scopes[0] if scopes else None
+
+
+def _first_text(item: dict, keys: tuple[str, ...]) -> str:
+    for key in keys:
+        value = item.get(key)
+        if value is None:
+            continue
+        text = str(value).strip()
+        if text:
+            return text
+    return ""

ai_knowledge_base_v2/apps/backend/app/services/knowledge_service.py

@@ -5,7 +5,6 @@ from dataclasses import dataclass
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
-from app.core.config import get_settings
 from app.models.knowledge import Knowledge
 from app.models.user import User
 
@@ -23,7 +22,7 @@ class KnowledgeAccessService:
     @staticmethod
     def get_allowed_knowledge(db: Session, user: User) -> list[KnowledgeScope]:
         rows = db.scalars(select(Knowledge).where(Knowledge.status == 1).order_by(Knowledge.id.asc()))
-        scopes = [
+        return [
             KnowledgeScope(
                 id=knowledge.id,
                 name=knowledge.name,
@@ -32,15 +31,3 @@ class KnowledgeAccessService:
             )
             for knowledge in rows
         ]
-        if scopes or not get_settings().mock_rag_enabled:
-            return scopes
-
-        return [
-            KnowledgeScope(
-                id=0,
-                name="开发阶段默认知识库",
-                feishu_space_id="mock-space",
-                feishu_node_id="mock-node",
-                is_mock=True,
-            )
-        ]

ai_knowledge_base_v2/apps/backend/app/services/model_service.py

@@ -87,8 +87,8 @@ def _mock_answer(rag_result: RagResult) -> str:
     return (
         "根据当前已开放知识库，整理到的信息如下：\n\n"
         f"{summaries}\n\n"
-        "当前仍是 mock RAG + mock 模型阶段，后续会把检索服务替换为飞书实时检索，"
-        "把模型服务替换为后台启用的大模型配置。"
+        "当前使用模型 mock 生成回答，知识片段来自搜索接口或飞书直读检索链路；"
+        "关闭模型 mock 后会使用后台启用的大模型配置。"
     )
 
 

ai_knowledge_base_v2/apps/backend/app/services/rag_service.py

@@ -42,7 +42,7 @@ class RagService:
     @staticmethod
     def build_result(db: Session, user: User, question: str) -> RagResult:
         scopes = KnowledgeAccessService.get_allowed_knowledge(db, user)
-        chunks = FeishuKnowledgeService.retrieve(question, scopes)
+        chunks = FeishuKnowledgeService.retrieve(question, scopes, db)
         prompt = PromptService.build_prompt(db, question, chunks)
         return RagResult(question=question, knowledge_scopes=scopes, chunks=chunks, prompt=prompt)