From RAG to AI Agent, 이론 적립(1)

From RAG to AI Agent, 이론 적립(1) 

---

들어가기 전

이런 뉴(?) 기술들을 공부할 땐, 역시 구글 검색이 더 좋은 것 같다.

Perplexity 같은 엔진을 통해 공부하는 것도 좋지만, 

글마다 다른 표현들, 조금씩 다른 설명 방식과 예시들도 공부가 되기 때문이다 

 

---

RAG(Retrieval Augmented Generation, 검색 증강 생성) 란?

 

한 줄 요약

RAG(검색 증강 생성)은 외부 소스(데이터)에서 가져온 사실을 이용해, 생성 AI 모델의 정확도와 신뢰성을 향상시키는 기술

 

예시를 보자

1. 아무리 똑똑한 ChatGPT라 해도, 내가 오늘 저녁에 먹은 메뉴까지는 알 수 없음.
2. 그래서 정확한 답변을 얻으려면, 최신 데이터가 담긴 식단표 DB를 보여줘함

마치 시험 공부를 전혀 하지 않은 채, 오픈북 시험에서 책을 뒤적이며 답을 찾는 학생과 같다

 

---

RAG 필요성, LLM 한계 

 

출처 : IBM Research

 

 

 

RAG는 대규모 언어 모델(LLM)의 문제점과 한계를 극복하는 기술이다

 

킹갓 LLM한테 문제점이 있다고?

1. No Source 

2. Out Of Date 

 

즉, 데이터가 오래 되었다와 대답에 근거가 없다는 문제점, 2가지가 있다

마치 깃허브에 코드를 올리듯, LLM 모델들도 매시간 데이터를 학습하고 업데이트해 배포하지 않는 이상,

그 안의 데이터는 실시간 정보를 반영하지 못하고 이론적으로 오래된 정보가 된다

---

그래서 RAG라는 개념이 필요한 것이다.

 

아래 예시를 다시 보자

ChatGPT는 내가 제공한 최신 데이터(그어오식DB)를 참고해 대답할 수 있고, 그 문서는 대답의 근거가 된다

 

---

 

RAG 프로세스

출처 : NVIDIA docs

 

RAG 파이프라인은 크게 2가지로 나뉜다

1. 데이터(문서) 수집(Document Ingestion) 단계

- 종종 여기서 인덱싱 한다고 표현을 하기도 함, 여기서 인덱싱이란 표현은 데이터를 효율적으로 검색할 수 있도록 준비하고 구조화하는 과정 

 

2. 검색 및 생성 단계  

- 사용자 쿼리(질문)를 받아 인덱스에서 관련 데이터를 검색한 다음 이를 모델에 전달하는 실제 RAG 체인

---

1. 데이터 수집 단계 뜯어보기 

 

뜯어보기 전에 많이 쓰는 오픈소스, LangChain에 대해 알고 가자

LangChain은
LLM 기반으로 애플리케이션을 구축을 위한 오픈 소스 프레임워크다.

기능도 많고 사용하기도 쉬워 엄청 많이 쓰는 라이브러리
예를 들어 데이터 수집의 세부적인 단계에서 필요한 모듈을 모두 지원한다

 

 

아래는 LangChain 커뮤니티에서 제시한, RAG 프로세스 중 '데이터 수집' 단계의 4가지 주요 과정이다

---

 

1.1. Load - 데이터(문서, 외부 소스)를 로드하는 단계

- 정확히는 " Load data sources to text" 단계 

 

Langchain에서는 다양한 문서들을 위해 많은 로더들을 제공하고 있다.

- 자주 사용되는 PDF는 물론이거니와, CSV, JSON까지

 

 

AWS S3 같은 클라우드 서비스에서도 가져올 수 있는 로더들도 있음

 

결론적으로 엄청 많다.  엄청 뒷북이긴 하지만 좀 놀랍다

Document loaders | 🦜️🔗 LangChain

 

---

다양한 종류의 로더 클래스를 제공하고, 사용법도 매우 간단함

어떤 로더 클래스든 공통적으로 load 메서드를 사용할 수 있어 일관된 인터페이스를 제공

결국, 문서 타입에 맞는 로더를 선택하기만 하면 된다는 점이 매우 훌륭하다.

loader = CSVLoader(file_path="./example_data/mlb_teams_2012.csv", source_column="Team")

