大模型RAG技术落地与优化实战

# 这是最基础的RAG实现
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.chains import RetrievalQA

# 加载文档
documents = load_documents()
# 切分
text_splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50)
chunks = text_splitter.split_documents(documents)
# 向量化存储
vectordb = Chroma.from_documents(chunks, HuggingFaceEmbeddings(model_name="BAAI/bge-large-zh-v1.5"))
# 检索问答
qa_chain = RetrievalQA.from_chain_type(llm=llm, retriever=vectordb.as_retriever())
result = qa_chain.run("什么是微服务架构？")

from langchain_text_splitters import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ("#", "header1"),
    ("##", "header2"), 
    ("###", "header3"),
]

markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
docs = markdown_splitter.split_text(markdown_content)

class HybridRetriever:
    def __init__(self, vector_store, bm25_index):
        self.vector_store = vector_store
        self.bm25_index = bm25_index
    
    def retrieve(self, query, top_k=5):
        # 向量检索
        vector_results = self.vector_store.similarity_search(query, k=top_k*2)
        # BM25关键词检索
        bm25_results = self.bm25_index.search(query, k=top_k*2)
        # RRF融合
        return self.reciprocal_rank_fusion(vector_results, bm25_results, top_k)

用户问题: "那个报警的事处理了没？"
改写后: "系统报警事件的处理状态和后续方案"

def rewrite_query(original_query, chat_history):
    prompt = f"""
    基于对话上下文，将用户的问题改写成更适合知识库检索的查询。
    
    历史对话：{chat_history[-3:]}
    用户问题：{original_query}
    
    改写要求：
    - 提取关键词，去掉口语化的表达
    - 补充必要的上下文信息
    - 保持问题的完整性
    
    改写结果：
    """
    return llm.generate(prompt)

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")

def rerank(query, candidates, top_k=3):
    pairs = [[query, doc.page_content] for doc in candidates]
    scores = reranker.predict(pairs)
    
    scored = list(zip(candidates, scores))
    scored.sort(key=lambda x: x[1], reverse=True)
    
    return [doc for doc, score in scored[:top_k]]

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever
)

compressed_docs = compression_retriever.get_relevant_documents(query)

class MultiSourceRetriever:
    def __init__(self):
        self.retrievers = {
            "vector": VectorRetriever(),
            "keyword": BM25Retriever(),
            "sql": SQLRetriever(),       # 从结构化数据中检索
            "graph": KnowledgeGraphRetriever(),  # 知识图谱检索
        }
    
    def retrieve(self, query, top_k=5):
        all_results = []
        for name, retriever in self.retrievers.items():
            results = retriever.retrieve(query, top_k=top_k)
            for r in results:
                r.source = name
                all_results.append(r)
        
        # RRF融合排序
        return self.rrf_merge(all_results, top_k)

class AdvancedRAGSystem:
    def __init__(self):
        self.query_rewriter = QueryRewriter(llm)
        self.retriever = MultiSourceRetriever()
        self.reranker = Reranker()
        self.response_generator = ResponseGenerator(llm)
    
    def answer(self, question, history=[]):
        # 1. 查询重写
        rewritten = self.query_rewriter.rewrite(question, history)
        
        # 2. 多路召回
        candidates = self.retriever.retrieve(rewritten, top_k=20)
        
        # 3. 重排序
        top_docs = self.reranker.rerank(rewritten, candidates, top_k=5)
        
        # 4. 生成回答
        answer = self.response_generator.generate(question, top_docs)
        
        return answer

RAG评估矩阵
├── 检索质量
│   ├── 命中率（Hit Rate）：相关文档是否被召回
│   └── MRR（Mean Reciprocal Rank）：相关文档的排名
├── 生成质量
│   ├── 忠实度（Faithfulness）：是否基于检索结果
│   ├── 相关性（Relevance）：是否回答了问题
│   └── 完整性（Completeness）：回答是否全面
└── 用户体验
    ├── 响应速度
    └── 引用准确性

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision

# 使用RAGAS框架做自动化评估
result = evaluate(
    dataset=test_dataset,
    metrics=[faithfulness, answer_relevancy, context_precision],
    llm=eval_llm,
    embeddings=eval_embeddings
)