本文详细介绍如何使用 Vue3 + Flask + 智谱AI 构建一个生产级的知识库问答系统。涵盖前后端分离架构、Prompt 工程、JWT 鉴权等核心技术，附完整源码。

---

目录

1. 项目背景与架构设计
2. 核心技术栈分析
3. 系统架构与数据库设计
4. 前端实现：Vue3 + Pinia 状态管理
5. 后端实现：Flask RESTful API
6. AI 问答模块：Prompt 工程实践
7. 部署与性能优化
8. 常见问题与解决方案

---

一、项目背景与架构设计

1.1 问题陈述

大语言模型（LLM）虽然知识丰富，但存在两个核心问题：

1. 幻觉问题：模型在不知道答案时会自信地编造信息
2. 隐私问题：企业私有文档不能上传到公共 API

本项目通过 Prompt 工程 和 知识库注入 解决这两个问题，实现基于私有文档的精准问答。

1.2 系统架构

采用经典的前后端分离 B/S 架构：

┌─────────────────────────────────────────┐
│         浏览器（Vue3 SPA）               │
│  - 用户界面                              │
│  - 状态管理（Pinia）                     │
│  - 路由管理（Vue Router）                │
└──────────────┬──────────────────────────┘
               │ HTTP/JSON
┌──────────────▼──────────────────────────┐
│      Flask RESTful API 服务器            │
│  - 用户认证（JWT）                       │
│  - 知识库管理                            │
│  - 问答业务逻辑                          │
└──────┬───────────────────┬──────────────┘
       │                   │
┌──────▼──────┐    ┌───────▼────────┐
│  SQLite DB  │    │  智谱 AI API   │
│  - 用户表   │    │  - GLM-4-Flash │
│  - 知识库表 │    │  - Prompt 工程 │
│  - 历史表   │    │  - 流式响应    │
└─────────────┘    └────────────────┘

---

1.3 系统截图

二、核心技术栈分析

2.1 前端技术栈

技术	版本	选择理由
Vue 3	3.4	Composition API 更灵活，性能更好
Vite	5.3	开发时热更新快，构建速度快
Pinia	2.1	Vue3 官方推荐，API 简洁
Vue Router	4.3	原生路由守卫，支持动态路由
Axios	1.7	请求拦截器，统一错误处理

关键特性：

• Composition API 组织代码更清晰
• 响应式系统自动追踪依赖
• 虚拟 DOM 高效更新

2.2 后端技术栈

技术	版本	用途
Flask	3.0	轻量级 Web 框架
SQLAlchemy	2.0	ORM，数据库操作
Flask-JWT-Extended	4.6	JWT 无状态鉴权
ZhipuAI SDK	2.1	大模型 API 调用

为什么选 Flask 而不是 Django：

• Django 过重，包含不需要的功能（模板、Admin）
• Flask 轻量，按需扩展
• 前后端分离场景下，只需要 API 层

2.3 AI 模型选择

使用智谱 AI 的 GLM-4-Flash 模型：

优势：
- 免费额度充足（注册即送）
- 响应速度快（适合实时问答）
- 支持 Prompt 工程
- 中文理解能力强

成本对比：
- OpenAI GPT-4：$0.03/1K tokens
- 智谱 GLM-4-Flash：免费（有额度限制）
- 本地 Llama 2：需要 GPU，部署复杂

---

三、系统架构与数据库设计

3.1 数据库 E-R 图

┌─────────────┐         ┌──────────────────┐
│    users    │         │ knowledge_bases  │
├─────────────┤         ├──────────────────┤
│ id (PK)     │────┐    │ id (PK)          │
│ username    │    │    │ user_id (FK)     │
│ password    │    │    │ name             │
│ created_at  │    │    │ file_path        │
└─────────────┘    │    │ char_count       │
                   │    │ created_at       │
                   │    └──────────────────┘
                   │           │
                   │           │ (1:N)
                   │           │
                   │    ┌──────▼──────────┐
                   │    │ chat_histories  │
                   │    ├─────────────────┤
                   │    │ id (PK)         │
                   │    │ user_id (FK)    │
                   │    │ kb_id (FK)      │
                   │    │ question        │
                   │    │ answer          │
                   │    │ tokens_used     │
                   │    │ created_at      │
                   │    └─────────────────┘
                   │
                   └─────────────────────┘

3.2 表结构详解

users 表：存储用户信息

CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    username VARCHAR(64) UNIQUE NOT NULL,
    password_hash VARCHAR(256) NOT NULL,  -- Werkzeug 加盐哈希
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

knowledge_bases 表：存储知识库元信息

