update

ruoyi-fastapi-backend/module_admin/service/search_service.py

@@ -57,6 +57,28 @@ class SearchService:
         '制造业', '制造业企业', '高新技术企业', '博士后科研工作站'
     )
 
+    # 问候/寒暄/致谢等“闲聊”短句不应触发公网搜索（防止意图模型误判）
+    _SMALLTALK_PATTERNS = (
+        r'^(hi|hello|hey|yo|sup)$',
+        r'^hello(?:there)?$',
+        r'^(goodmorning|goodafternoon|goodevening)$',
+        r'^(bye|goodbye|seeyou)$',
+        r'^(thankyou|thanks|thx)$',
+        r'^(你好|您好|嗨|哈喽|哈罗|早上好|上午好|下午好|晚上好|在吗|在不在|谢谢|多谢|谢了|再见|拜拜)$',
+        r'^(你是谁|你叫什么|你叫啥|介绍一下你自己)$',
+    )
+
+    @classmethod
+    def _is_smalltalk(cls, question: str) -> bool:
+        if not question:
+            return False
+        normalized = question.strip().lower()
+        # 仅保留中英文与数字，避免标点/空格影响匹配
+        compact = re.sub(r'[^0-9a-z\u4e00-\u9fff]+', '', normalized)
+        if not compact:
+            return False
+        return any(re.fullmatch(p, compact) for p in cls._SMALLTALK_PATTERNS)
+
     @classmethod
     def should_handle(cls, question: Optional[str]) -> bool:
         """判断是否应该使用搜索服务处理该问题"""
@@ -304,6 +326,10 @@ class SearchService:
         Classify user intent: 'SEARCH' (Public Internet) vs 'RAG' (Internal KB)
         Uses glm-4-flash for <300ms latency.
         """
+        # 先用本地规则兜底，避免问候类输入被误判为公网搜索
+        if cls._is_smalltalk(question):
+            return 'RAG'
+
         if not SearchConfig.ZHIPUAI_API_KEY:
             # Fallback to simple keyword check if no key
             return 'SEARCH' if cls.should_handle(question) else 'RAG'

ruoyi-fastapi-backend/test/test_fix_verification.py

@@ -1,8 +1,12 @@
+import os
 
+import pytest
 import asyncio
 import sys
-import os
-import json
+
+
+if os.getenv("RUN_INTEGRATION_TESTS") != "1":
+    pytest.skip("integration test: requires external RAGFlow service", allow_module_level=True)
 
 # Add project path
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)) + "/../")
@@ -40,7 +44,8 @@ async def test_fix_logic():
         
         async for chunk in result_stream:
             payload = chunk.get('data') if isinstance(chunk, dict) else chunk
-            if not payload: continue
+            if not payload:
+                continue
             
             body = payload if isinstance(payload, dict) else {'data': payload}
             

ruoyi-fastapi-backend/test/test_intent_guard_unit.py

@@ -0,0 +1,24 @@
+import pytest
+
+from config.env import SearchConfig
+from module_admin.service.search_service import SearchService
+
+
+@pytest.mark.asyncio
+async def test_classify_intent_smalltalk_forces_rag(monkeypatch):
+    # 防止触发外部请求：即便配置里有 key，也应被 smalltalk guard 拦住
+    monkeypatch.setattr(SearchConfig, "ZHIPUAI_API_KEY", "dummy")
+
+    assert await SearchService.classify_intent("hello") == "RAG"
+    assert await SearchService.classify_intent("你好") == "RAG"
+    assert await SearchService.classify_intent("谢谢") == "RAG"
+    assert await SearchService.classify_intent("再见") == "RAG"
+
+
+@pytest.mark.asyncio
+async def test_classify_intent_fallback_keyword_router(monkeypatch):
+    # 关闭意图模型，走 should_handle 分支
+    monkeypatch.setattr(SearchConfig, "ZHIPUAI_API_KEY", "")
+
+    assert await SearchService.classify_intent("今天上海的天气怎么样") == "SEARCH"
+    assert await SearchService.classify_intent("康达什么时候成立的") == "RAG"

ruoyi-fastapi-backend/test/test_intent_router.py

@@ -1,13 +1,18 @@
+import os
 
+import pytest
 import asyncio
 import sys
-import os
 
 # Add project path
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
 from module_admin.service.search_service import SearchService
 
+
+if os.getenv("RUN_INTEGRATION_TESTS") != "1":
+    pytest.skip("integration-style script: may call external intent model", allow_module_level=True)
+
 async def test_intent_router():
     print("=" * 60)
     print("Intent Router Test")

ruoyi-fastapi-backend/test/test_maxkb_performance.py

@@ -2,7 +2,14 @@
 MaxKB RAG 性能测试 - 测试首个token接收时间
 """
 import asyncio
