基于Node.js + MySQL + DeepSeek构建RAG用户检索系统

前言
本文将手把手教你构建一个基于RAG（检索增强生成）的用户信息检索系统。该系统使用Node.js作为后端，MySQL存储用户数据，DeepSeek提供AI能力，实现智能的用户信息查询。

技术栈：

Node.js + Express

MySQL数据库

DeepSeek API（向量嵌入 + 对话生成）

向量检索 + 关键词检索混合策略

最终效果： 用户通过HTTP接口提问，系统自动从数据库检索相关用户信息，结合DeepSeek生成智能回答。

一、项目概述

1.1 什么是RAG？

RAG（Retrieval-Augmented Generation）是一种结合信息检索和文本生成的技术（说白了就是增强检索，只不过大多数数据分析由ai完成，然后你只要规定ai怎么返回数据即可，主要是我们要规定ai返回的格式）。在本项目中：
检索阶段： 从MySQL数据库中查找与用户问题相关的用户信息
生成阶段： 将检索到的信息作为上下文，让DeepSeek生成自然语言回答
重点在4.6你只要会写接口直接看4.6即可
1.2 系统架构

用户提问 → HTTP接口 → 查询向量化 → 混合检索（向量+关键词）
    ↓
MySQL用户表 ← 检索结果 ← DeepSeek生成回答 → 返回给用户

二、环境准备

2.1 所需环境

Node.js 
MySQL 
DeepSeek API Key（官网申请）

2.2 创建项目

mkdir rag-user-system
cd rag-user-system
npm init -y

2.3 安装依赖

npm install express mysql2 dotenv openai cors helmet morgan express-rate-limit
npm install nodemon -D

三、项目结构

rag-user-system/
├── .env                    # 环境变量配置
├── package.json
├── src/
│   ├── app.js             # 应用入口
│   ├── config/
│   │   └── database.js    # 数据库配置
│   ├── services/
│   │   ├── userService.js # 用户数据服务
│   │   ├── embeddingService.js # 向量嵌入服务
│   │   ├── vectorStoreService.js # 向量存储服务
│   │   └── ragService.js  # RAG核心服务
│   ├── controllers/
│   │   └── queryController.js # 查询控制器
│   ├── routes/
│   │   └── queryRoutes.js # 路由定义
│   └── middleware/
│       └── errorHandler.js # 错误处理
└── test/
    └── api-test.http      # API测试文件

四、核心代码实现

4.1 环境配置 (.env)

# 服务器配置
PORT=3000
NODE_ENV=development

# DeepSeek API配置
DEEPSEEK_API_KEY=sk-your-api-key-here
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1
DEEPSEEK_MODEL=deepseek-chat

# MySQL数据库配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=your_password
DB_NAME=rag_demo
DB_CONNECTION_LIMIT=10

4.2 数据库配置 (src/config/database.js)

const mysql = require('mysql2/promise');
require('dotenv').config();

// 创建连接池
const pool = mysql.createPool({
  host: process.env.DB_HOST || 'localhost',
  port: parseInt(process.env.DB_PORT) || 3306,
  user: process.env.DB_USER || 'root',
  password: process.env.DB_PASSWORD || '',
  database: process.env.DB_NAME || 'rag_demo',
  waitForConnections: true,
  connectionLimit: parseInt(process.env.DB_CONNECTION_LIMIT) || 10,
  queueLimit: 0,
  charset: 'utf8mb4'
});

// 测试数据库连接
async function testConnection() {
  try {
    const connection = await pool.getConnection();
    console.log('✅ 数据库连接成功');
    connection.release();
    return true;
  } catch (error) {
    console.error('❌ 数据库连接失败:', error.message);
    return false;
  }
}

