Add async Feishu retrieval for chat RAG

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

@@ -16,6 +16,7 @@ from app.services.ai_request_log_service import AiRequestLogService
 from app.services.chat_service import ChatService, _title_from_question
 from app.services.external_errors import ExternalServiceError
 from app.services.model_stream_service import ModelStreamService
+from app.services.rag_async_service import AsyncRagService
 from app.services.rag_service import RagService
 
 
@@ -176,7 +177,13 @@ class ChatStreamService:
         answer_parts: list[str] = []
 
         try:
-            rag_result = _build_rag_result(db, user, normalized_question, history, getattr(session, "summary", None))
+            rag_result = await AsyncRagService.build_result(
+                db,
+                user,
+                normalized_question,
+                history=history,
+                session_summary=getattr(session, "summary", None),
+            )
             model_response = ModelStreamService.stream_async(db, rag_result)
             async for chunk in model_response.chunks:
                 if chunk:

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

@@ -0,0 +1,388 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from typing import TYPE_CHECKING, Any
+
+import httpx
+from sqlalchemy.orm import Session
+
+from app.services.external_errors import ExternalServiceError
+from app.services.feishu_service import (
+    FEISHU_OPEN_API,
+    FeishuKnowledgeService,
+    FeishuRetrievalConfig,
+    _build_search_payload,
+    _feishu_retrieval_config,
+    _parse_docx_content,
+    _parse_search_chunks,
+    _response_error_detail,
+    _token_cache,
+    _token_cache_key,
+)
+from app.services.knowledge_service import KnowledgeScope
+
+if TYPE_CHECKING:
+    from app.services.rag_service import RetrievedChunk
+
+
+class AsyncFeishuKnowledgeService:
+    """飞书知识库异步检索服务，保持和同步检索相同的结果结构。"""
+
+    @classmethod
+    async def retrieve(
+        cls,
+        question: str,
+        scopes: list[KnowledgeScope],
+        db: Session | None = None,
+    ) -> list[RetrievedChunk]:
+        if not scopes:
+            return []
+
+        config = _feishu_retrieval_config(db)
+        if config.search_url:
+            try:
+                return await cls._retrieve_from_search(question, scopes, config)
+            except ExternalServiceError as exc:
+                try:
+                    return await cls._retrieve_from_feishu(question, scopes, config)
+                except ExternalServiceError as fallback_exc:
+                    raise ExternalServiceError(
+                        f"{exc}；飞书直读也失败：{fallback_exc}",
+                        provider="feishu",
+                    ) from fallback_exc
+        return await cls._retrieve_from_feishu(question, scopes, config)
+
+    @classmethod
+    async def _retrieve_from_search(
+        cls,
+        question: str,
+        scopes: list[KnowledgeScope],
+        config: FeishuRetrievalConfig,
+    ) -> list[RetrievedChunk]:
+        from app.services.rag_service import RetrievedChunk
+
+        payload = _build_search_payload(question, scopes)
+        last_error: Exception | None = None
+        async with httpx.AsyncClient(timeout=config.timeout_seconds) as client:
+            for _ in range(max(1, config.retry_count + 1)):
+                try:
+                    response = await client.post(config.search_url, json=payload)
+                    response.raise_for_status()
+                    body = response.json()
+                    return _parse_search_chunks(body, scopes, RetrievedChunk)
+                except (ValueError, httpx.HTTPError) as exc:
+                    last_error = exc
+
+        raise ExternalServiceError(f"飞书搜索接口调用失败：{last_error}", provider="feishu-search")
+
+    @classmethod
+    async def _retrieve_from_feishu(
+        cls,
+        question: str,
+        scopes: list[KnowledgeScope],
+        config: FeishuRetrievalConfig,
+    ) -> list[RetrievedChunk]:
+        from app.services.rag_service import RetrievedChunk
+
+        if not config.app_id or not config.app_secret:
+            raise ExternalServiceError(
+                "未配置飞书应用凭证，无法读取飞书知识库。请配置 FEISHU_APP_ID/FEISHU_APP_SECRET，"
+                "或在系统设置中填写飞书 AppID/AppSecret。",
+                provider="feishu",
+            )
+
+        semaphore = asyncio.Semaphore(3)
+
+        async def load_scope(scope: KnowledgeScope):
+            async with semaphore:
+                return await cls._load_documents(scope, config)
+
+        results = await asyncio.gather(*(load_scope(scope) for scope in scopes), return_exceptions=True)
+
+        all_chunks: list[tuple[KnowledgeScope, str, str, str | None]] = []
+        load_errors: list[str] = []
+        for scope, result in zip(scopes, results, strict=False):
+            if isinstance(result, ExternalServiceError):
+                load_errors.append(f"{scope.name}：{result}")
+                continue
+            if isinstance(result, Exception):
+                load_errors.append(f"{scope.name}：{result}")
+                continue
+            for title, content, source_url in result:
+                all_chunks.append((scope, title, content, source_url))
+
+        if not all_chunks:
+            if load_errors:
+                raise ExternalServiceError(
+                    "飞书知识库读取失败：" + "；".join(load_errors[:3]),
+                    provider="feishu",
+                )
+            return []
+
+        scored_chunks = FeishuKnowledgeService._score_and_rank(question, all_chunks)
+        return [
+            RetrievedChunk(
+                knowledge_id=scope.id,
+                knowledge_name=scope.name,
+                title=title,
+                content=content,
+                source_url=source_url,
+            )
+            for scope, title, content, source_url in scored_chunks[:5]
+        ]
+
+    @classmethod
+    async def _load_documents(
+        cls,
+        scope: KnowledgeScope,
+        config: FeishuRetrievalConfig,
+    ) -> list[tuple[str, str, str | None]]:
+        node_id = scope.feishu_node_id
+        base_url = "https://zcn7hk6n047c.feishu.cn"
+
+        if not node_id:
+            return []
+
+        try:
+            node_info = await cls._get_node_info_with_cache(node_id, config)
+        except ExternalServiceError as node_exc:
+            try:
+                content = await cls._get_document_content_with_cache(node_id, config)
+                return [(node_id, content, f"{base_url}/docx/{node_id}")]
+            except ExternalServiceError as docx_exc:
+                raise ExternalServiceError(
+                    f"飞书节点读取失败：{node_exc}；按普通 docx token 读取也失败：{docx_exc}",
+                    provider="feishu",
+                ) from docx_exc
+
+        obj_type = node_info.get("obj_type", "")
+        obj_token = node_info.get("obj_token", "")
+        title = node_info.get("title", node_id)
+        space_id = node_info.get("space_id", scope.feishu_space_id)
+
+        if obj_type == "docx":
+            content = await cls._get_document_content_with_cache(obj_token, config)
+            return [(title, content, f"{base_url}/wiki/{node_id}")]
+        if obj_type == "file":
+            content = await cls._get_file_content_with_cache(obj_token, config)
+            return [(title, content, f"{base_url}/wiki/{node_id}")]
+
+        children = await cls._get_wiki_children_with_cache(space_id, node_id, config)
+        documents: list[tuple[str, str, str | None]] = []
+        for child in children:
+            child_node_token = child.get("node_token", "")
+            child_obj_type = child.get("obj_type", "")
+            child_obj_token = child.get("obj_token", "")
+            child_title = child.get("title", "")
+
+            try:
+                if child_obj_type == "docx" and child_obj_token:
+                    content = await cls._get_document_content_with_cache(child_obj_token, config)
+                elif child_obj_type == "file" and child_obj_token:
+                    content = await cls._get_file_content_with_cache(child_obj_token, config)
+                else:
+                    continue
+                documents.append((child_title, content, f"{base_url}/wiki/{child_node_token}"))
+            except ExternalServiceError:
+                continue
+
+        if obj_type == "docx" and obj_token:
+            try:
+                content = await cls._get_document_content_with_cache(obj_token, config)
+                documents.insert(0, (title, content, f"{base_url}/wiki/{node_id}"))
+            except ExternalServiceError:
+                pass
+
+        return documents
+
+    @classmethod
+    async def _get_node_info_with_cache(cls, node_token: str, config: FeishuRetrievalConfig) -> dict:
+        cache_key = f"{config.app_id}:node_info:{node_token}"
+        now = time.time()
+
+        if cache_key in FeishuKnowledgeService._content_cache:
+            cached_data, ts = FeishuKnowledgeService._content_cache[cache_key]
+            if now - ts < FeishuKnowledgeService._CACHE_TTL:
+                return eval(cached_data)  # noqa: S307
+
+        info = await _get_wiki_node_info_async(node_token, config)
+        FeishuKnowledgeService._content_cache[cache_key] = (str(info), now)
+        return info
+
+    @classmethod
+    async def _get_document_content_with_cache(cls, doc_token: str, config: FeishuRetrievalConfig) -> str:
+        cache_key = f"{config.app_id}:doc:{doc_token}"
+        now = time.time()
+
+        if cache_key in FeishuKnowledgeService._content_cache:
+            content, ts = FeishuKnowledgeService._content_cache[cache_key]
+            if now - ts < FeishuKnowledgeService._CACHE_TTL:
+                return content
+
+        content = await _get_docx_content_async(doc_token, config)
+        FeishuKnowledgeService._content_cache[cache_key] = (content, now)
+        return content
+
+    @classmethod
+    async def _get_file_content_with_cache(cls, file_token: str, config: FeishuRetrievalConfig) -> str:
+        cache_key = f"{config.app_id}:file:{file_token}"
+        now = time.time()
+
+        if cache_key in FeishuKnowledgeService._content_cache:
+            content, ts = FeishuKnowledgeService._content_cache[cache_key]
+            if now - ts < FeishuKnowledgeService._CACHE_TTL:
+                return content
+
+        content = await _get_file_content_async(file_token, config)
+        FeishuKnowledgeService._content_cache[cache_key] = (content, now)
+        return content
+
+    @classmethod
+    async def _get_wiki_children_with_cache(
+        cls,
+        space_id: str,
+        node_token: str,
+        config: FeishuRetrievalConfig,
+    ) -> list[dict]:
+        cache_key = f"{config.app_id}:children:{space_id}:{node_token}"
+        now = time.time()
+
+        if cache_key in FeishuKnowledgeService._content_cache:
+            cached_data, ts = FeishuKnowledgeService._content_cache[cache_key]
+            if now - ts < FeishuKnowledgeService._CACHE_TTL:
+                return eval(cached_data)  # noqa: S307
+
+        children = await _get_wiki_children_async(space_id, node_token, config)
+        FeishuKnowledgeService._content_cache[cache_key] = (str(children), now)
+        return children
+
+
+async def _get_tenant_token_async(config: FeishuRetrievalConfig) -> str:
+    now = time.time()
+    cache_key = _token_cache_key(config.app_id, config.app_secret)
+    cached = _token_cache.get(cache_key)
+
+    if cached and cached["token"] and now < cached["expires_at"]:
+        return cached["token"]
+
+    url = f"{FEISHU_OPEN_API}/open-apis/auth/v3/tenant_access_token/internal"
+    payload = {"app_id": config.app_id, "app_secret": config.app_secret}
+
+    try:
+        async with httpx.AsyncClient(timeout=config.timeout_seconds) as client:
+            resp = await client.post(url, json=payload)
+        resp.raise_for_status()
+        data = resp.json()
+        if data.get("code") != 0:
+            raise ExternalServiceError(
+                f"获取飞书Token失败：{data.get('msg', '未知错误')}",
+                provider="feishu",
+            )
+
+        token = data["tenant_access_token"]
+        expire = data.get("expire", 7200)
+        _token_cache[cache_key] = {"token": token, "expires_at": now + expire - 300}
+        return token
+    except httpx.HTTPError as exc:
+        raise ExternalServiceError(f"获取飞书Token网络错误：{exc}", provider="feishu") from exc
+
+
+async def _feishu_get_async(path: str, *, params: dict | None = None, config: FeishuRetrievalConfig) -> dict:
+    token = await _get_tenant_token_async(config)
+    url = f"{FEISHU_OPEN_API}{path}"
+    headers = {"Authorization": f"Bearer {token}"}
+
+    try:
+        async with httpx.AsyncClient(timeout=config.timeout_seconds) as client:
+            resp = await client.get(url, headers=headers, params=params)
+        resp.raise_for_status()
+        data = resp.json()
+    except httpx.HTTPStatusError as exc:
+        raise ExternalServiceError(
+            f"飞书HTTP请求失败 [{path}]：{exc.response.status_code} {_response_error_detail(exc.response)}",
+            provider="feishu",
+        ) from exc
+    except httpx.HTTPError as exc:
+        raise ExternalServiceError(f"飞书网络请求失败 [{path}]：{exc}", provider="feishu") from exc
+
+    if data.get("code") != 0:
+        raise ExternalServiceError(
+            f"飞书API调用失败 [{path}]：{data.get('msg', '未知错误')}",
+            provider="feishu",
+        )
+    return data
+
+
+async def _feishu_download_async(path: str, config: FeishuRetrievalConfig) -> bytes:
+    token = await _get_tenant_token_async(config)
+    url = f"{FEISHU_OPEN_API}{path}"
+    headers = {"Authorization": f"Bearer {token}"}
+
+    try:
+        async with httpx.AsyncClient(timeout=config.timeout_seconds) as client:
+            resp = await client.get(url, headers=headers)
+        resp.raise_for_status()
+        return resp.content
+    except httpx.HTTPStatusError as exc:
+        raise ExternalServiceError(
+            f"飞书文件下载失败 [{path}]：{exc.response.status_code} {_response_error_detail(exc.response)}",
+            provider="feishu",
+        ) from exc
+    except httpx.HTTPError as exc:
+        raise ExternalServiceError(f"飞书文件下载网络错误 [{path}]：{exc}", provider="feishu") from exc
+
+
+async def _get_docx_content_async(doc_token: str, config: FeishuRetrievalConfig) -> str:
+    data = await _feishu_get_async(f"/open-apis/docx/v1/documents/{doc_token}/blocks", config=config)
+    items = data.get("data", {}).get("items", [])
+    return _parse_docx_content(items)
+
+
+async def _get_file_content_async(file_token: str, config: FeishuRetrievalConfig) -> str:
+    content = await _feishu_download_async(f"/open-apis/drive/v1/files/{file_token}/download", config)
+    for encoding in ("utf-8-sig", "utf-8", "gb18030"):
+        try:
+            return content.decode(encoding).strip()
+        except UnicodeDecodeError:
+            continue
+    raise ExternalServiceError("飞书文件不是可解析的文本内容", provider="feishu")
+
+
+async def _get_wiki_node_info_async(node_token: str, config: FeishuRetrievalConfig) -> dict:
+    data = await _feishu_get_async("/open-apis/wiki/v2/spaces/get_node", params={"token": node_token}, config=config)
+    node = data.get("data", {}).get("node", {})
+    return {
+        "space_id": node.get("space_id", ""),
+        "obj_token": node.get("obj_token", ""),
+        "obj_type": node.get("obj_type", ""),
+        "title": node.get("title", ""),
+        "node_token": node.get("node_token", ""),
+    }
+
+
+async def _get_wiki_children_async(
+    space_id: str,
+    node_token: str,
+    config: FeishuRetrievalConfig,
+    page_size: int = 50,
+) -> list[dict]:
+    if not space_id:
+        raise ExternalServiceError("飞书知识库缺少 SpaceID，无法读取子节点", provider="feishu")
+
+    all_children = []
+    page_token = ""
+    while True:
+        params: dict[str, Any] = {"page_size": page_size, "parent_node_token": node_token}
+        if page_token:
+            params["page_token"] = page_token
+
+        data = await _feishu_get_async(f"/open-apis/wiki/v2/spaces/{space_id}/nodes", params=params, config=config)
+        children = data.get("data", {}).get("items", [])
+        all_children.extend(children)
+
+        page_token = data.get("data", {}).get("page_token", "")
+        if not page_token:
+            break
+
+    return all_children

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

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from inspect import signature
+
+from sqlalchemy.orm import Session
+
+from app.models.chat import ChatMessage
+from app.models.user import User
+from app.services.feishu_async_service import AsyncFeishuKnowledgeService
+from app.services.knowledge_service import KnowledgeAccessService
+from app.services.rag_service import PromptService, RagResult
+
+
+class AsyncRagService:
+    @staticmethod
+    async def build_result(
+        db: Session,
+        user: User,
+        question: str,
+        history: list[ChatMessage] | None = None,
+        session_summary: str | None = None,
+    ) -> RagResult:
+        scopes = KnowledgeAccessService.get_allowed_knowledge(db, user)
+        chunks = await AsyncFeishuKnowledgeService.retrieve(question, scopes, db)
+        prompt = _build_prompt(db, question, chunks, history, session_summary)
+        return RagResult(question=question, knowledge_scopes=scopes, chunks=chunks, prompt=prompt)
+
+
+def _build_prompt(
+    db: Session,
+    question: str,
+    chunks,
+    history: list[ChatMessage] | None,
+    session_summary: str | None,
+) -> str:
+    parameters = signature(PromptService.build_prompt).parameters
+    if "history" in parameters:
+        return PromptService.build_prompt(db, question, chunks, history, session_summary)
+    return PromptService.build_prompt(db, question, chunks)

ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md