data = loader.load()

print(data)

 

from langchain_community.document_loaders.csv_loader import UnstructuredCSVLoader

loader = UnstructuredCSVLoader(
    file_path="example_data/mlb_teams_2012.csv", mode="elements"
)
docs = loader.load()

print(docs)

---

로더가 다양한 만큼, 선택도 잘해야 한다.

다음은 Reddit의 게시글 **"What is the best document loader for PDFs? And other docs in general?"**에 달린 댓글이다.

 

 

하고 싶은 말은,

최고의 로더를 찾는 것보다는 로드하려는 문서의 특성이나 작업 환경 등 다양한 요소를 고려해 가장 최선의 로더를 선택하는 것이 더 중요하다고 생각함

---

 

테디노트님의 영상에서는 3가지 고려사항을 언급

여기에 추가로 개인적으로 생각하는 고려사항이 추가적으로 있지 않을까 싶다.

출처 : 테디노트 유튜브

---

로더마다 리턴해주는 메타데이터가 다르기 때문에 이 점도 주의하자

 

코드를 보면  "BasePDFLoader 또는 BaseLoader "를 상속받아서 사용하는 PDF Loader들이 있다. 

 

BasePDFLoader는 BaseLoader를 상속받아 사용하고

BasePDFLoader에는 load() 메서드가 재정의 되어있지 않기 때문에 결국 BaseLoader에 있는 load를 호출하게 된다

 

 

그렇다면 BaseLoader 클래스를 보면,

결론적으로 최하위 클래스에서 load() 메서드를 오버라이딩해서 사용하지 않는다면, lazy_load() 함수를 기본적으로 사용하게 된다.

class BaseLoader(ABC):  # noqa: B024
  ...
  ...
  ...
[docs]
    def load(self) -> list[Document]:
        """Load data into Document objects."""
        return list(self.lazy_load())


[docs]
    def lazy_load(self) -> Iterator[Document]:
        """A lazy loader for Documents."""
        if type(self).load != BaseLoader.load:
            return iter(self.load())
        msg = f"{self.__class__.__name__} does not implement lazy_load()"
        raise NotImplementedError(msg)

---

따라서 load 메서드를 재정의하지 않고 사용하는 로더들의 경우, Blob에서 생성된 메타데이터 값들이 서로 유사함

 

    def lazy_load(
        self,
    ) -> Iterator[Document]:
        """Lazy load given path as pages."""
        if self.web_path:
            blob = Blob.from_data(open(self.file_path, "rb").read(), path=self.web_path)  # type: ignore[attr-defined]
        else:
            blob = Blob.from_path(self.file_path)  # type: ignore[attr-defined]
        yield from self.parser.parse(blob)

 

---

하지만 아래 PyPDFDirectoryLoader처럼, load를 오버라이드할 때는 메타데이터 값이 다를 수 있음

class PyPDFDirectoryLoader(BaseLoader):
    """Load a directory with `PDF` files using `pypdf` and chunks at character level.

    Loader also stores page numbers in metadata. """

[docs]
    def load(self) -> list[Document]:
        p = Path(self.path)
        docs = []
        items = p.rglob(self.glob) if self.recursive else p.glob(self.glob)
        for i in items:
            if i.is_file():
                if self._is_visible(i.relative_to(p)) or self.load_hidden:
                    try:
                        loader = PyPDFLoader(str(i), extract_images=self.extract_images)
                        sub_docs = loader.load()
                        for doc in sub_docs:
                            doc.metadata["source"] = str(i)
                        docs.extend(sub_docs)
                    except Exception as e:
                        if self.silent_errors:
                            logger.warning(e)
                        else:
                            raise e
        return docs

 

 

결론은 사용하고자 하는 로더가 필요한 메타데이터를 리턴 값으로 주는지 고려하자

 

아래 영상에서는 테디 님이 추천하신 몇 가지 로더와 특징들을 확인할 수 있다. 

출처 : 테디노트 유튜브

---

1.2. Split : 로드된 텍스트를 더 작은 청크(문단)로 분할하는 단계

 

왜 문서를 분할할까?

출처 : langchain docs

 

위 내용을 결국 요약을 하면

