一.架构(Architecture)1.1 首次回答正确多次调用检索跑偏、答非所问#原代码 from langchain.messages import AIMessage ... last_query request.state[messages][-1].textrequest.state[messages]是完整对话历史LangGraph state 的 messages 列表会累积。你现在的测试是单轮列表里只有 1 条 user 消息所以[-1]碰巧取对。一旦多轮对话列表变成轮次messages 列表内容[-1]取到的是第 1 轮[HumanMessage(langchain安装…)]用户问题 ✅第 2 轮[Human, AI回答1, Human(pip怎么升级)]用户新问题 ✅第 2 轮AI 内部再次调用模型时[Human, AI回答1, Human, AI含tool_call…]上一轮 AI 的回复❌检索 query 变成 AI 自己上一句回答 → 召回错误文档 → 注入错误上下文 → 答非所问。而且这个 bug 是静默的——不报错只看结果不太对劲。#修改后 from langchain.messages import AIMessage, HumanMessage ... user_msgs [m for m in request.state[messages] if isinstance(m, HumanMessage)] last_query user_msgs[-1].text if user_msgs else 1.2 dynamic_prompt 模型调用单轮没问题加了工具或多轮导致API 调用 变慢、烧钱解决检索结果缓存或改用 Agentic 架构按需检索原代码修复后retriever在prompt_with_context内部每次调用都新建提到模块级只建一次检索逻辑retriever.invoke(last_query)裸调用每次模型调用都打一次 embedding API抽成retrieve_docs(query)lru_cache(maxsize128)同一 query 只检索一次空 query直接拿空串去检索embedding 一个空字符串兜底返回“无检索结果”检索打印每次都打印缓存命中时会误导移进retrieve_docs只在真实检索时打印——顺便成了缓存命中观察器虚拟环境激活CMD:.venv\Scripts\activate.batPowerShell:Activate.ps1Git Bashsource .venv\Scripts\activate二.RAG检索优化2.1 查询优化这三种都是检索前优化2_Step/Agentic 决定“检索怎么进流程”query_Optimize 决定“检索词质量”2.1.1 temperature0.7高温下复现困难、JSON 解析概率性失败#原代码 temperature0.7改写/拆解要的是稳定高温下同一问题每次改写结果不同Multi_hop 的拆解数量和 JSON 格式都飘#修改后 model ChatOpenAI(..., temperature0.7) # 最终回答保留多样性 rewrite_model ChatOpenAI(..., temperature0) # 改写/检索要稳定 rewrite_middleware QueryRewriteMiddleware(vectorstore, rewrite_model) # 用低温的 rewrite_agent create_agent(modelmodel, ...) # 回答用 0.7 的2.1.2 文档未去重导致后续拼上下文时重复内容膨胀原代码的两层问题第一层N 个子问题 × k3同一文档会被多个子问题召回 → 拼上下文时重复膨胀token 浪费、模型还可能误以为该文档特别重要。第二层setdefaultbugretrieved_docs.setdefault(sub_query, docs)setdefault的语义是键已存在就不动。原代码修复后数据结构dict[sub_query, docs]——按子问题去重dict[doc_id, (doc, [命中的子问题])]——按文档去重去重粒度只能处理两个子问题字符串完全相同处理两个不同子问题召回了同一篇文档这才是真实场景去重 keysub_query 字符串d.idku.py 里设的doc_X无则用内容前缀兜底附加信息只有 docs每文档记录被几个子问题命中多跳相关度信号输出按子问题分组打印重复文档混在里头打印合并前后数量对比 去重后按排名打印如果 LLM 拆出来的两个子问题字符串恰好相同第二次检索的 docs 会被整个丢弃不是合并。这种情况下你既丢了文档又不知道丢了——比重复更阴险。#修改后 for sub_query in sub_queries: for d in retriever.invoke(sub_query): key d.id or d.page_content[:32] if key not in retrieved_docs: retrieved_docs[key] (d, [sub_query]) # 首次见到该文档 else: retrieved_docs[key][1].append(sub_query) # 已见过只追加命中来源2.2 检索优化Dense 靠意思像Sparse 靠字面对WS 是两路分数加权投票RRF 是两路排名投票。RRF 更省心不用管分数方向和量纲WS 更灵活能调权重偏向某一路。2.2.1 检索结果对不上新文档且不报错#原代码 def create_bm25_index(metadata_corpus, index_pathSAMPLE_PATH, k11.5, b0.75): if os.path.exists(index_path): # ← 只看路径在不在 print(f索引文件已存在正在加载:{index_path}) retriever bm25s.BM25.load(index_path, load_corpusTrue) else: print(f未找到索引文件正在创建新索引...) corpus_tokens [jieba.lcut(doc[content]) for doc in metadata_corpus] retriever bm25s.BM25(k1k1, bb, corpusmetadata_corpus) retriever.index(corpus_tokens) retriever.save(index_path) print(f索引已保存至: {index_path}) return retriever # ← 无论哪条路都静默返回不报错维度说明检测逻辑os.path.exists只判断路径存不存在不看内容对不对触发场景你改了sample.md增/删/改文档但没手动删my_index.bm25/目录后果旧索引照样加载成功BM25 检索到的是改之前的文档结果对不上新语料且不报任何错误隐蔽性不报错 你根本不知道结果过期了debug 时方向完全跑偏#修改后 import hashlib # ← 新增 def corpus_hash(corpus: List[Dict]) - str: 生成语料库内容哈希指纹 content json.dumps(corpus, ensure_asciiFalse, sort_keysTrue) return hashlib.md5(content.encode()).hexdigest() def create_bm25_index(metadata_corpus, index_pathSAMPLE_PATH, k11.5, b0.75): current_hash corpus_hash(metadata_corpus) # ← 当前语料的指纹 hash_path str(index_path) .hash # ← 指纹文件路径 # 三重检查索引文件 hash 文件 hash 匹配 if os.path.exists(index_path) and os.path.exists(hash_path): with open(hash_path, r, encodingutf-8) as f: stored_hash f.read().strip() if stored_hash current_hash: # ✅ 索引存在 语料没变 → 直接复用 print(f索引文件已存在且语料未变更正在加载: {index_path}) return bm25s.BM25.load(index_path, load_corpusTrue) else: # ⚠️ 语料变了 → hash 不匹配重建 print(f语料已变更hash 不匹配重建索引...) # 重建索引 保存新 hash corpus_tokens [jieba.lcut(doc[content]) for doc in metadata_corpus] retriever bm25s.BM25(k1k1, bb, corpusmetadata_corpus) retriever.index(corpus_tokens) retriever.save(index_path) with open(hash_path, w, encodingutf-8) as f: f.write(current_hash) # ← 写入当前语料的指纹 print(f索引已保存至: {index_path}) return retrieveros.path.exists一个条件 →索引文件 hash 文件 hash 匹配三个条件全满足才复用否则一律重建。2.2.2 归一化失真 类型不一致#原代码 # 向量分不归一化 ranked_results weighted_sum_fusion([vector_rs, bm25_rs], [0.5, 0.5], [False, True]) # raw_docs 混存 Document 和 dict for doc, score in vector_results: raw_docs[doc.id] doc # Document 对象 for doc, score in bm25_results: raw_docs[doc[id]] doc # dict不是 Documentnormalized[False, True]向量相似度分范围约 01BM25 分范围约 030。不归一化向量分0.5 × 0.8 0.5 × 25 12.9向量分几乎被 BM25 淹没加权毫无意义。raw_docs混存 Document 对象和 dict最终print(raw_docs[doc_id])输出格式不一致。#修改后 # 两个检索器都先归一化到 [0,1] 再加权 ranked_results weighted_sum_fusion([vector_rs, bm25_rs], [0.5, 0.5], [True, True]) # BM25 返回的 dict 统一转 Document for doc, score in bm25_results: doc_id doc[id] if doc_id not in raw_docs: raw_docs[doc_id] Document(iddoc_id, page_contentdoc[content]) bm25_rs[doc_id] score2.3 重排序MMR 解决结果太像Cross-Encoder 解决排得不对。两者可以叠加——先用 MMR 保证多样性再用 Cross-Encoder 精排。2.3.1 Cross_Encoder.py 测试 query 你好不触发工具调用#原代码 res rag_agent.stream( {messages:[{role:user,content:你好}]}, stream_modemessages )system_prompt 说如果用户问了你不确定的问题或者涉及教资专业知识的问题你必须使用search_knowledge_base工具。#修改后 res rag_agent.stream( {messages:[{role:user,content:给教师的建议是谁写的}]}, stream_modemessages )三.高级RAG Agent汇总查询改写 多路召回 RRF融合 Cross-Encoder 组合成一个完整的高级RAG Agent。阶段做什么关键组件① 并行检索Dense BM25 同时跑各取 k3vectorstore.similarity_searchbm25_search② 融合RRF 按排名合并两路结果reciprocal_rank_fusion(k60)③ 精排Cross-Encoder 逐对 (query, doc) 打分过滤 取 top_kbge-reranker-v2-m3④ AgentLLM 拿到工具返回的文档内容生成最终回答create_agenttoolLangChain 生态做编排DashScope 做云端 LLM Embedding本地 sentence-transformers bm25s jieba 做检索与重排三层模型分工明确——便宜的在云端、重的在本地、纯算法的自己写。