@@ -303,3 +303,38 @@
 2. 不改变搜索 payload、结果解析、知识片段打分规则。
 3. 不改变用户端 SSE 协议。
 4. 本阶段仍不引入 Redis，全局缓存和全局队列放到阶段 4。
+
+## 阶段 3B 实施记录
+
+2026-07-08 完成异步飞书读取和异步 RAG 构建第一版。
+
+已完成：
+
+1. 新增 `AsyncFeishuKnowledgeService`：
+   - 飞书搜索接口改为异步 HTTP。
+   - 飞书 tenant token 获取改为异步 HTTP。
+   - Wiki node info、children、Docx blocks、Drive file download 改为异步 HTTP。
+   - 复用同步服务已有的缓存、搜索 payload、搜索结果解析和打分逻辑。
+2. 新增 `AsyncRagService`：
+   - 权限范围和 Prompt 读取仍使用当前同步 DB session。
+   - 飞书外部读取切到 `AsyncFeishuKnowledgeService`。
+   - 兼容当前历史对话/摘要 prompt 签名，旧签名自动回退。
+3. `ChatStreamService.stream_answer_async` 已切到 `AsyncRagService.build_result`。
+4. 同步 `ChatStreamService.stream_answer` 保留不变，继续作为回退路径。
+5. 飞书直读多知识库时增加并发上限，当前限制为最多 3 个知识库同时读取，降低触发飞书频控的概率。
+6. 飞书部分知识库读取失败但已有可用片段时，降级使用可用片段继续回答；所有知识库均失败时才返回错误。
+
+验证记录：
+
+1. 本机 `compileall` 检查通过。
+2. 容器内 `compileall` 检查通过。
+3. 后端健康检查通过。
+4. 第一次真实问答验证触发飞书 `frequency limit`，据此补充了飞书直读并发上限和部分失败降级策略。
+5. 第二次真实 `/chat/completions` 验证通过：首个事件为 `generating`，随后返回 `content`，最后返回 `[DONE]`。
+6. 真实会话历史验证通过：流式完成后用户消息和助手消息均已落库。
+
+阶段限制：
+
+1. DB session 仍是同步的，权限范围、Prompt 读取、消息落库还没有切到 async DB。
+2. 当前缓存仍是进程内缓存，多 worker 下不会共享缓存。
+3. 飞书直读并发上限是代码内默认值，后续可在系统配置中开放。