1. LLM 입력 토큰 제한이 있기 때문에
2. 정확도를 향상하기 위해
3. LLM이 참고할 문서를 더욱 명확하게 하기 위해 -> 정확도 향상의 목적(2번)과 동일
4. 비용을 절감하기 위해

---

Text splitters

그렇다면 어떤 텍스트 분할기를 써야 할까? 가 다음 문제이다

Langchain에서는 다양한 문서 로더뿐만 아니라 다양한 스플리터들도 지원하기 때문이다.

출처 : Langchain

---

각 분할 기법은 고유한 특징을 가지므로,

본인의 데이터 특성에 적합한 기법을 선택하는 것이 중요하다. 주요 분할 기법은 아래와 같다

1. 문자 기반 분할 (Character-Based Splitting)
   • 설정한 문자 수 기준으로 단순히 분할.
2. 재귀적 문자 분할 (Recursive Character Splitting)
   • 점진적으로 더 작은 구분자(\n\n, \n, , "")를 사용해 계층적으로 분할.
3. 의미 기반 분할 (Semantic Splitting)
   • 임베딩을 활용하여 의미에 따라 분할, 관련 있는 내용끼리 묶음.
4. 정규식 기반 분할 (Regex-Based Splitting)
   • 날짜나 특정 키워드와 같은 패턴을 기준으로 분할.
5. LLM 기반 분할 (LLM-Based)
   • AI 모델에 프롬프트를 제공해 콘텐츠 흐름에 따라 동적으로 분할 지점 결정.

출처: Rodrigo Nader, Medium

---

 

가장 범용적으로 사용되는 것이 "RecursiveCharacterTextSplitter"라고 함

from langchain_text_splitters import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=0)
texts = text_splitter.split_text(document)

 

 

 

별도의 구분자를 사용하려면:

• 원하는 구분자를 파라미터로 설정, is_separator_regex 값을 True로 변경

별도 구분자를 설정하지 않으면:

• 디폴트 구분자가 사용. 디폴트 값은 (\n\n, \n, " ", "")

 

출처 : langchain docs

 

---

재귀가 일어나는 부분은 빨간색 박스

1. 먼저 " \n\n"으로 첫 텍스트 스플릿 하고
2. 청크 사이즈 보다 길다 싶으면 재귀적으로 " \n" , " ", "" 순으로 스플릿 하고
3. 청크사이즈 및 청크 오버랩확인하면서 텍스트 Merge 

출처 : langchain docs

 

 

아래 글도 읽어보면 좋은 듯

Understanding LangChain's RecursiveCharacterTextSplitter

 

 

---

1.3. Embed : 분할된 청크(문단)를 벡터 표현으로 변경하기

- 임베딩을 하는 핵심은 벡터 공간에서 유사한 텍스트를 찾기 위함

쉽게 말하면,

1. LLM에게
   "사용자의 질문에 대해 A 문서를 참고하는 것이 적절해 보인다"라고 말해야 함
2. 문서의 벡터 표현
   A부터 Z까지의 문서가 각각 벡터로 표현되어 고유한 벡터 공간에 위치함
3. 사용자 질문의 벡터화
   사용자의 질문을 벡터로 변환한 뒤, 벡터 공간에서 해당 질문과 가까운 문서가 무엇인지 확인
4. 유사성 판단
   벡터 공간에서 근처에 위치한 문서를 사용자 질문과 가장 유사한 문서로 판단

---

Langchain에서 지원해 주는 임베딩 모델들 또한 많다, 유료도 있고 무료도 있음 

Embedding models | 🦜️🔗 LangChain

---

가장 많이 사용하고 사용성이 좋다고 하는 건, OpenAIEmbeddings

API 키만 있으면 딸깍 수준이니 매우 좋아 보인다, 하지만 유료라는 점

import getpass
import os

if not os.environ.get("OPENAI_API_KEY"):
  os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
embeddings.embed_query("Hello, world!")

---

임베딩 모델 선택 시 고려사항

첫 번째 당연 돈이요

두 번째는 한국어 대한 임베딩 모델의 성능이니라

출처 : 테디노트 유튜브

---

최고의 임베딩 모델이 뭔가요? 우문에

 

 

현답 하시는 christianweyer 좌 

---

1.4 VectorStore  : 벡터 데이터를 저장하고 유사성 검색

• 1.3 단계에서 데이터를 벡터로 표현했으면, 그 값들을 저장하는 스토어
• 문서 유사성 검색

