文脉定序代码实例：集成LangChain实现自动Rerank Chain的完整Demo

文脉定序代码实例：集成LangChain实现自动Rerank Chain的完整Demo

1. 引言：为什么需要智能语义重排序？

在日常的信息检索中，我们经常遇到这样的困境：搜索引擎返回了大量相关结果，但最精准的答案往往不在前列。传统的关键词匹配和向量相似度搜索虽然能找到相关内容，但在"精准排序"方面存在明显短板。

这就是「文脉定序」要解决的核心问题——它不是简单的搜索工具，而是专门为提升检索精度设计的AI重排序系统。基于顶尖的BGE语义模型，它能理解问题的深层含义，从大量候选结果中精准识别出真正相关的答案。

本文将带你通过完整的代码示例，学习如何将文脉定序集成到LangChain框架中，构建自动化的Rerank Chain，让你的检索系统实现从"搜得到"到"排得准"的质的飞跃。

2. 环境准备与快速部署

2.1 安装必要的依赖包

首先确保你的Python环境在3.8以上，然后安装以下依赖：

pip install langchain langchain-community torch transformers pip install sentence-transformers # 用于文本嵌入和重排序

2.2 获取文脉定序模型访问权限

文脉定序基于BGE-Reranker-v2-m3模型，你需要先获取相应的API密钥或模型访问权限：

import os from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.retrievers import BM25Retriever from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import CrossEncoderReranker from langchain_core.documents import Document # 设置模型路径或API端点 MODEL_NAME = "BAAI/bge-reranker-v2-m3"

3. 基础概念快速入门

3.1 什么是重排序（Reranking）？

重排序是信息检索中的关键步骤，它发生在初步检索之后。简单来说：

1. 初步检索：先用快速但相对粗糙的方法（如关键词匹配、向量相似度）找到大量相关文档
2. 重排序：再用更精准但计算量更大的方法对这些文档重新排序，把最相关的排到最前面

文脉定序使用的交叉注意力机制（Cross-Attention）能够深入理解问题和答案之间的语义关联，比简单的相似度计算准确得多。

3.2 LangChain中的Rerank Chain

LangChain提供了标准的重排序接口，我们可以轻松集成文脉定序：

from transformers import AutoModelForSequenceClassification, AutoTokenizer import torch class WenMaiReranker: def __init__(self, model_name=MODEL_NAME): self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForSequenceClassification.from_pretrained(model_name) self.model.eval() def rerank(self, query, documents, top_k=5): """对文档进行重排序""" pairs = [(query, doc.page_content) for doc in documents] with torch.no_grad(): inputs = self.tokenizer(pairs, padding=True, truncation=True, return_tensors="pt", max_length=512) scores = self.model(**inputs).logits.squeeze(dim=1) # 按得分排序 sorted_indices = torch.argsort(scores, descending=True) return [documents[i] for i in sorted_indices[:top_k]], [scores[i].item() for i in sorted_indices[:top_k]]

4. 完整实战：构建自动Rerank Chain

4.1 准备示例数据和文档库

首先，我们创建一个简单的文档库来演示整个流程：

# 示例文档库 - 模拟知识库内容 documents = [ Document(page_content="Python是一种高级编程语言，由Guido van Rossum创建", metadata={"source": "python_intro"}), Document(page_content="机器学习是人工智能的一个分支，专注于算法开发", metadata={"source": "ml_definition"}), Document(page_content="深度学习使用神经网络处理复杂模式识别任务", metadata={"source": "dl_definition"}), Document(page_content="LangChain是用于开发大语言模型应用的框架", metadata={"source": "langchain_info"}), Document(page_content="文脉定序是基于BGE模型的智能重排序系统", metadata={"source": "wenmai_info"}), Document(page_content="Transformer模型在自然语言处理中取得突破性进展", metadata={"source": "transformer_info"}), Document(page_content="向量数据库用于高效存储和检索嵌入向量", metadata={"source": "vector_db_info"}), Document(page_content="RAG（检索增强生成）结合检索和大语言模型", metadata={"source": "rag_definition"}), Document(page_content="BGE模型是智源研究院开发的高质量嵌入模型", metadata={"source": "bge_model_info"}), Document(page_content="交叉注意力机制能够捕捉序列间的复杂关系", metadata={"source": "cross_attention_info"}) ]

4.2 构建初步检索器

在重排序之前，我们需要一个初步检索器来获取候选文档：

from langchain.text_splitter import CharacterTextSplitter from langchain_community.vectorstores import FAISS from langchain.retrievers import BM25Retriever, EnsembleRetriever # 创建BM25检索器（基于关键词） bm25_retriever = BM25Retriever.from_documents(documents) bm25_retriever.k = 10 # 检索10个候选文档 # 创建向量检索器（基于语义相似度） embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-en-v1.5") vectorstore = FAISS.from_documents(documents, embeddings) vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10}) # 组合检索器（提高召回率） ensemble_retriever = EnsembleRetriever( retrievers=[bm25_retriever, vector_retriever], weights=[0.5, 0.5] )

4.3 集成文脉定序重排序

现在将文脉定序集成到LangChain的重排序管道中：

from langchain.retrievers.document_compressors import DocumentCompressorPipeline from langchain_community.document_transformers import EmbeddingsRedundantFilter # 创建文脉定序重排序器 wenmai_reranker = WenMaiReranker() # 创建重排序管道 pipeline = DocumentCompressorPipeline(transformers=[wenmai_reranker]) # 创建最终的压缩检索器 compression_retriever = ContextualCompressionRetriever( base_compressor=pipeline, base_retriever=ensemble_retriever )

4.4 完整流程测试

