如果你正在构建AI Agent应用，却经常遇到这样的困境：用户问了一个技术问题，Agent要么回答得模棱两可，要么直接说"我不知道"，甚至开始胡编乱造——那么问题很可能出在知识库构建环节。传统的AI模型虽然有强大的推理能力，但缺乏对特定领域知识的精准掌握。

这正是RAG技术要解决的核心问题。RAG不是简单的"检索+生成"，而是一套完整的知识管理架构。在AI Agent开发中，一个设计良好的RAG系统能让你的Agent从"通才"变成"专才"，真正掌握你希望它了解的领域知识。

本文将带你深入LangChain框架下的RAG系统构建，从基础概念到生产级实践，重点解决三个关键问题：如何构建高质量的知识库、如何设计高效的检索策略，以及如何将RAG与Agent工作流无缝集成。

1. RAG在AI Agent中的核心价值与定位

1.1 为什么单纯的LLM不够用？

大型语言模型在通用知识上表现优异，但在特定领域存在明显局限：

• 知识时效性 ：模型训练数据有截止日期，无法获取最新信息
• 领域专业性 ：对内部文档、专有API、企业流程等私有知识掌握有限
• 事实准确性 ：容易产生"幻觉"，编造看似合理但实际错误的信息
• 成本控制 ：为每个问题重新训练模型既不现实也不经济

1.2 RAG如何弥补这些缺陷？

RAG通过外部知识库为LLM提供实时、准确、专业的上下文信息：

用户问题 → 检索相关文档 → 增强提示词 → LLM生成答案

这种架构的优势在于：

• 知识可更新 ：只需更新向量数据库，无需重新训练模型
• 来源可追溯 ：每个回答都能追溯到具体的文档片段
• 成本可控 ：利用现有LLM能力，避免定制化训练的高成本
• 安全性更高 ：可以控制知识范围，避免泄露敏感信息

1.3 Agent与RAG的协同效应

在AI Agent架构中，RAG扮演着"长期记忆"和"专业知识库"的角色：

• 决策支持 ：Agent在制定行动计划时，可以检索相关知识作为决策依据
• 工具使用 ：RAG可以提供API文档、使用示例等工具相关信息的精准检索
• 多轮对话 ：在复杂的多轮交互中，保持知识上下文的一致性
• 子任务分发 ：Agent可以将知识检索任务委托给专门的RAG子模块

2. LangChain RAG核心组件深度解析

2.1 文档加载与处理流水线

文档处理是RAG系统的第一步，也是最容易出问题的环节。LangChain提供了丰富的文档加载器：

from langchain_community.document_loaders import (
    PyPDFLoader, 
    TextLoader,
    WebBaseLoader,
    UnstructuredFileLoader
)

# 加载本地PDF文档
def load_pdf_documents(pdf_paths):
    documents = []
    for pdf_path in pdf_paths:
        loader = PyPDFLoader(pdf_path)
        docs = loader.load()
        # 为每个文档添加元数据
        for doc in docs:
            doc.metadata["source"] = pdf_path
            doc.metadata["type"] = "pdf"
        documents.extend(docs)
    return documents

# 加载网页内容
def load_web_documents(urls):
    documents = []
    for url in urls:
        loader = WebBaseLoader(url)
        docs = loader.load()
        for doc in docs:
            doc.metadata["source"] = url
            doc.metadata["type"] = "web"
        documents.extend(docs)
    return documents

2.2 文本分块策略与最佳实践

文本分块的质量直接影响检索效果。常见的错误是使用固定的分块大小，忽略了文档的结构信息。

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
    MarkdownHeaderTextSplitter,
    Language
)

# 针对代码文档的专用分割器
def split_code_documents(documents):
    # 识别编程语言并进行语义分块
    code_splitter = RecursiveCharacterTextSplitter.from_language(
        language=Language.PYTHON,
        chunk_size=1000,
        chunk_overlap=200
    )
    return code_splitter.split_documents(documents)

