145_RAG应用论文(论文中附有源码)：检索增强生成 - 2025年向量数据库与LLM深度集成实践指南

用户查询 → 检索模块 → 检索结果 + 用户查询 → LLM → 生成回答

用户查询 → 查询扩展 → 初步检索 → 重排序 → 上下文构建 → LLM → 生成回答 → 验证

# HNSW索引配置示例（以Qdrant为例）
index_params = {
    "hnsw_config": {
        "m": 16,               # 每个节点的最大连接数
        "ef_construction": 200, # 构建时的搜索宽度
        "full_scan_threshold": 10000, # 低于此值时执行全扫描
        "max_indexing_threads": 0, # 0表示使用所有可用线程
        "on_disk": False,     # 是否存储在磁盘上
        "payload_m": 1000     # 用于负载存储的内存量
    }
}

# IVF索引配置示例（以Milvus为例）
index_params = {
    "index_type": "IVF_FLAT",  # IVF索引类型
    "metric_type": "L2",       # 距离度量类型
    "params": {
        "nlist": 1024           # 聚类中心数量
    }
}

# 查询参数
search_params = {
    "metric_type": "L2",
    "params": {
        "nprobe": 64            # 查询时检查的聚类数量
    }
}

# 智能分块实现示例
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyPDFLoader

# 递归字符分块器配置
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=600,               # 块大小（字符数）
    chunk_overlap=100,            # 块重叠大小
    length_function=len,          # 长度计算函数
    separators=["\n\n", "\n", ". ", " ", ""],  # 分割符优先级
    is_separator_regex=False      # 是否将分隔符视为正则表达式
)

# 加载文档并分块
def process_document(document_path):
    loader = PyPDFLoader(document_path)
    documents = loader.load()
    chunks = text_splitter.split_documents(documents)
    
    # 为每个块添加元数据
    for i, chunk in enumerate(chunks):
        chunk.metadata["chunk_id"] = f"{chunk.metadata['source'].split('/')[-1]}_{i}"
        chunk.metadata["chunk_size"] = len(chunk.page_content)
        chunk.metadata["chunk_position"] = i
    
    return chunks

# 文档预处理工具函数
def clean_text(text):
    """清理文本内容，准备分块和嵌入"""
    import re
    from bs4 import BeautifulSoup
    
    # 移除HTML/XML标签
    soup = BeautifulSoup(text, "html.parser")
    text = soup.get_text(separator=" ")
    
    # 移除多余空白
    text = re.sub(r'\s+', ' ', text)
    
    # 标准化换行符
    text = text.replace('\r\n', '\n').replace('\r', '\n')
    
    # 移除多余换行
    text = re.sub(r'\n{3,}', '\n\n', text)
    
    # 移除控制字符（保留制表符和换行符）
    text = re.sub(r'[\x00-\x08\x0B\x0C\x0E-\x1F]', '', text)
    
    return text.strip()

# 优化的RAG提示模板示例
def generate_rag_prompt(query, context_chunks):
    """生成优化的RAG提示模板"""
    template = """
    你是一个专业的问答助手，擅长基于提供的上下文信息回答用户问题。
    
    请严格遵循以下指令：
    1. 只使用提供的上下文信息来回答问题
    2. 如果上下文信息不足以回答问题，请明确说明
    3. 对于事实性问题，请提供准确的答案并引用相关上下文
    4. 对于解释性问题，请提供详细的分析
    5. 请保持回答简洁明了，避免无关信息
    
    # 上下文信息
    {context}
    
    # 用户问题
    {query}
    
    # 回答格式
    答案：[你的回答]
    引用：[引用的上下文来源，如有多个请分别列出]
    """
    
    # 组织上下文信息，添加元数据
    context = "\n\n".join([
        f"[文档来源: {chunk.metadata.get('source', '未知')}, 章节: {chunk.metadata.get('section', '未知')}]\n{chunk.page_content}"
        for chunk in context_chunks
    ])
    
    return template.format(context=context, query=query)

# OpenAI模型集成示例
import openai
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")

def generate_answer_with_openai(prompt, model="gpt-4-turbo"):
    """使用OpenAI模型生成回答"""
    try:
        response = openai.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": "你是一个基于检索信息的专业问答助手。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.1,  # 低温度提高回答一致性
            max_tokens=1000,
            top_p=0.9,
            frequency_penalty=0.0,
            presence_penalty=0.0
        )
        return response.choices[0].message.content.strip()
    except Exception as e:
        print(f"OpenAI API错误: {e}")
        return "生成回答时发生错误。"

# Claude模型集成示例
import anthropic
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

def generate_answer_with_claude(prompt, model="claude-3-opus-20240229"):
    """使用Claude模型生成回答"""
    try:
        message = client.messages.create(
            model=model,
            max_tokens=2000,
            temperature=0.1,
            system="你是一个基于检索信息的专业问答助手。请严格基于提供的上下文信息回答问题。",
            messages=[{"role": "user", "content": prompt}]
        )
        return message.content[0].text.strip()
    except Exception as e:
        print(f"Claude API错误: {e}")
        return "生成回答时发生错误。"

# 本地开源模型集成示例（使用vllm）
from vllm import LLM, SamplingParams

def setup_local_llm(model_name="meta-llama/Meta-Llama-3-8B-Instruct"):
    """设置本地LLM模型"""
    llm = LLM(
        model=model_name,
        quantization="awq",  # 使用AWQ量化加速推理
        max_num_batched_tokens=4096,
        tensor_parallel_size=1
    )
    return llm

def generate_answer_with_local_model(llm, prompt):
    """使用本地模型生成回答"""
    sampling_params = SamplingParams(
        temperature=0.1,
        top_p=0.9,
        max_tokens=1000
    )
    
    # 构建提示格式（根据模型要求）
    formatted_prompt = f"""<|begin_of_text|><|start_header_id|>system<|end_header_id|>
    你是一个基于检索信息的专业问答助手。请严格基于提供的上下文信息回答问题。<|eot_id|><|start_header_id|>user<|end_header_id|>
    {prompt}<|eot_id|><|start_header_id|>assistant<|end_header_id|>"""
    
    try:
        outputs = llm.generate([formatted_prompt], sampling_params)
        return outputs[0].outputs[0].text.strip()
    except Exception as e:
        print(f"本地模型错误: {e}")
        return "生成回答时发生错误。"

# LLM重排序实现示例
def llm_reranking(query, retrieved_docs, top_k=5, model="gpt-4-turbo"):
    """使用LLM对检索结果进行重排序"""
    import openai
    
    # 构建重排序提示
    rerank_prompt = f"""
    请对以下检索文档按照与查询"{query}"的相关性进行排序。
    
    对于每个文档，请给出相关性评分（1-10分）和简短理由。
    最后，请按照评分降序排列文档编号。
    
    文档列表：
    """
    
    for i, doc in enumerate(retrieved_docs):
        rerank_prompt += f"\n[{i+1}] {doc.page_content[:300]}..."
    
    rerank_prompt += "\n\n请返回排序结果，格式为：排序后的文档编号列表"
    
    try:
        response = openai.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": rerank_prompt}],
            temperature=0
        )
        
        # 解析排序结果
        sorted_indices = parse_rerank_result(response.choices[0].message.content)
        
        # 返回重排序后的文档
        return [retrieved_docs[i-1] for i in sorted_indices[:top_k]]
    except Exception as e:
        print(f"重排序错误: {e}")
        return retrieved_docs[:top_k]  # 失败时返回原始排序

def parse_rerank_result(result_text):
    """解析重排序结果"""
    import re
    # 尝试提取数字列表
    numbers = re.findall(r'\d+', result_text)
    return [int(num) for num in numbers]

# 交叉编码器重排序示例
from sentence_transformers import CrossEncoder

def cross_encoder_reranking(query, retrieved_docs, top_k=5):
    """使用交叉编码器对检索结果进行重排序"""
    # 加载交叉编码器模型
    model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
    
    # 准备配对数据
    pairs = [[query, doc.page_content] for doc in retrieved_docs]
    
    # 计算相关性分数
    scores = model.predict(pairs)
    
    # 根据分数排序
    sorted_docs = [doc for _, doc in sorted(zip(scores, retrieved_docs), reverse=True)]
    
    return sorted_docs[:top_k]

