diff --git a/.gitignore b/.gitignore index df920ad..5f26efd 100644 --- a/.gitignore +++ b/.gitignore @@ -23,7 +23,8 @@ !.gitea/** !tools/ !tools/** - +!models/ +!models/** # 3. 放行必要的根目录文件 !.gitignore !README.md diff --git a/backend/test/test_hybrid_router.py b/backend/test/test_hybrid_router.py deleted file mode 100644 index 8e64f1c..0000000 --- a/backend/test/test_hybrid_router.py +++ /dev/null @@ -1,171 +0,0 @@ -#!/usr/bin/env python3 -""" -完整的混合路由测试脚本 -""" -import sys -from pathlib import Path - -# 添加后端路径 -sys.path.insert(0, str(Path(__file__).parent.parent / "backend")) - - -def test_imports(): - """测试所有导入是否正常""" - print("="*70) - print("🧪 步骤 1/5 - 测试导入") - print("="*70) - - try: - from app.model_services.chat_services import get_chat_service, get_small_llm_service - print("✅ chat_services 导入成功") - - from app.main_graph.nodes.hybrid_router import ( - hybrid_router_node, - fast_chitchat_node, - route_from_hybrid_decision, - check_fast_path_success - ) - print("✅ hybrid_router 导入成功") - - from app.main_graph.utils.main_graph_builder import build_react_main_graph - print("✅ main_graph_builder 导入成功") - - from app.core.intent import react_reason, react_reason_async - print("✅ intent 导入成功") - - print("\n✅ 所有导入测试通过！") - return True - except Exception as e: - print(f"❌ 导入测试失败: {e}") - import traceback - traceback.print_exc() - return False - - -def test_small_llm(): - """测试小模型服务""" - print("\n" + "="*70) - print("🧪 步骤 2/5 - 测试小模型服务") - print("="*70) - - try: - from app.model_services.chat_services import get_small_llm_service - llm = get_small_llm_service() - print(f"✅ 小模型服务获取成功: {type(llm)}") - return True - except Exception as e: - print(f"❌ 小模型服务测试失败: {e}") - print("💡 小模型服务不可用是正常的，会自动降级到大模型") - return True - - -def test_rules_based_redirect(): - """测试规则分流""" - print("\n" + "="*70) - print("🧪 步骤 3/5 - 测试规则分流") - print("="*70) - - try: - from app.main_graph.nodes.hybrid_router import _rule_based_redirect - - # 测试 1: 问候 - result = _rule_based_redirect("你好") - if result and result.path == "fast_chitchat": - print(f"✅ 问候测试通过: path={result.path}") - else: - print(f"⚠️ 问候测试: result={result}") - - # 测试 2: 感谢 - result = _rule_based_redirect("谢谢") - if result and result.path == "fast_chitchat": - print(f"✅ 感谢测试通过: path={result.path}") - else: - print(f"⚠️ 感谢测试: result={result}") - - # 测试 3: 子图关键词 - result = _rule_based_redirect("查一下通讯录") - if result and result.path == "fast_tool": - print(f"✅ 通讯录关键词测试通过: path={result.path}") - else: - print(f"⚠️ 通讯录关键词测试: result={result}") - - # 测试 4: 复杂问题（不触发规则） - result = _rule_based_redirect("什么是 LangGraph？") - if result is None: - print(f"✅ 复杂问题测试通过: 规则不触发，走模型分类") - else: - print(f"⚠️ 复杂问题测试: result={result}") - - print("\n✅ 规则分流测试完成！") - return True - except Exception as e: - print(f"❌ 规则分流测试失败: {e}") - import traceback - traceback.print_exc() - return False - - -def test_build_graph(): - """测试图构建""" - print("\n" + "="*70) - print("🧪 步骤 4/5 - 测试图构建（混合路由模式）") - print("="*70) - - try: - from app.main_graph.utils.main_graph_builder import build_react_main_graph - - # 构建启用混合路由的图 - graph = build_react_main_graph(use_hybrid_router=True) - print(f"✅ 图构建成功（混合路由）") - - # 编译图 - compiled_graph = graph.compile() - print(f"✅ 图编译成功（混合路由）") - - # 构建纯 React 的图（兼容模式） - graph_react = build_react_main_graph(use_hybrid_router=False) - compiled_graph_react = graph_react.compile() - print(f"✅ 图构建成功（纯 React）") - - print("\n✅ 图构建测试完成！") - return True - except Exception as e: - print(f"❌ 图构建测试失败: {e}") - import traceback - traceback.print_exc() - return False - - -def test_summary(): - """测试总结""" - print("\n" + "="*70) - print("🎉 完整的混合路由优化已实现！") - print("="*70) - print("\n✅ 已完成的优化:") - print(" 1. 双模型服务 (llm + smallLLM)") - print(" 2. 规则快速分流 (无 LLM, 超快速)") - print(" 3. 轻量级意图分类 (smallLLM)") - print(" 4. 快速路径 (fast_chitchat, fast_rag, fast_tool)") - print(" 5. 自动升级机制 (快速路径失败 -> React 循环)") - print(" 6. SSE 事件增强 (intent_classified, path_decision, fast_path_*)") - print(" 7. 向后兼容 (可切换 use_hybrid_router=True/False)") - - -if __name__ == "__main__": - print("\n" + "🚀"*10) - print("🚀 混合路由系统测试") - print("🚀"*10 + "\n") - - results = [] - results.append(test_imports()) - results.append(test_small_llm()) - results.append(test_rules_based_redirect()) - results.append(test_build_graph()) - test_summary() - - if all(results): - print("\n✅ 所有测试通过！") - sys.exit(0) - else: - print("\n⚠️ 部分测试失败，请检查") - sys.exit(1) diff --git a/backend/test/test_mcp_simple.py b/backend/test/test_mcp_simple.py deleted file mode 100644 index 0ce736c..0000000 --- a/backend/test/test_mcp_simple.py +++ /dev/null @@ -1,112 +0,0 @@ -#!/usr/bin/env python3 -""" -简化版MCP测试 - 直接测试mcp模块 -""" -import asyncio -import sys -from pathlib import Path - -# 直接添加mcp模块路径 -sys.path.insert(0, str(Path(__file__).parent / "backend" / "app")) - - -async def test_mcp_direct(): - """直接测试MCP模块""" - print("=" * 70) - print("🧪 直接测试 MCP 模块") - print("=" * 70) - - # 直接导入mcp子模块 - print("\n[1/4] 导入MCP核心模块...") - try: - from mcp.mcp_manager import MCPManager, mcp_manager - from mcp.adapters.base_adapter import BaseAdapter, AdapterResult - print("✅ 核心模块导入成功") - except Exception as e: - print(f"❌ 核心模块导入失败: {e}") - import traceback - traceback.print_exc() - return False - - # 导入适配器 - print("\n[2/4] 导入适配器...") - try: - from mcp.adapters.contact_adapter import ContactAdapter - from mcp.adapters.dictionary_adapter import DictionaryAdapter - from mcp.adapters.news_adapter import NewsAdapter - print("✅ 适配器导入成功") - except Exception as e: - print(f"❌ 适配器导入失败: {e}") - import traceback - traceback.print_exc() - return False - - # 测试适配器 - print("\n[3/4] 测试各个适配器...") - - print("\n📖 测试 DictionaryAdapter...") - dict_adapter = DictionaryAdapter() - result = dict_adapter._fallback("query_word", word="ephemeral") - print(f" 来源: {result.source}") - print(f" 成功: {result.success}") - print(f" 单词: {result.data.get('word', '')}") - print(f" 释义: {result.data.get('definitions', [])}") - - print("\n📰 测试 NewsAdapter...") - news_adapter = NewsAdapter() - result = news_adapter._fallback("query_news", query="AI") - print(f" 来源: {result.source}") - print(f" 成功: {result.success}") - print(f" 数量: {len(result.data) if result.data else 0}") - - print("\n👥 测试 ContactAdapter...") - contact_adapter = ContactAdapter() - result = contact_adapter._fallback("list_contacts", user_id="test") - print(f" 来源: {result.source}") - print(f" 成功: {result.success}") - print(f" 数量: {len(result.data) if result.data else 0}") - - # 测试MCP管理器 - print("\n[4/4] 测试 MCP Manager...") - try: - mcp_manager.register_adapter(ContactAdapter()) - mcp_manager.register_adapter(DictionaryAdapter()) - mcp_manager.register_adapter(NewsAdapter()) - - await mcp_manager.initialize() - - print(f"✅ 可用适配器: {mcp_manager.get_available_adapters()}") - print(f"✅ 可用工具: {mcp_manager.get_available_tools()}") - - # 测试通过manager调用 - print("\n测试通过Manager调用...") - result = await mcp_manager.execute( - "dictionary", "query_word", word="serendipity", user_id="test" - ) - print(f" 成功: {result.success}") - print(f" 来源: {result.source}") - - except Exception as e: - print(f"❌ MCP Manager 测试失败: {e}") - import traceback - traceback.print_exc() - - print("\n" + "=" * 70) - print("🎉 直接测试完成！") - print("=" * 70) - print("\n✅ MCP集成系统架构已就绪:") - print(" ┌─────────────────────────────────────────────┐") - print(" │ MCP Manager (统一入口) │") - print(" ├─────────────────────────────────────────────┤") - print(" │ ContactAdapter │ DictionaryAdapter │ News │") - print(" ├─────────────────────────────────────────────┤") - print(" │ MCP Client -> Database -> Mock (降级) │") - print(" └─────────────────────────────────────────────┘") - print("\n📖 文档: docs/MCP_INTEGRATION.md") - print("⚙️ 配置: backend/app/mcp/mcp_config.example.yaml") - - return True - - -if __name__ == "__main__": - asyncio.run(test_mcp_direct())