Split View: Advanced RAG 파이프라인 완전 가이드 2025: 청킹 전략, 리랭킹, 에이전틱 RAG, 평가

User question → [Retrieval] → [Relevant documents] → [LLM + document context] → Answer generation

Naive RAG (Early 2023)
├── Fixed-size chunking
├── Single embedding retrieval
├── Top-K results passed directly to LLM
└── Problems: Low retrieval accuracy, context noise

Advanced RAG (2024)
├── Semantic chunking + metadata
├── Hybrid search (vector + keyword)
├── Re-ranking to refine search results
├── Query transformation (HyDE, Multi-Query)
└── Context compression

Modular RAG (2025)
├── Agentic RAG (dynamic routing)
├── Self-RAG (self-reflection)
├── CRAG (Corrective RAG)
├── Multi-modal RAG
├── Knowledge Graph + RAG
└── Modular composable pipelines

Bad chunking: Cut mid-sentence → meaning lost → retrieval fails → wrong answer
Good chunking: Split by meaning → rich context → accurate retrieval → correct answer

from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(
    chunk_size=500,      # chunk size
    chunk_overlap=50,    # overlap (preserves boundary info)
    separator="\n\n"     # separator
)

chunks = splitter.split_text(document_text)

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""]  # priority order
)

chunks = splitter.split_text(document_text)
# First tries \n\n, if chunk too large then \n, then period...

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Semantic chunking
chunker = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95  # split when similarity below 95th percentile
)

chunks = chunker.split_text(document_text)

Sentence 1: "Vector DBs are core AI infrastructure." → embedding [0.1, 0.3, ...]
Sentence 2: "HNSW is the fastest search algorithm." → similarity 0.85 (high → same chunk)
Sentence 3: "Meanwhile, the weather is sunny today." → similarity 0.15 (low → new chunk!)

from langchain.text_splitter import MarkdownHeaderTextSplitter

headers_to_split = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

splitter = MarkdownHeaderTextSplitter(headers_to_split)
chunks = splitter.split_text(markdown_text)

# Each chunk automatically includes header metadata
for chunk in chunks:
    print(f"Metadata: {chunk.metadata}")
    # {'Header 1': 'Vector Database', 'Header 2': 'Indexing Algorithms'}

from openai import OpenAI

client = OpenAI()

def agentic_chunk(text, max_chunk_size=1500):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """Split the given text into semantically independent chunks.
Each chunk should be self-contained and understandable on its own.
Return chunk texts as a JSON array."""},
            {"role": "user", "content": text}
        ],
        response_format={"type": "json_object"}
    )
    return response.choices[0].message.content

General text: 500-1000 tokens (10-20% overlap)
Technical docs: 800-1500 tokens (section-based)
Legal/Medical: 300-500 tokens (precision-focused)
Code: Function/class units (structure-based)
Conversations/QA: Per dialog turn

Cost-focused + general      → text-embedding-3-small
Top performance + free      → BGE-M3 (self-hosted)
Multilingual + search       → Cohere embed-v3.0
Code search                 → voyage-code-3
Long docs (8K+ tokens)      → Jina-embeddings-v3
Private data + on-premises  → BGE-M3 or Nomic

from ragatouille import RAGPretrainedModel

rag = RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0")

# Indexing
rag.index(
    collection=[doc1, doc2, doc3],
    index_name="my_index"
)

# Search (token-level matching)
results = rag.search(query="vector database performance comparison", k=5)

Original query: "RAG is slow" (ambiguous, short)
→ Transformed: "How to optimize RAG pipeline retrieval latency" (specific, retrieval-suitable)

from openai import OpenAI

client = OpenAI()

def hyde_search(query, collection):
    # 1. Generate hypothetical answer with LLM
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Write a detailed answer to the given question. It doesn't need to be accurate."},
            {"role": "user", "content": query}
        ]
    )
    hypothetical_doc = response.choices[0].message.content
    
    # 2. Embed the hypothetical answer and search
    embedding = get_embedding(hypothetical_doc)
    results = collection.search(query_vector=embedding, limit=5)
    
    return results

def multi_query_transform(original_query):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """Rewrite the given question from 3 different perspectives.
Each question should maintain the original intent but use different keywords.
Return only 3 questions separated by newlines."""},
            {"role": "user", "content": original_query}
        ]
    )
    queries = response.choices[0].message.content.strip().split("\n")
    return [original_query] + queries

# Usage
queries = multi_query_transform("RAG pipeline performance optimization")
# → ["RAG pipeline performance optimization",
#    "How to improve response time in retrieval augmented generation systems",
#    "Strategies to increase search accuracy in RAG architecture",
#    "LLM-based document retrieval system optimization techniques"]