# 混合重排序策略
def hybrid_reranking(query, retrieved_docs, top_k=5):
    """结合多种重排序方法"""
    # 1. 获取初步排序（向量相似度）
    initial_ranking = retrieved_docs
    
    # 2. 应用交叉编码器重排序
    cross_encoded_docs = cross_encoder_reranking(query, initial_ranking, top_k=2*top_k)
    
    # 3. 应用LLM重排序（对前2k结果）
    final_docs = llm_reranking(query, cross_encoded_docs, top_k=top_k)
    
    return final_docs

# 上下文压缩示例
def compress_context(context_chunks, max_tokens=4000, model="gpt-4-turbo"):
    """压缩上下文以适应模型窗口"""
    import tiktoken
    import openai
    
    # 计算当前token数量
    encoder = tiktoken.encoding_for_model(model)
    total_tokens = sum(len(encoder.encode(chunk.page_content)) for chunk in context_chunks)
    
    # 如果token数量超过限制，进行压缩
    if total_tokens > max_tokens:
        # 构建压缩提示
        compress_prompt = f"""
        请压缩以下文档内容，保留与查询相关的关键信息，同时保持信息的完整性和准确性。
        压缩后的内容应该可以直接用于回答用户问题。
        
        文档内容：
        """
        
        for chunk in context_chunks:
            compress_prompt += f"\n{chunk.page_content}"
        
        try:
            response = openai.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": compress_prompt}],
                max_tokens=max_tokens // 2  # 限制输出大小
            )
            
            # 返回压缩后的单一上下文块
            from langchain.schema import Document
            return [Document(page_content=response.choices[0].message.content, metadata={"compressed": True})]
        except Exception as e:
            print(f"上下文压缩错误: {e}")
            # 失败时，按原始顺序返回，直到达到token限制
            compressed_chunks = []
            current_tokens = 0
            for chunk in context_chunks:
                chunk_tokens = len(encoder.encode(chunk.page_content))
                if current_tokens + chunk_tokens <= max_tokens:
                    compressed_chunks.append(chunk)
                    current_tokens += chunk_tokens
                else:
                    # 尝试截断当前块
                    available_tokens = max_tokens - current_tokens
                    truncated_text = encoder.decode(encoder.encode(chunk.page_content)[:available_tokens])
                    truncated_chunk = Document(page_content=truncated_text, metadata=chunk.metadata)
                    compressed_chunks.append(truncated_chunk)
                    break
            return compressed_chunks
    
    return context_chunks

# 生成质量评估示例
import evaluate
from rouge_score import rouge_scorer

def evaluate_generation_quality(generated_answers, reference_answers):
    """评估生成回答的质量"""
    # 初始化评估指标
    rouge = evaluate.load('rouge')
    bleu = evaluate.load('bleu')
    meteor = evaluate.load('meteor')
    scorer = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)
    
    # 计算各项指标
    results = {
        'rouge': [],
        'bleu': [],
        'meteor': [],
        'exact_match': 0,
        'faithfulness': 0,  # 需要自定义实现或使用外部工具
        'relevance': 0      # 需要自定义实现或使用外部工具
    }
    
    # 逐对评估
    for gen, ref in zip(generated_answers, reference_answers):
        # 计算ROUGE分数
        rouge_scores = scorer.score(ref, gen)
        results['rouge'].append({
            'rouge1': rouge_scores['rouge1'].fmeasure,
            'rouge2': rouge_scores['rouge2'].fmeasure,
            'rougeL': rouge_scores['rougeL'].fmeasure
        })
        
        # 计算BLEU分数（需要分词）
        bleu_result = bleu.compute(predictions=[gen], references=[[ref]])
        results['bleu'].append(bleu_result['bleu'])
        
        # 计算METEOR分数
        meteor_result = meteor.compute(predictions=[gen], references=[ref])
        results['meteor'].append(meteor_result['meteor'])
        
        # 计算精确匹配
        if gen.strip() == ref.strip():
            results['exact_match'] += 1
    
    # 计算平均值
    avg_rouge = {
        'rouge1': sum(r['rouge1'] for r in results['rouge']) / len(results['rouge']),
        'rouge2': sum(r['rouge2'] for r in results['rouge']) / len(results['rouge']),
        'rougeL': sum(r['rougeL'] for r in results['rouge']) / len(results['rouge'])
    }
    
    return {
        'rouge': avg_rouge,
        'bleu': sum(results['bleu']) / len(results['bleu']),
        'meteor': sum(results['meteor']) / len(results['meteor']),
        'exact_match_rate': results['exact_match'] / len(generated_answers),
        'faithfulness': results['faithfulness'],
        'relevance': results['relevance']
    }

# 检索优化示例
def optimize_retrieval(query, retriever, top_k=5, rerank=True, hybrid=True):
    """优化检索过程"""
    # 1. 查询扩展
    expanded_queries = expand_query(query)
    
    # 2. 混合检索（向量搜索+关键词搜索）
    if hybrid:
        # 向量检索
        vector_results = retriever.similarity_search(query, k=top_k*2)
        
        # 关键词检索（假设有全文检索引擎）
        keyword_results = full_text_search(query, k=top_k*2)
        
        # 合并结果，去重
        combined_results = merge_and_deduplicate(vector_results, keyword_results)
    else:
        combined_results = retriever.similarity_search(query, k=top_k*2)
    
    # 3. 重排序
    if rerank and len(combined_results) > top_k:
        # 使用重排序技术
        final_results = llm_reranking(query, combined_results, top_k=top_k)
    else:
        final_results = combined_results[:top_k]
    
    return final_results

def expand_query(query):
    """查询扩展"""
    # 简单实现，可替换为更复杂的技术如词向量扩展
    synonyms = {
        "配置": ["设置", "部署", "配置"],
        "数据库": ["数据库", "存储", "数据仓库"],
        "集成": ["集成", "整合", "对接"]
    }
    
    expanded = [query]
    words = query.split()
    
    for i, word in enumerate(words):
        if word in synonyms:
            for syn in synonyms[word]:
                if syn != word:
                    new_query = " ".join(words[:i] + [syn] + words[i+1:])
                    expanded.append(new_query)
    
    return expanded

# 缓存系统实现示例
import hashlib
import pickle
import time
from functools import lru_cache

class RAGCache:
    """RAG系统缓存管理"""
    
    def __init__(self, max_size=1000, ttl=3600):
        """初始化缓存
        
        Args:
            max_size: 最大缓存条目数
            ttl: 缓存条目过期时间（秒）
        """
        self.max_size = max_size
        self.ttl = ttl
        self.cache = {}
        self.access_times = {}
    
    def _get_key(self, query, params):
        """生成缓存键"""
        data = f"{query}:{params}"
        return hashlib.md5(data.encode()).hexdigest()
    
    def get(self, query, params=None):
        """获取缓存项"""
        if params is None:
            params = {}
            
        key = self._get_key(query, params)
        
        if key in self.cache:
            value, timestamp = self.cache[key]
            # 检查是否过期
            if time.time() - timestamp <= self.ttl:
                # 更新访问时间
                self.access_times[key] = time.time()
                return value
            else:
                # 删除过期项
                del self.cache[key]
                if key in self.access_times:
                    del self.access_times[key]
        
        return None
    
    def set(self, query, result, params=None):
        """设置缓存项"""
        if params is None:
            params = {}
            
        key = self._get_key(query, params)
        
        # 检查是否需要清理缓存
        if len(self.cache) >= self.max_size and key not in self.cache:
            self._evict_oldest()
        
        # 存储结果和时间戳
        self.cache[key] = (result, time.time())
        self.access_times[key] = time.time()
    
    def _evict_oldest(self):
        """驱逐最久未使用的缓存项"""
        if not self.access_times:
            return
            
        # 找到最久未使用的键
        oldest_key = min(self.access_times.items(), key=lambda x: x[1])[0]
        
        # 删除
        del self.cache[oldest_key]
        del self.access_times[oldest_key]
    
    def clear(self):
        """清空缓存"""
        self.cache.clear()
        self.access_times.clear()
    
    def save(self, filepath):
        """保存缓存到文件"""
        data = {
            'cache': self.cache,
            'access_times': self.access_times
        }
        with open(filepath, 'wb') as f:
            pickle.dump(data, f)
    
    def load(self, filepath):
        """从文件加载缓存"""
        try:
            with open(filepath, 'rb') as f:
                data = pickle.load(f)
                self.cache = data.get('cache', {})
                self.access_times = data.get('access_times', {})
                # 清理过期项
                current_time = time.time()
                expired_keys = [k for k, (_, t) in self.cache.items() 
                              if current_time - t > self.ttl]
                for key in expired_keys:
                    del self.cache[key]
                    if key in self.access_times:
                        del self.access_times[key]
        except FileNotFoundError:
            pass

