没有捷径：RAG入门不推荐直接使用成熟框架

春节期间我在 Github 开源的 RAG 项目目前已经攒了 134 个 Star，盲猜可能也是因为最开始用的就是 Ollama 本地部署 DeepSeek-r1:7b 的方案，年后当本地部署知识库和 deepseek火了起来之后，被动蹭了一波流量。

1、为什么重复造轮子？

但是，在过去的一个月时间里也收到了很多网友的私信，询问关于为什么市面上已经有了类似 AnythingLLM、Cherry Studio、Dify、RAGFlow 等成熟的开源框架，还要重复造轮子去编一个不是很好用的 RAG 项目。

当然与此同时，也有很多网友在私信或者评论区中反馈上手调试过我这个简单的开源项目后，再去用其他框架更加得心应手了。

1.1 开箱即用的”不友好“

其实我从 24 年 6 月份开始，就逐一深度试用了上述常见的几个开源项目，但一段时间之后明显发现，作为一个刚入门的人来说，虽然 AnyThingLLM、RAGFlow 等成熟框架提供了便捷的"开箱即用"体验，但直接使用这些工具会让人陷入"知其然，而不知其所以然"的困境。

1.2 从零构建再到框架应用

换句话说，就像学编程不应该从框架开始，而是应该从基础语法入手一样。学习 RAG 技术实测也同样适合先构建基础认知框架，再应用封装工具。这不仅是技能学习的路径，更多也是培养解决问题能力的过程。

2、项目特点与局限性

2.1 项目优势

我这个开源项目采用简洁的代码展示了 RAG 的完整流程，通过亲手调试这些组件，可以：

更有助于建立具象认知

比如直观理解文本分块如何影响语义完整性，检索策略如何决定召回质量，以及重排序如何提升最终回答准确度

掌握核心决策点

亲身体验不同 chunk_size 对检索效果的影响，感受不同嵌入模型的语义表达差异，理解为什么相同的 RAG 系统在不同场景下表现迥异

培养调试直觉

当回答质量不理想时，可以通过控制变量法能相对准确判断是分块策略不当，还是检索精度不足，或是提示工程欠缺

掌握这些基础后，再转向 RAGFlow 或其他框架时，众多的配置选项就不再是很抽象的参数，再做 API 调优或者二次开发也会变得有的放矢。从而深度适配业务场景时，也就可以针对性地做调整框架配置，而不是非盲目尝试。

2.2 项目局限性

分块策略

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=20
)
chunks = text_splitter.split_text(text)

使用 LangChain 的 RecursiveCharacterTextSplitter

块大小：每块 200 字符（较小，一个字符相当于 1 个汉字）

重叠度：20 字符

特点：小块大小有利于精准定位信息，但可能损失上下文连贯性

检索策略

results = COLLECTION.query(
    query_embeddings=query_embedding,
    n_results=5,
    include=['documents', 'metadatas']
)

纯语义检索（没有混合 BM25 等关键词检索）

未使用过滤条件或高级查询功能

重排策略

代码中没有显式的重排序步骤，但设计了一些相关逻辑：

# 检测矛盾
conflict_detected = detect_conflicts(sources_for_conflict_detection)


# 获取可信源
if conflict_detected:
    credible_sources = [s for s in sources_for_conflict_detection 
                      if s['type'] == 'web' and evaluate_source_credibility(s) > 0.7]

通过 evaluate_source_credibility 函数对信息源进行可信度评分

在检测到冲突时，优先考虑可信度高的来源

3、进阶版本

前几天开通知识星球后，有些有一定实践经验的网友过来交流一些技术细节时发现，基础版本的开源项目已经不能满足他们当前的需求。

我花了半天时间重构了遍代码，相比上一版多了600行，也是敲到手酸。这个进阶版属于更符合 RAG 系统最佳实践的完整版本，一共有10 个大类的优化要点，下述展示会按照对最终回答质量的影响程度排序。需要说明的是，优先实施以下前 3-5 项将能带来最显著的效果提升。

注：下述代码优化示例是针对前述我提到的自己开源项目而言

3.1 分块策略优化（最高优先级）

200 字符的块过小，无法包含足够上下文

可能导致语义割裂和信息碎片化

