用 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 系统先用向量检索召回片段，再让大模型生成答案。降低幻觉的关键是限制模型只基于给定上下文回答。

import re

def monitor_competitor(question: str, ai_answer: str, sources: list[str], competitor: str) -> 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}