│ │ 1_simple_rag.ipynb
│ ── 2_semantic_chunking.ipynb
│
── 9_rse.ipynb
│ ── 10_contextual_compression.ipynb
│ ── 11_feedback_loop_rag.ipynb
│ ── 12_adaptive_rag.ipynb │
──
17_graph_rag.ipynb
│ ── 18_hierarchy_rag.ipynb
│ ── 19_HyDE_rag.ipynb
│ ── 20_crag.ipynb
└── data/
    └── val.json
    └── AI_information.pdf
    └──tention_is_all_you_need.pdf

test query:
How does AI’s reliance on massive data sets act as a double-edged sword?

True Answer:
It drives rapid learning and innovation while also
risking the amplification of inherent biases,
making it crucial to balance data volume with fairness and quality.

# Cloning the repo
git clone https://github.com/FareedKhan-dev/all-rag-techniques.git
cd all-rag-techniques

# Installing the required libraries
pip install -r requirements.txt

# 定义 PDF 文件的路径
pdf_path = "data/AI_information.pdf"

# 从 PDF 文件中提取文本，并创建更小、重叠的块。extracted_text
extracted_text = extract_text_from_pdf(pdf_path)
text_chunks = chunk_text(extracted_text, 1000, 200)

print("Number of text chunks:", len(text_chunks))

### OUTPUT ###
Number of text chunks: 42

# 为文本块创建嵌入
response = create_embeddings(text_chunks)

# 我们的测试查询，并执行语义搜索。query
query = '''How does AI's reliance on massive data sets act
           as a double-edged sword?'''
top_chunks = semantic_search(query, text_chunks, embeddings, k=2)

# 定义 AI 助手的系统提示
system_prompt = "You are an AI assistant that strictly answers based on the given context. If the answer cannot be derived directly from the provided context, respond with: 'I do not have enough information to answer that.'"

# 根据顶部块创建用户提示，并生成 AI 响应。
user_prompt = "\n".join([f"Context {i + 1}:\n{chunk}\n========\n" for i, chunk in enumerate(top_chunks)])
user_prompt = f"{user_prompt}\nQuestion: {query}"
ai_response = generate_response(system_prompt, user_prompt)
print(ai_response.choices[0].message.content)

# 定义评估系统的系统提示
evaluate_system_prompt = "You are an intelligent evaluation system tasked with assessing the AI assistant's responses. If the AI assistant's response is very close to the true response, assign a score of 1. If the response is incorrect or unsatisfactory in relation to the true response, assign a score of 0. If the response is partially aligned with the true response, assign a score of 0.5."

# 创建评估提示并生成评估响应
evaluation_prompt = f"User Query: {query}\nAI Response:\n{ai_response.choices[0].message.content}\nTrue Response: {data[0]['ideal_answer']}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)

### OUTPUT ###
... Therefore, the score of 0.3 being not very close to the
true response, and not perfectly aligned.

# 将文本拆分成句子（基本拆分）
sentences = extracted_text.split(". ")

# 为每个句子生成
embeddings = [get_embedding(sentence) for sentence in sentences]

print(f"Generated {len(embeddings)} sentence embeddings.")

### OUTPUT ###
233

# 计算连续句子之间的相似度
similarities = [cosine_similarity(embeddings[i], embeddings[i + 1]) for i in range(len(embeddings) - 1)]

# 使用百分位数法计算断点，阈值为 90
breakpoints = compute_breakpoints(similarities, method="percentile", threshold=90)

# 使用 split_into_chunks 函数创建块
text_chunks = split_into_chunks(sentences, breakpoints)
print(f"Number of semantic chunks: {len(text_chunks)}")

### OUTPUT ###
Number of semantic chunks: 145

# 使用 create_embeddings 函数创建块嵌入
chunk_embeddings = create_embeddings(text_chunks)

# Create the user prompt based on the top chunks
user_prompt = "\n".join([f"Context {i + 1}:\n{chunk}\n=====================================\n" for i, chunk in enumerate(top_chunks)])
user_prompt = f"{user_prompt}\nQuestion: {query}"

# Generate AI response
ai_response = generate_response(system_prompt, user_prompt)
print(ai_response.choices[0].message.content)