+import os
 import aiohttp
+import pytest
+
+
+if os.getenv("RUN_INTEGRATION_TESTS") != "1":
+    pytest.skip("integration test: requires external KB/LLM services", allow_module_level=True)
+
 import time
 import sys
 import os
@@ -61,8 +68,8 @@ async def test_deepseek_first_token(question: str = TEST_QUESTION):
                 print(f"    错误信息: {error_text}")
                 return
             
-            print(f"    ✓ 连接建立，开始接收数据...")
-            print(f"\n[2/2] 接收流式数据...")
+            print("    ✓ 连接建立，开始接收数据...")
+            print("\n[2/2] 接收流式数据...")
             print("-" * 70)
             
             async for chunk in response.content:
@@ -189,8 +196,8 @@ async def test_first_token_time(question: str = TEST_QUESTION):
                 print(f"    错误信息: {error_text}")
                 return
             
-            print(f"    ✓ 连接建立，开始接收数据...")
-            print(f"\n[3/3] 接收流式数据...")
+            print("    ✓ 连接建立，开始接收数据...")
+            print("\n[3/3] 接收流式数据...")
             print("-" * 70)
             
             async for chunk in response.content:
@@ -327,7 +334,7 @@ async def run_multiple_tests(n: int = 3, question: str = None):
                                     first_token_time = elapsed
                                     first_token_received = True
                                 token_count += 1
-                        except:
+                        except Exception:
                             pass
         
         total_time = time.time() - total_start_time
@@ -347,14 +354,14 @@ async def run_multiple_tests(n: int = 3, question: str = None):
         avg_first_token = sum(first_token_times) / len(first_token_times)
         min_first_token = min(first_token_times)
         max_first_token = max(first_token_times)
-        print(f"  首个Token时间:")
+        print("  首个Token时间:")
         print(f"    - 平均值: {avg_first_token:.4f} 秒")
         print(f"    - 最小值: {min_first_token:.4f} 秒")
         print(f"    - 最大值: {max_first_token:.4f} 秒")
     
     if total_times:
         avg_total = sum(total_times) / len(total_times)
-        print(f"  总响应时间:")
+        print("  总响应时间:")
         print(f"    - 平均值: {avg_total:.4f} 秒")
     print('='*70)
 

ruoyi-fastapi-backend/test/test_rag_internal.py

@@ -1,11 +1,16 @@
+import os
 
+import pytest
 import asyncio
 import sys
-import os
 
 # Add project path
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)) + "/../")
 
+
+if os.getenv("RUN_INTEGRATION_TESTS") != "1":
+    pytest.skip("integration test: requires external RAGFlow service", allow_module_level=True)
+
 from module_admin.service.ragflow_service import RAGFlowService
 from module_admin.entity.vo.ragflow_vo import ConverseWithChatAssistantModel
 

ruoyi-fastapi-backend/test/test_search_llm_standalone.py

