LLM & RAG 용어정리 (LangChain과 FAISS 실습)

---

실습을 통한 LangChain과 FAISS에 대해서 알아보자.

---

기본 설정

패키지 설치

기본적으로 FAISS와 LangChain을 사용하기 위해서 패키지의 설치를 진행한다.

faiss 같은 경우는 cpu와 gpu둘중 선택한다.

 

기본적으로는

pip install langchain langchain-openai faiss-cpu/gpu

 

 

현재 사용하고 있는 anaconda 가상환경에는 최대한 conda 패키지로 설치를 해주는 것이 좋다.

 

이때, conda패키지에 없을 수 있는데, 이는 conda-forge로 설치할 수 있다.

 

 

https://anaconda.org/

설치를 원하는 패키지의 이름을 검색하면 된다.

 

 conda install conda-forge::langchain
 conda install conda-forge::langchain-openai
 conda install conda-forge::faiss-gpu

 

 

---

API 호출

openai를 사용을 해야하기 때문에 api키를 가져와 준다.

 

나는 api 키를 윈도우에 설정을 해 뒀기 때문에 불러올 수 있지다.

setx GPT-KEY "~~"

이런 식으로 cmd 에 키의 값을 넣으면 GPT-KEY를  getenv로 부르게 되면 api 키를 자동으로 들고 올 수 있다.

보안상으로 이렇게 해 두는것이 좋다.

 

 

import os

os.environ["OPENAI_API_KEY"] = os.getenv("GPT-KEY")

 

위와같이 api키를 할당 할 수 있다.

이때의 os.enciron은 환경변수를 뜻한다.

 

즉 OPENAI_API_KEY의 환경변수의 값을 할당하여 후에 사용할 openai의 access를 받아오는 것이다.

 

 

 

---

기본 개념

언어 모델의 초기화

우선은 langchain_openai 에서 ChatOpenAI 를 호출한다.

 

그리고 langchain_core.messages 에서 HumanMessage를 호출하게 되는데

이는 메시지 관련하여 내부적으로 처리를 하는 메시지를 뜻한다.

 

메시지 관련한 패키지는 langchain.schema도 있는데, 이는 message보다 상위의 개념이며

파인튜닝과 같은 설정에서는 message를 사용한다.

 

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

 

 

 

패키지를 불러왔으면, 모델을 설정 한다.

model이라는 객체에 ChatOpenAI를 할당한다.

이를 우리는 초기화 라고 한다. (reset이 아닌 초기의 설정을 의미)

# 모델의 초기화
model = ChatOpenAI(model="gpt-4")

response = model.invoke([HumanMessage(content="안녕하세요, 무엇을 도와드릴까요?")])
print(response.content)

 

 

invoke는 어떠한 동작을 실행하거나 호출한다는 뜻으로, 뒤의 HumanMessage를 실행하는것이다.

 

 

이때 Message의 종류는 3가지 HumanMessage, AIMessage, SystemMessage가 있다.

 

HumanMessage는 사용자가 입력한 메시지를 뜻한다.

AIMessage는 AI모델이 생성한 응답의 메시지를 뜻한다.

SystemMessage는 시스템 설정이나 규칙을 설명하는 메시지이다.

 

이들을 이용해서 모델의 초기 설정 즉 프롬프팅을 할 수 있다.

 

 

print를 사용하여 content에 대한 반응을 확인 할 수 있다.

 

 

---

프롬프트 템플릿 설정하기

프롬프트 템플릿이란, 모델의 초기 세팅에서 프롬프트 엔지니어링을 하기위한 템플릿이다.

 

프롬프트 템플릿을 설정하기 위해서 패키지 부터 불러온다.

from langchain_core.prompts import ChatPromptTemplate

 

 

 

시스템 메시지를 설정하는데, 이는 유저의 요청이라기 보다 모델 시스템에 명령을 내리는 느낌이라고 보면된다.

 

template를 생성해서 사용자의 텍스트를 입력 받는다.

from_messages를 사용하여 메시지 객체를 조합하여 템플릿을 생성한다.

# 시스템 메시지 설정
system_template = "Translate the following sentence from English to {language}:"

# 사용자 텍스트 입력
prompt_template = ChatPromptTemplate.from_messages([
    ("system", system_template),
    ("user", "{text}")
])

# 프롬프트 생성
result = prompt_template.invoke({"language": "French", "text": "How are you?"})
print(result.to_messages())

 

프롬프트를 생성하면 이렇게 들어간다 라는걸 볼 수 있다.

 

위는 프롬프트의 템플릿 생성이지 여기까지만을 사용하여

translate를 할 수는 없다.

 

 

---

체인 연결

 

체인 연결을 위한 파서의 패키지를 불러옵니다.

from langchain_core.output_parsers import StrOutputParser

 

 

