[AI Agent] RAG를 이용한 문서 기반 LLM 생성하기

RAG

RAG ( Retrieval Augmented Generation ) 란,

LLM이 외부 지식(DB,문서 등)을 검색해서 답변을 생성하도록 하는 기법이다

 

[R] Retrieval (관련 문서 검색)

관련 문서를 검색하려면 다음과 같은 과정을 거쳐야 한다

    1. File을 Load할 수 있는 코드

    2. Load된 정보를 Chunk로 나누는 코드

    3. 나누어진 정보를 저장하는 Vector DB

    4. Vector DB를 탐색하는 Retriever

 

이렇게 하면 파일에서 정보들을 추출하여 문서기반 LLM을 생성할 수 있다

 

[A] Augmented (검색결과로 정보를 보강)

위에 Retriever가 물고온 context, 즉 문서 정보와 사용자의 질문을 합쳐 프롬포트를 생성한다는 것이다

- 이때 prompt는 langsmith에 있는 prompt opensource를 이용할 수 있따

(https://smith.langchain.com/hub/rlm/rag-prompt?organizationId=95157b0e-a178-4f08-ae15-acbde8a2b106)

 

[G] Generation (LLM이 답변 생성)

위에 생성된 prompt를 이용하며, 이는 langchain으로 쉽게 구현을 할 수 있다

* langchain : LLM 기반 애플리케이션을 더 쉽게 구축하기 위한 프레임워크

 

RAG 필수 요소

RAG를 구현하기 위해서 위에서 보았 듯이 필수 요소들이 존재한다

(나의 경우 ollama를 이용하여 LLM을 돌렸고 벡터 DB는 무료인 Chroma를 사용하였다)

• 벡터 DB : Chroma
• 텍스트 임베딩 모델: mxbai-embed-large:latest (Ollama)
• LLM (Generation, Retrieval) : gpt-4o (Ollama)

이제 기초 설명은 끝났으니 구현 하면 된다

 

1. File Loader

파일 그대로 Python으로 처리할 수 없으므로

 FileLoader를 통해 langchain Document를 만들어 준다

from langchain_excel_loader import StructuredExcelLoader
# Provide the path to your CSV file
file_path = "./data/saved_reason_solution_method_manual_sample.xlsx"
# Initialize the loader with your Excel file
loader = StructuredExcelLoader(file_path)

# Load all documents (one per sheet)
docs = loader.load()
print(f"loaded docs : {docs}")

이 외에도 여러가지 Loader 들이 존재 한다 

https://wikidocs.net/231364

Part 2. RAG (Retrieval-Augmented Generation) 기법

# CSV
from langchain_community.document_loaders.csv_loader import CSVLoader
loader = CSVLoader(file_path="./data/titanic.csv")

# PDF
from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader(FILE_PATH)

# Excel
from langchain_community.document_loaders import UnstructuredExcelLoader
loader = UnstructuredExcelLoader("./data/titanic.xlsx", mode="elements")

# Word
from langchain_community.document_loaders import Docx2txtLoader
loader = Docx2txtLoader("./data/sample-word-document.docx")  # 문서 로더 초기화

# PPT
from langchain_community.document_loaders import UnstructuredPowerPointLoader
loader = UnstructuredPowerPointLoader("./data/sample-ppt.pptx")

# txt
from langchain_community.document_loaders import TextLoader
loader = TextLoader("data/appendix-keywords.txt")

# WEB
import bs4
from langchain_community.document_loaders import WebBaseLoader
# 뉴스기사 내용을 로드합니다.
loader = WebBaseLoader(
    web_paths=("https://n.news.naver.com/article/437/0000378416",),
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            "div",
            attrs={"class": ["newsct_article _article_body", "media_end_head_title"]},
        )
    ),
    header_template={
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.0.0.0 Safari/537.36",
    },
)
docs = loader.load()

 

 

 

2. Chunking (Text Splitter)