# Search with each query and merge results (deduplicate)
all_results = set()
for q in queries:
    results = search(q)
    all_results.update(results)

def step_back_prompt(query):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """Generate a question that is one step more general and broader than the given question.
The answer to this general question should help answer the original question."""},
            {"role": "user", "content": query}
        ]
    )
    return response.choices[0].message.content

# Example
original = "What is the memory impact when setting HNSW M parameter to 32 in Qdrant?"
step_back = step_back_prompt(original)
# → "How do HNSW index parameters affect performance and resources in vector databases?"

def decompose_query(complex_query):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """Decompose the complex question into simpler sub-questions.
Each sub-question should be independently answerable.
Return as a JSON array."""},
            {"role": "user", "content": complex_query}
        ],
        response_format={"type": "json_object"}
    )
    return response.choices[0].message.content

# Example
complex_q = "Compare Pinecone and Qdrant on performance, cost, and operational ease at 10M vector scale?"
sub_questions = decompose_query(complex_q)
# → ["What is Pinecone's performance at 10M vector scale?",
#    "What is Qdrant's performance at 10M vector scale?",
#    "What is Pinecone's cost structure?",
#    "What is Qdrant's cost structure?",
#    "How easy is Pinecone to operate?",
#    "How easy is Qdrant to operate?"]

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Qdrant

# BM25 (keyword) retriever
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

# Vector retriever
vector_retriever = qdrant_vectorstore.as_retriever(
    search_kwargs={"k": 5}
)

# Ensemble (hybrid)
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # weight toward vector
)

results = ensemble_retriever.invoke("How to optimize RAG pipeline")

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

compressor = LLMChainExtractor.from_llm(llm)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vector_retriever
)

# Retrieval + compression: extracts only relevant parts
compressed_docs = compression_retriever.invoke("Role of M parameter in HNSW algorithm")
# Returns only question-relevant paragraphs, not the entire document

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

# Small chunks: for retrieval
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

# Large chunks: for context
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

store = InMemoryStore()

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# Precise search with 200-token chunks → return 2000-token parent documents
retriever.add_documents(documents)
results = retriever.invoke("HNSW parameter tuning")
# Retrieval is precise with small chunks, context is rich with large ones

from langchain.retrievers.multi_vector import MultiVectorRetriever

# Generate summaries + hypothetical questions per document
summaries = generate_summaries(documents)
hypothetical_questions = generate_questions(documents)

# Search by summaries/questions, return original documents
retriever = MultiVectorRetriever(
    vectorstore=vectorstore,  # stores summary/question vectors
    docstore=store,           # stores original documents
    id_key="doc_id"
)

# Searches via summary embeddings but returns full original documents

Stage 1 (Bi-encoder): query vector vs doc vector → fast but imprecise → Top 20
Stage 2 (Cross-encoder): (query, doc) pairs directly compared → slow but accurate → Top 5

from sentence_transformers import CrossEncoder

# Load cross-encoder model
model = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

# Initial search results
initial_results = vector_search(query, top_k=20)

# Re-rank
pairs = [(query, doc.content) for doc in initial_results]
scores = model.predict(pairs)

# Sort by score
reranked = sorted(
    zip(initial_results, scores),
    key=lambda x: x[1],
    reverse=True
)[:5]

import cohere

co = cohere.Client("YOUR_API_KEY")

# Initial search result texts
documents = [doc.content for doc in initial_results]

# Cohere re-ranking
response = co.rerank(
    model="rerank-v3.5",
    query="RAG pipeline optimization",
    documents=documents,
    top_n=5
)

for result in response.results:
    print(f"Index: {result.index}, Score: {result.relevance_score:.4f}")
    print(f"Text: {documents[result.index][:100]}...")

from ragatouille import RAGPretrainedModel

rag = RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0")

reranked = rag.rerank(
    query="How to optimize RAG pipeline",
    documents=[doc.content for doc in initial_results],
    k=5
)

def llm_rerank(query, documents, top_n=5):
    scored_docs = []
    
    for doc in documents:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "Rate the relevance between the question and document on a 0-10 scale. Return only the number."},
                {"role": "user", "content": f"Question: {query}\n\nDocument: {doc.content[:500]}"}
            ]
        )
        score = float(response.choices[0].message.content.strip())
        scored_docs.append((doc, score))
    
    return sorted(scored_docs, key=lambda x: x[1], reverse=True)[:top_n]