改进方案：

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,        # 增大到800-1000字符
    chunk_overlap=150,     # 增加重叠以保持连贯性
    separators=["\n\n", "\n", "。", "，", " "] # 优先在自然段落分割
)

预期效果：更连贯的上下文，减少信息丢失，提高回答质量和相关性。

3.2 混合检索策略

当前问题：

纯语义检索可能忽略关键词匹配

容易出现语义漂移

改进方案：

# 引入BM25关键词检索
from rank_bm25 import BM25Okapi
tokenized_corpus = [doc.split() for doc in corpus]
bm25 = BM25Okapi(tokenized_corpus)


# 混合检索结果
semantic_results = COLLECTION.query(query_embeddings, n_results=7)
bm25_results = bm25.get_top_n(query.split(), corpus, n=7)
combined_results = hybrid_merge(semantic_results, bm25_results)

预期效果：提高检索准确性，尤其对事实性和专业术语的问题。

3.3 重排序机制

当前问题：

缺乏真正的重排序机制

相关度评分单一依赖向量相似度

改进方案：

# 方案A: 使用交叉编码器重排序
from sentence_transformers import CrossEncoder
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
candidate_pairs = [[query, doc] for doc in retrieved_docs]
scores = reranker.predict(candidate_pairs)
reranked_results = [doc for _, doc in sorted(zip(scores, retrieved_docs), reverse=True)]


# 方案B: 使用LLM进行相关性评分
def llm_rerank(query, docs, llm_client):
    results = []
    for doc in docs:
        prompt = f"问题: {query}\n文档: {doc}\n这个文档与问题的相关性是否高? 0-10分打分:"
        score = float(llm_client.generate(prompt).strip())
        results.append((score, doc))
    return [doc for _, doc in sorted(results, reverse=True)]

预期效果：更精准的结果排序，将最相关的内容放在前面，显著提高答案质量。

3.4 递归检索与迭代查询

当前问题：

单次检索可能不足以回答复杂问题

缺乏根据初步检索结果调整查询的机制

改进方案：

def recursive_retrieval(initial_query, max_iteratinotallow=3):
    query = initial_query
    all_contexts = []
    
    for i in range(max_iterations):
        # 当前查询的检索结果
        current_results = retrieve_documents(query)
        all_contexts.extend(current_results)
        
        # 使用LLM分析还需要查询什么
        next_query_prompt = f"""
        基于原始问题: {initial_query}
        以及已检索信息: {summarize(current_results)}
        是否需要进一步查询? 如果需要,请提供新的查询问题:
        """
        
        next_query = llm_client.generate(next_query_prompt)
        if "不需要" in next_query or i == max_iterations-1:
            break
            
        query = next_query
        
    return all_contexts

预期效果：能够处理多跳推理问题，循序渐进获取所需信息。

3.5 上下文压缩与总结

当前问题：

检索文档过长时浪费 token

包含过多无关信息

改进方案：

def compress_context(query, documents, max_tokens=2000):
    compressed_docs = []
    
    for doc in documents:
        # 方案A: 使用map-reduce模式总结
        summary_prompt = f"原文: {doc}\n请提取与问题'{query}'最相关的信息,总结在100词以内:"
        compressed = llm_client.generate(summary_prompt)
        compressed_docs.append(compressed)
        
        # 方案B: 使用抽取式摘要选择关键句子
        # sentences = split_into_sentences(doc)
        # scores = sentence_similarity(query, sentences)
        # top_sentences = [s for s, _ in sorted(zip(sentences, scores), key=lambda x: x[1], reverse=True)[:5]]
        # compressed_docs.append(" ".join(top_sentences))
    
    return compressed_docs

预期效果：减少无关信息干扰，降低 token 消耗，提高模型对关键信息的关注度。

3.6 提示工程优化

当前问题：

现有提示模板相对简单

缺乏针对不同问题类型的专门指导

改进方案：

