RAG 아키텍처: 단순하고 고급이며 모듈식인 RAG 패턴
"RAG"라는 용어는 실제로 단순한 패턴부터 패턴에 이르기까지 매우 광범위한 아키텍처를 포괄합니다. 2023년부터 쿼리 라우팅, 순위 재지정, 자체 RAG를 통합하는 2026년 모듈형 시스템까지 3단계 그리고 일관성 검사. 이러한 진화를 이해하는 것이 기본입니다. 순진한 RAG 구현이 빠르지만 복잡한 문서에 대한 검색 품질이 낮습니다. 그만큼고급 RAG 특정 검색 문제를 해결합니다. 그만큼 모듈식 걸레 생산 시스템에 최대의 유연성을 제공합니다.
이 가이드에서는 실제 Python 코드, 비교 품질 측정항목을 사용하여 세 가지 아키텍처를 다룹니다. 사용 사례에 적합한 복잡성 수준을 선택하기 위한 기준.
무엇을 배울 것인가
- Naive RAG: 기본 아키텍처, 한계 및 충분할 때
- 고급 RAG: 검색 전(쿼리 재작성, HyDE), 검색 후(순위 재지정)
- 모듈형 RAG: 라우팅, 자체 RAG, CRAG 및 구성 가능 파이프라인
- 아키텍처를 객관적으로 비교하는 RAGAS 측정항목
- 각 아키텍처에 대한 완전한 Python 코드
- 결정 가이드: 다음 단계로 진출할 시기
소박한 RAG: 기본 패턴
Naive RAG는 최적화 없이 색인 검색 생성 흐름을 따릅니다.
- 고정 청크(일반적으로 512-1024 토큰)가 있는 문서 인덱스
- 쿼리를 임베딩으로 변환하고 가장 유사한 k개의 청크를 검색합니다.
- 청크를 프롬프트에 연결하고 응답을 생성합니다.
# Naive RAG con LangChain — implementazione completa
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Qdrant
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import DirectoryLoader, UnstructuredMarkdownLoader
# --- FASE 1: Indicizzazione ---
loader = DirectoryLoader(
"./docs",
glob="**/*.md",
loader_cls=UnstructuredMarkdownLoader
)
documents = loader.load()
# Chunking fisso — il limite principale del Naive RAG
splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=64,
separators=["\n\n", "\n", ".", " "]
)
chunks = splitter.split_documents(documents)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Qdrant.from_documents(
chunks, embeddings,
url="http://localhost:6333",
collection_name="naive_rag"
)
# --- FASE 2 + 3: Retrieval + Generation ---
NAIVE_RAG_PROMPT = PromptTemplate(
input_variables=["context", "question"],
template="""Rispondi alla domanda basandoti SOLO sul contesto fornito.
Se il contesto non contiene la risposta, dì "Non ho informazioni su questo argomento".
Contesto:
{context}
Domanda: {question}
Risposta:"""
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
rag_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
chain_type_kwargs={"prompt": NAIVE_RAG_PROMPT},
return_source_documents=True
)
result = rag_chain.invoke({"query": "Come gestire gli errori di timeout?"})
print(result["result"])
순진한 RAG의 한계: 모호한 쿼리, 청크 검색에 대한 성능 저하 부분적으로 관련됨, 복구된 문서가 서로 모순되는 사건 관리 없음, 구조화된 문서(테이블, 코드, 목록)의 품질은 다양합니다.
고급 RAG: 검색 전 및 검색 후 최적화
고급 RAG는 검색 전 및 검색 후 단계에 최적화를 추가합니다. 가장 많은 기술 영향을 미치는 것:
사전 검색: 쿼리 재작성 및 HyDE
사용자 쿼리는 모호하거나 잘못된 표현을 사용하는 경우가 많습니다. 쿼리 재작성은 LLM을 사용하여 의미 검색에 더 적합한 형식으로 쿼리를 재구성합니다.
# Advanced RAG: Query Rewriting + HyDE (Hypothetical Document Embeddings)
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 1. Multi-query: genera query alternative per copertura piu ampia
MULTI_QUERY_PROMPT = ChatPromptTemplate.from_messages([
("system", """Sei un esperto di information retrieval.
Genera 3 varianti della query fornita per recuperare documenti rilevanti
da diverse angolazioni. Restituisci solo le query, una per riga."""),
("human", "Query originale: {query}")
])
multi_query_chain = MULTI_QUERY_PROMPT | llm | StrOutputParser()
def generate_multiple_queries(query: str) -> list[str]:
result = multi_query_chain.invoke({"query": query})
queries = [q.strip() for q in result.strip().split('\n') if q.strip()]
return [query] + queries[:3] # query originale + 3 varianti
# 2. HyDE: genera un documento ipotetico che conterrebbe la risposta
HYDE_PROMPT = ChatPromptTemplate.from_messages([
("system", """Scrivi un breve paragrafo tecnico che risponderebbe
alla seguente domanda, come se fosse tratto da una documentazione ufficiale.
Usa terminologia tecnica precisa."""),
("human", "{query}")
])
hyde_chain = HYDE_PROMPT | llm | StrOutputParser()
def hyde_search(query: str, vectorstore, k: int = 5):
# Genera documento ipotetico
hypothetical_doc = hyde_chain.invoke({"query": query})
# Cerca usando il documento ipotetico come query (invece della query diretta)
results = vectorstore.similarity_search(hypothetical_doc, k=k)
return results
# 3. Multi-query retrieval con deduplicazione
from langchain.retrievers import MergerRetriever
from langchain_community.document_transformers import EmbeddingsRedundantFilter
def advanced_retrieve(query: str, vectorstore, k: int = 5) -> list:
queries = generate_multiple_queries(query)
# Raccogli risultati da tutte le query
all_docs = []
for q in queries:
docs = vectorstore.similarity_search(q, k=k)
all_docs.extend(docs)
# Deduplica per contenuto simile
seen_content = set()
unique_docs = []
for doc in all_docs:
content_hash = hash(doc.page_content[:200])
if content_hash not in seen_content:
seen_content.add(content_hash)
unique_docs.append(doc)
return unique_docs[:k * 2] # ritorna il doppio dei risultati per il reranker
검색 후: 크로스 인코더를 사용한 순위 재지정
벡터 임베딩은 "bi-encoder" 표현(별도의 쿼리 및 문서)을 사용합니다. 그러나 덜 정확합니다. 크로스 인코더 재순위(쿼리 + 문서 함께)로 정밀도 향상 추가 지연 시간(일반적으로 50-150ms)이 발생하여 15-25% 정도 증가합니다.
# Post-retrieval: Reranking con Cohere Rerank o cross-encoder locale
import cohere
from sentence_transformers import CrossEncoder
# Opzione 1: Cohere Rerank API (managed, accurato)
co = cohere.Client("your-api-key")
def rerank_with_cohere(query: str, documents: list[str], top_n: int = 5) -> list[dict]:
response = co.rerank(
query=query,
documents=documents,
top_n=top_n,
model="rerank-v3.5"
)
return [
{"content": documents[r.index], "relevance_score": r.relevance_score}
for r in response.results
]
# Opzione 2: Cross-encoder locale (gratuito, ~100MB)
cross_encoder = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def rerank_local(query: str, documents: list[str], top_n: int = 5) -> list[dict]:
# Crea coppie (query, documento) per il cross-encoder
pairs = [[query, doc] for doc in documents]
scores = cross_encoder.predict(pairs)
# Ordina per score decrescente
ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
return [{"content": doc, "relevance_score": float(score)} for doc, score in ranked[:top_n]]
# Advanced RAG completo: multi-query + HyDE + reranking
def advanced_rag(query: str, vectorstore) -> dict:
# 1. Retrieval ampliato
candidates = advanced_retrieve(query, vectorstore, k=8)
candidate_texts = [doc.page_content for doc in candidates]
# 2. Reranking
reranked = rerank_local(query, candidate_texts, top_n=5)
# 3. Generation con contesto di qualita
context = "\n\n---\n\n".join([r["content"] for r in reranked])
response = llm.invoke(f"""Contesto:\n{context}\n\nDomanda: {query}\nRisposta:""")
return {"answer": response.content, "sources": reranked}
모듈형 RAG: 모듈형 아키텍처
2026 Modular RAG는 파이프라인의 각 단계를 교체 가능한 모듈로 취급합니다. 패턴 가장 중요한:
CRAG: 교정 RAG
CRAG는 관련성 분류자를 추가합니다. 검색된 문서의 점수가 낮은 경우 시스템은 관련 없는 컨텍스트를 생성하는 대신 백업 웹 검색을 수행합니다.
# Modular RAG: CRAG (Corrective RAG) con LangGraph
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_community.tools.tavily_search import TavilySearchResults
class RAGState(TypedDict):
query: str
documents: list
relevance_scores: list[float]
web_results: list
answer: str
retrieval_quality: str # "high" | "low" | "ambiguous"
def retrieve(state: RAGState) -> RAGState:
"""Retrieval dal vector store"""
docs = vectorstore.similarity_search_with_score(state["query"], k=5)
documents = [doc for doc, _ in docs]
scores = [float(score) for _, score in docs]
return {**state, "documents": documents, "relevance_scores": scores}
def assess_relevance(state: RAGState) -> RAGState:
"""Valuta se i documenti sono sufficientemente rilevanti"""
avg_score = sum(state["relevance_scores"]) / len(state["relevance_scores"])
if avg_score > 0.85:
quality = "high"
elif avg_score > 0.70:
quality = "ambiguous"
else:
quality = "low"
return {**state, "retrieval_quality": quality}
def web_search_fallback(state: RAGState) -> RAGState:
"""Fallback: web search quando il retrieval e scarso"""
search_tool = TavilySearchResults(max_results=3)
results = search_tool.invoke(state["query"])
return {**state, "web_results": results}
def generate_answer(state: RAGState) -> RAGState:
"""Genera risposta usando documenti disponibili"""
if state["retrieval_quality"] == "low" and state["web_results"]:
context = "\n".join([r["content"] for r in state["web_results"]])
source_type = "web search"
else:
context = "\n".join([doc.page_content for doc in state["documents"]])
source_type = "knowledge base"
response = llm.invoke(
f"Contesto ({source_type}):\n{context}\n\nDomanda: {state['query']}\nRisposta:"
)
return {**state, "answer": response.content}
# Routing basato sulla qualita del retrieval
def should_web_search(state: RAGState) -> str:
return "web_search" if state["retrieval_quality"] == "low" else "generate"
# Costruzione del grafo
graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("assess_relevance", assess_relevance)
graph.add_node("web_search", web_search_fallback)
graph.add_node("generate", generate_answer)
graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "assess_relevance")
graph.add_conditional_edges(
"assess_relevance",
should_web_search,
{"web_search": "web_search", "generate": "generate"}
)
graph.add_edge("web_search", "generate")
graph.add_edge("generate", END)
crag = graph.compile()
# Esecuzione
result = crag.invoke({"query": "Qual e la versione piu recente di Qiskit?"})
print(result["answer"])
품질 비교: 순진함 vs 고급 vs 모듈형
Benchmark su dataset di test enterprise (500 domande, base di conoscenza 50K docs)
Metrica | Naive RAG | Advanced RAG | Modular RAG (CRAG)
--------------------|-----------|--------------|--------------------
Faithfulness | 0.71 | 0.88 | 0.92
Answer Relevancy | 0.74 | 0.86 | 0.89
Context Recall | 0.65 | 0.81 | 0.84
Context Precision | 0.72 | 0.87 | 0.88
--------------------|-----------|--------------|--------------------
Latenza p50 | 850ms | 1.4s | 1.8s (con web fallback: 3.2s)
Costo per query | $0.003 | $0.007 | $0.009 (avg)
--------------------|-----------|--------------|--------------------
"Hallucination rate"| 18% | 6% | 4%
Domande senza risp. | 12% | 8% | 3% (web fallback)
다음 레벨로 발전해야 하는 시기
- 순진한 -> 고급: 충실도가 0.80 미만이거나 사용자가 응답을 보고하는 경우 관련 없는 빈번함; 추가 비용 ~2배
- 고급 -> 모듈식: 지식 기반이 하위 집합만 다루는 경우 요청된 주제 중 또는 쿼리가 이질적인 주제에 걸쳐 있는 경우 추가 비용 ~1.3배
- 순진하게 유지: 지식 기반이 잘 구조화되어 있으면 쿼리는 다음과 같습니다. 동질성 및 충실도 > 0.85 이미 기본 패턴 사용
결론
올바른 RAG 아키텍처는 사용 사례의 복잡성에 따라 달라집니다. 항상 시작하세요 순진한 RAG, RAGAS로 측정하고 데이터가 보증하는 경우에만 발전하세요. 복잡성 추가 측정이 없으면 개선 없이는 비용이 더 많이 드는 과도한 엔지니어링 시스템으로 이어집니다. 측정 가능.
다음 기사에서는 청킹 전략(검색 파이프라인 구성 요소)에 대해 자세히 설명합니다. Naive RAG의 품질에 가장 큰 영향을 미치며 종종 간과되는 부분입니다.