# 并行处理优化示例
import concurrent.futures
from functools import partial

def parallel_retrieval(queries, retriever, top_k=5, max_workers=4):
    """并行执行多个查询的检索"""
    # 创建部分函数，固定top_k参数
    retrieval_func = partial(retriever.similarity_search, k=top_k)
    
    # 并行执行
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务
        futures = {executor.submit(retrieval_func, query): query for query in queries}
        
        # 收集结果
        results = {}
        for future in concurrent.futures.as_completed(futures):
            query = futures[future]
            try:
                results[query] = future.result()
            except Exception as e:
                print(f"查询 '{query}' 执行出错: {e}")
                results[query] = []
    
    return results

# 知识图谱增强RAG示例
class KnowledgeGraphEnhancedRAG:
    """知识图谱增强的RAG系统"""
    
    def __init__(self, vector_retriever, kg_retriever, llm):
        """初始化系统
        
        Args:
            vector_retriever: 向量检索器
            kg_retriever: 知识图谱检索器
            llm: 语言模型
        """
        self.vector_retriever = vector_retriever
        self.kg_retriever = kg_retriever
        self.llm = llm
    
    def retrieve(self, query, top_k=5):
        """混合检索
        
        Args:
            query: 用户查询
            top_k: 每类检索返回的结果数量
            
        Returns:
            合并后的检索结果
        """
        # 1. 向量检索
        vector_results = self.vector_retriever.similarity_search(query, k=top_k)
        
        # 2. 知识图谱检索
        kg_results = self.kg_retriever.query(query, top_k=top_k)
        
        # 3. 合并结果
        merged_results = self._merge_results(vector_results, kg_results)
        
        return merged_results
    
    def _merge_results(self, vector_results, kg_results):
        """合并不同类型的检索结果"""
        # 为向量检索结果添加类型标记
        for doc in vector_results:
            doc.metadata['type'] = 'text'
        
        # 为知识图谱结果添加类型标记并转换为Document格式
        kg_docs = []
        for triple in kg_results:
            # 转换三元组为文本描述
            content = f"{triple['subject']} {triple['predicate']} {triple['object']}"
            doc = Document(page_content=content, metadata={
                'type': 'knowledge_graph',
                'subject': triple['subject'],
                'predicate': triple['predicate'],
                'object': triple['object']
            })
            kg_docs.append(doc)
        
        # 合并并去重
        all_docs = vector_results + kg_docs
        # 简单去重 - 实际应用中可能需要更复杂的逻辑
        unique_docs = []
        seen = set()
        for doc in all_docs:
            content_hash = hashlib.md5(doc.page_content.encode()).hexdigest()
            if content_hash not in seen:
                seen.add(content_hash)
                unique_docs.append(doc)
        
        return unique_docs
    
    def generate_with_kg(self, query, context):
        """使用知识图谱增强的提示生成回答"""
        # 构建包含知识图谱信息的提示
        kg_info = "\n知识图谱信息:\n"
        text_info = "\n文本信息:\n"
        
        for doc in context:
            if doc.metadata.get('type') == 'knowledge_graph':
                kg_info += f"- {doc.page_content}\n"
            else:
                text_info += f"- {doc.page_content[:200]}...\n"
        
        prompt = f"""
        请根据以下知识图谱信息和文本信息回答用户问题。
        知识图谱信息提供了实体之间的关系，可以帮助回答涉及概念和关系的问题。
        文本信息提供了详细的上下文，可以帮助回答具体细节问题。
        
        {kg_info}
        {text_info}
        
        用户问题: {query}
        
        请提供全面、准确的回答，同时引用使用的信息类型。
        """
        
        # 生成回答
        response = self.llm.generate(prompt)
        
        return response

# 多模态RAG示例框架
class MultimodalRAG:
    """多模态RAG系统"""
    
    def __init__(self, text_retriever, image_retriever, multimodal_llm):
        """初始化多模态RAG系统
        
        Args:
            text_retriever: 文本检索器
            image_retriever: 图像检索器
            multimodal_llm: 多模态LLM
        """
        self.text_retriever = text_retriever
        self.image_retriever = image_retriever
        self.llm = multimodal_llm
    
    def retrieve(self, query, top_k=3):
        """执行多模态检索
        
        Args:
            query: 用户查询
            top_k: 每类检索返回的结果数量
            
        Returns:
            文本和图像检索结果
        """
        # 执行文本检索
        text_results = self.text_retriever.similarity_search(query, k=top_k)
        
        # 执行图像检索
        image_results = self.image_retriever.search(query, k=top_k)
        
        return {
            'text': text_results,
            'images': image_results
        }
    
    def generate_answer(self, query, context):
        """基于多模态上下文生成回答"""
        # 构建多模态提示
        messages = [
            {"role": "system", "content": "你是一个多模态问答助手，可以理解文本和图像内容。"},
            {"role": "user", "content": query}
        ]
        
        # 添加文本上下文
        if context.get('text'):
            text_context = "\n文本信息:\n"
            for i, doc in enumerate(context['text']):
                text_context += f"[{i+1}] {doc.page_content[:300]}...\n"
            messages[1]['content'] += text_context
        
        # 生成回答
        response = self.llm.generate(multimodal_messages=messages,
                                    images=context.get('images', []))
        
        return response

# 自适应RAG示例
class AdaptiveRAG:
    """自适应RAG系统"""
    
    def __init__(self, base_rag, query_classifier, performance_tracker):
        """初始化自适应RAG系统
        
        Args:
            base_rag: 基础RAG系统
            query_classifier: 查询分类器
            performance_tracker: 性能跟踪器
        """
        self.base_rag = base_rag
        self.query_classifier = query_classifier
        self.performance_tracker = performance_tracker
    
    def process_query(self, query, user_id=None):
        """处理用户查询并自适应调整参数
        
        Args:
            query: 用户查询
            user_id: 用户ID（可选，用于个性化）
            
        Returns:
            生成的回答
        """
        # 1. 分类查询类型
        query_type = self.query_classifier.classify(query)
        
        # 2. 根据查询类型和历史性能调整参数
        params = self._adjust_parameters(query_type, user_id)
        
        # 3. 执行RAG处理
        response = self.base_rag.process(query, **params)
        
        # 4. 记录性能
        self.performance_tracker.record(query, query_type, params, response)
        
        return response
    
    def _adjust_parameters(self, query_type, user_id=None):
        """根据查询类型和历史性能调整参数
        
        Args:
            query_type: 查询类型
            user_id: 用户ID
            
        Returns:
            调整后的参数
        """
        # 基础参数
        params = {
            'top_k': 5,
            'rerank': True,
            'max_tokens': 1000
        }
        
        # 根据查询类型调整
        if query_type == 'factual':
            # 事实查询 - 提高精确性
            params.update({
                'top_k': 3,
                'temperature': 0.1,
                'strict_context': True
            })
        elif query_type == 'complex':
            # 复杂查询 - 增加上下文和生成长度
            params.update({
                'top_k': 7,
                'temperature': 0.2,
                'max_tokens': 2000,
                'use_chain_of_thought': True
            })
        elif query_type == 'creative':
            # 创意查询 - 增加创造性
            params.update({
                'top_k': 5,
                'temperature': 0.7,
                'max_tokens': 1500
            })
        
        # 获取历史性能数据并进一步调整
        performance_data = self.performance_tracker.get_performance(
            query_type=query_type, user_id=user_id
        )
        
        if performance_data:
            # 根据历史成功率调整参数
            if performance_data['success_rate'] < 0.7:
                # 降低温度，提高精确性
                if 'temperature' in params:
                    params['temperature'] = max(0.1, params['temperature'] - 0.2)
                # 增加检索数量
                params['top_k'] = min(10, params['top_k'] + 2)
            elif performance_data['success_rate'] > 0.9:
                # 可以适当提高效率
                params['top_k'] = max(3, params['top_k'] - 1)
        
        return params

## 6. RAG系统部署与应用案例

### 6.1 部署架构设计

2025年，RAG系统的部署架构需要考虑高性能、高可用性和可扩展性：

#### 微服务架构

#### 关键组件配置

