[RAG 실험 #1] 개념은 나중에 — 내 Mac에서 RAG를 일단 돌려보기

개요

요즘 LangChain, LangGraph, RAG, AI Agent 등 AI 관련 기술들이 너무 많다. 당장 시작하지 않으면 뒤처질 것 같다는 걱정이 될 때가 자주 있다.

무엇을 해야 할지 몰라서 그냥 RAG로 시작해보려고 한다. 이유는 단순하다. '내가 가진 데이터를 바탕으로 대답하는 AI' 그게 제일 실무에 적용해 보기에 실용적으로 보였기 때문이다.

실험 #1 - 일단 실행시켜 보기

예전에 개념부터 파고드는 습관이 있었는데, 이 방식이 개발에서는 속도를 많이 늦춘다는 것을 깨달았다. 그래서 이번에는 반대로 해보려고 한다. 1) 일단 돌리고 2) 실행 로그를 남기고 3) 나중에 개념을 채우는 방식

이 글의 목표는 이론 이해가 아니며, 그저 'RAG'가 돌아가는 걸 내 Mac에서 확인하는 것이다. LangChain도 LangGraph도 모르고 심지어 RAG가 정확히 뭔지도 모르지만 돌아가는 걸 먼저 보고, 문제가 있으면 수정하고 실행한다. 그 결과를 바탕으로 패턴화 및 원리화를 탐험하고 시행착오하는 것이 개발 학습의 본질이라고 생각한다.

0. Overview

01-rag-minimal/
├── app.py            ← 진입점 (질문-응답 루프) 
├── rag_pipeline.py   ← RAG 핵심 로직
├── requirements.txt  ← 프로젝트가 돌아가기 위해 필요한 외부 라이브러리 목록
└── docs/
    └── sample.txt    ← 여기에 네가 참고시킬 문서들(.txt/.md) 넣으면 됨
└── .env              ← OPENAI_API_KEY=... 넣는 파일

코드 편집기를 열어서 위와 같은 디렉터리를 구성한다. 바로 실행해보고자 하면 해당 레포지토리를 로컬에 클론 하여 01-rag-minimal 폴더를 확인하면 된다. 참고로 레포지토리에는 .env 파일이 없으니 직접 생성하면 된다.

RAG의 핵심로직이 담긴 rag_pipeline.py가 있다. 이를 실행하는 app.py를 추가로 구성해서 진입점으로 사용한다. 여기서는 우리가 질문한 내용을 OpenAI API로 요청하여 응답을 받아 출력하는 전반적인 플로우를 담당한다. 이 흐름은 반복된다.

docs 에는 LLM에게 질문할 때의 문맥 정보를 담는 폴더로 생각했다.

1. 프로젝트 설정

데모 프로젝트 폴더를 생성하고자 하는 디렉터리도 이동해서 폴더를 생성한다. 여기서는 01-rag-minimal 폴더를 생성했다. 내부에 requirements.txt 을 생성해서 아래와 같이 입력한다.

openai>=1.0.0,<2.0.0
numpy>=1.26.0
python-dotenv>=1.0.0

requirements.txt는 pip install 명령어로 설치해야 하는 라이브러리(패키지)의 목록을 기록한 파일이다. 쉽게 말해 이 프로젝트를 다른 환경에서도 똑같이 실행하기 위해 필요한 재료 리스트라고 생각할 수 있다.

패키지 이름과 버전을 지정해, 다른 사람이 다음과 같은 명령어로 같은 환경을 재현할 수 있게 해 준다. 이후 코드실행하기 전에 이 명령어를 입력할 예정이다.

pip install -r requirements.txt

pip install 다음 r옵션은 pip install --help명령어를 터미널에 입력하면 확인해 볼 수 있다.

주어진 파일로부터 인스톨 명령어를 사용할 수 있고, 반복적으로 사용할 수 있다고 명시되어 있다. 이 명령어는 이미 설치된 패키지는 건너뛰기 때문에 반복 실행해도 문제가 없다.

