Retrieval-Augmented Generation: когда использовать RAG vs fine-tuning, гибридные подходы
Retrieval-Augmented Generation: когда использовать RAG vs fine-tuning, гибридные подходы
RAG (Retrieval-Augmented Generation) — подход, комбинирующий retrieval документов из базы знаний с генерацией ответа LLM. Вместо того чтобы полагаться только на знания модели, мы предоставляем релевантные документы в контексте.
Зачем RAG:
| Критерий | RAG |
|---|---|
| Данные часто обновляются | ✅ |
| Нужны citations/источники | ✅ |
| Приватные/внутренние данные | ✅ |
| Factual knowledge | ✅ |
| Бюджет ограничен | ✅ |
| Критерий | Fine-tuning |
|---|---|
| Domain-specific стиль/формат | ✅ |
| Специфичная терминология | ✅ |
| Данные стабильны | ✅ |
| Нужно "понимание" домена | ✅ |
| Есть бюджет на обучение | ✅ |
Fine-tuning для: style, format, domain terminology
+
RAG для: factual knowledge, frequently changing data
=
Best of both worlds
┌─────────────┐ ┌──────────────┐ ┌─────────────┐
│ User │────▶│ LLM │────▶│ Answer │
│ Question │ │ + Context │ │ + Sources │
└─────────────┘ └──────────────┘ └─────────────┘
▲
│
┌─────┴─────┐
│ Retrieve │
└─────┬─────┘
│
┌─────┴─────┐
│ Vector │
│ DB │
└───────────┘
pip install langchain langchain-community langchain-chroma
pip install chromadb # vector database
pip install sentence-transformers # embeddingsfrom langchain.document_loaders import (
DirectoryLoader,
PyPDFLoader,
TextLoader,
UnstructuredHTMLLoader
)
# Загрузка из директории
loader = DirectoryLoader(
"./docs/",
glob="**/*.pdf",
loader_cls=PyPDFLoader
)
documents = loader.load()
print(f"Loaded {len(documents)} documents")
# Загрузка одного файла
loader = PyPDFLoader("./docs/manual.pdf")
documents = loader.load_and_split()from langchain.text_splitter import (
RecursiveCharacterTextSplitter,
CharacterTextSplitter
)
# Recursive splitter (recommended)
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
length_function=len,
separators=["\n\n", "\n", " ", ""]
)
chunks = text_splitter.split_documents(documents)
print(f"Created {len(chunks)} chunks")
# Для кода
from langchain.text_splitter import Language, RecursiveCharacterTextSplitter
python_splitter = RecursiveCharacterTextSplitter.from_language(
language=Language.PYTHON,
chunk_size=500,
chunk_overlap=50
)from langchain.embeddings import HuggingFaceEmbeddings
# Локальные эмбеддинги
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2",
model_kwargs={'device': 'cpu'},
encode_kwargs={'normalize_embeddings': True}
)
# OpenAI embeddings
from langchain.embeddings import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# Создание эмбеддинга
vector = embeddings.embed_query("Hello, world!")
print(f"Vector dimension: {len(vector)}") # 384 for MiniLMfrom langchain.vectorstores import Chroma
# Создание векторной БД
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db"
)
# Поиск
query = "Как настроить аутентификацию?"
results = vectorstore.similarity_search(query, k=3)
for doc in results:
print(f"Score: {doc.metadata.get('score', 'N/A')}")
print(f"Content: {doc.page_content[:200]}...\n")
# Поиск с score
results_with_scores = vectorstore.similarity_search_with_score(query, k=3)
for doc, score in results_with_scores:
print(f"Score: {score:.4f}") # Lower = more similarfrom langchain.chains import RetrievalQA
from langchain.llms import Ollama
from langchain.vectorstores import Chroma
# Инициализация
llm = Ollama(model="llama3.1", temperature=0.2)
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
# Создание retriever
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 3}
)
# RAG цепочка
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff", # "stuff", "map_reduce", "refine"
retriever=retriever,
return_source_documents=True
)
# Запрос
query = "Как настроить JWT аутентификацию в FastAPI?"
result = qa_chain({"query": query})
print(f"Answer: {result['result']}")
print(f"Sources: {len(result['source_documents'])} documents")from langchain.prompts import PromptTemplate
template = """Ты — помощник для ответов на вопросы по документации.
Используй только предоставленный контекст для ответа.
Если ответа нет в контексте, скажи "Не могу найти ответ в документации".
Контекст:
{context}
Вопрос: {question}
Ответ:"""
prompt = PromptTemplate(
template=template,
input_variables=["context", "question"]
)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
chain_type_kwargs={"prompt": prompt},
return_source_documents=True
)LlamaIndex специализируется на RAG и имеет более продвинутые возможности.
pip install llama-index
pip install llama-index-vector-stores-chromafrom llama_index.core import (
VectorStoreIndex,
SimpleDirectoryReader,
Settings
)
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.llms.ollama import Ollama
# Настройки
Settings.embed_model = HuggingFaceEmbedding("sentence-transformers/all-MiniLM-L6-v2")
Settings.llm = Ollama(model="llama3.1", request_timeout=120.0)
# Загрузка документов
documents = SimpleDirectoryReader("./docs").load_data()
# Создание индекса
index = VectorStoreIndex.from_documents(documents)
# Query engine
query_engine = index.as_query_engine(similarity_top_k=3)
# Запрос
response = query_engine.query("Как настроить аутентификацию?")
print(f"Answer: {response}")
print(f"Sources: {response.source_nodes}")from llama_index.core import (
VectorStoreIndex,
StorageContext,
load_index_from_storage
)
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.retrievers import RecursiveRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.postprocessor.cohere_rerank import CohereRerank
# Node parser
parser = SentenceSplitter(chunk_size=512, chunk_overlap=50)
# Создание индекса
index = VectorStoreIndex.from_documents(
documents,
transformations=[parser]
)
# Сохранение
index.storage_context.persist("./storage")
# Загрузка
storage_context = StorageContext.from_defaults(persist_dir="./storage")
index = load_index_from_storage(storage_context)
# Retriever
retriever = index.as_retriever(
similarity_top_k=5,
vector_store_query_mode="hybrid" # keyword + semantic
)
# Reranker (опционально)
reranker = CohereRerank(
api_key="your-key",
top_n=3
)
# Query engine
query_engine = RetrieverQueryEngine(
retriever=retriever,
node_postprocessors=[reranker]
)
response = query_engine.query("Как настроить OAuth2?")Комбинация semantic + keyword поиска.
from langchain.vectorstores import Chroma
# Hybrid search
retriever = vectorstore.as_retriever(
search_type="mmr", # Maximal Marginal Relevance
search_kwargs={
"k": 5,
"fetch_k": 10,
"lambda_mult": 0.5 # 0 = keyword, 1 = semantic
}
)
# Или с разными retrievers
from langchain.retrievers import EnsembleRetriever
keyword_retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 5}
)
semantic_retriever = vectorstore.as_retriever(
search_type="mmr",
search_kwargs={"k": 5, "lambda_mult": 0.7}
)
ensemble_retriever = EnsembleRetriever(
retrievers=[keyword_retriever, semantic_retriever],
weights=[0.3, 0.7]
)from fastapi import FastAPI, HTTPException, UploadFile, File
from pydantic import BaseModel
from typing import List, Optional
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import Ollama
import tempfile
import os
app = FastAPI()
# Глобальные объекты
embeddings = None
vectorstore = None
llm = None
class QueryRequest(BaseModel):
query: str
k: int = 3
class QueryResponse(BaseModel):
answer: str
sources: List[str]
def initialize_rag():
"""Инициализация RAG компонентов"""
global embeddings, vectorstore, llm
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2"
)
llm = Ollama(model="llama3.1", temperature=0.2)
# Загрузка существующей БД или создание новой
if os.path.exists("./chroma_db"):
vectorstore = Chroma(
persist_directory="./chroma_db",
embedding_function=embeddings
)
else:
vectorstore = Chroma(embedding_function=embeddings)
@app.on_event("startup")
async def startup():
initialize_rag()
@app.post("/upload")
async def upload_document(file: UploadFile = File(...)):
"""Загрузка документа в RAG систему"""
global vectorstore
with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp:
content = await file.read()
tmp.write(content)
tmp_path = tmp.name
try:
# Загрузка и обработка
loader = PyPDFLoader(tmp_path)
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50
)
chunks = text_splitter.split_documents(documents)
# Добавление в векторную БД
vectorstore.add_documents(chunks)
vectorstore.persist()
return {"status": "ok", "chunks": len(chunks)}
finally:
os.unlink(tmp_path)
@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
"""Запрос к RAG системе"""
if vectorstore is None:
raise HTTPException(status_code=500, detail="RAG not initialized")
retriever = vectorstore.as_retriever(
search_kwargs={"k": request.k}
)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
return_source_documents=True
)
result = qa_chain({"query": request.query})
sources = [
doc.page_content[:200] + "..."
for doc in result["source_documents"]
]
return QueryResponse(
answer=result["result"],
sources=sources
)# Для документации
RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\n\n", "\n", " ", ""]
)
# Для кода
from langchain.text_splitter import Language
Language.PYTHON: {
"chunk_size": 500,
"chunk_overlap": 50,
"separators": ["\nclass ", "\ndef ", "\n\n", "\n", " ", ""]
}
# Для длинных документов
ParentDocumentRetriever(
text_splitter=RecursiveCharacterTextSplitter(chunk_size=500),
parent_splitter=RecursiveCharacterTextSplitter(chunk_size=2000)
)from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank
compressor = CohereRerank(
api_key="your-key",
top_n=3
)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=vectorstore.as_retriever()
)import hashlib
from functools import lru_cache
class RAGCache:
def __init__(self):
self.cache = {}
def _get_key(self, query: str, k: int) -> str:
return hashlib.md5(f"{query}:{k}".encode()).hexdigest()
@lru_cache(maxsize=1000)
def get(self, query: str, k: int) -> str:
key = self._get_key(query, k)
return self.cache.get(key)
def set(self, query: str, k: int, result: str):
key = self._get_key(query, k)
self.cache[key] = result
rag_cache = RAGCache()from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall
)
# Датасет для оценки
eval_dataset = {
"question": ["Как настроить JWT?", "Что такое OAuth2?"],
"answer": ["...", "..."],
"contexts": [["...", "..."], ["...", "..."]],
"ground_truth": ["...", "..."]
}
results = evaluate(
eval_dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall
]
)
print(results)RAG — мощный паттерн для работы с приватными и актуальными данными. Вы изучили:
В следующей теме вы изучите Безопасность LLM — защита от prompt injection и других уязвимостей.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.