@@ -3,9 +3,16 @@
 
 不需要启动完整的FastAPI服务，直接测试搜索和LLM集成功能
 """
+import os
+import json
+
+import pytest
 import asyncio
 import sys
-import os
+
+
+if os.getenv("RUN_INTEGRATION_TESTS") != "1":
+    pytest.skip("integration test: requires external search/LLM services", allow_module_level=True)
 
 # 添加项目路径
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
@@ -23,7 +30,7 @@ async def test_search_with_llm():
     print("=" * 70)
     
     # 导入服务
-    from module_admin.service.search_service import SearchService, SearchServiceError
+    from module_admin.service.search_service import SearchService
     
     # 测试问题
     test_question = "今天上海的天气怎么样"
@@ -64,7 +71,8 @@ async def test_search_with_llm():
              search_context = ""
              async for line in response.body_iterator:
                  line = line.strip()
-                 if not line: continue
+                 if not line:
+                     continue
                  if line.startswith('data:'):
                      data_str = line[5:].strip()
                      try:
@@ -73,7 +81,7 @@ async def test_search_with_llm():
                          if content:
                              print(content, end='', flush=True)
                              search_context += content
-                     except:
+                     except Exception:
                          pass
              print("\n\n✓ 搜索流式响应完成!")
              

ruoyi-fastapi-backend/test/test_search_simple.py

@@ -1,12 +1,17 @@
 """
 搜索服务测试 - 直接调用 SearchService
 """
+import pytest
 import asyncio
 import sys
 import os
 import json
 import time
 
+
+if os.getenv("RUN_INTEGRATION_TESTS") != "1":
+    pytest.skip("integration test: requires external search/LLM services", allow_module_level=True)
+
 # 添加项目路径
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
@@ -78,7 +83,7 @@ async def test_search_service():
                                 # print(f"Content: {content}")
                                 pass 
                                 
-                        except:
+                        except Exception:
                             pass
                             
             print("\n✓ 流式接收完成!")

ruoyi-fastapi-backend/test/test_semantic_cache.py

@@ -12,20 +12,23 @@ python test_semantic_cache.py
 作者：AI Assistant
 """
 
+import os
+
+import pytest
+
 import asyncio
 import sys
-import os
 import time
 
+
+if os.getenv("RUN_INTEGRATION_TESTS") != "1":
+    pytest.skip("integration test: requires Redis (and optional RAGFlow)", allow_module_level=True)
+
 # 添加项目根目录到Python路径
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 from utils.semantic_cache_service import (
-    SemanticCacheService, 
-    get_semantic_cache_service,
-    lookup_question,
-    store_qa_pair,
-    clear_cache
+    SemanticCacheService,
 )
 from config.get_redis import RedisUtil
 
@@ -122,7 +125,7 @@ async def test_semantic_matching():
         else:
             print(f"  [{desc}]")
             print(f"  Q: {question[:30]}...")
-            print(f"     未命中")
+            print("     未命中")
 
 
 async def test_performance():
@@ -199,9 +202,9 @@ async def test_cache_statistics():
     
     # 获取统计
     stats = await cache.get_stats("test_chat_stats", redis)
-    print(f"\n缓存统计信息:")
+    print("\n缓存统计信息:")
     print(f"  总条目数: {stats.get('total_entries', 0)}")
-    print(f"  配置信息:")
+    print("  配置信息:")
     print(f"    - 最大缓存大小: {stats['config']['max_cache_size']}")
     print(f"    - 相似度阈值: {stats['config']['similarity_threshold']}")
     print(f"    - 缓存过期时间: {stats['config']['cache_ttl_hours']}小时")
@@ -247,18 +250,18 @@ async def test_integration_with_ragflow():
         if cached:
             answer, similarity = cached
             print(f"  ✓ 缓存命中! (相似度={similarity:.2f})")
-            print(f"  → 直接返回缓存答案")
+            print("  → 直接返回缓存答案")
         else:
-            print(f"  ✗ 缓存未命中")
-            print(f"  → 调用RAG服务...")
+            print("  ✗ 缓存未命中")
+            print("  → 调用RAG服务...")
             
             # 2. 调用RAG
             answer = await rag_service.query(question)
-            print(f"  → RAG返回答案")
+            print("  → RAG返回答案")
             
             # 3. 存储到缓存
             await cache.store("test_integration", question, answer, redis)
-            print(f"  → 已存入缓存")
+            print("  → 已存入缓存")
         
         print(f"  答案预览: {answer[:40]}...")
 