이 데모에서는 openai, numpy, python-dotenv 세 가지의 패키지를 필요로 한다. openai는 openai의 api를 이용할 예정이기 때문에 해당 패키지를 넣었고, numpy는 RAG 로직의 '문서 임베딩 벡터 간 유사도 계산 등'에 사용된다고 한다. python-doenv는 이다음에 생성할 .env 파일에서 환경변수를 읽어오는 라이브러리이다.

.env 파일을 생성해서 아래와 같이 입력한다. 등호 다음에는 openai에서 생성한 본인의 키를 입력하면 된다. 참고로 public에 공개하면 안 되며 github에도 올라가선 안된다. api 키 생성 방법은 어렵지 않고 이미 많은 레퍼런스가 존재하기 때문에 생략한다.

OPENAI_API_KEY=sk-{YOUR_OPEN_API_KEY}

필요하면 아래 처럼 .gitignore 파일을 생성해 두면 리모트에 업로드 시 자동으로 파일이 제외된다:

.env
.venv/
__pycache__/

2. 핵심 로직

app.py 파일을 생성해서 아래 코드를 입력한다 :

# app.py
from rag_pipeline import RAGPipeline

def main():
    rag = RAGPipeline(docs_path="docs", top_k=3)
    rag.build_index()

    print("🔎 RAG 데모입니다. 'exit' 치면 종료.\n")
    while True:
        q = input("Q> ").strip()
        if q.lower() in ("exit", "quit"):
            break
        answer = rag.query(q)
        print("\nA>\n", answer, "\n")

if __name__ == "__main__":
    main()

여기까지 코드를 훑어보면서 이해한 흐름은 이렇다 :

RAGPipeline 클래스 인스턴스 생성
인덱스를 빌드 (RAG 연산에서 numpy 패키지를 사용하는 이유일 것 같다)
while 문 내부에서 openai api로 쿼리 요청하고, 응답을 출력
exit, quit이 나오기 전까지 3을 반복한다

rag_pipeline.py를 생성해서 다음 코드를 입력한다 :

# rag_pipeline.py
import os
import glob
from typing import List, Dict, Any, Tuple
from dataclasses import dataclass

import numpy as np
from dotenv import load_dotenv
from openai import OpenAI


@dataclass
class DocChunk:
    doc_id: str
    text: str
    embedding: np.ndarray


