增加语义缓存匹配测试
This commit is contained in:
parent
2c0ea3467b
commit
31c183c906
@ -3,6 +3,7 @@ asyncmy==0.2.10
|
||||
DateTime==5.5
|
||||
fastapi[all]==0.115.8
|
||||
httpx==0.27.2
|
||||
jieba==0.42.1
|
||||
loguru==0.7.3
|
||||
openpyxl==3.1.5
|
||||
pandas==2.2.3
|
||||
|
||||
43
ruoyi-fastapi-backend/test/test_similarity_debug.py
Normal file
43
ruoyi-fastapi-backend/test/test_similarity_debug.py
Normal file
@ -0,0 +1,43 @@
|
||||
"""
|
||||
测试语义缓存的相似度计算
|
||||
"""
|
||||
import sys
|
||||
import os
|
||||
|
||||
# 添加项目路径
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from utils.semantic_cache_service import SemanticCacheService
|
||||
|
||||
def test_similarity():
|
||||
"""测试相似度计算"""
|
||||
service = SemanticCacheService()
|
||||
|
||||
# 测试问题对
|
||||
test_cases = [
|
||||
("公司有什么产品？", "公司有什么产品？"),
|
||||
("公司有什么产品？", "公司有哪些产品？"),
|
||||
("公司有什么产品？", "你们公司有哪些产品？"),
|
||||
("公司有什么产品？", "今天天气怎么样？"),
|
||||
]
|
||||
|
||||
print("🔍 相似度计算测试")
|
||||
print("=" * 60)
|
||||
|
||||
for q1, q2 in test_cases:
|
||||
similarity = service._calculate_text_similarity(q1, q2)
|
||||
keywords1 = service._extract_keywords(q1)
|
||||
keywords2 = service._extract_keywords(q2)
|
||||
|
||||
threshold = service.SIMILARITY_THRESHOLD
|
||||
status = "✅ 命中" if similarity >= threshold else "❌ 未命中"
|
||||
|
||||
print(f"\n问题1: {q1}")
|
||||
print(f"问题2: {q2}")
|
||||
print(f"关键词1: {keywords1}")
|
||||
print(f"关键词2: {keywords2}")
|
||||
print(f"相似度: {similarity:.3f} (阈值: {threshold})")
|
||||
print(f"状态: {status}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_similarity()
|
||||
@ -49,6 +49,13 @@ class PerformanceTimer:
|
||||
self.start_time = time.perf_counter()
|
||||
return self
|
||||
|
||||
def restart(self):
|
||||
"""重新开始计时"""
|
||||
self.start_time = time.perf_counter()
|
||||
self.end_time = None
|
||||
self.elapsed_ms = 0
|
||||
return self
|
||||
|
||||
def stop(self):
|
||||
if self.start_time is None:
|
||||
raise ValueError("计时器未启动")
|
||||
@ -57,17 +64,23 @@ class PerformanceTimer:
|
||||
return self
|
||||
|
||||
def get_elapsed_ms(self) -> float:
|
||||
return self.elapsed_ms
|
||||
if self.start_time is not None and self.end_time is not None:
|
||||
return self.elapsed_ms
|
||||
elif self.start_time is not None:
|
||||
# 如果没有stop，返回当前经过的时间
|
||||
return (time.perf_counter() - self.start_time) * 1000
|
||||
return 0
|
||||
|
||||
def get_elapsed_seconds(self) -> float:
|
||||
return self.elapsed_ms / 1000
|
||||
return self.get_elapsed_ms() / 1000
|
||||
|
||||
def format(self) -> str:
|
||||
"""格式化输出时间"""
|
||||
if self.elapsed_ms < 1000:
|
||||
return f"{self.elapsed_ms:.2f}ms"
|
||||
elapsed = self.get_elapsed_ms()
|
||||
if elapsed < 1000:
|
||||
return f"{elapsed:.2f}ms"
|
||||
else:
|
||||
return f"{self.get_elapsed_seconds():.2f}s ({self.elapsed_ms:.0f}ms)"
|
||||
return f"{self.get_elapsed_seconds():.2f}s ({elapsed:.0f}ms)"
|
||||
|
||||
|
||||
async def login_get_token() -> str:
|
||||
@ -121,10 +134,10 @@ async def stream_chat_async(
|
||||
chat_id: str,
|
||||
session_id: str,
|
||||
request_number: int = 1
|
||||
) -> tuple[str, PerformanceTimer]:
|
||||
) -> dict:
|
||||
"""
|
||||
异步版（aiohttp），适合已有事件循环的应用，不阻塞主线程。
|
||||
返回完整回答和耗时
|
||||
返回完整回答和相关时间统计
|
||||
"""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"📤 请求 #{request_number}")
|
||||
@ -133,8 +146,14 @@ async def stream_chat_async(
|
||||
print(f"🔗 Session ID: {session_id}")
|
||||
print(f"{'='*60}")
|
||||
|
||||
# 创建计时器
|
||||
timer = PerformanceTimer().start()
|
||||
|
||||
token = await login_get_token()
|
||||
|
||||
# 重启计时器，只计算流式请求的时间（不包含登录时间）
|
||||
timer.restart()
|
||||
|
||||
payload = {
|
||||
"chat_id": chat_id,
|
||||
"question": question,
|
||||
@ -149,9 +168,9 @@ async def stream_chat_async(
|
||||
answer_parts = []
|
||||
event_type = "message"
|
||||
|
||||
# 性能计时
|
||||
timer = PerformanceTimer().start()
|
||||
# 性能计时 - timer 已在登录前创建并 restart
|
||||
first_token_time = None
|
||||
connection_time = None
|
||||
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(
|
||||
@ -160,16 +179,23 @@ async def stream_chat_async(
|
||||
print(
|
||||
f"# status={resp.status} content-type={resp.headers.get('content-type')}")
|
||||
resp.raise_for_status()
|
||||
|
||||
# 记录连接建立时间
|
||||
connection_time = timer.get_elapsed_ms()
|
||||
print(f"[debug] 连接建立时间: {connection_time:.0f}ms")
|
||||
|
||||
event_count = 0
|
||||
while True:
|
||||
raw_line = await resp.content.readline()
|
||||
if not raw_line:
|
||||
print(f"[debug] 收到 {event_count} 个事件，流结束")
|
||||
break
|
||||
line = raw_line.decode(errors="ignore").strip() # 逐行读取 SSE 数据
|
||||
if not line or line.startswith(":"):
|
||||
continue
|
||||
if line.startswith("event:"):
|
||||
event_type = line.split(":", 1)[1].strip() or "message"
|
||||
print(f"[debug] 事件类型: {event_type}")
|
||||
continue
|
||||
if not line.startswith("data:"):
|
||||
print(f"[skip] {line}")
|
||||
@ -177,16 +203,36 @@ async def stream_chat_async(
|
||||
data_str = line[len("data:"):].strip()
|
||||
if not data_str:
|
||||
continue
|
||||
event_count += 1
|
||||
try:
|
||||
payload_obj = json.loads(data_str)
|
||||
except json.JSONDecodeError:
|
||||
print(f"[{event_type}] {data_str}")
|
||||
event_type = "message"
|
||||
continue
|
||||
|
||||
# 打印所有收到的响应，便于调试
|
||||
print(f"[debug] 收到事件 #{event_count}: {list(payload_obj.keys())}")
|
||||
|
||||
# 记录首Token时间
|
||||
if first_token_time is None and "answer" in payload_obj:
|
||||
first_token_time = timer.get_elapsed_ms()
|
||||
# 记录首Token时间（遇到任何包含内容的字段时）
|
||||
if first_token_time is None:
|
||||
# 检测缓存命中情况
|
||||
if "from_cache" in payload_obj and payload_obj.get("from_cache") is True:
|
||||
# 缓存命中！首Token就是连接建立时间
|
||||
first_token_time = connection_time
|
||||
cache_total_time = payload_obj.get("total_time")
|
||||
print(f"[debug] 🔥 缓存命中! 连接时间: {connection_time:.0f}ms, 缓存总耗时: {cache_total_time}ms")
|
||||
elif "answer" in payload_obj:
|
||||
first_token_time = timer.get_elapsed_ms()
|
||||
print(f"[debug] 首Token时间(answer): {first_token_time:.0f}ms")
|
||||
elif "data" in payload_obj and payload_obj.get("data") is not True:
|
||||
# 缓存命中时可能直接返回完整答案
|
||||
first_token_time = timer.get_elapsed_ms()
|
||||
print(f"[debug] 首Token时间(data): {first_token_time:.0f}ms")
|
||||
elif payload_obj.get("status") == "completed" or event_type == "end":
|
||||
# 如果一开始就收到completed，可能是即时响应
|
||||
first_token_time = connection_time # 使用连接时间作为近似
|
||||
print(f"[debug] 首Token时间(即时响应): {first_token_time:.0f}ms")
|
||||
|
||||
if event_type == "end" or payload_obj.get("status") == "completed":
|
||||
print("\n[stream] completed")
|
||||
@ -199,47 +245,71 @@ async def stream_chat_async(
|
||||
answer_parts.append(piece)
|
||||
# 流式输出：每个片段都单独打印
|
||||
print(piece, end="", flush=True)
|
||||
time.sleep(0.05) # 每个片段间隔50ms，让流式效果更明显
|
||||
await asyncio.sleep(0.05) # 使用异步sleep而不是time.sleep
|
||||
else:
|
||||
print(f"[{event_type}] {payload_obj}")
|
||||
event_type = "message"
|
||||
|
||||
timer.stop()
|
||||
full_answer = "".join(answer_parts)
|
||||
|
||||
print(f"\n\n{'='*60}")
|
||||
print(f"✅ 请求 #{request_number} 完成")
|
||||
print(f"⏱️ 总耗时: {timer.format()}")
|
||||
if first_token_time:
|
||||
if connection_time is not None:
|
||||
print(f"🔗 连接建立: {connection_time:.0f}ms")
|
||||
if first_token_time is not None:
|
||||
print(f"🚀 首Token延迟: {first_token_time:.0f}ms")
|
||||
print(f"📝 回答长度: {len(''.join(answer_parts))} 字符")
|
||||
if connection_time is not None:
|
||||
ttft = first_token_time - connection_time # Time To First Token
|
||||
print(f"⏱️ 首Token耗时(不含连接): {ttft:.0f}ms")
|
||||
else:
|
||||
print(f"🚀 首Token延迟: 未记录")
|
||||
print(f"📝 回答长度: {len(full_answer)} 字符")
|
||||
print(f"{'='*60}")
|
||||
|
||||
full_answer = "".join(answer_parts)
|
||||
return full_answer, timer
|
||||
# 返回完整统计信息
|
||||
return {
|
||||
"answer": full_answer,
|
||||
"timer": timer,
|
||||
"first_token_ms": first_token_time,
|
||||
"connection_ms": connection_time
|
||||
}
|
||||
|
||||
|
||||
async def run_performance_test(
|
||||
question: str,
|
||||
questions: list[str],
|
||||
chat_id: str,
|
||||
session_id: str,
|
||||
num_requests: int = 3,
|
||||
delay_between_requests: float = 2.0
|
||||
):
|
||||
"""
|
||||
运行性能测试，多次请求以测试缓存效果
|
||||
|
||||
Args:
|
||||
questions: 问题列表，每个请求使用对应的问题
|
||||
chat_id: 聊天会话ID
|
||||
session_id: 会话ID
|
||||
delay_between_requests: 请求之间的间隔时间(秒)
|
||||
"""
|
||||
num_requests = len(questions)
|
||||
|
||||
print("\n" + "🚀"*30)
|
||||
print("🎯 性能测试开始")
|
||||
print(f"📊 测试配置:")
|
||||
print(f" - 请求次数: {num_requests}")
|
||||
print(f" - 请求间隔: {delay_between_requests}s")
|
||||
print(f" - 问题: {question}")
|
||||
print("📝 测试问题:")
|
||||
for i, q in enumerate(questions):
|
||||
print(f" {i+1}. {q}")
|
||||
print("🚀"*30)
|
||||
|
||||
results = []
|
||||
|
||||
for i in range(1, num_requests + 1):
|
||||
question = questions[i-1]
|
||||
try:
|
||||
answer, timer = await stream_chat_async(
|
||||
result = await stream_chat_async(
|
||||
question=question,
|
||||
chat_id=chat_id,
|
||||
session_id=session_id,
|
||||
@ -247,9 +317,13 @@ async def run_performance_test(
|
||||
)
|
||||
results.append({
|
||||
"request_number": i,
|
||||
"elapsed_ms": timer.get_elapsed_ms(),
|
||||
"elapsed_seconds": timer.get_elapsed_seconds(),
|
||||
"answer_length": len(answer)
|
||||
"question": question,
|
||||
"elapsed_ms": result["timer"].get_elapsed_ms(),
|
||||
"elapsed_seconds": result["timer"].get_elapsed_seconds(),
|
||||
"first_token_ms": result["first_token_ms"],
|
||||
"connection_ms": result["connection_ms"],
|
||||
"answer_length": len(result["answer"]),
|
||||
"from_cache": False # 后续可以根据响应判断
|
||||
})
|
||||
|
||||
# 如果不是最后一次请求，等待一段时间
|
||||
@ -261,8 +335,11 @@ async def run_performance_test(
|
||||
print(f"\n❌ 请求 #{i} 失败: {str(e)}")
|
||||
results.append({
|
||||
"request_number": i,
|
||||
"question": question,
|
||||
"elapsed_ms": -1,
|
||||
"elapsed_seconds": -1,
|
||||
"first_token_ms": None,
|
||||
"connection_ms": None,
|
||||
"answer_length": 0,
|
||||
"error": str(e)
|
||||
})
|
||||
@ -283,12 +360,12 @@ async def run_performance_test(
|
||||
min_time = min(elapsed_times)
|
||||
max_time = max(elapsed_times)
|
||||
|
||||
print(f"\n{'请求#':<8} {'耗时':<15} {'速度评估'}")
|
||||
print("-" * 50)
|
||||
print(f"\n{'请求#':<8} {'问题':<30} {'总耗时':<12} {'首Token':<10} {'速度评估'}")
|
||||
print("-" * 80)
|
||||
|
||||
for r in results:
|
||||
if r["elapsed_ms"] < 0:
|
||||
print(f"#{r['request_number']:<6} {'失败':<15} ❌")
|
||||
print(f"#{r['request_number']:<6} {r['question'][:28]:<30} {'失败':<12} {'-':<10} ❌")
|
||||
else:
|
||||
# 速度评估
|
||||
elapsed = r["elapsed_ms"]
|
||||
@ -301,9 +378,11 @@ async def run_performance_test(
|
||||
else:
|
||||
speed = "🐢 较慢"
|
||||
|
||||
print(f"#{r['request_number']:<6} {r['elapsed_ms']:.0f}ms".ljust(15) + speed)
|
||||
first_token = f"{r['first_token_ms']:.0f}ms" if r["first_token_ms"] else "-"
|
||||
question_short = r['question'][:28] + ".." if len(r['question']) > 28 else r['question']
|
||||
print(f"#{r['request_number']:<6} {question_short:<30} {r['elapsed_ms']:.0f}ms{'':<5} {first_token:<10} {speed}")
|
||||
|
||||
print("-" * 50)
|
||||
print("-" * 80)
|
||||
print(f"\n📊 统计信息:")
|
||||
print(f" 平均耗时: {avg_time:.0f}ms ({avg_time/1000:.2f}s)")
|
||||
print(f" 最快耗时: {min_time:.0f}ms")
|
||||
@ -322,6 +401,19 @@ async def run_performance_test(
|
||||
else:
|
||||
print(f" 🤔 响应时间相近，缓存效果不确定")
|
||||
|
||||
# 首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时间分析:")
|
||||
if last_ttft < first_ttft * 0.5:
|
||||
print(f" ✅ 后续请求首Token更快，可能命中缓存")
|
||||
elif last_ttft > first_ttft * 1.5:
|
||||
print(f" ⚠️ 后续请求首Token更慢，可能是新问题")
|
||||
else:
|
||||
print(f" 🤔 首Token时间相近")
|
||||
|
||||
print("📊"*30 + "\n")
|
||||
|
||||
|
||||
@ -393,12 +485,20 @@ if __name__ == "__main__":
|
||||
)
|
||||
else:
|
||||
# 性能测试模式
|
||||
# 定义三个测试问题
|
||||
# 问题1和2完全相同，用于测试精确缓存
|
||||
# 问题3与问题1意思相同但表述略不同，用于测试语义缓存
|
||||
questions = [
|
||||
"公司有什么产品？", # 问题1
|
||||
"公司有什么产品？", # 问题2（完全相同，测试精确缓存）
|
||||
"公司有哪些产品？", # 问题3（表述略不同，测试语义缓存）
|
||||
]
|
||||
|
||||
asyncio.run(
|
||||
run_performance_test(
|
||||
question=question,
|
||||
questions=questions,
|
||||
chat_id=args.chat_id,
|
||||
session_id=args.session_id,
|
||||
num_requests=args.num_requests,
|
||||
delay_between_requests=args.delay
|
||||
)
|
||||
)
|
||||
|
||||
@ -25,6 +25,14 @@ import logging
|
||||
from typing import Optional, Tuple, Dict, Any, List
|
||||
from dataclasses import dataclass, asdict
|
||||
|
||||
# 尝试导入jieba分词库
|
||||
try:
|
||||
import jieba
|
||||
JIEBA_AVAILABLE = True
|
||||
except ImportError:
|
||||
JIEBA_AVAILABLE = False
|
||||
logger.warning("jieba未安装，语义匹配将使用简单的关键词提取")
|
||||
|
||||
# 配置日志
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -107,18 +115,29 @@ class SemanticCacheService:
|
||||
提取问题关键词（用于语义匹配）
|
||||
|
||||
策略：
|
||||
1. 保留完整问题作为主要匹配依据
|
||||
2. 提取名词、动词等核心词汇
|
||||
3. 过滤停用词
|
||||
1. 使用jieba分词（如果可用）提取中文词语
|
||||
2. 保留完整问题作为主要匹配依据
|
||||
3. 提取名词、动词等核心词汇
|
||||
4. 过滤停用词
|
||||
"""
|
||||
# 简单停用词列表
|
||||
stopwords = {'的', '是', '了', '在', '有', '和', '与', '或', '吗', '呢', '吧', '啊', '哦', '请问', '能不能', '可以', '怎么', '如何', '什么', '多少', '几个'}
|
||||
stopwords = {'的', '是', '了', '在', '有', '和', '与', '或', '吗', '呢', '吧', '啊', '哦', '请问', '能不能', '可以', '怎么', '如何', '什么', '多少', '几个', '哪些', '那个', '这个'}
|
||||
|
||||
# 分词（简单按字符或词语切分）
|
||||
words = re.findall(r'[\w\u4e00-\u9fff]+', question.lower())
|
||||
|
||||
# 过滤停用词和过短的词
|
||||
keywords = [w for w in words if w not in stopwords and len(w) > 1]
|
||||
if JIEBA_AVAILABLE:
|
||||
# 使用jieba分词
|
||||
words = list(jieba.cut(question))
|
||||
|
||||
# 过滤停用词、过短的词和标点
|
||||
keywords = [w.strip() for w in words
|
||||
if w.strip() and w.strip() not in stopwords
|
||||
and len(w.strip()) > 1
|
||||
and not re.match(r'^[\s\d\W]+$', w)]
|
||||
else:
|
||||
# 分词（简单按字符或词语切分）
|
||||
words = re.findall(r'[\w\u4e00-\u9fff]+', question.lower())
|
||||
|
||||
# 过滤停用词和过短的词
|
||||
keywords = [w for w in words if w not in stopwords and len(w) > 1]
|
||||
|
||||
return keywords
|
||||
|
||||
@ -203,6 +222,7 @@ class SemanticCacheService:
|
||||
|
||||
# 遍历缓存，计算相似度
|
||||
best_match = None
|
||||
best_match_key = None # 记录最佳匹配的key
|
||||
best_similarity = 0.0
|
||||
|
||||
for key in cache_keys:
|
||||
@ -221,6 +241,7 @@ class SemanticCacheService:
|
||||
if similarity > best_similarity:
|
||||
best_similarity = similarity
|
||||
best_match = entry
|
||||
best_match_key = key # 保存正确的key
|
||||
|
||||
except (json.JSONDecodeError, KeyError) as e:
|
||||
logger.warning(f"[SemanticCache] 解析缓存失败: {key}, {e}")
|
||||
@ -231,7 +252,7 @@ class SemanticCacheService:
|
||||
logger.info(f"[SemanticCache] 语义命中 (相似度={best_similarity:.2f}): {question[:30]}...")
|
||||
# 更新命中次数
|
||||
best_match.hit_count += 1
|
||||
await redis.set(key, json.dumps(best_match.to_dict()), ex=self.CACHE_TTL_HOURS * 3600)
|
||||
await redis.set(best_match_key, json.dumps(best_match.to_dict()), ex=self.CACHE_TTL_HOURS * 3600)
|
||||
return best_match.answer, best_similarity
|
||||
|
||||
return None
|
||||
|
||||
Loading…
Reference in New Issue
Block a user