[AI 서비스 개발] RAG 파이프라인

1. RAG 파이프라인의 구성

1) 데이터 로드(Load Data)

- 외부 데이터 소스에서 정보를 수집하고, 필요한 형식으로 변환하여 시스템에 로드

- 공개 데이터셋, 웹 크롤링을 통해 얻은 데이터, 또는 사전에 정리된 자료

from langchain_community.document_loaders import WebBaseLoader

# 크롤링 하고 싶은 url
url = 'https://wikidocs.net/231393'

loader = WebBaseLoader(url)

docs = loader.load()
print(len(docs)) # 1
print(len(docs[0].page_content)) # 15735
print(docs[0].page_content)

 

2) 텍스트 분할(Text Split)

- 불러온 데이터를 작은 크기의 단위(chunk)로 분할하는 과정

- 자연어 처리(NLP) 기술을 활용해 문단, 문장, 또는 구 단위로 나누는 작업

- 검색 효율성을 높이기 위한 과정

- 왜 필요할까? LLM이나 API의 입력 크기 제한이 있고, 프롬프트가 지나치게 길면 중요한 정보가 희석될 수 있어서.

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits= text_splitter.split_documents(docs)

print(len(splits)) # 19
print(splits[10]) # Documenmt 객체 전체를 출력
splits[10].page_content # 본문 텍스트
splits[10].metadata # 추가 정보

# {'source': 'https://wikidocs.net/231393',
#  'title': '2-1. RAG 개요 - 랭체인(LangChain) 입문부터 응용까지',
#  'description': 'RAG(Retrieval-Augmented Generation) 파이프라인은 기존의 언어 모델에 검색 기능을 추가하여, 주어진 질문이나 문제에 대해 더 정확하고 풍부한 정보를 기…',
#  'language': 'ko'}

 

3) 인덱싱(Indexing)

* 임베딩: 텍스트를 벡터(숫자 배열)로 변환하는 과정

- 변환된 임베딩 벡터들을 검색이 용이하도록 저장하고, 이를 기반으로 유사성 검색(예: Chroma 벡터 DB활용)을 수행할 수 있게 만드는 과정

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

vectorsotre = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())
docs = vectorsotre.similarity_search("인덱싱에 대해 설명해주세요")
print(len(docs))
print(docs[0].page_content)

 

4) 검색(Retrieval)

- 사용자의 질문이나 주어진 컨텍스트에 가장 관련된 정보를 찾는 과정

- 사용자의 입력을 바탕으로 쿼리를 생성하고, 인덱싱된 데이터에서 가장 높은 정보를 검색 : retriever 메소드 사용

 

5) 생성(Generation)

- 검색된 정보를 바탕으로 사용자의 질문에 답변을 생성하는 최종 단계

- LLM 모델에 검색 결과와 함께 사용자의 입력을 전달

- 모델은 사전 학습된 지식 + 검색 결과를 결합해 질문에 가장 적절한 답변을 생성

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

vectorsotre = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())
docs = vectorsotre.similarity_search("인덱싱에 대해 설명해주세요")
print(len(docs))
print(docs[0].page_content)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

template = '''
    항상 다음의 context에 기반하여 답변하세요.
    : {context}

    Question : {question}    
'''

prompt = ChatPromptTemplate.from_template(template)


model = ChatOpenAI(model='gpt-4o-mini', temperature=0)

# Retriever
retriever = vectorsotre.as_retriever()

# combine Documents
def format_docs(docs):
    return '\n\n'.join(doc.page_content for doc in docs)

# RAG Chain 연결
rag_chain = (
    {
        'context': retriever | format_docs, 'question': RunnablePassthrough()
    }
    | prompt
    | model
    | StrOutputParser()
)

# Chain 실행
rag_chain.invoke("인덱싱에 대해 설명해주세요")

 

※ 참고 문서: https://wikidocs.net/231393