当人工智能遇上知识检索：RAG技术的深度解析与实践探索

RAG = 检索技术 + LLM生成 + 上下文融合

# 数据准备流程示例
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma

# 1. 数据加载
loader = PyPDFLoader("enterprise_knowledge.pdf")
documents = loader.load()

# 2. 文本分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", "。", "！", "？"]
)
texts = text_splitter.split_documents(documents)

# 3. 向量化处理
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=texts,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 实时查询处理流程
from langchain.llms import OpenAI
from langchain.chains import RetrievalQA

# 1. 用户查询向量化
query = "什么是深度学习的反向传播算法？"
query_embedding = embeddings.embed_query(query)

# 2. 相似性检索
relevant_docs = vectorstore.similarity_search(
    query, k=5, score_threshold=0.7
)

# 3. 构建增强提示
context = "\n".join([doc.page_content for doc in relevant_docs])
prompt = f"""
基于以下上下文信息回答问题：

上下文：{context}

问题：{query}

请基于上下文提供准确、详细的回答：
"""

# 4. LLM生成回答
llm = OpenAI(temperature=0.1)
response = llm(prompt)

# 向量相似性计算示例
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

def calculate_similarity(query_vector, doc_vectors):
    """
    计算查询向量与文档向量的余弦相似度
    """
    similarities = cosine_similarity(
        query_vector.reshape(1, -1), 
        doc_vectors
    )
    return similarities[0]

# 示例：计算相似度并排序
query_vec = np.array([0.1, 0.2, 0.3, 0.4])  # 查询向量
doc_vecs = np.array([
    [0.15, 0.25, 0.35, 0.45],  # 文档1向量
    [0.8, 0.1, 0.05, 0.05],   # 文档2向量
    [0.12, 0.18, 0.32, 0.38]  # 文档3向量
])

similarities = calculate_similarity(query_vec, doc_vecs)
ranked_docs = np.argsort(similarities)[::-1]  # 按相似度降序排列

# 多种Embedding模型对比
from sentence_transformers import SentenceTransformer
from openai import OpenAI

class EmbeddingComparison:
    def __init__(self):
        # 开源模型
        self.sentence_model = SentenceTransformer(
            'paraphrase-multilingual-MiniLM-L12-v2'
        )
        # 商业模型
        self.openai_client = OpenAI()
    
    def get_sentence_embedding(self, text):
        """使用SentenceTransformer获取向量"""
        return self.sentence_model.encode(text)
    
    def get_openai_embedding(self, text):
        """使用OpenAI获取向量"""
        response = self.openai_client.embeddings.create(
            model="text-embedding-ada-002",
            input=text
        )
        return response.data[0].embedding
    
    def compare_embeddings(self, text1, text2):
        """比较不同模型的向量相似度"""
        # SentenceTransformer结果
        sent_emb1 = self.get_sentence_embedding(text1)
        sent_emb2 = self.get_sentence_embedding(text2)
        sent_sim = cosine_similarity([sent_emb1], [sent_emb2])[0][0]
        
        # OpenAI结果
        openai_emb1 = self.get_openai_embedding(text1)
        openai_emb2 = self.get_openai_embedding(text2)
        openai_sim = cosine_similarity([openai_emb1], [openai_emb2])[0][0]
        
        return {
            'sentence_transformer': sent_sim,
            'openai': openai_sim
        }

# 企业知识管理系统示例
class EnterpriseKnowledgeRAG:
    def __init__(self):
        self.document_types = {
            'policy': '政策文件',
            'manual': '操作手册', 
            'report': '分析报告',
            'faq': '常见问题'
        }
        self.vectorstore = None
        self.llm = None
    
    def ingest_documents(self, file_paths, doc_type):
        """批量导入企业文档"""
        documents = []
        for file_path in file_paths:
            loader = self._get_loader(file_path)
            docs = loader.load()
            
            # 添加元数据
            for doc in docs:
                doc.metadata.update({
                    'doc_type': doc_type,
                    'source': file_path,
                    'ingestion_time': datetime.now().isoformat()
                })
            documents.extend(docs)
        
        # 文本分割和向量化
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=800,
            chunk_overlap=100
        )
        splits = text_splitter.split_documents(documents)
        
        # 存储到向量数据库
        self.vectorstore = Chroma.from_documents(
            documents=splits,
            embedding=OpenAIEmbeddings(),
            persist_directory="./enterprise_kb"
        )
    
    def query_knowledge(self, question, doc_type_filter=None):
        """智能知识查询"""
        # 构建过滤条件
        filter_dict = {}
        if doc_type_filter:
            filter_dict['doc_type'] = doc_type_filter
        
        # 检索相关文档
        relevant_docs = self.vectorstore.similarity_search(
            question, 
            k=5,
            filter=filter_dict
        )
        
        # 构建上下文
        context = self._build_context(relevant_docs)
        
        # 生成回答
        prompt = f"""
        你是一个企业知识助手，请基于以下企业内部资料回答问题。
        
        企业资料：
        {context}
        
        问题：{question}
        
        请提供准确、专业的回答，并注明信息来源：
        """
        
        response = self.llm(prompt)
        return {
            'answer': response,
            'sources': [doc.metadata['source'] for doc in relevant_docs],
            'doc_types': list(set([doc.metadata['doc_type'] for doc in relevant_docs]))
        }

