企业知识库 RAG 案例：答案质量取决于文档治理

# ❌ 错误做法：检索后再过滤权限，文档已经进了上下文
results = vector_store.similarity_search(query, k=10)
filtered = [r for r in results if user.has_permission(r.metadata['dept'])]
# 问题：如果 10 条里有 6 条被过滤掉，有效上下文只剩 4 条
# 更严重：日志里已经留下了越权文档的访问记录

# ✅ 正确做法：权限过滤下推到检索层，用 pre-filter 缩小范围
results = vector_store.similarity_search(
    query,
    k=10,
    filter={
        # 用户只能检索到自己角色允许的文档
        'access_level': {'$in': user.allowed_levels},
        'dept': {'$in': user.dept_scope},
        'status': 'active',  # 排除已归档文档
    }
)
# 好处：从源头控制可见范围，不会泄漏到上下文
# 日志也只记录权限范围内的检索行为

# ❌ 错误做法：所有文档统一固定长度切分
from langchain.text_splitter import RecursiveCharacterTextSplitter
 
splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    # 不区分文档类型，一律按 token 数切
    # 合同条款、操作手册、FAQ 被同样对待
)
chunks = splitter.split_text(document_text)
# 问题：完整条款可能在 chunk 边界被切断

# ✅ 正确做法：按文档类型选择切分策略
from langchain.text_splitter import RecursiveCharacterTextSplitter
 
def chunk_by_type(doc):
    if doc.type in ['contract', 'spec', 'regulation']:
        # 规范类文档：大窗口 + 语义分隔符优先
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=150,  # 约 15% overlap，NVIDIA 实验验证效果好
            separators=['\n## ', '\n### ', '\n\n', '\n', '。', '；'],
            # 优先按标题切，再按段落，最后按句子
        )
    elif doc.type in ['faq', 'notice', 'memo']:
        # 简短文档：小窗口保持简洁
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=400,
            chunk_overlap=60,
            separators=['\n\n', '\n', '。'],
        )
    else:
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=600,
            chunk_overlap=90,
            separators=['\n\n', '\n', '。'],
        )
 
    chunks = splitter.split_text(doc.content)
 
    # 关键：每个 chunk 都带上来源元数据
    for i, chunk in enumerate(chunks):
        chunk.metadata.update({
            'source': doc.title,
            'section': doc.current_section,
            'page': doc.page_number,
            'updated_at': doc.updated_at.isoformat(),
            'chunk_index': i,
        })
 
    return chunks

# ❌ 错误做法：生成回答时不附加引用信息
def generate_answer(query, context_chunks):
    context = '\n'.join([c.text for c in context_chunks])
    # 只把文本拼起来喂给模型，丢失了所有来源信息
    response = llm.generate(f"""
    基于以下信息回答问题：
    {context}
    问题：{query}
    """)
    return response  # 用户看到一段话，但不知道出处

# ✅ 正确做法：chunk 入库时保留引用链路，回答时附带来源
def generate_answer_with_citation(query, context_chunks):
    # 构建带引用标记的上下文
    context_parts = []
    for i, chunk in enumerate(context_chunks):
        ref = (
            f"[引用{i+1}] 来源：{chunk.metadata['source']}，"
            f"章节：{chunk.metadata.get('section', '未知')}，"
            f"页码：{chunk.metadata.get('page', '未知')}，"
            f"更新时间：{chunk.metadata.get('updated_at', '未知')}"
        )
        context_parts.append(f"{ref}\n{chunk.text}")
 
    context = '\n\n'.join(context_parts)
 
    response = llm.generate(f"""
    基于以下信息回答问题。回答中引用信息时标注 [引用N]。
    如果信息不足以回答，明确说明「当前知识库未找到相关规定」。
 
    {context}
 
    问题：{query}
    """)
 
    # 附上引用摘要
    citations = [
        {
            'ref': f'[引用{i+1}]',
            'source': c.metadata['source'],
            'section': c.metadata.get('section'),
            'page': c.metadata.get('page'),
            'updated_at': c.metadata.get('updated_at'),
        }
        for i, c in enumerate(context_chunks)
    ]
 
    return {'answer': response, 'citations': citations}

