SQLServer RAG笔记4：从服务层到前端交互

摘要

在之前的文章中，我们完成了 SQLServer 2025 向量数据库的搭建和基本配置。本文将深入介绍服务层和前端部分的三个核心模块：Embedding 生成、数据入库流程以及前端交互关键代码，完成一个基于SQLServer 2025的向量数据库的RAG系统流程。

一、Embedding 生成：文本向量化的核心

Embedding 是将自然语言转换为数值向量的关键步骤。我们使用 Ollama 本地部署的 nomic-embed-text 模型来完成这一任务。

代码实现

import httpx
import numpy as np
from typing import List
from vector_config import VectorDBConfig

class OllamaEmbeddingService:
    def __init__(self):
        self.base_url = VectorDBConfig.OLLAMA_BASE_URL
        self.model = VectorDBConfig.OLLAMA_EMBEDDING_MODEL
        self.dimension = VectorDBConfig.VECTOR_DIMENSION

    async def get_embedding(self, text: str) -> List[float]:
        async with httpx.AsyncClient(timeout=60.0) as client:
            response = await client.post(
                f"{self.base_url}/api/embeddings",
                json={"model": self.model, "prompt": text}
            )
            response.raise_for_status()
            data = response.json()
            return data.get("embedding", [])

    @staticmethod
    def chunk_text(text: str, chunk_size: int = 500, chunk_overlap: int = 50) -> List[str]:
        if len(text) <= chunk_size:
            return [text]

        chunks = []
        start = 0
        while start < len(text):
            end = start + chunk_size
            chunk = text[start:end]
            if chunk.strip():
                chunks.append(chunk)
            start = end - chunk_overlap
        return chunks

    @staticmethod
    def compute_hash(text: str) -> str:
        import hashlib
        return hashlib.md5(text.encode()).hexdigest()

embedding_service = OllamaEmbeddingService()

知识点笔记

1. 异步 HTTP 请求（httpx.AsyncClient）

   • 使用 httpx 库而非 requests，因为它原生支持异步操作
   • async with 确保请求完成后正确关闭连接，避免资源泄漏
   • 60 秒超时设置防止网络请求挂起

2. 文本分块策略（chunk_text）

   • 为什么需要分块：LLM 的上下文窗口有限，大文档必须拆分成小块
   • 滑动窗口机制：chunk_overlap 参数实现块间重叠，尽量保留上下文连贯性
   • 500字符块大小：500算是个经验值吧，平衡上下文完整性和检索精度

3. 文本向量化（get_embedding）

   • 为什么需要向量化：将文本转换为数值向量，方便后续的相似度计算
   • 嵌入模型的访问：通过异步 HTTP 请求调用 Ollama 模型，访问的路径默认是： http://localhost:11434/api/embeddings ，具体访问哪个模型通过post请求参数指定

4. 内容去重（compute_hash）

   • 使用 MD5 哈希记录文本块，避免重复导入相同内容
   • 后续可扩展为去重逻辑，提高存储效率

二、数据入库

数据入库是连接前端和后端的关键环节，涉及文档分块、向量化和持久化存储。

代码实现

from fastapi import APIRouter, HTTPException
from typing import List
from rag_models import DocumentCreate, DocumentResponse, IngestRequest, IngestResponse, TextChunkCreate
from vector_dao import DocumentDAO, TextChunkDAO, VectorDAO
from embedding_service import embedding_service

router = APIRouter(prefix="/api/rag/documents", tags=["rag-documents"])

@router.post("/ingest", response_model=IngestResponse)
async def ingest_document(ingest_req: IngestRequest):
    # 1. 创建文档记录
    doc = DocumentCreate(
        title=ingest_req.title,
        content=ingest_req.content,
        source=ingest_req.source
    )
    doc_id = DocumentDAO.create_document(doc)
    if not doc_id:
        raise HTTPException(status_code=400, detail="Failed to create document")

    # 2. 文本分块
    chunks = embedding_service.chunk_text(
        ingest_req.content,
        chunk_size=ingest_req.chunk_size,
        chunk_overlap=ingest_req.chunk_overlap
    )

    chunks_created = 0
    vectors_created = 0

    # 3. 批量处理每个块
    for i, chunk_text in enumerate(chunks):
        chunk_hash = embedding_service.compute_hash(chunk_text)
        chunk_id = TextChunkDAO.create_chunk(TextChunkCreate(
            document_id=doc_id,
            chunk_index=i,
            chunk_text=chunk_text,
            chunk_hash=chunk_hash
        ))
        if chunk_id:
            chunks_created += 1
            # 4. 生成 Embedding
            embedding = await embedding_service.get_embedding(chunk_text)
            vector_id = VectorDAO.create_vector(chunk_id, embedding)
            if vector_id:
                vectors_created += 1

    return IngestResponse(
        document_id=doc_id,
        chunks_created=chunks_created,
        vectors_created=vectors_created
    )

知识点笔记

1. 事务处理设计

   • 这里使用了分步提交而非单一事务，允许部分成功
   • 实际生产环境可根据需求调整：要么全成功要么全回滚

2. 数据模型三层架构

   • Documents 表：存储原始文档元数据
   • TextChunks 表：存储分块后的文本
   • VectorIndex 表：存储向量索引，与 TextChunks 一对一关联