# 针对Markdown文档的智能分块
def split_markdown_documents(documents):
    # 根据标题层级进行分块
    headers_to_split_on = [
        ("#", "Header 1"),
        ("##", "Header 2"), 
        ("###", "Header 3"),
    ]
    
    markdown_splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=headers_to_split_on,
        strip_headers=False
    )
    
    split_docs = []
    for doc in documents:
        splits = markdown_splitter.split_text(doc.page_content)
        for split in splits:
            split.metadata.update(doc.metadata)
            split_docs.append(split)
    
    return split_docs

# 通用文档分块配置
def get_optimal_text_splitter():
    return RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        length_function=len,
        separators=["\n\n", "\n", ". ", "! ", "? ", " ", ""]
    )

2.3 嵌入模型选择与向量化

嵌入模型的选择对检索质量有决定性影响。不同模型在语义理解、多语言支持、领域适应性方面表现各异。

from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings
import numpy as np

class EmbeddingManager:
    def __init__(self, model_name="text-embedding-3-small"):
        self.embeddings = OpenAIEmbeddings(model=model_name)
        
    def get_embedding_model(self):
        return self.embeddings
    
    def validate_embedding_quality(self, sample_texts):
        """验证嵌入模型的质量"""
        embeddings = self.embeddings.embed_documents(sample_texts)
        
        # 计算相似度矩阵
        similarity_matrix = np.dot(embeddings, np.array(embeddings).T)
        
        # 检查相似度是否合理
        print(f"嵌入维度: {len(embeddings[0])}")
        print(f"相似度范围: {similarity_matrix.min():.3f} - {similarity_matrix.max():.3f}")
        
        return embeddings

# 备用嵌入模型配置
def get_fallback_embeddings():
    """获取本地部署的嵌入模型"""
    return HuggingFaceEmbeddings(
        model_name="sentence-transformers/all-MiniLM-L6-v2",
        model_kwargs={'device': 'cpu'},
        encode_kwargs={'normalize_embeddings': True}
    )

3. 向量数据库选型与实战配置

3.1 主流向量数据库对比

根据应用场景选择合适的向量数据库至关重要：

数据库	适用场景	优点	缺点
Chroma	开发测试、小型应用	简单易用、无需外部依赖	生产环境需要持久化方案
Pinecone	云原生、企业级	全托管、高性能	付费服务、有vendor锁定风险
Qdrant	自部署、高性能	开源、功能丰富	运维复杂度较高
Weaviate	混合搜索、生产级	支持多种搜索方式	配置相对复杂

3.2 ChromaDB实战配置

对于大多数开发场景，ChromaDB是一个不错的起点：

import chromadb
from langchain_chroma import Chroma

class VectorStoreManager:
    def __init__(self, persist_directory="./chroma_db"):
        self.persist_directory = persist_directory
        self.client = chromadb.PersistentClient(path=persist_directory)
    
    def create_vector_store(self, documents, embeddings, collection_name="docs"):
        """创建向量存储"""
        vector_store = Chroma.from_documents(
            documents=documents,
            embedding=embeddings,
            persist_directory=self.persist_directory,
            collection_name=collection_name
        )
        return vector_store
    
    def get_retriever(self, vector_store, search_type="similarity", k=4):
        """配置检索器"""
        retriever = vector_store.as_retriever(
            search_type=search_type,
            search_kwargs={"k": k}
        )
        return retriever
    
    def hybrid_retrieval(self, vector_store, query, k=4, alpha=0.5):
        """混合检索：结合语义搜索和关键词搜索"""
        # 语义搜索
        semantic_results = vector_store.similarity_search(query, k=k)
        
        # 可以结合其他检索方式
        # keyword_results = keyword_search(query, k=k)
        
        # 结果融合逻辑
        return semantic_results