# 结合用户查询、AI 响应、真实响应和评估系统提示创建评估提示
evaluation_prompt = f"User Query: {query}\nAI Response:\n{ai_response.choices[0].message.content}\nTrue Response: {data[0]['ideal_answer']}\n{evaluate_system_prompt}"

# 使用评估系统提示和评估提示生成评估响应
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)

# 打印评估响应
print(evaluation_response.choices[0].message.content)

### OUTPUT
Based on the evaluation criteria,
I would assign a score of 0.2 to the AI assistant response.

def  context_enriched_search ( query, text_chunks, embeddings, k= 1 , context_size= 1 ): 
    """
    检索最相关的块及其相邻块。
    """ 
    # 将查询转换为嵌入向量
    query_embedding = create_embeddings(query).data[ 0 ].embedding 
    similarity_scores = [] 

    # 计算查询与每个文本块嵌入之间的相似度分数
    for i, chunk_embedding in  enumerate (embeddings): 
        # 计算查询嵌入与当前块嵌入之间的余弦相似度
        similarity_score = cosine_similarity(np.array(query_embedding), np.array(chunk_embedding.embedding)) 
        # 将索引和相似度分数存储为元组
        similarity_scores.append((i, similarity_score)) 

    # 按降序对相似度分数进行排序（相似度最高者优先）
     similarity_scores.sort(key= lambda x: x[ 1 ], reverse= True ) 

    # 获取最相关块的索引
    top_index = similarity_scores[ 0 ][ 0 ] 

    # 定义上下文包含的范围
    # 确保不低于 0 或超出 text_chunks 的长度
    start = max ( 0 , top_index - context_size) 
    end = min ( len (text_chunks), top_index + context_size + 1 ) 

    # 返回相关块及其相邻的上下文块
    return [text_chunks[i] for i in  range (start, end)]

# Create the user prompt based on the top chunks
user_prompt = "\n".join([f"Context {i + 1}:\n{chunk}\n=====================================\n" for i, chunk in enumerate(top_chunks)])
user_prompt = f"{user_prompt}\nQuestion: {query}"

# Generate AI response
ai_response = generate_response(system_prompt, user_prompt)
print(ai_response.choices[0].message.content)

# Create the evaluation prompt and generate the evaluation response
evaluation_prompt = f"User Query: {query}\nAI Response:\n{ai_response.choices[0].message.content}\nTrue Response: {data[0]['ideal_answer']}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)

### OUTUT ###
Based on the evaluation criteria,
I would assign a score of 0.6 to the AI assistant response.

# 将提取的文本分块，这次生成标题
text_chunks_with_headers = chunk_text_with_headers(extracted_text, 1000, 200)

# 打印一个样本以查看它是什么样子
print("Sample Chunk with Header:")
print("Header:", text_chunks_with_headers[0]['header'])
print("Content:", text_chunks_with_headers[0]['text'])

### OUTPUT ###
Sample Chunk with Header:
Header: A Description about AI Impact
Content: AI has been an important part of society since ...

# 为每个块生成嵌入（标题和文本）
embeddings = []
for chunk in tqdm(text_chunks_with_headers, desc="Generating embeddings"):
    text_embedding = create_embeddings(chunk["text"])
    header_embedding = create_embeddings(chunk["header"])
    embeddings.append({"header": chunk["header"], "text": chunk["text"], "embedding": text_embedding, "header_embedding": header_embedding})

# 使用查询和新的嵌入执行语义搜索
top_chunks = semantic_search(query, embeddings, k=2)

# 根据顶部块创建用户提示。注意：无需添加标题
# 因为上下文已经使用标题和块创建
user_prompt = "\n".join([f"Context {i + 1}:\n{chunk['text']}\n=====================================\n" for i, chunk in enumerate(top_chunks)])
user_prompt = f"{user_prompt}\nQuestion: {query}"

# Generate AI response
ai_response = generate_response(system_prompt, user_prompt)
print(ai_response.choices[0].message.content)

### OUTPUT ###
Evaluation Score: 0.5

# 处理文档（提取文本、创建块、生成问题、构建向量存储）
text_chunks, vector_store = process_document(
    pdf_path,
    chunk_size=1000,
    chunk_overlap=200,
    questions_per_chunk=3
)

print(f"Vector store contains {len(vector_store.texts)} items")

### OUTPUT ###
Vector store contains 214 items