# ❌ 错误做法：单路向量检索，无权限过滤，无引用
def simple_retrieve(query: str, vector_store) -> str:
    docs = vector_store.similarity_search(query, k=5)
    context = '\n'.join(d.text for d in docs)
    return llm.generate(f"基于以下信息回答：{context}\n问题：{query}")
    # 问题：
    # 1. 没有权限过滤，可能越权
    # 2. 只有向量检索，精确术语匹配不到
    # 3. 没有重排序，Top-5 可能有噪声
    # 4. 没有引用，用户无法核实

# ✅ 正确做法：混合检索 + 权限预过滤 + Rerank + 引用溯源
from typing import TypedDict
 
class RetrievedChunk(TypedDict):
    text: str
    source: str
    section: str
    page: int
    updated_at: str
    score: float
 
def retrieve_with_governance(
    query: str,
    user,
    vector_store,
    bm25_index,
    reranker,
) -> list[RetrievedChunk]:
 
    # 第一步：构建权限过滤条件
    permission_filter = {
        'access_level': {'$in': user.allowed_levels},
        'dept_scope': {'$in': user.dept_scope},
        'status': 'active',
    }
 
    # 第二步：双路召回 —— 语义向量 + 关键词 BM25
    vector_results = vector_store.similarity_search(
        query, k=20, filter=permission_filter
    )
    keyword_results = bm25_index.search(
        query, k=20, pre_filter=permission_filter
    )
 
    # 第三步：融合去重（RRF，Reciprocal Rank Fusion）
    merged = reciprocal_rank_fusion(vector_results, keyword_results)
 
    # 第四步：Rerank 精排，取 Top-5
    reranked = reranker.rerank(query, merged, top_k=5)
 
    # 第五步：组装引用信息
    chunks = []
    for i, doc in enumerate(reranked):
        chunks.append(RetrievedChunk(
            text=doc.text,
            source=doc.metadata.get('source', '未知'),
            section=doc.metadata.get('section', '未知'),
            page=doc.metadata.get('page', 0),
            updated_at=doc.metadata.get('updated_at', '未知'),
            score=doc.score,
        ))
 
    return chunks
    # 返回结果直接可以用于生成带引用的回答

# ❌ 错误做法：只记录反馈，不分类不处理
def log_feedback(query: str, answer: str, is_helpful: bool):
    db.insert({
        'query': query,
        'answer': answer,
        'is_helpful': is_helpful,
        # 只有一个布尔值，没法定位问题根因
        # 半年后积累了上万条反馈，没人知道哪些该优先处理
    })

# ✅ 正确做法：反馈自动分类，驱动不同治理动作
from enum import Enum
 
class FailureType(Enum):
    RETRIEVAL_MISS = 'retrieval_miss'     # 检索不到相关内容
    DOC_OUTDATED = 'doc_outdated'         # 文档已过期
    OUT_OF_SCOPE = 'out_of_scope'         # 问题超出知识库范围
    CHUNK_SPLIT_BAD = 'chunk_split_bad'   # 切分不合理
    PERMISSION_BLOCK = 'permission_block' # 有内容但用户无权看
 
def classify_and_route_feedback(
    query: str,
    answer: str,
    feedback: str,  # 用户填写的具体反馈
    retrieved_chunks: list,
    classifier,  # LLM 或规则分类器
):
    # 第一步：自动分类失败原因
    failure_type = classifier.classify(
        query=query,
        answer=answer,
        user_feedback=feedback,
        retrieved_context=[c.text for c in retrieved_chunks],
    )
 
    # 第二步：根据类型路由到不同处理流程
    match failure_type:
        case FailureType.RETRIEVAL_MISS:
            # 可能切分粒度有问题，或者缺少相关文档
            task_queue.enqueue('review_chunking', query=query)
        case FailureType.DOC_OUTDATED:
            # 触发文档更新流程，通知文档负责人
            doc_id = retrieved_chunks[0].metadata['doc_id']
            notify_owner(doc_id, action='review_and_update')
        case FailureType.OUT_OF_SCOPE:
            # 标记知识空白区，推动内容补充
            task_queue.enqueue('knowledge_gap', query=query)
        case FailureType.CHUNK_SPLIT_BAD:
            # 重新切分对应文档
            doc_id = retrieved_chunks[0].metadata['doc_id']
            task_queue.enqueue('re_chunk', doc_id=doc_id)
        case FailureType.PERMISSION_BLOCK:
            # 检查权限配置是否合理
            task_queue.enqueue('review_permission', query=query)
 
    # 第三步：记录闭环状态
    db.insert({
        'query': query,
        'failure_type': failure_type.value,
        'status': 'pending',
        'assigned_to': get_handler(failure_type),
    })