# 使用示例
def setup_rag_system():
    # 1. 加载文档
    documents = load_pdf_documents(["doc1.pdf", "doc2.pdf"])
    
    # 2. 文本分块
    splitter = get_optimal_text_splitter()
    chunks = splitter.split_documents(documents)
    
    # 3. 初始化嵌入模型
    embedding_manager = EmbeddingManager()
    embeddings = embedding_manager.get_embedding_model()
    
    # 4. 创建向量存储
    vs_manager = VectorStoreManager()
    vector_store = vs_manager.create_vector_store(chunks, embeddings)
    
    # 5. 配置检索器
    retriever = vs_manager.get_retriever(vector_store)
    
    return retriever, vector_store

4. 高级检索策略与优化技巧

4.1 多路检索与结果融合

单一检索策略往往无法满足复杂需求，多路检索能显著提升召回率：

from typing import List, Dict
from langchain.schema import Document

class AdvancedRetrievalSystem:
    def __init__(self, vector_store, max_results=6):
        self.vector_store = vector_store
        self.max_results = max_results
    
    def multi_retrieval(self, query: str) -> List[Document]:
        """多路检索策略"""
        results = []
        
        # 1. 基础语义搜索
        semantic_results = self.vector_store.similarity_search(
            query, k=self.max_results
        )
        results.extend(semantic_results)
        
        # 2. MMR搜索（最大化边际相关度）
        mmr_results = self.vector_store.max_marginal_relevance_search(
            query, k=self.max_results, fetch_k=10
        )
        results.extend([r for r in mmr_results if r not in results])
        
        # 3. 按特定字段过滤搜索（如果有元数据）
        # filtered_results = self.filtered_search(query)
        # results.extend(filtered_results)
        
        return self.deduplicate_documents(results)
    
    def deduplicate_documents(self, documents: List[Document]) -> List[Document]:
        """文档去重"""
        seen_content = set()
        unique_docs = []
        
        for doc in documents:
            content_hash = hash(doc.page_content[:200])  # 前200字符作为去重依据
            if content_hash not in seen_content:
                seen_content.add(content_hash)
                unique_docs.append(doc)
        
        return unique_docs[:self.max_results]
    
    def rerank_results(self, query: str, documents: List[Document]) -> List[Document]:
        """对检索结果进行重排序"""
        # 可以集成专门的重排序模型，如bge-reranker
        # 这里使用简单的基于查询相关性的启发式方法
        scored_docs = []
        
        for doc in documents:
            score = self.calculate_relevance_score(query, doc.page_content)
            scored_docs.append((score, doc))
        
        # 按分数降序排列
        scored_docs.sort(key=lambda x: x[0], reverse=True)
        return [doc for _, doc in scored_docs]
    
    def calculate_relevance_score(self, query: str, content: str) -> float:
        """简单的相关性评分（实际项目中可以使用更复杂的算法）"""
        query_terms = set(query.lower().split())
        content_terms = set(content.lower().split())
        
        intersection = query_terms & content_terms
        if not query_terms:
            return 0.0
        
        return len(intersection) / len(query_terms)

4.2 查询理解与改写

用户的原始查询往往不够精确，查询改写能显著提升检索效果：

class QueryUnderstanding:
    def __init__(self, llm):
        self.llm = llm
    
    def expand_query(self, original_query: str) -> List[str]:
        """查询扩展：生成相关的同义词和变体"""
        expansion_prompt = f"""
        根据用户查询生成3个相关的搜索查询变体，保持原意但使用不同的表达方式。
        
        原始查询: {original_query}
        
        生成变体:
        1.
        2. 
        3.
        """
        
        # 使用LLM生成查询变体
        response = self.llm.invoke(expansion_prompt)
        variants = self.parse_variants(response)
        
        return [original_query] + variants
    
    def parse_variants(self, response):
        """解析LLM生成的查询变体"""
        # 简化处理，实际需要更健壮的解析逻辑
        lines = response.strip().split('\n')
        variants = []
        for line in lines:
            if line.strip() and not line.startswith('原始查询'):
                variant = line.split('.', 1)[-1].strip()
                if variant:
                    variants.append(variant)
        return variants[:3]  # 最多返回3个变体
    
    def contextualize_query(self, query: str, conversation_history: List) -> str:
        """基于对话历史进行查询上下文化"""
        if not conversation_history:
            return query
        
        # 提取最近几轮对话作为上下文
        recent_history = conversation_history[-3:]  # 最近3轮
        context = " ".join([f"Q: {q} A: {a}" for q, a in recent_history])
        
        contextualized_prompt = f"""
        基于以下对话历史和当前查询，生成一个完整的、包含上下文的搜索查询。
        
        对话历史:
        {context}
        
        当前查询: {query}
        
        优化后的查询:
        """
        
        optimized_query = self.llm.invoke(contextualized_prompt)
        return optimized_query.strip()

