검색 증강 생성(RAG) - 개념 (1)

1. RAG

 

올라마(Ollama)에서 임베딩 개념을 이야기하면서 RAG 개념이 시작되었다.

 

임베딩 모델을 통해 어떤 input을 고차원의 숫자 리스트로 바꾸어줬고,

 

이 임베딩된 숫자를 통해 RAG라는 것을 할 수 있다고 공부했었다.

 

여기서 RAG는 Retrieval Augmented Generation의 약자로써, 검색 증강 생성을 의미한다.

 

생성형 AI가 임베딩한 외부 문서를 검색한 뒤 그 내용을 기반으로 답변한다.

 

1) 문서 처리 : DocumentLoader 

대표적으로, PyPDFLoader, JSONLoader, TextLoader, WebBaseLoader 등이 있다.

 

1. PyPDFLoader

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("../data/서울대 미대.pdf")
docs = loader.load()

print(f"총 페이지 수 : {len(docs)}")
print(f"첫 페이지 텍스트 : {docs[0].page_content[:100]}")
print(f"메타 데이타 : {docs[0].metadata}")

 

2. CSVLoader

from langchain_community.document_loaders import CSVLoader

csv_loader = CSVLoader("../data/restaurant_reviews.csv", encoding="utf-8", csv_args={'delimiter': ","})
csv_docs = csv_loader.load()

print(f"총 페이지 수 : {len(csv_docs)}")
print(f"첫 페이지 텍스트 : {csv_docs[0].page_content[:100]}")
print(f"메타 데이타 : {csv_docs[0].metadata}")

 

3. WebBaseLoader

# BeautifulSoup4 라이브러리 필요
from langchain_community.document_loaders import WebBaseLoader

web_loader = WebBaseLoader(web_paths=['https://python.org','https://langchain.com'])
web_docs = web_loader.load()

print(f"총 페이지 수 : {len(web_docs)}")
print(f"첫 페이지 텍스트 : {web_docs[0].page_content[:100]}")
print(f"메타 데이타 : {web_docs[0].metadata}")

 

4. DirectoryLoader

from langchain_community.document_loaders import DirectoryLoader

dir_loader = DirectoryLoader(path="../data", glob="**/*.pdf", loader_cls=PyPDFLoader, show_progress=True)
dir_docs = dir_loader.load()

print(f"총 페이지 수 : {len(dir_docs)}")
print(f"첫 페이지 텍스트 : {dir_docs[0].page_content[:100]}")
print(f"메타 데이타 : {dir_docs[0].metadata}")

 

5. 유튜브 스크립트 가져오기 예제

from youtube_transcript_api import YouTubeTranscriptApi

video_id = 'VRQzJMPf1ug'

yt_api = YouTubeTranscriptApi()

transcript = yt_api.fetch(video_id, languages=['ko'])

for idx, t in enumerate(transcript, 1):
    print(f"{idx} : {t.text}")
    print(f"{idx} : {t.start}")
    print(f"{idx} : {t.duration}")

text = " ".join(t.text for t in transcript)

print(text)

doc = Document(page_content=text, metadata={
    "source": f"https://www.youtube.com/watch?v={video_id}",
    "video_id": video_id
})

print(docs.page_content[:100])
print(docs.metadata)

 

2) 문서 쪼개기 : RecursiveCharacterTextSplitter

 

일반적인 RAG에서 주로 사용하며, LLM의 Context Window이 제한이 있어

 

문서를 한 번에 넣을 수 없으므로 TextSplitter 이용해 적절한 크기의 청크로 분할한다.

 

 chunk_size (청크 최대 크기), chunk_overlap (청크 겹치는 부분 - 문맥 유지) 개념이 있다.

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 1. 문서 로드
loader = PyPDFLoader("../data/서울대 미대.pdf")
docs = loader.load()

# 2. 문서 분할
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500, chunk_overlap=50, length_function=len, separators=['\n\n','\n',' ','']
)

chunks = splitter.split_documents(docs)

