
你要做的是入门级高星 GitHub 项目,之前纠结 Gradio 和 FastAPI 单独用的区别,现在想知道能不能 “后端用 FastAPI(高性能、可扩展性),前端用 Gradio(快速 UI 开发、LLM 友好组件)”,核心关注点是架构合理性、开发效率、部署难度、代码可维护性。
FastAPI+Gradio 的组合,能同时满足后端 API 规范、高性能、可扩展性和前端快速开发、LLM 友好组件、社区资源丰富的需求,是入门级高星 GitHub 项目的最佳选择之一。
职责划分清晰:
后端复用性高:
前端开发速度快:
Docker Compose 一键启动:
分层架构清晰:
app/目录下,前端代码放在gradio/目录下,便于代码的维护和扩展;结合之前的选题 1(本地知识库问答系统),给出 FastAPI+Gradio 的代码实现示例。
local-rag-chatbot/
├── .github/
│ ├── workflows/
│ │ └── ci.yml # GitHub Actions CI/CD配置
├── app/ # FastAPI后端代码
│ ├── api/
│ │ ├── __init__.py
│ │ ├── endpoints/
│ │ │ ├── __init__.py
│ │ │ ├── chat.py # 聊天接口
│ │ │ ├── document.py # 文档上传/解析/删除接口
│ │ └── schemas/
│ │ ├── __init__.py
│ │ ├── chat.py # 聊天接口的数据验证模型
│ │ ├── document.py # 文档上传/解析/删除接口的数据验证模型
│ ├── core/
│ │ ├── __init__.py
│ │ ├── config.py # 项目配置
│ │ ├── database.py # 数据库连接配置
│ │ └── security.py # 安全配置(如密码加密/验证)
│ ├── models/
│ │ ├── __init__.py
│ │ ├── chat.py # 聊天历史记录的ORM模型
│ │ └── document.py # 文档信息的ORM模型
│ ├── services/
│ │ ├── __init__.py
│ │ ├── chat_service.py # 聊天业务逻辑
│ │ ├── document_service.py # 文档业务逻辑
│ │ └── rag_service.py # RAG业务逻辑
│ ├── utils/
│ │ ├── __init__.py
│ │ ├── document_parser.py # 文档解析工具
│ │ └── vector_store.py # 向量库工具
│ └── main.py # FastAPI应用的主入口文件
├── gradio/ # Gradio前端代码
│ ├── __init__.py
│ └── app.py # Gradio应用的主入口文件
├── static/ # 静态资源(如CSS/JS/images,可选)
├── templates/ # 模板文件(如HTML,可选)
├── tests/ # 测试文件
├── .dockerignore # Docker忽略文件
├── .gitignore # Git忽略文件
├── alembic.ini # Alembic配置文件
├── docker-compose.yml # Docker Compose配置文件
├── Dockerfile.fastapi # FastAPI的Dockerfile配置文件
├── Dockerfile.gradio # Gradio的Dockerfile配置文件
├── requirements.fastapi.txt # FastAPI所需的依赖库
├── requirements.gradio.txt # Gradio所需的依赖库
└── README.md # 项目说明文档from fastapi import FastAPI, Depends, HTTPException, status, Form
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from langchain.llms import Ollama
from langchain.vectorstores import Milvus
from langchain.embeddings import OllamaEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoader, UnstructuredMarkdownLoader
from typing import List, Optional
from pydantic import BaseModel
from sqlalchemy.orm import Session
from database import get_db
from models import User, ChatHistory, Document
from schemas import UserCreate, UserInfo, UserUpdate, ChatRequest, ChatResponse, DocumentUploadRequest, DocumentInfo, DocumentDeleteRequest
from utils import get_password_hash, verify_password, create_access_token, authenticate_user, get_current_active_user, get_current_active_admin_user
import os
app = FastAPI(title="FastAPI LLM本地知识库问答系统")
# 挂载静态资源(可选)
app.mount("/static", StaticFiles(directory="static"), name="static")
# 初始化Jinja2模板引擎(可选)
templates = Jinja2Templates(directory="templates")
# 初始化Ollama LLM和嵌入模型
llm = Ollama(model="llama3:8b")
embeddings = OllamaEmbeddings(model="llama3:8b")
# 初始化向量库(Milvus Lite)
vector_store = Milvus(
embedding_function=embeddings,
collection_name="local_rag_chchatbot",
connection_args={"uri": "./milvus.db"}
)
# 初始化文档分块器
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len
)
# 初始化文档加载器
loaders = {
"pdf": PyPDFLoader,
"docx": Docx2txtLoader,
"txt": TextLoader,
"md": UnstructuredMarkdownLoader
}
# 文档上传/解析接口
@app.post("/api/documents/upload", response_model=DocumentInfo)
async def upload_document(request: DocumentUploadRequest, db: Session = Depends(get_db), current_user: User = Depends(get_current_active_user)):
"""
上传并解析文档的接口
:param request: 文档上传请求数据
:param db: 数据库会话(依赖注入)
:param current_user: 当前激活用户的信息(依赖注入)
:return: 文档信息的JSON响应
"""
try:
# 检查文件格式是否支持
file_extension = request.file.filename.split(".")[-1].lower()
if file_extension not in loaders:
raise HTTPException(status_code=400, detail="不支持的文件格式,只支持PDF/Word/TXT/Markdown格式")
# 保存文件到本地
file_path = f"./uploads/{request.file.filename}"
with open(file_path, "wb") as f:
f.write(await request.file.read())
# 解析文件
loader = loaders[file_extension](file_path)
documents = loader.load()
split_documents = text_splitter.split_documents(documents)
# 向量化并存储到向量库
vector_store.add_documents(split_documents)
# 保存文档信息到数据库
db_document = Document(
filename=request.file.filename,
file_path=file_path,
file_size=os.path.getsize(file_path),
file_extension=file_extension,
user_id=current_user.id
)
db.add(db_document)
db.commit()
db.refresh(db_document)
return db_document
except Exception as e:
raise HTTPException(status_code=500, detail=f"文档上传/解析失败:{e}")
# 聊天接口
@app.post("/api/chat", response_model=ChatResponse)
async def chat(request: ChatRequest, db: Session = Depends(get_db), current_user: User = Depends(get_current_active_user)):
"""
与本地知识库聊天的接口
:param request: 聊天请求数据
:param db: 数据库会话(依赖注入)
:param current_user: 当前激活用户的信息(依赖注入)
:return: 聊天响应数据的JSON响应
"""
try:
# 从向量库检索相关文档
retriever = vector_store.as_retriever(
search_kwargs={"k": 3, "score_threshold": 0.8}
)
relevant_documents = retriever.get_relevant_documents(request.message)
# 构造RAG prompt
full_prompt = "请根据以下知识库内容回答用户的问题,如果知识库内容没有提到相关信息,请回答“抱歉,我无法回答您的问题”:\n\n"
for doc in relevant_documents:
full_prompt += f"{doc.page_content}\n\n"
full_prompt += f"用户的问题:{request.message}\n\n"
full_prompt += "回答:"
# 调用LLM生成回答
response = llm.predict(full_prompt)
# 保存聊天历史记录到数据库
db_chat_history = ChatHistory(
user_id=current_user.id,
user_message=request.message,
bot_message=response
)
db.add(db_chat_history)
db.commit()
db.refresh(db_chat_history)
return {"response": response}
except Exception as e:
raise HTTPException(status_code=500, detail=f"聊天失败:{e}")
# 获取用户文档列表的接口
@app.get("/api/documents", response_model=List[DocumentInfo])
async def get_user_documents(db: Session = Depends(get_db), current_user: User = Depends(get_current_active_user)):
"""
获取用户文档列表的接口
:param db: 数据库会话(依赖注入)
:param current_user: 当前激活用户的信息(依赖注入)
:return: 用户文档列表的JSON响应
"""
try:
documents = db.query(Document).filter(Document.user_id == current_user.id).all()
return documents
except Exception as e:
raise HTTPException(status_code=500, detail=f"获取用户文档列表失败:{e}")
# 删除文档的接口
@app.post("/api/documents/delete", response_model=dict)
async def delete_document(request: DocumentDeleteRequest, db: Session = Depends(get_db), current_user: User = Depends(get_current_active_user)):
"""
删除文档的接口
:param request: 文档删除请求数据
:param db: 数据库会话(依赖注入)
:param current_user: 当前激活用户的信息(依赖注入)
:return: 删除成功的响应
"""
try:
document = db.query(Document).filter(Document.id == request.document_id, Document.user_id == current_user.id).first()
if not document:
raise HTTPException(status_code=404, detail="文档不存在")
# 从向量库删除文档的向量
# Milvus Lite不支持根据文档ID删除向量,所以需要先删除整个集合,然后重新添加其他文档的向量
# 这里简化处理,直接删除本地文件和数据库记录
os.remove(document.file_path)
db.delete(document)
db.commit()
return {"message": "文档删除成功"}
except Exception as e:
raise HTTPException(status_code=500, detail=f"删除文档失败:{e}")
# 获取用户聊天历史记录的接口
@app.get("/api/chat-history", response_model=List[dict])
async def get_chat_history(db: Session = Depends(get_db), current_user: User = Depends(get_current_active_user)):
"""
获取用户聊天历史记录的接口
:param db: 数据库会话(依赖注入)
:param current_user: 当前激活用户的信息(依赖注入)
:return: 用户聊天历史记录的JSON响应
"""
try:
chat_history = db.query(ChatHistory).filter(ChatHistory.user_id == current_user.id).all()
return [{"user_message": item.user_message, "bot_message": item.bot_message} for item in chat_history]
except Exception as e:
raise HTTPException(status_code=500, detail=f"获取聊天历史记录失败:{e}")
# 删除用户聊天历史记录的接口
@app.post("/api/chat-history/delete", response_model=dict)
async def delete_chat_history(db: Session = Depends(get_db), current_user: User = Depends(get_current_active_user)):
"""
删除用户聊天历史记录的接口
:param db: 数据库会话(依赖注入)
:param current_user: 当前激活用户的信息(依赖注入)
:return: 删除成功的响应
"""
try:
db.query(ChatHistory).filter(ChatHistory.user_id == current_user.id).delete()
db.commit()
return {"message": "聊天历史记录删除成功"}
except Exception as e:
raise HTTPException(status_code=500, detail=f"删除聊天历史记录失败:{e}")
# 用户注册接口
@app.post("/api/users/register", response_model=UserInfo)
async def register(request: UserCreate, db: Session = Depends(get_db)):
"""
用户注册的接口
:param request: 用户注册请求数据
:param db: 数据库会话(依赖注入)
:return: 新创建用户的信息
"""
try:
# 检查用户名和邮箱是否已经存在
db_user = db.query(User).filter(User.username == request.username).first()
if db_user:
raise HTTPException(status_code=400, detail="用户名已存在")
db_email = db.query(User).filter(User.email == request.email).first()
if db_email:
raise HTTPException(status_code=400, detail="邮箱已存在")
# 加密密码
hashed_password = get_password_hash(request.password)
# 创建新用户
db_user = User(
username=request.username,
email=request.email,
hashed_password=hashed_password,
age=request.age,
is_admin=request.is_admin
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
except Exception as e:
raise HTTPException(status_code=500, detail=f"用户注册失败:{e}")
# 用户登录接口(获取JWT访问令牌)
@app.post("/api/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
"""
用户登录的接口(获取JWT访问令牌)
:param form_data: OAuth2密码模式的请求数据(用户名和密码)
:param db: 数据库会话(依赖注入)
:return: JWT访问令牌的JSON响应
"""
try:
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户名或密码错误",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
except Exception as e:
raise HTTPException(status_code=500, detail=f"用户登录失败:{e}")import gradio as gr
import requests
# FastAPI后端的API接口地址
API_BASE_URL = "http://localhost:8000/api"
# 发送消息到FastAPI后端的函数
def send_message(message, history, token):
"""
发送消息到FastAPI后端
:param message: 用户输入的消息
:param history: 聊天历史记录(格式:[(user_msg1, bot_msg1), (user_msg2, bot_msg2)])
:param token: JWT访问令牌
:return: LLM的回复
"""
try:
url = f"{API_BASE_URL}/chat"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {token}"
}
data = {
"message": message,
"history": history
}
response = requests.post(url, headers=headers, json=data)
response.raise_for_status()
return response.json()["response"]
except Exception as e:
return f"聊天失败:{e}"
# 上传文档到FastAPI后端的函数
def upload_document(file, token):
"""
上传文档到FastAPI后端
:param file: 用户上传的文件
:param token: JWT访问令牌
:return: 文档上传成功的消息
"""
try:
url = f"{API_BASE_URL}/documents/upload"
headers = {
"Authorization": f"Bearer {token}"
}
files = {
"file": file
}
response = requests.post(url, headers=headers, files=files)
response.raise_for_status()
return f"文档上传成功:{response.json()['filename']}"
except Exception as e:
return f"文档上传失败:{e}"
# 获取用户文档列表的函数
def get_user_documents(token):
"""
获取用户文档列表
:param token: JWT访问令牌
:return: 用户文档列表的HTML表格
"""
try:
url = f"{API_BASE_URL}/documents"
headers = {
"Authorization": f"Bearer {token}"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
documents = response.json()
if not documents:
return "您还没有上传任何文档"
table = "<table border='1'><tr><th>文件名</th><th>文件大小</th><th>文件格式</th><th>上传时间</th></tr>"
for doc in documents:
table += f"<tr><td>{doc['filename']}</td><td>{doc['file_size']}字节</td><td>{doc['file_extension']}</td><td>{doc['created_at']}</td></tr>"
table += "</table>"
return table
except Exception as e:
return f"获取文档列表失败:{e}"
# 删除文档的函数
def delete_document(document_id, token):
"""
删除文档
:param document_id: 文档的唯一标识符
:param token: JWT访问令牌
:return: 文档删除成功的消息
"""
try:
url = f"{API_BASE_URL}/documents/delete"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {token}"
}
data = {
"document_id": document_id
}
response = requests.post(url, headers=headers, json=data)
response.raise_for_status()
return f"文档删除成功:{document_id}"
except Exception as e:
return f"删除文档失败:{e}"
# 获取聊天历史记录的函数
def get_chat_history(token):
"""
获取聊天历史记录
:param token: JWT访问令牌
:return: 聊天历史记录的HTML表格
"""
try:
url = f"{API_BASE_URL}/chat-history"
headers = {
"Authorization": f"Bearer {token}"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
chat_history = response.json()
if not chat_history:
return "您还没有任何聊天历史记录"
table = "<table border='1'><tr><th>用户消息</th><th>机器人消息</th></tr>"
for msg in chat_history:
table += f"<tr><td>{msg['user_message']}</td><td>{msg['bot_message']}</td></tr>"
table += "</table>"
return table
except Exception as e:
return f"获取聊天历史记录失败:{e}"
# 删除聊天历史记录的函数
def delete_chat_history(token):
"""
删除聊天历史记录
:param token: JWT访问令牌
:return: 聊天历史记录删除成功的消息
"""
try:
url = f"{API_BASE_URL}/chat-history/delete"
headers = {
"Authorization": f"Bearer {token}"
}
response = requests.post(url, headers=headers)
response.raise_for_status()
return "聊天历史记录删除成功"
except Exception as e:
return f"删除聊天历史记录失败:{e}"
# 用户注册的函数
def register(username, email, password, age, is_admin):
"""
用户注册
:param username: 用户名
:param email: 邮箱
:param password: 密码
:param age: 年龄
:param is_admin: 是否是管理员
:return: 注册成功的消息
"""
try:
url = f"{API_BASE_URL}/users/register"
headers = {
"Content-Type": "application/json"
}
data = {
"username": username,
"email": email,
"password": password,
"age": age,
"is_admin": is_admin
}
response = requests.post(url, headers=headers, json=data)
response.raise_for_status()
return f"注册成功:{response.json()['username']}"
except Exception as e:
return f"注册失败:{e}"
# 用户登录的函数
def login(username, password):
"""
用户登录
:param username: 用户名
:param password: 密码
:return: JWT访问令牌
"""
try:
url = f"{API_BASE_URL}/token"
headers = {
"Content-Type": "application/x-www-form-urlencoded"
}
data = {
"username": username,
"password": password
}
response = requests.post(url, headers=headers, data=data)
response.raise_for_status()
return response.json()["access_token"]
except Exception as e:
return f"登录失败:{e}"
# 创建Gradio应用
with gr.Blocks(title="FastAPI+Gradio本地知识库问答系统", theme="soft") as demo:
# 登录/注册组件
with gr.Tab("登录"):
with gr.Row():
with gr.Column():
username_input = gr.Textbox(label="用户名", placeholder="请输入您的用户名")
password_input = gr.Textbox(label="密码", placeholder="请输入您的密码", type="password")
login_btn = gr.Button("登录")
login_output = gr.Textbox(label="登录结果", interactive=False)
login_btn.click(fn=login, inputs=[username_input, password_input], outputs=login_output)
with gr.Tab("注册"):
with gr.Row():
with gr.Column():
reg_username_input = gr.Textbox(label="用户名", placeholder="请输入您的用户名")
reg_email_input = gr.Textbox(label="邮箱", placeholder="请输入您的邮箱")
reg_password_input = gr.Textbox(label="密码", placeholder="请输入您的密码", type="password")
reg_age_input = gr.Number(label="年龄", placeholder="请输入您的年龄", minimum=1, maximum=120)
reg_is_admin_input = gr.Checkbox(label="是否是管理员")
reg_btn = gr.Button("注册")
reg_output = gr.Textbox(label="注册结果", interactive=False)
reg_btn.click(fn=register, inputs=[reg_username_input, reg_email_input, reg_password_input, reg_age_input, reg_is_admin_input], outputs=reg_output)
# 聊天组件(需要登录后才能使用)
with gr.Tab("聊天"):
with gr.Row():
with gr.Column():
token_input = gr.Textbox(label="JWT访问令牌", placeholder="请输入您的JWT访问令牌", type="password")
message_input = gr.Textbox(label="消息", placeholder="请输入您的消息")
send_btn = gr.Button("发送")
chat_output = gr.Chatbot(label="聊天历史记录")
send_btn.click(fn=send_message, inputs=[message_input, chat_output, token_input], outputs=chat_output)
# 文档管理组件(需要登录后才能使用)
with gr.Tab("文档管理"):
with gr.Row():
with gr.Column():
doc_token_input = gr.Textbox(label="JWT访问令牌", placeholder="请输入您的JWT访问令牌", type="password")
file_input = gr.File(label="上传文档", file_types=["pdf", "docx", "txt", "md"])
upload_btn = gr.Button("上传")
upload_output = gr.Textbox(label="上传结果", interactive=False)
get_docs_btn = gr.Button("获取文档列表")
docs_output = gr.HTML(label="文档列表")
delete_doc_id_input = gr.Number(label="文档ID", placeholder="请输入要删除的文档ID")
delete_doc_btn = gr.Button("删除文档")
delete_doc_output = gr.Textbox(label="删除结果", interactive=False)
upload_btn.click(fn=upload_document, inputs=[file_input, doc_token_input], outputs=upload_output)
get_docs_btn.click(fn=get_user_documents, inputs=[doc_token_input], outputs=docs_output)
delete_doc_btn.click(fn=delete_document, inputs=[delete_doc_id_input, doc_token_input], outputs=delete_doc_output)
# 聊天历史记录管理组件(需要登录后才能使用)
with gr.Tab("聊天历史记录管理"):
with gr.Row():
with gr.Column():
history_token_input = gr.Textbox(label="JWT访问令牌", placeholder="请输入您的JWT访问令牌", type="password")
get_history_btn = gr.Button("获取聊天历史记录")
history_output = gr.HTML(label="聊天历史记录")
delete_history_btn = gr.Button("删除聊天历史记录")
delete_history_output = gr.Textbox(label="删除结果", interactive=False)
get_history_btn.click(fn=get_chat_history, inputs=[history_token_input], outputs=history_output)
delete_history_btn.click(fn=delete_chat_history, inputs=[history_token_input], outputs=delete_history_output)
if __name__ == "__main__":
demo.launch(server_name="0.0.0.0", server_port=7860, share=False)version: "3.8"
services:
fastapi:
build:
context: .
dockerfile: Dockerfile.fastapi
ports:
- "8000:8000"
volumes:
- ./uploads:/app/uploads
- ./milvus.db:/app/milvus.db
- ./app:/app/app
environment:
- DATABASE_URL=sqlite:///./app.db
command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
gradio:
build:
context: .
dockerfile: Dockerfile.gradio
ports:
- "7860:7860"
volumes:
- ./gradio:/app/gradio
environment:
- API_BASE_URL=http://fastapi:8000/api
command: python -m gradio app.gradio.app --host 0.0.0.0 --port 7860 --reloadFROM python:3.11-slim
WORKDIR /app
# 安装项目所需的依赖库
COPY requirements.fastapi.txt .
RUN pip install --no-cache-dir -r requirements.fastapi.txt
# 复制项目代码
COPY app /app/app
COPY database.py /app/database.py
COPY models.py /app/models.py
COPY schemas.py /app/schemas.py
COPY utils.py /app/utils.py
# 创建uploads文件夹
RUN mkdir -p /app/uploads
# 暴露端口
EXPOSE 8000
# 启动FastAPI应用
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]FROM python:3.11-slim
WORKDIR /app
# 安装项目所需的依赖库
COPY requirements.gradio.txt .
RUN pip install --no-cache-dir -r requirements.gradio.txt
# 复制项目代码
COPY gradio /app/gradio
# 暴露端口
EXPOSE 7860
# 启动Gradio应用
CMD ["python", "-m", "gradio", "gradio.app", "--host", "0.0.0.0", "--port", "7860"]fastapi==0.109.0
uvicorn==0.27.0
python-multipart==0.0.6
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
sqlalchemy==2.0.25
alembic==1.13.1
pymysql==1.1.0
psycopg2-binary==2.9.9
langchain==0.1.0
langchain-community==0.0.17
langchain-core==0.1.16
ollama==0.2.1
pymilvus==2.4.3
pypdf2==3.0.0
python-docx==0.8.11
unstructured==0.10.30
unstructured[pdf]==0.10.30
unstructured[docx]==0.10.30
unstructured[md]==0.10.30gradio==4.14.0
requests==2.31.0在终端中进入项目根目录,输入以下命令启动项目:
docker-compose up -d如果条件允许,可以将项目部署到云平台(如 AWS、GCP、Azure、阿里云、腾讯云),提供在线访问地址。建议使用AWS Lightsail或阿里云轻量应用服务器,因为它们的价格便宜,操作简单。
FastAPI+Gradio 是 “1+1>2” 的黄金组合,能同时满足后端 API 规范、高性能、可扩展性和前端快速开发、LLM 友好组件、社区资源丰富的需求,是入门级高星 GitHub 项目的最佳选择之一。
如果你的项目对 API 规范有要求,或者对后端性能有要求,或者之前已经用 FastAPI 写过核心业务逻辑,那么建议使用 FastAPI+Gradio 的组合。如果你的项目对界面有极高定制化要求,或者项目需要和已有 FastAPI 后端深度集成,那么可以考虑用 FastAPI + 前端框架(如 React、Vue、Angular)的组合。