企业官网建设流程全解析

普通 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%免费】