RAG的核心挑战
RAG（检索增强生成）让LLM能够基于私有知识回答问题。但一个简单的RAG原型——文档切块→向量化→相似度检索→拼接到prompt——在实际使用中往往效果不佳。2026年的RAG优化需要从文档处理、检索质量、上下文管理和生成控制四个维度系统提升。
文档处理优化
智能分块
from langchain.text_splitter import RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter
class SmartChunker:
def __init__(self, chunk_size=512, chunk_overlap=64):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
# 通用分割器
self.general_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""]
)
# Markdown分割器
self.md_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=[
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3"),
]
)
def chunk(self, text, doc_type="auto"):
if doc_type == "auto":
doc_type = self.detect_type(text)
if doc_type == "markdown":
# 先按标题分块，再按大小细分
md_chunks = self.md_splitter.split_text(text)
final_chunks = []
for chunk in md_chunks:
if len(chunk.page_content) > self.chunk_size * 2:
sub_chunks = self.general_splitter.split_text(chunk.page_content)
for sc in sub_chunks:
sc.metadata = chunk.metadata
final_chunks.append(sc)
else:
final_chunks.append(chunk)
return final_chunks
else:
return self.general_splitter.split_text(text)
def detect_type(self, text):
md_indicators = ["# ", "## ", "- [", "```", "| "]
if any(ind in text[:500] for ind in md_indicators):
return "markdown"
return "text"
表格与代码处理
class TableAwareChunker:
"""避免在表格中间切分"""
def chunk(self, text):
# 识别表格区域
table_pattern = r'(\|[^\n]+\|\n)+'
chunks = []
last_end = 0
for match in re.finditer(table_pattern, text):
# 表格前的文本正常切分
pre_text = text[last_end:match.start()]
if pre_text.strip():
chunks.extend(self.split_text(pre_text))
# 表格作为完整块
chunks.append(text[match.start():match.end()])
last_end = match.end()
# 剩余文本
if last_end < len(text):
chunks.extend(self.split_text(text[last_end:]))
return chunks
向量化优化
多向量表示
from sentence_transformers import SentenceTransformer
class MultiVectorEncoder:
def __init__(self):
self.dense_model = SentenceTransformer('BAAI/bge-large-zh-v1.5')
# 稀疏向量用于关键词匹配
self.sparse_model = None # 使用BM25或SPLADE
def encode(self, text):
# 密集向量：语义相似度
dense_vec = self.dense_model.encode(text, normalize_embeddings=True)
# 稀疏向量：精确匹配
sparse_vec = self.sparse_model.encode(text) if self.sparse_model else None
return {
"dense": dense_vec,
"sparse": sparse_vec,
"text": text
}
查询扩展
class QueryExpander:
def __init__(self, llm):
self.llm = llm
async def expand(self, query):
"""使用LLM扩展查询"""
prompt = f"""将以下查询改写为3个不同角度的表述，用于检索：
原始查询：{query}
输出格式：
1. [改写1]
2. [改写2]
3. [改写3]"""
response = await self.llm.ainvoke(prompt)
expansions = self.parse_expansions(response)
expansions.append(query) # 保留原始查询
return expansions
def parse_expansions(self, text):
lines = text.strip().split('\n')
return [line.split('.', 1)[1].strip() for line in lines if '.' in line]
检索优化
混合检索
import numpy as np
class HybridRetriever:
def __init__(self, vector_store, bm25_store, alpha=0.7):
self.vector_store = vector_store # 密集向量检索
self.bm25_store = bm25_store # BM25稀疏检索
self.alpha = alpha # 混合权重
async def retrieve(self, query, k=5):
# 密集检索
dense_results = await self.vector_store.asimilarity_search_with_score(
query, k=k*2
)
# 稀疏检索
sparse_results = self.bm25_store.search(query, k=k*2)
# 分数归一化
dense_scores = self.normalize_scores([s for _, s in dense_results])
sparse_scores = self.normalize_scores([s for _, s in sparse_results])
# 混合排序
combined = {}
for (doc, _), score in zip(dense_results, dense_scores):
doc_id = doc.metadata["id"]
combined[doc_id] = combined.get(doc_id, 0) + self.alpha * score
combined.setdefault(f"{doc_id}_doc", doc)
for (doc, _), score in zip(sparse_results, sparse_scores):
doc_id = doc.metadata["id"]
combined[doc_id] = combined.get(doc_id, 0) + (1 - self.alpha) * score
combined.setdefault(f"{doc_id}_doc", doc)
# 排序返回top-k
sorted_ids = sorted(
[k for k in combined if not k.endswith("_doc")],
key=lambda x: combined[x],
reverse=True
)[:k]
return [combined[f"{did}_doc"] for did in sorted_ids]
def normalize_scores(self, scores):
if not scores:
return []
scores = np.array(scores)
if scores.max() > scores.min():
return (scores - scores.min()) / (scores.max() - scores.min())
return np.ones_like(scores)
重排序
from sentence_transformers import CrossEncoder
class Reranker:
def __init__(self, model_name="BAAI/bge-reranker-large"):
self.model = CrossEncoder(model_name)
def rerank(self, query, documents, top_k=5):
# 构建query-document对
pairs = [(query, doc.page_content) for doc in documents]
# 计算相关性分数
scores = self.model.predict(pairs)
# 排序
ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
return [doc for doc, _ in ranked[:top_k]]
上下文管理
动态上下文窗口
class ContextManager:
def __init__(self, max_tokens=4096):
self.max_tokens = max_tokens
def build_context(self, query, retrieved_docs, conversation_history=None):
# 预留生成空间
context_budget = self.max_tokens - 512
# 分配：历史对话30%，检索文档70%
history_budget = int(context_budget * 0.3)
docs_budget = context_budget - history_budget
# 构建历史上下文
history_text = self.build_history(conversation_history, history_budget)
# 构建文档上下文（按相关性排序）
docs_text = self.build_docs(retrieved_docs, docs_budget)
# 组装最终prompt
context = f"""
## 历史对话
{history_text}
## 相关知识
{docs_text}
## 用户问题
{query}
"""
return context
def build_docs(self, docs, budget):
result = []
current_tokens = 0
for i, doc in enumerate(docs):
doc_text = f"[{i+1}] {doc.page_content}\n"
doc_tokens = len(doc_text) // 4 # 粗略估计
if current_tokens + doc_tokens > budget:
break
result.append(doc_text)
current_tokens += doc_tokens
return "\n".join(result)
引用标注
class CitationGenerator:
def __init__(self, llm):
self.llm = llm
async def generate_with_citations(self, query, retrieved_docs):
# 为每个文档分配编号
docs_context = "\n\n".join([
f"[{i+1}] {doc.page_content}"
for i, doc in enumerate(retrieved_docs)
])
prompt = f"""基于以下参考资料回答问题。在回答中使用 [编号] 标注信息来源。
参考资料：
{docs_context}
问题：{query}
要求：
1. 只使用参考资料中的信息
2. 用 [编号] 标注每条信息的来源
3. 如果资料中没有相关信息，说明"根据现有资料无法回答"
"""
response = await self.llm.ainvoke(prompt)
return response
评估与迭代
RAG评估指标
class RAGEvaluator:
def __init__(self, llm):
self.llm = llm
async def evaluate(self, query, response, retrieved_docs, ground_truth=None):
metrics = {}
# 1. 检索相关性
metrics["retrieval_relevance"] = await self.eval_retrieval(
query, retrieved_docs
)
# 2. 回答忠实度（是否基于检索内容）
metrics["faithfulness"] = await self.eval_faithfulness(
response, retrieved_docs
)
# 3. 回答完整性
if ground_truth:
metrics["completeness"] = await self.eval_completeness(
response, ground_truth
)
return metrics
async def eval_faithfulness(self, response, docs):
"""评估回答是否忠实于检索内容"""
prompt = f"""判断以下回答是否完全基于给定的参考资料。
参考资料：{' '.join(d.page_content[:200] for d in docs)}
回答：{response}
请输出：
1. 忠实/不忠实
2. 不忠实的部分（如有）
"""
result = await self.llm.ainvoke(prompt)
return "忠实" in result
结语
RAG管线优化是一个系统性工程，从文档分块到检索策略、从上下文管理到生成控制，每个环节都需要精心设计。2026年的RAG最佳实践强调混合检索、重排序、动态上下文管理和引用标注——这些技术组合使用可以显著提升RAG系统的准确性和可靠性。
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞，认知差异创造无限可能。