// 初始化数据库表
async function initializeDatabase() {
  const connection = await pool.getConnection();
  try {
    // 创建用户表
    await connection.execute(`
      CREATE TABLE IF NOT EXISTS users (
        id INT PRIMARY KEY AUTO_INCREMENT,
        name VARCHAR(100) NOT NULL COMMENT '姓名',
        email VARCHAR(255) UNIQUE NOT NULL COMMENT '邮箱',
        phone VARCHAR(20) COMMENT '电话',
        department VARCHAR(100) COMMENT '部门',
        position VARCHAR(100) COMMENT '职位',
        bio TEXT COMMENT '个人简介',
        skills TEXT COMMENT '技能标签',
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
        INDEX idx_department (department),
        INDEX idx_position (position),
        FULLTEXT INDEX ft_search (name, email, department, position, bio, skills)
      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户信息表'
    `);

    // 创建向量存储表
    await connection.execute(`
      CREATE TABLE IF NOT EXISTS user_embeddings (
        id INT PRIMARY KEY AUTO_INCREMENT,
        user_id INT NOT NULL COMMENT '用户ID',
        content TEXT NOT NULL COMMENT '原始文本内容',
        embedding LONGTEXT NOT NULL COMMENT '向量数据(JSON格式)',
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
        INDEX idx_user_id (user_id)
      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户向量存储表'
    `);

    console.log('✅ 数据库表初始化完成');
  } catch (error) {
    console.error('❌ 数据库初始化失败:', error);
    throw error;
  } finally {
    connection.release();
  }
}

module.exports = {
  pool,
  testConnection,
  initializeDatabase
};

4.3 用户数据服务 (src/services/userService.js)