```yaml
# Docker Compose示例配置
version: '3.8'

services:
  # API网关
  api-gateway:
    image: nginx:latest
    ports:
      - "80:80"
    volumes:
      - ./nginx/conf.d:/etc/nginx/conf.d
    depends_on:
      - rag-service
    restart: always
  
  # RAG服务
  rag-service:
    build: ./rag-service
    environment:
      - VECTOR_DB_URL=http://vector-db:8900
      - LLM_SERVICE_URL=http://llm-service:8000
      - CACHE_URL=redis://cache:6379
    depends_on:
      - vector-db
      - llm-service
      - cache
    deploy:
      replicas: 3
      resources:
        limits:
          cpus: '2'
          memory: 4G
  
  # LLM服务
  llm-service:
    build: ./llm-service
    environment:
      - MODEL_NAME=gpt-4-turbo
      - API_KEY=${OPENAI_API_KEY}
    deploy:
      replicas: 2
      resources:
        limits:
          cpus: '4'
          memory: 8G
  
  # 向量数据库
  vector-db:
    image: pinecone/pinecone-server:latest
    environment:
      - PINECONE_API_KEY=${PINECONE_API_KEY}
      - INDEX_NAME=rag-index
    volumes:
      - vector-data:/data
    deploy:
      resources:
        limits:
          cpus: '4'
          memory: 16G
  
  # 缓存服务
  cache:
    image: redis:latest
    volumes:
      - redis-data:/data
    deploy:
      resources:
        limits:
          cpus: '1'
          memory: 2G
  
  # 监控服务
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus/prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus-data:/prometheus
    ports:
      - "9090:9090"
  
  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    depends_on:
      - prometheus
    volumes:
      - grafana-data:/var/lib/grafana

volumes:
  vector-data:
  redis-data:
  prometheus-data:
  grafana-data:

# Prometheus监控指标定义示例
from prometheus_client import Counter, Histogram, Gauge, start_http_server

# 定义指标
REQUEST_COUNT = Counter('rag_requests_total', 'Total RAG requests', ['endpoint', 'status'])
RESPONSE_TIME = Histogram('rag_response_time_seconds', 'RAG response time', ['endpoint'])
CACHE_HIT_RATE = Gauge('rag_cache_hit_rate', 'RAG cache hit rate')
RETRIEVAL_QUALITY = Gauge('rag_retrieval_quality', 'RAG retrieval quality score')
LLM_ERROR_RATE = Gauge('rag_llm_error_rate', 'LLM error rate')

# 启动指标服务器
start_http_server(8000)

# 自动扩缩容脚本示例
import requests
import time
import json

class AutoScaler:
    """自动扩缩容管理器"""
    
    def __init__(self, prometheus_url, threshold_cpu=70, threshold_memory=75, 
                 min_instances=2, max_instances=10, check_interval=60):
        """初始化自动扩缩容管理器
        
        Args:
            prometheus_url: Prometheus服务地址
            threshold_cpu: CPU使用率阈值(%)
            threshold_memory: 内存使用率阈值(%)
            min_instances: 最小实例数
            max_instances: 最大实例数
            check_interval: 检查间隔(秒)
        """
        self.prometheus_url = prometheus_url
        self.threshold_cpu = threshold_cpu
        self.threshold_memory = threshold_memory
        self.min_instances = min_instances
        self.max_instances = max_instances
        self.check_interval = check_interval
    
    def get_metrics(self):
        """获取监控指标"""
        # 获取CPU使用率
        cpu_query = 'avg(rate(container_cpu_usage_seconds_total{name="rag-service"}[5m])) * 100'
        cpu_response = requests.get(
            f"{self.prometheus_url}/api/v1/query",
            params={"query": cpu_query}
        )
        cpu_data = cpu_response.json()
        cpu_usage = float(cpu_data['data']['result'][0]['value'][1]) if cpu_data['data']['result'] else 0
        
        # 获取内存使用率
        mem_query = 'avg(container_memory_usage_bytes{name="rag-service"} / container_spec_memory_limit_bytes{name="rag-service"}) * 100'
        mem_response = requests.get(
            f"{self.prometheus_url}/api/v1/query",
            params={"query": mem_query}
        )
        mem_data = mem_response.json()
        mem_usage = float(mem_data['data']['result'][0]['value'][1]) if mem_data['data']['result'] else 0
        
        # 获取当前实例数
        instances_query = 'count(up{job="rag-service"})'
        instances_response = requests.get(
            f"{self.prometheus_url}/api/v1/query",
            params={"query": instances_query}
        )
        instances_data = instances_response.json()
        current_instances = int(instances_data['data']['result'][0]['value'][1]) if instances_data['data']['result'] else 0
        
        return {
            "cpu_usage": cpu_usage,
            "memory_usage": mem_usage,
            "current_instances": current_instances
        }
    
    def scale(self, metrics):
        """根据指标决定扩缩容操作"""
        cpu_usage = metrics["cpu_usage"]
        memory_usage = metrics["memory_usage"]
        current = metrics["current_instances"]
        
        # 决定扩缩容操作
        if (cpu_usage > self.threshold_cpu or memory_usage > self.threshold_memory) and current < self.max_instances:
            # 扩展 - 增加1个实例
            new_count = min(current + 1, self.max_instances)
            self.update_instances(new_count)
            print(f"扩展实例数: {current} -> {new_count}")
        elif (cpu_usage < self.threshold_cpu * 0.5 and memory_usage < self.threshold_memory * 0.5) and current > self.min_instances:
            # 缩容 - 减少1个实例
            new_count = max(current - 1, self.min_instances)
            self.update_instances(new_count)
            print(f"缩容实例数: {current} -> {new_count}")
    
    def update_instances(self, count):
        """更新实例数量（实际环境中调用K8s API或其他编排工具）"""
        # 示例：调用Docker Compose API更新实例数
        # 实际实现需要根据部署环境调整
        pass
    
    def run(self):
        """启动自动扩缩容循环"""
        print("启动自动扩缩容服务...")
        while True:
            try:
                metrics = self.get_metrics()
                print(f"当前指标 - CPU: {metrics['cpu_usage']:.2f}%, 内存: {metrics['memory_usage']:.2f}%, 实例数: {metrics['current_instances']}")
                self.scale(metrics)
            except Exception as e:
                print(f"自动扩缩容检查失败: {e}")
            
            time.sleep(self.check_interval)

故障排查决策树:

1. 用户报告响应缓慢
   ├─ 检查系统负载
   │  ├─ 负载高 → 检查资源使用情况
   │  │  ├─ CPU高 → 检查计算密集型操作
   │  │  ├─ 内存高 → 检查内存泄漏
   │  │  └─ 网络高 → 检查网络连接
   │  └─ 负载正常 → 检查RAG流程各阶段
   │     ├─ 检索慢 → 检查向量数据库
   │     └─ 生成慢 → 检查LLM服务
   └─ 检查缓存状态
      ├─ 缓存未命中 → 优化缓存策略
      └─ 缓存命中 → 检查其他组件

### 6.5 RAG系统MVP实现

基于前述内容，下面提供一个完整的RAG系统最小可行产品(MVP)实现。这个MVP涵盖了从文档处理、向量数据库配置到LLM集成的完整流程，并提供了一个简单的Web界面。

#### MVP系统架构

#### 1. 项目结构

#### 2. 依赖配置 (requirements.txt)

```python
# requirements.txt
fastapi==0.104.1
uvicorn==0.24.0.post1
langchain==0.1.3
langchain-openai==0.0.5
pinecone-client==2.2.2
pydantic==2.5.0
python-dotenv==1.0.0
sentence-transformers==2.2.2
pandas==2.1.4
matplotlib==3.8.2
pillow==10.1.0
tiktoken==0.5.1

# config.py
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

class Config:
    # OpenAI配置
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "your-api-key")
    OPENAI_MODEL = "gpt-4-turbo"
    
    # Pinecone配置
    PINECONE_API_KEY = os.getenv("PINECONE_API_KEY", "your-pinecone-key")
    PINECONE_ENVIRONMENT = os.getenv("PINECONE_ENVIRONMENT", "gcp-starter")
    PINECONE_INDEX_NAME = "rag-mvp-index"
    
    # 嵌入模型配置
    EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
    EMBEDDING_DIMENSIONS = 384
    
    # 分块配置
    CHUNK_SIZE = 512
    CHUNK_OVERLAP = 50
    
    # 检索配置
    TOP_K = 5
    
    # Web服务配置
    HOST = "0.0.0.0"
    PORT = 8000
    
    # 缓存配置
    CACHE_TTL = 3600  # 1小时
    
    # 数据目录
    DATA_DIR = "./data"
    
    # 日志级别
    LOG_LEVEL = "INFO"