# 执行语义搜索，找到相关内容
search_results = semantic_search(query, vector_store, k=5)

print("Query:", query)
print("\nSearch Results:")

# 按类型组织结果
chunk_results = []
question_results = []

for result in search_results:
    if result["metadata"]["type"] == "chunk":
        chunk_results.append(result)
    else:
        question_results.append(result

# 从搜索结果准备上下文
context = prepare_context(search_results) 

# 生成响应
response_text = generate_response(query, context) 

# 从验证数据中获取参考答案
reference_answer = data[ 0 ][ 'ideal_answer' ] 

# 评估响应
evaluation = evaluate_response(query, response_text, reference_answer)

print("\nEvaluation:")
print(evaluation)

### OUTPUT ###
Based on the evaluation criteria, I would assign a
score of 0.8 to the AI assistants response.

# 查询重写
rewritten_query = rewrite_query(query)

# 后退提示
step_back_query = generate_step_back_query(query)

# 子查询分解
sub_queries = decompose_query(query, num_subqueries=4)

def rag_with_query_transformation(pdf_path, query, transformation_type=None):
    """
    使用可选的查询转换运行完整的 RAG 管道。Args 

    ：
        pdf_path（str）：PDF 文档的路径
        query（str）：用户查询
        transformation_type（str）：转换类型（无、'rewrite'、'step_back' 或 'decompose'）

    返回：
        Dict：包括查询、转换后的查询、上下文和响应的结果
    """ 
    # 处理文档以创建向量存储
    vector_store = process_document(pdf_path) 

    # 应用查询转换和搜索
    if transformation_type：
        # 使用转换后的查询执行搜索
        results = formed_embeddings(query, vector_store, transformation_type) 
    else：
        # 执行不进行转换的常规搜索
        query_embeddings = create_embeddings(query) 
        results = vector_store.similarity_search(query_embedding, k= 3 ) 

    # 从搜索结果中合并上下文
    context = "\n\n" .join([ f"PASSAGE {i+ 1 } :\n {result[ 'text' ]} "  for i, result in  enumerate (results)]) 

    # 根据查询和组合上下文生成响应
    response = generate_response(query, context) 

    # 返回包括原始查询、转换类型、上下文和响应的结果
    return {
        "original_query": query,
        "transformation_type": transformation_type,
        "context": context,
        "response": response
    }

# Run evaluation
evaluation_results = evaluate_transformations(pdf_path, query, reference_answer)
print(evaluation_results)

### OUTPUT ###
Evaluation Score: 0.5

def  rag_with_reranking ( query, vector_store, reranking_method= "llm" , top_n= 3 , model= "meta-llama/Llama-3.2-3B-Instruct" ): 
    """
    完成包含重新排名的 RAG 管道。
    """ 
    # 创建查询嵌入
    query_embedding = create_embeddings(query) 
    
    # 初始检索（获取比重新排名所需更多的数据）
     initial_results = vector_store.similarity_search(query_embedding, k= 10 ) 
    
    # 应用重新排名
    if reranking_method == "llm" : 
        reranked_results = rerank_with_llm(query, initial_results, top_n=top_n) 
    elif reranking_method == "keywords" : 
        reranked_results = rerank_with_keywords(query, initial_results, top_n=top_n) # 我们不使用它。
    else : 
        # 不重新排名，只使用初始检索中的顶级结果
        reranked_results = initial_results[:top_n] 
    
    # 结合重新排名结果中的上下文
    context = "\n\n===\n\n" .join([result[ "text" ] for result in reranked_results]) 
    
    # 根据上下文生成响应
    response = generate_response(query, context, model) 
    
    return { 
        "query" : query, 
        "reranking_method" : reranking_method, 
        "initial_results" : initial_results[:top_n], 
        "reranked_results" : reranked_results, 
        "context" : context, 
        "response" : response 
    }

# 使用基于 LLM 的重新排名运行 RAG
llm_reranked_result = rag_with_reranking(query, vector_store, reranking_method="llm")

# Evaluate.
evaluation_prompt = f"User Query: {query}\nAI Response:\n{llm_reranked_result['response']}\nTrue Response: {reference_answer}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)

### OUTPUT ###
Evaluation score is 0.7

# Run RAG with RSE
rse_result = rag_with_rse(pdf_path, query)

# Evaluate
evaluation_prompt = f"User Query: {query}\nAI Response:\n{rse_result['response']}\nTrue Response: {reference_answer}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)

### OUTPUT ###
However, the Response from Standard Retrieval includes ...
0.8 is the score I would assign to AI Response

def  rag_with_compression ( pdf_path, query, k= 10 , compression_type= "selective" , model= "meta-llama/Llama-3.2-3B-Instruct" ): 
    """
    带有上下文压缩的 RAG（检索增强生成）管道。

    参数：
        pdf_path（str）：PDF 文档的路径。query 
        （str）：用于检索的用户查询。k 
        （int）：要检索的相关块数。默认值为 10。compression_type 
        （str）：应用于检索到的块的压缩类型。默认值为“selective”。model 
        （str）：用于生成响应的语言模型。默认值为“meta-llama/Llama-3.2-3B-Instruct”。

    返回：
        dict：包含查询、原始和压缩块、压缩统计数据和最终响应的字典。
    """ 
    
    print ( f"\n=== RAG WITH COMPRESSION ===\nQuery: {query} | 压缩：{compression_type} " ) 
    
    # 处理文档以提取、分块和嵌入文本
    vector_store = process_document(pdf_path) 
    
    # 根据查询相似度检索前 k 个相关块
    results = vector_store.similarity_search(create_embeddings(query), k=k) 
    removed_chunks = [r[ "text" ] for r in results] 

    # 对检索到的块应用压缩
    compressed = batch_compress_chunks(retrieved_chunks, query, compression_type, model) 
    
    # 过滤掉空的压缩块；如果所有都是空的，则回退到原始
    compressed_chunks, compression_ratios = zip ([(c, r) for c, r in compressed if c.strip()] or [(chunk, 0.0 ) for chunk in removed_chunks]) 
    
    # 组合压缩块以形成用于生成响应的上下文
    context = "\n\n---\n\n" .join(compressed_chunks) 
    
    # 使用压缩上下文生成响应 response
     = generate_response(query, context, model) 

    print ( f"\n=== RESPONSE ===\n {response} " ) 
    
    # 返回详细结果
    return { 
        "query" : query, 
        "original_chunks" : removed_chunks, 
        "compressed_chunks" : compressed_chunks, 
        "compression_ratios" :压缩比，
        “context_length_reduction”：f“ {总和（压缩率）/ len（压缩率）：. 2 f}％”，
        “响应”：响应
    }

# Run RAG with contextual compression (using 'selective' mode)
compression_result = rag_with_compression(pdf_path, query, compression_type="selective")

# Evaluate.
evaluation_prompt = f"User Query: {query}\nAI Response:\n{compression_result['response']}\nTrue Response: {reference_answer}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)

### OUTPUT ###
Evaluation Score 0.75

def  full_rag_workflow ( pdf_path, query, feedback_data= None , feedback_file= "feedback_data.json" , fine_tune= False ): 
    """
    执行带有反馈集成的完整 RAG 工作流程以持续改进。

    """ 
    # 步骤 1：如果未明确提供，则加载历史反馈以进行相关性调整
    if feedback_data is  None : 
        feedback_data = load_feedback_data(feedback_file) 
        print ( f"Loaded { len (feedback_data)} feedback entry from {feedback_file} " ) 
    
    # 步骤 2：通过提取、分块和嵌入管道处理文档
    chunks, vector_store = process_document(pdf_path) 
    
    # 步骤 3：通过合并高质量的过去交互来微调向量索引
    # 这将从成功的问答对中创建增强的可检索内容
    if fine_tune and feedback_data: 
        vector_store = fine_tune_index(vector_store, chunks, feedback_data) 
    
    # 步骤 4：使用反馈感知检索
    # 注意：这取决于 rag_with_feedback_loop 函数，该函数应在其他地方定义
    result = rag_with_feedback_loop(query, vector_store, feedback_data) 
    
    # 步骤 5：收集用户反馈以提高未来的性能
    print ( “\n=== 您想对此回复提供反馈吗？ ===" ) 
    print ( "评价相关性（1-5，5 为最相关）：" ) 
    relevance = input () 
    
    print ( "评价质量（1-5，5 为最优质）：" ) 
    quality = input () 
    
    print ( "有任何评论吗？（可选，按 Enter 跳过）" ) 
    comments = input () 
    
    # 步骤 6：将反馈格式化为结构化数据
    feedback = get_user_feedback( 
        query=query, 
        response=result[ "response" ], 
        relevance= int (relevance), 
        quality= int (quality), 
        comments=comments 
    ) 
    
    # 步骤 7：保留反馈以实现持续的系统学习
    store_feedback(feedback, feedback_file) 
    print ( "反馈已记录。谢谢！" ) 
    
    return result

# 我们没有之前的反馈，因此“fine_tune=False”
 result = full_rag_workflow(pdf_path=pdf_path, query=query, fine_tune= False ) 

# 评估。evaluation_prompt
 = f“用户查询：{query} \nAI 响应：\n {result[ 'response' ]} \nTrue 响应：{reference_answer} \n {evaluate_system_prompt} ”
 evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt) 
print (evaluation_response.choices[ 0 ].message.content) 

### OUTPUT ###
评估分数为 0.7，因为......

def  rag_with_adaptive_retrieval ( pdf_path, query, k= 4 , user_context= None ): 
    """
    使用自适应检索完成 RAG 管道。

    """ 
    print ( "\n=== RAG WITH ADAPTIVE RETRIEVAL ===" ) 
    print ( f"Query: {query} " ) 
    
    # 处理文档以提取文本、对其进行分块并创建嵌入
    chunks, vector_store = process_document(pdf_path) 
    
    # 对查询进行分类以确定其类型
    query_type = classify_query(query) 
    print ( f"Query classified as: {query_type} " ) 
    
    # 根据查询类型使用自适应检索策略检索文档
    retrieved_docs = adaptive_retrieval(query, vector_store, k, user_context) 
    
    # 根据查询、检索到的文档和查询类型生成响应
    response = generate_response(query, retained_docs, query_type) 
    
    # 编译结果放入字典中
    result = {
        "query": query,
        "query_type": query_type,
        "retrieved_documents": retrieved_docs,
        "response": response
    }

    print("\n=== RESPONSE ===")
    print(response)

    return result

# 运行自适应 RAG 管道
result = rag_with_adaptive_retrieval(pdf_path, query)

# Evaluate.
evaluation_prompt = f"User Query: {query}\nAI Response:\n{result['response']}\nTrue Response: {reference_answer}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)

### OUTPUT ###
Evaluation score is 0.86

def  determine_if_retrieval_needed ( query ): 
    """ 
    (说明性示例 - 并非完全功能)
    确定给定查询是否需要检索。
    """
     system_prompt = """您是 AI 助手，可确定是否需要检索来回答查询。
    对于事实问题、具体信息请求或有关事件、人物或概念的问题，请回答“是”。
    对于意见、假设情景或具有常识的简单查询，请回答“否”。
    仅回答“是”或“否”。"""

     user_prompt = f"查询：{query} \n\n是否需要检索才能准确回答此查询？"

     response = client.chat.completions.create( 
        model= "meta-llama/Llama-3.2-3B-Instruct" , 
        messages=[ 
            { "role" : "system" , "content" : system_prompt}, 
            { "role" : "user" , "content" : user_prompt} 
        ], 
        temperature= 0
     ) 
    answer = response.choices[ 0 ].message.content.strip().lower()
    return "yes" in answer

def  assess_relevance ( query, context ): 
    """ 
    (说明性示例 - 并非完全功能)
    评估上下文与查询的相关性。
    """
     system_prompt = """您是 AI 助手。确定文档是否与查询相关。
    仅用“相关”或“不相关”回答。"""

     user_prompt = f"""查询：{query}
    文档内容：
    {context[: 500 ]} ... [truncated]

    此文档是否与查询相关？仅用“相关”或“不相关”回答。
    """

    response = client.chat.completions.create(
        model="meta-llama/Llama-3.2-3B-Instruct",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0
    )
    answer = response.choices[0].message.content.strip().lower()
    return answer

# 我们可以调用 `self_rag` 函数进行自我整理，它会自动
# 决定何时检索，何时不检索。result
result = self_rag(query, vector_store)

print(result["response"])

### OUTPUT ###
Evaluation score for the AI Response is 0.65

def  build_knowledge_graph ( chunks ): 
    """
    使用嵌入和概念提取从文本块构建知识图谱。

    参数：
        chunks（字典列表）：文本块列表，每个块包含一个“text”字段。

    返回：
        tuple：（以节点为文本块的图，嵌入列表）
    """
     graph, texts = nx.Graph(), [c[ "text" ] for c in chunks] 
    embeddings = create_embeddings(texts)   # 计算嵌入

    # 添加带有提取概念和嵌入的节点
    for i, (chunk, emb) in  enumerate ( zip (chunks, embeddings)): 
        graph.add_node(i, text=chunk[ "text" ], ideas := extract_concepts(chunk[ "text" ]), embedding=emb) 

    # 根据共享概念和嵌入相似性创建边
    for i, j in ((i, j) for i in range(len(chunks)) for j in range(i + 1, len(chunks))):
        if shared_concepts := set(graph.nodes[i]["concepts"]) & set(graph.nodes[j]["concepts"]):
            sim = np.dot(embeddings[i], embeddings[j]) / (np.linalg.norm(embeddings[i])  np.linalg.norm(embeddings[j]))
            weight = 0.7 * sim + 0.3 * (len(shared_concepts) / min(len(graph.nodes[i]["concepts"]), len(graph.nodes[j]["concepts"])))
            if weight > 0.6:
                graph.add_edge(i, j, weight=weight, similarity=sim, shared_concepts=list(shared_concepts))

    print(f"Graph built: {graph.number_of_nodes()} nodes, {graph.number_of_edges()} edges")
    return graph, embeddings

def graph_rag_pipeline(pdf_path, query, chunk_size=1000, chunk_overlap=200, top_k=3):
		"""
    从文档到答案的完整 Graph RAG 管道。
    """ 
    # 从 PDF 文档中提取文本
    text = extract_text_from_pdf(pdf_path) 
    
    # 将提取的文本拆分为重叠块
    chunks = chunk_text(text, chunk_size, chunk_overlap) 
    
    # 从文本块构建知识图谱
    graph, embeddings = build_knowledge_graph(chunks) 
    
    # 遍历知识图谱以查找与查询相关的信息
    related_chunks, traversal_path = traverse_graph(query, graph, embeddings, top_k) 
    
    # 根据查询和相关块生成响应
    response = generate_response(query, related_chunks) 
        
    # 返回查询、响应、相关块、遍历路径和图形
    return {
        "query": query,
        "response": response,
        "relevant_chunks": relevant_chunks,
        "traversal_path": traversal_path,
        "graph": graph
    }

# 执行 Graph RAG 管道来处理文档并回答查询
results = graph_rag_pipeline(pdf_path, query)

# Evaluate.
evaluation_prompt = f"User Query: {query}\nAI Response:\n{results['response']}\nTrue Response: {reference_answer}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)

### OUTPUT
0.78

def hierarchical_rag(query, pdf_path, chunk_size=1000, chunk_overlap=200,
                     k_summaries=3, k_chunks=5, regenerate=False):
    """
    完成分层检索增强生成 (RAG) 管道。

    参数：
        query (str)：用户查询。pdf_path 
        (str)：PDF 文档的路径。chunk_size 
        (int)：要处理的文本块的大小。chunk_overlap 
        (int)：连续块之间的重叠。k_summaries 
        (int)：要检索的顶级摘要的数量。k_chunks 
        (int)：每个摘要要检索的详细块数。regenerate 
        (bool)：是否重新处理文档。

    返回：
        dict：包含查询、生成的响应、检索到的块
              以及摘要和详细块的计数。
    """ 
    # 定义用于缓存摘要和详细向量存储的文件名
    summary_store_file = f"{os.path.basename(pdf_path)}_summary_store.pkl"
    detailed_store_file = f"{os.path.basename(pdf_path)}_detailed_store.pkl"

    # 如果需要重新生成或缓存文件丢失，则处理文档，
    if regenerate or not os.path.exists(summary_store_file) or not os.path.exists(detailed_store_file):
        print("Processing document and creating vector stores...")
        summary_store, detailed_store = process_document_hierarchically(pdf_path, chunk_size, chunk_overlap)

        # Save processed stores for future use
        with open(summary_store_file, 'wb') as f:
            pickle.dump(summary_store, f)
        with open(detailed_store_file, 'wb') as f:
            pickle.dump(detailed_store, f)
    else:
        # Load existing vector stores from cache
        print("Loading existing vector stores...")
        with open(summary_store_file, 'rb') as f:
            summary_store = pickle.load(f)
        with open(detailed_store_file, 'rb') as f:
            detailed_store = pickle.load(f)

    # 使用分层搜索检索相关块
    retrieved_chunks = retrieve_hierarchically(query, summary_store, detailed_store, k_summaries, k_chunks)

    # 根据检索到的块生成响应
    response = generate_response(query, retrieved_chunks)

    # 返回带有元数据的结果
    return {
        "query": query,
        "response": response,
        "retrieved_chunks": retrieved_chunks,
        "summary_count": len(summary_store.texts),
        "detailed_count": len(detailed_store.texts)
    }

# Run the hierarchical RAG pipeline
result = hierarchical_rag(query, pdf_path)

# Evaluate.
evaluation_prompt = f"User Query: {query}\nAI Response:\n{result['response']}\nTrue Response: {reference_answer}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)

### OUTPUT
0.84

def generate_hypothetical_document(query, desired_length=1000):
    """
    生成一个假设文档来回答查询。
    """ 
    """您是专家文档创建者。
    给定一个问题，生成一份详细的文档来直接回答这个问题。
    文档长度应约为{desired_length}个字符，并提供对该问题的深入、
    翔实的答案。写作时要像这份文件来自
    该主题的权威来源一样。包括具体的细节、事实和解释。
    不要提到这是一份假设文件 - 只需直接写出内容即可。""" 
    # 定义系统提示以指导模型如何生成文档
    system_prompt = f"""You are an expert document creator.
    Given a question, generate a detailed document that would directly answer this question.
    The document should be approximately {desired_length} characters long and provide an in-depth,
    informative answer to the question. Write as if this document is from an authoritative source
    on the subject. Include specific details, facts, and explanations.
    Do not mention that this is a hypothetical document - just write the content directly."""

    # 使用查询定义用户提示
    user_prompt = f"Question: {query}\n\nGenerate a document that fully answers this question:"

    # 向 OpenAI API 发出请求以生成假设文档
    response = client.chat.completions.create(
        model="meta-llama/Llama-3.2-3B-Instruct",  # Specify the model to use
        messages=[
            {"role": "system", "content": system_prompt},  # System message to guide the assistant
            {"role": "user", "content": user_prompt}  # User message with the query
        ],
        temperature=0.1  # Set the temperature for response generation
    )

    # 返回生成的文档内容
    return response.choices[0].message.content

def hyde_rag(query, vector_store, k=5, should_generate_response=True):
    """
    使用假设文档嵌入执行 RAG。
    """ 
    print(f"\n=== Processing query with HyDE: {query} ===\n")

    # 步骤 1：生成一个回答查询的假设文档
    print("Generating hypothetical document...")
    hypothetical_doc = generate_hypothetical_document(query)
    print(f"Generated hypothetical document of {len(hypothetical_doc)} characters")

    # 步骤 2：为假设文档创建嵌入
    print("Creating embedding for hypothetical document...")
    hypothetical_embedding = create_embeddings([hypothetical_doc])[0]

    # 步骤 3：根据假设文档检索相似的块
    print(f"Retrieving {k} most similar chunks...")
    retrieved_chunks = vector_store.similarity_search(hypothetical_embedding, k=k)

    # 准备结果字典
    results = {
        "query": query,
        "hypothetical_document": hypothetical_doc,
        "retrieved_chunks": retrieved_chunks
    }

    # 步骤 4：如果要求，生成响应
    if should_generate_response:
        print("Generating final response...")
        response = generate_response(query, retrieved_chunks)
        results["response"] = response

    return results

# Run HyDE RAG
hyde_result = hyde_rag(query, vector_store)

# Evaluate.
evaluation_prompt = f"User Query: {query}\nAI Response:\n{hyde_result['response']}\nTrue Response: {reference_answer}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)

### OUTPUT
0.5

import numpy as np

def fusion_retrieval(query, chunks, vector_store, bm25_index, k=5, alpha=0.5):
    """通过结合基于向量和 BM25 的搜索结果执行融合检索。""" 
    
    # 为查询生成嵌入
    query_embedding = create_embeddings(query) 

    # 执行向量搜索并将结果存储在字典中（索引 -> 相似度得分）
     vector_results = { 
        r[ "metadata" ][ "index" ]: r[ "similarity" ] 
        for r in vector_store.similarity_search_with_scores(query_embedding, len (chunks)) 
    } 

    # 执行 BM25 搜索并将结果存储在字典中（索引 -> BM25 得分）
     bm25_results = { 
        r[ "metadata" ][ "index" ]: r[ "bm25_score" ] 
        for r in bm25_search(bm25_index, chunks, query, len (chunks)) 
    } 

    # 从向量存储中检索所有文档
    all_docs = vector_store.get_all_documents() 

    # 使用向量和 BM25 分数的加权和计算每个文档的综合分数
    scores = [ 
        (i, alpha * vector_results.get(i, 0 ) + ( 1 - alpha) * bm25_results.get(i, 0 )) 
        for i in  range ( len (all_docs)) 
    ] 

    # 按综合分数降序对文档进行排序，并保留前 k 个结果
    top_docs = sorted (scores, key= lambda x: x[ 1 ], reverse= True )[:k] 

    # 返回包含文本、元数据和综合分数的前 k 个文档
    return [
        {"text": all_docs[i]["text"], "metadata": all_docs[i]["metadata"], "score": s}
        for i, s in top_docs
    ]

# 首先，处理文档以创建块、向量存储和 BM25 索引
chunks,vector_store,bm25_index = process_document(pdf_path) 

# 使用融合检索运行 RAG
 fusion_result = answer_with_fusion_rag(query,chunks,vector_store,bm25_index) 
print (fusion_result[ "response" ]) 

# 评估。evaluate_prompt
 = f"User Query: {query} \nAI Response:\n {fusion_result[ 'response' ]} \nTrue Response: {reference_answer} \n {evaluate_system_prompt} "
 evaluate_response =generate_response(evaluate_system_prompt,evaluation_prompt) 
print (evaluation_response.choices[ 0 ].message.content) 

### OUTPUT
Evaluation score for AI Response is 0.83

def process_document(pdf_path, chunk_size=1000, chunk_overlap=200):
    """
    为多模态 RAG 处理文档。

    """ 
    # 为提取的图像创建目录
    image_dir = "extracted_images"
     os.makedirs(image_dir, exist_ok= True ) 
    
    # 从 PDF 中提取文本和图像
    text_data, image_paths = extract_content_from_pdf(pdf_path, image_dir) 
    
    # 对提取的文本进行分
    块 chunked_text = chunk_text(text_data, chunk_size, chunk_overlap) 
    
    # 处理提取的图像以生成标题
    image_data = process_images(image_paths) 
    
    # 组合所有内容项（文本块和图像标题）
     all_items = chunked_text + image_data 
    
    # 提取用于嵌入的内容
    contents = [item[ "content" ] for item in all_items] 
    
    # 为所有内容创建嵌入
    print ( "Creating embeddings for all content..." ) 
    embeddings = create_embeddings(contents) 
    
    # 构建向量存储并添加带有嵌入的项目
    vector_store = MultiModalVectorStore() 
    vector_store.add_items(all_items, embeddings) 
    
    # 准备包含文本块和图像标题计数的文档信息
    doc_info = { 
        “text_count” : len (chunked_text), 
        “image_count” : len (image_data), 
        “total_items” : len (all_items), 
    } 
    
    # 打印添加项目的摘要
    print(f"Added {len(all_items)} items to vector store ({len(chunked_text)} text chunks, {len(image_data)} image captions)")

    # 返回向量存储和文档信息
    return vector_store, doc_info

# 处理文档以创建向量存储。我们为此创建了一个新的 PDF
 pdf_path = "data/attention_is_all_you_need.pdf"
 vector_store, doc_info = process_document(pdf_path) 

# 运行多模态 RAG 管道。这与之前非常相似！
 result = query_multimodal_rag(query,vector_store) 

# 评估。evaluation_prompt
evaluation_prompt = f"User Query: {query}\nAI Response:\n{result['response']}\nTrue Response: {reference_answer}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)

### OUTPUT
0.79

# Run CRAG
crag_result = rag_with_compression(pdf_path, query, compression_type="selective")

# Evaluate.
evaluation_prompt = f"User Query: {query}\nAI Response:\n{crag_result['response']}\nTrue Response: {reference_answer}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)

### OUTPUT ###
0.824