RAG Architecture Patterns: Паттерны архитектуры RAG

Введение

Retrieval-Augmented Generation (RAG) — архитектура, которая комбинирует retrieval информации с generation текста. Существует множество паттернов RAG, каждый для своих use cases.

---

Базовый RAG

Simple RAG

Simple RAG Pipeline:
  Query → Embed → Retrieve → Context + Query → LLM → Response

  Steps:
  1. User query
  2. Convert query to embedding
  3. Search vector DB for similar chunks
  4. Append chunks to prompt
  5. Generate response

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.llms import OpenAI

def simple_rag(query: str) -> str:
    # 1. Embed query
    embeddings = OpenAIEmbeddings()
    
    # 2. Retrieve
    vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
    docs = vectorstore.similarity_search(query, k=4)
    
    # 3. Build context
    context = "\n\n".join([doc.page_content for doc in docs])
    
    # 4. Generate
    prompt = f"""Answer the question based on the context.
    
Context:
{context}

Question: {query}
Answer:"""
    
    llm = OpenAI(temperature=0)
    return llm(prompt)

---

Advanced RAG Patterns

1. Query Transformation

Query Transformation RAG:
  Query → Rewrite → Embed → Retrieve → Context → LLM → Response
        → Expand → Embed → Retrieve → Context → LLM → Response
        → Decompose → Embed → Retrieve → Context → LLM → Response

def query_transformation_rag(query: str) -> str:
    llm = ChatOpenAI(model="gpt-4o")
    
    # Rewrite: переформулирование для лучшего поиска
    rewrite_prompt = ChatPromptTemplate.from_template("""
    Rewrite the following query to improve retrieval:
    Original: {query}
    Rewritten:
    """)
    
    rewrite_chain = rewrite_prompt | llm
    rewritten_query = rewrite_chain.invoke({"query": query}).content
    
    # Expand: добавление связанных вопросов
    expand_prompt = ChatPromptTemplate.from_template("""
    Generate 3 related questions for: {query}
    Return as numbered list.
    """)
    
    expand_chain = expand_prompt | llm
    expanded_queries = expand_chain.invoke({"query": query}).content.split('\n')
    
    # Decompose: разложение на подзапросы
    decompose_prompt = ChatPromptTemplate.from_template("""
    Decompose this question into sub-questions: {query}
    Return as numbered list.
    """)
    
    decompose_chain = decompose_prompt | llm
    sub_questions = decompose_chain.invoke({"query": query}).content.split('\n')
    
    # Retrieve для всех вариантов
    all_docs = []
    for q in [rewritten_query] + expanded_queries + sub_questions:
        docs = vectorstore.similarity_search(q, k=3)
        all_docs.extend(docs)
    
    # Deduplicate
    unique_docs = list({doc.page_content: doc for doc in all_docs}.values())
    
    # Generate
    context = "\n\n".join([doc.page_content for doc in unique_docs[:8]])
    return generate_response(query, context)

2. Multi-Vector RAG

Multi-Vector RAG:
  Document → Split into multiple representations
            → Question vectors
            → Answer vectors
            → Metadata vectors
  
  Retrieve: question → answer/metadata

from langchain.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

def multi_vector_rag():
    # Загрузка документов
    loader = DirectoryLoader("./docs")
    documents = loader.load()
    
    # Разбиение на chunks
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=100
    )
    chunks = splitter.split_documents(documents)
    
    # Создание нескольких представлений
    # 1. Question vectors — вопросы, на которые отвечает chunk
    # 2. Answer vectors — сами ответы
    # 3. Metadata vectors — заголовки, ключевые слова
    
    # Embedding для каждого типа
    question_embeddings = create_question_embeddings(chunks)
    answer_embeddings = create_answer_embeddings(chunks)
    metadata_embeddings = create_metadata_embeddings(chunks)
    
    # Хранение в multi-vector store
    store = MultiVectorStore(
        question_vectors=question_embeddings,
        answer_vectors=answer_embeddings,
        metadata_vectors=metadata_embeddings,
        original_docs=chunks
    )
    
    return store

3. Parent Document Retriever

Parent Document Retriever:
  Large Document → Split into small chunks
                  → Store small chunks for retrieval
                  → Keep parent documents for context
  
  Retrieve: small chunk → return parent document

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

