16c42787
 tangwang
feat: implement r...
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
#!/usr/bin/env python3
"""
测试修复后的查询解析功能
验证翻译和向量生成是否正常工作
"""
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
def test_fixed_query_parsing():
"""测试修复后的查询解析"""
print("🧪 测试修复后的查询解析功能")
print("=" * 60)
try:
from context.request_context import create_request_context
from query.query_parser import QueryParser
from config import CustomerConfig
from config.config_loader import ConfigLoader
# 加载配置
print("📝 加载配置...")
loader = ConfigLoader()
config = loader.load_customer_config("customer1")
print(f"✅ 配置加载成功: {config.customer_id}")
print(f" - 翻译功能: {'启用' if config.query_config.enable_translation else '禁用'}")
print(f" - 向量功能: {'启用' if config.query_config.enable_text_embedding else '禁用'}")
# 创建解析器和上下文
parser = QueryParser(config)
context = create_request_context("test_fixed", "test_user")
# 测试查询
test_query = "推车"
print(f"\n🔍 测试查询: '{test_query}'")
# 执行解析
result = parser.parse(
test_query,
context=context,
generate_vector=config.query_config.enable_text_embedding
)
# 显示结果
print(f"\n📊 查询解析结果:")
print(f" 原查询: {result.original_query}")
print(f" 标准化: {result.normalized_query}")
print(f" 重写后: {result.rewritten_query}")
print(f" 检测语言: {result.detected_language}")
print(f" 域: {result.domain}")
print(f" 翻译结果: {result.translations}")
if result.query_vector is not None:
print(f" 向量: ✅ 已生成 (形状: {result.query_vector.shape})")
print(f" 向量类型: {type(result.query_vector)}")
print(f" 向量前5个值: {result.query_vector[:5]}")
else:
print(f" 向量: ❌ 未生成")
# 检查翻译质量
if result.translations:
print(f"\n🌍 翻译质量检查:")
for lang, translation in result.translations.items():
if translation:
print(f" {lang}: '{translation}' ✅")
else:
print(f" {lang}: 翻译失败 ❌")
else:
print(f"\n🌍 翻译: 无翻译结果")
# 测试上下文存储
print(f"\n💾 上下文存储检查:")
stored_query = context.get_intermediate_result('normalized_query')
stored_lang = context.get_intermediate_result('detected_language')
stored_translations = context.get_intermediate_result('translations')
print(f" 存储的查询: {stored_query}")
print(f" 存储的语言: {stored_lang}")
print(f" 存储的翻译: {stored_translations}")
# 性能摘要
summary = context.get_summary()
print(f"\n📈 性能摘要:")
print(f" 请求ID: {summary['request_info']['reqid']}")
print(f" 用户ID: {summary['request_info']['uid']}")
print(f" 有错误: {summary['request_info']['has_error']}")
print(f" 警告数量: {summary['request_info']['warnings_count']}")
print(f" 查询有向量: {summary['query_analysis']['has_vector']}")
# 判断修复是否成功
print(f"\n🎯 修复结果评估:")
translation_success = (
result.translations and
any(translation is not None and translation != result.original_query
for translation in result.translations.values())
)
vector_success = result.query_vector is not None
print(f" 翻译功能: {'✅ 修复成功' if translation_success else '❌ 仍有问题'}")
print(f" 向量功能: {'✅ 修复成功' if vector_success else '❌ 仍有问题'}")
if translation_success and vector_success:
print(f"\n🎉 所有功能修复成功！")
return True
else:
print(f"\n⚠️ 还有功能需要修复")
return False
except Exception as e:
print(f"❌ 测试失败: {e}")
import traceback
traceback.print_exc()
return False
if __name__ == "__main__":
success = test_fixed_query_parsing()
if success:
print(f"\n✨ 修复验证完成 - 系统正常运行！")
else:
print(f"\n💥 修复验证失败 - 需要进一步检查")
sys.exit(0 if success else 1)
|