普通 RAG 工作逻辑

你提问→直接去向量数据库搜相似度最高 4 篇文档→丢给 DeepSeek 大模型直接回答，一步到位，一锤定音。

分层多级 RAG 核心思路

不用一次选出完美文档，分三道关卡层层过滤，越往后筛选越精细，供给 DeepSeek 的资料质量越高：

• 粗召回：大范围捞鱼，保证不漏关键资料（速度快、成本低）

• 精排：精准打分，筛掉无关垃圾（精度更高）

• 多样性重排：去重复，保留多角度信息（给 DeepSeek 丰富素材）最后再交给 DeepSeek 生成答案

  整体流程：用户提问 → 粗召回 → 精排 → MMR 重排 → DeepSeek 生成回答

1. 粗召回

   向量检索（语义匹配）+ BM25 关键词检索双管齐下向量：看懂句子意思，比如 “RAG 发展趋势” 能匹配 “检索增强生成技术未来方向”；BM25：抓关键词，精准匹配带 RAG、检索增强的文字；两者合并去重，一次性捞出 20~30 份候选，宁可多捞不能漏掉关键内容。

2. Cross-Encoder 精排

   普通向量是分开编码问句和文档再对比相似度；交叉编码器是把你的问题 + 文档拼在一起同时输入打分，判断这段文字到底能不能回答你的问题，准确度高很多。给 30 份候选全部打分，只保留得分最高前 8 篇，直接过滤大半无关内容。

3. MMR 多样性重排

   上一步 8 篇高分文档可能内容高度重合，比如全讲 RAG 向量优化，没有落地、多模态相关内容。MMR 算法平衡两点：文档和问题的相关度 + 文档之间的差异化，剔除重复段落，最后只留 5 篇信息互不重叠、覆盖多维度的优质文档。

4. DeepSeek 生成回答只用最后筛选完的 5 篇高质量资料输入 DeepSeek，模型不会被垃圾文本误导，答案准确率大幅提升。

LangGraph 流水线调度器

把粗召回、精排、重排、DeepSeek 生成封装成独立节点，用图结构串起来，流程清晰，支持自定义分支逻辑：

• 标准流程：全部 4 步走完；
• 智能分支（自适应路由）：简单常识问题直接跳过检索，丢给 DeepSeek 直接回答，省时间

完整简化代码讲解

1. 环境依赖替换

import os
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
# 替换：DeepSeek大模型
from langchain_openai import OpenAIEmbeddings
from langchain_deepseek import ChatDeepSeek
from langchain_community.vectorstores import Chroma
from langchain_community.retrievers import BM25Retriever
from sentence_transformers import CrossEncoder
# 配置DeepSeek密钥
os.environ["DEEPSEEK_API_KEY"] = "你的DeepSeek密钥"
# LangSmith观测（可选，用来调试每一步检索结果）
os.environ["LANGSMITH_TRACING"] = "true"

2. 统一状态结构体

class RAGState(TypedDict):
query: str               # 用户原始问题
candidate_docs: list     # 粗召回捞出的30篇候选
reranked_docs: list     # Cross精排后的8篇
final_docs: list         # MMR去重后的5篇最终文档
answer: str              # DeepSeek输出的最终回答

3. 阶段 1：粗召回（向量 + BM25 混合检索）

def stage1_recall(state: RAGState) -> RAGState:
query = state["query"]
# 1. 向量语义检索，捞15篇
vector_docs = vector_store.similarity_search(query, k=15)
vector_texts = [d.page_content for d in vector_docs]
# 2. BM25关键词检索，捞15篇
bm25_docs = bm25_retriever.invoke(query)
bm25_texts = [d.page_content for d in bm25_docs][:15]
# 合并两段结果，简单哈希去重，最多保留30篇
seen = set()
candidates = []
for text in vector_texts + bm25_texts:
text_hash = hash(text[:100])
if text_hash not in seen:
seen.add(text_hash)
candidates.append(text)
return {"candidate_docs": candidates[:30]}

4. 阶段 2：Cross-Encoder 精排打分