# 智能客服RAG系统
class CustomerServiceRAG:
    def __init__(self):
        self.knowledge_base = {
            'product_info': '产品信息库',
            'troubleshooting': '故障排除指南',
            'policy': '服务政策',
            'history': '历史对话记录'
        }
    
    def handle_customer_query(self, query, customer_id=None):
        """处理客户查询"""
        # 1. 查询意图识别
        intent = self._classify_intent(query)
        
        # 2. 检索相关知识
        relevant_docs = self._retrieve_knowledge(query, intent)
        
        # 3. 个性化上下文（如果有客户历史）
        if customer_id:
            customer_context = self._get_customer_context(customer_id)
            relevant_docs.extend(customer_context)
        
        # 4. 生成回答
        response = self._generate_response(query, relevant_docs, intent)
        
        # 5. 记录对话历史
        self._log_interaction(customer_id, query, response)
        
        return response
    
    def _classify_intent(self, query):
        """查询意图分类"""
        intent_prompts = {
            'product_inquiry': '产品咨询',
            'technical_support': '技术支持',
            'billing_question': '账单问题',
            'complaint': '投诉建议'
        }
        
        classification_prompt = f"""
        请将以下客户查询分类到合适的意图类别：
        
        查询：{query}
        
        可选类别：{list(intent_prompts.keys())}
        
        只返回类别名称：
        """
        
        intent = self.llm(classification_prompt).strip()
        return intent if intent in intent_prompts else 'general'

# 教育RAG系统
class EducationalRAG:
    def __init__(self):
        self.subject_areas = {
            'mathematics': '数学',
            'physics': '物理',
            'chemistry': '化学',
            'computer_science': '计算机科学'
        }
    
    def create_personalized_explanation(self, concept, student_level, learning_style):
        """创建个性化概念解释"""
        # 检索相关教学材料
        relevant_materials = self.vectorstore.similarity_search(
            f"{concept} {student_level} level explanation",
            k=3
        )
        
        # 根据学习风格调整解释方式
        style_prompts = {
            'visual': '请用图表、图像和视觉化方式解释',
            'auditory': '请用对话和声音相关的比喻解释',
            'kinesthetic': '请用动手实践和身体感知的方式解释',
            'reading': '请用详细的文字描述和阅读材料解释'
        }
        
        context = "\n".join([doc.page_content for doc in relevant_materials])
        
        prompt = f"""
        你是一位经验丰富的教师，请为{student_level}水平的学生解释以下概念：
        
        概念：{concept}
        学生水平：{student_level}
        学习风格：{learning_style}
        
        参考教学材料：
        {context}
        
        解释要求：{style_prompts.get(learning_style, '请用清晰易懂的方式解释')}
        
        请提供：
        1. 概念的核心定义
        2. 具体例子和应用
        3. 常见误区和注意事项
        4. 进一步学习建议
        """
        
        explanation = self.llm(prompt)
        return explanation

# 实时知识更新系统
class RealTimeKnowledgeUpdater:
    def __init__(self, vectorstore):
        self.vectorstore = vectorstore
        self.update_queue = []
    
    def add_new_document(self, document, metadata=None):
        """添加新文档到知识库"""
        # 文档预处理
        processed_doc = self._preprocess_document(document)
        
        # 添加时间戳和版本信息
        if metadata is None:
            metadata = {}
        metadata.update({
            'added_time': datetime.now().isoformat(),
            'version': self._get_next_version(),
            'status': 'active'
        })
        
        # 向量化并存储
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        chunks = text_splitter.split_text(processed_doc)
        
        for chunk in chunks:
            doc_obj = Document(
                page_content=chunk,
                metadata=metadata
            )
            self.vectorstore.add_documents([doc_obj])
        
        print(f"成功添加文档，共{len(chunks)}个片段")
    
    def update_document(self, doc_id, new_content):
        """更新现有文档"""
        # 标记旧版本为非活跃
        self._deactivate_document(doc_id)
        
        # 添加新版本
        metadata = {
            'original_doc_id': doc_id,
            'update_type': 'content_revision'
        }
        self.add_new_document(new_content, metadata)
    
    def remove_outdated_info(self, keywords, cutoff_date):
        """移除过时信息"""
        # 搜索包含特定关键词的文档
        docs = self.vectorstore.similarity_search(
            " ".join(keywords),
            k=100
        )
        
        # 筛选出过时的文档
        outdated_docs = []
        for doc in docs:
            doc_date = datetime.fromisoformat(doc.metadata.get('added_time', '1970-01-01'))
            if doc_date < cutoff_date:
                outdated_docs.append(doc)
        
        # 标记为非活跃
        for doc in outdated_docs:
            doc.metadata['status'] = 'inactive'
            doc.metadata['deactivated_time'] = datetime.now().isoformat()
        
        print(f"标记{len(outdated_docs)}个过时文档为非活跃状态")

