agentic-parallelism/
    ├── 01_parallel_tool_use.ipynb
    ├── 02_parallel_hypothesis.ipynb
    ...
    ├── 06_competitive_agent_ensembles.ipynb
    ├── 07_agent_assembly_line.ipynb
    ├── 08_decentralized_blackboard.ipynb
    ...
    ├── 13_parallel_context_preprocessing.ipynb
    └── 14_parallel_multi_hop_retrieval.ipynb

from langchain_core.pydantic_v1 import BaseModel, Field
from typing import List

class ExpandedQueries(BaseModel):
    """Pydantic 模型定义了一组不同的扩展查询，以提高检索召回率"""
    # 生成的假设文档段落，在语义上与可能的答案相似
    hypothetical_document: str = Field(description="A generated, paragraph-length hypothetical document that directly answers the user's question, which will be used for semantic search.", alias="hyde_query")
    # 将原始查询分解为更小、更具体的问题列表
    sub_questions: List[str] = Field(description="A list of 2-3 smaller, more specific questions that break down the original query.")
    # 用于精确词法搜索的核心关键字和实体列表
    keywords: List[str] = Field(description="A list of 3-5 core keywords and entities extracted from the user's query.")

from typing import TypedDict, List, Optional
from langchain_core.documents import Document

class RAGGraphState(TypedDict):
    original_question: str
    expanded_queries: Optional[ExpandedQueries]
    retrieved_docs: List[Document]
    final_answer: str

# 节点 1: 查询扩展代理
query_expansion_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a query expansion specialist. Your goal is to transform a user's question into a diverse set of search queries to maximize retrieval recall. Generate a hypothetical document, sub-questions, and keywords."),
    ("human", "Please expand the following question: {question}")
])

# 该链将提示符通过管道链接到 LLM，并基于 Pydantic 模型构建其输出
query_expansion_chain = query_expansion_prompt | llm.with_structured_output(ExpandedQueries)
def query_expansion_node(state: RAGGraphState):
    """图中的第一个节点：获取原始问题并生成一组扩展查询"""
    print("--- [Expander] Generating parallel queries... ---")
    expanded_queries = query_expansion_chain.invoke({"question": state['original_question']})
    return {"expanded_queries": expanded_queries}

from concurrent.futures import ThreadPoolExecutor

def retrieval_node(state: RAGGraphState):
    """第二个节点：接受且并行执行扩展查询"""
    print("--- [Retriever] Executing parallel searches... ---")
    
    # 创建一个包含所有要执行的查询列表
    all_queries = []
    expanded = state['expanded_queries']
    all_queries.append(expanded.hypothetical_document)
    all_queries.extend(expanded.sub_questions)
    all_queries.extend(expanded.keywords)
    
    all_docs = []

    # 用 ThreadPoolExecutor 并发运行所有检索器调用
    with ThreadPoolExecutor(max_workers=5) as executor:
        results = executor.map(retriever.invoke, all_queries)
        for docs in results:
            all_docs.extend(docs)
    
    # 最后一步是对检索到的文档删除重复数据，以创建干净、唯一的上下文
    unique_docs = {doc.page_content: doc for doc in all_docs}.values()
    print(f"--- [Retriever] Found {len(unique_docs)} unique documents from {len(all_queries)} queries. ---")
    return {"retrieved_docs": list(unique_docs)}

from langgraph.graph import StateGraph, END

workflow = StateGraph(RAGGraphState)

# 将节点加入图
workflow.add_node("expand_queries", query_expansion_node)
workflow.add_node("retrieve_docs", retrieval_node)
workflow.add_node("generate_answer", generation_node)

# 定义线性工作流
workflow.set_entry_point("expand_queries")
workflow.add_edge("expand_queries", "retrieve_docs")
workflow.add_edge("retrieve_docs", "generate_answer")
workflow.add_edge("generate_answer", END)

# 用户查询使用一般术语（“big and fast”）而不是知识库中的技术术语
user_query = "How do modern AI systems get so big and fast at the same time? I've heard about attention but I'm not sure how it's optimized."

# --- 执行简单 RAG 系统 ---
print("--- [SIMPLE RAG] Retrieving documents...")
# 拦截检索步骤以检查简单系统找到的内容
simple_retrieved_docs = retriever.invoke(user_query)
print(f"--- [SIMPLE RAG] Documents Retrieved: {len(simple_retrieved_docs)}")
simple_rag_answer = simple_rag_chain.invoke(user_query)

# --- 执行高级 RAG 系统 ---
# --- 最终分析 ---
print("\n" + "="*60)
print("            RETRIEVED DOCUMENTS COMPARISON")
print("="*60)
print(f"\n--- Simple RAG Retrieved {len(simple_retrieved_docs)} document(s) ---")
for i, doc in enumerate(simple_retrieved_docs):
    print(f"{i+1}. {doc.page_content}")
print(f"\n--- Advanced RAG Retrieved {len(advanced_rag_result['retrieved_docs'])} document(s) ---")
for i, doc in enumerate(advanced_rag_result['retrieved_docs']):
    print(f"{i+1}. {doc.page_content}")
print("\n" + "="*60)
print("                     ACCURACY & QUALITY ANALYSIS")

#### 输出 ####
============================================================
            RETRIEVED DOCUMENTS COMPARISON
============================================================

--- Simple RAG Retrieved 1 document(s) ---
1. **Multi-headed Attention Mechanism**: The core component of the Transformer architecture is the multi-headed self-attention mechanism...
--- Advanced RAG Retrieved 3 document(s) ---
1. **FlashAttention Optimization**: ...FlashAttention is an I/O-aware algorithm that reorders the computation to reduce the number of read/write operations...
2. **Mixture of Experts (MoE) Layers**: ...a router network dynamically selects a small subset of 'expert' sub-networks to process each input token...
3. **Multi-headed Attention Mechanism**: The core component of the Transformer architecture...