5. RAG与AI Agent的深度集成

5.1 Agent中的RAG工具设计

将RAG系统封装成Agent可用的工具：

from langchain.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

class RagSearchInput(BaseModel):
    query: str = Field(description="搜索查询")
    max_results: int = Field(default=4, description="最大返回结果数")

class RagSearchTool(BaseTool):
    name = "knowledge_base_search"
    description = "从知识库中搜索相关信息"
    args_schema: Type[BaseModel] = RagSearchInput
    
    def __init__(self, retriever, **kwargs):
        super().__init__(**kwargs)
        self.retriever = retriever
    
    def _run(self, query: str, max_results: int = 4) -> str:
        """执行搜索"""
        try:
            # 获取检索结果
            docs = self.retriever.get_relevant_documents(query)[:max_results]
            
            if not docs:
                return "未找到相关信息"
            
            # 格式化结果
            result = f"找到 {len(docs)} 个相关文档:\n\n"
            for i, doc in enumerate(docs, 1):
                source = doc.metadata.get('source', '未知来源')
                result += f"{i}. 来源: {source}\n"
                result += f"内容: {doc.page_content[:300]}...\n\n"
            
            return result
            
        except Exception as e:
            return f"搜索过程中出现错误: {str(e)}"

class AdvancedRagAgent:
    def __init__(self, llm, retriever, tools=[]):
        self.llm = llm
        self.retriever = retriever
        
        # 创建RAG搜索工具
        rag_tool = RagSearchTool(retriever=retriever)
        
        # 组合所有工具
        self.tools = [rag_tool] + tools
        
        # 创建Agent
        self.agent = self._create_agent()
    
    def _create_agent(self):
        """创建带有RAG能力的Agent"""
        from langchain.agents import AgentType, initialize_agent
        
        agent = initialize_agent(
            tools=self.tools,
            llm=self.llm,
            agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
            verbose=True,
            handle_parsing_errors=True
        )
        
        return agent
    
    def invoke(self, query: str):
        """执行查询"""
        return self.agent.invoke({"input": query})

5.2 多智能体协作的RAG架构

对于复杂任务，可以采用多智能体架构，让专门的智能体处理知识检索：

