"""
FastAPI + React 18 单文件全栈应用
专为 Kaggle/Colab 环境设计，展示企业级前后端分离架构
功能特点：
1. 后端：FastAPI (异步、高性能、自动文档)
2. 前端：React 18 + Tailwind CSS (现代化UI、组件化)
3. 部署：单文件运行，自动处理静态资源，支持 ngrok 穿透
使用方法：
python server.py
"""
import os
import sys
import uvicorn
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional
import shutil
# 导入项目核心模块
# 确保项目根目录在 sys.path 中
sys.path.append(os.getcwd())
try:
from config import ENABLE_MULTIMODAL
except Exception:
ENABLE_MULTIMODAL = False
# ============================================================
# 1. FastAPI 后端定义
# ============================================================
app = FastAPI(
title="Adaptive RAG Enterprise API",
description="基于 FastAPI 和 React 构建的企业级 RAG 系统演示",
version="1.0.0"
)
# 允许跨域 (虽然单体部署不需要，但为了开发规范加上)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 全局 RAG 系统实例
rag_system = None
def get_rag_system():
global rag_system
if rag_system is None:
try:
print("🔄 初始化 RAG 系统...")
from main import AdaptiveRAGSystem
rag_system = AdaptiveRAGSystem()
print("✅ RAG 系统初始化完成")
except Exception as e:
print(f"❌ RAG 系统初始化失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
return rag_system
# --- 数据模型 ---
class ChatRequest(BaseModel):
message: str
history: List[dict] = []
class ChatResponse(BaseModel):
answer: str
sources: List[str] = []
metrics: Optional[dict] = None
images: List[str] = []
# --- API 路由 ---
@app.get("/api/health")
async def health_check():
"""健康检查接口"""
return {"status": "ok", "service": "Adaptive RAG", "multimodal": ENABLE_MULTIMODAL}
@app.post("/api/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
"""聊天接口"""
system = get_rag_system()
try:
# 调用 RAG 系统的主查询方法
# 注意：这里假设 main.py 中的 AdaptiveRAGSystem 有 query 方法
# 如果是 main_graphrag.py，可能需要调整调用逻辑
result = system.query(request.message)
# 解析结果
answer = result.get("answer", "无法生成回答")
sources = [doc.page_content[:200] + "..." for doc in result.get("source_documents", [])]
metrics = result.get("retrieval_metrics", {})
# 处理多模态图片结果 (如果有)
images = []
if ENABLE_MULTIMODAL and "images" in result:
# 这里简化处理，实际可能需要返回图片URL或Base64
images = result["images"]
return ChatResponse(
answer=answer,
sources=sources,
metrics=metrics,
images=images
)
except Exception as e:
import traceback
traceback.print_exc()
raise HTTPException(status_code=500, detail=f"处理请求时出错: {str(e)}")
@app.post("/api/upload")
async def upload_file(file: UploadFile = File(...)):
"""文件上传接口"""
try:
# 确保上传目录存在
upload_dir = "./data/uploads"
os.makedirs(upload_dir, exist_ok=True)
file_path = os.path.join(upload_dir, file.filename)
with open(file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
return {"filename": file.filename, "status": "success", "message": "文件上传成功，将在下次索引重建时生效"}
except Exception as e:
raise HTTPException(status_code=500, detail=f"文件上传失败: {str(e)}")
# ============================================================
# 2. 前端 React 应用 (嵌入在 Python 字符串中)
# ============================================================
HTML_CONTENT = """
Enterprise RAG System (React)
"""
@app.get("/", response_class=HTMLResponse)
async def read_root():
"""返回单页应用前端"""
return HTMLResponse(content=HTML_CONTENT)
if __name__ == "__main__":
print("="*60)
print("🚀 企业级 RAG 服务器启动中...")
print(" 后端: FastAPI")
print(" 前端: Vue 3 + Tailwind")
print(" 地址: http://0.0.0.0:8000")
print(" 文档: http://0.0.0.0:8000/docs")
print("="*60)
uvicorn.run(app, host="0.0.0.0", port=8000)