当前位置：首页 > news >正文

37、RAG系统架构与实现：知识增强型AI的完整构建

news 2025/9/20 17:30:14

核心学习目标：深入理解检索增强生成的系统架构，掌握向量检索和知识融合技术，建立知识增强型AI系统的完整能力，实现实用的RAG应用。

检索增强生成（RAG）代表了人工智能领域的重要进展，它将大语言模型的生成能力与外部知识库的检索能力结合，解决了模型知识时效性和领域专业性的关键挑战。通过构建高效的向量检索系统、智能的文档处理流程和精确的知识融合机制，我们能够创建真正实用的知识增强型AI应用，为用户提供准确、及时、可靠的智能问答服务。

一、RAG核心概念与架构设计：知识增强的智能范式

1.1 检索增强生成的核心机制

> 传统生成模型的局限性分析

参数化知识的固化问题是传统大语言模型面临的根本挑战。模型的知识完全封装在参数中，一旦训练完成就无法更新，导致信息时效性差。当现实世界发生变化时，模型无法获取最新信息，容易产生过时或错误的回答。这种知识固化还表现在领域专业性不足上，通用模型难以覆盖所有专业领域的深度知识。

幻觉现象的产生机制源于模型在缺乏相关知识时的"创造性"填补。当面对超出训练数据范围的问题时，模型倾向于基于模糊的模式匹配生成看似合理但实际错误的内容。这种幻觉在专业咨询、事实核查等高准确性要求场景中尤为危险，可能导致误导性的信息传播。

知识边界的模糊性使得用户难以判断模型回答的可靠性。传统模型无法明确指出自己的知识边界，也无法提供信息来源，这在需要可追溯性和可验证性的应用中构成重大障碍。

> RAG架构的创新突破

动态知识注入机制通过实时检索外部知识源为模型提供最新、相关的信息。这种机制将模型的参数化知识与外部结构化知识有机结合，既保持了生成的流畅性，又确保了内容的准确性和时效性。检索到的知识片段作为生成的依据，显著降低了幻觉风险。

知识来源的可追溯性是RAG系统的重要优势。每个生成的答案都可以追溯到具体的源文档，用户可以验证信息的可靠性。这种透明度不仅提升了系统的可信度，也为持续的知识库优化提供了反馈机制。

领域适应的灵活性通过更换或扩展知识库实现。无需重新训练模型，只需构建特定领域的高质量知识库，就能快速适应新的应用场景。这种模块化设计大大降低了系统维护成本和部署复杂度。

1.2 系统架构的分层设计

> 分层架构的职责边界

用户交互层负责查询接收、参数解析、结果格式化和用户体验优化。这一层需要处理多种查询类型，支持结构化和非结构化输入，同时提供直观的结果展示和交互反馈。设计重点在于降低使用门槛和提升响应体验。

应用逻辑层实现核心业务逻辑，包括查询意图识别、检索策略选择、结果融合和质量控制。这一层是系统的"大脑"，决定了如何最有效地利用底层资源回答用户问题。需要根据查询特点动态调整检索和生成策略。

检索服务层提供多样化的检索能力，支持向量检索、关键词检索、语义检索等多种方式。不同检索方式各有优势：向量检索擅长语义相似性匹配，关键词检索擅长精确匹配，混合检索能够综合两者优势。

知识存储层管理多种形式的知识资源，包括向量表示、原始文档和结构化元数据。存储设计需要平衡查询性能、存储成本和数据一致性，支持增量更新和版本管理。

计算引擎层提供核心的AI能力，包括文本向量化、内容生成和相关性排序。这一层的性能直接影响整个系统的响应速度和准确性，需要根据硬件条件和性能要求进行优化配置。

二、向量数据库设计与实现：高效检索的技术基石

2.1 向量索引的算法原理

> HNSW算法的层次导航机制

分层图结构的构建原理基于小世界网络理论，通过构建多层导航图实现高效的近似最近邻搜索。底层包含所有向量节点，上层节点数量递减，形成金字塔结构。每个节点在不同层级都有不同的连接模式：上层连接跨度大，用于快速导航到目标区域；下层连接密集，用于精确定位。

搜索路径的优化策略从最顶层开始搜索，利用长距离连接快速逼近目标区域，然后逐层下降进行精细搜索。这种分层搜索避免了传统算法的局部最优陷阱，在保证高召回率的同时显著提升搜索速度。搜索复杂度从线性降低到对数级别。

动态更新的平衡维护新向量插入时需要维护图结构的平衡性。插入位置通过搜索确定，连接策略确保局部和全局连通性。删除操作需要重新连接相关节点，保持图的完整性。合理的更新策略能够在动态环境中保持索引性能。

> IVF（Inverted File）索引的聚类分割思想