---

VectorStore 선택하기

• LangChain에서는 역시 다양한 VectorStore를 통합 지원하며, 각 VectorStore는 고유한 강점/장점/단점을 가
  따라서 목적과 요구사항에 맞는 VectorStore를 선택하는 것이 중요 

많이 사용되는 것 중 하나가 FAISS

1. retriever 객체로 사용자의 쿼리(질문)와 관련된 유사한 문서들을 벡터스토어에서 검색하게 된다.
2. 이때, 유사한 문서들을 찾아내는 알고리즘이 각 라이브러리마다 다르기 때문에 결과가 다를 수 있다는 점 고려할 것

import getpass
import os
from langchain_openai import OpenAIEmbeddings
import faiss
from langchain_community.docstore.in_memory import InMemoryDocstore
from langchain_community.vectorstores import FAISS

index = faiss.IndexFlatL2(len(embeddings.embed_query("hello world")))

vector_store = FAISS(
    embedding_function=embeddings,
    index=index,
    docstore=InMemoryDocstore(),
    index_to_docstore_id={},
)
...
...
...
retriever = vector_store.as_retriever()

 

---

부끄럽지만 벡터 DB와 벡터 스토어가 그냥 같은 건 줄 알았지만, 차이가 있었다

한번 읽어보면 좋음

Exploring Vector Store vs. Vector Database: Which is Right for You?

---

2. 검색 및 생성 단계  

데이터 수집을 위한 4단계가 끝났다면, 

이제 사용자의 입력을 받아 답변을 하는 프로세스가 필요함

---

2.1 Runnable interface

본격적으로 시작하기 전에 "Runnable interface"를 이해할 필요가 있을 것 같음

 

나는 RAG를 공부할 때 코드 예제부터 봤는데, 속으로 생각했음

"튜플 자료형에 invoke..? 저게 뭐시여"

 

"LangChain Expression Language (LCEL)"을 알면 의문점이 풀린다

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)
chain.invoke(
    "Who is the main character in `Tokyo Ghoul` and does he transform into a ghoul?"
)

---

LangChain Expression Language (LCEL)

langchain Docs에서는 LCEL을 "Langchain 컴포넌트를 체인(연결)하는 선언적(declarative) 방법 "이라고 말하고 있음

 

내 이해를 정리해 보자면 

- "Runnable"은 LangChain에서 데이터를 입력받고, 처리하고, 결과를 반환하는 단위

- "Chain" 은여러 단계를 연결한 실행가능한 작업 흐름

- 결론은 "Chain"은 단순히 데이터 흐름만 표현하는 게 아니라 결과를 반환할 수 있는 실행 가능한 단일 객체로 취급

 

---

LCEL 장점

특히, LangSmith로 결괏값들을 편리하게 확인할 수 있는 점이 엄청 편리해 보임

---

3가지 가이드라인

• 단일 액션을 할 때는 굳이 사용할 필요 없음
• 간단한 체인에 적합
• 엄청 복잡한 체인 만들 거면 LangGraph 사용하셈
  • AI Agent 같은 서비스를 만들 때 LangGraph, CrewAI, AG를 사용하는 것 같던데 뒤에서 알아보자

---

어떻게 사용할까? 를 알아보기 전에 까먹지 말아야 하는 것은

- Langchain에서 제공하는 모든 컴포넌트들은 "Runnable 한 특성을 가진다"

 

가장 궁금했던 것은 "|" 오퍼레이터임

1. Runnables 한 특성을 가진 객체들을 연결할 때 간단하게 사용할 수 있는 문법
2. 사실은 RunnableSequence 클래스를 사용한 것임 
3. RunnableSequence 특징은 앞에 실행 결과를 뒤에 입력으로 넣음

 

앞에서 봤던 코드가 사실 이거였다는 말이다, 보기 불편하긴 하네..

chain = RunnableSequence([
    # context와 question을 가져오는 단계
    RunnableSequence([
        {
            "context": RunnableSequence([
                retriever,
                format_docs
            ]),
            "question": RunnablePassthrough()
        },
        prompt
    ]),
    llm,
    StrOutputParser()
])

---

타입 강제 변환 

LCEL에서는 타입을 강제로 변환하여 체인을 더 쉽게 만들 수 있게 했다고 함

 