CREATE TABLE knowledge_bases (
    id INTEGER PRIMARY KEY,
    user_id INTEGER NOT NULL FOREIGN KEY,
    name VARCHAR(128) NOT NULL,           -- 用户友好名称
    filename VARCHAR(256) NOT NULL,       -- UUID 化文件名
    file_path VARCHAR(512) NOT NULL,      -- 完整路径
    file_size INTEGER,                    -- 字节数
    char_count INTEGER,                   -- 字符数
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

chat_histories 表：存储问答记录

CREATE TABLE chat_histories (
    id INTEGER PRIMARY KEY,
    user_id INTEGER NOT NULL FOREIGN KEY,
    kb_id INTEGER NOT NULL FOREIGN KEY,
    question TEXT NOT NULL,
    answer TEXT NOT NULL,
    tokens_used INTEGER,                  -- 消耗的 token 数
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

3.3 索引优化

# SQLAlchemy 中定义索引
class ChatHistory(db.Model):
    __tablename__ = 'chat_histories'
    
    # 复合索引：加速按用户和知识库的查询
    __table_args__ = (
        Index('idx_user_kb', 'user_id', 'kb_id'),
        Index('idx_created_at', 'created_at'),
    )

---

四、前端实现：Vue3 + Pinia 状态管理

4.1 Composition API 最佳实践

// src/views/ChatView.vue
import { ref, computed, watch, nextTick } from 'vue'
import { useRoute } from 'vue-router'
import { useAuthStore } from '@/stores/auth'
import { sendChat } from '@/api/chat'

export default {
  setup() {
    const route = useRoute()
    const auth = useAuthStore()
    
    // 响应式状态
    const messages = ref([])
    const inputText = ref('')
    const thinking = ref(false)
    const selectedKbId = ref('')
    
    // 计算属性：判断是否可以发送
    const canSend = computed(() =>
      selectedKbId.value &&
      inputText.value.trim().length > 0 &&
      inputText.value.length <= 1000 &&
      !thinking.value
    )
    
    // 方法：发送消息
    async function handleSend() {
      if (!canSend.value) return
      
      const question = inputText.value.trim()
      inputText.value = ''
      
      // 添加用户消息到 UI
      messages.value.push({
        id: Date.now(),
        role: 'user',
        content: question,
        time: new Date(),
      })
      
      scrollToBottom()
      thinking.value = true
      
      try {
        // 调用后端 API
        const res = await sendChat(selectedKbId.value, question)
        
        // 添加 AI 回答
        messages.value.push({
          id: Date.now() + 1,
          role: 'ai',
          content: res.data.answer,
          tokens: res.data.tokens_used,
          time: new Date(),
        })
      } catch (e) {
        // 错误处理
        messages.value.push({
          id: Date.now() + 1,
          role: 'ai',
          content: `错误：${e.message}`,
          time: new Date(),
        })
      } finally {
        thinking.value = false
        await nextTick()
        scrollToBottom()
      }
    }
    
    // 监听知识库切换，清空对话
    watch(selectedKbId, () => {
      messages.value = []
    })
    
    return {
      messages,
      inputText,
      thinking,
      selectedKbId,
      canSend,
      handleSend,
    }
  }
}

4.2 Pinia 状态管理

// src/stores/auth.js
import { defineStore } from 'pinia'
import { ref, computed } from 'vue'
import { login as loginApi } from '@/api/auth'

export const useAuthStore = defineStore('auth', () => {
  // 状态
  const token = ref(localStorage.getItem('token') || '')
  const user = ref(JSON.parse(localStorage.getItem('user') || 'null'))
  
  // 计算属性
  const isLoggedIn = computed(() => !!token.value)
  
  // 方法
  async function login(username, password) {
    const res = await loginApi(username, password)
    token.value = res.data.token
    user.value = res.data.user
    
    // 持久化到 localStorage
    localStorage.setItem('token', res.data.token)
    localStorage.setItem('user', JSON.stringify(res.data.user))
  }
  
  function logout() {
    token.value = ''
    user.value = null
    localStorage.removeItem('token')
    localStorage.removeItem('user')
  }
  
  return { token, user, isLoggedIn, login, logout }
})

4.3 Axios 拦截器

// src/api/request.js
import axios from 'axios'

const request = axios.create({
  baseURL: '/api',
  timeout: 60000,
})

// 请求拦截：自动注入 JWT Token
request.interceptors.request.use((config) => {
  const token = localStorage.getItem('token')
  if (token) {
    config.headers.Authorization = `Bearer ${token}`
  }
  return config
})

// 响应拦截：统一错误处理
request.interceptors.response.use(
  (response) => response.data,
  (error) => {
    // 401 未授权，跳转登录
    if (error.response?.status === 401) {
      localStorage.removeItem('token')
      window.location.href = '/login'
    }
    
    const msg = error.response?.data?.msg || error.message
    return Promise.reject(new Error(msg))
  }
)

export default request

4.4 路由守卫

// src/router/index.js
import { createRouter, createWebHistory } from 'vue-router'
import { useAuthStore } from '@/stores/auth'

const router = createRouter({
  history: createWebHistory(),
  routes: [
    {
      path: '/login',
      component: () => import('@/views/LoginView.vue'),
      meta: { public: true },  // 公开路由
    },
    {
      path: '/',
      component: () => import('@/views/LayoutView.vue'),
      children: [
        {
          path: 'kb',
          component: () => import('@/views/KbView.vue'),
        },
        {
          path: 'chat',
          component: () => import('@/views/ChatView.vue'),
        },
      ],
    },
  ],
})

// 路由守卫：检查登录状态
router.beforeEach((to) => {
  const auth = useAuthStore()
  
  // 公开路由直接通过
  if (to.meta.public) return true
  
  // 需要登录的路由，检查 token
  if (!auth.isLoggedIn) {
    return { name: 'Login' }
  }
})

export default router

---

五、后端实现：Flask RESTful API

5.1 JWT 鉴权机制

# app.py
from flask_jwt_extended import JWTManager, create_access_token, jwt_required

app.config['JWT_SECRET_KEY'] = os.getenv('JWT_SECRET_KEY')
jwt = JWTManager(app)

@app.route('/api/auth/login', methods=['POST'])
def login():
    data = request.get_json()
    username = data.get('username')
    password = data.get('password')
    
    # 查询用户
    user = User.query.filter_by(username=username).first()
    
    # 验证密码（Werkzeug 加盐哈希）
    if not user or not check_password_hash(user.password_hash, password):
        return {'code': 401, 'msg': '用户名或密码错误'}, 401
    
    # 生成 JWT Token
    token = create_access_token(identity=str(user.id))
    
    return {
        'code': 200,
        'msg': '登录成功',
        'data': {
            'token': token,
            'user': user.to_dict(),
        }
    }

# 保护的接口
@app.route('/api/kb', methods=['GET'])
@jwt_required()
def list_kb():
    user_id = get_jwt_identity()
    kbs = KnowledgeBase.query.filter_by(user_id=user_id).all()
    return {
        'code': 200,
        'data': [kb.to_dict() for kb in kbs]
    }

5.2 文件上传安全处理

from werkzeug.utils import secure_filename
import uuid

@app.route('/api/kb/upload', methods=['POST'])
@jwt_required()
def upload_kb():
    user_id = get_jwt_identity()
    
    if 'file' not in request.files:
        return {'code': 400, 'msg': '未找到文件'}, 400
    
    file = request.files['file']
    
    # 1. 文件类型校验
    if not file.filename.endswith('.txt'):
        return {'code': 400, 'msg': '只支持 .txt 格式'}, 400
    
    # 2. 文件名安全处理
    safe_name = secure_filename(file.filename)
    
    # 3. UUID 重命名，防止覆盖和路径注入
    unique_filename = f"{uuid.uuid4().hex}_{safe_name}"
    file_path = os.path.join(UPLOAD_FOLDER, unique_filename)
    
    # 4. 保存文件
    file.save(file_path)
    
    # 5. 统计文件信息
    file_size = os.path.getsize(file_path)
    with open(file_path, 'r', encoding='utf-8') as f:
        char_count = len(f.read())
    
    # 6. 保存到数据库
    kb = KnowledgeBase(
        user_id=user_id,
        name=os.path.splitext(safe_name)[0],
        filename=unique_filename,
        file_path=file_path,
        file_size=file_size,
        char_count=char_count,
    )
    db.session.add(kb)
    db.session.commit()
    
    return {
        'code': 201,
        'msg': '上传成功',
        'data': kb.to_dict()
    }

5.3 统一响应格式

def success(data=None, msg='success', code=200):
    """统一成功响应"""
    resp = {'code': code, 'msg': msg}
    if data is not None:
        resp['data'] = data
    return jsonify(resp), code

def fail(msg='error', code=400):
    """统一失败响应"""
    return jsonify({'code': code, 'msg': msg}), code

# 使用示例
@app.route('/api/kb/<int:kb_id>', methods=['DELETE'])
@jwt_required()
def delete_kb(kb_id):
    user_id = get_jwt_identity()
    kb = KnowledgeBase.query.filter_by(id=kb_id, user_id=user_id).first()
    
    if not kb:
        return fail('知识库不存在', 404)
    
    # 删除文件
    if os.path.exists(kb.file_path):
        os.remove(kb.file_path)
    
    # 删除数据库记录
    db.session.delete(kb)
    db.session.commit()
    
    return success(msg='删除成功')

---

六、AI 问答模块：Prompt 工程实践

6.1 Prompt 工程核心

这是系统的灵魂所在。通过精心设计 System Prompt，约束模型的行为。

# ai_service.py
def build_system_prompt(kb_content: str) -> str:
    """构建系统 Prompt，将知识库内容注入"""
    return f"""你是一个专业的知识库问答助手。
请严格根据以下知识库内容回答用户的问题。

规则：
1. 只根据知识库内容作答，不要编造知识库中没有的信息。
2. 如果知识库中没有相关内容，请明确告知用户"知识库中未找到相关信息"。
3. 回答要简洁、准确、有条理。
4. 使用中文回答。

========== 知识库内容 ==========
{kb_content}
================================
"""

6.2 处理流程

def ask_question(file_path: str, question: str) -> dict:
    """
    完整的问答流程
    """
    try:
        # 1. 读取知识库文件
        kb_content = load_knowledge_base(file_path)
        
        # 2. 构建 Prompt
        system_prompt = build_system_prompt(kb_content)
        
        # 3. 调用大模型 API
        response = _client.chat.completions.create(
            model='glm-4-flash',
            messages=[
                {'role': 'system', 'content': system_prompt},
                {'role': 'user', 'content': question},
            ],
            temperature=0.3,      # 低温度，保证准确性
            max_tokens=2048,
        )
        
        # 4. 提取回答和 token 消耗
        answer = response.choices[0].message.content
        tokens_used = response.usage.total_tokens
        
        # 5. 返回结果
        return {
            'answer': answer,
            'tokens_used': tokens_used,
            'success': True,
        }
        
    except Exception as e:
        return {
            'answer': '',
            'tokens_used': 0,
            'success': False,
            'error': str(e),
        }

6.3 编码兼容性处理

def load_knowledge_base(file_path: str) -> str:
    """读取知识库，自动处理编码"""
    try:
        # 优先尝试 UTF-8
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
    except UnicodeDecodeError:
        # 降级到 GBK（兼容中文 Windows 系统）
        with open(file_path, 'r', encoding='gbk', errors='replace') as f:
            content = f.read()
    
    # 超长内容截断
    MAX_CONTEXT_CHARS = 12000
    if len(content) > MAX_CONTEXT_CHARS:
        content = content[:MAX_CONTEXT_CHARS] + '\n\n[...内容过长，已截断...]'
    
    return content

6.4 Prompt 工程最佳实践

设计原则：
1. 角色设定：明确告诉模型它是什么
   "你是一个专业的知识库问答助手"

2. 内容注入：将知识库内容嵌入 Prompt
   "根据以下知识库内容回答..."

3. 规则约束：明确要求和禁止
   "只根据知识库内容作答"
   "不要编造信息"

4. 兜底策略：处理知识库无相关内容的情况
   "如果知识库中没有相关内容，请明确告知用户"

5. 温度参数：控制输出的随机性
   temperature=0.3  低温度，保证准确性
   temperature=0.7  高温度，增加创意性

---

七、部署与性能优化

7.1 生产环境部署

# 后端部署（使用 Gunicorn）
pip install gunicorn
gunicorn -w 4 -b 0.0.0.0:5001 app:app

# 前端构建
cd kb-qa-frontend
npm run build
# 将 dist/ 目录部署到 Nginx 或 CDN

7.2 Nginx 反向代理配置

server {
    listen 80;
    server_name your-domain.com;
    
    # 前端静态文件
    location / {
        root /path/to/kb-qa-frontend/dist;
        try_files $uri $uri/ /index.html;
    }
    
    # 后端 API 代理
    location /api/ {
        proxy_pass http://127.0.0.1:5001;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        # 长连接支持（AI 问答可能耗时）
        proxy_connect_timeout 60s;
        proxy_send_timeout 60s;
        proxy_read_timeout 60s;
    }
}

7.3 性能优化建议

# 1. 数据库查询优化
# 使用 eager loading 避免 N+1 查询
kbs = KnowledgeBase.query.options(
    joinedload(KnowledgeBase.owner)
).all()

# 2. 缓存知识库内容
from functools import lru_cache

@lru_cache(maxsize=128)
def load_knowledge_base(file_path: str) -> str:
    """缓存知识库内容，避免重复读取"""
    # ...

# 3. 异步处理长耗时操作
from celery import Celery

@app.route('/api/chat', methods=['POST'])
def chat():
    # 立即返回，后台处理
    task = process_chat.delay(kb_id, question)
    return {'task_id': task.id}

# 4. 分页查询历史记录
@app.route('/api/chat/history', methods=['GET'])
def get_history():
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    
    pagination = ChatHistory.query.paginate(
        page=page,
        per_page=per_page,
        error_out=False
    )
    
    return {
        'items': [h.to_dict() for h in pagination.items],
        'total': pagination.total,
        'pages': pagination.pages,
    }

---

八、常见问题与解决方案

8.1 大模型幻觉问题

问题：模型有时会编造知识库中没有的信息。

解决方案：

1. 降低 temperature 参数（0.3 而不是 0.7）
2. 在 Prompt 中明确禁止编造
3. 引入向量数据库做语义检索（RAG 方案）

# 改进的 Prompt
system_prompt = """
你是一个严谨的知识库问答助手。

重要规则：
- 只能根据提供的知识库内容回答
- 如果知识库中没有相关信息，必须回答"知识库中未找到相关信息"
- 不允许推测、猜测或编造任何信息
- 即使你知道答案，如果知识库中没有，也不能回答
"""

8.2 超长文档处理

问题：知识库文档超过模型上下文窗口（通常 4K-8K tokens）。

解决方案：

# 方案 1：截断（当前实现）
MAX_CONTEXT_CHARS = 12000
if len(content) > MAX_CONTEXT_CHARS:
    content = content[:MAX_CONTEXT_CHARS]

# 方案 2：分块 + 向量检索（推荐）
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma

splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_text(content)

# 向量化存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_texts(chunks, embeddings)

# 检索相关片段
relevant_chunks = vectorstore.similarity_search(question, k=3)

8.3 并发控制

问题：多个用户同时提问，API 调用频率限制。

解决方案：

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    app=app,
    key_func=get_remote_address,
    default_limits=['200 per day', '50 per hour']
)

@app.route('/api/chat', methods=['POST'])
@limiter.limit('10 per minute')  # 每分钟最多 10 个请求
@jwt_required()
def chat():
    # ...

8.4 Token 成本控制

问题：大模型 API 按 token 计费，需要控制成本。

解决方案：

# 1. 监控 token 消耗
@app.route('/api/chat', methods=['POST'])
def chat():
    result = ask_question(kb.file_path, question)
    tokens_used = result['tokens_used']
    
    # 保存到数据库，便于统计
    history = ChatHistory(
        tokens_used=tokens_used,
        # ...
    )
    db.session.add(history)
    db.session.commit()

# 2. 设置用户配额
class User(db.Model):
    monthly_token_quota = db.Column(db.Integer, default=100000)
    tokens_used_this_month = db.Column(db.Integer, default=0)

# 3. 检查配额
if user.tokens_used_this_month + tokens_used > user.monthly_token_quota:
    return fail('本月 token 配额已用尽', 429)

---

九、总结与展望

9.1 项目亮点

1. Prompt 工程：通过精心设计 System Prompt，有效解决大模型幻觉问题
2. 前后端分离：清晰的架构，易于维护和扩展
3. 安全设计：JWT 鉴权、密码哈希、文件名 UUID 化
4. 工程规范：统一的 API 响应格式、完整的错误处理

9.2 改进方向

1. 向量数据库：引入 Chroma/Faiss，实现 RAG 方案，支持超长文档
2. 多模型支持：支持 PDF、Word、Markdown 等格式
3. 多轮对话：保存对话历史，支持连续问答
4. 容器化部署：Docker + Kubernetes，支持云端部署
5. 监控告警：集成 Prometheus + Grafana，监控系统性能

9.3 学习资源

• Vue 3 官方文档：https://cn.vuejs.org
• Flask 官方文档：https://flask.palletsprojects.com
• SQLAlchemy 文档：https://www.sqlalchemy.org
• 智谱 AI 文档：https://open.bigmodel.cn/docs
• Prompt 工程最佳实践：https://platform.openai.com/docs/guides/prompt-engineering

---

附录：快速开始

环境要求

• Python 3.11+
• Node.js 18+
• 智谱 AI API Key（免费申请）

后端启动

cd kb-qa-backend
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
cp .env.example .env  # 填入 ZHIPUAI_API_KEY
python app.py

前端启动

cd kb-qa-frontend
npm install
npm run dev

默认账号

• 用户名：admin
• 密码：admin123

---

作者：沙蒿同学
发布时间：2026 年
更新时间：持续更新中

如有问题，欢迎在评论区讨论！