이때 파서 즉 parser란 데이터를 분석하고 구조화되는 형식으로 변환을 시켜주는 라이브러리이다.

 

텍스트 파일과 같은 입력데이터를 구문 분석을 하며, 모델이 잘 받아 들일 수 있는 형식으로 변환을 한다.

 

 

아래의 코드의 parser는 입력데이터가 모델의 출력으로 받아서 str즉 문자열로 변환을 시키는 것이다.

 

즉 위에 설명한 프롬프트 템플릿, 모델, 파서를 chain으로 연결시켜서

번역을 실제로 하게 만드는 것이며, 값을 유저가 읽을수 있도록 변환시키는 것이다.

 

# 응답을 파싱하는 파서 초기화
parser = StrOutputParser()

# 템플릿, 모델, 파서를 체인으로 연결
chain = prompt_template | model | parser

# 체인 실행
response = chain.invoke({"language": "Spanish", "text": "Where is the library?"})
print(response)

 

 

---

FAISS 활용 벡터 데이터베이스 구성

FAISS는 벡터의 유사성 검색을 위한 라이브러리이다.

이는 Vector DB와 같은 데이터베이스이며, 유사성검색이 가능한 데이터베이스라고 보면된다.

 

벡터 데이터베이스를 생성해보는 과정에 대해서 알아보자.

 

---

벡터 임베딩 생성

 

우선 OpenAIEmbeddings 를 불러온다.

이는 임베딩 모델이다.

from langchain_openai import OpenAIEmbeddings

 

 

embeddings 라는 객체에 임베딩 모델을 초기화한다.

이때 text-embedding-ada-002 라는 모델을 호출한다.

# OpenAI 임베딩 모델 초기화
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")

 

---

인덱스 초기화

우선 FAISS 벡터 데이터베이스 생성을 위한 패키지들을 불러온다.

import faiss
from langchain_community.vectorstores import FAISS
from langchain_community.docstore.in_memory import InMemoryDocstore

 

 

 

FAISS의 인덱스를 설정하게 된다.

이때 IndexFlatL2는 hello world라는 데이터를 임베딩하여 나온 벡터의 관계를

유클리드 거리를 이용하여 유사성을 계산하는 것이다.

 

이때 vector_store라는 빈 데이터베이스에 FAISS의 데이터를 할당 하는 것이다.

 

이때의 embedding_function은 인베딩의 형식이 어떻게 되어있는지 설정하는 것이며

이는 서치를 하게 됬을때, 같은 임베딩을 통하여 벡터화를 하여야 하기 때문이다.

 

 

여기서 docstore_id에 index를 할당 하는 이유는,

index.에는 데이터가 들어가 있는것이 아닌, 유사도를 나타내는 숫자로

만약 FAISS를 이용해서 유사도가 높은 데이터를 찾게 된다면, 이 index와 연결된 docstore_id즉

벡터 db에 있는 실제 데이터와 빠르게 연결을 하기 위함이다.

 

# FAISS 인덱스 생성
index = faiss.IndexFlatL2(len(embeddings.embed_query("hello world")))
vector_store = FAISS(
    embedding_function=embeddings,
    index=index,
    docstore=InMemoryDocstore(),
    index_to_docstore_id={}
)

 

 

---

데이터 추가

벡터 데이터베이스에 문서데이터를 추가하기 위해 필요한 패키지를 불러온다.

이때 uuid는 문서 데이터에 고유한 식별 id 를 생성하기 위해 필요한 패키지이다.

from langchain_core.documents import Document
from uuid import uuid4

 

 

document객체에 문서데이터를 넣는다.

 

document의 숫자만큼 uuid를 생성하고 vector_store에 문서와 각각의 uuid를 할당하여 저장한다.

# 문서 생성
documents = [
    Document(page_content="LangChain을 사용해 프로젝트를 구축하고 있습니다!", metadata={"source": "tweet"}),
    Document(page_content="내일 날씨는 맑고 따뜻할 예정입니다.", metadata={"source": "news"}),
    Document(page_content="오늘 아침에는 팬케이크와 계란을 먹었어요.", metadata={"source": "personal"}),
    Document(page_content="주식 시장이 경기 침체 우려로 하락 중입니다.", metadata={"source": "news"}),
]

# 고유 ID 생성 및 문서 추가
uuids = [str(uuid4()) for _ in range(len(documents))]
vector_store.add_documents(documents=documents, ids=uuids)

 

---

벡터 데이터베이스 의 서치

 

유사성 검색을 사용하는 코드이다.

 

filter로 조건을 할당하고, 입력값을 넣는다.

vector_store로 인하여 임베딩 되고, 그의 유사성을 찾아 표시하게 한다.

 

similarity_search_with_score로 유사도의 점수까지 포함하게 한다.

 