아래 예시를 보자

1. mappiong은 runnable1과 runnable2를 포함하는 딕셔너리
2. 이 딕셔너리는 자동으로 RunnableParallel로 변환
3. runnable1과 runnable2를 병렬로 실행하고, 그 결과를 runnable3에 전달 

mapping = {
    "key1": runnable1,
    "key2": runnable2,
}

chain = mapping | runnable3

 

만약 강제로 변환이 안 됐다면..

chain = RunnableSequence([
    RunnableParallel(mapping),
    runnable3
])

 

예시의 이해를 좀 더 쉽게 하기 위해 만들어봄

# 가상의 러너블 함수들
def runnable1(question):
    return f"Answer from runnable1 for '{question}'"

def runnable2(question):
    return f"Documents related to '{question}'"

def runnable3(answer, documents):
    return f"Final answer: {answer} with supporting documents: {documents}"

# 병렬 실행을 위한 매핑
mapping = {
    "answer": runnable1,
    "documents": runnable2
}

chain = mapping | runnable3

---

 

2.2 체인 만들기

Runnable interface에 대해 알아봤으니 이번에는 Chain을 먼저 만들어보자

아래 검색 및 생성 단계가 하나의 체인이 될 것이다

---

RunnablePassthrough()
- RunnablePassthrough 클래스는 invoke() 메서드를 통해 입력된 값을 그대로 반환함

 

순서

1. 사용자의 입력(question)을 invoke 메서드를 사용해서 체인에 전달
2. Runnable 한 특성을 가진 딕셔너리기 때문에 사실 "RunnableParallel" 
   1. question 값은 <1.4 벡터스토어> 단계에서 만들어 논 "retriever" 객체로 전달되어서 사용자 질문과 유사한 문서를 벡터 스토어에서 찾아서 결과를 반환함 
   2. RunaablePassthrough()가 question을 그대로 반환해 밸류값을 완성시킴
3. 2번에서 나온 결괏값을 우리가 지정한 프롬프트로 전달
4. LLM이 프롬프트를 실행
5. StrOutputParser()를 사용하여 LLM이 실행한 결과를 문자열 형식으로 변환 

question = "돈이 가장 많은 사람은?"

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)


response = chain.invoke(question)

 

 

아직 prompt와 llm 단계를 하지 않았으니 밑에서 마무리하겠음

---

2.3 프롬프트 작성하기 

프롬프트는 자유롭게 설정하면 되지만, 여기서는 Docs 예시를 따를 예정 

 

Context에는 대답에 참고할 문서들을 넣어줌

Question에는 사용자의 쿼리를 넣어줌

from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_template(
    """
    Answer the question based only on the context provided.
    Context: {context}
    Question: {question}
    """
)

 

 

langchain 프롬프트 허브도 있음

from langchain import hub

prompt = hub.pull("rlm/rag-prompt")

---

이 외에도 좋은 프롬프트가 많아서, 필요시 Pull 해서 사용하면 됨

 

---

2.4 LLM :  언어모델 생성하기 

Langchain을 통해 사용가능한 모델들은 많으며, 역시나 선택의 문제이다 

---

import getpass
import os

if not os.environ.get("OPENAI_API_KEY"):
  os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

---

마무리

간단하게 개념만 정리해 보자라고 시작했는데, 굉장히 진이 빠진다

2편에서는 AI Agent 관련해서 정리를 빨리 해보자

---

참고

https://www.datacamp.com/blog/the-top-5-vector-databases#rdl

Introduction | 🦜️🔗 LangChain

 

Text Splitting: Beyond Basic RAG

 

Understanding LangChain's RecursiveCharacterTextSplitter

 

What Is Retrieval-Augmented Generation aka RAG?

 

What is Retrieval Augmented Generation (RAG)? | A Comprehensive RAG Guide

 

Retrieval Augmented Generation: Streamlining the creation of intelligent natural language processing models

 

Langchain Embedding Model Overview

 

Which Vector Database Should You Use? Choosing the Best One for Your Needs

 

Best 17 Vector Databases for 2025 [Top Picks]

 

RAG 인덱싱 완벽 가이드: 구현부터 최적화까지 5단계 실전 노하우 | 프롬프트해커 대니

 

[LCC-9] LangChain Runnable 이해