from langchain_text_splitters import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=300, 
    chunk_overlap=20,
    length_function=len, #길이를 보고 싶을 때 사용하는 함수
    is_separator_regex=True #구분자를 text로 볼건지 정규표현식으로 볼 건지 
    )
all_splits = text_splitter.split_documents(docs)
print(f"Type of Chunk : {type(all_splits)}, {type(all_splits[0])}")
for i, split in enumerate(all_splits):
    print(f"Split {i+1}")
    print(split)
    break

위에서 불러온  Docs를 Chunk 단위로 나누어 준다. (임베딩을 하기 위해서)

- 한 페이지를 한꺼번에 임베딩하게 되면 벡터 데이터베이스에서 관련된 정보를 찾기 어려움

- 의미 있는 정보가 묶인 덩어리 (Chunk)로 문서를 분할해야 한다

 

RecursiveCharacterTextSplitter Hyperparameter

- chunk_size : 각 청크의 최대 길이

- chunk_overlap : 문맥을 유지하기 위한 인접한 청크 사이에 중복되는 영역

- length_function : 청크에 길이를 측정하는 함수 정의

- is_seperator_regex : True :정규식표현을 통해 구분자 처리 / False : 구분자를 단순한 문자열로 해석

 

3. Text Embedding (Embedding Model )

Text로 되어진 데이터를 Vector DB에 넣기 위해서 Vector로 바꿔준다

- 문서에 연관된 정보를 정확하게 추출하기 위함

이를 위해 Embedding Model을 가져오자

import os
from dotenv import load_dotenv
load_dotenv()
base_url = os.getenv('base_url')
api_key = os.getenv('api_key')

from langchain_openai import OpenAIEmbeddings
embedding_model = OpenAIEmbeddings(
    model='jeffh/intfloat-multilingual-e5-large-instruct:f32', 
    api_key=api_key,
    base_url=base_url,
    # OpenAI 호환 서버 안정 옵션
    tiktoken_enabled=False, # OpenAI 제공 api 아니면 필요
    check_embedding_ctx_length=False, # OpenAI 제공 api 아니면 필요
    )

tiktoken_enbled 과 check_embedding_ctx_length 같은 경우

OpenAI  API를 직접 사용하는 게 아니라

Ollama를 통해 사용하고 있으면 추가해줘야 하는 옵션인다

BadRequestError: Error code: 400 - {'error': {'message': 'invalid input type', 'type': 'api_error', 'param': None, 'code': None}}

없으면 이 에러 뜸

tiktoken_enbled

- Langchain이 tiktoken으로 입력 텍스트의 토큰 수를 미리 계산 (여기에서 입력 타입이 꼬이거나 호환 문제가 생김)

check_embedding_ctx_length

- embedding 모델의 최대 토큰 길이 초과 여부를 미리 계산 (여기에서도 tiktoken 경로를 탐)

여튼 Ollama 서버를 쓰면 추가 하도록 하자

 

4. Vector DB (Vector Store)

from langchain_chroma import Chroma
# -------------------------------
persist_directory = "./chroma_store"
collection_name = "reason_manual_v1"

vectorstore = Chroma.from_documents(
    documents=all_splits,
    embedding=embedding_model,
    persist_directory=persist_directory,
    collection_name=collection_name,
)

Chroma : 무료 오픈 소스

https://wikidocs.net/234094

01. Chroma

 

 

5. Retriever

Retriever 를 활용하여서 여러가지 Chunk들 중 사용자의 질문과 가장 관련있는 정보만 가져오는 방법이다

즉 검색기를 이용하여 Vector DB 를 탐색하자

# Multi-Query Retriever
# 사용자의 질문을 여러 개의 유사 질문으로 재생성 

# langchain v0.x
#from langchain.retrievers.multi_query import MultiQueryRetriever
# langchain v1.x
from langchain_classic.retrievers import MultiQueryRetriever
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
        openai_api_base=base_url,
        openai_api_key=api_key,
        model="gpt-oss:20b",
        temperature=0
    )