# 轻量高效重排模型，不用大模型，成本低
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")
def stage2_rerank(state: RAGState) -> RAGState:
query = state["query"]
candidates = state["candidate_docs"]
if not candidates:
return {"reranked_docs": []}
# 拼接【问题+单篇文档】送入模型打分
query_doc_pairs = [[query, doc] for doc in candidates]
scores = reranker.predict(query_doc_pairs)
# 按分数从高到低排序，取前8
sorted_docs = sorted(zip(candidates, scores), key=lambda x:x[1], reverse=True)
top8_docs = [doc for doc, score in sorted_docs[:8]]
return {"reranked_docs": top8_docs}

5. 阶段 3：MMR 多样性重排，剔除重复信息

def stage3_diversity(state: RAGState) -> RAGState:
raw_docs = state["reranked_docs"]
# MMR重排：lambda_mult=0.7，七分看重相关性，三分看重多样性
mmr_result = vector_store.max_marginal_relevance_search(
state["query"],
k=5,
fetch_k=8,
lambda_mult=0.7
)
final_texts = [d.page_content for d in mmr_result]
# 过滤太短无效片段，最终保留5篇
final_texts = [t for t in final_texts if len(t) > 100][:5]
return {"final_docs": final_texts}

6. 阶段 4：大模型 生成答案

def stage4_generate(state: RAGState) -> RAGState:
# 拼接所有参考文档
context = "\n=====分割线=====\n".join(
f"参考文档{i+1}：{doc[:1500]}"
for i, doc in enumerate(state["final_docs"])
)
# 提示词：约束DeepSeek只能基于文档作答，无资料如实说明
prompt = f"""
请严格根据下方参考文档回答用户问题，如果文档没有对应信息，直接说明无法找到相关内容，禁止编造信息。
参考文档：
{context}
用户问题：{state["query"]}
你的回答：
"""
# 调用DeepSeek模型
llm = ChatDeepSeek(model="deepseek-chat", temperature=0)
res = llm.invoke(prompt)
return {"answer": res.content}

7. LangGraph 搭建流水线，串联全部节点

# 初始化图结构
graph_builder = StateGraph(RAGState)
# 注册4个执行节点
graph_builder.add_node("粗召回", stage1_recall)
graph_builder.add_node("精排打分", stage2_rerank)
graph_builder.add_node("多样性重排", stage3_diversity)
graph_builder.add_node("DeepSeek生成", stage4_generate)
# 固定执行顺序
graph_builder.add_edge(START, "粗召回")
graph_builder.add_edge("粗召回", "精排打分")
graph_builder.add_edge("精排打分", "多样性重排")
graph_builder.add_edge("多样性重排", "DeepSeek生成")
graph_builder.add_edge("DeepSeek生成", END)
# 编译工作流
rag_workflow = graph_builder.compile()
# 执行问答
result = rag_workflow.invoke({"query": "2026年RAG技术的发展趋势"})
print("DeepSeek回答：", result["answer"])

8. 智能路由，简单问题跳过检索

def judge_need_retrieval(state: RAGState):
classify_llm = ChatDeepSeek(model="deepseek-chat", temperature=0)
res = classify_llm.invoke(f"""
判断用户问题是否需要检索外部知识库：
1. 需要查资料（行业、技术、数据、专业知识）→ 返回 retrieve
2. 基础常识、简单计算、无需外部资料 → 返回 direct
仅输出关键词，不要多余文字
问题：{state["query"]}
""")
return res.content.strip()
# 新增判断节点+条件分支
graph_builder.add_node("问题分类", judge_need_retrieval)
graph_builder.add_edge(START, "问题分类")
# 分支逻辑：需要检索走完整流程，不需要直接调用DeepSeek生成
graph_builder.add_conditional_edges(
"问题分类",
judge_need_retrieval,
{
"retrieve": "粗召回",
"direct": "DeepSeek生成"
}
)

完整代码

pip install langgraph langchain-deepseek langchain-community langchain-openai chromadb sentence-transformers