def parent_document_retriever():
    # Small chunks для retrieval
    small_splitter = RecursiveCharacterTextSplitter(
        chunk_size=200,
        chunk_overlap=50
    )
    
    # Large chunks для контекста
    large_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200
    )
    
    # Загрузка
    loader = DirectoryLoader("./docs")
    docs = loader.load()
    
    # Splitting
    small_docs = small_splitter.split_documents(docs)
    large_docs = large_splitter.split_documents(docs)
    
    # Vector store для small chunks
    vectorstore = Chroma(embedding_function=OpenAIEmbeddings())
    
    # Storage для parent documents
    store = InMemoryStore()
    
    # Retriever
    retriever = ParentDocumentRetriever(
        vectorstore=vectorstore,
        docstore=store,
        child_splitter=small_splitter,
        parent_splitter=large_splitter,
    )
    
    retriever.add_documents(large_docs)
    
    # Retrieve small chunks → get parent documents
    docs = retriever.similarity_search("query", k=4)

4. HyDE (Hypothetical Document Embeddings)

HyDE:
  Query → Generate hypothetical answer → Embed hypothetical → Retrieve
  
  Idea: Embed a hypothetical answer instead of the query
  This puts the embedding in the same space as documents

from langchain.retrievers import HyDERetriever

def hyde_retrieval():
    # Гипотетический документ
    llm = ChatOpenAI(model="gpt-4o")
    
    hyde_prompt = ChatPromptTemplate.from_template("""
    Given the question, generate a hypothetical document that answers it.
    The document should be detailed and factual.
    
    Question: {question}
    Hypothetical document:
    """)
    
    # Генерация гипотетического документа
    hypothetical_chain = hyde_prompt | llm
    
    def generate_hypothetical(question: str) -> str:
        response = hypothetical_chain.invoke({"question": question})
        return response.content
    
    # Embedding гипотетического документа
    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
    
    # Поиск по гипотетическому документу
    hypothetical = generate_hypothetical("What is the capital of France?")
    docs = vectorstore.similarity_search(hypothetical, k=4)
    
    return docs

---

Agentic RAG

RAG с агентом

Agentic RAG:
  Query → Agent → Decide: Retrieve? Rewrite? Search? → Action → Observe → Response
  
  Agent может:
  - Решать, нужно ли retrieval
  - Переформулировать query
  - Использовать multiple tools
  - Итеративно уточнять ответ

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import Tool

def agentic_rag():
    # Tools для агента
    retrieval_tool = Tool(
        name="retriever",
        func=lambda x: vectorstore.similarity_search(x, k=4),
        description="Search the knowledge base"
    )
    
    search_tool = Tool(
        name="web_search",
        func=lambda x: web_search_api(x),
        description="Search the web"
    )
    
    calculator_tool = Tool(
        name="calculator",
        func=lambda x: calculate(x),
        description="Perform calculations"
    )
    
    # Agent
    llm = ChatOpenAI(model="gpt-4o")
    prompt = create_agent_prompt()
    
    agent = create_openai_functions_agent(
        llm=llm,
        tools=[retrieval_tool, search_tool, calculator_tool],
        prompt=prompt
    )
    
    agent_executor = AgentExecutor(
        agent=agent,
        tools=[retrieval_tool, search_tool, calculator_tool],
        verbose=True
    )
    
    return agent_executor.invoke({"input": "What is the population of Paris?"})

Multi-Agent RAG

Multi-Agent RAG:
  Query → Router Agent
              → Research Agent → Retrieve → Synthesize
              → Verify Agent → Fact-check → Validate
              → Writer Agent → Compose → Final Answer

class MultiAgentRAG:
    def __init__(self):
        self.router = create_router_agent()
        self.researcher = create_research_agent()
        self.verifier = create_verify_agent()
        self.writer = create_writer_agent()
    
    def process(self, query: str) -> str:
        # 1. Router определяет тип запроса
        route = self.router.invoke({"query": query})
        
        # 2. Research agent ищет информацию
        research = self.researcher.invoke({
            "query": query,
            "route": route
        })
        
        # 3. Verifier проверяет факты
        verification = self.verifier.invoke({
            "query": query,
            "research": research
        })
        
        # 4. Writer compose final answer
        answer = self.writer.invoke({
            "query": query,
            "research": research,
            "verification": verification
        })
        
        return answer

---

Self-RAG

Self-Reflection RAG

Self-RAG:
  Query → Retrieve → Generate → Reflect:
    - Is response supported by context?
    - Is response relevant to query?
    - Is response complete?
  
  If reflection fails → Retrieve again → Regenerate

def self_rag(query: str) -> str:
    max_iterations = 3
    
    for i in range(max_iterations):
        # Retrieve
        docs = vectorstore.similarity_search(query, k=4)
        context = "\n\n".join([d.page_content for d in docs])
        
        # Generate
        llm = ChatOpenAI(model="gpt-4o")
        response = generate(llm, query, context)
        
        # Reflect
        reflection = reflect(
            llm,
            query=query,
            response=response,
            context=context
        )
        
        # Check reflection scores
        if reflection["supported"] > 0.8 and \
           reflection["relevant"] > 0.8 and \
           reflection["complete"] > 0.8:
            return response
        
        # If not good, retrieve more
        query = refine_query(query, reflection)
    
    return response