print(f"원본 페이지 수 : {len(docs)}")
print(f"분할한 청크 수 : {len(chunks)}")
print(f"첫 청크 길이 : {len(chunks[0].page_content)}")
print(f"첫 청크 내용 : {chunks[0].page_content}")
print(f"메타 데이타 : {chunks[0].metadata}")

 

3) 청크 파라미터

 

overlap이 0일 경우, 앞의 문맥에 대해 알지 못해

 

LLM에게 앞선 컨텍스트를 전달하지 못하는 문제를 확인하기 위함이다.

 

overlap이 많을수록 앞선 내용에 대해 기억하는 경우가 많을 것이다.

# chunk 파라미터 비교

text = "파이썬은 1991년에 발표된 언어이다" * 20

configs = [
    { "chunk_size": 100, "chunk_overlap": 0},
    { "chunk_size": 100, "chunk_overlap": 20},
    { "chunk_size": 200, "chunk_overlap": 50}
]

for cfg in configs:
    sp = RecursiveCharacterTextSplitter(**cfg)
    chunks = sp.split_text(text)
    print(f"size={cfg['chunk_size']}, overlap={cfg['chunk_overlap']}")
    print(f"{len(chunks)}개 청크, 첫 청크 : {chunks[0][:30]}....")
    if len(chunks) > 1:
        print(f"첫 청크 끝 : '{chunks[0][-20:]}'")
        print(f"둘째 청크 시작 : '{chunks[1][:20]}'")

 

4) 임베딩(Embedding)

 

텍스트를 고차원 벡터로 변환하는 과정이며, 의미가 비슷할수록 벡터가 가깝다.

 

RAG에서 임베딩은 다음과 같을 때 사용한다.

• 문서 청크를 저장할 때 사용
• 질문 검색할 때 사용

from langchain_ollama import OllamaEmbeddings

ollama_embedding = OllamaEmbeddings(model='nomic-embed-text-v2-moe')

# 단일 텍스트 쿼리
vector = ollama_embedding.embed_query("파이썬이란?")

print(f"벡터 차원 : {len(vector)}")
print(f"첫 5개 값 : {vector[:5]}")

texts = ["파이썬이란?", "자바란 무엇인가?", "오늘 날씨는?"]

# 여러 개 쿼리
vectors = ollama_embedding.embed_documents(texts)

print(f"임베딩 문서 수 : {len(vectors)}")

 

이제 이 벡터값을 가지고 벡터 유사도를 추출해 볼수 있다. (여기선 코사인 유사도 사용)

# 벡터 유사도
import numpy as np

def cosine_similarity(a, b):
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

vec1 = ollama_embedding.embed_query("파이썬이란?")
vec2 = ollama_embedding.embed_query("python이란?")
vec3 = ollama_embedding.embed_query("오늘 저녁 뭐 먹지?")

print(f"파이썬 vs Python : {cosine_similarity(vec1, vec2):.3f}")
print(f"파이썬 vs 저녁 : {cosine_similarity(vec1, vec3):.3f}")

 

5) 벡터 DB

 

메모리가 아닌 DB에 벡터 값을 벡터 저장소에 저장한다. 여기서는 chromaDB와 faiss를 사용할 것이다.

chromaDB

from langchain_chroma import Chroma

# 1. 문서 로드
loader = PyPDFLoader("../data/서울대 미대.pdf")
docs = loader.load()

# 2. 문서 분할
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50, length_function=len, separators=['\n\n','\n',' ',''])

chunks = splitter.split_documents(docs)

# 3. 임베딩 처리
ollama_embedding = OllamaEmbeddings(model='nomic-embed-text-v2-moe')

# 4. 벡터 스토어에 저장(chromaDB)
vector_store = Chroma.from_documents(
    documents=chunks,
    embedding=ollama_embedding,
    persist_directory="./db/chroma_db",
    collection_name="my_docs"
)

 

# 5. 검색(Retrieval)
# 질문도 벡터로 변경해야 함, 유사도 높은 걸로 몇개 추출할 것인지 설정

results = vector_store.similarity_search("근로장학생 신청 절차는?",k=3)