# 创建全局配置实例
config = Config()

# data_processing/document_loader.py
import os
from langchain.document_loaders import (PDFLoader, TextLoader, 
                                     Docx2txtLoader, UnstructuredExcelLoader)
from typing import List
from langchain.schema import Document

class DocumentLoader:
    """文档加载器，支持多种格式"""
    
    @staticmethod
    def load_document(file_path: str) -> List[Document]:
        """加载单个文档
        
        Args:
            file_path: 文档路径
            
        Returns:
            文档对象列表
        """
        file_extension = os.path.splitext(file_path)[1].lower()
        
        try:
            if file_extension == ".pdf":
                loader = PDFLoader(file_path)
            elif file_extension == ".txt":
                loader = TextLoader(file_path, encoding="utf-8")
            elif file_extension == ".docx":
                loader = Docx2txtLoader(file_path)
            elif file_extension in [".xlsx", ".xls"]:
                loader = UnstructuredExcelLoader(file_path)
            else:
                raise ValueError(f"不支持的文件格式: {file_extension}")
            
            return loader.load()
        except Exception as e:
            print(f"加载文档失败 {file_path}: {str(e)}")
            return []
    
    @staticmethod
    def load_documents_from_directory(directory_path: str) -> List[Document]:
        """加载目录中的所有文档
        
        Args:
            directory_path: 目录路径
            
        Returns:
            文档对象列表
        """
        documents = []
        
        if not os.path.exists(directory_path):
            print(f"目录不存在: {directory_path}")
            return documents
        
        for root, _, files in os.walk(directory_path):
            for file in files:
                file_path = os.path.join(root, file)
                docs = DocumentLoader.load_document(file_path)
                documents.extend(docs)
        
        return documents

# data_processing/chunker.py
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
from typing import List

class DocumentChunker:
    """文档分块器"""
    
    def __init__(self, chunk_size: int = 512, chunk_overlap: int = 50):
        """初始化分块器
        
        Args:
            chunk_size: 块大小（字符数）
            chunk_overlap: 块重叠大小（字符数）
        """
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", " ", "", "。", "！", "？"]  # 支持中英文分割
        )
    
    def split_document(self, document: Document) -> List[Document]:
        """分割单个文档
        
        Args:
            document: 文档对象
            
        Returns:
            分割后的文档块列表
        """
        return self.text_splitter.split_documents([document])
    
    def split_documents(self, documents: List[Document]) -> List[Document]:
        """分割多个文档
        
        Args:
            documents: 文档对象列表
            
        Returns:
            分割后的文档块列表
        """
        return self.text_splitter.split_documents(documents)
    
    def add_chunk_metadata(self, chunks: List[Document]) -> List[Document]:
        """为文档块添加额外元数据
        
        Args:
            chunks: 文档块列表
            
        Returns:
            添加元数据后的文档块列表
        """
        for i, chunk in enumerate(chunks):
            chunk.metadata["chunk_id"] = i
            chunk.metadata["chunk_size"] = len(chunk.page_content)
        
        return chunks

# vector_db/db_manager.py
import pinecone
from langchain.vectorstores import Pinecone
from langchain.embeddings import HuggingFaceEmbeddings
from typing import List, Optional
from langchain.schema import Document
from config import config
import time

class VectorDBManager:
    """向量数据库管理器"""
    
    def __init__(self):
        """初始化向量数据库管理器"""
        # 初始化嵌入模型
        self.embeddings = HuggingFaceEmbeddings(
            model_name=config.EMBEDDING_MODEL,
            model_kwargs={'device': 'cpu'},
            encode_kwargs={'normalize_embeddings': True}
        )
        
        # 初始化Pinecone
        pinecone.init(
            api_key=config.PINECONE_API_KEY,
            environment=config.PINECONE_ENVIRONMENT
        )
        
        # 检查并创建索引
        self._ensure_index_exists()
        
        # 创建向量存储
        self.vector_store = Pinecone.from_existing_index(
            index_name=config.PINECONE_INDEX_NAME,
            embedding=self.embeddings
        )
    
    def _ensure_index_exists(self):
        """确保索引存在，如果不存在则创建"""
        if config.PINECONE_INDEX_NAME not in pinecone.list_indexes():
            print(f"创建索引: {config.PINECONE_INDEX_NAME}")
            pinecone.create_index(
                name=config.PINECONE_INDEX_NAME,
                dimension=config.EMBEDDING_DIMENSIONS,
                metric="cosine"
            )
            # 等待索引创建完成
            time.sleep(1)
    
    def add_documents(self, documents: List[Document]) -> List[str]:
        """添加文档到向量数据库
        
        Args:
            documents: 文档对象列表
            
        Returns:
            添加的文档ID列表
        """
        try:
            return self.vector_store.add_documents(documents)
        except Exception as e:
            print(f"添加文档到向量数据库失败: {str(e)}")
            return []
    
    def similarity_search(self, query: str, k: int = 5) -> List[Document]:
        """相似性搜索
        
        Args:
            query: 查询文本
            k: 返回结果数量
            
        Returns:
            搜索结果列表
        """
        try:
            return self.vector_store.similarity_search(query, k=k)
        except Exception as e:
            print(f"相似性搜索失败: {str(e)}")
            return []
    
    def hybrid_search(self, query: str, k: int = 5, 
                     filter_criteria: Optional[dict] = None) -> List[Document]:
        """混合搜索（向量搜索 + 元数据过滤）
        
        Args:
            query: 查询文本
            k: 返回结果数量
            filter_criteria: 元数据过滤条件
            
        Returns:
            搜索结果列表
        """
        try:
            if filter_criteria:
                return self.vector_store.similarity_search(
                    query, k=k, filter=filter_criteria
                )
            else:
                return self.vector_store.similarity_search(query, k=k)
        except Exception as e:
            print(f"混合搜索失败: {str(e)}")
            return []
    
    def delete_all_documents(self):
        """删除所有文档"""
        try:
            # Pinecone目前不支持直接删除所有文档
            # 这里我们可以重建索引来实现
            pinecone.delete_index(config.PINECONE_INDEX_NAME)
            self._ensure_index_exists()
            # 重新初始化向量存储
            self.vector_store = Pinecone.from_existing_index(
                index_name=config.PINECONE_INDEX_NAME,
                embedding=self.embeddings
            )
            print("所有文档已删除")
        except Exception as e:
            print(f"删除所有文档失败: {str(e)}")
    
    def close(self):
        """关闭Pinecone连接"""
        pinecone.deinit()

# llm_integration/llm_client.py
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from typing import Dict, Any, Optional
from config import config