让我们测试整个重排序流程：

# 测试查询 query = "什么是文脉定序系统？" print("=== 初步检索结果（重排序前）===") initial_results = ensemble_retriever.invoke(query) for i, doc in enumerate(initial_results): print(f"{i+1}. {doc.page_content[:80]}... [来源: {doc.metadata['source']}]") print("\n=== 文脉定序重排序后结果 ===") final_results, scores = wenmai_reranker.rerank(query, initial_results, top_k=3) for i, (doc, score) in enumerate(zip(final_results, scores)): print(f"{i+1}. 得分: {score:.4f} - {doc.page_content}... [来源: {doc.metadata['source']}]")

5. 高级应用：自定义重排序策略

5.1 多维度评分策略

文脉定序支持复杂的重排序策略，你可以结合多个维度进行评分：

class AdvancedWenMaiReranker(WenMaiReranker): def advanced_rerank(self, query, documents, top_k=5, relevance_weight=0.7, novelty_weight=0.3): """ 高级重排序：结合相关性和新颖性 """ # 基础相关性评分 pairs = [(query, doc.page_content) for doc in documents] with torch.no_grad(): inputs = self.tokenizer(pairs, padding=True, truncation=True, return_tensors="pt", max_length=512) relevance_scores = self.model(**inputs).logits.squeeze(dim=1) # 简单的新颖性评分（基于内容多样性） contents = [doc.page_content for doc in documents] novelty_scores = self._calculate_novelty_scores(contents) # 组合评分 combined_scores = (relevance_weight * relevance_scores + novelty_weight * novelty_scores) sorted_indices = torch.argsort(combined_scores, descending=True) return [documents[i] for i in sorted_indices[:top_k]] def _calculate_novelty_scores(self, contents): """计算内容新颖性分数（简化版）""" # 这里可以使用更复杂的新颖性计算逻辑 scores = torch.ones(len(contents)) return scores

5.2 批量处理优化

对于大量文档，我们可以优化批量处理：

def batch_rerank(self, queries, document_batches, batch_size=8, top_k=5): """批量重排序优化""" results = [] for i in range(0, len(queries), batch_size): batch_queries = queries[i:i+batch_size] batch_docs = document_batches[i:i+batch_size] # 处理当前批次 batch_results = [] for query, docs in zip(batch_queries, batch_docs): ranked_docs, scores = self.rerank(query, docs, top_k) batch_results.append((ranked_docs, scores)) results.extend(batch_results) return results

6. 实际应用场景与效果对比

6.1 RAG系统中的应用

在检索增强生成（RAG）系统中，文脉定序可以显著提升回答质量：

from langchain.llms import OpenAI from langchain.chains import RetrievalQA # 创建带重排序的RAG系统 llm = OpenAI(temperature=0) # 使用实际API密钥 # 不带重排序的RAG basic_qa = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=ensemble_retriever ) # 带文脉定序重排序的RAG advanced_qa = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=compression_retriever ) # 对比测试 test_question = "文脉定序基于什么技术？有什么优势？" print("基础RAG回答:", basic_qa.run(test_question)) print("高级RAG回答:", advanced_qa.run(test_question))

6.2 效果对比分析

我们通过实际测试来展示文脉定序的效果提升：

7. 常见问题与解决方案

7.1 性能优化建议

文脉定序虽然准确，但计算量较大，以下是一些优化建议：

# 1. 使用FP16精度加速 model = AutoModelForSequenceClassification.from_pretrained( MODEL_NAME, torch_dtype=torch.float16 ) # 2. 限制输入长度 def truncate_text(text, max_length=400): return text[:max_length] + "..." if len(text) > max_length else text # 3. 缓存频繁查询的结果 from functools import lru_cache @lru_cache(maxsize=100) def cached_rerank(query, doc_contents_tuple): """缓存重排序结果""" documents = [Document(page_content=content) for content in doc_contents_tuple] return wenmai_reranker.rerank(query, documents)

7.2 处理特殊场景

针对不同场景的特殊处理：

# 处理长文档 def rerank_long_documents(self, query, documents, max_chunk_length=500): """处理超长文档的重排序""" results = [] for doc in documents: if len(doc.page_content) > max_chunk_length: # 分割长文档为多个 chunk chunks = self._split_long_document(doc, max_chunk_length) chunk_results, chunk_scores = self.rerank(query, chunks) # 选择最佳 chunk 或组合结果 best_chunk = chunk_results[0] results.append(Document( page_content=best_chunk.page_content, metadata=doc.metadata )) else: results.append(doc) return self.rerank(query, results) def _split_long_document(self, document, max_length): """分割长文档""" content = document.page_content chunks = [content[i:i+max_length] for i in range(0, len(content), max_length)] return [Document(page_content=chunk, metadata=document.metadata) for chunk in chunks]

8. 总结与下一步建议

通过本文的完整示例，你已经学会了如何将文脉定序集成到LangChain中，构建强大的自动重排序链。关键收获包括：

1. 理解了重排序的价值：文脉定序能够显著提升检索系统的精准度
2. 掌握了集成方法：学会了如何将BGE-Reranker-v2-m3模型与LangChain框架结合
3. 实现了完整流程：从初步检索到重排序的端到端实现
4. 了解了优化策略：性能优化和特殊场景处理的实用技巧

下一步学习建议：

• 尝试在不同的数据集上测试文脉定序的效果
• 探索与其他重排序模型的对比实验
• 考虑将文脉定序集成到你的实际项目中
• 关注模型更新，及时升级到最新版本

文脉定序为信息检索提供了"最后一公里"的精准校准，希望这个完整的代码示例能帮助你在实际项目中实现更智能的检索体验。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。