feat(rag): 实现路由层 RAG 前置管线与实时思考事件推送

- vector_store.py: 新增 compute_route_score 快速相关度评分方法
- chat.py: 实现路由判断逻辑（score > 0.5 走 RAG 管线）
- 通过 asyncio.Queue 实时推送 RAG 思考事件到前端
- RAG 检索结果注入 Agent 的 rag_context 参数

Author: RMA-MUN

backend/app/rag/vector_store.py

@@ -121,6 +121,26 @@ def _get_embed_model():
         """获取嵌入模型（延迟加载包装器，模型在首次调用时解析）"""
         return _LazyEmbedding()
 
+    async def compute_route_score(self, query: str, user_id: str) -> float:
+        """快速计算查询与用户知识库的相关度（<10ms）
+
+        用 ChromaDB 做 Top-1 检索，返回 L2 距离转换后的相似度分数。
+        分数越高表示与用户知识库越相关，用于路由层判断是否需要 RAG 前置管线。
+        """
+        try:
+            results = await asyncio.to_thread(
+                self.vectors_store.similarity_search_with_score,
+                query,
+                k=1,
+                filter={"user_id": user_id}
+            )
+            if not results:
+                return 0.0
+            distance = results[0][1]
+            return 1 / (1 + distance)
+        except Exception:
+            return 0.0
+
     async def get_bm25_retriever(self, user_id: str = None):
         return await self.hybrid_retriever.get_bm25_retriever(user_id)
 

backend/app/router/chat.py

@@ -1,3 +1,5 @@
+import asyncio
+import json
 import uuid
 
 from fastapi import Depends
@@ -23,8 +25,109 @@ async def query_stream(
     """查询Agent流式响应"""
     session_id = request.session_id or str(uuid.uuid4())
 
+    from app.core.logger_handler import logger
+    from app.rag.vector_store import VectorStoreService
+
+    vector_store = VectorStoreService()
+
+    # ---- 路由判断（快速，~50ms）----
+    score = await vector_store.compute_route_score(
+        request.query, user_id
+    )
+
+    # 查询 Top-1 文档详情，用于日志输出
+    top1_docs = await asyncio.to_thread(
+        vector_store.vectors_store.similarity_search_with_score,
+        request.query, k=1, filter={"user_id": user_id}
+    )
+    if top1_docs:
+        top1_doc, top1_distance = top1_docs[0]
+        source_type = "笔记库" if top1_doc.metadata.get("source_type") == "note" else "知识库"
+        source_name = top1_doc.metadata.get("title") or top1_doc.metadata.get("original_filename", "未知")
+        preview = top1_doc.page_content[:80].replace("\n", " ")
+        logger.info(
+            f"【路由决策】查询: 「{request.query}」 | "
+            f"score: {score:.4f} (距离: {top1_distance:.4f}) | "
+            f"Top-1来源: {source_type}《{source_name}》 | "
+            f"预览: {preview}... | "
+            f"决策: {'→ RAG 前置管线' if score > 0.5 else '→ 跳过 RAG'}"
+        )
+    else:
+        logger.info(
+            f"【路由决策】查询: 「{request.query}」 | "
+            f"score: {score:.4f} | "
+            f"Top-1: 无文档 | "
+            f"决策: → 跳过 RAG"
+        )
+
+    async def stream_with_rag_thinking():
+        """包装生成器：RAG 管线在内部实时推送思考事件，再转发 Agent 流式响应"""
+        rag_context = ""
+
+        if score > 0.5:
+            from app.rag.rag_service import RagService
+
+            # RAG 管线与 SSE 推送共用的队列
+            thinking_queue = asyncio.Queue()
+            rag_done = asyncio.Event()
+
+            async def thinking_callback(data: dict):
+                await thinking_queue.put(data)
+
+            async def run_rag_pipeline():
+                """在后台执行 RAG 管线，thinking 事件通过队列实时推送"""
+                try:
+                    rag_service = RagService(user_id, thinking_callback=thinking_callback)
+                    documents = await rag_service.retrieve_document(request.query)
+
+                    def _format_doc(doc):
+                        if doc.metadata.get("source_type") == "note":
+                            title = doc.metadata.get("title", "无标题")
+                            return f"[来源：笔记《{title}》]\n{doc.page_content}"
+                        else:
+                            filename = doc.metadata.get("original_filename", "知识库文档")
+                            return f"[来源：知识库《{filename}》]\n{doc.page_content}"
+
+                    doc_contents = [_format_doc(doc) for doc in documents]
+                    reordered = await rag_service.reorder_documents(request.query, doc_contents)
+                    nonlocal rag_context
+                    rag_context = "\n\n".join(reordered[:3])
+                    logger.info(f"【RAG前置】检索到 {len(documents)} 个文档，重排序后取前 {min(3, len(reordered))} 个注入 Agent")
+                except Exception as e:
+                    logger.error(f"【RAG前置】管线执行失败: {e}", exc_info=True)
+                finally:
+                    rag_done.set()
+
+            # 启动 RAG 管线（后台任务）
+            rag_task = asyncio.create_task(run_rag_pipeline())
+
+            # 实时推送 RAG 思考事件：边跑边推，不等管线结束
+            while not rag_done.is_set() or not thinking_queue.empty():
+                try:
+                    event = thinking_queue.get_nowait()
+                    yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
+                except asyncio.QueueEmpty:
+                    # 队列暂时为空，等 RAG 管线产出新事件
+                    try:
+                        event = await asyncio.wait_for(thinking_queue.get(), timeout=0.1)
+                        yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
+                    except (asyncio.TimeoutError, asyncio.QueueEmpty):
+                        continue
+
+            # 确保 RAG 任务完成，再 drain 一次队列防止竞态丢失事件
+            await rag_task
+            while not thinking_queue.empty():
+                event = thinking_queue.get_nowait()
+                yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
+
+        # 转发 Agent 流式响应
+        async for chunk in get_agent_stream_response(
+            request.query, session_id, user_id, rag_context=rag_context
+        ):
+            yield chunk
+
     return StreamingResponse(
-        get_agent_stream_response(request.query, session_id, user_id),
+        stream_with_rag_thinking(),
         media_type="text/event-stream",
         headers={
             "Cache-Control": "no-cache",