```javascript
const { pool } = require('../config/database');

class UserService {
  /**
   * 获取所有用户
   */
  async getAllUsers() {
    const [rows] = await pool.execute(
      'SELECT id, name, email, phone, department, position, bio, skills FROM users ORDER BY id'
    );
    return rows;
  }

  /**
   * 根据ID获取用户
   */
  async getUserById(id) {
    const [rows] = await pool.execute(
      'SELECT * FROM users WHERE id = ?',
      [id]
    );
    return rows[0] || null;
  }

  /**
   * 批量插入用户
   */
  async insertUsers(users) {
    const connection = await pool.getConnection();
    try {
      await connection.beginTransaction();
      
      const sql = `INSERT INTO users (name, email, phone, department, position, bio, skills) 
                   VALUES (?, ?, ?, ?, ?, ?, ?)`;
      
      for (const user of users) {
        await connection.execute(sql, [
          user.name,
          user.email,
          user.phone || null,
          user.department || null,
          user.position || null,
          user.bio || null,
          user.skills || null
        ]);
      }
      
      await connection.commit();
      console.log(`✅ 成功插入 ${users.length} 条用户数据`);
    } catch (error) {
      await connection.rollback();
      throw error;
    } finally {
      connection.release();
    }
  }

  /**
   * 获取测试数据
   */
  getSeedData() {
    return [
      {
        name: '张三',
        email: 'zhangsan@company.com',
        phone: '13800138001',
        department: '技术部',
        position: '高级工程师',
        bio: '10年全栈开发经验，精通前后端技术栈，曾主导多个大型项目架构设计',
        skills: 'Node.js, React, Vue.js, MySQL, MongoDB, Docker, Kubernetes, AWS'
      },
      {
        name: '李四',
        email: 'lisi@company.com',
        phone: '13800138002',
        department: '技术部',
        position: '前端工程师',
        bio: '5年前端开发经验，专注React生态，对性能优化有深入研究',
        skills: 'React, TypeScript, Next.js, Webpack, CSS3, Canvas, WebGL'
      },
      {
        name: '王五',
        email: 'wangwu@company.com',
        phone: '13800138003',
        department: '产品部',
        position: '高级产品经理',
        bio: '8年产品经验，擅长B端SaaS产品设计，有从0到1的产品孵化经验',
        skills: '产品设计, 数据分析, 用户研究, Axure, Figma, SQL, Python'
      },
      {
        name: '赵六',
        email: 'zhaoliu@company.com',
        phone: '13800138004',
        department: '技术部',
        position: '后端工程师',
        bio: '6年Java后端开发经验，擅长微服务架构和分布式系统设计',
        skills: 'Java, Spring Boot, MySQL, Redis, RabbitMQ, Elasticsearch, Docker'
      },
      {
        name: '孙七',
        email: 'sunqi@company.com',
        phone: '13800138005',
        department: '设计部',
        position: 'UI设计师',
        bio: '4年UI设计经验，擅长B端产品设计，对用户体验有独到见解',
        skills: 'UI设计, UX设计, Sketch, Figma, Photoshop, Illustrator'
      },
      {
        name: '周八',
        email: 'zhouba@company.com',
        phone: '13800138006',
        department: '产品部',
        position: '产品经理',
        bio: '3年产品经验，专注用户增长和数据分析',
        skills: '产品策划, A/B测试, 数据分析, SQL, Google Analytics'
      },
      {
        name: '吴九',
        email: 'wujiu@company.com',
        phone: '13800138007',
        department: '技术部',
        position: 'DevOps工程师',
        bio: '7年运维开发经验，精通CI/CD流程和云原生技术',
        skills: 'Linux, Docker, Kubernetes, Jenkins, Terraform, AWS, Prometheus'
      },
      {
        name: '郑十',
        email: 'zhengshi@company.com',
        phone: '13800138008',
        department: '市场部',
        position: '市场总监',
        bio: '10年市场营销经验，擅长数字营销和品牌建设',
        skills: '数字营销, 品牌策划, SEO, SEM, 数据分析, 团队管理'
      }
    ];
  }

  /**
   * 将用户信息转换为可检索的文本
   */
  userToText(user) {
    return [
      `姓名: ${user.name}`,
      `部门: ${user.department || '未分配'}`,
      `职位: ${user.position || '未设置'}`,
      `技能: ${user.skills || '无'}`,
      `简介: ${user.bio || '无'}`,
      `联系方式: ${user.email} ${user.phone || ''}`
    ].join(' | ');
  }

  /**
   * 全文搜索用户
   */
  async fullTextSearch(keyword, limit = 5) {
    const [rows] = await pool.execute(
      `SELECT *, MATCH(name, email, department, position, bio, skills) 
       AGAINST(? IN NATURAL LANGUAGE MODE) as relevance
       FROM users 
       WHERE MATCH(name, email, department, position, bio, skills) 
       AGAINST(? IN NATURAL LANGUAGE MODE)
       ORDER BY relevance DESC
       LIMIT ?`,
      [keyword, keyword, limit]
    );
    return rows;
  }
}

module.exports = new UserService();

4.4 向量嵌入服务 (src/services/embeddingService.js)

const OpenAI = require('openai');
require('dotenv').config();

class EmbeddingService {
  constructor() {
    this.openai = new OpenAI({
      apiKey: process.env.DEEPSEEK_API_KEY,
      baseURL: process.env.DEEPSEEK_BASE_URL || 'https://api.deepseek.com/v1',
    });
  }

  /**
   * 获取单个文本的向量嵌入
   */
  async getEmbedding(text) {
    try {
      const response = await this.openai.embeddings.create({
        model: process.env.DEEPSEEK_MODEL || 'deepseek-chat',
        input: text,
      });
      return response.data[0].embedding;
    } catch (error) {
      console.error('获取嵌入向量失败:', error.message);
      throw error;
    }
  }

  /**
   * 批量获取文本的向量嵌入
   */
  async getEmbeddings(texts) {
    try {
      const response = await this.openai.embeddings.create({
        model: process.env.DEEPSEEK_MODEL || 'deepseek-chat',
        input: texts,
      });
      return response.data.map(item => item.embedding);
    } catch (error) {
      console.error('批量获取嵌入向量失败:', error.message);
      throw error;
    }
  }

  /**
   * 计算余弦相似度
   */
  cosineSimilarity(vecA, vecB) {
    if (vecA.length !== vecB.length) {
      throw new Error('向量维度不匹配');
    }

    let dotProduct = 0;
    let normA = 0;
    let normB = 0;

    for (let i = 0; i < vecA.length; i++) {
      dotProduct += vecA[i] * vecB[i];
      normA += vecA[i] * vecA[i];
      normB += vecB[i] * vecB[i];
    }

    if (normA === 0 || normB === 0) return 0;
    
    return dotProduct / (Math.sqrt(normA) * Math.sqrt(normB));
  }
}

module.exports = new EmbeddingService();

4.5 向量存储服务 (src/services/vectorStoreService.js)

const { pool } = require('../config/database');
const embeddingService = require('./embeddingService');
const userService = require('./userService');

class VectorStoreService {
  /**
   * 为用户生成并存储嵌入向量
   */
  async storeUserEmbedding(userId, userText) {
    const embedding = await embeddingService.getEmbedding(userText);
    
    const connection = await pool.getConnection();
    try {
      await connection.execute(
        'INSERT INTO user_embeddings (user_id, content, embedding) VALUES (?, ?, ?)',
        [userId, userText, JSON.stringify(embedding)]
      );
      console.log(`✅ 用户ID ${userId} 的向量已存储`);
    } finally {
      connection.release();
    }
  }

  /**
   * 为所有用户生成嵌入向量
   */
  async embedAllUsers() {
    const users = await userService.getAllUsers();
    
    for (const user of users) {
      // 检查是否已有嵌入向量
      const connection = await pool.getConnection();
      try {
        const [existing] = await connection.execute(
          'SELECT id FROM user_embeddings WHERE user_id = ?',
          [user.id]
        );
        
        if (existing.length === 0) {
          const userText = userService.userToText(user);
          await this.storeUserEmbedding(user.id, userText);
        } else {
          console.log(`⏭️ 用户ID ${user.id} 的向量已存在，跳过`);
        }
      } finally {
        connection.release();
      }
    }
    
    console.log(`✅ 所有用户向量处理完成`);
  }

  /**
   * 向量检索相似用户
   */
  async vectorSearch(query, topK = 5) {
    // 获取查询向量
    const queryEmbedding = await embeddingService.getEmbedding(query);
    
    // 获取所有用户向量
    const connection = await pool.getConnection();
    try {
      const [embeddings] = await connection.execute(
        `SELECT ue.id as embedding_id, ue.user_id, ue.content, ue.embedding,
                u.name, u.email, u.phone, u.department, u.position, u.bio, u.skills
         FROM user_embeddings ue 
         JOIN users u ON ue.user_id = u.id`
      );

      // 计算相似度
      const results = embeddings.map(item => {
        const userEmbedding = JSON.parse(item.embedding);
        const similarity = embeddingService.cosineSimilarity(queryEmbedding, userEmbedding);
        
        return {
          user: {
            id: item.user_id,
            name: item.name,
            email: item.email,
            phone: item.phone,
            department: item.department,
            position: item.position,
            bio: item.bio,
            skills: item.skills
          },
          similarity,
          searchType: 'vector'
        };
      });

      // 排序并返回Top K
      return results
        .sort((a, b) => b.similarity - a.similarity)
        .slice(0, topK);
        
    } finally {
      connection.release();
    }
  }

  /**
   * 混合检索：向量检索 + 关键词检索
   */
  async hybridSearch(query, topK = 5) {
    // 并行执行两种检索
    const [vectorResults, keywordResults] = await Promise.all([
      this.vectorSearch(query, topK * 2), // 获取更多候选
      userService.fullTextSearch(query, topK)
    ]);

    // 合并结果（加权融合）
    const mergedMap = new Map();
    
    // 添加向量检索结果（权重0.6）
    vectorResults.forEach(item => {
      mergedMap.set(item.user.id, {
        user: item.user,
        score: item.similarity * 0.6,
        vectorScore: item.similarity,
        keywordScore: 0
      });
    });

    // 添加关键词检索结果（权重0.4）
    keywordResults.forEach(user => {
      const relevance = user.relevance || 0;
      // 归一化关键词得分
      const normalizedScore = Math.min(relevance / 10, 1);
      
      if (mergedMap.has(user.id)) {
        const existing = mergedMap.get(user.id);
        existing.score += normalizedScore * 0.4;
        existing.keywordScore = normalizedScore;
      } else {
        mergedMap.set(user.id, {
          user,
          score: normalizedScore * 0.4,
          vectorScore: 0,
          keywordScore: normalizedScore
        });
      }
    });

    // 转换并排序
    const mergedResults = Array.from(mergedMap.values())
      .map(item => ({
        ...item,
        searchType: item.vectorScore > 0 && item.keywordScore > 0 
          ? 'hybrid' 
          : item.vectorScore > 0 ? 'vector' : 'keyword'
      }))
      .sort((a, b) => b.score - a.score)
      .slice(0, topK);

    return mergedResults;
  }

  /**
   * 重建所有向量（用于数据更新后）
   */
  async rebuildAllEmbeddings() {
    const connection = await pool.getConnection();
    try {
      // 清空现有向量
      await connection.execute('DELETE FROM user_embeddings');
      console.log('🗑️ 已清空现有向量数据');
      
      // 重新生成
      await this.embedAllUsers();
      console.log('✅ 向量重建完成');
    } finally {
      connection.release();
    }
  }
}

module.exports = new VectorStoreService();

4.6 RAG核心服务 (src/services/ragService.js)重点在这里只要你会写接口看这里就行了

const OpenAI = require('openai');
require('dotenv').config();
const vectorStoreService = require('./vectorStoreService');

class RAGService {
  constructor() {
    this.openai = new OpenAI({
      apiKey: process.env.DEEPSEEK_API_KEY,
      baseURL: process.env.DEEPSEEK_BASE_URL || 'https://api.deepseek.com/v1',
    });
  }

  /**
   * 构建上下文提示词
   */
  buildContext(searchResults) {
    if (!searchResults || searchResults.length === 0) {
      return '未找到相关用户信息。';
    }

    let context = '以下是从数据库中检索到的相关用户信息：\n\n';
    
    searchResults.forEach((item, index) => {
      const user = item.user;
      context += `【用户${index + 1}】\n`;
      context += `- 姓名：${user.name}\n`;
      context += `- 部门：${user.department || '未分配'}\n`;
      context += `- 职位：${user.position || '未设置'}\n`;
      if (user.skills) {
        context += `- 技能：${user.skills}\n`;
      }
      if (user.bio) {
        context += `- 简介：${user.bio}\n`;
      }
      context += `- 邮箱：${user.email}\n`;
      if (user.phone) {
        context += `- 电话：${user.phone}\n`;
      }
      context += `- 匹配度：${(item.score * 100).toFixed(1)}%\n\n`;
    });

    return context;
  }

  /**
   * 构建系统提示词
   */
  buildSystemPrompt(context) {
    return `你是一个专业的企业信息查询助手。你的任务是基于提供的用户信息，准确地回答用户的问题。