# 기본 유사성 검색
results = vector_store.similarity_search("내일 날씨는 어떨까요?", k=2, filter={"source": "news"})
for res in results:
    print(f"* {res.page_content} [{res.metadata}]")

# 점수와 함께 유사성 검색
results_with_scores = vector_store.similarity_search_with_score("LangChain에 대해 이야기해주세요.", k=2, filter={"source": "tweet"})
for res, score in results_with_scores:
    print(f"* [SIM={score:.3f}] {res.page_content} [{res.metadata}]")

 

 

---

RAG체인에 FAISS 통합

FAISS를 retriever로 변환

FAISS의 벡터 데이터를 연결한다.

이때 search_type은 유사성을 기준으로 서치를 한다는 것을 의미한다.

search_kwargs는 검색의 아웃풋으로 최대 결과값의 숫자를 뜻한다.

 

즉 이 코드에서는, 검색시 가장 유사한 1개의 문서데이터만을 가져오도록 설정하는 것이다.

retriever = vector_store.as_retriever(search_type="similarity", search_kwargs={"k": 1})

 

 

---

RAG 체인 생성

프롬프트 템플릿과 FAISS와 연결된 Retriever 을 연결하기 위한 패키지들을 불러온다.

 

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

 

 

 

프롬프트 템플릿을 정의함으로써, 이 모델에 지시사항을 내리는 것과 같다.

즉 system 메시지를 기준으로 행동을 해야한다.

 

 

 

디버그 클래스는 입력값을 그대로 출력하지만,

그 값을 한번 프린트 함으로써, 문제가 있는지 한번 유저에게 보여주는것이다.

만약 출력값에 문제가 있다면, 이전의 단계에서 수정을 하게끔 하는 것이다.

 

불러온 문서의 리스트를 하나의 데이터 즉 텍스트로 변환을 하는 단계를 추가한다.

하나의 값으로 변환하여 모델이 문서들을 입력 받을수 있도록 하는 것이다.

이때, 형식은 context : 문서들의 내용, question : 유저의 질문 이 되게 한다.

 

retriever로 유저의 입력을 토대로 찾은 데이터를 context로 저장한다.

이때 가져온 context를 contexttotext를 통하여 텍스트로 변환 시킨다.

유저의 질문이 debugpassthrough를 지나 제대로 전달이 되고 있는지 확인한다.

 

| DebugPassThrough() | ContextToText()|   contextual_prompt | model

이는 모델의 플로우를 정하는 것이다.

즉 유저의 입력값이 디버그를 통하고 입력값에 따른 검색을 하여 찾은 문서를 텍스트로 변환하고

이를 프롬프트 템플릿에 할당하여 모델에 전달해 LLM에서 데이터를 기반으로 답변을 생성하게 한다.

 

# 프롬프트 템플릿 정의
contextual_prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer the question using only the following context."),
    ("user", "Context: {context}\\n\\nQuestion: {question}")
])


class DebugPassThrough(RunnablePassthrough):
    def invoke(self, *args, **kwargs):
        output = super().invoke(*args, **kwargs)
        print("Debug Output:", output)
        return output
        
# 문서 리스트를 텍스트로 변환하는 단계 추가
class ContextToText(RunnablePassthrough):
    def invoke(self, inputs, config=None, **kwargs):  # config 인수 추가
        # context의 각 문서를 문자열로 결합
        context_text = "\n".join([doc.page_content for doc in inputs["context"]])
        return {"context": context_text, "question": inputs["question"]}

# RAG 체인에서 각 단계마다 DebugPassThrough 추가
rag_chain_debug = {
    "context": retriever,                    # 컨텍스트를 가져오는 retriever
    "question": DebugPassThrough()        # 사용자 질문이 그대로 전달되는지 확인하는 passthrough
}  | DebugPassThrough() | ContextToText()|   contextual_prompt | model

# 질문 실행 및 각 단계 출력 확인
response = rag_chain_debug.invoke("강사이름은?")
print("Final Response:")
print(response.content)

 

---

FAISS 저장

save_local을 사용해서 (" ")안에 설정한 이름으로 저장한다.

(" ")안에 path를 지정 할 수 있다.

 

load_local로 index를 부르고 임베딩을 통하여 유사성을 검색한다.

 

# 인덱스 저장
vector_store.save_local("faiss_index")

# 저장된 인덱스 로드
new_vector_store = FAISS.load_local("faiss_index", embeddings)

 

 

---

FAISS 데이터베이스 병합

두개 또는 여러개의 FAISS 데이터베이스를 병합할 수 있다.

db1 = FAISS.from_texts(["문서 1 내용"], embeddings)
db2 = FAISS.from_texts(["문서 2 내용"], embeddings)

# 병합
db1.merge_from(db2)