ruoyi-fastapi-backend/test/test_sse.py

@@ -2,12 +2,16 @@ import argparse  # 解析命令行参数
 import asyncio  # 异步事件循环
 import json  # 处理 JSON 文本
 import os  # 读取环境变量
-import sys  # 系统参数
 import time  # 时间统计
 import uuid  # 生成随机 chat/session id
 
+import pytest
+
+
+if os.getenv("RUN_INTEGRATION_TESTS") != "1":
+    pytest.skip("integration test: requires running backend + credentials", allow_module_level=True)
+
 import aiohttp  # 异步 HTTP 客户端
-import requests  # 同步 HTTP 客户端
 
 # 下面的默认参数可以通过环境变量覆盖，避免把机密写死在代码里。
 API_URL = os.environ.get(
@@ -264,7 +268,7 @@ async def stream_chat_async(
             ttft = first_token_time - connection_time  # Time To First Token
             print(f"⏱️  首Token耗时(不含连接): {ttft:.0f}ms")
     else:
-        print(f"🚀 首Token延迟: 未记录")
+        print("🚀 首Token延迟: 未记录")
     print(f"📝 回答长度: {len(full_answer)} 字符")
     print(f"{'='*60}")
     
@@ -296,7 +300,7 @@ async def run_performance_test(
     
     print("\n" + "🚀"*30)
     print("🎯 性能测试开始")
-    print(f"📊 测试配置:")
+    print("📊 测试配置:")
     print(f"   - 请求次数: {num_requests}")
     print(f"   - 请求间隔: {delay_between_requests}s")
     print("📝 测试问题:")
@@ -383,7 +387,7 @@ async def run_performance_test(
             print(f"#{r['request_number']:<6} {question_short:<30} {r['elapsed_ms']:.0f}ms{'':<5} {first_token:<10} {speed}")
     
     print("-" * 80)
-    print(f"\n📊 统计信息:")
+    print("\n📊 统计信息:")
     print(f"   平均耗时: {avg_time:.0f}ms ({avg_time/1000:.2f}s)")
     print(f"   最快耗时: {min_time:.0f}ms")
     print(f"   最慢耗时: {max_time:.0f}ms")
@@ -393,26 +397,26 @@ async def run_performance_test(
         first_time = valid_results[0]["elapsed_ms"]
         last_time = valid_results[-1]["elapsed_ms"]
         
-        print(f"\n💡 缓存效果分析:")
+        print("\n💡 缓存效果分析:")
         if last_time < first_time * 0.7:
             print(f"   ✅ 缓存可能生效! 后端请求({first_time:.0f}ms)比首次请求快 {((first_time - last_time) / first_time * 100):.1f}%")
         elif last_time > first_time * 1.3:
-            print(f"   ⚠️ 缓存可能未生效，后续请求比首次请求慢")
+            print("   ⚠️ 缓存可能未生效，后续请求比首次请求慢")
         else:
-            print(f"   🤔 响应时间相近，缓存效果不确定")
+            print("   🤔 响应时间相近，缓存效果不确定")
     
     # 首Token时间分析
     first_tokens = [r["first_token_ms"] for r in valid_results if r["first_token_ms"] is not None]
     if len(first_tokens) >= 2:
         first_ttft = first_tokens[0]
         last_ttft = first_tokens[-1]
-        print(f"\n🚀 首Token时间分析:")
+        print("\n🚀 首Token时间分析:")
         if last_ttft < first_ttft * 0.5:
-            print(f"   ✅ 后续请求首Token更快，可能命中缓存")
+            print("   ✅ 后续请求首Token更快，可能命中缓存")
         elif last_ttft > first_ttft * 1.5:
-            print(f"   ⚠️ 后续请求首Token更慢，可能是新问题")
+            print("   ⚠️ 后续请求首Token更慢，可能是新问题")
         else:
-            print(f"   🤔 首Token时间相近")
+            print("   🤔 首Token时间相近")
     
     print("📊"*30 + "\n")