class RAGPipeline:
    """
    아주 단순한 in-memory RAG 파이프라인
    - docs 폴더의 .txt/.md 읽어서
    - chunk로 자르고
    - OpenAI 임베딩으로 바꿔서
    - 질의 들어오면 코사인 유사도로 top_k 뽑고
    - 그걸 컨텍스트로 LLM에 보내서 답변 생성
    """

    def __init__(
        self,
        docs_path: str = "docs",
        top_k: int = 3,
        embedding_model: str = "text-embedding-3-small",
        chat_model: str = "gpt-4o-mini",
    ):
        load_dotenv()
        api_key = os.getenv("OPENAI_API_KEY")
        if not api_key:
            raise RuntimeError("OPENAI_API_KEY 가 .env 에 없습니다.")
        self.client = OpenAI(api_key=api_key)

        self.docs_path = docs_path
        self.top_k = top_k
        self.embedding_model = embedding_model
        self.chat_model = chat_model

        self.chunks: List[DocChunk] = []

    # ---------- public API ----------
    def build_index(self) -> None:
        """docs/ 폴더의 파일들을 읽어 in-memory index를 만든다."""
        texts = self._load_docs()
        chunks = self._chunk_all(texts)
        embedded = self._embed_chunks(chunks)
        self.chunks = embedded
        print(f"[index] {len(self.chunks)} 개의 chunk 로 인덱스 구성 완료")

    def query(self, question: str) -> str:
        """질문 하나를 받아서 RAG 로 답한다."""
        q_emb = self._embed_query(question)
        top_chunks = self._search(q_emb, top_k=self.top_k)
        context = self._build_context(top_chunks)
        answer = self._call_llm(question, context)
        return answer

    # ---------- internal ----------
    def _load_docs(self) -> Dict[str, str]:
        """docs/ 이하의 .txt, .md 파일을 전부 읽어온다."""
        os.makedirs(self.docs_path, exist_ok=True)
        paths = glob.glob(os.path.join(self.docs_path, "*.txt")) + glob.glob(
            os.path.join(self.docs_path, "*.md")
        )
        docs = {}
        for p in paths:
            with open(p, "r", encoding="utf-8") as f:
                docs[os.path.basename(p)] = f.read()
        return docs

    def _chunk_all(self, docs: Dict[str, str]) -> List[Tuple[str, str]]:
        """아주 단순한 chunking: 글자 수 기준으로 자름."""
        all_chunks: List[Tuple[str, str]] = []
        chunk_size = 500  # 글자 수 기준
        overlap = 100

        for doc_id, text in docs.items():
            start = 0
            while start < len(text):
                end = start + chunk_size
                chunk = text[start:end]
                all_chunks.append((doc_id, chunk))
                start += chunk_size - overlap

        return all_chunks

    def _embed_chunks(self, chunks: List[Tuple[str, str]]) -> List[DocChunk]:
        embedded: List[DocChunk] = []
        for doc_id, text in chunks:
            emb = self._embed_text(text)
            embedded.append(DocChunk(doc_id=doc_id, text=text, embedding=emb))
        return embedded

    def _embed_query(self, query: str) -> np.ndarray:
        return self._embed_text(query)

    def _embed_text(self, text: str) -> np.ndarray:
        # OpenAI 임베딩 호출
        resp = self.client.embeddings.create(
            model=self.embedding_model,
            input=text,
        )
        vec = np.array(resp.data[0].embedding, dtype=np.float32)
        return vec

    def _search(self, query_emb: np.ndarray, top_k: int = 3) -> List[DocChunk]:
        """코사인 유사도 기반 top_k 검색"""
        scored = []
        for chunk in self.chunks:
            score = self._cosine_similarity(query_emb, chunk.embedding)
            scored.append((score, chunk))

        scored.sort(key=lambda x: x[0], reverse=True)
        return [c for _, c in scored[:top_k]]

    @staticmethod
    def _cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-10))

    def _build_context(self, chunks: List[DocChunk]) -> str:
        parts = []
        for i, c in enumerate(chunks, start=1):
            parts.append(f"[출처 {i} - {c.doc_id}]\n{c.text.strip()}\n")
        return "\n".join(parts)

    def _call_llm(self, question: str, context: str) -> str:
        system_prompt = (
            "너는 업로드된 문서를 바탕으로만 대답하는 RAG 어시스턴트다. "
            "모르면 모른다고 말해. 출처에 없는 내용은 추측하지 마."
        )
        user_prompt = (
            f"다음은 너에게 주어진 참고 문서이다:\n\n{context}\n\n"
            f"위 문서를 근거로 다음 질문에 한국어로 답해라:\n{question}"
        )

        resp = self.client.chat.completions.create(
            model=self.chat_model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt},
            ],
            temperature=0.2,
        )
        return resp.choices[0].message.content.strip()

내가 현재 이해할 수 있는 부분은 다음과 같다.

클래스를 초기화할 수 있다
인덱스를 빌드한다 (무엇의?)
쿼리를 요청한다
문서를 로드한다 (우리가 생성하는 문서를 나타내는 것 같다)

여기서 말하는 인덱스는 RDB의 B-Tree 인덱스랑 완전히 같지는 않아 보인다. 다만 '나중에 빠르게 찾기 위해 미리 구조를 만들어둔다'는 점에서 비슷하다. build_index()는 문서를 읽고, 청킹 하고, 임베딩하는 모든 과정을 포함한다. 이 글에서는 거기까지만 이해하고 넘어가려고 한다. 청킹 임베딩은 다음에 다룬다.