def create_advanced_prompt(query, context, question_type):
    if question_type == "factual":
        template = """
        你是一个精确的事实回答助手。以下是与问题相关的信息:
        {context}
        
        问题: {query}
        请基于以上信息提供精确的事实回答。如信息不足,请明确指出。请在回答末尾标明信息来源。
        """
    elif question_type == "analytical":
        template = """
        你是一个分析型助手。以下是相关参考信息:
        {context}
        
        问题: {query}
        请分析以上信息,提供深入见解。注意分析信息的一致性与可靠性,标明不同来源间的差异。最后给出综合结论并注明信息来源。
        """
    # 更多问题类型...
    
    # 添加思维链指导
    template += """
    思考步骤:
    1. 提取问题关键点
    2. 识别相关文档中最有价值的信息
    3. 对比不同来源信息
    4. 形成清晰、全面的回答
    """
    
    return template.format(cnotallow=context, query=query)

预期效果：更加结构化和针对性的回答，提高回答质量和可信度。

3.7 元数据增强与过滤

当前问题：

元数据利用不足

缺乏基于元数据的预过滤和后过滤

改进方案：

# 丰富元数据
metadatas = [{
    "source": file_name,
    "doc_id": doc_id,
    "date_processed": datetime.now().isoformat(),
    "chunk_index": i,
    "total_chunks": len(chunks),
    "document_type": detect_document_type(text),
    "language": detect_language(chunk),
    "entities": extract_entities(chunk)
} for i, chunk in enumerate(chunks)]


# 检索时使用元数据过滤
results = COLLECTION.query(
    query_embeddings=query_embedding,
    n_results=10,
    where={"document_type": {"$in": ["report", "research_paper"]}},
    include=['documents', 'metadatas']
)

预期效果：更精准的检索结果筛选，减少噪音，提高答案质量。

3.8 向量化模型升级

当前问题：

all-MiniLM-L6-v2 维度较低(384 维)

中文表示能力有限

改进方案：

# 升级为更强大的双语模型
from sentence_transformers import SentenceTransformer
EMBED_MODEL = SentenceTransformer('BAAI/bge-large-zh-v1.5') # 1024维,中文效果更好


# 或考虑OpenAI嵌入模型
import openai
embeddings = openai.Embedding.create(
    input=chunks, 
    model="text-embedding-ada-002"
)

预期效果：提高语义理解能力和检索精度，尤其是中文文档。

3.9 评估与反馈机制

当前问题：

缺乏回答质量评估

无法迭代改进

改进方案：

def evaluate_response(query, response, retrieved_docs):
    # LLM自评估
    evaluation_prompt = f"""
    问题: {query}
    回答: {response}
    请评估这个回答的质量(1-10分),考虑以下标准:
    1. 准确性: 回答是否符合检索文档的事实?
    2. 完整性: 回答是否全面覆盖问题的各个方面?
    3. 相关性: 回答是否切中问题要点?
    4. 一致性: 回答内部是否存在矛盾?
    请详细说明评分理由。
    """
    
    feedback = llm_client.generate(evaluation_prompt)
    
    # 存储评估结果用于系统优化
    save_evaluation_result(query, response, retrieved_docs, feedback)
    
    return feedback

预期效果：建立持续评估和改进机制，累积数据用于系统优化。

3.10 缓存与预计算策略

当前问题：

重复问题重复计算

实时响应不够快

改进方案：

# 问题-检索结果缓存
import hashlib
import pickle
from functools import lru_cache


@lru_cache(maxsize=100)
def cached_retrieval(query):
    # 计算查询哈希
    query_hash = hashlib.md5(query.encode()).hexdigest()
    cache_file = f"cache/{query_hash}.pkl"
    
    # 检查缓存
    try:
        with open(cache_file, 'rb') as f:
            return pickle.load(f)
    except FileNotFoundError:
        # 执行检索
        results = perform_retrieval(query)
        # 保存到缓存
        os.makedirs("cache", exist_ok=True)
        with open(cache_file, 'wb') as f:
            pickle.dump(results, f)
        return results

4、写在最后

升级版本的项目源码在目前开源版本的基础上，主要完成了前三个方面（分块策略、混合检索策略、重排序机制）的代码升级，源码发布在了知识星球内，供有一定实践经验的盆友测试。剩余其他 7 个维度的示例代码，后续会持续发布，并结合真实案例进行演示。

另外有个彩蛋是，为了让大家更加可视化的了解整个调优过程，我在代码升级的同时做了 UI 优化，相比开源项目新增了一个”分块可视化“的选项卡，其中直观展示了当前代码中的核心模型和方法，以及分块之后的实际内容预览，对调试过程会更加便利些。