普通 RAG 工作逻辑你提问→直接去向量数据库搜相似度最高 4 篇文档→丢给 DeepSeek 大模型直接回答一步到位一锤定音。分层多级 RAG 核心思路不用一次选出完美文档分三道关卡层层过滤越往后筛选越精细供给 DeepSeek 的资料质量越高粗召回大范围捞鱼保证不漏关键资料速度快、成本低精排精准打分筛掉无关垃圾精度更高多样性重排去重复保留多角度信息给 DeepSeek 丰富素材最后再交给 DeepSeek 生成答案整体流程用户提问 → 粗召回 → 精排 → MMR 重排 → DeepSeek 生成回答粗召回向量检索语义匹配 BM25 关键词检索双管齐下向量看懂句子意思比如 “RAG 发展趋势” 能匹配 “检索增强生成技术未来方向”BM25抓关键词精准匹配带 RAG、检索增强的文字两者合并去重一次性捞出 20~30 份候选宁可多捞不能漏掉关键内容。Cross-Encoder 精排普通向量是分开编码问句和文档再对比相似度交叉编码器是把你的问题 文档拼在一起同时输入打分判断这段文字到底能不能回答你的问题准确度高很多。给 30 份候选全部打分只保留得分最高前 8 篇直接过滤大半无关内容。MMR 多样性重排上一步 8 篇高分文档可能内容高度重合比如全讲 RAG 向量优化没有落地、多模态相关内容。MMR 算法平衡两点文档和问题的相关度 文档之间的差异化剔除重复段落最后只留 5 篇信息互不重叠、覆盖多维度的优质文档。DeepSeek 生成回答只用最后筛选完的 5 篇高质量资料输入 DeepSeek模型不会被垃圾文本误导答案准确率大幅提升。LangGraph流水线调度器把粗召回、精排、重排、DeepSeek 生成封装成独立节点用图结构串起来流程清晰支持自定义分支逻辑标准流程全部 4 步走完智能分支自适应路由简单常识问题直接跳过检索丢给 DeepSeek 直接回答省时间完整简化代码讲解环境依赖替换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统一状态结构体class RAGState(TypedDict): query: str # 用户原始问题 candidate_docs: list # 粗召回捞出的30篇候选 reranked_docs: list # Cross精排后的8篇 final_docs: list # MMR去重后的5篇最终文档 answer: str # DeepSeek输出的最终回答阶段 1粗召回向量 BM25 混合检索def stage1_recall(state: RAGState) - RAGState: query state[query] # 1. 向量语义检索捞15篇 vector_docs vector_store.similarity_search(query, k15) 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]}阶段 2Cross-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), keylambda x:x[1], reverseTrue) top8_docs [doc for doc, score in sorted_docs[:8]] return {reranked_docs: top8_docs}阶段 3MMR 多样性重排剔除重复信息def stage3_diversity(state: RAGState) - RAGState: raw_docs state[reranked_docs] # MMR重排lambda_mult0.7七分看重相关性三分看重多样性 mmr_result vector_store.max_marginal_relevance_search( state[query], k5, fetch_k8, lambda_mult0.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}阶段 4大模型 生成答案def stage4_generate(state: RAGState) - RAGState: # 拼接所有参考文档 context \n分割线\n.join( f参考文档{i1}{doc[:1500]} for i, doc in enumerate(state[final_docs]) ) # 提示词约束DeepSeek只能基于文档作答无资料如实说明 prompt f 请严格根据下方参考文档回答用户问题如果文档没有对应信息直接说明无法找到相关内容禁止编造信息。 参考文档 {context} 用户问题{state[query]} 你的回答 # 调用DeepSeek模型 llm ChatDeepSeek(modeldeepseek-chat, temperature0) res llm.invoke(prompt) return {answer: res.content}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])智能路由简单问题跳过检索def judge_need_retrieval(state: RAGState): classify_llm ChatDeepSeek(modeldeepseek-chat, temperature0) 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-transformersimport 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_mult0.7平衡相关性与多样性数值越大越看重匹配度越小越看重内容差异化通用场景推荐0.6~0.8。, CrossEncoder重排原理将query与文档拼接输入模型相比双塔向量相似度打分精度更高轻量BGE重排模型算力消耗远小于大模型。 ] # 文本切分 splitter RecursiveCharacterTextSplitter(chunk_size600, chunk_overlap100) 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, k15) 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} # 节点2CrossEncoder精排打分 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_barFalse) # 按分数降序取前8 sorted_items sorted(zip(candidates, scores), keylambda x: x[1], reverseTrue) top8 [doc for doc, score in sorted_items[:8]] return {reranked_docs: top8} # 节点3MMR多样性重排去冗余 def stage3_diversity(state: RAGState) - RAGState: query state[query] # MMR检索从8条里选出5条高相关且不重复 mmr_res vector_store.max_marginal_relevance_search( queryquery, k5, fetch_k8, lambda_mult0.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} # 节点4DeepSeek生成回答 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【参考文档{idx1}】\n{text[:1500]}) context \n\n分隔线\n\n.join(context_parts) prompt f 你是专业技术问答助手请严格依据下方参考文档回答用户问题 1. 文档没有相关信息时直接回复暂无相关资料无法解答该问题 2. 禁止编造、脑补文档不存在的内容 3. 回答条理清晰分点说明。 参考文档 {context} 用户问题{query} 回答 llm ChatDeepSeek(modeldeepseek-chat, temperature0) resp llm.invoke(prompt) return {answer: resp.content} # 分支判断是否需要检索知识库 def judge_retrieval_branch(state: RAGState) - str: query state[query] classifier ChatDeepSeek(modeldeepseek-chat, temperature0) 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(modeldeepseek-chat, temperature0) 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( sourceclassifier, pathjudge_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%免费】