3. 문서 생성

docs폴더를 생성하여 내부에 txt 혹은 md파일을 생성한다. 위 rag_pipeline.py 파일 내부의 _load_docs 함수를 보면 아래와 같은데 md 밑 txt 파일만 읽도록 되어있다. 하위 폴더는 탐색하지 않는다. PDF까지 넣고 싶다면 해당 내용을 수정하면 된다.

여기서는 간단하게 sample.txt 파일을 생성해서 아래와 같이 단순한 데이터를 넣었다. 인사 시스템에서 가장 활용도가 좋은 예는 LLM 인사 봇을 만들어서 고객사의 인사팀의 로드를 줄여주는 게 가장 비즈니스 관점에서 중요해 보였기 때문이다.

우리 회사의 휴가 규정은 연 15일이고, 미사용분은 다음해 5월까지 이월 가능하다.

4. 실행

다음 명령어들을 차례로 입력하면 실행 결과를 얻을 수 있다. 아래 명령어들은 반드시 프로젝트 루트(여기서는 01-rag-minimal)에서 실행해야 한다.

python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
python app.py

각 명령어의 의미는 다음과 같이 정리할 수 있다:

python3 -m venv .venv
- 가상환경(Virtual Environment)을 생성하는 명령어
  - python3는 macOS에서 Python3 버전 실행기(인터프리터)를 명시적으로 호출
  - -m venv는 파이썬 내장 모듈인 venv를 실행
  - .venv는 현재 디렉토리에 .venv라는 이름의 폴더를 만들어 가상환경 생성
source .venv/bin/activate
- 가상환경을 활성화하는 명령어
  - source는 현재 쉘 세션에서 지정된 스크립트를 실행
  - .venv/bin/activate는 가상환경의 활성화 스크립트 위치
- 터미널에서 (.venv) 이렇게 표시되면 활성화 성공
pip install -r requirements.txt
- 필요한 패키지(의존성) 일괄 설치
  - pip install은 Python 패키지 매니저 명령
  - -r requirements.txt는 'requirements.txt 파일 안의 패키지를 모두 설치
- 결과적으로 .venv/lib/python3.x/site-packages/안에 위에서 언급한 라이브러리들이 설치됨
- 명령어 입력 시 해당 파일이 위치한 디렉터리에서 진행하기
python app.py
- 현재 활성화된 Python 인터프리터로 메인 스크립트(app.py) 실행

5. 결과

결과를 확인하면 확실히 여기서 입력한 docs 내의 txt 파일 기반으로 대답하는 것을 확인할 수 있었다.

비용 관점에서는 대시보드를 확인해 보면 0.01 달러보다 적게 썼음을 확인할 수 있다. 5달러 정도 결제해 둔 상태인데 걱정 없이 여러 번 테스트해 봐도 큰 문제가 없을 것으로 보인다.

다음 스텝

왜 이렇게 귀찮게 가상환경을 써야 할까?
왜 문서를 잘게 쪼개야 하는지? (chunking)
백터 DB를 쓰면 이 코드가 어떻게 바뀌는지? (지금 코드는 in-memory)

위 순서대로 하나씩 정리할 예정이 이며, 이 시리즈는 ‘실행 → 기록 → 개념화’ 순서로 진행된다.

Gukin Han

[RAG 실험 #1] 개념은 나중에 — 내 Mac에서 RAG를 일단 돌려보기

개요

실험 #1 - 일단 실행시켜 보기

0. Overview

1. 프로젝트 설정

2. 핵심 로직

3. 문서 생성

4. 실행

5. 결과

다음 스텝

티스토리툴바

[RAG 실험 #1] 개념은 나중에 — 내 Mac에서 RAG를 일단 돌려보기

개요

실험 #1 - 일단 실행시켜 보기

0. Overview

1. 프로젝트 설정

2. 핵심 로직

3. 문서 생성

4. 실행

5. 결과

다음 스텝

관련글

티스토리툴바