diff --git a/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py b/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py index a9d7b98..52ba8f2 100644 --- a/ai_knowledge_base_v2/apps/backend/app/services/chat_stream_service.py +++ b/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: diff --git a/ai_knowledge_base_v2/apps/backend/app/services/feishu_async_service.py b/ai_knowledge_base_v2/apps/backend/app/services/feishu_async_service.py new file mode 100644 index 0000000..ca4f2c3 --- /dev/null +++ b/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 diff --git a/ai_knowledge_base_v2/apps/backend/app/services/rag_async_service.py b/ai_knowledge_base_v2/apps/backend/app/services/rag_async_service.py new file mode 100644 index 0000000..2f50720 --- /dev/null +++ b/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) diff --git a/ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md b/ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md index dcb829d..1aec2ef 100644 --- a/ai_knowledge_base_v2/development_records/2026-07-08-concurrency-optimization-plan.md +++ b/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. 飞书直读并发上限是代码内默认值,后续可在系统配置中开放。