# 可解释性RAG系统
class ExplainableRAG:
    def __init__(self, vectorstore, llm):
        self.vectorstore = vectorstore
        self.llm = llm
    
    def query_with_explanation(self, question):
        """带解释的查询"""
        # 1. 检索相关文档
        relevant_docs = self.vectorstore.similarity_search_with_score(
            question, k=5
        )
        
        # 2. 分析检索结果
        retrieval_analysis = self._analyze_retrieval_results(relevant_docs)
        
        # 3. 生成回答
        context = "\n".join([doc.page_content for doc, score in relevant_docs])
        
        prompt = f"""
        基于以下信息回答问题，并说明你的推理过程：
        
        上下文信息：
        {context}
        
        问题：{question}
        
        请提供：
        1. 直接回答
        2. 推理过程
        3. 信心度评估（1-10分）
        """
        
        response = self.llm(prompt)
        
        # 4. 构建完整的解释
        explanation = {
            'answer': response,
            'sources': self._format_sources(relevant_docs),
            'retrieval_analysis': retrieval_analysis,
            'confidence_factors': self._calculate_confidence_factors(relevant_docs)
        }
        
        return explanation
    
    def _analyze_retrieval_results(self, docs_with_scores):
        """分析检索结果质量"""
        analysis = {
            'total_docs': len(docs_with_scores),
            'avg_similarity': np.mean([score for _, score in docs_with_scores]),
            'source_diversity': len(set([doc.metadata.get('source', 'unknown') 
                                       for doc, _ in docs_with_scores])),
            'recency': self._analyze_document_recency(docs_with_scores)
        }
        return analysis
    
    def _format_sources(self, docs_with_scores):
        """格式化信息源"""
        sources = []
        for doc, score in docs_with_scores:
            source_info = {
                'content_preview': doc.page_content[:200] + "...",
                'similarity_score': round(score, 3),
                'source': doc.metadata.get('source', '未知来源'),
                'timestamp': doc.metadata.get('added_time', '未知时间')
            }
            sources.append(source_info)
        return sources

# 成本效益分析工具
class RAGCostAnalyzer:
    def __init__(self):
        self.cost_factors = {
            'embedding_cost': 0.0001,  # 每1K tokens的embedding成本
            'storage_cost': 0.001,     # 每GB向量存储成本/月
            'query_cost': 0.002,       # 每次查询成本
            'llm_generation_cost': 0.02 # 每1K tokens生成成本
        }
    
    def calculate_setup_cost(self, document_count, avg_doc_size_kb):
        """计算初始设置成本"""
        total_tokens = document_count * avg_doc_size_kb * 0.75  # 估算token数
        
        costs = {
            'embedding_processing': total_tokens * self.cost_factors['embedding_cost'] / 1000,
            'initial_storage': (total_tokens * 1536 * 4) / (1024**3) * self.cost_factors['storage_cost'],  # 假设1536维向量
            'setup_time': document_count * 0.1  # 假设每文档0.1小时人工成本
        }
        
        return costs
    
    def calculate_operational_cost(self, monthly_queries, avg_response_tokens):
        """计算运营成本"""
        monthly_costs = {
            'query_processing': monthly_queries * self.cost_factors['query_cost'],
            'llm_generation': monthly_queries * avg_response_tokens * self.cost_factors['llm_generation_cost'] / 1000,
            'storage_maintenance': 10  # 固定存储维护成本
        }
        
        return monthly_costs
    
    def compare_with_fine_tuning(self, model_size_gb, training_hours):
        """与微调成本对比"""
        fine_tuning_costs = {
            'compute_cost': training_hours * 50,  # 假设每小时50美元GPU成本
            'data_preparation': training_hours * 0.5 * 100,  # 数据准备人工成本
            'model_storage': model_size_gb * 0.1,  # 模型存储成本
            'deployment_cost': 200  # 部署成本
        }
        
        return fine_tuning_costs

