diff --git a/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py b/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py index d697cff..b04e5b7 100644 --- a/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py +++ b/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py @@ -140,28 +140,22 @@ async def converse_with_chat_assistant( cache_similarity = static_sim cache_source = 'static_qa' logger.info(f'[RAG_SOURCE] 命中静态FAQ | chat_id={converse_params.chat_id} | question={converse_params.question} | similarity={cache_similarity:.2f} | answer_length={len(cached_answer)}') - - # 非流式：直接返回静态问答答案 - if not converse_params.stream: - return ResponseUtil.success(data={'answer': cached_answer, 'from_cache': True, 'similarity': cache_similarity, 'source': 'static_qa'}) - - # 流式：使用静态问答答案进行流式响应 - if converse_params.stream: - logger.info(f'[StaticQA] 流式响应使用静态问答答案，chat_id={converse_params.chat_id}') - return StreamingResponse( - stream_cached_response(cached_answer, converse_params.chat_id, start_time, cache_source='static_qa'), - media_type='text/event-stream', - headers={ - 'Cache-Control': 'no-cache', - 'Connection': 'keep-alive', - 'X-Accel-Buffering': 'no', - 'Transfer-Encoding': 'chunked' - } - ) + logger.info(f'[StaticQA] 流式响应使用静态问答答案，chat_id={converse_params.chat_id}') + return StreamingResponse( + stream_cached_response(cached_answer, converse_params.chat_id, start_time, cache_source='static_qa'), + media_type='text/event-stream', + headers={ + 'Cache-Control': 'no-cache', + 'Connection': 'keep-alive', + 'X-Accel-Buffering': 'no', + 'Transfer-Encoding': 'chunked' + } + ) except Exception as e: logger.warning(f'[StaticQA] 静态问答匹配失败: {e}') # ========== 2. RAG历史缓存查找 ========== + logger.info(f'[SemanticCache] 准备执行RAG历史缓存查找 | redis={redis is not None} | chat_id={converse_params.chat_id}') if redis: lookup_hash = _get_question_hash(converse_params.question) logger.info(f'[SemanticCache] 开始查找 | chat_id={converse_params.chat_id} | question={converse_params.question} | hash={lookup_hash} | threshold=0.60') @@ -175,24 +169,17 @@ async def converse_with_chat_assistant( cached_answer, cache_similarity = cache_result cache_source = 'rag_history' logger.info(f'[RAG_SOURCE] 命中RAG会话历史 | chat_id={converse_params.chat_id} | question={converse_params.question} | similarity={cache_similarity:.2f} | answer_length={len(cached_answer)}') - - # 非流式：直接返回缓存答案 - if not converse_params.stream: - return ResponseUtil.success(data={'answer': cached_answer, 'from_cache': True, 'similarity': cache_similarity, 'source': 'rag_history'}) - - # 流式：使用缓存答案进行流式响应 - if converse_params.stream: - logger.info(f'[SemanticCache] 流式响应使用RAG历史缓存答案，chat_id={converse_params.chat_id}') - return StreamingResponse( - stream_cached_response(cached_answer, converse_params.chat_id, start_time, cache_source='rag_history'), - media_type='text/event-stream', - headers={ - 'Cache-Control': 'no-cache', - 'Connection': 'keep-alive', - 'X-Accel-Buffering': 'no', - 'Transfer-Encoding': 'chunked' - } - ) + logger.info(f'[SemanticCache] 流式响应使用RAG历史缓存答案，chat_id={converse_params.chat_id}') + return StreamingResponse( + stream_cached_response(cached_answer, converse_params.chat_id, start_time, cache_source='rag_history'), + media_type='text/event-stream', + headers={ + 'Cache-Control': 'no-cache', + 'Connection': 'keep-alive', + 'X-Accel-Buffering': 'no', + 'Transfer-Encoding': 'chunked' + } + ) # 检查是否应该使用搜索服务 try: @@ -204,17 +191,6 @@ async def converse_with_chat_assistant( except Exception as e: logger.warning(f"意图分类失败，使用RAG服务: {e}") - # 缓存检查（原有的精确缓存）- 非流式 - if not converse_params.stream and redis: - cache_key = build_chat_cache_key(converse_params.chat_id, converse_params.question) - try: - cached = await redis.get(cache_key) - if cached: - logger.info('ragflow对话命中缓存: chat=%s', converse_params.chat_id) - return ResponseUtil.success(json.loads(cached)) - except Exception as e: - logger.warning(f"缓存获取失败: {e}") - # 直接使用同步RAGFlow服务 try: # 恢复原始问题（包含语言风格提示词） @@ -225,36 +201,32 @@ async def converse_with_chat_assistant( logger.info(f'[RAG_SOURCE] 调用原生RAG服务 | chat_id={converse_params.chat_id} | question={converse_params.question}') result = RAGFlowService.converse_with_chat_assistant_services(converse_params) - # 流式响应 - if converse_params.stream: - # 注意：存储时使用清理后的问题（与查找时保持一致） - cache_question = cleaned_question if style_removed else converse_params.question - store_hash = _get_question_hash(cache_question) - logger.info(f'[RAG_CACHE] 准备存储 | chat_id={converse_params.chat_id} | question={cache_question} | hash={store_hash}') - - # 创建缓存存储回调函数 - async def make_cache_store(chat_id: str, question: str): - async def store_answer(answer: str): - if redis and answer and len(answer.strip()) >= 10: - try: - await _async_store_qa(chat_id, question, answer, redis) - logger.info(f'[RAG_CACHE] 语义缓存存储成功 | chat_id={chat_id} | answer_length={len(answer)}') - except Exception as e: - logger.warning(f'语义缓存存储失败: {e}') - return store_answer - - cache_store = await make_cache_store(converse_params.chat_id, cache_question) - - return StreamingResponse( - stream_ragflow_response(result, converse_params.chat_id, start_time, cache_store_func=cache_store), - media_type='text/event-stream', - headers={ - 'Cache-Control': 'no-cache', - 'Connection': 'keep-alive', - 'X-Accel-Buffering': 'no', - 'Transfer-Encoding': 'chunked' - } - ) + cache_question = cleaned_question if style_removed else converse_params.question + store_hash = _get_question_hash(cache_question) + logger.info(f'[RAG_CACHE] 准备存储 | chat_id={converse_params.chat_id} | question={cache_question} | hash={store_hash}') + + async def make_cache_store(chat_id: str, question: str): + async def store_answer(answer: str): + if redis and answer and len(answer.strip()) >= 10: + try: + await _async_store_qa(chat_id, question, answer, redis) + logger.info(f'[RAG_CACHE] 语义缓存存储成功 | chat_id={chat_id} | answer_length={len(answer)}') + except Exception as e: + logger.warning(f'语义缓存存储失败: {e}') + return store_answer + + cache_store = await make_cache_store(converse_params.chat_id, cache_question) + + return StreamingResponse( + stream_ragflow_response(result, converse_params.chat_id, start_time, cache_store_func=cache_store), + media_type='text/event-stream', + headers={ + 'Cache-Control': 'no-cache', + 'Connection': 'keep-alive', + 'X-Accel-Buffering': 'no', + 'Transfer-Encoding': 'chunked' + } + ) except Exception as e: logger.exception(f'[RAG_PROFILE] ragflow对话异常: {e}') diff --git a/ruoyi-fastapi-backend/utils/semantic_cache_service.py b/ruoyi-fastapi-backend/utils/semantic_cache_service.py index bcb5a80..0f28e2c 100644 --- a/ruoyi-fastapi-backend/utils/semantic_cache_service.py +++ b/ruoyi-fastapi-backend/utils/semantic_cache_service.py @@ -419,10 +419,13 @@ def get_semantic_cache_service() -> SemanticCacheService: # 便捷函数 async def lookup_question(chat_id: str, question: str, redis_client=None) -> Optional[Tuple[str, float]]: """查找缓存问题""" - logger.info(f"[SemanticCache.lookup_question] 入参 | chat_id={chat_id} | question={question}") + logger.info(f"[SemanticCache.lookup_question] 🔍 函数被调用 | chat_id={chat_id} | question={question} | redis_client={redis_client is not None}") service = get_semantic_cache_service() + logger.info(f"[SemanticCache.lookup_question] 获取服务实例完成 | service={service is not None}") result = await service.lookup(chat_id, question, redis_client) logger.info(f"[SemanticCache.lookup_question] 返回 | result={result is not None}") + if result: + logger.info(f"[SemanticCache.lookup_question] 命中结果 | answer_length={len(result[0])} | similarity={result[1]:.2f}") return result