125 lines
4.6 KiB
Python
125 lines
4.6 KiB
Python
"""
|
||
智能舷号映射 + RAG检索 完整测试脚本
|
||
直接调用 search_with_ship_number_strategy，展示真实检索结果
|
||
"""
|
||
import asyncio
|
||
import sys
|
||
import os
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from utils.ship_number_search import search_with_ship_number_strategy, smart_ship_number_mapping
|
||
|
||
|
||
async def interactive_test():
|
||
"""交互式完整测试"""
|
||
print("=" * 60)
|
||
print("智能舷号映射 + RAG检索 完整测试")
|
||
print("=" * 60)
|
||
print("输入 'exit' 或 'quit' 退出")
|
||
print("输入 'help' 查看说明")
|
||
print("=" * 60)
|
||
|
||
while True:
|
||
print("\n" + "-" * 60)
|
||
ship_input = input("请输入舷号/型号/舰名: ").strip()
|
||
|
||
if ship_input.lower() in ['exit', 'quit', 'q']:
|
||
print("退出测试")
|
||
break
|
||
|
||
if ship_input.lower() == 'help':
|
||
print("\n使用说明:")
|
||
print(" 先输入舷号/型号/舰名（如 163、南昌舰、055驱逐舰）")
|
||
print(" 再输入查询内容（如 主机排烟温度高）")
|
||
print(" 系统会执行完整的智能映射 + RAG三步检索")
|
||
continue
|
||
|
||
if not ship_input:
|
||
print("请输入有效内容")
|
||
continue
|
||
|
||
query_input = input("请输入查询内容: ").strip()
|
||
if not query_input:
|
||
query_input = "主机排烟温度高的维修方案"
|
||
print(f"使用默认查询: {query_input}")
|
||
|
||
search_type = input("搜索类型 (fault/operate，回车默认fault): ").strip()
|
||
if search_type not in ["fault", "operate"]:
|
||
search_type = "fault"
|
||
|
||
print("\n" + "=" * 60)
|
||
print("第一步：智能映射")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
matched_ship_number, matched_model, all_numbers, match_type = await smart_ship_number_mapping(ship_input)
|
||
print(f"输入: '{ship_input}'")
|
||
print(f"匹配类型: {match_type}")
|
||
print(f"匹配舷号: {matched_ship_number}")
|
||
print(f"匹配型号: {matched_model}")
|
||
print(f"该型号所有舷号: {all_numbers}")
|
||
|
||
if match_type == "none":
|
||
print("未匹配到，将使用原始输入进行检索")
|
||
except Exception as e:
|
||
print(f"映射失败: {str(e)}")
|
||
matched_ship_number = None
|
||
matched_model = None
|
||
all_numbers = None
|
||
match_type = "none"
|
||
|
||
print("\n" + "=" * 60)
|
||
print("第二步：RAG检索（三步策略）")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
rag_results, matched_kb_name, matched_kb_id = await search_with_ship_number_strategy(
|
||
base_query=query_input,
|
||
ship_number=ship_input,
|
||
top_k=5,
|
||
search_type=search_type
|
||
)
|
||
|
||
print(f"\n检索完成!")
|
||
print(f"知识库: {matched_kb_name or '未匹配'}")
|
||
print(f"知识库ID: {matched_kb_id or '未匹配'}")
|
||
print(f"结果数量: {len(rag_results)}")
|
||
|
||
if rag_results:
|
||
print(f"\n{'=' * 60}")
|
||
print("第三步：检索结果详情")
|
||
print("=" * 60)
|
||
for i, item in enumerate(rag_results, 1):
|
||
if isinstance(item, dict):
|
||
text = item.get("text", "")
|
||
kb_name = item.get("kb_name", "")
|
||
score = item.get("score", "")
|
||
source = item.get("source", "")
|
||
|
||
print(f"\n--- 结果 {i} ---")
|
||
if kb_name:
|
||
print(f"知识库: {kb_name}")
|
||
if score:
|
||
print(f"相似度: {score}")
|
||
if source:
|
||
print(f"来源: {source}")
|
||
if text:
|
||
preview = text[:300] + "..." if len(text) > 300 else text
|
||
print(f"内容预览:\n{preview}")
|
||
elif isinstance(item, str):
|
||
preview = item[:300] + "..." if len(item) > 300 else item
|
||
print(f"\n--- 结果 {i} ---")
|
||
print(f"内容预览:\n{preview}")
|
||
else:
|
||
print("\n未检索到任何结果")
|
||
|
||
except Exception as e:
|
||
print(f"\n检索失败: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(interactive_test())
|