倒排索引的向量化扩展将传统文本检索的倒排索引概念扩展到向量空间。通过聚类算法将向量空间划分为若干子区域，每个子区域维护一个倒排列表。查询时只需搜索最相关的几个子区域，大幅减少计算量。

聚类中心的自适应优化聚类质量直接影响检索效果。K-means等算法用于确定聚类中心，需要考虑数据分布特点和查询模式。动态调整聚类参数可以适应数据分布的变化，保持索引的有效性。

探查深度的动态调整根据查询要求的精度和速度平衡调整探查的聚类数量。高精度要求时增加探查深度，实时性要求高时减少探查范围。这种自适应机制让系统能够灵活应对不同应用场景。

2.2 相似度计算的数学基础

> 距离度量的选择原则

余弦相似度的几何意义测量向量间的夹角，不受向量长度影响，适合文本向量等归一化场景。计算公式为两向量内积除以长度乘积，值域为[-1,1]，1表示完全相似，-1表示完全相反。在文档检索中，余弦相似度能够有效捕捉内容相关性，忽略文档长度差异。

欧氏距离的空间直觉测量向量在高维空间中的直线距离，适合坐标信息重要的场景。距离越小表示越相似，但受向量维度影响较大。在图像特征等场景中，欧氏距离能够保持空间关系的直观性。

内积计算的效率优势直接计算向量内积，计算复杂度最低，适合大规模检索场景。内积值越大表示越相似，但受向量长度影响。在经过归一化的向量空间中，内积等价于余弦相似度，兼具计算效率和语义合理性。

> 量化压缩的性能权衡

乘积量化的压缩原理将高维向量分割为若干子向量，每个子向量独立量化为码本中的码字。原向量用码字索引表示，大幅减少存储空间。检索时通过查表计算近似距离，平衡了存储效率和计算精度。

量化误差的控制策略通过增加码本大小或子向量数量来降低量化误差。需要在存储压缩比和检索精度之间找到最优平衡点。训练阶段的码本优化对最终效果至关重要，需要使用具代表性的样本数据。

三、文档处理与向量化流程：知识获取的智能管道

3.1 文档智能分割的策略设计

> 基于语义的分割算法

句子边界的智能识别超越简单的句号分割，考虑缩写、数字、特殊格式等复杂情况。使用规则和机器学习相结合的方法，准确识别真实的句子边界。对于技术文档、法律条文等专业内容，需要特殊的边界识别规则。

语义完整性的保持策略确保分割后的每个片段都包含完整的语义信息，避免切断重要的上下文关系。通过分析句间相关性、实体引用关系和逻辑连接词，判断最佳分割位置。平衡片段长度和语义完整性是关键挑战。

重叠窗口的设计原则在相邻片段间设置适当重叠，保持上下文的连续性。重叠长度需要根据内容类型和检索需求调整，技术文档可能需要更大重叠，新闻资讯可能需要较小重叠。重叠策略有助于避免检索时的边界效应。

> 多格式文档的统一处理

PDF文档的复杂解析处理文字、图片、表格、公式等混合内容，保持原始格式和逻辑结构。使用OCR技术处理扫描版PDF，结合版面分析算法恢复文档结构。特别关注科学论文、技术手册等结构复杂的文档类型。

Word文档的结构还原利用文档的内置结构信息，如标题层级、样式、批注等，构建文档的逻辑组织。保持表格、列表、图表等结构化信息的完整性。处理版本兼容性和格式转换中的信息损失问题。

网页内容的净化提取去除广告、导航、版权等无关信息，提取核心内容。处理JavaScript动态生成的内容，使用无头浏览器或API接口获取完整信息。识别和保持文章结构、段落层次和超链接关系。

3.2 向量化编码的技术实现

> 预训练模型的选择与优化

通用语言模型的适应性BERT、RoBERTa等通用模型在多数场景下表现良好，但在特定领域可能需要进一步优化。通过领域数据微调或使用领域专用的预训练模型可以提升向量化质量。平衡模型复杂度和推理性能是重要考虑因素。

多语言模型的跨语言能力在处理多语言内容时，使用multilingual-BERT等跨语言模型实现统一向量空间。这种统一性有助于跨语言检索和知识关联，但可能在单一语言上的表现不如专用模型。

领域特化模型的性能提升BioBERT、FinBERT等领域专用模型在特定领域内能提供更准确的向量表示。这些模型通过领域语料的预训练，更好地理解专业术语和领域知识。选择模型时需要平衡领域适应性和通用性。

> 向量质量的评估机制

语义相似性的验证测试构建测试集验证向量化模型对语义相似性的捕捉能力。测试集应包含同义词、释义、上下位关系等多种语义关系类型。通过相似度计算和人工评估相结合的方式评价向量质量。

聚类质量的定量分析使用聚类算法分析向量空间的结构，评估同类内容的向量聚合程度和不同类内容的向量分离度。通过轮廓系数、戴维斯-博尔丁指数等指标定量评估聚类质量。