3. 异步 vs 同步混合使用

   • get_embedding 是异步的（网络 IO）
   • 数据库操作是同步的（pyodbc 限制）
   • FastAPI 自动处理这种混合场景

三、前端交互：三处关键代码解析

1. 向量检索功能（SearchPanel.vue）

<script setup>
import { ref } from 'vue'
import { ElMessage } from 'element-plus'
import api from '../services/api'

const queryText = ref('')
const topK = ref(5)
const minScore = ref(0.0)
const searching = ref(false)
const searched = ref(false)
const searchResults = ref([])
const total = ref(0)

const doSearch = async () => {
  if (!queryText.value.trim()) {
    ElMessage.warning('请输入检索内容')
    return
  }

  searching.value = true
  searched.value = true
  try {
    const response = await api.searchDocuments(queryText.value, topK.value, minScore.value)
    searchResults.value = response.results || []
    total.value = response.total || 0
  } catch (error) {
    ElMessage.error('检索失败')
    searchResults.value = []
    total.value = 0
  } finally {
    searching.value = false
  }
}
</script>

知识点笔记

1. 响应式状态管理

   • 使用 Vue 3 Composition API 的 ref 管理状态
   • searching 状态控制按钮加载动画，防止重复点击
   • searched 状态区分初始状态和搜索后状态

2. 错误处理模式

   • try-catch-finally 经典模式：
     • try：执行核心逻辑
     • catch：处理异常，清空结果
     • finally：无论成功失败都重置 loading 状态

3. 用户体验优化

   • 输入验证：空搜索直接提示，不发送请求
   • 即时反馈：使用 ElMessage 显示操作结果

2. 文档导入功能（DocumentList.vue）

<script setup>
import { ref, onMounted, watch } from 'vue'
import { ElMessage, ElMessageBox } from 'element-plus'
import api from '../services/api'

const props = defineProps({
  refreshTrigger: Number
})

const emit = defineEmits(['document-deleted'])

const docForm = ref({
  title: '',
  content: '',
  source: '',
  chunk_size: 500,
  chunk_overlap: 50
})

const addDocument = async () => {
  if (!docForm.value.title || !docForm.value.content) {
    ElMessage.warning('请填写标题和内容')
    return
  }
  ingesting.value = true
  try {
    await api.ingestDocument(docForm.value)
    ElMessage.success('文档导入成功')
    showAddDialog.value = false
    docForm.value = { title: '', content: '', source: '', chunk_size: 500, chunk_overlap: 50 }
    loadDocuments()
  } catch (error) {
    ElMessage.error('文档导入失败')
  } finally {
    ingesting.value = false
  }
}

watch(() => props.refreshTrigger, loadDocuments)
onMounted(loadDocuments)
</script>

知识点笔记

1. 父子组件通信

   • defineProps 接收父组件传递的 refreshTrigger
   • defineEmits 定义事件，通知父组件刷新数据
   • watch 监听触发器变化，自动重新加载文档列表

2. 表单重置机制

   • 成功提交后手动重置表单对象
   • 避免遗留数据干扰下一次输入

3. 生命周期钩子

   • onMounted：组件挂载时自动加载数据
   • 这种模式确保页面打开就能看到数据

3. API 层封装（api.js）

import axios from 'axios'

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

api.interceptors.response.use(
  response => response.data,
  error => {
    console.error('API Error:', error)
    return Promise.reject(error)
  }
)

export default {
  searchDocuments(queryText, topK = 5, minScore = 0.0) {
    return api.post('/rag/search/', {
      query_text: queryText,
      top_k: topK,
      min_score: minScore
    })
  },

  ingestDocument(ingestData) {
    return api.post('/rag/documents/ingest', ingestData)
  },

  getDocuments(limit = 100, offset = 0) {
    const params = new URLSearchParams({ limit, offset })
    return api.get(`/rag/documents/?${params}`)
  }
}

知识点笔记

1. Axios 实例化

   • 创建独立实例，避免污染全局配置
   • baseURL: '/api' 配合 Vite 代理，解决跨域问题

2. 响应拦截器

   • 统一处理响应，直接返回 response.data
   • 组件中不需要每次写 .data，代码更简洁
   • 统一错误日志记录

3. 函数参数设计

   • 提供合理默认值，调用更灵活
   • 使用 URLSearchParams 构建查询字符串，避免手动拼接

4. 命名约定

   • JavaScript 使用驼峰命名（queryText）
   • Python 使用下划线命名（query_text）
   • API 层负责转换，保持各自语言习惯

四、完整数据流总结

请求流程

1. 文档导入流程

   前端表单 → api.ingestDocument() → 后端 /ingest 接口 
   → DocumentDAO.create_document() → TextChunkDAO.create_chunk() 
   → embedding_service.get_embedding() → VectorDAO.create_vector()
   

2. 向量检索流程

   前端搜索框 → api.searchDocuments() → 后端 /search 接口 
   → embedding_service.get_embedding() → VectorDAO.vector_search() 
   → VECTOR_SEARCH SQL → 返回结果
   

五、功能演示

为了掩饰核心功能，简化周边功能，所以暂时没有封装文档上传的功能，文章的内容可以通过UI进行输入，系统会自动将其分块并且转换为向量并存储到数据库中。

总结

此篇只讲述了服务层和前端关键处的几段代码，完整的项目代码可以从以下站点获取：

SQLServerRAG