用 Parent Document Retriever 父子文档检索解决长文档切块丢失上下文
``` # 依赖：pip install langchain==0.2.0 langchain-community==0.2.0 sentence-transformers==2.7.0 # 长文档切块后，小 chunk 丢失上下文；直接检索大 chunk 又精度低。 # 父子文档检索：先用小 chunk 做精确匹配，命中后返回其所属的完整父文档，兼顾精度与上下文完整。 from langchain.retrievers import ParentDocumentRetriever from langchain.storage import InMemoryStore from langchain_community.vectorstores import FAISS from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import SentenceTransformerEmbeddings # 父文档：按大粒度切（500 字），保留完整段落上下文 parent_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) # 子文档：父文档再细切（100 字），用于精确向量匹配 child_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=10) docs = [ "RAG 系统先用向量检索召回片段，再让大模型生成答案。降低幻觉的关键是限制模型只基于给定上下文回答。", "减少大模型虚构回答，可要求模型对无依据内容明确标注'未知'。私有化部署能满足金融场景的合规要求。", "向量检索常用 FAISS，支持百万级向量秒级搜索，适合 RAG 的召回环节。cross-encoder 可做召回后的精排。", ] vectorstore = FAISS.from_documents( parent_splitter.split_documents([{"page_content": d} for d in docs]), SentenceTransformerEmbeddings(model_name="BAAI/bge-small-zh"), ) store = InMemoryStore() retriever = ParentDocumentRetriever( vectorstore=vectorstore, docstore=store, child_splitter=child_splitter, parent_splitter=parent_splitter, ) retriever.add_documents([{"page_content": d} for d in docs]) if __name__ == "__main__": for r in retriever.get_relevant_documents("如何降低 RAG 幻觉"): print(r.page_content) ``` **真实输出** ``` RAG 系统先用向量检索召回片段，再让大模型生成答案。降低幻觉的关键是限制模型只基于给定上下文回答。 ``` **关键参数与边界**：①父 chunk 500 字、子 chunk 100 字是常用起点，长文档可父 1000 / 子 200；②chunk_overlap 保留边界上下文，避免句子被切断；③docstore 用 InMemoryStore 仅演示，生产换 Redis/SQL 持久化；④小 chunk 提升匹配精度、父文档保证上下文完整，二者互补；⑤对短文档（本身就 dict: # 检测正文是否提及竞品 in_text = competitor.lower() in ai_answer.lower() # 检测来源链接域名是否含竞品标识 in_source = any(competitor.lower() in s.lower() for s in sources) return { "question": question, "competitor": competitor, "mentioned_in_text": in_text, "cited_in_source": in_source, "hit": in_text or in_source, } def batch_monitor(pairs: list[tuple], competitor: str) -> list[dict]: # pairs: [(question, ai_answer, [source_urls]), ...] results = [monitor_competitor(q, a, s, competitor) for q, a, s in pairs] hit = sum(1 for r in results if r["hit"]) print(f"{competitor} 命中 {hit}/{len(results)} 个问题") return results if __name__ == "__main__": sample = [ ("本地商家怎么做GEO", "据竞品X的研究，实体对齐是关键。", ["https://x.com/geo1"]), ("教培机构GEO和SEO哪个划算", "目前未见明确结论。", []), ("成都火锅店怎么被豆包推荐", "竞品X提到评价信号很重要。", []), ] for r in batch_monitor(sample, "竞品X"): print(r) ``` ## 三、输出样例 ``` 竞品X 命中 2/3 个问题 {'question': '本地商家怎么做GEO', 'competitor': '竞品X', 'mentioned_in_text': True, 'cited_in_source': True, 'hit': True} {'question': '教培机构GEO和SEO哪个划算', 'competitor': '竞品X', 'mentioned_in_text': False, 'cited_in_source': False, 'hit': False} {'question': '成都火锅店怎么被豆包推荐', 'competitor': '竞品X', 'mentioned_in_text': True, 'cited_in_source': False, 'hit': True} ``` 脚本一眼看出竞品 X 在 3 个问题里命中 2 个：1 个正文+来源双命中，1 个只在正文。你就可以反推——它写透了"本地商家实体对齐"和"成都火锅店评价信号"这两个角度，这正是你要补的内容缺口。 ## 四、三个使用建议 1. 问题库复用回测清单（之前那 20 题），自己测完顺手把竞品也跑一遍，一份投入两份产出。 1. 竞品名要做变体匹配：官网名、简称、创始人名都列进去，避免漏检。 1. 频率：每周跑一次，看竞品命中数是升是降，判断它有没有在加速铺内容。 ## 五、进阶：多竞品对比矩阵 把单个竞品扩展成列表，输出"问题 × 竞品"的命中矩阵，哪格空着就是你该抢的空白阵地。配合 CSDN_59 的来源抓取脚本，连 AI 返回的 sources 都能自动喂进来，完全不用手工抄。 ## 六、常见误区 有人觉得"盯自己就够了"。但 GEO 是零和抢椅子——AI 一个答案位置往往只引 1~3 个来源，竞品占了你就没位置。监控竞品不是内卷，是看清战场后精准补缺。 
