[LangChain] RAG with Pinecone

 

 LLM이 사전 학습한 정보를 벗어난 데이터의 범위를 벗어나는 경우, 원하는 대답을 얻을 수 없다. 또한, 생성 모델의 한계상 할루시네이션이 발생해 정확한 정보를 얻기 어려울 때도 있다.

 

 우리가 원하는 정보를 얻기 위해 LLM을 Supervised fine-tuning하거나, Instruction fine-tuning을 하거나, RAG를 사용할 수 있다. SFT의 경우, 현실적으로 비용 문제 때문에 접근성이 떨어지기 때문에 제외하고 Instruction fine-tuning이나 RAG를 사용하게 된다.

 

 Instruction fine-tuning은 LLM을 Instruction-output 데이터셋으로 사용자 질의와 답변을 align 시켜주는 방법이다. RAG는 사용자 질문 의도에 적절한 외부 문서를 가져와 프롬프트에 넣어서 외부 정보를 맥락(in-context)으로 모델이 참고할 수 있도록 도와준다.

이번에는 랭체인을 이용한 간단한 RAG를 구성해보자. RAG를 위한 프로세스는 다음과 같은 순서로 진행된다.

 

Indexing(Load, Split, Store) → Retrieve → Generate

 

1. Load: 랭체인의 document_loaders 로 여러 유형의 파일을 parsing
2. Split: 큰 문서를 작은 chunk로 분할. 큰 chunk는 검색하기 어렵고 모델의 context window를 고려해야하기 때문에 필요
3. Store: 문서를 검색할 수 있도록 chunk를 저장하고 인덱싱하기 위해 벡터스토어에 임베딩 벡터 형태로 저장
4. Retrieve: 사용자 input을 받아, 연관있는 top-k개의 문서를 가져옴
5. Generate: LLM이 가져온 데이터와 input을 고려한 prompt로부터 답변을 생성

1. Pinecone에 문서 저장하기

 

[키보드] 키 위치 변경하기

 

 여기서는 위 글에 작성된 본문을 복사해서 test.txt에 저장했다고 가정한다(WebBaseLoader로 웹으로부터 데이터를 가져오거나, 에이전트를 활용해 검색한 자료를 가져오는 등의 작업을 수행하는 등으로 확장가능하다).

 

 Pinecone과 OpenAIChat모델을 사용할 예정이므로, 먼저 Pinecone의 계정을 생성하고 API key를 발급 받아야 한다. 그리고, Pinecone DataBase에 Index를 하나 생성하고 해당 값들을 .env 파일에 저장해놓자. 해당 Index는 사용할 LLM 모델이 사용하는 임베딩 모델의 차원을 고려해서 임베딩 차원을 설정해놓자.

 

# .env
OPENAI_API_KEY=sk-...  
INDEX_NAME=<인덱스명>
PINECONE_API_KEY=...

1.1 Load

 먼저 TextLoader로 txt 타입의 파일을 로드한다.

loader = TextLoader("test.txt", encoding="UTF8")  
document = loader.load()

 

 document는 리스트로 하나의 파일만 불러왔으므로 1개의 Document 객체만 갖는다.

