1.项目介绍
大家好我是文淇AI开发者,是一位全栈开发初学者。经过一段时间的学习,我的第一个项目文淇ai聊天终于迎来1.0版本。
首先介绍运行这个项目,该项目为可以本地部署,引用阿里云千问大模型api,具有流式输出,qq登录注册,历史记录的ai聊天网页。前端技术选型简单即:HTML,CSS,JS。后端主要是Python 3.9,fastapi,SQLite。
项目定位
「轻量级 AI 聊天助手」是一个基于浏览器即可使用的对话式 AI 前端应用。用户无需安装任何软件,打开网页即可与后端大模型进行自然语言交互,实现问答、写作、翻译、代码生成等常见场景。
核心功能
匿名/账号双模式:auth.html 支持注册/登录,匿名模式直接跳过。
流式对话:前端 script.js 通过 EventSource / fetch-stream 逐字打印 AI 回答,体验更丝滑。
会话持久化:聊天记录写入 test.db,页面刷新不丢失。
环境隔离:所有敏感配置(OpenAI Key、端口、数据库 URI)统一放到 .env。
响应式 UI:CSS Flex + 媒体查询,手机/平板/桌面多端适配。
项目结构

2.架构分析
整体视图

分层架构
层 |
名称 |
技术 |
端口/协议 |
主要职责 |
表现层 |
Browser SPA |
HTML5+CSS3+ JS |
443/80 (TLS) |
渲染 UI、收集输入、流式展示回答 |
网关层 |
Nginx / Vercel Edge |
反向代理 |
443 |
静态资源直出、TLS 终止、压缩、缓存 |
服务层 |
Python API |
Flask (或 FastAPI) |
5000/8000 |
会话管理、鉴权、LLM 调用、持久化 |
数据层 |
SQLite |
SQLAlchemy |
file |
用户表、会话表、消息表 |
外部依赖 |
OpenAI API |
HTTPS |
443 |
大模型推理 |
配置中心 |
dotenv |
.env 文件 |
N/A |
环境变量、API Key、DB 路径 |
数据库设计
物理模型(DDL)
CREATE TABLE users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
is_verified INTEGER DEFAULT 0,
verification_token TEXT UNIQUE,
verification_token_expiry DATETIME,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE email_verifications (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email TEXT UNIQUE NOT NULL,
token TEXT NOT NULL,
expiry DATETIME NOT NULL
);
CREATE TABLE chat_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
message TEXT NOT NULL,
response TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);
CREATE INDEX idx_users_username ON users(username);
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_chat_user_time ON chat_history(user_id, timestamp DESC);
字段 & 约束说明
表 |
字段 |
类型 |
业务含义 |
约束/索引 |
users |
username |
TEXT |
登录名 |
UNIQUE, 长度 3-20 |
|
password |
TEXT |
登录密码 |
现用明文→需改为哈希 |
|
email |
TEXT |
注册邮箱 |
UNIQUE, 必须 @qq.com |
|
is_verified |
INT |
邮箱验证开关 |
0/1 |
|
verification_token |
TEXT |
邮箱验证链接 token |
UNIQUE, 24h 过期 |
email_verifications |
token |
TEXT |
6 位验证码 |
5 分钟过期 |
chat_history |
message |
TEXT |
用户原始问题 |
不加密 |
|
response |
TEXT |
AI 返回内容 |
不加密 |
|
timestamp |
DATETIME |
创建时间 |
与 user_id 联合索引 |
ER 图
相关界面展示
聊天页面

登录页