检索效果的端到端验证在实际检索任务中验证向量化效果，关注检索准确率、召回率和排序质量。通过A/B测试比较不同向量化策略的实际效果，建立从向量质量到业务指标的完整评估体系。

四、检索策略优化设计：精准召回的算法架构

4.1 多轮检索的渐进优化

> 粗召回与精排序的分离设计

高效粗召回的实现策略第一阶段采用高效但相对粗糙的检索方法，如近似最近邻搜索，从海量文档中快速筛选出候选集合。这一阶段优先考虑召回率和计算效率，允许一定的精度损失。候选集合规模通常是最终结果的10-100倍。

精确精排序的算法选择第二阶段对候选集合进行精确的相关性计算，使用更复杂的相似度算法或机器学习模型。可以引入更多特征，如文档权威性、时效性、用户偏好等。这一阶段重点提升排序精度，计算复杂度相对可控。

多阶段融合的策略设计不同阶段可以采用不同的相似度度量和排序策略，通过加权融合得到最终排序。融合权重可以根据查询类型、用户画像或历史效果动态调整。建立端到端的优化目标，确保各阶段协同工作。

> 查询扩展与重写机制

同义词扩展的语义增强通过词典、词向量或语言模型识别查询词的同义词和近义词，扩展查询范围。特别关注专业术语的同义表达，如"机器学习"和"ML"。扩展策略需要控制噪声引入，避免语义漂移。

上下文理解的查询重写分析用户历史查询和对话上下文，理解当前查询的真实意图。处理代词指代、省略信息和隐含条件等语言现象。利用对话历史补全查询信息，提升检索的准确性。

多角度查询的并行处理从不同角度理解用户查询，生成多个候选查询版本并行执行检索。例如，既考虑字面意思，也考虑深层意图；既检索直接答案，也检索相关背景。通过结果融合获得更全面的检索覆盖。

4.2 相关性评估的多维融合

> 语义相似度的计算优化

多层次语义匹配在词汇、短语、句子和文档多个层次计算语义相似度。词汇层面关注精确匹配和同义替换，短语层面关注概念对应，句子层面关注语义等价，文档层面关注主题相关。综合多层次信息得到更准确的相似度评估。

上下文感知的相似度调整考虑词汇和短语在特定上下文中的语义变化，如"苹果"在科技和水果语境中的不同含义。使用上下文相关的词向量或语言模型捕捉这种语义变化。动态调整相似度权重，提升匹配精度。

领域知识的融入策略在特定领域内引入专业知识图谱或本体，增强语义理解能力。识别专业术语间的上下位关系、同义关系和关联关系。利用领域知识修正通用模型的相似度计算，提升专业查询的检索效果。

> 多特征融合的排序模型

传统特征的有效利用结合TF-IDF、BM25等传统检索特征与向量相似度特征，发挥各自优势。传统特征擅长捕捉精确匹配和词频信息，向量特征擅长捕捉语义相关性。通过特征工程和权重学习优化融合效果。

文档质量特征的引入考虑文档的权威性、完整性、准确性等质量指标。权威性可以通过来源评级、作者声誉等判断；完整性可以通过文档长度、结构完整性等评估；准确性可以通过事实核查、引用数量等衡量。质量特征有助于提升检索结果的可信度。

用户行为特征的个性化利用用户历史查询、点击、停留时间等行为数据，学习用户偏好模式。识别用户对不同内容类型、难度级别、时效性的偏好。通过个性化调整排序权重，提升用户满意度和系统粘性。

五、知识融合与生成控制：智能回答的精准制导

5.1 知识片段的智能提取

> 相关性判断的细粒度控制

段落级相关性的精确评估不仅考虑整体文档相关性，更要识别文档中与查询最相关的具体段落。通过滑动窗口和局部相似度计算，定位包含答案的关键段落。避免引入无关信息稀释答案质量，同时确保不遗漏重要信息。

实体关系的语义匹配识别查询中的关键实体和关系，在检索文档中寻找对应的实体提及和关系表达。利用命名实体识别和关系抽取技术，建立查询实体与文档实体间的对应关系。这种细粒度匹配能够提升答案的准确性和具体性。

证据链的完整性构建对于复杂查询，需要多个证据片段支撑答案。识别证据间的逻辑关系，构建完整的推理链条。确保证据的时间一致性、逻辑一致性和事实一致性。通过证据链的完整性评估，提升答案的可信度。

> 上下文窗口的动态管理

长度限制下的信息最大化在生成模型的上下文长度限制下，最大化有用信息的包含。通过重要性排序和压缩算法，在有限空间内放入最相关、最关键的信息。平衡信息完整性和计算效率，避免信息截断造成的理解偏差。