retriever_from_llm = MultiQueryRetriever.from_llm(
    retriever= vectorstore.as_retriever(),
    llm = llm
)

user_prompt = "MIB2.GET.ETH_SW-VER.의 원인과 조치 방안을 알려줘"
docs = retriever_from_llm.invoke(user_prompt)
print(f" 검색된 Chunk의 개수 : {len(docs)}")
print(docs[0].page_content)

# 검색된 문서들을 하나로 합쳐 줌
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

 

 

MultiQueryRetriever

추가로  MultiQueryRetriever를 이용하여

LLM을 사용하여 사용자 질문을 다양한 가짓수로 생성함으로써, 보다 관련성 높은 결과를 제공한다

만약 어떤 질문들을 생성하는지 보고 싶으면 MultiQueryRetriever 에 구현되어 있는 query 생성 체인을 이용하여 확인 가능 하다

user_prompt = "MIB2 ETH_SW-VER의 원인과 조치 방안을 알려줘"
pipeline = RAGPipeline(file_path,StructuredExcelLoader)
pipeline.setup_rag_chain()
pipeline.retriever_from_llm.llm_chain.invoke({"question" : user_prompt})

 

https://asidefine.tistory.com/298

LangChain RAG Retriever 방법 정리 (Multi-Query, Parent Document, Ensemble Retriever, ... )

만약 여기서 생성한 쿼리들이 Noise가 될거 같거나 만족 스럽지 않는다면

1. BM25Retriever + VectorRetriever

2. EnsembleRetriever

3. Reranker

4. 최종 top_n 문서 뽑기

등으로 고도화할 수 있다

 

Error Shooting ( ModuleNotFoundError )

추가로 옛날 책을 보면

from langchain.retrievers.multi_query import MultiQueryRetriever

이런 식으로 import 하는데 아래와 같은 오류가 나오므로

ModuleNotFoundError
: No module named 'langchain.retrievers'

Langchain ver1.0 이상 부터는 langchain_classic을 이용해서 import 해주자

from langchain_classic.retrievers import MultiQueryRetriever

 

 

6. Prompt Augmentation 

# from langchain import hub
# (v1.x)ImportError: cannot import name 'hub' from 'langchain'
from langchain_classic import hub
prompt = hub.pull("rlm/rag-prompt")

답변을 생성하기 위한 프롬포트 템플릿을 "Prompt Hub"를 통해 가져온다

(프롬포트에 정보를 붙이는 과정이라 생각하면 된다)
다른 프롬포트들도 구경하고 싶으면 아래 링크를 참고하자

https://smith.langchain.com/hub/rlm/rag-prompt?organizationId=95157b0e-a178-4f08-ae15-acbde8a2b106

 

 

ErrorShooting (ImportError)

기존 자료 들 (langchain v0.x)의 경우는 from langchain import hub 를 통해 가져오지만

v1.x 으로 langchain이 업데이트 되면서 langchain_classic으로 빠졌다

기존에는 있었는데 지금은 왜 import 가 안되지?

ImportError
: cannot import name 'hub' from 'langchain' (/home/seungjong.yoo/.pyenv/versions/Agent/lib/python3.10/site-packages/langchain/__init__.py)

하면 대부분 langchain_classic 에 빠져있다고 생각하면 되다

 

 

7. Generator 구현

from langchain_core.runnables import RunnablePassthrough
#RunnablePassthrough : invoke에 있는 사용자 입력을 그대로 전달
from langchain_core.output_parsers import StrOutputParser #출력파서
rag_chain = (
    {"context": retriever_from_llm | format_docs, 
     "question" : RunnablePassthrough()
    }
     | prompt
     | llm
     | StrOutputParser()
)

이제 Langchain을 이용해서 답변기 (Generator)를 만들면 된다

- Generator :답변을 생성하는 생성기

StrOutputParser의 경우 출력 파서 이고

RunnablePassthrough 인스턴스는 invoke()함수의 사용자 입력을 그대로 전달하는 역활이다

 

8. invoke(실행)

result = rag_chain.invoke(user_prompt)
result