Reflection prompts

def reflect(llm, query, response, context):
    reflection_prompt = ChatPromptTemplate.from_template("""
    Evaluate the response:
    
    Query: {query}
    Context: {context}
    Response: {response}
    
    Score 0-1 for:
    1. Support: Is the response fully supported by context?
    2. Relevance: Is the response relevant to the query?
    3. Completeness: Does the response fully answer the query?
    
    Return JSON: {"support": 0.9, "relevance": 0.8, "completeness": 0.7}
    """)
    
    result = llm.invoke(reflection_prompt.format(
        query=query,
        context=context,
        response=response
    ))
    
    return json.loads(result.content)

---

Adaptive RAG

Adaptive RAG

Adaptive RAG:
  Query → Classifier → Choose strategy:
    - Simple QA → Direct answer (no retrieval)
    - Factual QA → Retrieve → Generate
    - Multi-hop → Multi-step retrieval
    - Creative → Generate without retrieval

def adaptive_rag(query: str) -> str:
    llm = ChatOpenAI(model="gpt-4o")
    
    # Classify query type
    classifier_prompt = ChatPromptTemplate.from_template("""
    Classify the query type:
    - simple_qa: Simple question with direct answer
    - factual: Requires factual information
    - multi_hop: Requires multiple steps/retrievals
    - creative: Requires creative generation
    - reasoning: Requires logical reasoning
    
    Query: {query}
    Type:
    """)
    
    type_response = llm.invoke(classifier_prompt.format(query=query))
    query_type = type_response.content.strip()
    
    # Choose strategy
    if query_type == "simple_qa":
        return direct_answer(llm, query)
    elif query_type == "factual":
        return simple_rag(query)
    elif query_type == "multi_hop":
        return multi_hop_rag(query)
    elif query_type == "creative":
        return creative_generation(llm, query)
    elif query_type == "reasoning":
        return reasoning_rag(query)

---

Graph RAG

Knowledge Graph RAG

Graph RAG:
  Documents → Extract entities & relations → Knowledge Graph
  Query → Graph traversal → Context → LLM → Response
  
  Benefits:
  - Better for structured knowledge
  - Can answer multi-hop questions
  - Explicit entity relationships

from langchain.graphs import Neo4jGraph

def graph_rag(query: str):
    # Подключение к графу
    graph = Neo4jGraph()
    
    # Query графа
    cypher_query = """
    MATCH (n:Entity)
    WHERE n.name CONTAINS $query
    MATCH (n)-[r:RELATED_TO]-(m:Entity)
    RETURN n.name AS source, type(r) AS relation, m.name AS target
    """
    
    results = graph.query(cypher_query, params={"query": query})
    
    # Build context from graph
    context = build_graph_context(results)
    
    # Generate response
    return generate_response(query, context)

Entity extraction for Graph RAG

def extract_entities(documents):
    """Извлечение сущностей из документов"""
    llm = ChatOpenAI(model="gpt-4o")
    
    extraction_prompt = ChatPromptTemplate.from_template("""
    Extract entities and relationships from the text.
    
    Text: {text}
    
    Return JSON:
    {
      "entities": [{"name": "Entity1", "type": "Person/Org/Location"}],
      "relationships": [
        {"source": "Entity1", "relation": "WORKS_FOR", "target": "Entity2"}
      ]
    }
    """)
    
    for doc in documents:
        result = llm.invoke(extraction_prompt.format(text=doc.page_content))
        yield json.loads(result.content)

---

RAG Evaluation

Оценка качества RAG

RAG Evaluation Metrics:
  1. Context Relevance: Насколько retrieved context релевантен
  2. Answer Relevance: Насколько ответ релевантен query
  3. Answer Correctness: Насколько ответ корректен
  4. Context Recall: Насколько хорошо context покрывает ground truth

from ragas import evaluate
from datasets import Dataset

def evaluate_rag(questions, answers, contexts, ground_truths):
    # Создание dataset
    data = {
        "questions": questions,
        "answers": answers,
        "contexts": contexts,
        "ground_truths": ground_truths
    }
    
    dataset = Dataset.from_dict(data)
    
    # Evaluation
    result = evaluate(
        dataset,
        metrics=[
            context_relevance,
            answer_relevance,
            answer_correctness,
            context_recall
        ]
    )
    
    return result.to_pandas()

---

RAG Patterns Comparison

Pattern              | Use Case                    | Complexity
---------------------|-----------------------------|----------
Simple RAG           | Basic QA                    | Low
Query Transformation | Ambiguous queries           | Medium
Multi-Vector         | Multi-faceted docs          | High
Parent Document