结构化信息的有序组织将检索到的知识片段按照逻辑顺序和重要性排列，形成结构化的上下文。区分背景信息、核心事实和支撑证据，分别放置在上下文的不同位置。这种组织方式有助于生成模型更好地理解和利用知识。

动态扩展与压缩策略根据查询复杂度和生成需求动态调整上下文长度。简单查询使用精简上下文，复杂查询扩展更多背景信息。在生成过程中根据需要动态调整上下文内容，实现知识的按需供给。

5.2 生成质量的多维控制

> 事实一致性的验证机制

知识源的可信度评估建立文档和信息源的可信度评级体系，优先使用权威、可靠的知识源。考虑发布机构的声誉、文档的引用情况、信息的核查状态等因素。在生成答案时，明确标注信息来源和可信度级别。

交叉验证的多源确认对于关键事实，尝试从多个独立源进行验证。识别信息间的一致性和矛盾性，对矛盾信息进行标注和说明。通过多源验证提升答案的可靠性，降低单一源错误的影响。

时效性检查的自动化检查信息的发布时间和更新状态，优先使用最新信息。对于时效性敏感的查询，如股价、政策、新闻等，特别关注信息的时间标签。建立信息过期提醒机制，避免使用过时信息。

> 回答完整性的结构化保障

答案结构的逻辑组织根据查询类型设计相应的答案结构模板。事实性查询提供直接答案和支撑证据，分析性查询提供多角度观点和逻辑推理，操作性查询提供步骤说明和注意事项。结构化的答案更易于理解和使用。

关键信息的完整覆盖确保答案涵盖查询所需的所有关键信息点，避免遗漏重要细节。通过查询分析识别必需信息点，检查答案的覆盖程度。对于复杂查询，使用检查清单确保答案的完整性。

不确定性的诚实表达当知识不足或存在争议时，诚实表达不确定性和局限性。明确区分已确认事实、可能推论和个人观点。提供相关的进一步查询建议，引导用户获取更准确信息。

六、RAG系统示例代码：端到端智能问答平台

基于前面章节的理论基础，现在构建一个示例的RAG系统，集成文档处理、向量检索、知识融合和答案生成等核心功能，实现高质量的智能问答服务。