注册页 
主要功能分析
用户会话管理(登录 / 注册 / 退出)
async function login(username, password) {
const res = await fetch('/auth/login', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ username, password })
});
const data = await res.json();
if (res.ok) {
localStorage.setItem('username', username);
location.href = '/index.html';
}
return data;
}
async function register(username, password, email, code) {
const res = await fetch('/auth/register', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ username, password, email, verification_code: code })
});
return res.json();
}
function logout() {
localStorage.removeItem('username');
location.href = '/index.html';
}
发送消息 + 流式打字机效果
let abortCtrl = null;
async function sendMessage(text) {
if (abortCtrl) abortCtrl.abort();
abortCtrl = new AbortController();
const payload = {
message: text,
username: localStorage.getItem('username')
};
const res = await fetch('/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload),
signal: abortCtrl.signal
});
const { response } = await res.json();
return response;
}
async function typeWriter(targetEl, text, speed = 30) {
targetEl.textContent = '';
for (let c of text) {
targetEl.textContent += c;
await new Promise(r => setTimeout(r, speed));
}
}
聊天记录持久化
async function saveChat(username, message, response) {
await fetch('/api/chat/save', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ username, message, response, timestamp: new Date().toISOString() })
});
}
async function loadChatHistory(username, limit = 20) {
const res = await fetch(`/api/chat/history?username=${encodeURIComponent(username)}&limit=${limit}`);
const { history } = await res.json();
return history;
}
邮箱验证码倒计时
let timer = null;
async function sendCode(email) {
const res = await fetch('/auth/send-verification-code', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ email })
});
if (!res.ok) return false;
let sec = 60;
const btn = document.getElementById('send-verification');
btn.disabled = true;
timer = setInterval(() => {
btn.textContent = `重新发送(${--sec}s)`;
if (sec === 0) {
clearInterval(timer);
btn.disabled = false;
btn.textContent = '发送验证码';
}
}, 1000);
return true;
}
会话中断(停止回答)
let abortCtrl = null;
let typingTimer = null;
function stopAnswer() {
if (abortCtrl) {
abortCtrl.abort();
abortCtrl = null;
}
if (typingTimer) {
clearTimeout(typingTimer);
typingTimer = null;
}
const typingEl = document.getElementById('typing');
if (typingEl) typingEl.remove();
}
登录态与 UI 联动(退出、历史侧栏)
const $ = sel => document.querySelector(sel);
function logout() {
localStorage.removeItem('username');
location.href = '/auth.html';
}
async function syncUIWithAuth() {
const username = localStorage.getItem('username');
const loginBtn = $('#authButton');
const historyBox = $('#historyList');
if (!username) {
loginBtn.textContent = '登录';
loginBtn.onclick = () => (location.href = '/auth.html');
historyBox.innerHTML = '<div class="no-history">请先登录查看历史记录</div>';
return;
}
loginBtn.textContent = '退出登录';
loginBtn.onclick = logout;
historyBox.innerHTML = '加载中...';
try {
const res = await fetch(`/api/chat/history?username=${encodeURIComponent(username)}`);
const { history } = await res.json();
renderHistory(history);
} catch (e) {
historyBox.innerHTML = '<div class="error">加载失败</div>';
}
}
function renderHistory(history) {
const box = $('#historyList');
box.innerHTML = '';
if (!history.length) {
box.innerHTML = '<div class="no-history">暂无聊天记录</div>';
return;
}
history.reverse().forEach(h => {
const item = document.createElement('div');
item.className = 'history-item';
item.innerHTML = `
<div class="history-time">${new Date(h.timestamp).toLocaleString()}</div>
<div class="history-msg">${h.message.slice(0, 30)}…</div>
`;
box.appendChild(item);
});
}
window.addEventListener('DOMContentLoaded', syncUIWithAuth);
Python 代码功能 & 结构提炼
切片 |
关键函数 / 类 |
主要作用 |
1. 配置加载 |
load_dotenv() |
读取 .env 注入全局变量 |
2. 数据库初始化 |
engine / SessionLocal / Base |
SQLite 建表;Base.metadata.create_all() |
3. 模型定义 |
User / EmailVerification / ChatHistory |
SQLAlchemy ORM 三张表 |
4. 用户注册 |
register() |
验证码校验 → 写 User |
5. 用户登录 |
login() |
支持用户名/邮箱 + 密码 |
6. 邮箱验证码 |
send_code_email() |
QQ SMTP 发送 6 位数字码 |
7. 聊天接口 |
chat() & chat_stream() |
单轮 & SSE 流式 |
8. 聊天记录 |
save_chat_history() / get_chat_history() |
CRUD |
9. AI 调用 |
get_ai_response() |
向 DashScope 发 HTTP 请求 |
10. 静态托管 |
app.mount("/static") |
FastAPI 托管 HTML/JS/CSS |
详细代码
from fastapi import FastAPI, Form, Depends, Request, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse,JSONResponse
import requests
import os
from dotenv import load_dotenv
from sqlalchemy import create_engine, Column, String, Integer, ForeignKey, Text, DateTime
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from datetime import datetime, timedelta
from pydantic import BaseModel, EmailStr
import smtplib
from email.mime.text import MIMETex
from email.header import Header
import uuid
from smtplib import SMTPAuthenticationError, SMTPConnectError
import secrets
from fastapi.responses import StreamingResponse,HTMLResponse, RedirectResponse
import asyncio
load_dotenv()
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
API_KEY = os.getenv("API_KEY", "sk-8c3c8f762df84d9eaa4bce5e2d97ead9")
API_URL = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"
class UserAuth(BaseModel):
username: str
password: str
class ChatSaveRequest(BaseModel):
username: str
message: str
response: str
timestamp: str
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True)
password = Column(String)
email = Column(String, unique=True, index=True)
is_verified = Column(Integer, default=0)
verification_token = Column(String, unique=True, nullable=True)
verification_token_expiry = Column(DateTime, nullable=True)
class EmailVerification(Base):
__tablename__ = "email_verifications"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
token = Column(String, unique=True, index=True)
expiry = Column(DateTime)
class ChatHistory(Base):
__tablename__ = "chat_history"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id"))
message = Column(Text)
response = Column(Text)
timestamp = Column(DateTime, default=datetime.utcnow)
Base.metadata.create_all(bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/", response_class=RedirectResponse)
async def root():
return RedirectResponse(url="/auth.html")
@app.get("/index.html")
async def read_index():
with open("static/index.html", "r", encoding="utf-8") as f:
return HTMLResponse(f.read())
@app.get("/auth.html", response_class=HTMLResponse)
async def read_auth():
with open("static/auth.html", "r", encoding="utf-8") as f:
return f.read()
@app.post("/chat")
async def chat(request: Request):
try:
data = await request.json()
user_message = data.get("message")
response_text = get_ai_response(user_message)
return {"response": response_text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/chat/stream")
async def chat_stream(request: Request):
data = await request.json()
user_message = data.get("message")
async def generate():
response_text = get_ai_response(user_message)
for word in response_text.split(" "):
yield word + " "
await asyncio.sleep(0.1)
return StreamingResponse(generate(), media_type="text/event-stream")
class EmailRequest(BaseModel):
email: str
@app.post("/auth/send-verification-code")
async def send_verification_code(request: EmailRequest, db: Session = Depends(get_db)):
email = request.email
if not email or not email.endswith("@qq.com"):
raise HTTPException(status_code=400, detail="请输入有效的QQ邮箱")
existing_user = db.query(User).filter(User.email == email).first()
if existing_user and existing_user.is_verified:
raise HTTPException(status_code=400, detail="该邮箱已注册并验证,请直接登录")
import random
verification_code = ''.join([str(random.randint(0, 9)) for _ in range(6)])
expiry = datetime.utcnow() + timedelta(minutes=5)
code_record = db.query(EmailVerification).filter(EmailVerification.email == email).first()
if code_record:
code_record.token = verification_code
code_record.expiry = expiry
else:
code_record = EmailVerification(email=email, token=verification_code, expiry=expiry)
db.add(code_record)
db.commit()
success, message = send_code_email(email, verification_code)
if not success:
raise HTTPException(status_code=500, detail=message)
return {"status": "success", "message": "验证码已发送至您的邮箱,5分钟内有效"}
@app.post("/auth/login")
async def login(user: UserAuth, db: Session = Depends(get_db)):
db_user = db.query(User).filter(
(User.username == user.username) | (User.email == user.username)
).first()
if not db_user or db_user.password != user.password:
raise HTTPException(status_code=400, detail="用户名/邮箱或密码错误")
if db_user.is_verified == 0:
raise HTTPException(status_code=400, detail="邮箱尚未验证,请先验证邮箱")
return {"status": "success", "message": "登录成功"}
@app.get("/verify-email")
async def verify_email(token: str, db: Session = Depends(get_db)):
user = db.query(User).filter(User.verification_token == token).first()
if not user:
return HTMLResponse("<h1>验证失败</h1><p>无效的验证链接</p><script>setTimeout(() => window.location.href = '/auth.html', 3000);</script>")
if datetime.utcnow() > user.verification_token_expiry:
return HTMLResponse("<h1>验证失败</h1><p>验证链接已过期,请重新发送验证邮件</p><script>setTimeout(() => window.location.href = '/auth.html', 3000);</script>")
user.is_verified = 1
user.verification_token = None
user.verification_token_expiry = None
db.commit()
return HTMLResponse("<h1>验证成功</h1><p>您的邮箱已验证,3秒后自动跳转到登录页面</p><script>setTimeout(() => window.location.href = '/auth.html?verified=true', 3000);</script>")
@app.get("/api/chat/history")
async def get_chat_history(
username: str,
limit: int = 20,
db: Session = Depends(get_db)
):
user = db.query(User).filter(
(User.username == username) | (User.email == username)
).first()
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
history = (
db.query(ChatHistory)
.filter(ChatHistory.user_id == user.id)
.order_by(ChatHistory.timestamp.desc())
.limit(limit)
.all()
)
return {
"history": [{
"message": item.message,
"response": item.response,
"timestamp": item.timestamp.isoformat()
} for item in reversed(history)]
}
@app.post("/api/chat/save")
async def save_chat_history(
request: ChatSaveRequest,
db: Session = Depends(get_db)
):
user = db.query(User).filter(
(User.username == request.username) | (User.email == request.username)
).first()
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
try:
timestamp = datetime.fromisoformat(request.timestamp)
except ValueError:
timestamp = datetime.utcnow()
chat_record = ChatHistory(
user_id=user.id,
message=request.message,
response=request.response,
timestamp=timestamp
)
db.add(chat_record)
db.commit()
db.refresh(chat_record)
return {"status": "success", "message": "聊天记录已保存"}
def get_ai_response(message: str):
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {API_KEY}"
}
payload = {
"model": "qwen-turbo",
"input": {
"prompt": message
},
"parameters": {
"temperature": 0.7,
"top_p": 0.9
}
}
response = requests.post(API_URL, json=payload, headers=headers)
response_data = response.json()
return response_data["output"]["text"] if "output" in response_data and "text" in response_data["output"] else "抱歉,无法获取AI响应。"
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=8082, reload=True)
QQ_SMTP_SERVER = 'smtp.qq.com'
QQ_SMTP_PORT = 465
QQ_EMAIL = os.getenv('QQ_EMAIL')
QQ_SMTP_PASSWORD = os.getenv('QQ_SMTP_PASSWORD', 'mmupqtegwnqsdaea')
def send_code_email(email: str, code: str):
subject = "注册验证码"
body = f"您的注册验证码是:{code}\n验证码有效期为5分钟,请尽快完成验证。"
msg = MIMEText(body, 'plain', 'utf-8')
msg['Subject'] = Header(subject, 'utf-8')
msg['From'] = f"<{QQ_EMAIL}>"
msg['To'] = email
try:
server = smtplib.SMTP_SSL(QQ_SMTP_SERVER, QQ_SMTP_PORT)
server.set_debuglevel(1)
server.login(QQ_EMAIL, QQ_SMTP_PASSWORD)
server.sendmail(QQ_EMAIL, [email], msg.as_string())
server.quit()
return True, "邮件发送成功"
except SMTPAuthenticationError:
return False, "邮箱认证失败,请检查SMTP配置"
except SMTPConnectError:
return False, "无法连接到SMTP服务器,请检查网络连接"
except Exception as e:
return False, f"发送失败:{str(e)}"
class RegisterRequest(BaseModel):
username: str
password: str
email: str
verification_code: str
@app.post("/auth/register")
async def register(request: RegisterRequest, db: Session = Depends(get_db)):
code_record = db.query(EmailVerification).filter(EmailVerification.email == request.email).first()
if not code_record or code_record.token != request.verification_code or datetime.utcnow() > code_record.expiry:
raise HTTPException(status_code=400, detail="验证码无效或已过期")
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="邮箱已被注册")
if len(request.username) < 3 or len(request.username) > 20:
raise HTTPException(status_code=400, detail="用户名长度必须在3-20个字符之间")
if len(request.password) < 6:
raise HTTPException(status_code=400, detail="密码长度不能少于6位")
def generate_verification_token():
"""Generate a secure random verification token"""
return secrets.token_urlsafe(32)
expiry = datetime.utcnow() + timedelta(hours=24)
db_user = User(
username=request.username,
password=request.password,
email=request.email,
is_verified=1
)
db.add(db_user)
db.commit()
db.refresh(db_user)
db.delete(code_record)
db.commit()
return {"status": "success", "message": "注册成功,请登录"}
@app.exception_handler(500)
async def internal_server_error_handler(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={"detail": "服务器内部错误,请稍后重试"}
)
总结与改进优化方案
一、安全加固
问题 |
现状 |
优化方案 |
明文密码 |
password 字段明文 |
passlib[bcrypt] 哈希 |
暴力破解 |
无登录限流 |
slowapi + IP 限流 5/min |
CSRF |
无防护 |
登录后下发 HttpOnly Cookie,前端每次带 X-CSRF-Token |
XSS |
用户输入直接渲染 |
前端 DOMPurify 清洗;后端 bleach 二次过滤 |
邮件验证码 |
固定 6 位数字 |
使用 secrets.token_urlsafe(16) ,有效期 5 min |
JWT |
无签名 |
登录成功后返回 JWT,RS256 签名,有效期 2 h,刷新令牌 7 d |
二、性能与并发
问题 |
现状 |
优化方案 |
SQLite 并发瓶颈 |
单线程写入 |
迁移 PostgreSQL + 读写分离 |
连接池 |
默认 5 条 |
SQLALCHEMY_POOL_SIZE=20 , max_overflow=30 |
AI 请求阻塞 |
同步 requests |
改为 httpx.AsyncClient 并设置 5 s 超时 |
缓存 |
无 |
Redis 缓存「用户-历史」列表 60 s |
静态文件 |
FastAPI 直出 |
Nginx + gzip + 缓存 1 d |
负载均衡 |
单机 |
Docker-Compose + Traefik + 2 实例 |
三、代码与架构
模块 |
现状 |
优化方案 |
单文件 main.py |
300 行 |
拆分 routers/ , services/ , schemas/ , models/ |
ORM |
原生 SQLAlchemy |
引入 SQLModel + 自动生成 CRUD |
配置 |
全局变量 |
Pydantic Settings 类,支持 .env + 12-Factor |
日志 |
无 |
structlog + loguru JSON 日志,接入 Loki |
测试 |
无 |
pytest-asyncio + pytest-cov > 90 % |
文档 |
FastAPI 默认 |
补充 OpenAPI example ,生成前端 SDK |
异常 |
500 统一返回 |
自定义 AppException ,统一响应格式 |
四、数据库演进
阶段 |
动作 |
当前 |
SQLite 单文件 |
阶段 1 |
PostgreSQL + 迁移脚本 Alembic |
阶段 2 |
分表:chat_history 按月分区 |
阶段 3 |
只读副本 + pgBouncer 连接池 |
阶段 4 |
读写分离 + 逻辑复制到 ClickHouse 做分析 |
五、前端优化
项目 |
现状 |
优化 |
框架 |
原生 JS |
迁移 Vite + Vue3(保留接口) |
打包 |
无 |
Rollup 分包,gzip < 200 KB |
流式 |
SSE |
支持 ReadableStream + AbortController |
主题 |
无 |
Dark/Light 切换,TailwindCSS |
国际化 |
无 |
vue-i18n 中英双语 |
PWA |
无 |
Service Worker 离线缓存首屏 |
源码下载
所有评论(0)