网站里做任务wordpress 爱情

网站里做任务,wordpress 爱情,公司变更地址需要多少钱,好听的网络公司名称本文系统介绍RAG高级技术体系#xff0c;涵盖Query改写、联网搜索、混合检索与Rerank重排序等核心技术#xff0c;解决AI回答不准确、无法处理实时问题等痛点。通过知识库问题生成、对话知识沉淀、健康检查与版本管理等机制#xff0c;实现知识库自我进化。以迪士尼客服为例…本文系统介绍RAG高级技术体系涵盖Query改写、联网搜索、混合检索与Rerank重排序等核心技术解决AI回答不准确、无法处理实时问题等痛点。通过知识库问题生成、对话知识沉淀、健康检查与版本管理等机制实现知识库自我进化。以迪士尼客服为例提供完整代码实现帮助开发者构建企业级RAG系统将检索准确率提升90%。前排提示文末有大模型AGI-CSDN独家资料包哦本篇公众号更适合电脑来看下面附带了很多代码。手机看可能不是很方便阅读。前言为什么要学RAG高阶技术 你是否遇到过这些问题❌AI回答不准确“迪士尼门票大概300-500元”❌无法回答实时问题“今天开放吗”AI不知道因为知识库是静态的无法理解今天是什么 ❌无法理解上下文“还有其他项目吗”AI不知道”其他”指什么❌检索结果不精准问门票价格却返回交通信息❌知识库难维护手动更新容易出错传统的AI系统往往无法获取最新信息只能依赖训练数据中的静态知识这显然无法满足实际业务需求。学习RAG高级技术后你可以✅信息准确基于最新知识库准确率90%✅实时查询智能判断何时需要联网获取实时信息✅理解上下文5种Query改写处理复杂对话✅检索精准混合检索Rerank召回率提升35%✅自动维护对话沉淀健康检查维护成本降低70%摘要本文档系统性地阐述了检索增强生成RAG从入门基础到高级应用的完整技术体系。文章首先对比了入门篇RAG与高级篇RAG在架构设计、功能模块和性能表现上的核心差异揭示了企业级应用对技术深度的必然要求。随后深入剖析了RAG的核心概念即索引构建、检索和生成三大基础步骤。在实战篇章中本文以ChatPDF为例展示了基础RAG的构建流程及其局限性。在此基础上重点转向高级RAG的核心技术栈。首先详细探讨了五种关键的Query改写技术包括上下文依赖型、对比型、模糊指代型、多意图型和反问型改写这些是解决查询理解难题的基石。接着介绍了Query联网搜索模块用于解决静态知识库无法处理实时信息的问题。在召回优化方面本文深入分析了混合检索结合BM25与向量检索与Rerank重排序技术如何协同工作以大幅提升检索的精准度。文章进一步探讨了知识库的全生命周期管理涵盖了利用LLM进行知识库问题生成以优化索引、通过对话知识沉淀实现知识库的自我完善、定期的知识库健康度检查以及确保系统稳定性的知识库版本管理。在高级增强部分本文介绍了GraphRAG图谱增强技术用以克服传统RAG在多跳推理和全局性总结问题上的局限性。最后文章提供了完整的系统集成架构、循序渐进的学习路径、常见问题FAQ以及面向生产环境的部署指南。第一章入门篇VS高级篇技术对比1.1 技术演进全景图RAG技术的发展呈现出两个明显的技术阶段入门篇与高级篇。入门篇RAG通常作为概念验证或小型项目其架构相对简单流程高度线性。它始于单一的数据源如PDF文档通过设定固定的chunk_size进行粗略的文本分割接着使用单一的嵌入模型进行向量化并存储在如FAISS这样的本地向量数据库中。在查询阶段它直接使用用户的原始问题进行简单的向量相似度检索例如Top-3然后将检索到的文本块直接拼接后交给大模型生成答案。入门篇的优势在于实现简单、快速上手但其功能局限性非常明显难以应对真实世界中复杂的查询场景和多变的知识形态。相比之下高级篇RAG是面向企业级生产环境的系统工程它在入门篇的每个环节上都进行了深度的优化和扩展。在数据源层面高级篇RAG需要处理来自PDF、网页、数据库等异构的多源数据并进行智能化的预处理、清洗与去重。在查询处理上它不再直接使用原始问题而是引入了复杂的Query改写引擎该引擎包含多种改写策略并能智能判断是否需要联网搜索。在检索策略上高级篇摒弃了单一的向量检索转而采用BM25关键词检索与向量语义检索相结合的混合检索策略并通过Rerank重排序模型进一步提升检索结果的精准度。更关键的区别在于系统的动态维护能力。入门篇RAG的知识库是静态的依赖人工手动更新而高级篇RAG具备完善的知识库自动维护机制包括从对话中自动沉淀新知识、定期的知识库健康度检查以及严格的版本管理。此外高级篇还引入了GraphRAG等技术来处理入门篇无法解决的复杂推理问题。这种从简单功能堆叠到模块化、可维护、高鲁棒性的企业级架构的演进是RAG技术从“玩具”走向“工具”的必然路径。【以下是Mermaid的代码公众号采集并上传后只有代码没有效果这个同时也不方便转化为图片所以这一个流程图如果大家感兴趣可以去Mermaid的官网粘贴过去看看后面的流程图我都会用截图图片来展示。有我微信的uu们可以直接找我要pdf文档~】graph TB subgraph 入门篇[ 入门篇基础RAGRAG入门篇] A1[PDF文档] -- A2[文本分割br/chunk_size1000] A2 -- A3[向量化br/text-embedding] A3 -- A4[FAISS存储] A4 -- A5[简单检索br/Top-3] A5 -- A6[直接生成答案] style A1 fill:#e3f2fd style A6 fill:#e3f2fd end subgraph 高级篇[ 高级篇企业级RAG本教程] B1[多源数据br/PDF/网页/DB] -- B2[智能预处理br/清洗去重] B2 -- B3[Query改写引擎br/5种类型自动识别] B3 -- B4[联网搜索判断br/8种场景] B4 -- B5[混合检索br/BM25Vector] B5 -- B6[Rerank重排序br/BGE/Cohere] B6 -- B7[知识库维护br/沉淀健康检查] B7 -- B8[GraphRAG增强br/图谱推理] B8 -- B9[高质量生成br/来源置信度] style B1 fill:#fff3e0 style B9 fill:#fff3e0 end A6 -.技术升级.- B1 Note1[入门篇特点:br/✅ 简单易学br/✅ 快速上手br/❌ 功能有限] -- 入门篇 Note2[高级篇特点:br/✅ 企业级br/✅ 功能完整br/✅ 生产可用] -- 高级篇1.2 适用场景对比第二章RAG核心概念2.1 什么是RAG大白话解释RAG Retrieval检索 Augmented增强 Generation生成【具体内容可以看上一篇RAG技术深度解析与实战入门篇这一章只有相关代码没有解释。】场景模拟迪士尼客服❌ 没有RAG的AI传统方式游客: 迪士尼门票多少钱AI靠记忆回答: 根据我的知识迪士尼门票大概300-500元问题1. ❌ 信息可能过时训练数据是去年的2. ❌ 不够具体哪个迪士尼什么日期3. ❌ 容易瞎编AI幻觉问题4. ❌ 无法验证没有信息来源5. ❌ 难以更新需要重新训练模型成本极高✅ 有RAG的AI智能客服游客: 迪士尼门票多少钱【系统内部流程】步骤1 - 检索Retrieval: 在知识库搜索迪士尼门票价格 找到文档上海迪士尼乐园门票价格2024-11-01更新 平日成人票399元周末499元儿童票299-374元步骤2 - 增强Augmented: 将检索到的文档作为参考资料 提取关键信息价格、日期类型、儿童政策步骤3 - 生成Generation: AI基于参考文档生成答案AI: 您好根据最新的价格信息 成人票价格 • 平日399元 • 周末/节假日499元 儿童票价格身高1.0-1.4米 • 平日299元 • 周末/节假日374元 免费政策 • 1.0米以下儿童免费入园 温馨提示 建议提前3-7天在官方渠道预订高峰期可能售罄哦 信息来源官方价格文档更新于2024-11-01优势✅ 信息准确基于最新知识库✅ 有据可查标注来源和更新时间✅ 不会瞎编有什么说什么✅ 易于更新只需更新知识库文档✅ 可追溯可以查看原始文档2.2 RAG三大步骤详解步骤1Indexing索引构建- 只需做一次步骤2Retrieval检索- 每次查询都执行步骤3Generation生成答案- 每次查询都执行代码示例from langchain.chains.question_answering import load_qa_chainfrom langchain_community.llms import Tongyi# 初始化大模型llm Tongyi( model_namedeepseek-v3, dashscope_api_keyAPI_KEY)# 加载问答链chain load_qa_chain(llm, chain_typestuff)# chain_typestuff: 把所有文档一起给模型处理# 准备输入input_data {input_documents: docs, # 检索到的3个文档 question: user_query # 用户问题}# 生成答案response chain.invoke(input_data)answer response[output_text]print(answer)第三章ChatPDF入门实战3.1 完整系统架构3.2代码实现chatpdf_basic.py核心代码片段# # ChatPDF核心流程# defmain():主流程从PDF到问答# 第1步读取PDF print( 第1步读取PDF文件) pdf_reader PdfReader(迪士尼介绍.pdf) print(f✅ PDF共 {len(pdf_reader.pages)} 页)# 第2步提取文本和页码 print(\n 第2步提取文本内容) text, page_numbers extract_text_with_page_numbers(pdf_reader) print(f✅ 提取 {len(text)} 个字符)# 第3步处理文本分割向量化存储 print(\n 第3步建立向量索引) knowledgeBase process_text_with_splitter( text, page_numbers, save_path./vector_db )# 第4步初始化大模型 print(\n 第4步初始化大语言模型) llm Tongyi( model_namedeepseek-v3, dashscope_api_keyAPI_KEY )# 第5步问答测试 print(\n 第5步测试问答功能) questions [迪士尼在哪里,门票多少钱,有哪些热门项目 ]for question in questions: answer, sources ask_question( knowledgeBase, question, llm, k3 )第四章Query改写技术4.1 为什么Query改写是RAG的核心想象一下这个场景游客问迪士尼客服还有其他项目吗这是一个再正常不过的问题。但传统的RAG系统却可能完全无法理解——其他指的是什么项目又是什么项目这就是Query改写技术要解决的核心问题搭建用户自然语言与知识库规范文本之间的翻译桥梁。Query改写技术是高级RAG系统的智能翻译官专门解决用户口语化、模糊的查询表达与知识库中规范化、陈述性文本之间的语义鸿沟。Query改写的核心工作就是当用户说“黑话”、说“半截话”、或者“话里有话”时把这个“人话”甚至“黑话”翻译成知识库能听懂的“官话”Prompt大大提高检索的命中率。现实中的理解困境在日常对话中我们会说“那个项目刺激吗”指代不明“还有其他好玩的吗”缺少上下文“创极速光轮和矿山车哪个更好玩”对比模糊“门票多少钱需要预约吗停车方便吗”多意图混合而在知识库中信息是这样组织的创极速光轮明日世界园区过山车身高要求122cm刺激程度五星 七个小矮人矿山车梦幻世界园区家庭过山车身高要求97cm刺激程度三星Query改写的核心工作就是把用户的人话甚至黑话翻译成知识库能听懂的官话。对于企业级RAG系统来说Query改写不是锦上添花而是雪中送炭。它直接决定了系统能否理解真实用户的真实问题是整个RAG流程的第一道关卡。4.3 类型1上下文依赖型 —— “阅读理解式翻译”# # 上下文依赖型Query改写器# classContextDependentQueryRewriter:上下文依赖型Query改写器def__init__(self, modelqwen-turbo-latest): self.model modeldefrewrite(self, current_query, conversation_history):将依赖上下文的查询改写为独立完整的查询 instruction 你是一个智能的查询优化助手。 【任务】:分析用户的当前问题是否依赖于对话历史如果依赖则补全信息。 【识别标志】: - 还有、其他、更多需要知道相对于什么 - 也、另外需要补充主体信息 - 问题很短但需要上下文才能理解 【改写步骤】: 1. 从对话历史中提取主题、对象、地点、已提到的内容 2. 将这些信息补充到当前问题中 3. 确保改写后的问题完全独立不需要任何上下文即可理解 prompt f ## 指令 ##{instruction} ## 对话历史 ##{conversation_history} ## 当前问题 ##{current_query} ## 改写后的问题 ## response get_completion(prompt, self.model)return response.strip()测试效果对比# 测试代码rewriter ContextDependentQueryRewriter()test_cases [ { history: 用户: 疯狂动物城有什么好玩的AI: 有警察局互动、训练营、冰淇淋店, query: 还有其他设施吗 }, { history: 用户: 门票多少钱AI: 平日399元周末499元, query: 儿童票呢 }]for test in test_cases: print(f原始: {test[query]}) rewritten rewriter.rewrite(test[query], test[history]) print(f改写: {rewritten}) print(f效果: ✅ 补全了上下文变成独立完整的问题\n)# 输出# 原始: 还有其他设施吗# 改写: 除了警察局互动体验、朱迪警官训练营和尼克狐的冰淇淋店疯狂动物城园区还有其他设施吗# 效果: ✅ 补全了上下文变成独立完整的问题# 原始: 儿童票呢# 改写: 上海迪士尼乐园的儿童票价格是多少# 效果: ✅ 补全了上下文变成独立完整的问题4.4 类型2对比型Query改写业务场景常见对比型Query的改写策略 —— “破译谜语式翻译”# # 对比型Query改写器# classComparisonQueryRewriter: 对比型Query改写器 核心功能: 1. 识别对比词哪个、更、比较 2. 明确对比对象从Query或历史中提取 3. 细化对比维度补充具体的评估标准def__init__(self, modelqwen-turbo-latest): self.model modeldefrewrite(self, query, context_info): 改写对比型Query 改写逻辑: 1. 保留对比对象如果明确 2. 补充对比对象如果在上下文中 3. 细化对比维度添加具体评估标准 4. 添加场景信息地点、用途等 参数: query: 原始查询 context_info: 上下文信息 返回: 改写后的对比性查询 instruction 你是一个查询分析专家专门处理对比型问题。【任务】:分析用户的对比型问题明确对比对象和对比维度。【识别标志】:- 对比词: 哪个、更、比较、还是、vs- 对比结构: A和B哪个...、A比B...、选A还是B【改写步骤】:步骤1 - 明确对比对象: • 如果Query中已有对比对象 → 保留 • 如果对比对象在上下文中 → 提取并补充 • 如果对比对象不明确 → 保持原样或标注需要补充步骤2 - 细化对比维度: • 适合小孩 → 刺激程度、身高限制、互动性、安全性 • 好玩 → 项目丰富度、体验时长、独特性、用户评价 • 方便 → 交通便利、时间成本、经济成本 • 值得 → 性价比、体验质量、独特性步骤3 - 补充完整信息: • 添加地点: 上海迪士尼乐园的... • 添加用途: 带小孩游玩、快速到达等 • 添加场景: 具体的使用情境【示例1】:原始: 疯狂动物城和宝藏湾哪个更适合小孩上下文: 用户计划带5岁孩子游玩改写: 比较上海迪士尼乐园的疯狂动物城园区和宝藏湾园区哪个更适合带5岁小孩游玩 请从以下方面对比分析 1. 游乐项目的刺激程度是否有过山车等刺激项目 2. 身高和年龄限制孩子是否能玩所有项目 3. 互动性和趣味性是否有适合儿童的互动体验 4. 安全性和家长陪同是否需要成人陪同【示例2】:原始: 地铁和打车哪个更方便上下文: 用户从浦东机场去迪士尼改写: 从浦东机场到上海迪士尼乐园乘坐地铁和打车哪个更方便 请从以下方面对比 1. 交通便利性是否需要换乘、是否有直达 2. 时间成本预计所需时间、是否会堵车 3. 经济成本票价/车费对比 4. 舒适度是否拥挤、是否方便携带行李 prompt f### 指令 ### {instruction}### 上下文信息 ### {context_info}### 原始查询 ### {query}### 改写后的查询 ### response get_completion(prompt, self.model)return response.strip()对比型改写的3个层次测试效果对比# # 对比型Query改写测试# rewriter ComparisonQueryRewriter()# 测试案例1: 园区对比test_case_1 {query: 疯狂动物城和宝藏湾哪个更适合小孩,context: 用户带5岁孩子第一次来迪士尼}result_1 rewriter.rewrite(test_case_1[query], test_case_1[context])print(原始Query:, test_case_1[query])print(\n改写后:)print(result_1)print(\n改进点:)print(✅ 补充了年龄信息5岁)print(✅ 细化了4个对比维度)print(✅ 添加了具体的评估标准)# 输出示例:原始Query: 疯狂动物城和宝藏湾哪个更适合小孩改写后:比较上海迪士尼乐园的疯狂动物城园区和宝藏湾园区哪个更适合带5岁小孩游玩请从以下方面进行详细对比【刺激程度对比】:- 疯狂动物城: 是否有刺激的过山车类项目- 宝藏湾: 海盗船等项目的刺激程度如何【身高年龄限制】:- 疯狂动物城: 各项目的具体身高要求- 宝藏湾: 加勒比海盗等项目的身高限制【互动性与趣味性】:- 疯狂动物城: 是否有适合儿童的互动体验- 宝藏湾: 互动项目的类型和难度【综合建议】:考虑5岁儿童的年龄特点给出最适合的推荐。改进点:✅ 补充了年龄信息5岁✅ 细化了4个对比维度✅ 添加了具体的评估标准✅ 要求给出综合建议# 测试案例2: 项目对比test_case_2 {query: 创极速光轮和七个小矮人矿山车哪个更刺激,context: 用户想体验刺激的过山车}result_2 rewriter.rewrite(test_case_2[query], test_case_2[context])# 输出:改写后:比较上海迪士尼乐园的创极速光轮和七个小矮人矿山车这两个过山车项目哪个更刺激请从以下维度对比1. 速度对比: 最高时速分别是多少2. 高度对比: 轨道最高点和落差距离3. 失重感: 是否有垂直下降的失重体验4. 旋转程度: 是否有翻转、螺旋等动作5. 综合刺激度: 整体刺激程度排名目标: 为喜欢刺激体验的游客推荐最合适的项目。4.5 类型3模糊指代型Query改写多轮对话必需指代消歧的3种策略—— “列清单式翻译完整实现代码带详细注释# # 模糊指代型Query改写器# classReferenceQueryRewriter: 模糊指代型Query改写器 解决的核心问题: - 它指代什么 - 都包括哪些 - 这个具体是什么 技术难点: 1. 指代词识别NLP任务 2. 指代对象定位需要理解上下文 3. 多义性消解可能有多个候选对象 def__init__(self, modelqwen-turbo-latest): self.model modeldefrewrite(self, current_query, conversation_history): 消除模糊指代生成明确的Query 工作流程: 1. 识别Query中的所有指代词 2. 在对话历史中定位指代对象 3. 用明确的名词替换指代词 4. 验证改写后的Query语义完整 参数: current_query: 当前查询包含指代词 conversation_history: 对话历史 返回: 消歧后的明确查询 instruction 你 是一个语言歧义消除专家专门处理指代词问题。 【常见指代词分类】:1. 单数指代指代单个对象: 1.它、他、她 → 具体的人或物 2. 这个、那个 → 前文提到的事物 该 → 正在讨论的对象2. 复数指代指代多个对象: • 都、全部、所有 → 前文提到的多个对象 • 它们、他们、这些、那些 → 一组对象 3. 时间指代指代前文内容: 刚才、之前、前面说的 → 前文的话题或对象 【消歧步骤】: 步骤1 - 识别指代词:扫描Query标注所有指代词及其位置 步骤2 - 定位指代对象 • 对于单数指代: 找最近提到的同类名词• 对于复数指代: 找所有相关的名词并列举• 对于时间指代: 定位到具体的对话内容 步骤3 - 验证指代关系 • 检查名词类型是否匹配项目/园区/服务等• 检查语义是否合理• 处理多义性如果有多个候选对象选择最可能的 步骤4 - 执行替换 • 用明确的名词替换指代词• 保持句子的流畅性• 确保语义完整无歧义【示例1 - 单数指代】:对话历史: 用户: 创极速光轮是什么项目 AI: 创极速光轮是明日世界主题园区的过山车项目是全球最快的迪士尼过山车。当前Query: 它有身高要求吗分析: • 指代词: 它 • 指代对象: 创极速光轮最近提到的项目名词 • 验证: ✅ 类型匹配都是游乐项目改写: 创极速光轮过山车项目有身高要求吗【示例2 - 复杂指代】:对话历史: 用户: 我看到有个海盗船的项目 AI: 您说的应该是宝藏湾的加勒比海盗战争之潮这是全球首个海盗主题迪士尼项目。 用户: 这个项目好玩吗 AI: 非常好玩采用了先进的机器人和投影技术。当前Query: 那个适合小朋友吗分析: • 指代词: 那个 • 需要追溯: 对话已经进行了3轮 • 指代对象: 加勒比海盗战争之潮初始提到的项目改写: 宝藏湾的加勒比海盗战争之潮项目适合小朋友吗 prompt f ### 指令 ###{instruction} ### 对话历史 ###{conversation_history} ### 当前查询包含指代词###{current_query} ### 消歧后的查询 ### response get_completion(prompt, self.model)return response.strip()实战测试与效果分析# # 完整测试指代消歧效果对比# deftest_reference_disambiguation():测试指代消歧的效果 print(*80) print( 模糊指代型Query改写 - 效果对比测试) print(*80) rewriter ReferenceQueryRewriter()# 准备测试用例 test_cases [ {name: 都的指代最常见,history: 用户: 疯狂动物城和宝藏湾有什么好玩的AI: 疯狂动物城有警察局互动、训练营宝藏湾有加勒比海盗、独木舟。, query: 都有身高限制吗, expected: 两个园区的游乐项目, difficulty: ⭐⭐⭐ }, { name: 它的指代次常见, history: 用户: 创极速光轮是什么项目AI: 创极速光轮是明日世界的摩托过山车全球最快, query: 它有身高要求吗, expected: 创极速光轮, difficulty: ⭐⭐ }, { name: 那个的远距离指代较难, history: 用户: 我看到有个海盗船项目AI: 是宝藏湾的加勒比海盗战争之潮。用户: 这个项目好玩吗AI: 非常好玩采用了先进技术。, query: 那个适合小朋友吗, expected: 加勒比海盗项目需要追溯到第1轮, difficulty: ⭐⭐⭐⭐ } ] # 执行测试 for i, test in enumerate(test_cases, 1): print(f\n{*80}) print(f 测试案例 {i}: {test[name]} (难度: {test[difficulty]})) print(f{*80}) print(f\n 对话历史:) print(test[history]) print(f\n❓ 用户Query: {test[query]}) print(f 期待识别: {test[expected]}) # 执行改写 result rewriter.rewrite(test[query], test[history]) print(f\n✅ 改写结果:) print(f {result}) # 效果分析 if test[expected].split()[0] in result: print(f\n✨ 效果评估: ✅ 成功识别指代对象) else: print(f\n⚠️ 效果评估: 需要人工复核) if i len(test_cases): input(\n按回车继续下一个测试...) print(f\n{*80}) print(✨ 所有测试完成) print(f{*80}\n)4.6 类型4多意图型Query改写效率优化关键多意图识别的3种模式 —— “分诊台式翻译”完整实现代码# # 多意图型Query改写器# classMultiIntentQueryRewriter: 多意图型Query改写器 核心能力: 1. 识别一个Query中包含的多个意图 2. 拆分为独立的子Query 3. 保留原始Query的上下文信息 4. 标注每个子Query的优先级 应用场景: • 用户连续提问价格时间地点• 复杂查询对比列举建议 • 效率优化一次问多个问题def__init__(self, modelqwen-turbo-latest): self.model modeldefrewrite(self, query, context): 拆分多意图Query为多个独立子Query 拆分策略: 1. 显式标记识别: 问号、顿号、分号等 2. 语义主题识别: 通过LLM理解不同意图 3. 上下文保留: 每个子Query都包含必要的上下文 4. 优先级标注: 标注哪些问题更重要 参数: query: 原始查询可能包含多个意图 context: 上下文信息 instruction 你是一个查询意图分析专家专门识别和拆分多意图查询。【任务】:判断用户Query是否包含多个意图如果包含则拆分为独立的子查询。 【识别标志】: 1. 显式标记最明显: 多个问号、顿号分隔、分号分隔、并列连词 2. 语义层面需要理解:不同主题、不同层次、不同对象 【拆分原则】: 原则1 - 独立性: 每个子Query必须是一个完整、独立的问题不依赖其他子Query 原则2 - 上下文保留: 每个子Query都要包含必要的上下文地点、对象等 原则3 - 优先级标注:- high: 主要问题、核心需求- medium: 次要问题、补充信息- low: 可选问题、额外信息 prompt f ### 指令 ### {instruction} ### 上下文信息 ### {context} ### 用户查询 ###{query} ### 分析结果JSON格式### response get_completion(prompt, self.model)# 解析JSON try: result json.loads(preprocess_json_response(response))return resultexcept Exception as e:# 如果解析失败返回原始Query return {intent_count: 1,sub_queries: [],original_query: query,reasoning: f解析失败: {str(e)} }拆分后如何处理完整工作流4.7 类型5反问型Query改写情绪处理反问句改写的核心策略—— “情商提升式翻译”策略核心情绪分离 意图提取 同理心回应完整实现代码# # 反问型Query改写器# classRhetoricalQueryRewriter: 反问型Query改写器 核心功能: 1. 识别反问句式不会...吧、难道等 2. 分析用户情绪焦虑、不满、质疑等 3. 转换为客观疑问句 4. 标注情绪用于生成同理心回复 应用价值: • 提高检索准确性去除情绪干扰词 • 改善用户体验理解并回应情绪• 降低投诉率及时情绪安抚 def__init__(self, modelqwen-turbo-latest): self.model modeldefrewrite(self, query, context): 将反问句改写为客观疑问句并标注情绪 改写逻辑: 1. 识别反问句式和情绪词 2. 去除否定词和情绪修饰 3. 转换为直接的疑问句 4. 保留核心查询意图 5. 附加情绪标签 参数: query: 原始查询可能是反问句 context: 上下文信息 instruction 你是一个情绪识别和Query转换专家专门处理用户的反问句。【任务】:1. 判断用户Query是否为反问句2. 分析用户的情绪类型和强度3. 将反问句转换为客观的疑问句4. 提供情绪回应的关键词 【反问句识别标志】: 1. 反问句式:不会...吧 → 表示担心、不希望发生 2. 难道... → 表示质疑、不相信 3. 怎么可能... → 表示惊讶、怀疑 4. 就这样 / 只有... → 表示失望、不满 5. 居然... → 表示意外、惊讶 6. 情绪词汇: • 太...太贵、太远、太难 → 不满 7. 这么...这么少、这么慢 → 失望 8. 还...还要等、还得 → 焦虑 【转换规则】: 规则1 - 去除否定:不会要排队2小时吧 → 实际排队时间是多久 规则2 - 去除夸张:怎么这么贵 → 门票价格是多少价格构成如何 规则3 - 转换语气:难道只有这些项目 → 园区有哪些游乐项目 规则4 - 补充完整:就这样 → XX园区还有其他项目或服务吗 【情绪分类】: 不满/抱怨 (complaining):• 标志: 太贵、太远、太久• 强度: 根据语气词判断居然高强度有点低强度• 回应: 表示理解给出合理解释 焦虑/担心 (anxious):• 标志: 不会...吧、会不会• 强度: 根据问题严重程度• 回应: 安抚情绪给出确定答案 惊讶/质疑 (surprised/doubtful):• 标志: 怎么可能、真的• 强度: 中等• 回应: 提供证据消除疑虑 失望/遗憾 (disappointed):• 标志: 只有、就这样• 强度: 根据期望差距• 回应: 提供额外选项弥补期望【示例1 - 焦虑型反问】:原始Query: 创极速光轮不会要排队2小时吧分析:{ is_rhetorical: true, emotion: anxious, emotion_intensity: medium, rewritten_query: 创极速光轮过山车的实际排队时间是多久平日和周末分别需要等待多长时间, original_query: 创极速光轮不会要排队2小时吧, empathy_keywords: [理解您对排队时间的担心, 让我为您查询实际情况, 其实], reasoning: 用户使用不会...吧反问句式表达对长时间排队的担心和焦虑。通过转换为客观询问实际排队时间并在回答时给出减少排队的建议可以有效缓解用户焦虑。}【示例2 - 不满型反问】:原始Query: 门票怎么这么贵分析:{ is_rhetorical: true, emotion: complaining, emotion_intensity: medium, rewritten_query: 上海迪士尼乐园的门票价格是多少门票价格包含哪些内容和服务, original_query: 门票怎么这么贵, empathy_keywords: [确实相对较高, 但是包含, 性价比], reasoning: 用户使用怎么这么贵表达对价格的不满。转换为询问价格和包含内容回答时说明价值构成有助于用户理解价格合理性。}【示例3 - 失望型反问】:原始Query: 疯狂动物城就只有这几个项目分析:{ is_rhetorical: true, emotion: disappointed, emotion_intensity: low, rewritten_query: 疯狂动物城园区有哪些游乐项目和体验内容除了主要项目还有其他活动吗, original_query: 疯狂动物城就只有这几个项目, empathy_keywords: [园区虽然不大但, 还有很多细节, 推荐], reasoning: 用户用就只有表达失望期望更多项目。转换为全面询问项目和活动回答时强调隐藏内容和体验细节可以提升满意度。}【示例4 - 非反问句】:原始Query: 迪士尼门票多少钱分析:{ is_rhetorical: false, emotion: neutral, emotion_intensity: none, rewritten_query: 迪士尼门票多少钱, original_query: 迪士尼门票多少钱, empathy_keywords: [], reasoning: 正常的疑问句无情绪色彩无需转换。}【输出格式】:必须返回JSON格式包含所有必需字段 prompt f ### 指令 ###{instruction} ### 上下文信息 ###{context} ### 用户查询 ###{query} ### 分析结果JSON格式### response get_completion(prompt, self.model)# 解析JSON try: result json.loads(preprocess_json_response(response))return resultexcept Exception as e:# 如果解析失败返回原始Query return {is_rhetorical: False,emotion: neutral,emotion_intensity: none,rewritten_query: query,original_query: query,empathy_keywords: [],reasoning: f解析失败: {str(e)} }4.8 Query改写自动引擎AutoQueryRewriter为什么需要统一引擎我们已经实现了5种独立的Query改写器上下文依赖型、对比型、模糊指代型、多意图型、反问型。在真实客服场景中用户的提问往往同时具备多种特征。如果人工判断调用顺序不但效率低而且容易漏掉关键改写。因此需要一个统一的智能引擎将所有改写器按优先级串联输出可溯源的改写轨迹。·✅ 自动识别Query类型支持多标签·✅ 按优先级顺序调用改写器避免逻辑冲突·✅ 输出改写日志便于调试与分析·✅ 携带情绪标签帮助生成阶段输出同理心回复·✅ 易扩展未来新增改写类型只需插入模块改写优先级多意图型 模糊指代型 对比型 上下文依赖型 反问型AutoQueryRewriter 核心代码# # AutoQueryRewriter: 统一Query改写引擎# import jsonfrom typing import Dict, Any, ListclassAutoQueryRewriter:统一的Query改写调度器 PRIORITY [multi_intent, # 多意图型必须先拆分reference, # 模糊指代型让语义清晰comparison, # 对比型补充对比维度context_dependent, # 上下文依赖型补全信息rhetorical# 反问型最后处理情绪 ]def__init__(self, llm_model: str qwen-turbo-latest): self.llm_model llm_model self.context_rewriter context_rewriter or ContextDependentQueryRewriter(llm_model) self.comparison_rewriter comparison_rewriter or ComparisonQueryRewriter(llm_model) self.reference_rewriter reference_rewriter or ReferenceQueryRewriter(llm_model) self.multi_intent_rewriter multi_intent_rewriter or MultiIntentQueryRewriter(llm_model) self.rhetorical_rewriter rhetorical_rewriter or RhetoricalQueryRewriter(llm_model)# --------------- # Step1: 类型识别 # ------------- defanalyze_query_type(self, query: str, conversation_history: str ) - Dict[str, Any]: 调用LLM识别Query类型支持多标签 instruction 你是一个RAG系统的Query类型识别器请判断当前Query是否属于以下类型:1. multi_intent: 多意图2. reference: 模糊指代3. comparison: 对比型4. context_dependent: 上下文依赖5. rhetorical: 反问型请输出JSON字段包括 query_type, confidence, detected_keywords, reasoning。 prompt f ### 指令 ###{instruction} ### 对话历史 ###{conversation_history} ### 当前Query ###{query} ### 输出JSON ### response get_completion(prompt, self.llm_model)return json.loads(preprocess_json_response(response))# ------------- # Step2: 调度改写器 # ------------- defrewrite(self, query: str, conversation_history: str ) - Dict[str, Any]: analysis self.analyze_query_type(query, conversation_history) query_types analysis.get(query_type, [])ifnot query_types:return {final_query: query,query_types: [],sub_queries: [],emotion: {emotion: neutral,emotion_intensity: none,empathy_keywords: [] },rewrite_steps: [],analysis: analysis }# 先处理多意图 ifmulti_intentin query_types: multi_result self.multi_intent_rewriter.rewrite(query, conversation_history)if multi_result[intent_count] 1: final_sub_queries []for item in multi_result[sub_queries]: processed self._process_single_query( queryitem[query], conversation_historyconversation_history, query_types[qt for qt in query_types if qt ! multi_intent] ) processed.update({priority: item.get(priority, medium),topic: item.get(topic, general) }) final_sub_queries.append(processed)return {final_query: query,query_types: query_types,sub_queries: final_sub_queries,emotion: {emotion: neutral,emotion_intensity: none,empathy_keywords: [] },rewrite_steps: [{type: multi_intent, result: multi_result}],analysis: analysis }# 单Query改写流程 result self._process_single_query( queryquery, conversation_historyconversation_history, query_typesquery_types ) result.update({query_types: query_types,sub_queries: [],analysis: analysis })return result# ---------- # Step3: 针对单个Query按优先级改写 # ------- def_process_single_query(self, query: str, conversation_history: str, query_types: List[str]): current_query query rewrite_steps [] emotion_info {emotion: neutral,emotion_intensity: none,empathy_keywords: [] }for query_type in self.PRIORITY[1:]: # 多意图已在外层处理 if query_type notin query_types:continueif query_type reference: rewritten self.reference_rewriter.rewrite(current_query, conversation_history) rewrite_steps.append({type: reference, rewrite: rewritten}) current_query rewrittenelif query_type comparison: rewritten self.comparison_rewriter.rewrite(current_query, conversation_history) rewrite_steps.append({type: comparison, rewrite: rewritten}) current_query rewrittenelif query_type context_dependent: rewritten self.context_rewriter.rewrite(current_query, conversation_history) rewrite_steps.append({type: context, rewrite: rewritten}) current_query rewrittenelif query_type rhetorical: result self.rhetorical_rewriter.rewrite(current_query, conversation_history) rewrite_steps.append({type: rhetorical, rewrite: result}) current_query result[rewritten_query] emotion_info {emotion: result[emotion],emotion_intensity: result[emotion_intensity],empathy_keywords: result[empathy_keywords] }return {final_query: current_query,emotion: emotion_info,rewrite_steps: rewrite_steps }调试辅助打印改写链路defprint_rewrite_log(result: Dict[str, Any]): print(*80) print(️ Query改写流水线) print(*80) print(f识别类型: {, .join(result.get(query_types, []))}) print(f情绪标签: {result[emotion][emotion]} ({result[emotion][emotion_intensity]}))for step in result.get(rewrite_steps, []): print(-*80) print(f步骤: {step[type]}) print(f结果: {step[rewrite]})if result.get(sub_queries): print(\n 子Query列表:)for idx, sub in enumerate(result[sub_queries], 1): print(f [{idx}] ({sub[priority]}) {sub[final_query]}) print(f 情绪: {sub[emotion][emotion]}) print(*80)第五章Query联网搜索实时信息融合5.1 为什么需要联网搜索虽然知识库能够覆盖大部分静态信息但迪士尼智能客服还必须回答实时问题例如” 现在排队多久”“今晚有烟花吗”“今天会下雨吗”。这些信息在知识库中无法及时更新因此需要合理判断何时联网并将联网结果与知识库信息融合。这个模块的工作流通常包含三个核心环节。首先是联网识别逻辑。系统并非对所有查询都执行联网而是通过一个由LLM驱动的“联网判断器”来进行决策。这个判断器会分析用户的查询是否包含了“今天”、“现在”、“实时”、“最新”、“票价变动”或“天气”等具有明显时效性特征的关键词。同时它也会评估本地知识库对该查询的覆盖度和置信度。只有当查询明确需要实时信息或者本地知识库无法提供高置信度答案时系统才会触发联网搜索。其次是搜索策略生成。一旦决定联网系统会利用LLM作为搜索策略专家为原始查询制定详细的搜索计划。这包括将用户的口语化问题改写为更适合搜索引擎的精炼关键词组合例如将“今天去迪士尼人多吗”改写为“上海迪士尼 实时人流量 2025年11月16日”。此外LLM还会根据问题类型推荐最佳的数据来源例如对于官方活动优先搜索官方网站或官方社交媒体账号对于用户评价则优先搜索旅游平台或点评类网站。最后是联网执行与结果融合。系统根据生成的搜索策略调用外部的搜索API如Tavily或Google Search来获取最新的网络信息片段。这一步至关重要系统必须将联网获取的实时信息与本地知识库检索到的静态信息进行融合。融合过程中可能涉及去重、事实交叉验证和冲突消解例如本地知识库的票价是399元而联网搜索的最新公告是499元。LLM会基于这个融合了多方信源的、最全面的上下文来生成最终答案并清晰地标注出实时信息的时间戳和来源从而保证了答案的权威性与时效性。5.2 联网策略总览5.3 联网判断器实现classWebSearchDetector: 联网判断器def__init__(self, modelqwen-turbo-latest): self.model modeldefanalyze(self, query: str, context: str ) - Dict[str, Any]:判断用户问题是否需要联网 instruction 你是一个智能联网策略分析器请判断用户问题是否需要联网。 【必须联网的场景包括】 - 实时信息今天、现在、当前、实时、最新 - 动态状态排队、天气、交通、 - 时效内容活动、优惠、公告、新闻。 如果需要联网请生成搜索语句和推荐数据源。输出JSON: need_web_search, scene, confidence, reason, search_queries, data_sources。 prompt f ### 指令 ###{instruction} ### 对话历史 ###{context} ### 用户Query ###{query} ### 输出JSON ### response get_completion(prompt, self.model)return json.loads(preprocess_json_response(response))5.4 联网执行与结果融合classWebSearchExecutor: def__init__(self, search_api): self.search_api search_apidefexecute(self, strategy: Dict[str, Any]) - Dict[str, Any]: all_results []for sq in strategy[search_queries]: raw self.search_api.search(sq) clean self._clean(raw) normalized self._normalize(clean) all_results.extend(normalized)ifnot all_results:return {status: fallback,message: 暂未获取到实时数据 } top_results sorted(all_results, keylambda x: -x[confidence])[:3] prompt f请根据以下实时数据生成摘要:场景: {strategy[scene]}数据: {json.dumps(top_results, ensure_asciiFalse, indent2)}输出JSON: summary, sources, advice。 summary get_completion(prompt, modeldeepseek-v3)return json.loads(preprocess_json_response(summary))def_clean(self, raw_results):return [item for item in raw_results if广告notin item.get(title, )]def_normalize(self, results): normalized []for item in results: normalized.append({title: item.get(title),summary: item.get(snippet),source: item.get(source),url: item.get(url),published_time: item.get(time),confidence: self._estimate_confidence(item) })return normalizeddef_estimate_confidence(self, item): weight {official: 0.95,news: 0.85,social: 0.65 }return weight.get(item.get(source_type, social), 0.6)defmerge_kb_web(kb_docs, web_result): merged []if kb_docs: merged.append({type: knowledge_base,content: kb_docs,description: 知识库基础事实 })if web_result and web_result.get(status) ! fallback: merged.append({type: web_search,content: web_result,description: 实时数据含来源与时间 })return merged5.5 完整联网搜索流程端到端的工作流defcomplete_web_search_pipeline(user_query, conversation_history, knowledge_base): 完整的联网搜索流程# 1. 判断是否需要联网 detector WebSearchDetector() web_decision detector.analyze(user_query, conversation_history)ifnot web_decision[need_web_search]:return {action: local_only,reason: web_decision[reason],context: knowledge_base.retrieve(user_query) }# 2. 生成搜索策略 strategy_generator SearchStrategyGenerator() search_strategy strategy_generator.generate_strategy(user_query, conversation_history)# 3. 执行联网搜索 search_executor WebSearchExecutor(search_apiTavilySearchAPI()) web_results search_executor.execute_search(search_strategy)# 4. 检索本地知识库 kb_docs knowledge_base.retrieve(user_query)# 5. 融合信息 fusion_engine InformationFusion() fused_context fusion_engine.fuse_knowledge(kb_docs, web_results, user_query)return {action: web_search,web_strategy: search_strategy,web_results: web_results,fused_context: fused_context,final_context: self._format_final_context(fused_context) }def_format_final_context(self, fused_context):格式化最终上下文供LLM使用 context_parts []for ctx in fused_context:if ctx[type] knowledge_base: context_parts.append(f【基础知识】\n{ctx[content]})elif ctx[type] web_search: context_parts.append(f【实时信息 - 更新于{ctx[timestamp]}】\n{ctx[content]})return\n\n.join(context_parts)第六章召回优化6.1 为什么单一向量检索“偏科”入门篇中我们使用了纯向量检索Embedding Similarity虽然能捕捉语义相似性但单一向量检索就像只会意会不会言传的朋友——它能理解刺激好玩的项目但记不住创极速光轮这个具体名字。 检索Retrieval即召回是RAG流程的生命线召回内容的质量直接决定了最终生成答案的上限。在高级RAG系统中召回优化是一个多阶段、多策略协同的过程远比单一的向量检索复杂。 基础的向量检索虽然擅长捕捉语义的相似性例如能理解“适合小孩的项目”与“家庭友好型游乐设施”是相近的但它在处理精确匹配时存在天然缺陷。例如它对“122cm身高限制”这样的具体数字或“创极速光轮”这样的专有名词不敏感容易导致召回遗漏。 为了克服这一缺陷高级RAG系统普遍采用**混合检索Hybrid Retrieval策略。混合检索的核心思想是取长补短它将两种截然不同的检索技术结合起来。一方面是传统的关键词检索如BM25。另一方面是向量语义检索**它利用嵌入模型的强大能力捕捉深层语义确保了召回的相关性。系统会并行执行这两种检索然后通过特定的算法如Reciprocal Rank Fusion, RRF或简单的加权求和将两路召回的结果进行融合生成一个既包含精确匹配又包含语义相关的候选文档列表。结论需要结合关键词检索BM25和向量检索Embedding取长补短。6.2 BM25原理与实现BM25Best Matching 25是经典的关键词检索算法基于稀疏索引TF-IDF改进。BM 代表最佳匹配Best Match。BM 后面的数字 25 表示 BM 算法的迭代次数。该算法经过 25 次迭代后才被广泛采用和接受尽管现在也有 BM25 和其他变体。BM25的核心思想是计算查询中每个词项与文档的相关性然后综合所有词项得到最终分数。BM25得分计算公式包含多个组件。IDF逆文档频率衡量词项的重要性罕见词比常见词具有更高的区分度。TF词项频率表示词项在文档中出现的次数出现越多相关性越高但通过参数k1进行平滑处理避免过度依赖单一词项。 长度归一化组件通过参数b调节文档长度的影响防止长文档仅仅因为包含更多词项而获得不公平的优势。dl表示文档长度avgdl是平均文档长度b控制 长度归一化的强度通常设为0.75。 实现BM25检索器需要先对文档集合进行预处理。中文文档需要使用分词工具如jieba进行分词将文本转换为词项序列。然后基于分词后的文档构建BM25模型计算每个词项的统计信息。 检索时先对查询进行分词然后计算查询与每个文档的BM25分数。分数计算涉及查询中每个词项在文档中的频率、词项的逆文档频率以及文档长度等因素。最后根据分数排序返回Top-K相关文档。 BM25的优势在于对精确匹配和特定术语的敏感性能够有效捕捉用户查询中的关键词信息。与向量检索结合使用可以弥补向量模型在精确匹配方面的不足提供更全面的检索覆盖。核心公式实现示例from rank_bm25 import BM25Okapiimport jiebaclassBM25Retriever:BM25关键词检索器def__init__(self, documents): self.tokenized_docs [list(jieba.cut(doc)) for doc in documents] self.bm25 BM25Okapi(self.tokenized_docs) 参数: documents: 文档列表每个文档是一个字符串 # 中文分词 self.tokenized_docs [list(jieba.cut(doc)) for doc in documents]# 初始化BM25 self.bm25 BM25Okapi(self.tokenized_docs)# 保存原始文档 self.documents documentsdefsearch(self, query: str, top_k: int 10):精准关键词匹配 tokenized_query list(jieba.cut(query)) scores self.bm25.get_scores(tokenized_query) # 分词 tokenized_query list(jieba.cut(query))# 计算BM25分数 scores self.bm25.get_scores(tokenized_query)# 排序并返回Top-K top_indices sorted(range(len(scores)), keylambda i: -scores[i])[:top_k] results []for idx in top_indices: results.append({index: idx,score: float(scores[idx]),content: self.documents[idx] })return results测试示例# 迪士尼知识库文档docs [ 创极速光轮是明日世界园区的过山车项目身高要求122cm以上。,疯狂动物城警察局是互动体验项目适合5-12岁儿童。,宝藏湾的加勒比海盗项目身高要求97cm以上。,七个小矮人矿山车是梦幻世界的过山车身高要求97cm以上。]bm25_retriever BM25Retriever(docs)# 测试1精确数字匹配query1 122cm身高要求results1 bm25_retriever.search(query1, top_k2)print(Query:, query1)for r in results1: print(f 分数: {r[score]:.2f} | {r[content]})# 输出# Query: 122cm身高要求# 分数: 3.45 | 创极速光轮是明日世界园区的过山车项目身高要求122cm以上。# 分数: 0.82 | 宝藏湾的加勒比海盗项目身高要求97cm以上。# 测试2专有名词匹配query2 疯狂动物城警察局results2 bm25_retriever.search(query2, top_k2)print(\nQuery:, query2)for r in results2: print(f 分数: {r[score]:.2f} | {r[content]})# 输出# Query: 疯狂动物城警察局# 分数: 8.12 | 疯狂动物城警察局是互动体验项目适合5-12岁儿童。# 分数: 0.00 | 创极速光轮是明日世界园区的过山车项目身高要求122cm以上。6.3 混合检索架构混合检索架构通过结合多种检索技术来解决单一方法的局限性。让向量检索和BM25组队就像让文科生和理科生搭档——一个负责理解深层含义一个负责记住具体细节。整个架构从用户查询开始通过检索策略分析确定合适的检索方法。系统并行执行BM25关键词检索和向量语义检索分别获取两种方法的前20个结果。 BM25检索充当文科生基于传统的信息检索技术擅长处理精确匹配、专有名词和长尾词汇。它计算查询词项与文档的统计相关性对数字、特定术语等精确匹配效果很好。向量检索充当理科生基于深度学习模型擅长语义理解和概念匹配能够找到语义相关但用词不同的文档。 结果合并环节将两种检索方法的结果进行融合。首先对BM25分数和向量相似度分数进行归一化处理使它们处于可比较的范围内。然后根据预设的权重进行加权融合生成每个文档的混合分数。权重的设置需要根据具体场景调整通常向量权重略高于BM25权重。6.4 混合检索实现混合检索器的实现需要协调BM25和向量检索两种方法确保它们协同工作而不是相互干扰。初始化时需要准备文档集合、向量存储实例以及设置两种方法的权重参数。权重参数决定了BM25和向量检索在最终结果中的相对重要性需要根据具体场景进行调整。检索过程首先并行执行BM25检索和向量检索分别获取各自的前20个结果。BM25检索返回文档索引、BM25分数和文档内容向量检索返回文档和相似度分数。由于两种方法的分数范围不同需要进行归一化处理。BM25分数归一化通常采用最大最小值归一化将分数转换到0-1范围。向量相似度分数通常已经在0-1范围内可以直接使用。归一化后根据预设的权重计算每个文档的混合分数。结果合并需要建立统一的文档标识体系。BM25结果使用文档索引作为标识向量结果需要映射到相同的文档索引。对于只在一种检索方法中出现的文档另一种方法的分数设为0。混合分数计算完成后根据分数排序选择Top-K文档。返回结果包含文档内容、BM25分数、向量分数和混合分数便于后续分析和调试。这种透明的结果格式有助于理解混合检索的工作机制优化权重参数。混合检索的优势在于结合了两种方法的优点。BM25确保关键词匹配的准确性向量检索保证语义相关的完整性两者结合提供了更全面的检索覆盖。实际应用中混合检索通常比单一方法获得更好的效果。classHybridRetriever: 混合检索器BM25 向量检索def__init__(self, documents, vector_store, bm25_weight0.3, vector_weight0.7):参数: documents: 文档列表、vector_store: FAISS向量库、 bm25_weight: BM25权重 vector_weight: 向量权重 self.bm25_retriever BM25Retriever(documents) self.vector_store vector_store self.bm25_weight bm25_weight self.vector_weight vector_weightdefsearch(self, query: str, top_k: int 10):混合检索策略:1. BM25检索Top-20 2. 向量检索Top-20 3. 归一化分数 4. 加权融合 5. 返回Top-K# BM25检索 bm25_results self.bm25_retriever.search(query, top_k20)# 向量检索 vector_results self.vector_store.similarity_search_with_score(query, k20)# 归一化BM25分数 bm25_scores [r[score] for r in bm25_results]if max(bm25_scores) 0: bm25_scores_norm [s / max(bm25_scores) for s in bm25_scores]else: bm25_scores_norm bm25_scores# 归一化向量分数余弦相似度已在0-1之间 vector_scores_norm [score for _, score in vector_results]# 构建文档分数字典 doc_scores {}for i, r in enumerate(bm25_results): doc_id r[index] doc_scores[doc_id] {bm25_score: bm25_scores_norm[i] * self.bm25_weight,vector_score: 0,content: r[content] }for i, (doc, score) in enumerate(vector_results): doc_id i # 需要映射到实际文档ID if doc_id in doc_scores: doc_scores[doc_id][vector_score] vector_scores_norm[i] * self.vector_weightelse: doc_scores[doc_id] {bm25_score: 0,vector_score: vector_scores_norm[i] * self.vector_weight,content: doc.page_content }# 计算混合分数 for doc_id in doc_scores: doc_scores[doc_id][hybrid_score] ( doc_scores[doc_id][bm25_score] doc_scores[doc_id][vector_score] )# 排序并返回Top-K sorted_docs sorted( doc_scores.items(), keylambda x: -x[1][hybrid_score] )[:top_k] results []for doc_id, scores in sorted_docs: results.append({doc_id: doc_id,content: scores[content],bm25_score: scores[bm25_score],vector_score: scores[vector_score],hybrid_score: scores[hybrid_score] })return results6.5 Rerank重排序混合检索后得到Top-20或Top-40候选文档但这些文档的相关性排序可能仍不够精准。Rerank通过更强大的模型如BGE-Rerank、Cohere Rerank对候选文档重新打分排序。Rerank模型通常基于交叉编码器架构能够同时处理查询和文档进行深度的注意力交互。与双编码器架构的向量检索相比交叉编码器计算代价更高但效果更好适合对少量候选进行精排。BGE-Rerank是常用的开源Rerank模型基于预训练的语言模型微调得到。使用时需要加载模型和分词器对查询-文档对进行编码获取相关性分数。由于计算量较大通常只对Top-20或Top-40候选进行Rerank。实现Rerank器需要处理输入格式和批量计算。查询和文档列表构造成对输入通过分词器转换为模型可接受的格式。模型推理获取每个对的相关性分数然后根据分数对文档重新排序。Rerank的代价是增加响应时间通常需要200-500毫秒额外的计算时间。因此需要在效果和性能之间权衡对于实时性要求高的场景可以跳过Rerank对于准确性要求高的场景则值得使用。重排序任务类似于一个智能过滤器。当检索器从索引集合中检索到多个上下文时这些上下文与用户查询的相关性可能不同。有些上下文可能非常相关图 1 中以红色方框突出显示而另一些上下文可能只有轻微的相关性甚至完全不相关图 1 中以绿色和蓝色方框突出显示。图1RAG 中的重排序重排序的任务是评估这些上下文的相关性并优先考虑最有可能提供准确相关答案的上下文红色方框。BGE-Rerank实现from transformers import AutoModelForSequenceClassification, AutoTokenizerimport torchclassBGEReranker:BGE-Rerank重排序器def__init__(self, model_nameBAAI/bge-reranker-large): self.tokenizer AutoTokenizer.from_pretrained(model_name) self.model AutoModelForSequenceClassification.from_pretrained(model_name) self.model.eval()defrerank(self, query: str, documents: List[str], top_k: int 3): 重排序 参数: query: 用户查询 、documents: 候选文档列表、top_k: 返回前K个 返回: List[Tuple[文档索引, Rerank分数, 文档内容]]# 构建输入对 pairs [[query, doc] for doc in documents]# Tokenize with torch.no_grad(): inputs self.tokenizer( pairs, paddingTrue, truncationTrue, return_tensorspt, max_length512 )# 前向传播 scores self.model(**inputs, return_dictTrue).logits.view(-1).float()# 排序 sorted_indices torch.argsort(scores, descendingTrue).tolist() results []for idx in sorted_indices[:top_k]: results.append({index: idx,score: float(scores[idx]),content: documents[idx] })return results完整召回流程defadvanced_retrieval_pipeline(query: str, hybrid_retriever, reranker, top_k_hybrid20, top_k_final3): 高级召回流程混合检索 Rerank print(f*80) print(f 查询: {query}) print(f*80)# 步骤1混合检索 print(\n 步骤1混合检索BM25 Vector) hybrid_results hybrid_retriever.search(query, top_ktop_k_hybrid) print(f 获得 {len(hybrid_results)} 个候选文档)for i, r in enumerate(hybrid_results[:5], 1): print(f [{i}] 混合分数: {r[hybrid_score]:.3f} f(BM25: {r[bm25_score]:.3f}, Vector: {r[vector_score]:.3f})) print(f {r[content][:50]}...)# 步骤2Rerank重排序 print(f\n 步骤2Rerank重排序) documents [r[content] for r in hybrid_results] rerank_results reranker.rerank(query, documents, top_ktop_k_final) print(f 重排序后Top-{top_k_final}:)for i, r in enumerate(rerank_results, 1): print(f [{i}] Rerank分数: {r[score]:.3f}) print(f {r[content]}) print(f\n{*80}\n)return rerank_results第七章知识库问题生成7.1 从“精准检索”到“智能匹配”上一章我们打造了强大的混合检索引擎让系统能够精准找到相关信息。但有一个关键问题依然存在**如果用户的提问方式与知识库表述差异太大再好的检索引擎也无力回天**。这就好比拥有了最先进的搜索引擎但用户却用方言提问——匹配度依然上不去。所以要先解决传统知识库现实困境 ❌ **覆盖不全**人工很难穷举所有可能的提问方式 ❌ **更新滞后**知识更新后相应问题未及时补充 ❌ **检索失效**用户提问方式与FAQ不匹配导致检索失败解决方案利用LLM为每个知识片段Chunk自动生成多样化的问题建立”问题→答案”映射提高检索覆盖率。从技术架构看知识片段首先通过LLM问题生成器产生多种类型的问题。标准问法如创极速光轮身高要求直接对应知识点口语化问法如小孩能玩创极速光轮吗更符合用户习惯对比问法如创极速光轮和七个小矮人哪个身高限制高涉及多个知识点场景化问法如带5岁孩子能玩创极速光轮吗结合具体使用情境。生成的问题被构建成问题索引通过BM25等关键词检索技术支持用户查询。当用户提出问题时系统在问题索引中找到最匹配的表述然后关联到对应的知识片段。这种间接检索方式虽然增加了一个步骤但显著提升了检索的成功率。7.2 问题生成策略让知识库说用户的语言直接提问是最基础的生成类型会直接询问知识点中的关键信息比如创极速光轮的身高要求是多少。这种问法虽然简单明了但可能听起来不够自然哦。我们的核心思路是为每个知识片段预生成各种可能的用户提问建立提问-答案映射库。classQuestionGenerator: 知识库问题生成器def__init__(self, modelqwen-turbo-latest): self.model modeldefgenerate_questions(self, chunk: str, num_questions: int 5): 为一个知识片段生成多样化问题 生成类型: 1. 直接提问直接问这个知识点 2. 场景提问结合具体场景 3. 对比提问与其他知识点对比 4. 反问提问用反问句 5. 口语提问日常口语表达 instruction 你是一个智能问题生成器请为给定的知识片段生成5种不同类型的问题。 【生成要求】: 1. 直接提问最直接的问法 2. 场景提问结合用户使用场景 3. 对比提问与其他事物对比 4. 口语提问日常口语化表达 5. 反问提问用反问句表达 prompt f ### 指令 ### {instruction} ### 知识片段 ### {chunk} ### 生成的问题JSON数组### response get_completion(prompt, self.model) questions json.loads(preprocess_json_response(response))return questions7.3 构建智能问题索引# # 知识库问题生成与检索优化完整流程# from rank_bm25 import BM25Okapiimport jiebaimport json# 迪士尼知识库示例knowledge_base [ {id: chunk_001,content: 创极速光轮是明日世界园区的摩托过山车是全球最快的迪士尼过山车。身高要求122cm以上刺激程度较高适合喜欢刺激的青少年和成人。 }, {id: chunk_002,content: 疯狂动物城警察局互动体验是一个室内互动项目适合5-12岁儿童。无身高限制家长可陪同。项目时长约15分钟。 }, {id: chunk_003,content: 上海迪士尼乐园门票价格平日成人票399元周末/节假日499元。儿童票1.0-1.4米平日299元周末374元。1.0米以下免费。 }]# 步骤1为每个知识片段生成问题 generator QuestionGenerator()question_index [] # 问题索引[{question: ..., chunk_id: ...}]print(*80)print( 步骤1为知识片段生成问题)print(*80)for chunk in knowledge_base: print(f\n知识片段ID: {chunk[id]}) print(f内容: {chunk[content][:50]}...) questions generator.generate_questions(chunk[content]) print(f生成问题:)for question in questions: question_index.append({question: question,chunk_id: chunk[id],original_content: chunk[content],question_type: self._classify_question_type(question) })# 步骤2构建BM25索引print(f\n{*80})print( 步骤2构建问题BM25索引)print(*80)questions_list [item[question] for item in question_index]tokenized_questions [list(jieba.cut(q)) for q in questions_list]bm25 BM25Okapi(tokenized_questions)print(f✅ 索引构建完成共 {len(questions_list)} 个问题)# 步骤3用户查询测试print(f\n{*80})print( 步骤3用户查询测试)print(*80)test_queries [小孩能玩创极速光轮吗,迪士尼门票多少钱,有适合小朋友的项目吗]for user_query in test_queries: print(f\n用户Query: {user_query})# BM25检索 tokenized_query list(jieba.cut(user_query)) scores bm25.get_scores(tokenized_query) top_idx scores.argmax() matched_question question_index[top_idx] matched_chunk_id matched_question[chunk_id] matched_chunk next(c for c in knowledge_base if c[id] matched_chunk_id) print(f 匹配到问题: {matched_question[question]}) print(f 对应知识片段: {matched_chunk[content][:80]}...) print(f BM25分数: {scores[top_idx]:.2f})7.4 基于问题库的智能检索现在检索流程变成了两步在问题库中匹配最相似的用户提问通过匹配到的问题找到对应知识点classSmartQuestionRetriever: 智能问题检索器 - 理解用户的各种问法def__init__(self, question_index): self.question_index question_index self.bm25 self._build_question_index(question_index)def_build_question_index(self, question_index):构建问题检索索引 questions [item[question] for item in question_index] tokenized_questions [list(jieba.cut(q)) for q in questions]return BM25Okapi(tokenized_questions)defretrieve(self, user_query: str, top_k: int 3):通过问题库检索相关知识 tokenized_query list(jieba.cut(user_query)) scores self.bm25.get_scores(tokenized_query)# 找到最匹配的预生成问题 best_match_idx scores.argmax() best_match_score scores[best_match_idx]if best_match_score 1.0: # 置信度阈值 matched_question self.question_index[best_match_idx]return {matched_question: matched_question[question],chunk_content: matched_question[original_content],confidence: best_match_score,question_type: matched_question[question_type] }else:return {status: no_good_match, confidence: best_match_score}7.5评估检索效果# 测试用例对比test_queries [小孩多高能玩摩托过山车, # 口语化提问122cm是哪个项目的要求, # 反向提问创极速光轮会不会太刺激# 担忧型提问]print( 检索效果对比测试)for query in test_queries:# 传统检索 traditional_result hybrid_retriever.search(query) traditional_score traditional_result[0][score] if traditional_result else0# 问题库检索 smart_result question_retriever.retrieve(query) smart_score smart_result.get(confidence, 0) print(f查询: {query}) print(f 传统检索得分: {traditional_score:.3f}) print(f 问题库检索得分: {smart_score:.3f})if smart_score traditional_score: print(f 问题库检索更优匹配到: {smart_result[matched_question]})第八章对话知识沉淀8.1 从静态知识库到活的知识生态经过前七章的优化我们的RAG系统已经具备了精准检索和理解用户意图的能力。但还有一个根本性问题没有解决**知识库更新滞后于现实变化**。在实际运营中**用户与客服的对话本身就是宝贵的知识来源** ✅ **发现知识盲区**用户提出的问题知识库可能没有覆盖 ✅ **丰富表达方式**用户的提问方式多样补充到知识库 ✅ **及时性信息**用户反馈的最新情况如:今天某项目维护 ✅ **用户需求**用户关心什么可以指导知识库建设方向这就是对话知识沉淀要解决的终极问题让知识库具备自我进化能力。8.2 对话知识提取流程对话知识提取是一个系统化的过程需要从非结构化的对话记录中抽取出结构化的知识内容。整个过程开始于原始对话记录的收集这些记录包含了用户与客服的完整交互过程包括多轮对话的上下文关系。对话记录需要经过预处理包括去除敏感信息、标准化格式、分割对话轮次等步骤为后续的知识提取做好准备。知识提取环节使用大语言模型的强大理解能力从对话中识别有价值的知识点。事实类知识提取关注对话中提到的客观事实如具体项目的运营参数、价格信息、时间安排等。这些事实通常有明确的表述和依据可以直接转化为知识库内容。需求类知识提取则聚焦于用户表达的需求和关切如用户频繁询问的问题类型、特别关注的方面等这些信息反映了用户的真实需求。 流程类知识提取关注对话中涉及的操作步骤和解决方法如购票流程、预约方式、问题处理流程等。这类知识往往分散在多个对话轮次中需要系统性地梳理和整合。注意事项类知识提取则捕捉对话中提到的各种提示、限制和特殊情况这些内容对用户体验有重要影响。 知识过滤环节对提取的知识进行质量把关去除质量不高或可信度不足的内容。过滤标准包括知识的准确性、完整性、时效性和重要性。准确性确保信息真实可靠完整性要求知识表达充分完整时效性排除过时信息重要性评估知识对用户的价值程度。去重与合并环节处理提取知识中的重复和相似内容。相同的知识可能从不同对话中多次提取需要识别并合并这些重复内容。相似的知识可能存在细微差异需要进行内容融合形成更全面准确的表达。这个过程确保知识库的简洁性和一致性。最终经过处理的知识被更新到知识库中完成从对话到知识的转化。这个流程可以配置人工审核环节对敏感或重要的知识进行人工确认确保知识质量。整个提取过程可以定期自动执行形成持续的知识更新机制。8.3 完整实现# # 对话知识沉淀系统# classConversationKnowledgeExtractor:对话知识提取器def__init__(self, modelqwen-turbo-latest): self.model modeldefextract_knowledge(self, conversation: str):从对话中提取结构化知识 提取类型: 1. 事实类知识客观事实如价格、时间、要求 2. 需求类知识用户关注点、高频问题 3. 流程类知识办事流程、使用步骤 4. 注意事项提示、限制、特殊情况 instruction 你是一个智能知识提取器请从客服对话中提取有价值的知识。 【提取要求】: 1. 只提取客观、准确的知识有依据的 2. 去除临时性信息如今天下雨 3. 提取用户关注的问题需求 4. 提取流程和注意事项 【输出格式】JSON: prompt f ### 指令 ### {instruction }### 对话记录 ### {conversation} ### 提取的知识JSON### response get_completion(prompt, self.model) knowledge json.loads(preprocess_json_response(response))return knowledgeclassKnowledgeMerger:知识合并器def__init__(self, modelqwen-turbo-latest): self.model modeldefmerge_similar_knowledge(self, knowledge_list):合并相似的知识点策略: 1. 识别相同主题的知识 2. 合并重复信息 3. 保留最完整、最准确的版本 instruction 你是一个知识合并专家请合并相似的知识点。 【合并原则】: 1. 相同主题的知识合并为一条 2. 保留最完整、最准确的信息 3. 去除重复和矛盾的内容 4. 标注置信度 prompt f ### 指令 ###{instruction} ### 知识列表 ### {json.dumps(knowledge_list, ensure_asciiFalse, indent2)} ### 合并结果JSON### response get_completion(prompt, self.model) merged json.loads(preprocess_json_response(response))return merged# 完整流程示例def knowledge_deposition_pipeline(conversation_logs):对话知识沉淀完整流程 extractor ConversationKnowledgeExtractor() merger KnowledgeMerger() all_knowledge {facts: [],needs: [],processes: [],tips: [] } print(*80) print( 对话知识沉淀流程) print(*80)# 步骤1从多个对话中提取知识 print(\n步骤1知识提取)for i, conversation in enumerate(conversation_logs, 1): print(f\n 对话{i}:) print(f {conversation[:80]}...) knowledge extractor.extract_knowledge(conversation)for k_type in [facts, needs, processes, tips]: all_knowledge[k_type].extend(knowledge.get(k_type, [])) print(f ✅ 提取: {len(knowledge.get(facts, []))}条事实, f{len(knowledge.get(needs, []))}条需求, f{len(knowledge.get(tips, []))}条提示)# 步骤2合并相似知识 print(f\n步骤2知识合并与去重) merged_knowledge {}for k_type in [facts, needs, processes, tips]:if all_knowledge[k_type]: contents [k[content] for k in all_knowledge[k_type]] merged merger.merge_similar_knowledge(contents) merged_knowledge[k_type] merged print(f {k_type}: {len(contents)}条 → 合并后{len(merged.get(items, []))}条)# 步骤3更新知识库 print(f\n步骤3更新知识库) print(f ✅ 新增/更新知识完成)return merged_knowledge# 测试数据conversation_logs [用户: 创极速光轮现在排队多久客服: 当前排队约60分钟。创极速光轮身高要求122cm以上哦。用户: 刺激吗客服: 刺激程度较高是全球最快的迪士尼过山车。, 用户: 疯狂动物城有什么好玩的客服: 疯狂动物城有警察局互动体验适合5-12岁儿童无身高限制。用户: 排队要很久吗客服: 平日排队约20-30分钟周末可能需要40-50分钟。, 用户: 如果下雨创极速光轮会关闭吗客服: 大雨天气可能临时关闭建议关注官方APP实时通知。]# 执行流程result knowledge_deposition_pipeline(conversation_logs)8.4 知识质量评估提取的知识需要进行质量评估避免低质量知识污染知识库评估过程需要从多个维度综合考量确保知识的准确性、完整性、时效性、实用性和安全性。除了这些核心维度还需要考虑知识的适用性、可读性、合规性等方面。适用性确保知识符合目标用户的使用场景和认知水平可读性保证知识表达清晰易懂合规性检查知识内容是否符合相关法规和政策要求。classKnowledgeQualityEvaluator: 知识质量评估器defevaluate(self, knowledge_item):评估知识质量评估维度: 1. 准确性信息是否准确 2. 完整性信息是否完整 3. 时效性是否为过时信息 4. 重要性对用户的价值 返回: score: 0-1之间的分数 accept: 是否接受这条知识 reason: 评估理由 # 准确性检查 accuracy self._check_accuracy(knowledge_item)# 完整性检查 completeness self._check_completeness(knowledge_item)# 时效性检查 timeliness self._check_timeliness(knowledge_item)# 重要性评分 importance self._check_importance(knowledge_item)# 综合评分 score (accuracy * 0.4 completeness * 0.2 timeliness * 0.2 importance * 0.2 ) accept score 0.7# 阈值 return {score: score,accept: accept,dimensions: {accuracy: accuracy,completeness: completeness,timeliness: timeliness,importance: importance },reason: self._generate_reason(score, accept) }def_check_accuracy(self, item):# 检查是否包含数字、具体信息 # 实际应结合知识库验证 return 0.9 if any(char.isdigit() for char in item[content]) else 0.7 def _check_completeness(self, item):# 检查信息是否完整长度、结构 return min(len(item[content]) / 50, 1.0)def_check_timeliness(self, item):# 检查是否包含时间敏感词 temp_words [今天, 现在, 刚才, 临时] has_temp any(word in item[content] for word in temp_words)return0.5if has_temp else1.0def_check_importance(self, item):# 根据type和frequency判断重要性 if item.get(frequency) 高:return1.0elif item.get(frequency) 中:return0.7else:return0.5def_generate_reason(self, score, accept):if accept:returnf知识质量良好{score:.2f}建议采纳else:returnf知识质量不足{score:.2f}建议人工审核8.5 知识合并器classKnowledgeMerger: 知识合并器 - 消除重复和冲突def__init__(self, modelqwen-turbo-latest): self.model modeldefmerge_similar_knowledge(self, knowledge_list: List[Dict]) - Dict:合并相似知识点# 按内容相似度分组 groups self._group_similar_knowledge(knowledge_list) merged_results []for group in groups:if len(group) 1:# 单独知识点直接保留 merged_results.append(group[0])else:# 合并相似知识点 merged self._merge_group(group) merged_results.append(merged)return merged_resultsdef_group_similar_knowledge(self, knowledge_list: List[Dict]) - List[List[Dict]]:按相似度分组 groups []for knowledge in knowledge_list: placed Falsefor group in groups:if self._is_similar(knowledge, group[0]): group.append(knowledge) placed Truebreakifnot placed: groups.append([knowledge])return groupsdef_is_similar(self, knowledge1: Dict, knowledge2: Dict) - bool:判断两个知识点是否相似 content1 knowledge1[content] content2 knowledge2[content]# 简单相似度计算实际可用更复杂的算法 words1 set(jieba.cut(content1)) words2 set(jieba.cut(content2)) intersection words1 words2 union words1 | words2 similarity len(intersection) / len(union) if union else0return similarity 0.6第九章知识库健康检查9.1 什么是知识库健康度知识库和人是一样的也需要系统的体检来确保长期稳定的运行。知识库健康度是衡量知识库整体质量和可用性的综合指标反映了知识库满足用户需求的能力和持续发展的潜力。一个健康的知识库不仅要有丰富的内容覆盖还要保证信息的准确性、时效性和一致性同时具备良好的可维护性和可扩展性。知识库需要定期”体检”确保✅完整性核心知识是否覆盖✅时效性信息是否过时✅一致性是否存在矛盾✅可用性能否被有效检索9.2 健康度检查维度9.3 完整实现# # 知识库健康度检查系统# classKnowledgeBaseHealthChecker:知识库健康度检查器def__init__(self, knowledge_base, query_logs, modelqwen-turbo-latest): self.kb knowledge_base self.query_logs query_logs self.model modeldefcheck_completeness(self):检查完整性 print(\n *80) print( 完整性检查) print(*80)# 核心知识点列表预定义 core_topics [门票价格, 营业时间, 交通方式,热门项目介绍, 身高限制, 餐饮信息,住宿推荐, 特殊服务, 退改规则 ] covered_topics [] missing_topics []for topic in core_topics:# 在知识库中搜索该主题 found any(topic in chunk[content] for chunk in self.kb)if found: covered_topics.append(topic)else: missing_topics.append(topic) completeness_score len(covered_topics) / len(core_topics) print(f\n核心知识点覆盖率: {completeness_score:.1%}) print(f ✅ 已覆盖 ({len(covered_topics)}): {, .join(covered_topics[:5])}...)if missing_topics: print(f ❌ 缺失 ({len(missing_topics)}): {, .join(missing_topics)})return {score: completeness_score,covered: covered_topics,missing: missing_topics }defcheck_timeliness(self):检查时效性 print(\n *80) print(⏰ 时效性检查) print(*80)from datetime import datetime, timedelta outdated_threshold timedelta(days90) # 90天未更新视为过时 now datetime.now() outdated_chunks []for chunk in self.kb: last_update chunk.get(last_update)if last_update: update_time datetime.fromisoformat(last_update)if now - update_time outdated_threshold: outdated_chunks.append(chunk) timeliness_score 1 - (len(outdated_chunks) / len(self.kb)) print(f\n时效性得分: {timeliness_score:.1%}) print(f 总计: {len(self.kb)}个知识片段) print(f 过时: {len(outdated_chunks)}个 (超过90天未更新))if outdated_chunks: print(f\n 需要更新的知识:)for chunk in outdated_chunks[:3]: print(f • {chunk[content][:50]}...)return {score: timeliness_score,outdated_count: len(outdated_chunks),outdated_chunks: outdated_chunks }defcheck_consistency(self):检查一致性 print(\n *80) print( 一致性检查) print(*80)# 检测矛盾信息 contradictions self._detect_contradictions()# 检测重复知识 duplicates self._detect_duplicates() total_issues len(contradictions) len(duplicates) consistency_score 1 - min(total_issues / len(self.kb), 1.0) print(f\n一致性得分: {consistency_score:.1%}) print(f 矛盾信息: {len(contradictions)}处) print(f 重复知识: {len(duplicates)}处)if contradictions: print(f\n 发现矛盾:)for c in contradictions[:2]: print(f • {c[chunk1][:40]}... vs {c[chunk2][:40]}...)return {score: consistency_score,contradictions: contradictions,duplicates: duplicates }defcheck_usability(self):检查可用性 print(\n *80) print( 可用性检查) print(*80)# 使用历史Query测试检索效果 from rank_bm25 import BM25Okapiimport jieba docs [chunk[content] for chunk in self.kb] tokenized_docs [list(jieba.cut(doc)) for doc in docs] bm25 BM25Okapi(tokenized_docs) hit_count 0 total_queries min(len(self.query_logs), 100) #测试100个Quer for query in self.query_logs[:total_queries]: tokenized_query list(jieba.cut(query)) scores bm25.get_scores(tokenized_query) max_score max(scores) if len(scores) 0else0# 如果最高分数阈值认为检索成功 if max_score 1.0: hit_count 1 usability_score hit_count / total_queries if total_queries 0else0 print(f\n可用性得分: {usability_score:.1%}) print(f 测试Query数: {total_queries}) print(f 成功检索: {hit_count}) print(f 检索失败: {total_queries - hit_count})return {score: usability_score,total_queries: total_queries,hit_count: hit_count,miss_count: total_queries - hit_count }def_detect_contradictions(self):检测矛盾信息简化版# 实际应使用LLM进行语义对比 contradictions []# 示例查找价格相关矛盾 price_chunks [c for c in self.kb if价格in c[content] or元in c[content]]for i, c1 in enumerate(price_chunks):for c2 in price_chunks[i1:]:# 简单检查如果都提到门票但价格不同 if门票in c1[content] and门票in c2[content]:# 提取数字 import re nums1 re.findall(r\d, c1[content]) nums2 re.findall(r\d, c2[content])if nums1 and nums2 and nums1[0] ! nums2[0]: contradictions.append({chunk1: c1[content],chunk2: c2[content],reason: 门票价格不一致 })return contradictionsdef_detect_duplicates(self):检测重复知识from difflib import SequenceMatcher duplicates []for i, c1 in enumerate(self.kb):for j, c2 in enumerate(self.kb[i1:], i1): similarity SequenceMatcher(None, c1[content], c2[content]).ratio()if similarity 0.85: # 相似度85%认为重复 duplicates.append({chunk1: c1[content],chunk2: c2[content],similarity: similarity })return duplicatesdefgenerate_report(self):生成完整健康度报告 print(\n *80) print( 知识库健康度检查报告) print(*80)# 执行所有检查 completeness self.check_completeness() timeliness self.check_timeliness() consistency self.check_consistency() usability self.check_usability()# 计算综合得分 overall_score (completeness[score] * 0.3 timeliness[score] * 0.2 consistency[score] * 0.2 usability[score] * 0.3 ) print(\n *80) print( 综合评分) print(*80) print(f\n 总分: {overall_score:.1%}) print(f 完整性: {completeness[score]:.1%}) print(f 时效性: {timeliness[score]:.1%}) print(f 一致性: {consistency[score]:.1%}) print(f 可用性: {usability[score]:.1%})# 给出建议 print(\n *80) print( 优化建议) print(*80)if completeness[missing]: print(f\n 1. 补充缺失的核心知识点:)for topic in completeness[missing][:3]: print(f • {topic})if timeliness[outdated_count] 0: print(f\n 2. 更新{timeliness[outdated_count]}个过时的知识片段)if consistency[contradictions]: print(f\n 3. 解决{len(consistency[contradictions])}处矛盾信息)if usability[miss_count] usability[total_queries] * 0.3: print(f\n 4. 优化知识库结构提高检索成功率) print(\n *80)return {overall_score: overall_score,completeness: completeness,timeliness: timeliness,consistency: consistency,usability: usability }# 使用示例knowledge_base [ {id: 001, content: 创极速光轮身高要求122cm以上, last_update: 2024-01-15}, {id: 002, content: 门票价格平日399元, last_update: 2024-10-01}, {id: 003, content: 营业时间9:00-21:00, last_update: 2023-05-20}, # 过时]query_logs [门票多少钱,创极速光轮身高要求,今天开门吗,# ... 更多Query]checker KnowledgeBaseHealthChecker(knowledge_base, query_logs)report checker.generate_report()第十章知识库版本管理与AB测试10.1 为什么需要版本管理想象一个真实场景运营团队吭哧吭哧更新了迪士尼知识库新增了”疯狂动物城新区”的介绍但上线后发现用户满意度下降了5%。问题出在哪能否快速回滚如何科学地评估新旧版本的效果没有版本管理的情况下很难快速定位问题原因也无法立即恢复到之前的稳定状态。版本管理通过保存历史版本和变更记录使团队能够快速对比分析找出导致问题的具体变更。真实血泪教训# 没有版本管理的痛苦经历pain_points { 盲目更新: 一顿操作猛如虎上线直接二百五, 问题定位: 到底哪个改动搞砸了完全不知道, 回滚困难: 想回到昨天稳定版本抱歉没备份, 效果评估: 新版本到底好不好全靠感觉猜}10.2 版本管理架构知识库的时光机“后悔药”知识库版本管理架构设计需要平衡功能的完整性和实现的复杂性。基础版本V1.0作为起点包含知识库的初始状态。每次运营操作如新增知识点、修改内容或结构调整都会生成新的版本如V2.0在V1.0基础上新增了疯狂动物城相关内容。AB测试环节是版本管理的关键部分将用户流量分配到不同版本的知识库。跟前阵子有些人微信可以评论图片但有些人干着急也没到自己用新的版本。用户组A使用新版本V2.0用户组B继续使用旧版本V1.0确保测试的公平性和可比性。流量分配应该随机且一致同一用户在不同请求中应该使用相同版本的知识库。 指标收集系统监控各版本的运行效果包括用户满意度、检索召回率、答案准确率等关键指标。这些指标应该实时收集和分析为版本比较提供数据支持。指标系统需要确保数据的准确性和完整性避免偏差影响决策。 效果对比模块分析各版本的表现差异使用统计方法判断差异的显著性。对比应该考虑多个维度的指标综合评估版本的优劣。对于关键业务指标需要设置最小可检测效应确保测试的灵敏度。 决策环节基于对比结果决定下一步行动。如果新版本V2.0表现更好可以逐步扩大流量直至全量发布。如果表现不如旧版本需要回滚到V1.0并分析问题原因。回滚后的分析应该深入查找根本原因为后续改进提供输入。 改进方案基于问题分析结果制定可能涉及知识内容优化、检索策略调整或生成提示词改进等。改进后的知识库形成V3.0版本重新开始测试验证循环。这种迭代优化机制确保知识库的持续改进。 架构实现上版本管理系统需要与知识库存储、检索系统、用户界面等组件集成。版本切换应该对用户透明不影响使用体验。元数据管理记录每个版本的详细信息支持快速的版本查询和对比操作。10.3 版本管理核心要素给每次改动上户口一版本元数据设计每个版本需要记录的信息# 版本元数据设计version_metadata {version: v2.1.3, # 版本号create_time: 2024-11-10 14:30:00, # 创建时间author: 运营_张三, # 负责人change_type: 新增知识, # 变更类型change_summary: 新增疯狂动物城新区介绍包含5个新项目, # 变更摘要impact_scope: 主题园区模块, # 影响范围parent_version: v2.1.2, # 父版本基于哪个版本改的tags: [新区上线, 紧急发布], # 标签status: 灰度测试中, # 状态traffic_percentage: 20, # 流量百分比rollback_flag: False# 是否可快速回滚}二变更追踪Diff记录每次变更的具体内容类似Git# 变更记录Diff信息change_log {新增内容: [疯狂动物城新区占地XX平方米包含朱迪训练营、尼克冰淇淋店等,朱迪训练营适合7-12岁儿童互动体验项目,尼克狐冰淇淋店营业时间10:00-20:00 ],修改内容: [ {before: 疯狂动物城包含警察局互动体验,after: 疯狂动物城包含警察局互动体验、朱迪训练营、尼克冰淇淋店,reason: 补充新区信息 } ],删除内容: [ {content: 该区域正在建设中敬请期待,reason: 项目已完工信息过时 } ]}三版本比较维度如何科学地对比两个版本需要从多个维度评估10.4 AB测试实施方案用数据说话不靠拍脑门流量分配策略classGradualRolloutStrategy: 渐进式发布策略 - 不把鸡蛋放在一个篮子里defget_rollout_plan(self):获取发布计划return [ {stage: 内部尝鲜,traffic: 1, # 1%流量duration: 1-2天, focus: [错误率, 崩溃率], # 重点关注success_criteria: 错误率1%, # 通过标准risk_level: 极低# 风险等级 }, {stage: 小范围灰度, traffic: 10, # 10%流量duration: 2-3天,focus: [核心功能, 用户体验],success_criteria: 核心指标无显著下降,risk_level: 低 }, {stage: 大规模测试,traffic: 50, # 50%流量 duration: 3-7天,focus: [业务指标, 用户反馈],success_criteria: 关键指标正向提升,risk_level: 中 }, {stage: 全量发布,traffic: 100, # 100%流量duration: 持续监控,focus: [稳定性, 性能],success_criteria: 平稳运行一周,risk_level: 高 } ]用户分组策略确保AB测试的科学性用户分组需要考虑随机性用户ID哈希取模确保随机一致性同一用户始终看到同一版本隔离性不同组用户之间相互不影响defassign_knowledge_version(user_id: str, experiment_config: Dict) - str: 为用户分配知识库版本# 基于用户ID哈希确保同一用户始终看到同一版本 user_hash hash(user_id) % 100# 根据流量配置分配版本 traffic_ranges {v1.0: (0, experiment_config[v1_traffic]),v2.0: (experiment_config[v1_traffic], 100) }for version, (start, end) in traffic_ranges.items():if start user_hash end:return versionreturnv1.0 # 默认版本10.5 回归测试上线前的压力测试每次知识库更新都需要跑一遍回归测试集确保✅ 旧的核心Query仍然能正确回答✅ 没有引入新的错误✅ 性能没有显著下降回归测试集构建defbuild_smart_test_suite(): 构建智能回归测试集return {high_frequency: [门票多少钱, # 最高频问题营业时间几点, # 核心业务问题创极速光轮身高要求# 关键事实问题 ],historical_issues: [儿童票怎么买, # 曾经出过问题的快速通行证多少钱, # 用户经常搞错的下雨天哪些项目关闭# 容易混淆的 ],edge_cases: [, # 空查询啊啊啊啊, # 无意义输入a * 1000, # 超长输入门票#价格# 特殊字符 ],business_critical: [怎么去迪士尼, # 关键业务路径无新增的Critical错误带小孩能玩什么, # 核心用户场景 今天有什么活动# 实时信息需求 ] } 测试集来源1. 历史高频Query Top-100必测2. 曾经出错的Query回归测试3. 核心业务场景Query关键路径4. 边界Case如空Query、超长Query# 测试标准 - 通过率 ≥ 95%允许5%的边界Case失败 - 平均响应时间不超过旧版本的110% - 无新增的Critical错误 回归测试报告示例defgenerate_regression_report(test_results: List) - Dict: 生成智能回归测试报告# 分析测试结果 passed [r for r in test_results if r[status] passed] degraded [r for r in test_results if r[status] degraded] failed [r for r in test_results if r[status] failed]# 性能对比 performance_compare {avg_response_time: {current: 1.23,baseline: 1.15, change: 6.9%,acceptable: True },p95_response_time: {current: 2.45,baseline: 2.10,change: 16.7%, acceptable: False# P95超标需要关注 },error_rate: {current: 0.8,baseline: 0.5, change: 60%,acceptable: False# 错误率上升明显 } }# 生成建议 suggestions []if len(failed) 0: suggestions.append(修复失败用例后再发布)if performance_compare[p95_response_time][acceptable] False: suggestions.append(优化性能瓶颈降低P95响应时间)if performance_compare[error_rate][acceptable] False: suggestions.append(排查错误率上升原因)return {summary: f通过率: {len(passed)}/{len(test_results)} ({len(passed)/len(test_results):.1%}),risk_level: 中等风险if len(failed) 0else低风险,performance: performance_compare,critical_issues: failed[:3], # 只显示前3个关键问题suggestions: suggestions,release_recommendation: 建议修复问题后重新测试if len(failed) 0else可以进入灰度发布 }10.6 版本回滚机制知识库的安全气囊快速回滚三步走步骤1检测异常 - 实时监控系统指标错误率、响应时间 - 设置自动告警阈值 - 人工观察用户反馈步骤2决策回滚 - 触发条件错误率5% 或 用户投诉10条/小时 - 决策人技术负责人 运营负责人 - 执行时间发现问题后5分钟内步骤3执行回滚 - 切换知识库版本指针秒级完成 - 清理缓存确保旧版本生效 - 通知相关人员记录回滚原因classSmartRollbackManager: 智能回滚管理器 - 7x24小时保驾护航def__init__(self): self.alert_thresholds {error_rate: 0.05, # 错误率超过5%告警user_complaints: 10, # 每小时投诉超过10条response_time_p95: 3.0, # P95响应时间超过3秒satisfaction_drop: 0.1# 用户满意度下降10% }defmonitor_real_time_metrics(self, metrics: Dict) - Dict:实时监控指标 alerts [] should_rollback False# 检查各项指标if metrics[error_rate] self.alert_thresholds[error_rate]: alerts.append(f 错误率超标: {metrics[error_rate]:.1%}) should_rollback Trueif metrics[user_complaints] self.alert_thresholds[user_complaints]: alerts.append(f 用户投诉激增: {metrics[user_complaints]}条/小时) should_rollback Trueif metrics[response_time_p95] self.alert_thresholds[response_time_p95]: alerts.append(f 响应时间变慢: P95{metrics[response_time_p95]}s)if metrics[satisfaction_drop] self.alert_thresholds[satisfaction_drop]: alerts.append(f 用户满意度下降: {metrics[satisfaction_drop]:.1%}) should_rollback Truereturn {should_rollback: should_rollback,alerts: alerts,severity: criticalif should_rollback elsewarning }defexecute_emergency_rollback(self, target_version: str):执行紧急回滚 print(f 触发紧急回滚切换到版本: {target_version})# 三步回滚法 steps [1. 切换知识库版本指针,2. 清理所有相关缓存, 3. 验证回滚效果 ]for step in steps: print(f {step})# 实际执行回滚操作 time.sleep(0.5) # 模拟操作耗时 print(✅ 回滚完成系统已恢复稳定)回滚后的复盘defconduct_post_mortem(rollback_event: Dict): 事故复盘 - 不甩锅只改进return {复盘时间: 回滚后24小时内,参与人员: [技术负责人, 产品经理, 运营同学, 测试同学],讨论议题: [技术层面到底是哪行代码/哪个配置搞砸了,流程层面为什么测试没发现流程有什么漏洞, 业务层面影响了多少用户业务损失多大,改进层面怎么避免类似问题再次发生 ],输出成果: {根本原因: 新知识片段格式不规范导致解析异常,改进措施: [️ 短期修复问题补充测试用例, 中期完善发布checklist增加自动化检查,️ 长期建设知识质量检测平台 ],经验沉淀: 所有知识入库前必须经过格式校验和语义检查,后续计划: 一周内完成所有改进措施 } }总结经过前10章的系统学习我们已经共同完成了一场从RAG基础认知到企业级架构设计的完整技术进阶。这段旅程让我们从最初理解RAG为何能够解决AI瞎编问题开始逐步掌握了构建智能问答系统的核心技术体系。 技术进阶全景图基础筑基1-3章✅ 理解了RAG的核心价值告别AI瞎编让回答有据可查✅ 掌握了三大核心步骤索引构建 → 精准检索 → 可信生成✅ 亲手实现了ChatPDF原型体验了从0到1的成就感核心技术突破4-6章 Query改写让AI听懂用户的黑话和潜台词 联网搜索打破知识壁垒获取实时最新信息 混合检索Rerank从大概相关到精准命中的质变系统智能化7-10章 知识自进化让知识库在服务中不断学习成长 健康度检查建立知识库的定期体检体系 版本管理AB测试像互联网产品一样科学迭代读者福利倘若大家对大模型感兴趣那么这套大模型学习资料一定对你有用。针对0基础小白如果你是零基础小白快速入门大模型是可行的。大模型学习流程较短学习内容全面需要理论与实践结合学习计划和方向能根据资料进行归纳总结包括大模型学习线路汇总、学习阶段大模型实战案例大模型学习视频人工智能、机器学习、大模型书籍PDF。带你从零基础系统性的学好大模型有需要的小伙伴可以保存图片到wx扫描二v码免费领取【保证100%免费】AI大模型学习路线汇总大模型学习路线图整体分为7个大的阶段全套教程文末领取哈第一阶段从大模型系统设计入手讲解大模型的主要方法第二阶段在通过大模型提示词工程从Prompts角度入手更好发挥模型的作用第三阶段大模型平台应用开发借助阿里云PAI平台构建电商领域虚拟试衣系统第四阶段大模型知识库应用开发以LangChain框架为例构建物流行业咨询智能问答系统第五阶段大模型微调开发借助以大健康、新零售、新媒体领域构建适合当前领域大模型第六阶段以SD多模态大模型为主搭建了文生图小程序案例第七阶段以大模型平台应用与开发为主通过星火大模型文心大模型等成熟大模型构建大模型行业应用。大模型实战案例光学理论是没用的要学会跟着一起做要动手实操才能将自己的所学运用到实际当中去这时候可以搞点实战案例来学习。大模型视频和PDF合集这里我们能提供零基础学习书籍和视频。作为最快捷也是最有效的方式之一跟着老师的思路由浅入深从理论到实操其实大模型并不难。学会后的收获• 基于大模型全栈工程实现前端、后端、产品经理、设计、数据分析等通过这门课可获得不同能力• 能够利用大模型解决相关实际项目需求大数据时代越来越多的企业和机构需要处理海量数据利用大模型技术可以更好地处理这些数据提高数据分析和决策的准确性。因此掌握大模型应用开发技能可以让程序员更好地应对实际项目需求• 基于大模型和企业数据AI应用开发实现大模型理论、掌握GPU算力、硬件、LangChain开发框架和项目实战技能学会Fine-tuning垂直训练大模型数据准备、数据蒸馏、大模型部署一站式掌握• 能够完成时下热门大模型垂直领域模型训练能力提高程序员的编码能力大模型应用开发需要掌握机器学习算法、深度学习框架等技术这些技术的掌握可以提高程序员的编码能力和分析能力让程序员更加熟练地编写高质量的代码。获取方式有需要的小伙伴可以保存图片到wx扫描二v码免费领取【保证100%免费】