import os
import hashlib
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_deepseek import ChatDeepSeek
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.retrievers import BM25Retriever
from langchain.text_splitter import RecursiveCharacterTextSplitter
from sentence_transformers import CrossEncoder
# ====================== 1. 全局配置 ======================
# DeepSeek密钥
os.environ["DEEPSEEK_API_KEY"] = "sa-xxxxxxxxxxxxxxxxxxx"
# LangSmith 链路追踪（可选，调试用）
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "LangSmith密钥"
# 初始化Embedding、重排模型
embedding = OpenAIEmbeddings()
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")
# 全局缓存类（高频查询加速）
class RAGCache:
def __init__(self):
self._cache = {}
def get(self, query: str):
key = hashlib.md5(query.encode("utf-8")).hexdigest()
return self._cache.get(key)
def set(self, query: str, docs: list):
key = hashlib.md5(query.encode("utf-8")).hexdigest()
self._cache[key] = docs
rag_cache = RAGCache()
# ====================== 2. 定义图状态 ======================
class RAGState(TypedDict):
query: str                  # 用户原始问题
candidate_docs: list        # 粗召回结果
reranked_docs: list         # CrossEncoder精排结果
final_docs: list            # MMR多样性重排后最终文档
answer: str                 # DeepSeek输出答案
# ====================== 3. 模拟知识库构建（可替换为自己文档） ======================
def build_knowledge_base():
# 模拟业务文档，实际替换为你的文件读取逻辑
raw_texts = [
"2026年RAG技术发展趋势：多模态RAG成为主流，支持图文音视频联合检索；轻量化本地RAG普及，终端设备可离线部署；检索与Agent深度融合，实现自主规划检索步骤。",
"RAG分层检索优势：粗召回+精排+MMR重排三段式流水线，相比单次向量检索准确率提升15%以上，解决文档重复、低质量素材干扰大模型问题。",
"DeepSeek大模型适配RAG最佳实践：temperature设为0，严格限制模型仅参考提供文档，禁止编造未知信息；长文档分段截取1500字以内避免上下文溢出。",
"MMR算法参数说明：lambda_mult=0.7平衡相关性与多样性，数值越大越看重匹配度，越小越看重内容差异化，通用场景推荐0.6~0.8。",
"CrossEncoder重排原理：将query与文档拼接输入模型，相比双塔向量相似度打分精度更高，轻量BGE重排模型算力消耗远小于大模型。"
]
# 文本切分
splitter = RecursiveCharacterTextSplitter(chunk_size=600, chunk_overlap=100)
split_docs = splitter.create_documents(raw_texts)
# 向量库
vector_store = Chroma.from_documents(split_docs, embedding, persist_directory="./chroma_db")
vector_store.persist()
# BM25检索器
bm25_retriever = BM25Retriever.from_documents(split_docs)
return vector_store, bm25_retriever
# 初始化向量库与BM25（全局变量供各节点调用）
vector_store, bm25_retriever = build_knowledge_base()
# ====================== 4. LangGraph 各节点函数 ======================
# 节点1：粗召回（向量+BM25混合检索）
def stage1_recall(state: RAGState) -> RAGState:
query = state["query"]
# 先查缓存
cache_res = rag_cache.get(query)
if cache_res:
return {"candidate_docs": cache_res}
# 向量检索15条
vec_docs = vector_store.similarity_search(query, k=15)
vec_texts = [d.page_content for d in vec_docs]
# BM25关键词检索15条
bm25_docs = bm25_retriever.invoke(query)
bm25_texts = [d.page_content for d in bm25_docs][:15]
# 合并去重
seen_hash = set()
candidates = []
for text in vec_texts + bm25_texts:
h = hash(text[:100])
if h not in seen_hash:
seen_hash.add(h)
candidates.append(text)
# 最多保留30条候选
final_candidates = candidates[:30]
# 写入缓存
rag_cache.set(query, final_candidates)
return {"candidate_docs": final_candidates}
# 节点2：CrossEncoder精排打分
def stage2_rerank(state: RAGState) -> RAGState:
candidates = state["candidate_docs"]
if not candidates:
return {"reranked_docs": []}
query = state["query"]
# 构造query-doc对
pairs = [[query, doc] for doc in candidates]
scores = reranker.predict(pairs, show_progress_bar=False)
# 按分数降序，取前8
sorted_items = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
top8 = [doc for doc, score in sorted_items[:8]]
return {"reranked_docs": top8}
# 节点3：MMR多样性重排，去冗余
def stage3_diversity(state: RAGState) -> RAGState:
query = state["query"]
# MMR检索，从8条里选出5条高相关且不重复
mmr_res = vector_store.max_marginal_relevance_search(
query=query,
k=5,
fetch_k=8,
lambda_mult=0.7
)
text_list = [doc.page_content for doc in mmr_res]
# 过滤过短无效文本
final_texts = [t for t in text_list if len(t) > 100][:5]
return {"final_docs": final_texts}
# 节点4：DeepSeek生成回答
def stage4_generate(state: RAGState) -> RAGState:
docs = state["final_docs"]
query = state["query"]
# 拼接上下文
context_parts = []
for idx, text in enumerate(docs):
context_parts.append(f"【参考文档{idx+1}】\n{text[:1500]}")
context = "\n\n======分隔线======\n\n".join(context_parts)
prompt = f"""
你是专业技术问答助手，请严格依据下方参考文档回答用户问题：
1. 文档没有相关信息时，直接回复：暂无相关资料，无法解答该问题；
2. 禁止编造、脑补文档不存在的内容；
3. 回答条理清晰，分点说明。
参考文档：
{context}
用户问题：{query}
回答：
"""
llm = ChatDeepSeek(model="deepseek-chat", temperature=0)
resp = llm.invoke(prompt)
return {"answer": resp.content}
# 分支判断：是否需要检索知识库
def judge_retrieval_branch(state: RAGState) -> str:
query = state["query"]
classifier = ChatDeepSeek(model="deepseek-chat", temperature=0)
prompt = f"""
判断用户问题是否需要检索外部知识库，仅输出单词：
需要查资料 → retrieve
常识/简单计算无需资料 → direct
只返回一个单词，不要多余文字
问题：{query}
"""
res = classifier.invoke(prompt)
return res.content.strip().lower()
# 无检索直接回答节点
def direct_generate(state: RAGState) -> RAGState:
query = state["query"]
llm = ChatDeepSeek(model="deepseek-chat", temperature=0)
resp = llm.invoke(f"直接回答问题：{query}")
return {"answer": resp.content, "final_docs": []}
# ====================== 5. 组装LangGraph工作流 ======================
def build_rag_graph():
builder = StateGraph(RAGState)
# 注册所有节点
builder.add_node("classifier", judge_retrieval_branch)
builder.add_node("recall", stage1_recall)
builder.add_node("rerank", stage2_rerank)
builder.add_node("diversity", stage3_diversity)
builder.add_node("generate", stage4_generate)
builder.add_node("direct_gen", direct_generate)
# 入口
builder.add_edge(START, "classifier")
# 条件分支
builder.add_conditional_edges(
source="classifier",
path=judge_retrieval_branch,
path_map={
"retrieve": "recall",
"direct": "direct_gen"
}
)
# 检索完整流水线链路
builder.add_edge("recall", "rerank")
builder.add_edge("rerank", "diversity")
builder.add_edge("diversity", "generate")
# 两个出口统一到END
builder.add_edge("generate", END)
builder.add_edge("direct_gen", END)
graph = builder.compile()
return graph
# ====================== 6. 执行测试 ======================
if __name__ == "__main__":
rag_graph = build_rag_graph()
# 测试1：需要检索知识库的专业问题
print("=====测试1：专业技术问题=====")
result1 = rag_graph.invoke({"query": "2026年RAG技术有哪些发展趋势？分层检索有什么好处？"})
print(result1["answer"])
print("\n=====测试2：无需检索的简单常识=====")
# 测试2：简单常识，直接跳过检索
result2 = rag_graph.invoke({"query": "100加200等于多少？"})
print(result2["answer"])

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】