# 检索质量优化系统
class RetrievalQualityOptimizer:
    def __init__(self, vectorstore):
        self.vectorstore = vectorstore
        self.retrieval_metrics = {
            'precision': [],
            'recall': [],
            'f1_score': [],
            'mrr': []  # Mean Reciprocal Rank
        }
    
    def hybrid_search(self, query, alpha=0.7):
        """混合检索：语义搜索 + 关键词搜索"""
        # 语义搜索
        semantic_results = self.vectorstore.similarity_search_with_score(
            query, k=10
        )
        
        # 关键词搜索（BM25）
        keyword_results = self._bm25_search(query, k=10)
        
        # 结果融合
        fused_results = self._fuse_results(
            semantic_results, 
            keyword_results, 
            alpha
        )
        
        return fused_results[:5]  # 返回top-5结果
    
    def _bm25_search(self, query, k=10):
        """BM25关键词搜索实现"""
        from rank_bm25 import BM25Okapi
        
        # 获取所有文档
        all_docs = self._get_all_documents()
        
        # 分词处理
        tokenized_docs = [doc.split() for doc in all_docs]
        bm25 = BM25Okapi(tokenized_docs)
        
        # 搜索
        query_tokens = query.split()
        scores = bm25.get_scores(query_tokens)
        
        # 排序并返回top-k
        top_indices = np.argsort(scores)[::-1][:k]
        
        results = []
        for idx in top_indices:
            results.append((all_docs[idx], scores[idx]))
        
        return results
    
    def _fuse_results(self, semantic_results, keyword_results, alpha):
        """结果融合算法"""
        # 归一化分数
        semantic_scores = self._normalize_scores([score for _, score in semantic_results])
        keyword_scores = self._normalize_scores([score for _, score in keyword_results])
        
        # 创建文档到分数的映射
        doc_scores = {}
        
        # 语义搜索结果
        for i, (doc, _) in enumerate(semantic_results):
            doc_id = self._get_doc_id(doc)
            doc_scores[doc_id] = alpha * semantic_scores[i]
        
        # 关键词搜索结果
        for i, (doc, _) in enumerate(keyword_results):
            doc_id = self._get_doc_id(doc)
            if doc_id in doc_scores:
                doc_scores[doc_id] += (1 - alpha) * keyword_scores[i]
            else:
                doc_scores[doc_id] = (1 - alpha) * keyword_scores[i]
        
        # 按融合分数排序
        sorted_docs = sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)
        
        return [(self._get_doc_by_id(doc_id), score) for doc_id, score in sorted_docs]
    
    def evaluate_retrieval_quality(self, test_queries, ground_truth):
        """评估检索质量"""
        total_precision = 0
        total_recall = 0
        total_f1 = 0
        reciprocal_ranks = []
        
        for query, relevant_docs in zip(test_queries, ground_truth):
            # 执行检索
            retrieved_docs = self.hybrid_search(query)
            retrieved_ids = [self._get_doc_id(doc) for doc, _ in retrieved_docs]
            
            # 计算指标
            relevant_ids = set(relevant_docs)
            retrieved_set = set(retrieved_ids)
            
            # Precision和Recall
            if len(retrieved_set) > 0:
                precision = len(relevant_ids & retrieved_set) / len(retrieved_set)
            else:
                precision = 0
            
            if len(relevant_ids) > 0:
                recall = len(relevant_ids & retrieved_set) / len(relevant_ids)
            else:
                recall = 0
            
            # F1 Score
            if precision + recall > 0:
                f1 = 2 * (precision * recall) / (precision + recall)
            else:
                f1 = 0
            
            # MRR (Mean Reciprocal Rank)
            for i, doc_id in enumerate(retrieved_ids):
                if doc_id in relevant_ids:
                    reciprocal_ranks.append(1 / (i + 1))
                    break
            else:
                reciprocal_ranks.append(0)
            
            total_precision += precision
            total_recall += recall
            total_f1 += f1
        
        # 计算平均指标
        num_queries = len(test_queries)
        avg_metrics = {
            'precision': total_precision / num_queries,
            'recall': total_recall / num_queries,
            'f1_score': total_f1 / num_queries,
            'mrr': np.mean(reciprocal_ranks)
        }
        
        return avg_metrics

# 上下文管理优化器
class ContextManager:
    def __init__(self, max_context_length=4000):
        self.max_context_length = max_context_length
        self.tokenizer = tiktoken.get_encoding("cl100k_base")  # GPT-4 tokenizer
    
    def optimize_context(self, query, retrieved_docs):
        """优化上下文以适应长度限制"""
        # 1. 计算查询和模板的token数
        base_prompt_tokens = self._count_tokens(
            f"基于以下信息回答问题：\n\n问题：{query}\n\n请提供详细回答："
        )
        
        available_tokens = self.max_context_length - base_prompt_tokens - 500  # 预留生成空间
        
        # 2. 文档重要性评分
        scored_docs = self._score_document_importance(query, retrieved_docs)
        
        # 3. 贪心选择文档
        selected_docs = []
        current_tokens = 0
        
        for doc, score in scored_docs:
            doc_tokens = self._count_tokens(doc.page_content)
            
            if current_tokens + doc_tokens <= available_tokens:
                selected_docs.append(doc)
                current_tokens += doc_tokens
            else:
                # 尝试截断文档
                remaining_tokens = available_tokens - current_tokens
                if remaining_tokens > 100:  # 至少保留100个token
                    truncated_content = self._truncate_document(
                        doc.page_content, remaining_tokens
                    )
                    truncated_doc = Document(
                        page_content=truncated_content,
                        metadata=doc.metadata
                    )
                    selected_docs.append(truncated_doc)
                break
        
        return selected_docs
    
    def _score_document_importance(self, query, docs):
        """文档重要性评分"""
        scored_docs = []
        
        for doc in docs:
            # 多维度评分
            similarity_score = self._calculate_similarity(query, doc.page_content)
            recency_score = self._calculate_recency_score(doc.metadata)
            authority_score = self._calculate_authority_score(doc.metadata)
            
            # 综合评分
            total_score = (
                0.5 * similarity_score + 
                0.3 * recency_score + 
                0.2 * authority_score
            )
            
            scored_docs.append((doc, total_score))
        
        # 按分数降序排列
        return sorted(scored_docs, key=lambda x: x[1], reverse=True)
    
    def _truncate_document(self, content, max_tokens):
        """智能文档截断"""
        tokens = self.tokenizer.encode(content)
        
        if len(tokens) <= max_tokens:
            return content
        
        # 尝试在句子边界截断
        sentences = content.split('。')
        truncated_content = ""
        current_tokens = 0
        
        for sentence in sentences:
            sentence_tokens = len(self.tokenizer.encode(sentence + '。'))
            if current_tokens + sentence_tokens <= max_tokens:
                truncated_content += sentence + '。'
                current_tokens += sentence_tokens
            else:
                break
        
        # 如果没有完整句子，则硬截断
        if not truncated_content:
            truncated_tokens = tokens[:max_tokens-10]  # 预留省略号空间
            truncated_content = self.tokenizer.decode(truncated_tokens) + "..."
        
        return truncated_content