for idx, doc in enumerate(results, 1):
    print(f"\n[결과] {idx} 출처 : {doc.metadata}")
    print(doc.page_content[:300])

# DB 확인

data = vector_store.get()

print(data['documents'][0][:300])
print(data['metadatas'][0])
print(f"청크 개수 : {vector_store._collection.count()}")

# 저장 후 불러오기
chromadb = Chroma(persist_directory="./db/chroma_db", embedding_function=ollama_embedding, collection_name="my_docs")

 

faiss

# faiss
from langchain_community.vectorstores import FAISS

# 1. 문서 로드
loader = PyPDFLoader("../data/서울대 미대.pdf")
docs = loader.load()

# 2. 문서 분할
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50, length_function=len, separators=['\n\n','\n',' ',''])

chunks = splitter.split_documents(docs)

# 3. 임베딩 처리
ollama_embedding = OllamaEmbeddings(model='nomic-embed-text-v2-moe')

# 벡터 스토어에 저장(faiss)
faiss_vector_store = FAISS.from_documents(documents=chunks, embedding=ollama_embedding)

results_score = faiss_vector_store.similarity_search_with_score("수강신청 내역 확인 방법은?",k=3)

for doc, score in results_score:
    print(f"유사도 : {score:.4f}, {doc.page_content[:20]}")

# 로컬 저장
faiss_vector_store.save_local("./db/faiss_index")

# 저장 후 불러오기
faissdb = FAISS.load_local("./db/faiss_index",embeddings=ollama_embedding, allow_dangerous_deserialization=True)

 

6) LLM에게 전송

 

추가적으로 위에서 나온 결과를 LLM에게 전송해야 한다. (다듬는 과정)

# 1. 문서 로드
pdf_doc = PyPDFLoader("../data/Summary of ChatGPTGPT-4 Research.pdf")
loader = pdf_doc.load()

# 2. 문서 분할
splitter = RecursiveCharacterTextSplitter(chunk_size=500,chunk_overlap=50)
chunks = splitter.split_documents(loader)

# 3. 인덱싱 (임베딩)
embeddings = OllamaEmbeddings(model='nomic-embed-text-v2-moe')

# 4. 벡터 스토어(chromaDB)
chroma_store = Chroma.from_documents(documents=chunks,embedding=embeddings,persist_directory="./db/chroma_db", collection_name="research")

# 5. as_retriever() : 벡터 스토어를 검색 가능한 Retriever 형태로 변환 (랭체인과 연동)
retriever = chroma_store.as_retriever(search_type="similarity", search_kwargs={"k": 3})

system_message = """\
다음 컨텍스트를 참고하여 질문에 답하세요.
컨텍스트에 없는 내용은 모른다고 답하세요.

컨텍스트:
{context}

"""

# 6. RAG 프롬프트 생성
prompt = ChatPromptTemplate.from_messages([
    ("system",system_message),
    ("human","{question}")
])

qwen_llm = ChatOllama(model='qwen2.5')

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = {
    "context": retriever | format_docs,
    "question": RunnablePassthrough()
} | prompt | qwen_llm | StrOutputParser()

ans = rag_chain.invoke("Where can I use ChatGPT?")

print(ans)

 

출처 페이지를 함께 병렬 계산하고 싶은 경우 아래와 같이 처리한다.

# 출처 페이지 정보와 답변 함께 반환
def format_docs_with_source(docs):
    result = []
    for doc in docs:
        src = doc.metadata.get("source","알 수 없음")
        page = doc.metadata.get("page","")
        result.append(f"[출처 : {src} p.{page}\n{doc.page_content}]")
    return "\n\n".join(result)
    
rag_chain = {
    "context": retriever | format_docs_with_source,
    "question": RunnablePassthrough()
} | prompt | qwen_llm | StrOutputParser()

rag_with_source = RunnableParallel(answer=rag_chain, sources=retriever)
result = rag_with_source.invoke("where can I use ChatGPT?")

print("======답변======")
print(result['answer'])
print("======출처======")
for doc in result['sources']:
    print(f" - {doc.metadata.get('source')} p.{doc.metadata.get('page','')}")