class LLMTool:
    """LLM工具类"""
    
    def __init__(self):
        """初始化LLM工具"""
        self.llm = ChatOpenAI(
            api_key=config.OPENAI_API_KEY,
            model=config.OPENAI_MODEL,
            temperature=0.1,  # 低温度提高回答一致性
            max_tokens=1000
        )
        
        # 初始化RAG提示模板
        self.rag_prompt_template = ChatPromptTemplate.from_template("""
        你是一个专业的问答助手，擅长基于提供的上下文信息回答用户问题。
        
        请严格遵循以下指令：
        1. 只使用提供的上下文信息来回答问题
        2. 如果上下文信息不足以回答问题，请明确说明
        3. 对于事实性问题，请提供准确的答案并引用相关上下文
        4. 对于解释性问题，请提供详细的分析
        5. 请保持回答简洁明了，避免无关信息
        
        # 上下文信息
        {context}
        
        # 用户问题
        {query}
        
        # 回答格式
        答案：[你的回答]
        引用：[引用的上下文来源，如有多个请分别列出]
        """)
    
    def generate_answer(self, query: str, context: str) -> Dict[str, Any]:
        """生成回答
        
        Args:
            query: 用户查询
            context: 上下文信息
            
        Returns:
            包含生成回答的字典
        """
        try:
            # 格式化提示
            prompt = self.rag_prompt_template.format(
                context=context,
                query=query
            )
            
            # 生成回答
            response = self.llm.predict(prompt)
            
            return {
                "success": True,
                "answer": response,
                "error": None
            }
        except Exception as e:
            print(f"生成回答失败: {str(e)}")
            return {
                "success": False,
                "answer": None,
                "error": str(e)
            }
    
    def generate_with_custom_prompt(self, prompt: str, **kwargs) -> Dict[str, Any]:
        """使用自定义提示生成内容
        
        Args:
            prompt: 自定义提示
            **kwargs: 额外参数
            
        Returns:
            包含生成内容的字典
        """
        try:
            # 生成内容
            response = self.llm.predict(prompt, **kwargs)
            
            return {
                "success": True,
                "content": response,
                "error": None
            }
        except Exception as e:
            print(f"使用自定义提示生成内容失败: {str(e)}")
            return {
                "success": False,
                "content": None,
                "error": str(e)
            }
    
    def rerank_documents(self, query: str, documents: list) -> list:
        """使用LLM对文档进行重排序
        
        Args:
            query: 用户查询
            documents: 文档列表
            
        Returns:
            重排序后的文档列表
        """
        # 构建重排序提示
        rerank_prompt = f"""
        请对以下检索文档按照与查询"{query}"的相关性进行排序。
        
        对于每个文档，请给出相关性评分（1-10分）和简短理由。
        最后，请按照评分降序排列文档编号。
        
        文档列表：
        """
        
        for i, doc in enumerate(documents):
            content_preview = doc.page_content[:300] + ("..." if len(doc.page_content) > 300 else "")
            rerank_prompt += f"\n[{i+1}] {content_preview}"
        
        rerank_prompt += "\n\n请返回排序结果，格式为：排序后的文档编号列表"
        
        # 获取重排序结果
        result = self.generate_with_custom_prompt(rerank_prompt, temperature=0)
        
        if result["success"]:
            # 解析结果（简单实现，实际应用中可能需要更复杂的解析）
            import re
            sorted_indices = re.findall(r'\d+', result["content"])
            sorted_indices = [int(idx) - 1 for idx in sorted_indices if idx.isdigit()]
            
            # 确保索引有效
            valid_indices = [idx for idx in sorted_indices if 0 <= idx < len(documents)]
            
            # 返回重排序后的文档
            if valid_indices:
                return [documents[idx] for idx in valid_indices]
        
        # 失败时返回原始文档
        return documents

# rag_service/retriever.py
from typing import List, Dict, Any, Optional
from langchain.schema import Document
from vector_db.db_manager import VectorDBManager
from llm_integration.llm_client import LLMTool

class RAGRetriever:
    """RAG检索器"""
    
    def __init__(self, vector_db_manager: VectorDBManager, 
                 llm_tool: LLMTool, top_k: int = 5, 
                 use_reranking: bool = True):
        """初始化检索器
        
        Args:
            vector_db_manager: 向量数据库管理器
            llm_tool: LLM工具
            top_k: 返回结果数量
            use_reranking: 是否使用重排序
        """
        self.vector_db_manager = vector_db_manager
        self.llm_tool = llm_tool
        self.top_k = top_k
        self.use_reranking = use_reranking
    
    def retrieve(self, query: str, 
                filter_criteria: Optional[Dict[str, Any]] = None) -> List[Document]:
        """检索相关文档
        
        Args:
            query: 用户查询
            filter_criteria: 过滤条件
            
        Returns:
            检索到的文档列表
        """
        # 执行相似性搜索
        documents = self.vector_db_manager.hybrid_search(
            query=query,
            k=self.top_k * 2,  # 获取更多结果用于重排序
            filter_criteria=filter_criteria
        )
        
        # 执行重排序
        if self.use_reranking and documents:
            documents = self.llm_tool.rerank_documents(query, documents)
        
        # 返回前top_k个结果
        return documents[:self.top_k]
    
    def format_documents_for_context(self, documents: List[Document]) -> str:
        """格式化文档用于上下文
        
        Args:
            documents: 文档列表
            
        Returns:
            格式化后的上下文文本
        """
        context_parts = []
        
        for i, doc in enumerate(documents):
            # 获取文档元数据
            source = doc.metadata.get('source', '未知来源')
            page = doc.metadata.get('page', '未知页')
            
            # 格式化文档内容
            context_part = f"""
[文档 {i+1}]来源: {source}, 页: {page}
内容: {doc.page_content}
"""
            context_parts.append(context_part)
        
        return "\n".join(context_parts)

# rag_service/generator.py
from typing import Dict, Any, List
from langchain.schema import Document
from llm_integration.llm_client import LLMTool

class RAGGenerator:
    """RAG生成器"""
    
    def __init__(self, llm_tool: LLMTool):
        """初始化生成器
        
        Args:
            llm_tool: LLM工具
        """
        self.llm_tool = llm_tool
    
    def generate(self, query: str, documents: List[Document]) -> Dict[str, Any]:
        """基于检索到的文档生成回答
        
        Args:
            query: 用户查询
            documents: 检索到的文档列表
            
        Returns:
            生成结果字典
        """
        # 格式化上下文
        context = "\n\n".join([f"[文档{i+1}]: {doc.page_content}" 
                             for i, doc in enumerate(documents)])
        
        # 生成回答
        result = self.llm_tool.generate_answer(query, context)
        
        # 添加源文档信息
        if result["success"]:
            source_info = []
            for i, doc in enumerate(documents):
                source_info.append({
                    "id": i+1,
                    "content": doc.page_content,
                    "metadata": doc.metadata
                })
            
            result["sources"] = source_info
        
        return result
    
    def generate_with_feedback(self, query: str, documents: List[Document],
                             feedback: str) -> Dict[str, Any]:
        """基于用户反馈优化生成回答
        
        Args:
            query: 用户查询
            documents: 检索到的文档列表
            feedback: 用户反馈
            
        Returns:
            优化后的生成结果
        """
        # 格式化上下文
        context = "\n\n".join([f"[文档{i+1}]: {doc.page_content}" 
                             for i, doc in enumerate(documents)])
        
        # 创建带有反馈的提示
        enhanced_prompt = f"""
用户问题: {query}

上下文信息:
{context}

用户反馈: {feedback}

请根据用户反馈和提供的上下文，重新生成更准确、更全面的回答。
"""
        
        # 生成优化回答
        result = self.llm_tool.generate_with_custom_prompt(enhanced_prompt)
        
        return {
            "success": result["success"],
            "answer": result.get("content"),
            "error": result.get("error")
        }

# rag_service/rag_engine.py
from typing import Dict, Any, List, Optional
from langchain.schema import Document
from data_processing.document_loader import DocumentLoader
from data_processing.chunker import DocumentChunker
from vector_db.db_manager import VectorDBManager
from llm_integration.llm_client import LLMTool
from rag_service.retriever import RAGRetriever
from rag_service.generator import RAGGenerator
from config import config