# 多模态RAG系统
class MultiModalRAG:
    def __init__(self):
        self.text_embedder = OpenAIEmbeddings()
        self.image_embedder = self._init_image_embedder()
        self.audio_embedder = self._init_audio_embedder()
        self.multimodal_vectorstore = None
    
    def _init_image_embedder(self):
        """初始化图像编码器"""
        from transformers import CLIPProcessor, CLIPModel
        
        model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
        processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
        
        return {'model': model, 'processor': processor}
    
    def _init_audio_embedder(self):
        """初始化音频编码器"""
        from transformers import Wav2Vec2Processor, Wav2Vec2Model
        
        processor = Wav2Vec2Processor.from_pretrained("facebook/wav2vec2-base-960h")
        model = Wav2Vec2Model.from_pretrained("facebook/wav2vec2-base-960h")
        
        return {'model': model, 'processor': processor}
    
    def process_multimodal_document(self, file_path, doc_type):
        """处理多模态文档"""
        if doc_type == 'image':
            return self._process_image_document(file_path)
        elif doc_type == 'audio':
            return self._process_audio_document(file_path)
        elif doc_type == 'video':
            return self._process_video_document(file_path)
        else:
            return self._process_text_document(file_path)
    
    def _process_image_document(self, image_path):
        """处理图像文档"""
        from PIL import Image
        import torch
        
        # 加载图像
        image = Image.open(image_path)
        
        # 图像描述生成
        description = self._generate_image_description(image)
        
        # 图像特征提取
        inputs = self.image_embedder['processor'](images=image, return_tensors="pt")
        with torch.no_grad():
            image_features = self.image_embedder['model'].get_image_features(**inputs)
        
        # 创建多模态文档
        multimodal_doc = {
            'content_type': 'image',
            'file_path': image_path,
            'description': description,
            'text_embedding': self.text_embedder.embed_query(description),
            'image_embedding': image_features.numpy().flatten(),
            'metadata': {
                'source': image_path,
                'type': 'image',
                'processed_time': datetime.now().isoformat()
            }
        }
        
        return multimodal_doc
    
    def _process_audio_document(self, audio_path):
        """处理音频文档"""
        import librosa
        import torch
        
        # 加载音频
        audio, sr = librosa.load(audio_path, sr=16000)
        
        # 语音转文本
        transcript = self._speech_to_text(audio, sr)
        
        # 音频特征提取
        inputs = self.audio_embedder['processor'](
            audio, sampling_rate=sr, return_tensors="pt"
        )
        with torch.no_grad():
            audio_features = self.audio_embedder['model'](**inputs).last_hidden_state
            audio_embedding = torch.mean(audio_features, dim=1).numpy().flatten()
        
        # 创建多模态文档
        multimodal_doc = {
            'content_type': 'audio',
            'file_path': audio_path,
            'transcript': transcript,
            'text_embedding': self.text_embedder.embed_query(transcript),
            'audio_embedding': audio_embedding,
            'metadata': {
                'source': audio_path,
                'type': 'audio',
                'duration': len(audio) / sr,
                'processed_time': datetime.now().isoformat()
            }
        }
        
        return multimodal_doc
    
    def multimodal_search(self, query, query_type='text', k=5):
        """多模态搜索"""
        if query_type == 'text':
            query_embedding = self.text_embedder.embed_query(query)
            # 在文本嵌入空间中搜索
            results = self._search_in_embedding_space(
                query_embedding, 'text_embedding', k
            )
        elif query_type == 'image':
            # 处理图像查询
            query_embedding = self._encode_image_query(query)
            results = self._search_in_embedding_space(
                query_embedding, 'image_embedding', k
            )
        elif query_type == 'audio':
            # 处理音频查询
            query_embedding = self._encode_audio_query(query)
            results = self._search_in_embedding_space(
                query_embedding, 'audio_embedding', k
            )
        
        return results
    
    def _generate_image_description(self, image):
        """生成图像描述"""
        # 使用BLIP或类似模型生成图像描述
        # 这里简化为示例
        return "这是一张包含多个对象的图像"
    
    def _speech_to_text(self, audio, sr):
        """语音转文本"""
        # 使用Whisper或类似模型
        # 这里简化为示例
        return "这是音频的转录文本"