그러고 langchain을 invoke() 함수를 통해 실행시켜주면 된다

 

 

실행 코드 정리

지금까지 설명한 코드를

이용하기 쉽게 정리한다면

전체 코드

'''
1. Docs Load 
'''

from langchain_excel_loader import StructuredExcelLoader
# Provide the path to your CSV file
file_path = "./data/saved_reason_solution_method_manual_sample.xlsx"
# Initialize the loader with your Excel file
loader = StructuredExcelLoader(file_path)

# Load all documents (one per sheet)
docs = loader.load()
print(f"loaded docs : {docs}")

'''
2. Split with Chunk
'''
from langchain_text_splitters import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=300, 
    chunk_overlap=20,
    length_function=len, #길이를 보고 싶을 때 사용하는 함수
    is_separator_regex=True #구분자를 text로 볼건지 정규표현식으로 볼 건지 
    )
all_splits = text_splitter.split_documents(docs)
print(f"Type of Chunk : {type(all_splits)}, {type(all_splits[0])}")
for i, split in enumerate(all_splits):
    print(f"Split {i+1}")
    print(split)
    break

'''
3. Text Embedding
'''
import os

from dotenv import load_dotenv
load_dotenv()
base_url = os.getenv('base_url')
api_key = os.getenv('api_key')




from langchain_openai import OpenAIEmbeddings
embedding_model = OpenAIEmbeddings(
    model='jeffh/intfloat-multilingual-e5-large-instruct:f32', 
    api_key=api_key,
    base_url=base_url,

    # OpenAI 호환 서버 안정 옵션
    tiktoken_enabled=False, # OpenAI 제공 api 아니면 필요
    check_embedding_ctx_length=False, # OpenAI 제공 api 아니면 필요
    )

'''
4. Vector Store 호출
'''
from langchain_chroma import Chroma
# -------------------------------
persist_directory = "./chroma_store"
collection_name = "reason_manual_v1"

vectorstore = Chroma.from_documents(
    documents=all_splits,
    embedding=embedding_model,
    persist_directory=persist_directory,
    collection_name=collection_name,
)

'''
5. [R]검색기(Retriever 구현)
    - MultiQueryRetruever를 통한 질문 증강
'''
# Multi-Query Retriever
# 사용자의 질문을 여러 개의 유사 질문으로 재생성 

# langchain v0.x
#from langchain.retrievers.multi_query import MultiQueryRetriever
# langchain v1.x
from langchain_classic.retrievers import MultiQueryRetriever
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
        openai_api_base=base_url,
        openai_api_key=api_key,
        model="gpt-oss:20b",
        temperature=0
    )

retriever_from_llm = MultiQueryRetriever.from_llm(
    retriever= vectorstore.as_retriever(),
    llm = llm
)

user_prompt = "MIB2.GET.ETH_SW-VER.의 원인과 조치 방안을 알려줘"
docs = retriever_from_llm.invoke(user_prompt)
print(f" 검색된 Chunk의 개수 : {len(docs)}")
print(docs[0].page_content)

# 검색된 문서들을 하나로 합쳐 줌
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)


'''
6. [A] Augmentation 질문 증강 (prompt 강화)
'''
# from langchain import hub
# (v1.x)ImportError: cannot import name 'hub' from 'langchain'
from langchain_classic import hub
prompt = hub.pull("rlm/rag-prompt")

'''
7. [G] 생성기(Generator) 구현
    - langchain 구성
'''
from langchain_core.runnables import RunnablePassthrough
#RunnablePassthrough : invoke에 있는 사용자 입력을 그대로 전달
from langchain_core.output_parsers import StrOutputParser #출력파서
rag_chain = (
    {"context": retriever_from_llm | format_docs, 
     "question" : RunnablePassthrough()
    }
     | prompt
     | llm
     | StrOutputParser()
)

'''
8. 실행부
'''
result = rag_chain.invoke(user_prompt)
result

 

 

class로 구현하기