[Document(page_content='직장과 집, 휴대용 이런저런 키보드를 ...)]

 

 Document 객체는 page_content와 metadata 를 키로 갖는데, page_content는 말그대로 해당 문서의 내용이며, 기본적으로 metadata에는 해당 파일의 source를 갖는다. metadata에는 이외의 원하는 다른 내용도 추가할 수 있다. 더 자세한 내용은 앞선 글(https://sean-j.tistory.com/37)을 참고하자.

1.2. Split

 위에서 살펴봤듯, 불러온 문서를 적당한 크기로 분할해야 한다. 이때, Chunk의 크기는 적당해야 하며 너무 크지도 너무 작지도 않아야 한다. Gemini처럼 백만 개씩 토큰을 사용할 수 있는 모델이 있지만, chunk가 큰 경우 오히려 불필요한 정보까지 전달해 GIGO의 결과를 낼 수 있다. 또한, chunk가 너무 작다면 의미를 담지 못할 수 있다. 따라서, 적당히 하나의 의미를 담는 크기를 정해야 하는데, 이는 실험적으로 결정해야 한다.

 

 문서를 분할하는 방법에는 여러 가지가 있다. chunk 크기가 작아질 때까지 재귀적으로 분할하는RecursiveCharacterTextSplitter, 의미론적 유사성에 기반해 분할하는 #SemanticChunker 등이 있다. 여기서는 단순히 "\n\n" 을 기준으로 텍스트를 분할하고, chunk 크기를 문자 수로 측정하는 CharacterTextSplitter를 사용해보자.

 

text_splitter = CharacterTextSplitter(chunk_size=500, chunk_overlap=0)  
texts = text_splitter.split_documents(document) # List[Document]

[Document(page_content='직장과 집, 휴대용 이런저런 키보드를...', metadata={'source': 'test.txt'}), 
 Document(page_content='위 - 68배열 (tester68, 저소음 회축) ...',)
]

1.3. Store

 Pinecone에 임베딩 벡터로 저장해보자. from_documents 메소드는 Document 객체들을 원소로 갖는 리스트를 입력 받아 반복문을 순회하며 임베딩 모델로 각 Document.page_content를 임베딩 벡터로 변환하고, metadata와 함께 Pinecone 벡터 스토어에 저장한다.

from langchain_pinecone import PineconeVectorStore


embeddings = OpenAIEmbeddings()

PineconeVectorStore.from_documents(  
    documents=texts, embedding=embeddings, index_name=os.getenv("INDEX_NAME")  
)

 

 

---

 

만약 RAG를 사용하지 않는다면..

 앞서 Pincone에 불러온 text를 Chunking해서 임베딩된 벡터로 변환하여 저장했다. 이제 저장한 벡터를 수집(retreive)해와 prompt에 추가(augmentation)하고 LLM을 통해 답변을 생성(Generation)해보자.

 

 그 전에, 만약 RAG를 사용하지 않은 경우의 답변을 살펴보자.

embeddings = OpenAIEmbeddings()
llm = ChatOpenAI()

query = "내가 원하는 대로 키매핑하려면 무슨 코드를 사용해야 돼?"

chain = PromptTemplate.from_template(query) | llm

print(chain.invoke(input={}).content)

키매핑이란 특정 키에 대응하는 값을 지정하는 것을 말합니다. 파이썬에서는 딕셔너리를 사용하여 키매핑을 할 수 있습니다. 

예를 들어, 다음과 같이 딕셔너리를 사용하여 키매핑을 할 수 있습니다.

\```python
key_mapping = {
    'apple': '사과',
    'banana': '바나나',
    'cherry': '체리'
}

# 'apple'에 대응하는 값을 출력
print(key_mapping['apple'])  # 출력 결과: '사과'
\```

이 코드에서는 'apple'이라는 키에 대응하는 값을 '사과'로 지정하고, 이 값을 출력하는 예제입니다. 이와 같이 딕셔너리를 사용하여 원하는 대로 키매핑을 할 수 있습니다.

 

 chatgpt-3.5-turbo 모델은 해당 정보를 학습하지 않았기 때문에, 엉뚱한 얘기만 하고 있다. 내가 작성한 글에서 원하는 키매핑을 위해 Scancode Map 값에 hex코드를 입력하면 된다고 설명하고 있다.

 

 따라서, 우리가 앞서 저장한 벡터 스토어로부터 관련 있는 문서를 가져와 프롬프트에 넣어주고, 이로부터 답변을 생성하게끔 지시해 우리가 원하는 답변을 얻을 수 있도록 해야 한다.

 

---

Chain 구성하기

1. 초기화

embeddings = OpenAIEmbeddings()  
llm = ChatOpenAI()

# Pinecone vector store 초기화  
vectorstore = PineconeVectorStore(  
index_name=os.getenv("INDEX_NAME"), embedding=embeddings  
)

query = "What is Pincone in machine learning?"

2. 프롬프트 템플릿

 RAG에 맞는 템플릿을 langchain hub로부터 불러오자. 이 예제에서는 langchain-ai/retrieval-qa-chat을 사용해보자.

from langchain import hub

retrieval_qa_prompt = hub.pull("langchain-ai/retrieval-qa-chat")

 

 해당 프롬프트는 다음과 같은 모습을 하고 있다.

[
SystemMessagePromptTemplate(
    prompt=PromptTemplate(
        input_variables=['context'], 
        template='Answer any use questions based solely on the context below:\n\n<context>\n{context}\n</context>')), 

MessagesPlaceholder(variable_name='chat_history', optional=True), 

HumanMessagePromptTemplate(
    prompt=PromptTemplate(
        input_variables=['input'], 
        template='{input}'))
]

1. SyetemMessage:
   • Answer any use questions based solely on the context below:
   • <context>
     {context}
     </context>
2. MasseagesPlaceholder:
   • chat_history
3. HumanMessage:
   • {input}

작동 순서를 고려하면 다음과 같이 생각해볼 수 있다.

1. 사용자 {input}을 받아 이와 유사한 top-k의 문서를 벡터 스토어로부터 수집
2. {context} 플레이스홀더 부분에 벡터 스토어로부터 가져온 chunk 삽입
3. 만약 chat_history 값이 있다면 이를 프롬프트에 삽입

이후에는 위처럼 만들어진 최종 프롬프트를 LLM에 넣어주고 답변을 받는 순서로 작동한다.

3. 체인 구성

 랭체인에서는 Documents의 리스트를 모델에게 전달하는 체인을 손쉽게 만들 수 있도록 create_stuff_documents_chain 함수를 제공한다. 해당 함수는 가져온 page_content 정보를 그대로 모두 집어넣기 때문에 stuff 방식이라고 한다. 물론 직접 구현하거나, 문서(chunk)를 그대로 다 넣는 것이 아니라 요약해서 집어넣거나, 순서를 재정렬하는 등의 방법도 있지만 여기서는 RAG의 이해를 위해 가장 간단한 버전으로 테스트해보자.

 

 create_stuff_documents_chain은 {context} 플레이스홀더를 갖는 프롬프트를 인자로 받아 해당 위치에 List[Documents]를 문자열로 전달해준다. 문자열로 전달할 때는 page_content만을 전달하는 것이 기본값이나, document_prompt 인자를 통해 meta_data를 포함하는 등 조절할 수 있다.

 

 이렇게 프롬프트에 context를 채운 뒤 llm과 output_parser(기본값: StrOutputParser)를 파이프(|)로 연결한 LCEL Runnable 객체를 반환한다.

combine_docs_chain = create_stuff_documents_chain(llm=llm, prompt=retrieval_qa_prompt)

 

 사용자 input과 유사한 문서를 Pinecone에서 가져올 수 있도록 연결해줘야 한다. 이때 사용할 수 있는 사전 정의 chain으로 create_retrieval_chain 이 있다. create_retrieval_chain 은 List[Document]를 반환하는 retriever와 input을 입력받는 Runnable 객체를 인자로 받는다.

 

 해당 체인은 retrieval를 사용하여 문서를 검색하고, 그 결과를 context 키에 할당한다. 그런 다음 이들을 combine_docs_chain에 넘겨 답변을 생성하고, 이를 answer 키에 할당하는 Runnable 객체를 반환한다.

retrieval_chain = create_retrieval_chain(  
    retriever=vectorstore.as_retriever(),  # 결과를 context로 넘겨줌
    combine_docs_chain=combine_docs_chain  
)

4. 결과 비교

 앞서 query = "내가 원하는 대로 키매핑하려면 무슨 코드를 사용해야 돼?"에 대해 RAG를 사용하지 않은 LLM은 이상한 답변을 했었다. 그럼 내가 작성한 블로그 글을 참고할 수 있도록 구성한 RAG chain에서는 어떤 답변이 나오는 지 확인해보자.

retrieval_chain.invoke({"input": query})

원하는 대로 키매핑을 하려면 레지스트리 경로 [HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\Keyboard Layout] 에서 Scancode Map이라는 값에 hex코드로 원하는 변경사항을 입력해야 합니다. 아래와 같이 메모장에 입력 후 .reg 확장자로 변경하여 실행하면 됩니다.

\```
Windows Registry Editor Version 5.00

[HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\Keyboard Layout]
 "Scancode Map"= hex:00,00,00,00,\
                                     00,00,00,00,\
                                     03,00,00,00,\    ← 아래 2개의 줄을 입력 받아 변경한다. (2개 변경이라면 3, n개 변경 시 n+1)
                                     38,00,5b,e0,\    ← 좌측 ALT키(00 38)를 좌측 Window 키(e0 5b)로 할당
                                     5b,e0,38,00,\    ← Window 키(e0 5b)를 좌측 좌측 ALT키(00 38)로 할당

                                     00,00,00,00
\```

 

LLM이 원하는 답변을 전달한 것을 확인할 수 있다.

---

코드

# ingestion.py

"""  
1. text 파일을 load  
2. chunking  
3. embedding  
4. store in vector DB  
"""  

import os  
from dotenv import load_dotenv  
from langchain_community.document_loaders import TextLoader  
from langchain_text_splitters import CharacterTextSplitter  
from langchain_openai import OpenAIEmbeddings  
from langchain_pinecone import PineconeVectorStore  

load_dotenv()  

if __name__ == "__main__":  
print("Ingesting...")  

loader = TextLoader("test.txt", encoding="UTF8")  
document = loader.load()  

print("splitting...")  
text_splitter = CharacterTextSplitter(chunk_size=500, chunk_overlap=0)  
texts = text_splitter.split_documents(document) # List[Document]  
print(f"created {len(texts)} chunks")  

embeddings = OpenAIEmbeddings()  

print("ingesting...")  
PineconeVectorStore.from_documents(  
documents=texts, embedding=embeddings, index_name=os.getenv("INDEX_NAME")  
)  
print("finish")

# main.py
import os  
from dotenv import load_dotenv  
from langchain_openai import ChatOpenAI, OpenAIEmbeddings  
from langchain_pinecone import PineconeVectorStore  
from langchain import hub  

from langchain.chains.combine_documents import create_stuff_documents_chain  
from langchain.chains.retrieval import create_retrieval_chain  


load_dotenv()  


if __name__ == "__main__":  
print("Retrieving..")  

embeddings = OpenAIEmbeddings()  
llm = ChatOpenAI()  

# Pinecone vector store 초기화  
vectorstore = PineconeVectorStore(  
index_name=os.getenv("INDEX_NAME"), embedding=embeddings  
)  

query = "내가 원하는 대로 키매핑하려면 무슨 코드를 사용해야 돼?"  

# Not using RAG
# chain = PromptTemplate.from_template(query) | llm  
# print(chain.invoke(input={}).content)  

retrieval_qa_prompt = hub.pull("langchain-ai/retrieval-qa-chat")  
combine_docs_chain = create_stuff_documents_chain(  
    llm=llm, prompt=retrieval_qa_prompt  
)  

retrieval_chain = create_retrieval_chain(  
    retriever=vectorstore.as_retriever(), combine_docs_chain=combine_docs_chain  
)  

print(retrieval_chain.invoke({"input": query})['answer'])

 

 

다음 글에서는

1. RAG를 위한 Chain의 내부 동작 과정
2. 인터넷 검색을 통한 RAG 서비스 만들기

순서로 진행하려고 한다.

 

 

Reference

---

[Udemy] LangChain- Develop LLM powered applications with LangChain

[LangChain Q&A with RAG] https://python.langchain.com/v0.1/docs/use_cases/question_answering/