# 自适应RAG系统
class AdaptiveRAG:
    def __init__(self):
        self.query_classifier = self._init_query_classifier()
        self.retrieval_strategies = {
            'factual': self._factual_retrieval,
            'analytical': self._analytical_retrieval,
            'creative': self._creative_retrieval,
            'comparative': self._comparative_retrieval
        }
        self.performance_tracker = {}
    
    def adaptive_query(self, query, user_context=None):
        """自适应查询处理"""
        # 1. 查询分类
        query_type = self._classify_query(query)
        
        # 2. 用户画像分析
        user_profile = self._analyze_user_profile(user_context)
        
        # 3. 选择最优检索策略
        strategy = self._select_optimal_strategy(query_type, user_profile)
        
        # 4. 执行检索
        retrieved_docs = strategy(query, user_profile)
        
        # 5. 动态调整上下文
        optimized_context = self._optimize_context_dynamically(
            query, retrieved_docs, query_type
        )
        
        # 6. 生成回答
        response = self._generate_adaptive_response(
            query, optimized_context, query_type, user_profile
        )
        
        # 7. 记录性能数据
        self._track_performance(query_type, strategy.__name__, response)
        
        return response
    
    def _classify_query(self, query):
        """查询类型分类"""
        classification_prompt = f"""
        请将以下查询分类到最合适的类型：
        
        查询：{query}
        
        类型选项：
        - factual: 事实性查询，需要准确的信息
        - analytical: 分析性查询，需要深度思考
        - creative: 创造性查询，需要灵感和想象
        - comparative: 比较性查询，需要对比分析
        
        只返回类型名称：
        """
        
        query_type = self.query_classifier(classification_prompt).strip().lower()
        return query_type if query_type in self.retrieval_strategies else 'factual'
    
    def _factual_retrieval(self, query, user_profile):
        """事实性检索策略"""
        # 高精度检索，优先权威来源
        return self.vectorstore.similarity_search(
            query,
            k=3,
            filter={'authority_score': {'$gte': 0.8}}
        )
    
    def _analytical_retrieval(self, query, user_profile):
        """分析性检索策略"""
        # 多角度检索，包含不同观点
        diverse_results = []
        
        # 主要观点
        main_results = self.vectorstore.similarity_search(query, k=3)
        diverse_results.extend(main_results)
        
        # 反对观点
        counter_query = f"反对 {query} 的观点"
        counter_results = self.vectorstore.similarity_search(counter_query, k=2)
        diverse_results.extend(counter_results)
        
        return diverse_results
    
    def _creative_retrieval(self, query, user_profile):
        """创造性检索策略"""
        # 扩散性检索，包含相关但不直接的信息
        expanded_queries = self._generate_creative_queries(query)
        
        all_results = []
        for expanded_query in expanded_queries:
            results = self.vectorstore.similarity_search(expanded_query, k=2)
            all_results.extend(results)
        
        return all_results[:5]
    
    def _select_optimal_strategy(self, query_type, user_profile):
        """选择最优检索策略"""
        # 基于历史性能选择策略
        if query_type in self.performance_tracker:
            best_strategy = max(
                self.performance_tracker[query_type].items(),
                key=lambda x: x[1]['avg_satisfaction']
            )[0]
            return self.retrieval_strategies[best_strategy]
        
        return self.retrieval_strategies[query_type]