class MultiAgentRagSystem:
    def __init__(self, llm, vector_store):
        self.llm = llm
        self.vector_store = vector_store
        self.setup_agents()
    
    def setup_agents(self):
        """设置不同类型的智能体"""
        # 检索智能体 - 专门负责知识检索
        self.retrieval_agent = self.create_retrieval_agent()
        
        # 分析智能体 - 负责内容分析
        self.analysis_agent = self.create_analysis_agent()
        
        # 合成智能体 - 负责最终答案合成
        self.synthesis_agent = self.create_synthesis_agent()
    
    def create_retrieval_agent(self):
        """创建检索智能体"""
        retrieval_prompt = """你是一个专业的检索专家。你的任务是：
        1. 理解用户问题的核心信息需求
        2. 生成有效的搜索查询
        3. 从知识库中检索最相关的信息
        
        用户问题: {question}
        
        请生成3个搜索查询，然后执行检索："""
        
        # 简化的检索智能体实现
        class RetrievalAgent:
            def __init__(self, vector_store):
                self.vector_store = vector_store
            
            def retrieve(self, question):
                # 多查询检索
                queries = self.generate_queries(question)
                all_results = []
                
                for query in queries:
                    results = self.vector_store.similarity_search(query, k=2)
                    all_results.extend(results)
                
                return self.deduplicate(all_results)
            
            def generate_queries(self, question):
                # 简单的查询生成逻辑
                return [question, f"详细说明 {question}", f"{question} 的具体步骤"]
        
        return RetrievalAgent(self.vector_store)
    
    def process_query(self, question):
        """处理用户查询的多智能体流程"""
        # 1. 检索阶段
        retrieved_docs = self.retrieval_agent.retrieve(question)
        
        # 2. 分析阶段
        analysis_results = self.analyze_documents(question, retrieved_docs)
        
        # 3. 合成阶段
        final_answer = self.synthesize_answer(question, analysis_results)
        
        return final_answer
    
    def analyze_documents(self, question, documents):
        """分析检索到的文档"""
        analysis_prompt = f"""
        基于以下文档内容，分析它们如何回答用户问题。
        
        用户问题: {question}
        
        文档内容:
        {self.format_documents(documents)}
        
        请提取关键信息点：
        """
        
        analysis = self.llm.invoke(analysis_prompt)
        return analysis
    
    def format_documents(self, documents):
        """格式化文档用于提示词"""
        return "\n\n".join([
            f"文档 {i+1} (来源: {doc.metadata.get('source', '未知')}):\n{doc.page_content}"
            for i, doc in enumerate(documents)
        ])

6. 生产环境部署与优化

6.1 性能优化策略

import time
from functools import wraps
from concurrent.futures import ThreadPoolExecutor