class RAGEngine:
    """RAG引擎"""
    
    def __init__(self):
        """初始化RAG引擎"""
        # 初始化各个组件
        self.document_loader = DocumentLoader()
        self.document_chunker = DocumentChunker(
            chunk_size=config.CHUNK_SIZE,
            chunk_overlap=config.CHUNK_OVERLAP
        )
        self.vector_db_manager = VectorDBManager()
        self.llm_tool = LLMTool()
        self.retriever = RAGRetriever(
            vector_db_manager=self.vector_db_manager,
            llm_tool=self.llm_tool,
            top_k=config.TOP_K
        )
        self.generator = RAGGenerator(llm_tool=self.llm_tool)
    
    def index_documents(self, directory_path: str) -> Dict[str, Any]:
        """索引目录中的文档
        
        Args:
            directory_path: 文档目录路径
            
        Returns:
            索引结果统计
        """
        # 加载文档
        documents = self.document_loader.load_documents_from_directory(directory_path)
        
        if not documents:
            return {
                "success": False,
                "message": "未找到或加载任何文档",
                "stats": None
            }
        
        # 分块
        chunks = self.document_chunker.split_documents(documents)
        chunks = self.document_chunker.add_chunk_metadata(chunks)
        
        # 索引到向量数据库
        document_ids = self.vector_db_manager.add_documents(chunks)
        
        return {
            "success": True,
            "message": f"成功索引 {len(document_ids)} 个文档块",
            "stats": {
                "total_documents": len(documents),
                "total_chunks": len(chunks),
                "indexed_chunks": len(document_ids)
            }
        }
    
    def query(self, query_text: str, 
              filter_criteria: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """执行查询
        
        Args:
            query_text: 查询文本
            filter_criteria: 过滤条件
            
        Returns:
            查询结果
        """
        # 检索相关文档
        retrieved_documents = self.retriever.retrieve(
            query=query_text,
            filter_criteria=filter_criteria
        )
        
        if not retrieved_documents:
            return {
                "success": False,
                "answer": "未找到相关信息来回答您的问题。",
                "sources": []
            }
        
        # 生成回答
        result = self.generator.generate(query_text, retrieved_documents)
        
        return {
            "success": result["success"],
            "answer": result.get("answer"),
            "sources": result.get("sources", []),
            "error": result.get("error")
        }
    
    def get_stats(self) -> Dict[str, Any]:
        """获取系统统计信息
        
        Returns:
            系统统计信息
        """
        # 注意：Pinecone API限制了获取索引统计的能力
        # 这里返回基本配置信息
        return {
            "config": {
                "embedding_model": config.EMBEDDING_MODEL,
                "llm_model": config.OPENAI_MODEL,
                "chunk_size": config.CHUNK_SIZE,
                "top_k": config.TOP_K
            },
            "vector_store": config.PINECONE_INDEX_NAME
        }
    
    def clear_index(self) -> Dict[str, Any]:
        """清除索引中的所有文档
        
        Returns:
            操作结果
        """
        try:
            self.vector_db_manager.delete_all_documents()
            return {
                "success": True,
                "message": "索引已清除"
            }
        except Exception as e:
            return {
                "success": False,
                "message": f"清除索引失败: {str(e)}"
            }

# web_interface/app.py
from fastapi import FastAPI, UploadFile, File, Form, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse, JSONResponse
from typing import List, Dict, Any
import os
import tempfile
import shutil
from rag_service.rag_engine import RAGEngine
from config import config

# 创建FastAPI应用
app = FastAPI(title="RAG MVP系统", description="检索增强生成最小可行产品")

# 初始化RAG引擎
rag_engine = RAGEngine()

# 创建临时目录用于文件上传
temp_dir = tempfile.mkdtemp()

@app.on_event("shutdown")
async def shutdown_event():
    """关闭时清理临时文件"""
    shutil.rmtree(temp_dir)

@app.get("/", response_class=HTMLResponse)
async def root():
    """根路径，返回简单的HTML界面"""
    html_content = """
    <!DOCTYPE html>
    <html>
    <head>
        <title>RAG MVP系统</title>
        <style>
            body {
                font-family: Arial, sans-serif;
                max-width: 1200px;
                margin: 0 auto;
                padding: 20px;
            }
            h1 {
                color: #333;
                text-align: center;
            }
            .section {
                margin-bottom: 30px;
                padding: 20px;
                border: 1px solid #ddd;
                border-radius: 8px;
            }
            .form-group {
                margin-bottom: 15px;
            }
            label {
                display: block;
                margin-bottom: 5px;
                font-weight: bold;
            }
            input[type="text"], textarea {
                width: 100%;
                padding: 10px;
                border: 1px solid #ddd;
                border-radius: 4px;
                font-size: 16px;
            }
            textarea {
                height: 150px;
                resize: vertical;
            }
            button {
                padding: 10px 20px;
                background-color: #4CAF50;
                color: white;
                border: none;
                border-radius: 4px;
                cursor: pointer;
                font-size: 16px;
            }
            button:hover {
                background-color: #45a049;
            }
            #results {
                margin-top: 20px;
                padding: 15px;
                background-color: #f9f9f9;
                border-radius: 4px;
            }
            .source {
                margin-top: 10px;
                padding: 10px;
                background-color: #fff;
                border: 1px solid #eee;
                border-radius: 4px;
            }
            .tab {
                overflow: hidden;
                border: 1px solid #ccc;
                background-color: #f1f1f1;
                border-radius: 4px 4px 0 0;
            }
            .tab button {
                background-color: inherit;
                float: left;
                border: none;
                outline: none;
                cursor: pointer;
                padding: 14px 16px;
                transition: 0.3s;
                color: #333;
                border-radius: 0;
            }
            .tab button:hover {
                background-color: #ddd;
            }
            .tab button.active {
                background-color: #ccc;
            }
            .tabcontent {
                display: none;
                padding: 20px;
                border: 1px solid #ddd;
                border-top: none;
                border-radius: 0 0 4px 4px;
            }
        </style>
    </head>
    <body>
        <h1>RAG MVP系统</h1>
        
        <div class="tab">
            <button class="tablinks" onclick="openTab(event, 'query')" id="defaultOpen">查询</button>
            <button class="tablinks" onclick="openTab(event, 'index')">索引文档</button>
            <button class="tablinks" onclick="openTab(event, 'stats')">系统统计</button>
        </div>
        
        <div id="query" class="tabcontent">
            <div class="section">
                <h2>自然语言查询</h2>
                <div class="form-group">
                    <label for="query-input">输入您的问题:</label>
                    <textarea id="query-input" placeholder="请输入您想查询的问题..."></textarea>
                </div>
                <button onclick="submitQuery()">提交查询</button>
                
                <div id="results" style="display: none;">
                    <h3>查询结果</h3>
                    <div id="answer"></div>
                    <h3>引用来源</h3>
                    <div id="sources"></div>
                </div>
            </div>
        </div>
        
        <div id="index" class="tabcontent">
            <div class="section">
                <h2>索引文档</h2>
                <div class="form-group">
                    <label for="documents">上传文档 (支持PDF, TXT, DOCX, XLSX):</label>
                    <input type="file" id="documents" multiple accept=".pdf,.txt,.docx,.xlsx">
                </div>
                <button onclick="uploadDocuments()">上传并索引</button>
                <div id="upload-status" style="margin-top: 20px;"></div>
            </div>
        </div>
        
        <div id="stats" class="tabcontent">
            <div class="section">
                <h2>系统统计信息</h2>
                <div id="system-stats"></div>
                <button onclick="getStats()">刷新统计</button>
                <button onclick="clearIndex()" style="background-color: #f44336; margin-left: 10px;">清除索引</button>
            </div>
        </div>
        
        <script>
            // 默认打开查询标签
            document.getElementById("defaultOpen").click();
            
            function openTab(evt, tabName) {
                var i, tabcontent, tablinks;
                tabcontent = document.getElementsByClassName("tabcontent");
                for (i = 0; i < tabcontent.length; i++) {
                    tabcontent[i].style.display = "none";
                }
                tablinks = document.getElementsByClassName("tablinks");
                for (i = 0; i < tablinks.length; i++) {
                    tablinks[i].className = tablinks[i].className.replace(" active", "");
                }
                document.getElementById(tabName).style.display = "block";
                evt.currentTarget.className += " active";
                
                // 如果打开统计标签，自动加载统计信息
                if (tabName === 'stats') {
                    getStats();
                }
            }
            
            function submitQuery() {
                const query = document.getElementById("query-input").value;
                if (!query.trim()) {
                    alert("请输入查询内容");
                    return;
                }
                
                document.getElementById("results").style.display = "none";
                
                fetch('/api/query', {
                    method: 'POST',
                    headers: {
                        'Content-Type': 'application/json',
                    },
                    body: JSON.stringify({ query: query }),
                })
                .then(response => response.json())
                .then(data => {
                    if (data.success) {
                        document.getElementById("answer").textContent = data.answer;
                        
                        // 显示来源
                        const sourcesDiv = document.getElementById("sources");
                        sourcesDiv.innerHTML = "";
                        
                        if (data.sources && data.sources.length > 0) {
                            data.sources.forEach((source, index) => {
                                const sourceDiv = document.createElement("div");
                                sourceDiv.className = "source";
                                sourceDiv.innerHTML = `<strong>来源 ${index + 1}:</strong> ${source.content.substring(0, 300)}${source.content.length > 300 ? '...' : ''}`;
                                sourcesDiv.appendChild(sourceDiv);
                            });
                        } else {
                            sourcesDiv.innerHTML = "<p>无引用来源</p>";
                        }
                        
                        document.getElementById("results").style.display = "block";
                    } else {
                        alert("查询失败: " + (data.error || "未知错误"));
                    }
                })
                .catch(error => {
                    console.error('Error:', error);
                    alert("查询失败，请重试");
                });
            }
            
            function uploadDocuments() {
                const files = document.getElementById("documents").files;
                if (files.length === 0) {
                    alert("请选择文件上传");
                    return;
                }
                
                const formData = new FormData();
                for (let i = 0; i < files.length; i++) {
                    formData.append("files", files[i]);
                }
                
                const statusDiv = document.getElementById("upload-status");
                statusDiv.innerHTML = "正在上传并索引，请稍候...";
                
                fetch('/api/index', {
                    method: 'POST',
                    body: formData,
                })
                .then(response => response.json())
                .then(data => {
                    if (data.success) {
                        statusDiv.innerHTML = `<p style="color: green;">成功: ${data.message}</p>`;
                        if (data.stats) {
                            statusDiv.innerHTML += `<p>文档数: ${data.stats.total_documents}</p>`;
                            statusDiv.innerHTML += `<p>文档块数: ${data.stats.total_chunks}</p>`;
                            statusDiv.innerHTML += `<p>已索引块数: ${data.stats.indexed_chunks}</p>`;
                        }
                    } else {
                        statusDiv.innerHTML = `<p style="color: red;">失败: ${data.message}</p>`;
                    }
                })
                .catch(error => {
                    console.error('Error:', error);
                    statusDiv.innerHTML = "<p style='color: red;'>上传失败，请重试</p>";
                });
            }
            
            function getStats() {
                fetch('/api/stats')
                .then(response => response.json())
                .then(data => {
                    const statsDiv = document.getElementById("system-stats");
                    
                    if (data.config) {
                        let html = "<h3>配置信息</h3><ul>";
                        html += `<li>嵌入模型: ${data.config.embedding_model}</li>`;
                        html += `<li>LLM模型: ${data.config.llm_model}</li>`;
                        html += `<li>分块大小: ${data.config.chunk_size}</li>`;
                        html += `<li>检索数量: ${data.config.top_k}</li>`;
                        html += `</ul>`;
                        
                        if (data.vector_store) {
                            html += `<h3>向量存储</h3><p>${data.vector_store}</p>`;
                        }
                        
                        statsDiv.innerHTML = html;
                    } else {
                        statsDiv.innerHTML = "<p>无法获取统计信息</p>";
                    }
                })
                .catch(error => {
                    console.error('Error:', error);
                    document.getElementById("system-stats").innerHTML = "<p>获取统计失败，请重试</p>";
                });
            }
            
            function clearIndex() {
                if (confirm("确定要清除所有索引的文档吗？此操作不可恢复。")) {
                    fetch('/api/clear', {
                        method: 'POST'
                    })
                    .then(response => response.json())
                    .then(data => {
                        if (data.success) {
                            alert("索引已清除");
                            getStats();
                        } else {
                            alert("清除索引失败: " + data.message);
                        }
                    })
                    .catch(error => {
                        console.error('Error:', error);
                        alert("清除索引失败，请重试");
                    });
                }
            }
        </script>
    </body>
    </html>
    """
    return html_content

@app.post("/api/query")
async def api_query(query_data: Dict[str, str]) -> Dict[str, Any]:
    """API端点：执行查询"""
    query = query_data.get("query")
    if not query:
        raise HTTPException(status_code=400, detail="查询不能为空")
    
    return rag_engine.query(query)

@app.post("/api/index")
async def api_index(files: List[UploadFile] = File(...)) -> Dict[str, Any]:
    """API端点：索引上传的文件"""
    # 创建临时目录保存上传的文件
    upload_dir = os.path.join(temp_dir, str(os.getpid()))
    os.makedirs(upload_dir, exist_ok=True)
    
    try:
        # 保存上传的文件
        for file in files:
            file_path = os.path.join(upload_dir, file.filename)
            with open(file_path, "wb") as buffer:
                shutil.copyfileobj(file.file, buffer)
        
        # 索引文件
        result = rag_engine.index_documents(upload_dir)
        return result
    finally:
        # 清理临时文件
        shutil.rmtree(upload_dir, ignore_errors=True)

@app.get("/api/stats")
async def api_stats() -> Dict[str, Any]:
    """API端点：获取系统统计信息"""
    return rag_engine.get_stats()

@app.post("/api/clear")
async def api_clear() -> Dict[str, Any]:
    """API端点：清除索引"""
    return rag_engine.clear_index()

# 启动应用的代码（在实际部署时使用）
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "web_interface.app:app",
        host=config.HOST,
        port=config.PORT,
        reload=True
    )