## 可用信息
${context}

## 回答规则
1. 严格基于提供的用户信息回答，不要编造信息
2. 如果信息不足，明确告知用户
3. 回答要结构清晰，重点突出
4. 当用户询问联系方式时，提供完整的邮箱和电话
5. 当用户询问技能时，列出所有相关技能
6. 当用户询问部门或职位时，提供完整信息
7. 如果有多个匹配用户，分别列出
8. 使用友好、专业的语气

## 回答格式
- 先给出总结性回答
- 然后列出具体的用户信息
- 最后可以询问是否需要更多帮助`;
  }

  /**
   * 普通查询
   */
  async query(question, topK = 5) {
    console.log(`\n🔍 收到查询: "${question}"`);
    
    // 1. 检索相关用户
    console.log('📊 正在检索相关用户...');
    const searchResults = await vectorStoreService.hybridSearch(question, topK);
    console.log(`✅ 检索到 ${searchResults.length} 个相关用户`);
    
    // 打印检索结果
    searchResults.forEach((item, index) => {
      console.log(`  ${index + 1}. ${item.user.name} (${item.searchType}) - 得分: ${item.score.toFixed(4)}`);
    });
    
    // 2. 构建上下文和提示词
    const context = this.buildContext(searchResults);
    const systemPrompt = this.buildSystemPrompt(context);
    
    // 3. 调用DeepSeek生成回答
    console.log('🤖 正在调用DeepSeek生成回答...');
    const startTime = Date.now();
    
    const response = await this.openai.chat.completions.create({
      model: process.env.DEEPSEEK_MODEL || 'deepseek-chat',
      messages: [
        { role: 'system', content: systemPrompt },
        { role: 'user', content: question }
      ],
      temperature: 0.7,
      max_tokens: 2000,
    });
    
    const answer = response.choices[0].message.content;
    const processingTime = Date.now() - startTime;
    
    console.log(`⏱️ 生成耗时: ${processingTime}ms`);
    
    return {
      question,
      answer,
      relevantUsers: searchResults.map(item => ({
        user: item.user,
        score: item.score,
        searchType: item.searchType
      })),
      processingTime
    };
  }

  /**
   * 流式查询
   */
  async queryStream(question, topK = 5, onChunk) {
    console.log(`\n🔍 收到流式查询: "${question}"`);
    
    // 检索相关用户
    const searchResults = await vectorStoreService.hybridSearch(question, topK);
    
    // 先发送检索到的用户信息
    onChunk({
      type: 'search_results',
      data: searchResults.map(item => ({
        user: item.user,
        score: item.score,
        searchType: item.searchType
      }))
    });
    
    // 构建上下文
    const context = this.buildContext(searchResults);
    const systemPrompt = this.buildSystemPrompt(context);
    
    // 流式调用DeepSeek
    try {
      const stream = await this.openai.chat.completions.create({
        model: process.env.DEEPSEEK_MODEL || 'deepseek-chat',
        messages: [
          { role: 'system', content: systemPrompt },
          { role: 'user', content: question }
        ],
        temperature: 0.7,
        max_tokens: 2000,
        stream: true,
      });

      for await (const chunk of stream) {
        const content = chunk.choices[0]?.delta?.content || '';
        if (content) {
          onChunk({ type: 'answer', content });
        }
      }
      
      onChunk({ type: 'done' });
    } catch (error) {
      onChunk({ type: 'error', message: 'AI服务调用失败' });
    }
  }
}

module.exports = new RAGService();

4.7 查询控制器 (src/controllers/queryController.js)

const ragService = require('../services/ragService');
const userService = require('../services/userService');

class QueryController {
  /**
   * 普通查询接口
   */
  async query(req, res, next) {
    try {
      const { question, topK } = req.body;

      // 参数验证
      if (!question || typeof question !== 'string' || !question.trim()) {
        return res.status(400).json({
          success: false,
          message: '请输入有效的问题',
          data: null
        });
      }

      if (question.length > 500) {
        return res.status(400).json({
          success: false,
          message: '问题长度不能超过500个字符',
          data: null
        });
      }

      // 执行查询
      const result = await ragService.query(
        question.trim(),
        Math.min(parseInt(topK) || 5, 10) // 限制最多返回10个
      );

      // 返回结果
      res.json({
        success: true,
        message: '查询成功',
        data: result
      });

    } catch (error) {
      console.error('查询失败:', error);
      next(error);
    }
  }

  /**
   * 流式查询接口
   */
  async streamQuery(req, res, next) {
    try {
      const { question, topK } = req.body;

      if (!question || typeof question !== 'string' || !question.trim()) {
        return res.status(400).json({
          success: false,
          message: '请输入有效的问题'
        });
      }

      // 设置SSE响应头
      res.setHeader('Content-Type', 'text/event-stream');
      res.setHeader('Cache-Control', 'no-cache');
      res.setHeader('Connection', 'keep-alive');
      res.setHeader('X-Accel-Buffering', 'no');

      // 执行流式查询
      await ragService.queryStream(
        question.trim(),
        Math.min(parseInt(topK) || 5, 10),
        (chunk) => {
          res.write(`data: ${JSON.stringify(chunk)}\n\n`);
        }
      );

      res.end();

    } catch (error) {
      console.error('流式查询失败:', error);
      if (!res.headersSent) {
        next(error);
      } else {
        res.write(`data: ${JSON.stringify({ type: 'error', message: error.message })}\n\n`);
        res.end();
      }
    }
  }

  /**
   * 获取所有用户
   */
  async getAllUsers(req, res, next) {
    try {
      const users = await userService.getAllUsers();
      res.json({
        success: true,
        data: users,
        total: users.length
      });
    } catch (error) {
      next(error);
    }
  }

  /**
   * 获取单个用户
   */
  async getUserById(req, res, next) {
    try {
      const { id } = req.params;
      const user = await userService.getUserById(id);
      
      if (!user) {
        return res.status(404).json({
          success: false,
          message: '用户不存在'
        });
      }

      res.json({
        success: true,
        data: user
      });
    } catch (error) {
      next(error);
    }
  }
}

module.exports = new QueryController();

4.8 路由定义 (src/routes/queryRoutes.js)

const express = require('express');
const router = express.Router();
const queryController = require('../controllers/queryController');

/**
 * @swagger
 * /api/query:
 *   post:
 *     summary: RAG智能查询接口
 *     description: 通过自然语言查询用户信息，系统会自动检索并生成智能回答
 *     tags: [查询]
 *     requestBody:
 *       required: true
 *       content:
 *         application/json:
 *           schema:
 *             type: object
 *             required:
 *               - question
 *             properties:
 *               question:
 *                 type: string
 *                 description: 用户问题
 *                 example: "技术部有哪些高级工程师？"
 *               topK:
 *                 type: integer
 *                 description: 返回的相关用户数量
 *                 default: 5
 *     responses:
 *       200:
 *         description: 查询成功
 */
router.post('/query', queryController.query);

/**
 * @swagger
 * /api/query/stream:
 *   post:
 *     summary: 流式查询接口
 *     description: 使用SSE实现流式返回查询结果
 *     tags: [查询]
 *     requestBody:
 *       required: true
 *       content:
 *         application/json:
 *           schema:
 *             type: object
 *             required:
 *               - question
 *             properties:
 *               question:
 *                 type: string
 *                 example: "找一下会React的开发人员"
 *               topK:
 *                 type: integer
 *                 default: 5
 *     responses:
 *       200:
 *         description: 流式响应
 *         content:
 *           text/event-stream:
 *             schema:
 *               type: string
 */
router.post('/query/stream', queryController.streamQuery);

/**
 * @swagger
 * /api/users:
 *   get:
 *     summary: 获取所有用户
 *     tags: [用户]
 *     responses:
 *       200:
 *         description: 用户列表
 */
router.get('/users', queryController.getAllUsers);

/**
 * @swagger
 * /api/users/{id}:
 *   get:
 *     summary: 获取单个用户
 *     tags: [用户]
 *     parameters:
 *       - in: path
 *         name: id
 *         required: true
 *         schema:
 *           type: integer
 *     responses:
 *       200:
 *         description: 用户信息
 */
router.get('/users/:id', queryController.getUserById);

module.exports = router;

4.9 错误处理中间件 (src/middleware/errorHandler.js)

/**
 * 全局错误处理中间件
 */
function errorHandler(err, req, res, next) {
  console.error('❌ 错误:', {
    message: err.message,
    stack: process.env.NODE_ENV === 'development' ? err.stack : undefined,
    path: req.path,
    method: req.method
  });

  // 处理不同类型的错误
  if (err.type === 'entity.parse.failed') {
    return res.status(400).json({
      success: false,
      message: '请求体格式错误，请使用有效的JSON格式',
      error: 'INVALID_JSON'
    });
  }

  if (err.code === 'ECONNREFUSED' || err.code === 'ER_ACCESS_DENIED_ERROR') {
    return res.status(503).json({
      success: false,
      message: '数据库连接失败，请稍后再试',
      error: 'DB_CONNECTION_ERROR'
    });
  }

  if (err.code === 'ER_DUP_ENTRY') {
    return res.status(409).json({
      success: false,
      message: '数据已存在，请检查后重试',
      error: 'DUPLICATE_ENTRY'
    });
  }

  // API调用错误
  if (err.status === 429) {
    return res.status(429).json({
      success: false,
      message: 'API调用频率超限，请稍后再试',
      error: 'RATE_LIMIT'
    });
  }

  // 默认错误
  const statusCode = err.status || err.statusCode || 500;
  res.status(statusCode).json({
    success: false,
    message: err.message || '服务器内部错误',
    error: process.env.NODE_ENV === 'development' ? err.stack : 'INTERNAL_ERROR'
  });
}

module.exports = errorHandler;

4.10 应用入口文件 (src/app.js)

const express = require('express');
const cors = require('cors');
const helmet = require('helmet');
const morgan = require('morgan');
const rateLimit = require('express-rate-limit');
require('dotenv').config();

const { testConnection, initializeDatabase } = require('./config/database');
const queryRoutes = require('./routes/queryRoutes');
const errorHandler = require('./middleware/errorHandler');
const userService = require('./services/userService');
const vectorStoreService = require('./services/vectorStoreService');

// 创建Express应用
const app = express();
const PORT = process.env.PORT || 3000;

// ==================== 中间件配置 ====================

// 安全中间件
app.use(helmet());

// 跨域配置
app.use(cors({
  origin: '*',
  methods: ['GET', 'POST'],
  allowedHeaders: ['Content-Type', 'Authorization']
}));

// 请求日志
app.use(morgan('[:date[iso]] :method :url :status :response-time ms'));

// 请求体解析
app.use(express.json({ limit: '10mb' }));
app.use(express.urlencoded({ extended: true, limit: '10mb' }));

// API限流
const apiLimiter = rateLimit({
  windowMs: 15 * 60 * 1000, // 15分钟
  max: 100, // 限制100个请求
  message: {
    success: false,
    message: '请求过于频繁，请15分钟后再试',
    error: 'RATE_LIMIT'
  },
  standardHeaders: true,
  legacyHeaders: false,
});

// 查询接口特殊限流
const queryLimiter = rateLimit({
  windowMs: 1 * 60 * 1000, // 1分钟
  max: 10, // 限制10个请求
  message: {
    success: false,
    message: '查询请求过于频繁，请1分钟后再试',
    error: 'QUERY_RATE_LIMIT'
  }
});

app.use('/api/', apiLimiter);
app.use('/api/query', queryLimiter);

// ==================== 路由配置 ====================

// API路由
app.use('/api', queryRoutes);

// 健康检查
app.get('/health', async (req, res) => {
  const dbConnected = await testConnection();
  res.json({
    success: true,
    message: 'RAG系统运行正常',
    database: dbConnected ? 'connected' : 'disconnected',
    timestamp: new Date().toISOString(),
    uptime: process.uptime()
  });
});

// 根路径
app.get('/', (req, res) => {
  res.json({
    name: 'RAG用户检索系统',
    version: '1.0.0',
    description: '基于DeepSeek的智能用户信息检索系统',
    endpoints: {
      query: 'POST /api/query',
      streamQuery: 'POST /api/query/stream',
      users: 'GET /api/users',
      userById: 'GET /api/users/:id',
      health: 'GET /health'
    },
    documentation: 'https://github.com/your-repo/rag-system'
  });
});

// 404处理
app.use((req, res) => {
  res.status(404).json({
    success: false,
    message: `接口 ${req.method} ${req.path} 不存在`,
    error: 'NOT_FOUND'
  });
});

// 错误处理
app.use(errorHandler);

// ==================== 系统初始化 ====================

async function initializeSystem() {
  console.log('\n' + '='.repeat(60));
  console.log('🚀 RAG用户检索系统启动中...');
  console.log('='.repeat(60) + '\n');

  try {
    // 1. 测试数据库连接
    const dbConnected = await testConnection();
    if (!dbConnected) {
      throw new Error('数据库连接失败，请检查配置');
    }

    // 2. 初始化数据库表
    await initializeDatabase();

    // 3. 检查并插入测试数据
    const users = await userService.getAllUsers();
    if (users.length === 0) {
      console.log('📝 插入测试数据...');
      const seedData = userService.getSeedData();
      await userService.insertUsers(seedData);
    }
    console.log(`📊 数据库中共有 ${users.length || 8} 个用户`);

    // 4. 生成向量嵌入
    console.log('\n🧮 检查用户向量嵌入...');
    await vectorStoreService.embedAllUsers();

    console.log('\n' + '='.repeat(60));
    console.log('✅ 系统初始化完成');
    console.log('='.repeat(60) + '\n');

  } catch (error) {
    console.error('\n❌ 系统初始化失败:', error.message);
    process.exit(1);
  }
}

// ==================== 启动服务器 ====================

async function startServer() {
  await initializeSystem();

  app.listen(PORT, () => {
    console.log('='.repeat(60));
    console.log(`🌐 服务器启动成功！`);
    console.log(`📍 地址: http://localhost:${PORT}`);
    console.log(`📋 API接口:`);
    console.log(`   POST /api/query        - RAG智能查询`);
    console.log(`   POST /api/query/stream - 流式查询`);
    console.log(`   GET  /api/users        - 获取所有用户`);
    console.log(`   GET  /api/users/:id    - 获取单个用户`);
    console.log(`   GET  /health           - 健康检查`);
    console.log('='.repeat(60) + '\n');
    
    console.log('💡 快速测试:');
    console.log(`   curl -X POST http://localhost:${PORT}/api/query \\`);
    console.log(`     -H "Content-Type: application/json" \\`);
    console.log(`     -d '{"question":"技术部有哪些工程师？"}'`);
    console.log('\n' + '='.repeat(60) + '\n');
  });

  // 优雅关闭
  process.on('SIGTERM', () => {
    console.log('\n👋 收到关闭信号，正在退出...');
    process.exit(0);
  });
}

// 启动
startServer().catch(console.error);

module.exports = app;

最后使用npm start启动（这里只说一个大概的思路需要具体的项目可以私聊，我这里有一个基于LLM的论文检索系统，功能要更加丰富一点）