RAG 系统建设

import uuid
import numpy as np
​
# 存储完整父 Chunk 的 NoSQL/KV 库
parent_document_store = {}
# 存储子 Chunk 向量的向量数据库
vector_database = []
​
def mock_get_embedding(text: str) -> list:
    np.random.seed(abs(hash(text)) % (10 ** 8))
    return np.random.rand(128).tolist()  # 模拟128维向量
​
def ingest_document(raw_text: str):
    # 1. 存入完整的父 Chunk
    parent_id = str(uuid.uuid4())
    parent_document_store[parent_id] = raw_text
​
    # 2. 切分成高内聚的子 Chunk 并做向量化
    child_chunks = [
        "大模型在2026年迎来了爆发式增长。",
        "企业级RAG系统成为了各家公司落地AI的核心技术方案。",
        "通过多模态解析和父子分块，可以大幅度降低RAG的幻觉率。"
    ]
​
    for child_text in child_chunks:
        child_id = str(uuid.uuid4())
        child_vector = mock_get_embedding(child_text)
        
        vector_record = {
            "id": child_id,
            "vector": child_vector,
            "text": child_text,
            "parent_id": parent_id  # 关联父 ID
        }
        vector_database.append(vector_record)
​
def retrieve_and_generate(user_query: str):
    query_vector = np.array(mock_get_embedding(user_query))
    
    # 计算最匹配的子 Chunk
    results = []
    for record in vector_database:
        record_vec = np.array(record["vector"])
        sim = np.dot(query_vector, record_vec) / (np.linalg.norm(query_vector) * np.linalg.norm(record_vec))
        results.append((sim, record))
    results.sort(key=lambda x: x[0], reverse=True)
    
    best_match_record = results[0][1]
    
    # 通过 parent_id 取出完整的父 Chunk
    target_parent_id = best_match_record["parent_id"]
    full_context = parent_document_store[target_parent_id]
    
    return full_context

[ 100万个原始文本块 ]
        │
        ├── (BM25 检索)     ──── 召回 Top-100
        ├── (向量检索)       ──── 召回 Top-100
        │
        ▼
   [ RRF 融合排序 ]   ──── 合并并筛选 Top-30
        │
        ▼
 [ Rerank 模型精排 ]  ──── 输出高精度 Top-5 给 LLM

from collections import defaultdict
​
# 模拟两路召回的文档 ID 排名（从高到低）
bm25_ranked_list = ["doc1", "doc2", "doc3"]
vector_ranked_list = ["doc2", "doc3", "doc1"]
​
def reciprocal_rank_fusion(ranked_lists: list, k: int = 60):
    """
    RRF 算法：
    Score(d) = sum( 1 / (k + rank_d) )
    """
    rrf_scores = defaultdict(float)
    for r_list in ranked_lists:
        for rank, doc_id in enumerate(r_list, start=1):
            rrf_scores[doc_id] += 1.0 / (k + rank)
    
    # 按照分数由高到低重新排序
    sorted_docs = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_docs
​
# 执行融合
merged_results = reciprocal_rank_fusion([bm25_ranked_list, vector_ranked_list], k=60)
print("RRF 融合结果:", merged_results)

def ushape_reorder(chunks: list) -> list:
    """
    输入降序排列的 Chunks，重排为 U 型（两头得分最高，中间得分最低）
    """
    sorted_chunks = sorted(chunks, key=lambda x: x["score"], reverse=True)
    reordered = [None] * len(sorted_chunks)
    left, right = 0, len(sorted_chunks) - 1
    
    for i, chunk in enumerate(sorted_chunks):
        if i % 2 == 0:
            reordered[left] = chunk
            left += 1
        else:
            reordered[right] = chunk
            right -= 1
    return reordered
​
def build_enterprise_prompt(query: str, ordered_chunks: list) -> str:
    context_str = "<search_results>\n"
    for chunk in ordered_chunks:
        context_str += f'  <doc id="{chunk["doc_id"]}">\n'
        context_str += f'    <content>{chunk["text"]}</content>\n'
        context_str += "  </doc>\n"
    context_str += "</search_results>"
    
    prompt = f"""你是一个严谨的企业知识库助手。请根据下面给出的 <search_results> 中的内容回答用户的问题。
规则：
1. 只能根据 <search_results> 中提及的事实回答，若背景知识不足则直接说“抱歉，根据公司现有文档无法回答”，切勿编造。
2. 句末必须强制标注参考的文档ID（例如：“李雷在研发部[doc_1]。”）。
​
以下是背景知识：
{context_str}
​
当前用户提问：{query}
回答："""
    return prompt

import numpy as np
​
# 模拟向量缓存库
semantic_cache_store = []
SIMILARITY_THRESHOLD = 0.92
​
def get_embedding(text: str) -> list:
    np.random.seed(abs(hash(text)) % (10 ** 8))
    return np.random.rand(64).tolist()
​
# ----------------- 语义缓存判定 -----------------
def check_and_get_cache(new_query: str):
    new_vector = np.array(get_embedding(new_query))
    
    for cache_entry in semantic_cache_store:
        cached_vector = np.array(cache_entry["query_vector"])
        sim = np.dot(new_vector, cached_vector) / (np.linalg.norm(new_vector) * np.linalg.norm(cached_vector))
        
        if sim >= SIMILARITY_THRESHOLD:
            return cache_entry["answer"]
    return None
​
# ----------------- 前置权限过滤检索 -----------------
mock_vector_db = [
    {"id": "doc_hr_01", "text": "高层薪资标准...", "vector": get_embedding("薪资标准"), "allowed_departments": ["HR"]},
    {"id": "doc_tech_01", "text": "Git分支规范...", "vector": get_embedding("Git规范"), "allowed_departments": ["HR", "R&D"]}
]
​
def search_with_acl(user_query: str, user_dept: str):
    query_vector = np.array(get_embedding(user_query))
    valid_results = []
    
    for doc in mock_vector_db:
        # 权限判定 (Pre-filtering)
        if user_dept in doc["allowed_departments"]:
            doc_vec = np.array(doc["vector"])
            sim = np.dot(query_vector, doc_vec) / (np.linalg.norm(query_vector) * np.linalg.norm(doc_vec))
            valid_results.append((sim, doc))
            
    valid_results.sort(key=lambda x: x[0], reverse=True)
    return valid_results

                    [ 用户的提问 ]
                     /          \
      答案相关性(QA)                上下文相关性(QC)
                   /              \
        [ 生成的答案 ]  ─────── [ 检索出的背景 ]
                        忠实度(AC)

import json
​
def evaluate_faithfulness(retrieved_context: str, generated_answer: str) -> float:
    # 模拟发送给裁判大模型（如 GPT-4）的 Prompt 与输出解析
    # 裁判模型将生成的回答拆解为陈述，判断是否能从上下文推导
    judge_output_mock = {
        "statements_and_verdicts": [
            {"statement": "李雷在研发部上班。", "verdict": "Yes"},
            {"statement": "他需要在周五主持技术评审会。", "verdict": "No"}  # 属于上下文未提及的幻觉
        ]
    }
    
    verdicts = judge_output_mock["statements_and_verdicts"]
    total = len(verdicts)
    yes_count = sum(1 for v in verdicts if v["verdict"] == "Yes")
    
    return yes_count / total if total > 0 else 0.0
​
# 自动化评估输出得分 0.50
print("Faithfulness 得分:", evaluate_faithfulness("...", "..."))