pip install -r requirements.txt

# .env
OPENAI_API_KEY=your_openai_api_key
PINECONE_API_KEY=your_pinecone_api_key
PINECONE_ENVIRONMENT=your_pinecone_environment

# 方式1：直接运行Python脚本
python -m web_interface.app

# 方式2：使用uvicorn
uvicorn web_interface.app:app --host 0.0.0.0 --port 8000 --reload

顶层: 文档级检索 (快速筛选相关文档集合)
↓
中层: 章节级检索 (缩小范围到相关章节)
↓
底层: 段落级检索 (精确获取相关段落)

第一阶段: 向量检索 (召回top-100候选)
↓
第二阶段: 轻量级重排序 (过滤到top-10候选)
↓
第三阶段: LLM重排序 (得到最终top-k结果)

文档内容 → 实体识别 → 关系抽取 → 知识图谱构建 → 图检索增强

查询复杂度分析 → 选择检索策略 → 动态调整参数 → 执行检索

### 6.3 企业级应用案例

#### 金融行业案例

**案例背景**：某大型银行实施RAG系统，用于智能客服和风险评估。

**系统架构**：
- 向量数据库：Milvus分布式部署，分片存储
- 嵌入模型：定制化金融领域BERT模型
- LLM：GPT-4金融领域微调版本
- 部署：Kubernetes集群，多区域冗余

**关键挑战与解决方案**：

| 挑战 | 解决方案 | 效果 |
|------|----------|------|
| 敏感数据安全 | 私有化部署+端到端加密 | 符合金融监管要求 |
| 实时性要求 | 多级缓存+异步处理 | 响应时间<0.8秒 |
| 数据量大 | 分片存储+增量更新 | 支持100TB+数据 |
| 精确性要求 | 知识图谱增强+专家反馈 | 准确率提升30% |

**实施成果**：
- 客服效率提升60%
- 风险评估准确率达到95%
- 运营成本降低40%

#### 医疗行业案例

**案例背景**：某三甲医院部署RAG系统辅助临床决策支持。

**技术架构**：
- 文档处理：医学文献自动提取与分块
- 向量数据库：Pinecone + 医疗知识库
- 安全层：HIPAA合规加密与访问控制
- 监控：全流程审计日志

**应用场景**：

1. **临床指南查询**：医生可快速检索最新治疗指南
2. **病例相似性分析**：基于患者病历检索相似病例
3. **药物相互作用**：实时查询药物兼容性信息
4. **医学文献辅助**：自动总结最新研究文献

**实施效果**：
- 医生查询效率提升75%
- 诊断支持准确率达到92%
- 患者满意度提升35%

#### 制造行业案例

**案例背景**：全球制造企业部署RAG系统用于设备维护与故障诊断。

**系统特点**：
- 多模态支持：文本手册+图表+技术图纸
- 实时索引：设备数据自动更新到向量库
- 边缘部署：部分功能在工厂本地运行
- 多语言支持：支持8种语言界面

**核心功能**：

```python
# 设备故障诊断RAG示例
class EquipmentDiagnosticRAG:
    """设备故障诊断RAG系统"""
    
    def __init__(self, vector_db, equipment_api, llm):
        """初始化系统"""
        self.vector_db = vector_db
        self.equipment_api = equipment_api
        self.llm = llm
    
    def diagnose(self, equipment_id, fault_symptoms):
        """诊断设备故障
        
        Args:
            equipment_id: 设备ID
            fault_symptoms: 故障症状描述
            
        Returns:
            诊断结果与维修建议
        """
        # 1. 获取设备历史数据
        equipment_data = self.equipment_api.get_equipment_data(equipment_id)
        
        # 2. 构建查询
        query = f"{fault_symptoms} 设备型号: {equipment_data['model']} 运行时间: {equipment_data['runtime']}"
        
        # 3. 检索相关文档
        context = self.vector_db.similarity_search(query, k=5)
        
        # 4. 检索相似故障案例
        similar_cases = self._find_similar_cases(fault_symptoms, equipment_data['model'])
        
        # 5. 构建提示
        prompt = self._build_diagnosis_prompt(query, context, similar_cases, equipment_data)
        
        # 6. 生成诊断结果
        diagnosis = self.llm.generate(prompt)
        
        return diagnosis
    
    def _find_similar_cases(self, symptoms, model):
        """查找相似故障案例"""
        # 实现略
        return []
    
    def _build_diagnosis_prompt(self, query, context, cases, equipment_data):
        """构建诊断提示"""
        # 构建详细提示
        # 实现略
        return ""