# RAG Agent系统
class RAGAgent:
    def __init__(self):
        self.tools = {
            'search': self._search_tool,
            'calculate': self._calculate_tool,
            'visualize': self._visualize_tool,
            'summarize': self._summarize_tool
        }
        self.memory = []
        self.vectorstore = None
        self.llm = None
    
    def execute_complex_task(self, task_description):
        """执行复杂任务"""
        # 1. 任务分解
        subtasks = self._decompose_task(task_description)
        
        # 2. 执行计划生成
        execution_plan = self._generate_execution_plan(subtasks)
        
        # 3. 逐步执行
        results = []
        for step in execution_plan:
            step_result = self._execute_step(step)
            results.append(step_result)
            
            # 更新记忆
            self.memory.append({
                'step': step,
                'result': step_result,
                'timestamp': datetime.now().isoformat()
            })
        
        # 4. 结果整合
        final_result = self._integrate_results(results, task_description)
        
        return final_result
    
    def _decompose_task(self, task):
        """任务分解"""
        decomposition_prompt = f"""
        请将以下复杂任务分解为具体的子任务：
        
        任务：{task}
        
        请按照逻辑顺序列出子任务，每个子任务应该是具体可执行的：
        """
        
        response = self.llm(decomposition_prompt)
        subtasks = [line.strip() for line in response.split('\n') if line.strip()]
        
        return subtasks
    
    def _execute_step(self, step):
        """执行单个步骤"""
        # 1. 确定需要的工具
        required_tools = self._identify_required_tools(step)
        
        # 2. 检索相关信息
        relevant_info = self._retrieve_step_info(step)
        
        # 3. 执行工具调用
        tool_results = {}
        for tool_name in required_tools:
            if tool_name in self.tools:
                tool_result = self.tools[tool_name](step, relevant_info)
                tool_results[tool_name] = tool_result
        
        # 4. 生成步骤结果
        step_result = self._generate_step_result(step, relevant_info, tool_results)
        
        return step_result
    
    def _search_tool(self, query, context):
        """搜索工具"""
        # 结合上下文进行搜索
        enhanced_query = f"{query} {' '.join([mem['result'][:100] for mem in self.memory[-3:]])}"
        
        results = self.vectorstore.similarity_search(enhanced_query, k=5)
        return [doc.page_content for doc in results]
    
    def _calculate_tool(self, expression, context):
        """计算工具"""
        # 安全的数学计算
        try:
            # 提取数学表达式
            math_expr = self._extract_math_expression(expression)
            result = eval(math_expr)  # 在实际应用中应使用更安全的计算方法
            return f"计算结果：{result}"
        except Exception as e:
            return f"计算错误：{str(e)}"
    
    def _visualize_tool(self, data_description, context):
        """可视化工具"""
        # 生成可视化代码
        viz_code = self._generate_visualization_code(data_description, context)
        
        # 执行可视化（这里简化为返回代码）
        return f"可视化代码：\n{viz_code}"
    
    def _summarize_tool(self, content, context):
        """摘要工具"""
        summary_prompt = f"""
        请对以下内容进行摘要：
        
        内容：{content}
        
        上下文：{context}
        
        请提供简洁而全面的摘要：
        """
        
        summary = self.llm(summary_prompt)
        return summary

# 生产级RAG系统架构
class ProductionRAGSystem:
    def __init__(self, config):
        self.config = config
        self.document_processor = DocumentProcessor()
        self.vector_store = self._init_vector_store()
        self.llm_client = self._init_llm_client()
        self.cache = self._init_cache()
        self.monitoring = self._init_monitoring()
    
    def _init_vector_store(self):
        """初始化向量数据库"""
        if self.config['vector_store']['type'] == 'pinecone':
            import pinecone
            pinecone.init(
                api_key=self.config['vector_store']['api_key'],
                environment=self.config['vector_store']['environment']
            )
            return pinecone.Index(self.config['vector_store']['index_name'])
        elif self.config['vector_store']['type'] == 'weaviate':
            import weaviate
            return weaviate.Client(
                url=self.config['vector_store']['url'],
                auth_client_secret=weaviate.AuthApiKey(
                    api_key=self.config['vector_store']['api_key']
                )
            )
        else:
            # 默认使用Chroma
            return Chroma(
                persist_directory=self.config['vector_store']['persist_directory']
            )
    
    def _init_cache(self):
        """初始化缓存系统"""
        import redis
        return redis.Redis(
            host=self.config['cache']['host'],
            port=self.config['cache']['port'],
            db=self.config['cache']['db']
        )
    
    def _init_monitoring(self):
        """初始化监控系统"""
        from prometheus_client import Counter, Histogram, Gauge
        
        return {
            'query_counter': Counter('rag_queries_total', 'Total RAG queries'),
            'response_time': Histogram('rag_response_time_seconds', 'RAG response time'),
            'cache_hit_rate': Gauge('rag_cache_hit_rate', 'Cache hit rate'),
            'error_counter': Counter('rag_errors_total', 'Total RAG errors', ['error_type'])
        }
    
    async def process_query(self, query, user_id=None):
        """异步查询处理"""
        start_time = time.time()
        
        try:
            # 1. 查询预处理
            processed_query = await self._preprocess_query(query)
            
            # 2. 缓存检查
            cache_key = self._generate_cache_key(processed_query, user_id)
            cached_result = await self._check_cache(cache_key)
            
            if cached_result:
                self.monitoring['cache_hit_rate'].inc()
                return cached_result
            
            # 3. 检索相关文档
            relevant_docs = await self._retrieve_documents(processed_query)
            
            # 4. 上下文优化
            optimized_context = await self._optimize_context(
                processed_query, relevant_docs
            )
            
            # 5. 生成回答
            response = await self._generate_response(
                processed_query, optimized_context
            )
            
            # 6. 后处理
            final_response = await self._postprocess_response(response)
            
            # 7. 缓存结果
            await self._cache_result(cache_key, final_response)
            
            # 8. 记录指标
            response_time = time.time() - start_time
            self.monitoring['query_counter'].inc()
            self.monitoring['response_time'].observe(response_time)
            
            return final_response
            
        except Exception as e:
            self.monitoring['error_counter'].labels(error_type=type(e).__name__).inc()
            raise
    
    async def _retrieve_documents(self, query):
        """异步文档检索"""
        # 并行执行多种检索策略
        tasks = [
            self._semantic_search(query),
            self._keyword_search(query),
            self._hybrid_search(query)
        ]
        
        results = await asyncio.gather(*tasks)
        
        # 结果融合和去重
        all_docs = []
        for result_set in results:
            all_docs.extend(result_set)
        
        # 去重并按相关性排序
        unique_docs = self._deduplicate_documents(all_docs)
        ranked_docs = self._rank_documents(query, unique_docs)
        
        return ranked_docs[:self.config['retrieval']['max_docs']]
    
    def _deduplicate_documents(self, documents):
        """文档去重"""
        seen_content = set()
        unique_docs = []
        
        for doc in documents:
            content_hash = hashlib.md5(doc.page_content.encode()).hexdigest()
            if content_hash not in seen_content:
                seen_content.add(content_hash)
                unique_docs.append(doc)
        
        return unique_docs