def timing_decorator(func):
    """执行时间统计装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"{func.__name__} 执行时间: {end_time - start_time:.2f}秒")
        return result
    return wrapper

class OptimizedRagSystem:
    def __init__(self, vector_store, llm, cache_size=1000):
        self.vector_store = vector_store
        self.llm = llm
        self.cache = {}  # 简单的缓存实现
        self.cache_size = cache_size
    
    @timing_decorator
    def optimized_retrieve(self, query, use_cache=True):
        """带缓存的优化检索"""
        if use_cache:
            cache_key = hash(query)
            if cache_key in self.cache:
                print("缓存命中!")
                return self.cache[cache_key]
        
        # 并行处理多个检索策略
        with ThreadPoolExecutor() as executor:
            future_semantic = executor.submit(
                self.vector_store.similarity_search, query, k=3
            )
            future_mmr = executor.submit(
                self.vector_store.max_marginal_relevance_search, query, k=3
            )
            
            semantic_results = future_semantic.result()
            mmr_results = future_mmr.result()
        
        # 结果融合
        all_results = semantic_results + mmr_results
        final_results = self.deduplicate_results(all_results)[:4]
        
        # 更新缓存
        if use_cache and len(self.cache) < self.cache_size:
            self.cache[hash(query)] = final_results
        
        return final_results
    
    def deduplicate_results(self, documents):
        """结果去重"""
        seen = set()
        unique_docs = []
        for doc in documents:
            content_hash = hash(doc.page_content[:500])
            if content_hash not in seen:
                seen.add(content_hash)
                unique_docs.append(doc)
        return unique_docs

6.2 监控与评估体系

class RagMonitor:
    def __init__(self):
        self.metrics = {
            "retrieval_times": [],
            "cache_hit_rate": 0,
            "queries_processed": 0
        }
    
    def log_retrieval(self, query, results, retrieval_time):
        """记录检索日志"""
        self.metrics["retrieval_times"].append(retrieval_time)
        self.metrics["queries_processed"] += 1
        
        print(f"查询: {query}")
        print(f"返回结果数: {len(results)}")
        print(f"检索时间: {retrieval_time:.2f}秒")
        
        # 记录到文件或监控系统
        self._write_to_log(query, results, retrieval_time)
    
    def evaluate_retrieval_quality(self, query, results, ground_truth=None):
        """评估检索质量"""
        if not results:
            return {"recall": 0, "precision": 0, "f1": 0}
        
        # 简单的质量评估指标
        relevance_scores = []
        for doc in results:
            score = self._calculate_relevance(query, doc.page_content)
            relevance_scores.append(score)
        
        avg_relevance = sum(relevance_scores) / len(relevance_scores)
        
        return {
            "avg_relevance": avg_relevance,
            "results_count": len(results),
            "relevance_scores": relevance_scores
        }
    
    def _calculate_relevance(self, query, content):
        """计算查询与内容的相关性"""
        # 简化的相关性计算
        query_terms = set(query.lower().split())
        content_terms = set(content.lower().split())
        common_terms = query_terms & content_terms
        
        if not query_terms:
            return 0
        
        return len(common_terms) / len(query_terms)
    
    def generate_report(self):
        """生成性能报告"""
        if not self.metrics["retrieval_times"]:
            return "尚无数据"
        
        avg_time = sum(self.metrics["retrieval_times"]) / len(self.metrics["retrieval_times"])
        max_time = max(self.metrics["retrieval_times"])
        min_time = min(self.metrics["retrieval_times"])
        
        report = f"""
        RAG系统性能报告:
        - 总查询数: {self.metrics["queries_processed"]}
        - 平均检索时间: {avg_time:.2f}秒
        - 最快检索: {min_time:.2f}秒
        - 最慢检索: {max_time:.2f}秒
        - 缓存命中率: {self.metrics['cache_hit_rate']:.1%}
        """
        
        return report

7. 常见问题与解决方案

7.1 检索质量不佳的排查流程

当RAG系统表现不佳时，可以按照以下流程排查：

class RagTroubleshooter:
    def __init__(self, rag_system):
        self.rag_system = rag_system
    
    def diagnose_issues(self, query, expected_results=None):
        """诊断RAG系统问题"""
        print("=== RAG系统问题诊断 ===")
        print(f"测试查询: {query}")
        
        issues = []
        
        # 1. 检查检索环节
        retrieval_issues = self._check_retrieval(query)
        issues.extend(retrieval_issues)
        
        # 2. 检查嵌入质量
        embedding_issues = self._check_embeddings()
        issues.extend(embedding_issues)
        
        # 3. 检查文档处理
        processing_issues = self._check_document_processing()
        issues.extend(processing_issues)
        
        # 4. 生成修复建议
        recommendations = self._generate_recommendations(issues)
        
        return {
            "issues": issues,
            "recommendations": recommendations
        }
    
    def _check_retrieval(self, query):
        """检查检索环节"""
        issues = []
        
        try:
            results = self.rag_system.retriever.get_relevant_documents(query)
            
            if len(results) == 0:
                issues.append("检索返回结果为空")
            elif len(results) < 3:
                issues.append("检索结果数量较少")
            
            # 检查结果相关性
            for i, doc in enumerate(results):
                relevance = self._calculate_relevance(query, doc.page_content)
                if relevance < 0.1:
                    issues.append(f"结果{i+1}相关性较低: {relevance:.2f}")
        
        except Exception as e:
            issues.append(f"检索过程出错: {str(e)}")
        
        return issues
    
    def _check_embeddings(self):
        """检查嵌入模型"""
        issues = []
        
        # 测试嵌入一致性
        test_texts = ["这是一个测试", "这是另一个测试"]
        try:
            embeddings = self.rag_system.embeddings.embed_documents(test_texts)
            if len(embeddings[0]) < 100:
                issues.append("嵌入维度较低，可能影响检索精度")
        except Exception as e:
            issues.append(f"嵌入生成失败: {str(e)}")
        
        return issues

7.2 典型问题及解决方案

问题现象	可能原因	解决方案
检索结果不相关	分块策略不合理、嵌入模型不适合	调整分块大小、尝试不同嵌入模型
回答缺乏细节	检索结果过于笼统	减小分块大小、增加检索数量
回答包含矛盾信息	知识库中存在冲突内容	实施知识去重、添加置信度评分
响应速度慢	向量数据库性能问题、网络延迟	使用本地向量数据库、添加缓存层

8. 最佳实践与进阶技巧

8.1 知识库维护策略

class KnowledgeBaseMaintainer:
    def __init__(self, vector_store, embedding_model):
        self.vector_store = vector_store
        self.embedding_model = embedding_model
    
    def incremental_update(self, new_documents, batch_size=100):
        """增量更新知识库"""
        print(f"开始增量更新，处理 {len(new_documents)} 个文档")
        
        for i in range(0, len(new_documents), batch_size):
            batch = new_documents[i:i + batch_size]
            print(f"处理批次 {i//batch_size + 1}")
            
            # 检查重复内容
            unique_docs = self.filter_duplicates(batch)
            
            if unique_docs:
                self.vector_store.add_documents(unique_docs)
                print(f"添加了 {len(unique_docs)} 个唯一文档")
    
    def filter_duplicates(self, documents, similarity_threshold=0.95):
        """过滤重复文档"""
        unique_docs = []
        existing_contents = set()
        
        for doc in documents:
            content_hash = self.get_content_hash(doc.page_content)
            if content_hash not in existing_contents:
                existing_contents.add(content_hash)
                unique_docs.append(doc)
        
        print(f"去重后保留 {len(unique_docs)}/{len(documents)} 个文档")
        return unique_docs
    
    def get_content_hash(self, content):
        """生成内容哈希值"""
        import hashlib
        return hashlib.md5(content.encode()).hexdigest()
    
    def quality_check(self, sample_queries):
        """知识库质量检查"""
        print("执行知识库质量检查...")
        
        for query in sample_queries:
            results = self.vector_store.similarity_search(query, k=3)
            print(f"查询: {query}")
            print(f"返回结果: {len(results)} 个")
            
            for i, doc in enumerate(results):
                relevance = self.calculate_relevance(query, doc.page_content)
                print(f"  结果 {i+1}: 相关性 {relevance:.2f}")

8.2 安全与权限控制

class SecureRagSystem:
    def __init__(self, vector_store, auth_provider):
        self.vector_store = vector_store
        self.auth_provider = auth_provider
    
    def secure_retrieve(self, query, user_context):
        """带权限控制的检索"""
        # 验证用户权限
        if not self.auth_provider.has_access(user_context, "knowledge_base"):
            raise PermissionError("用户没有访问知识库的权限")
        
        # 执行检索
        results = self.vector_store.similarity_search(query)
        
        # 过滤敏感信息
        filtered_results = self.filter_sensitive_content(results, user_context)
        
        return filtered_results
    
    def filter_sensitive_content(self, documents, user_context):
        """过滤敏感内容"""
        filtered_docs = []
        
        for doc in documents:
            # 检查文档权限标签
            doc_permissions = doc.metadata.get('permissions', [])
            user_permissions = user_context.get('permissions', [])
            
            # 只有用户有权限查看的文档才返回
            if self.has_required_permissions(doc_permissions, user_permissions):
                # 进一步过滤敏感内容
                filtered_content = self.redact_sensitive_info(doc.page_content)
                filtered_doc = doc.copy()
                filtered_doc.page_content = filtered_content
                filtered_docs.append(filtered_doc)
        
        return filtered_docs
    
    def has_required_permissions(self, doc_permissions, user_permissions):
        """检查权限匹配"""
        if not doc_permissions:  # 文档没有权限要求
            return True
        
        return any(perm in user_permissions for perm in doc_permissions)

构建生产级的RAG系统需要综合考虑检索质量、性能、安全性和可维护性。本文介绍的LangChain RAG架构提供了从基础到高级的完整解决方案，但实际项目中还需要根据具体需求进行调整和优化。

关键是要建立持续改进的机制：定期评估检索质量、监控系统性能、更新知识库内容。只有这样，才能确保RAG系统在AI Agent应用中发挥最大的价值。