From a04cfcabce2ed865d24d4749655235dc37719b96 Mon Sep 17 00:00:00 2001 From: Tian jianyong <11429339@qq.com> Date: Wed, 24 Dec 2025 18:41:42 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E8=B0=83=E8=AF=95=E8=AF=AD?= =?UTF-8?q?=E5=8F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../controller/ragflow_controller.py | 60 +++++++------------ ruoyi-fastapi-backend/test/test_sse.py | 17 +++--- ruoyi-fastapi-backend/utils/match_service.py | 6 ++ 3 files changed, 38 insertions(+), 45 deletions(-) diff --git a/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py b/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py index f0e5ff0..7e59392 100644 --- a/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py +++ b/ruoyi-fastapi-backend/module_admin/controller/ragflow_controller.py @@ -218,8 +218,21 @@ async def converse_with_chat_assistant( # 流式响应 if converse_params.stream: + # 创建缓存存储回调函数 + 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, converse_params.question) + return StreamingResponse( - stream_ragflow_response(result, converse_params.chat_id, start_time), + 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', @@ -229,47 +242,12 @@ async def converse_with_chat_assistant( } ) - # 非流式响应 - response = parse_result(result) - - # 记录原生RAG响应日志 - if isinstance(result, dict) and result.get('code') == 0: - answer_data = result.get('data', {}) - answer_text = answer_data.get('answer') if isinstance(answer_data, dict) else str(answer_data) - logger.info(f'[RAG_SOURCE] 原生RAG非流式响应完成 | chat_id={converse_params.chat_id} | answer_length={len(answer_text) if answer_text else 0}') - - # 设置语义缓存（存储问答对） - if not converse_params.stream and redis and isinstance(result, dict) and result.get('code') == 0: - try: - answer_data = result.get('data', {}) - answer_text = answer_data.get('answer') if isinstance(answer_data, dict) else str(answer_data) - if answer_text and len(answer_text.strip()) >= 10: - # 异步存储，不阻塞响应返回 - asyncio.create_task(_async_store_qa( - converse_params.chat_id, - converse_params.question, - answer_text, - redis - )) - except Exception as e: - logger.warning(f"语义缓存存储失败: {e}") - - # 设置原有精确缓存 - if redis and cache_key and isinstance(result, dict) and result.get('code') == 0: - try: - await redis.set(cache_key, json.dumps(result.get('data', {}), ensure_ascii=False), ex=60) - except Exception as e: - logger.warning(f"缓存设置失败: {e}") - - logger.info(f'[RAG_PROFILE] Total Sync Duration ({converse_params.chat_id}): {time.perf_counter() - start_time:.3f}s') - return response - except Exception as e: logger.exception(f'[RAG_PROFILE] ragflow对话异常: {e}') return ResponseUtil.error(msg=f'对话服务异常: {str(e)}') -def stream_ragflow_response(result: Generator, chat_id: str, start_time: float) -> Generator[str, None, None]: +def stream_ragflow_response(result: Generator, chat_id: str, start_time: float, cache_store_func=None) -> Generator[str, None, None]: """ 流式处理RAGFlow响应 - 同步版本，修复首token延迟 """ @@ -361,6 +339,14 @@ def stream_ragflow_response(result: Generator, chat_id: str, start_time: float) finally: total_time = time.perf_counter() - start_time logger.info(f'[RAG_SERVER {time.time():.3f}] ⏱️ Total Stream Duration ({chat_id}): {total_time:.3f}s') + + # 流结束后存储缓存 + if cache_store_func and last_answer and len(last_answer.strip()) >= 10: + try: + cache_store_func(last_answer) + logger.info(f'[RAG_CACHE] 流式响应缓存已存储 | chat_id={chat_id} | answer_length={len(last_answer)}') + except Exception as cache_err: + logger.warning(f'[RAG_CACHE] 流式响应缓存存储失败: {cache_err}') # 聊天助手列表 diff --git a/ruoyi-fastapi-backend/test/test_sse.py b/ruoyi-fastapi-backend/test/test_sse.py index a7b66ac..157ef0f 100644 --- a/ruoyi-fastapi-backend/test/test_sse.py +++ b/ruoyi-fastapi-backend/test/test_sse.py @@ -485,14 +485,15 @@ if __name__ == "__main__": ) else: # 性能测试模式 - # 定义三个测试问题 - # 问题1和2完全相同，用于测试精确缓存 - # 问题3与问题1意思相同但表述略不同，用于测试语义缓存 - questions = [ - "公司有什么产品？", # 问题1 - "公司有什么产品？", # 问题2（完全相同，测试精确缓存） - "公司有哪些产品？", # 问题3（表述略不同，测试语义缓存） - ] + if question and question != DEFAULT_QUESTION: + questions = [question, question, question] + print(f"📝 使用命令行参数的问题进行性能测试: {question}") + else: + questions = [ + "公司有什么产品？", # 问题1 + "公司有什么产品？", # 问题2（完全相同，测试精确缓存） + "公司有哪些产品？", # 问题3（表述略不同，测试语义缓存） + ] asyncio.run( run_performance_test( diff --git a/ruoyi-fastapi-backend/utils/match_service.py b/ruoyi-fastapi-backend/utils/match_service.py index 4dbb52c..5ba2f72 100644 --- a/ruoyi-fastapi-backend/utils/match_service.py +++ b/ruoyi-fastapi-backend/utils/match_service.py @@ -169,6 +169,9 @@ class MatchService: kw1 = set(self.extract_keywords(q1)) kw2 = set(self.extract_keywords(q2)) + logger.info(f'[MatchDebug] q1={q1} | n1={n1} | kw1={kw1}') + logger.info(f'[MatchDebug] q2={q2} | n2={n2} | kw2={kw2}') + if not kw1 or not kw2: return 0.0 @@ -192,11 +195,14 @@ class MatchService: for candidate_text, candidate_data in candidates: similarity = self.calculate_similarity(query, candidate_text) + logger.info(f'[MatchDebug] compare | query="{query}" | candidate="{candidate_text}" | sim={similarity:.3f}') if similarity > best_similarity: best_similarity = similarity best_match = candidate_data + logger.info(f'[MatchDebug] best | query="{query}" | best_sim={best_similarity:.3f} | threshold={threshold}') + if best_similarity >= threshold: return best_match, best_similarity