🤖 自适应RAG智能问答

""" Kaggle Flask 智能问答界面在 Kaggle Notebook 中使用 Flask 创建交互式 RAG 系统 """ from flask import Flask, render_template_string, request, jsonify import sys import os import threading # 添加项目路径 if '/kaggle/working/adaptive_RAG' not in sys.path: sys.path.insert(0, '/kaggle/working/adaptive_RAG') from main import AdaptiveRAGSystem # 创建Flask应用 app = Flask(__name__) # 全局RAG系统实例 rag_system = None initialization_error = None # HTML模板 HTML_TEMPLATE = """ 🤖 自适应RAG智能问答

🤖

你好！我是自适应RAG智能助手。我可以帮你回答关于LLM、Agent、Prompt Engineering等问题。请输入你的问题！

💡 示例问题

⏳ AI正在思考中...

""" @app.route('/') def index(): """主页""" return render_template_string( HTML_TEMPLATE, system_ok=(rag_system is not None), error=initialization_error or "" ) @app.route('/api/query', methods=['POST']) def query(): """处理查询请求""" if rag_system is None: return jsonify({ 'success': False, 'error': f'系统未初始化: {initialization_error}' }) try: data = request.get_json() question = data.get('question', '').strip() if not question: return jsonify({ 'success': False, 'error': '问题不能为空' }) # 查询RAG系统 result = rag_system.query(question, verbose=False) answer = result.get('answer', '无法生成答案') return jsonify({ 'success': True, 'answer': answer, 'metrics': result.get('retrieval_metrics') }) except Exception as e: return jsonify({ 'success': False, 'error': str(e) }) def initialize_rag(): """初始化RAG系统""" global rag_system, initialization_error print("🚀 正在初始化RAG系统...") try: rag_system = AdaptiveRAGSystem() print("✅ RAG系统初始化成功") except Exception as e: initialization_error = str(e) print(f"❌ RAG系统初始化失败: {e}") def run_flask(host='0.0.0.0', port=5000): """ 启动Flask应用 Args: host: 主机地址 port: 端口号 """ print("=" * 60) print("🚀 启动 Flask RAG 智能问答系统") print("=" * 60) # 初始化RAG系统 initialize_rag() # 检查是否在Kaggle环境 is_kaggle = os.path.exists('/kaggle/working') if is_kaggle: print("🎯 检测到 Kaggle 环境") print(f"💡 访问地址: http://localhost:{port}") print("⚠️ 注意: Kaggle无法从外部访问，只能在Notebook内查看") print(f"\n🌐 正在启动服务...") print(f" 地址: http://{host}:{port}") print("=" * 60) # 启动Flask app.run(host=host, port=port, debug=False, threaded=True) if __name__ == "__main__": run_flask(port=5000)