import os
import json
import logging
import asyncio
import uuid
from typing import List, Dict, Optional, Tuple, Any
from dataclasses import dataclass, asdict
from datetime import datetime
import hashlib# 基础依赖
import numpy as np
import torch
import torch.nn.functional as F
from transformers import AutoTokenizer, AutoModel
import faiss
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity# 文档处理
import PyPDF2
import docx
from bs4 import BeautifulSoup
import re
import nltk
from nltk.tokenize import sent_tokenize, word_tokenize
from nltk.corpus import stopwords# Web框架
from fastapi import FastAPI, HTTPException, UploadFile, File
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import uvicorn# 确保NLTK数据下载
try:nltk.data.find('tokenizers/punkt')nltk.data.find('corpora/stopwords')
except LookupError:nltk.download('punkt', quiet=True)nltk.download('stopwords', quiet=True)# ===== 配置与数据结构 =====@dataclass
class DocumentChunk:"""文档片段数据结构"""id: strcontent: strmetadata: Dict[str, Any]vector: Optional[np.ndarray] = Nonedef to_dict(self) -> Dict:"""转换为字典格式"""result = asdict(self)if self.vector is not None:result['vector'] = self.vector.tolist()return result@classmethoddef from_dict(cls, data: Dict) -> 'DocumentChunk':"""从字典创建对象"""vector = Noneif 'vector' in data and data['vector'] is not None:vector = np.array(data['vector'])return cls(id=data['id'],content=data['content'],metadata=data['metadata'],vector=vector)@dataclass
class RetrievalResult:"""检索结果数据结构"""chunk_id: strcontent: strscore: floatmetadata: Dict[str, Any]@dataclass
class RAGResponse:"""RAG系统响应结构"""answer: strsources: List[Dict[str, Any]]confidence: floatprocessing_time: floatclass RAGConfig:"""RAG系统配置"""def __init__(self):self.embedding_model = "sentence-transformers/all-MiniLM-L6-v2"self.chunk_size = 512self.chunk_overlap = 50self.top_k_retrieval = 5self.similarity_threshold = 0.3self.max_context_length = 2048self.vector_dim = 384  # MiniLM-L6的向量维度# ===== 文档处理模块 =====class DocumentProcessor:"""文档处理器：支持多格式文档解析和智能分割"""def __init__(self, config: RAGConfig):self.config = configself.stop_words = set(stopwords.words('english'))def extract_text(self, file_path: str, file_type: str) -> str:"""从不同格式文件提取文本"""try:if file_type.lower() == 'pdf':return self._extract_pdf_text(file_path)elif file_type.lower() in ['docx', 'doc']:return self._extract_docx_text(file_path)elif file_type.lower() == 'txt':with open(file_path, 'r', encoding='utf-8') as file:return file.read()elif file_type.lower() == 'html':return self._extract_html_text(file_path)else:raise ValueError(f"Unsupported file type: {file_type}")except Exception as e:logging.error(f"Error extracting text from {file_path}: {e}")return ""def _extract_pdf_text(self, file_path: str) -> str:"""提取PDF文本"""text = ""try:with open(file_path, 'rb') as file:pdf_reader = PyPDF2.PdfReader(file)for page in pdf_reader.pages:text += page.extract_text() + "\n"except Exception as e:logging.error(f"PDF extraction error: {e}")return textdef _extract_docx_text(self, file_path: str) -> str:"""提取Word文档文本"""try:doc = docx.Document(file_path)return '\n'.join([paragraph.text for paragraph in doc.paragraphs])except Exception as e:logging.error(f"DOCX extraction error: {e}")return ""def _extract_html_text(self, file_path: str) -> str:"""提取HTML文本"""try:with open(file_path, 'r', encoding='utf-8') as file:soup = BeautifulSoup(file.read(), 'html.parser')# 移除脚本和样式内容for script in soup(["script", "style"]):script.decompose()return soup.get_text()except Exception as e:logging.error(f"HTML extraction error: {e}")return ""def clean_text(self, text: str) -> str:"""文本清洗"""# 移除多余空白字符text = re.sub(r'\s+', ' ', text)# 移除特殊字符text = re.sub(r'[^\w\s\.\,\!\?\;\:]', ' ', text)# 移除过短的行lines = [line.strip() for line in text.split('\n') if len(line.strip()) > 10]return '\n'.join(lines)def intelligent_split(self, text: str, metadata: Dict[str, Any]) -> List[DocumentChunk]:"""智能文档分割"""# 清洗文本text = self.clean_text(text)# 按句子分割sentences = sent_tokenize(text)chunks = []current_chunk = ""current_length = 0for sentence in sentences:sentence_length = len(sentence)# 检查是否需要开始新的块if (current_length + sentence_length > self.config.chunk_size and current_chunk):# 创建当前块chunk_id = str(uuid.uuid4())chunk = DocumentChunk(id=chunk_id,content=current_chunk.strip(),metadata={**metadata,'chunk_length': len(current_chunk),'created_at': datetime.now().isoformat()})chunks.append(chunk)# 开始新块，保持重叠if self.config.chunk_overlap > 0:# 保留当前块的最后部分作为重叠overlap_text = current_chunk[-self.config.chunk_overlap:]current_chunk = overlap_text + " " + sentencecurrent_length = len(current_chunk)else:current_chunk = sentencecurrent_length = sentence_lengthelse:# 添加到当前块if current_chunk:current_chunk += " " + sentenceelse:current_chunk = sentencecurrent_length += sentence_length# 处理最后一个块if current_chunk.strip():chunk_id = str(uuid.uuid4())chunk = DocumentChunk(id=chunk_id,content=current_chunk.strip(),metadata={**metadata,'chunk_length': len(current_chunk),'created_at': datetime.now().isoformat()})chunks.append(chunk)return chunksdef process_document(self, file_path: str, file_type: str, doc_metadata: Optional[Dict] = None) -> List[DocumentChunk]:"""完整文档处理流程"""# 提取文本text = self.extract_text(file_path, file_type)if not text:return []# 构建元数据metadata = {'source_file': os.path.basename(file_path),'file_type': file_type,'processed_at': datetime.now().isoformat(),'file_hash': hashlib.md5(text.encode()).hexdigest()}if doc_metadata:metadata.update(doc_metadata)# 智能分割chunks = self.intelligent_split(text, metadata)logging.info(f"Processed {file_path}: {len(chunks)} chunks created")return chunks# ===== 向量化模块 =====class EmbeddingModel:"""向量化模型：负责文本的向量表示"""def __init__(self, config: RAGConfig):self.config = configself.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')self.tokenizer = Noneself.model = Noneself._load_model()def _load_model(self):"""加载预训练模型"""try:self.tokenizer = AutoTokenizer.from_pretrained(self.config.embedding_model)self.model = AutoModel.from_pretrained(self.config.embedding_model)self.model.to(self.device)self.model.eval()logging.info(f"Embedding model loaded: {self.config.embedding_model}")except Exception as e:logging.error(f"Failed to load embedding model: {e}")raisedef encode_texts(self, texts: List[str]) -> np.ndarray:"""批量编码文本为向量"""if not texts:return np.array([])embeddings = []batch_size = 32  # 批处理大小with torch.no_grad():for i in range(0, len(texts), batch_size):batch_texts = texts[i:i + batch_size]# Tokenizeencoded = self.tokenizer(batch_texts,padding=True,truncation=True,max_length=512,return_tensors='pt')# 移动到设备encoded = {k: v.to(self.device) for k, v in encoded.items()}# 获取embeddingsoutputs = self.model(**encoded)# 使用[CLS] token的embedding或平均池化batch_embeddings = outputs.last_hidden_state.mean(dim=1)embeddings.append(batch_embeddings.cpu().numpy())return np.vstack(embeddings) if embeddings else np.array([])def encode_single(self, text: str) -> np.ndarray:"""编码单个文本"""return self.encode_texts([text])[0]# ===== 向量数据库模块 =====class VectorDatabase:"""向量数据库：高效的向量检索系统"""def __init__(self, config: RAGConfig):self.config = configself.index = Noneself.chunks = {}  # chunk_id -> DocumentChunkself.chunk_ids = []  # 维护ID与索引位置的对应关系self.tfidf_vectorizer = TfidfVectorizer(stop_words='english', max_features=5000)self.tfidf_matrix = Noneself._initialize_index()def _initialize_index(self):"""初始化FAISS索引"""self.index = faiss.IndexFlatIP(self.config.vector_dim)  # 使用内积相似度logging.info("Vector database initialized")def add_chunks(self, chunks: List[DocumentChunk], embeddings: np.ndarray):"""添加文档片段和对应向量"""if len(chunks) != len(embeddings):raise ValueError("Chunks and embeddings length mismatch")# 归一化向量用于余弦相似度normalized_embeddings = embeddings.copy()norms = np.linalg.norm(normalized_embeddings, axis=1, keepdims=True)normalized_embeddings = normalized_embeddings / np.maximum(norms, 1e-12)# 添加到FAISS索引self.index.add(normalized_embeddings.astype('float32'))# 存储chunks和更新ID映射for i, chunk in enumerate(chunks):chunk.vector = normalized_embeddings[i]self.chunks[chunk.id] = chunkself.chunk_ids.append(chunk.id)# 更新TF-IDF矩阵self._update_tfidf_index()logging.info(f"Added {len(chunks)} chunks to database")def _update_tfidf_index(self):"""更新TF-IDF索引"""if not self.chunks:returntexts = [chunk.content for chunk in self.chunks.values()]try:self.tfidf_matrix = self.tfidf_vectorizer.fit_transform(texts)except Exception as e:logging.error(f"TF-IDF update failed: {e}")def vector_search(self, query_vector: np.ndarray, top_k: int = None) -> List[RetrievalResult]:"""向量检索"""if top_k is None:top_k = self.config.top_k_retrievalif self.index.ntotal == 0:return []# 归一化查询向量query_vector = query_vector.copy()norm = np.linalg.norm(query_vector)if norm > 1e-12:query_vector = query_vector / norm# 执行检索scores, indices = self.index.search(query_vector.reshape(1, -1).astype('float32'), min(top_k, self.index.ntotal))results = []for score, idx in zip(scores[0], indices[0]):if idx >= 0 and score >= self.config.similarity_threshold:chunk_id = self.chunk_ids[idx]chunk = self.chunks[chunk_id]result = RetrievalResult(chunk_id=chunk_id,content=chunk.content,score=float(score),metadata=chunk.metadata)results.append(result)return resultsdef keyword_search(self, query: str, top_k: int = None) -> List[RetrievalResult]:"""关键词检索"""if top_k is None:top_k = self.config.top_k_retrievalif self.tfidf_matrix is None or len(self.chunks) == 0:return []try:# 向量化查询query_vector = self.tfidf_vectorizer.transform([query])# 计算相似度similarities = cosine_similarity(query_vector, self.tfidf_matrix)[0]# 获取top-k结果top_indices = np.argsort(similarities)[::-1][:top_k]results = []chunk_list = list(self.chunks.values())for idx in top_indices:if similarities[idx] >= self.config.similarity_threshold * 0.5:  # 降低阈值chunk = chunk_list[idx]result = RetrievalResult(chunk_id=chunk.id,content=chunk.content,score=float(similarities[idx]),metadata=chunk.metadata)results.append(result)return resultsexcept Exception as e:logging.error(f"Keyword search failed: {e}")return []def hybrid_search(self, query: str, query_vector: np.ndarray, vector_weight: float = 0.7) -> List[RetrievalResult]:"""混合检索：结合向量和关键词检索"""# 执行两种检索vector_results = self.vector_search(query_vector)keyword_results = self.keyword_search(query)# 合并结果combined_results = {}# 添加向量检索结果for result in vector_results:combined_results[result.chunk_id] = RetrievalResult(chunk_id=result.chunk_id,content=result.content,score=result.score * vector_weight,metadata=result.metadata)# 添加关键词检索结果for result in keyword_results:if result.chunk_id in combined_results:# 融合分数combined_results[result.chunk_id].score += result.score * (1 - vector_weight)else:combined_results[result.chunk_id] = RetrievalResult(chunk_id=result.chunk_id,content=result.content,score=result.score * (1 - vector_weight),metadata=result.metadata)# 排序并返回results = list(combined_results.values())results.sort(key=lambda x: x.score, reverse=True)return results[:self.config.top_k_retrieval]def get_stats(self) -> Dict[str, Any]:"""获取数据库统计信息"""return {'total_chunks': len(self.chunks),'index_size': self.index.ntotal if self.index else 0,'vector_dimension': self.config.vector_dim,'memory_usage_mb': self.index.ntotal * self.config.vector_dim * 4 / (1024 * 1024)}# ===== 简化生成模块 =====class SimpleGenerator:"""简化的答案生成器"""def __init__(self, config: RAGConfig):self.config = configdef generate_answer(self, query: str, context_chunks: List[RetrievalResult]) -> str:"""基于检索上下文生成答案"""if not context_chunks:return "抱歉，我没有找到相关信息来回答您的问题。"# 构建上下文context_parts = []for i, chunk in enumerate(context_chunks):source_info = chunk.metadata.get('source_file', '未知来源')context_parts.append(f"来源 {i+1} ({source_info}): {chunk.content}")context = "\n\n".join(context_parts)# 简单的模板生成（实际应用中应使用真正的生成模型）answer_template = f"""基于检索到的信息，针对问题："{query}"相关信息：
{context}根据以上信息，我可以为您提供如下回答："""# 这里应该调用实际的生成模型，现在使用简单的摘要逻辑# 提取关键句子作为答案key_sentences = []for chunk in context_chunks[:2]:  # 使用前两个最相关的chunksentences = sent_tokenize(chunk.content)if sentences:key_sentences.append(sentences[0])  # 取首句summary = " ".join(key_sentences)return answer_template + summary + "\n\n请注意：以上答案基于检索到的文档内容生成，建议您查看原始资料以获取更完整的信息。"# ===== 核心RAG系统 =====class RAGSystem:"""完整的RAG系统"""def __init__(self, config: RAGConfig):self.config = configself.doc_processor = DocumentProcessor(config)self.embedding_model = EmbeddingModel(config)self.vector_db = VectorDatabase(config)self.generator = SimpleGenerator(config)logging.info("RAG System initialized successfully")def add_document(self, file_path: str, file_type: str, metadata: Optional[Dict] = None) -> Dict[str, Any]:"""添加文档到知识库"""try:# 处理文档chunks = self.doc_processor.process_document(file_path, file_type, metadata)if not chunks:return {"success": False, "message": "No content extracted from document"}# 向量化texts = [chunk.content for chunk in chunks]embeddings = self.embedding_model.encode_texts(texts)# 添加到向量数据库self.vector_db.add_chunks(chunks, embeddings)return {"success": True, "chunks_added": len(chunks),"message": f"Successfully processed {len(chunks)} chunks"}except Exception as e:logging.error(f"Error adding document: {e}")return {"success": False, "message": str(e)}def query(self, question: str, search_type: str = "hybrid") -> RAGResponse:"""处理查询请求"""start_time = datetime.now()try:# 向量化查询query_vector = self.embedding_model.encode_single(question)# 执行检索if search_type == "vector":results = self.vector_db.vector_search(query_vector)elif search_type == "keyword":results = self.vector_db.keyword_search(question)else:  # hybridresults = self.vector_db.hybrid_search(question, query_vector)# 生成答案answer = self.generator.generate_answer(question, results)# 计算置信度（基于检索结果质量）confidence = 0.0if results:confidence = min(1.0, sum(r.score for r in results) / len(results))# 构建源信息sources = []for result in results:sources.append({"content": result.content[:200] + "...",  # 截断显示"score": result.score,"metadata": result.metadata})processing_time = (datetime.now() - start_time).total_seconds()return RAGResponse(answer=answer,sources=sources,confidence=confidence,processing_time=processing_time)except Exception as e:logging.error(f"Query processing error: {e}")return RAGResponse(answer="抱歉，处理您的查询时出现了错误。",sources=[],confidence=0.0,processing_time=(datetime.now() - start_time).total_seconds())def get_system_status(self) -> Dict[str, Any]:"""获取系统状态"""return {"database_stats": self.vector_db.get_stats(),"config": {"embedding_model": self.config.embedding_model,"chunk_size": self.config.chunk_size,"top_k_retrieval": self.config.top_k_retrieval},"timestamp": datetime.now().isoformat()}# ===== Web API接口 =====# Pydantic模型定义
class QueryRequest(BaseModel):question: strsearch_type: str = "hybrid"class QueryResponse(BaseModel):answer: strsources: List[Dict]confidence: floatprocessing_time: floatclass DocumentUploadResponse(BaseModel):success: boolmessage: strchunks_added: Optional[int] = None# 初始化系统
config = RAGConfig()
rag_system = RAGSystem(config)# 创建FastAPI应用
app = FastAPI(title="RAG Question Answering System",description="A complete RAG system for document-based question answering",version="1.0.0"
)# 添加CORS中间件
app.add_middleware(CORSMiddleware,allow_origins=["*"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)@app.post("/upload", response_model=DocumentUploadResponse)
async def upload_document(file: UploadFile = File(...), title: Optional[str] = None,category: Optional[str] = None):"""上传文档到知识库"""try:# 保存上传文件file_path = f"temp_{file.filename}"with open(file_path, "wb") as buffer:content = await file.read()buffer.write(content)# 确定文件类型file_type = file.filename.split('.')[-1].lower()# 构建元数据metadata = {"title": title or file.filename,"category": category or "general","upload_time": datetime.now().isoformat()}# 处理文档result = rag_system.add_document(file_path, file_type, metadata)# 清理临时文件try:os.remove(file_path)except:passreturn DocumentUploadResponse(**result)except Exception as e:logging.error(f"Upload error: {e}")return DocumentUploadResponse(success=False, message=str(e))@app.post("/query", response_model=QueryResponse)
async def ask_question(request: QueryRequest):"""处理问答请求"""try:response = rag_system.query(request.question, request.search_type)return QueryResponse(answer=response.answer,sources=response.sources,confidence=response.confidence,processing_time=response.processing_time)except Exception as e:logging.error(f"Query error: {e}")raise HTTPException(status_code=500, detail=str(e))@app.get("/status")
async def get_status():"""获取系统状态"""return rag_system.get_system_status()@app.get("/health")
async def health_check():"""健康检查"""return {"status": "healthy","timestamp": datetime.now().isoformat(),"system": "RAG Question Answering System"}# ===== 命令行接口 =====class RAGCommandLine:"""RAG系统命令行界面"""def __init__(self):self.rag_system = RAGSystem(RAGConfig())def run_interactive(self):"""运行交互式问答"""print("=== RAG智能问答系统 ===")print("输入 'help' 查看帮助，输入 'quit' 退出")print()while True:try:user_input = input("请输入您的问题: ").strip()if user_input.lower() in ['quit', 'exit', '退出']:print("感谢使用！")breakelif user_input.lower() == 'help':self._show_help()elif user_input.lower() == 'status':self._show_status()elif user_input.lower().startswith('add '):self._add_document_interactive(user_input[4:])elif user_input:self._process_query(user_input)except KeyboardInterrupt:print("\n感谢使用！")breakexcept Exception as e:print(f"错误: {e}")def _show_help(self):"""显示帮助信息"""help_text = """
可用命令:
- help: 显示此帮助信息
- status: 显示系统状态
- add <文件路径>: 添加文档到知识库
- quit/exit: 退出系统直接输入问题即可开始问答。"""print(help_text)def _show_status(self):"""显示系统状态"""status = self.rag_system.get_system_status()print(f"知识库文档数: {status['database_stats']['total_chunks']}")print(f"向量维度: {status['database_stats']['vector_dimension']}")print(f"内存使用: {status['database_stats']['memory_usage_mb']:.2f} MB")def _add_document_interactive(self, file_path: str):"""交互式添加文档"""if not os.path.exists(file_path):print(f"文件不存在: {file_path}")returnfile_type = file_path.split('.')[-1].lower()result = self.rag_system.add_document(file_path, file_type)if result['success']:print(f"成功添加文档，生成 {result['chunks_added']} 个知识片段")else:print(f"添加失败: {result['message']}")def _process_query(self, question: str):"""处理查询"""print("正在搜索相关信息...")response = self.rag_system.query(question)print(f"\n答案 (置信度: {response.confidence:.2f}):")print("-" * 50)print(response.answer)if response.sources:print(f"\n参考来源 ({len(response.sources)}个):")for i, source in enumerate(response.sources, 1):source_file = source['metadata'].get('source_file', '未知来源')print(f"{i}. {source_file} (相关度: {source['score']:.3f})")print(f"\n处理用时: {response.processing_time:.2f}秒")print("-" * 50)# ===== 主程序 =====def start_web_server():"""启动Web服务器"""logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s')print("启动RAG问答系统Web服务...")print("API文档: http://localhost:8000/docs")print("系统状态: http://localhost:8000/status")uvicorn.run(app, host="0.0.0.0", port=8000)def start_cli():"""启动命令行界面"""logging.basicConfig(level=logging.WARNING)cli = RAGCommandLine()cli.run_interactive()if __name__ == "__main__":import sysif len(sys.argv) > 1 and sys.argv[1] == "web":start_web_server()else:start_cli()