Traditional RAG: Question → Retrieve → Generate (fixed pipeline)
Agentic RAG: Question → [Agent decides]
              ├── Is retrieval needed? → Retrieve → Results sufficient?
              │                              ├── Yes → Generate
              │                              └── No → Search different source / modify query
              └── Can answer without retrieval → Generate directly

def self_rag(query):
    # 1. Judge if retrieval is needed
    need_retrieval = judge_retrieval_need(query)
    
    if not need_retrieval:
        return generate_without_context(query)
    
    # 2. Perform retrieval
    documents = retrieve(query)
    
    # 3. Judge relevance of each document
    relevant_docs = []
    for doc in documents:
        if is_relevant(query, doc):
            relevant_docs.append(doc)
    
    if not relevant_docs:
        # No relevant docs found, refine query and re-search
        refined_query = refine_query(query)
        documents = retrieve(refined_query)
        relevant_docs = [d for d in documents if is_relevant(refined_query, d)]
    
    # 4. Generate answer
    answer = generate_with_context(query, relevant_docs)
    
    # 5. Self-evaluate answer quality
    if not is_supported(answer, relevant_docs):
        # If answer not grounded in docs, regenerate
        answer = regenerate(query, relevant_docs)
    
    return answer

def corrective_rag(query):
    # 1. Initial retrieval
    documents = retrieve(query)
    
    # 2. Evaluate retrieval quality
    quality = evaluate_retrieval_quality(query, documents)
    
    if quality == "CORRECT":
        # Good results → refine knowledge and generate
        refined_knowledge = refine_knowledge(documents)
        return generate(query, refined_knowledge)
    
    elif quality == "AMBIGUOUS":
        # Ambiguous results → supplement with web search
        web_results = web_search(query)
        combined = documents + web_results
        refined = refine_knowledge(combined)
        return generate(query, refined)
    
    elif quality == "INCORRECT":
        # Poor results → replace with web search
        web_results = web_search(query)
        refined = refine_knowledge(web_results)
        return generate(query, refined)

def adaptive_rag(query):
    # Classify query complexity
    complexity = classify_query(query)
    
    if complexity == "SIMPLE":
        # Simple factual question → direct retrieve + generate
        docs = simple_retrieve(query, top_k=3)
        return generate(query, docs)
    
    elif complexity == "MODERATE":
        # Moderate → multi-query + re-ranking
        queries = multi_query_transform(query)
        docs = hybrid_search(queries)
        reranked = rerank(query, docs)
        return generate(query, reranked)
    
    elif complexity == "COMPLEX":
        # Complex → query decomposition + multi-step reasoning
        sub_queries = decompose_query(query)
        sub_answers = []
        for sq in sub_queries:
            docs = search(sq)
            sub_answers.append(generate(sq, docs))
        return synthesize(query, sub_answers)

def query_router(query):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """Analyze the query and select the appropriate data source:
- VECTOR_DB: When internal document search is needed
- WEB_SEARCH: When latest or external information is needed
- SQL_DB: When structured data queries are needed
- DIRECT: When the LLM can answer directly
Return a single word only."""},
            {"role": "user", "content": query}
        ]
    )
    route = response.choices[0].message.content.strip()
    
    if route == "VECTOR_DB":
        return vector_db_search(query)
    elif route == "WEB_SEARCH":
        return web_search(query)
    elif route == "SQL_DB":
        return sql_query(query)
    else:
        return direct_answer(query)

from langchain_community.document_loaders import UnstructuredPDFLoader

# Extract text + images + tables from PDF
loader = UnstructuredPDFLoader(
    "document.pdf",
    mode="elements",
    strategy="hi_res"  # high-res image/table extraction
)
elements = loader.load()

# Generate descriptions for image elements using GPT-4o
for element in elements:
    if element.metadata.get("type") == "Image":
        description = describe_image_with_vision(element)
        element.page_content = description  # convert image description to text

# Store all elements (text + image descriptions) in vector DB

def process_table(table_element):
    """Convert table to searchable format"""
    # 1. Convert table to markdown
    md_table = table_element.metadata.get("text_as_html", "")
    
    # 2. Generate table summary with LLM
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Summarize the following table in natural language."},
            {"role": "user", "content": md_table}
        ]
    )
    summary = response.choices[0].message.content
    
    # 3. Store summary with metadata
    return {
        "content": summary,
        "metadata": {
            "type": "table",
            "original_html": md_table,
            "source": table_element.metadata.get("source")
        }
    }

import base64

def describe_image_with_vision(image_path):
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode()
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": "Describe this image in detail. For diagrams explain the structure, for charts explain the data."},
                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_data}"}}
                ]
            }
        ]
    )
    return response.choices[0].message.content