from langchain_excel_loader import StructuredExcelLoader
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_classic.retrievers import MultiQueryRetriever
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser #출력파서
from langchain_classic import hub
import os
from dotenv import load_dotenv
load_dotenv()
base_url = os.getenv('base_url')
api_key = os.getenv('api_key')
file_path = "./data/saved_reason_solution_method_manual_sample.xlsx"

class RAGPipeline:
    def __init__(self, file_path, loader, chunk_size=300, chunk_overlap=20):
        self.file_path = file_path
        self.persist_directory = "./chroma_store"
        self.collection_name = "reason_manual_v1"
        self.model_name = "gpt-oss:20b"
        self.embedding_model_name = 'jeffh/intfloat-multilingual-e5-large-instruct:f32'
        self.langchain_hub_prompt = "rlm/rag-prompt"
        self.docs = self.make_docs_with_loader(loader)
                    
        embedding_model = OpenAIEmbeddings(
            model= self.embedding_model_name,
            api_key=api_key,
            base_url=base_url,
            tiktoken_enabled=False,
            check_embedding_ctx_length=False,
        )
        
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size, 
            chunk_overlap=chunk_overlap,
            length_function=len,
            is_separator_regex=True
        )
        
        self.vectorstore = Chroma.from_documents(
            documents=text_splitter.split_documents(self.docs),
            embedding=embedding_model,
            persist_directory=self.persist_directory,
            collection_name=self.collection_name,
        )
        
        self.llm = ChatOpenAI(
            openai_api_base=base_url,
            openai_api_key=api_key,
            model=self.model_name,
            temperature=0
        )
        
        self.retriever_from_llm = MultiQueryRetriever.from_llm(
            retriever=self.vectorstore.as_retriever(),
            llm=self.llm
        )

        
    def __str__(self):
        return f"""
    base model name [Retriever, Genrator]] : {self.model_name}
    embedding model name : {self.embedding_model_name}
    langchain_hub_prompt : {self.langchain_hub_prompt}
    """

    def make_docs_with_loader(self,loader):
        if loader == PyPDFLoader:
            return loader(self.file_path).load_and_split()
        else:
            return loader(self.file_path).load()

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

    def setup_rag_chain(self):
        prompt = hub.pull(self.langchain_hub_prompt)
        self.rag_chain = (
            {"context": self.retriever_from_llm | self.format_docs, 
             "question": RunnablePassthrough()
            }
             | prompt
             | self.llm
             | StrOutputParser()
        )

    def run(self, user_prompt):
        docs = self.retriever_from_llm.invoke(user_prompt)
        print(f" 검색된 Chunk의 개수 : {len(docs)}")
        print(docs[0].page_content)
        result = self.rag_chain.invoke(user_prompt)
        return result
        
        
if __name__ == "__main__":  
    pipeline = RAGPipeline(file_path,StructuredExcelLoader)
    pipeline.setup_rag_chain()
    user_prompt = "MIB2 ETH_SW-VER의 원인과 조치 방안을 알려줘"
    result = pipeline.run(user_prompt)
    print(result)

 

 

참고 자료

https://velog.io/@one_two_three/RAG%EB%9E%80-%EB%AC%B4%EC%97%87%EC%9D%B8%EA%B0%80

RAG는 무엇인가? RAG를 간단하게 구현해보자

https://wikidocs.net/234094

01. Chroma

https://wikidocs.net/234016

01. 벡터스토어 기반 검색기(VectorStore-backed Retriever)

https://smith.langchain.com/hub/rlm/rag-prompt?organizationId=95157b0e-a178-4f08-ae15-acbde8a2b106

LangSmith

https://wikidocs.net/253706

01. 도큐먼트(Document) 의 구조

https://wikidocs.net/231364

Part 2. RAG (Retrieval-Augmented Generation) 기법

https://mininkorea.tistory.com/83

Python으로 검색 엔진 성능 비교하기: FAISS vs ChromaDB

https://reference.langchain.com/python#MultiQueryRetriever.generate_queries