### 6.2 性能优化策略：提升系统响应速度

生产环境中的RAG系统需要在保证准确性的同时优化响应速度：

```python
# 性能优化组件
class RAGPerformanceOptimizer:
    def __init__(self):
        self.query_cache = LRUCache(maxsize=1000)
        self.embedding_cache = LRUCache(maxsize=5000)
        self.connection_pool = self._init_connection_pool()
    
    def _init_connection_pool(self):
        """初始化连接池"""
        import aiohttp
        connector = aiohttp.TCPConnector(
            limit=100,  # 总连接数限制
            limit_per_host=30,  # 每个主机连接数限制
            ttl_dns_cache=300,  # DNS缓存时间
            use_dns_cache=True
        )
        return aiohttp.ClientSession(connector=connector)
    
    async def batch_embedding_generation(self, texts):
        """批量生成嵌入向量"""
        # 检查缓存
        cached_embeddings = {}
        uncached_texts = []
        
        for text in texts:
            text_hash = hashlib.md5(text.encode()).hexdigest()
            if text_hash in self.embedding_cache:
                cached_embeddings[text] = self.embedding_cache[text_hash]
            else:
                uncached_texts.append(text)
        
        # 批量处理未缓存的文本
        if uncached_texts:
            batch_embeddings = await self._generate_embeddings_batch(uncached_texts)
            
            # 更新缓存
            for text, embedding in zip(uncached_texts, batch_embeddings):
                text_hash = hashlib.md5(text.encode()).hexdigest()
                self.embedding_cache[text_hash] = embedding
                cached_embeddings[text] = embedding
        
        return [cached_embeddings[text] for text in texts]
    
    async def parallel_document_processing(self, documents):
        """并行文档处理"""
        # 将文档分批处理
        batch_size = 50
        batches = [documents[i:i+batch_size] for i in range(0, len(documents), batch_size)]
        
        # 并行处理每个批次
        tasks = [self._process_document_batch(batch) for batch in batches]
        results = await asyncio.gather(*tasks)
        
        # 合并结果
        processed_docs = []
        for batch_result in results:
            processed_docs.extend(batch_result)
        
        return processed_docs
    
    def implement_query_optimization(self, query):
        """查询优化"""
        optimizations = {
            'query_expansion': self._expand_query(query),
            'query_rewriting': self._rewrite_query(query),
            'intent_detection': self._detect_intent(query)
        }
        
        return optimizations
    
    def _expand_query(self, query):
        """查询扩展"""
        # 使用同义词和相关词扩展查询
        expansion_prompt = f"""
        为以下查询生成3-5个相关的同义词或相关词：
        
        查询：{query}
        
        请只返回词汇，用逗号分隔：
        """
        
        expanded_terms = self.llm(expansion_prompt).strip().split(',')
        return [term.strip() for term in expanded_terms]

# 质量监控系统
class RAGQualityMonitor:
    def __init__(self):
        self.metrics_collector = MetricsCollector()
        self.alert_manager = AlertManager()
        self.quality_thresholds = {
            'response_time': 2.0,  # 秒
            'accuracy_score': 0.85,
            'user_satisfaction': 0.8,
            'cache_hit_rate': 0.6
        }
    
    def monitor_system_health(self):
        """系统健康监控"""
        health_metrics = {
            'vector_store_status': self._check_vector_store_health(),
            'llm_service_status': self._check_llm_service_health(),
            'cache_status': self._check_cache_health(),
            'memory_usage': self._get_memory_usage(),
            'cpu_usage': self._get_cpu_usage()
        }
        
        # 检查是否需要告警
        for metric, value in health_metrics.items():
            if self._should_alert(metric, value):
                self.alert_manager.send_alert(metric, value)
        
        return health_metrics
    
    def evaluate_response_quality(self, query, response, retrieved_docs):
        """评估响应质量"""
        quality_scores = {
            'relevance': self._calculate_relevance_score(query, response),
            'completeness': self._calculate_completeness_score(query, response),
            'accuracy': self._calculate_accuracy_score(response, retrieved_docs),
            'coherence': self._calculate_coherence_score(response)
        }
        
        overall_score = np.mean(list(quality_scores.values()))
        
        # 记录质量指标
        self.metrics_collector.record_quality_metrics(quality_scores)
        
        return overall_score, quality_scores