Standard RAG: Retrieves document chunks independently
GraphRAG: Uses entity relationships to retrieve connected information too

Example: "Which vector databases use HNSW?"
Standard RAG: Returns only HNSW chunks
GraphRAG: HNSW → (used_by) → Qdrant, Weaviate, Milvus
         + Returns HNSW configuration chunks for each DB

from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain

graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="password"
)

# Knowledge graph + LLM QA chain
chain = GraphCypherQAChain.from_llm(
    llm=llm,
    graph=graph,
    verbose=True
)

result = chain.invoke("What indexing algorithms does Qdrant support?")
# LLM generates Cypher queries to explore relationships in Neo4j

def hybrid_graph_vector_rag(query):
    # 1. Vector search for relevant document chunks
    vector_results = vector_search(query, top_k=5)
    
    # 2. Extract entities from chunks
    entities = extract_entities(query)
    
    # 3. Explore related entities in knowledge graph
    graph_results = graph_query(entities, depth=2)
    
    # 4. Combine both results
    combined_context = merge_results(vector_results, graph_results)
    
    # 5. LLM generation
    return generate(query, combined_context)

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# Prepare evaluation dataset
eval_data = {
    "question": ["What is RAG?", "How does HNSW algorithm work?"],
    "answer": ["RAG is retrieval augmented generation...", "HNSW is a hierarchical graph..."],
    "contexts": [
        ["RAG (Retrieval-Augmented Generation) is...", "A retrieval-based generation technique..."],
        ["HNSW stands for Hierarchical Navigable...", "A graph-based ANN algorithm..."]
    ],
    "ground_truth": ["RAG enables LLMs to retrieve external...", "HNSW is a multi-layer graph structure..."]
}

dataset = Dataset.from_dict(eval_data)

# Run RAGAS evaluation
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)

print(results)
# {'faithfulness': 0.89, 'answer_relevancy': 0.92,
#  'context_precision': 0.85, 'context_recall': 0.78}

from trulens.apps.langchain import TruChain
from trulens.core import Feedback, TruSession
from trulens.providers.openai import OpenAI as TruOpenAI

session = TruSession()
provider = TruOpenAI()

# Define feedback functions
f_relevance = Feedback(provider.relevance).on_input_output()
f_groundedness = Feedback(provider.groundedness_measure_with_cot_reasons).on(
    source=context, statement=output
)

# Wrap RAG chain
tru_chain = TruChain(
    rag_chain,
    app_name="RAG Pipeline v1",
    feedbacks=[f_relevance, f_groundedness]
)

# Run evaluation
with tru_chain as recording:
    response = rag_chain.invoke("How to optimize RAG pipelines?")

# Check dashboard
session.get_leaderboard()

def llm_judge(question, answer, context, ground_truth):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": """Evaluate the RAG system's answer.
Score 1-5 on these criteria:
1. Accuracy: Is the answer factually correct
2. Relevance: Is it a fitting answer to the question
3. Groundedness: Is the answer based on the provided context
4. Completeness: Does it fully answer the question
Return JSON with each score and reasoning."""},
            {"role": "user", "content": f"""Question: {question}
Answer: {answer}
Context: {context}
Ground Truth: {ground_truth}"""}
        ],
        response_format={"type": "json_object"}
    )
    return response.choices[0].message.content

import hashlib
import json

class RAGCache:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 3600  # 1 hour
    
    def _hash_query(self, query):
        return hashlib.md5(query.encode()).hexdigest()
    
    def get_cached_response(self, query):
        key = self._hash_query(query)
        cached = self.redis.get(f"rag:{key}")
        if cached:
            return json.loads(cached)
        return None
    
    def cache_response(self, query, response):
        key = self._hash_query(query)
        self.redis.setex(
            f"rag:{key}",
            self.ttl,
            json.dumps(response)
        )

# Semantic caching: cache hits for similar queries too
class SemanticCache:
    def __init__(self, vectorstore, threshold=0.95):
        self.store = vectorstore
        self.threshold = threshold
    
    def get(self, query):
        results = self.store.similarity_search_with_score(query, k=1)
        if results and results[0][1] >= self.threshold:
            return results[0][0].metadata["response"]
        return None

from openai import OpenAI

client = OpenAI()

def stream_rag_response(query, context):
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"Context:\n{context}"},
            {"role": "user", "content": query}
        ],
        stream=True
    )
    
    for chunk in stream:
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content

# FastAPI streaming endpoint
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/rag/stream")
async def rag_stream(query: str):
    context = retrieve_context(query)
    return StreamingResponse(
        stream_rag_response(query, context),
        media_type="text/event-stream"
    )