블로그 Q&A 챗봇(Chatbot) RAG 만들어보기 - LangChain + Ollama + FastAPI + Streamlit + PGVector

포스팅 개요

이번 포스팅은 저의 티스토리 블로그 글을 활용한 AI Q&A 챗봇(Chatbot) RAG를 만들어본 포스팅입니다. PostgreSQL의 PGVector를 사용해서 벡터 데이터베이스(vector database)로 사용했고, Python의 랭체인(langchain)과 ollama, FastAPI, Streamlit을 활용해서 데이터를 저장, LLM 통신, 챗봇 Q&A 화면을 구성했습니다.

 

이번 포스팅은 다음과 같은 순서로 진행됩니다.

 

1. 데이터베이스 테이블 구성

2. 티스토리 블로그 크롤링 및 postgresql 데이터베이스에 저장

3. 데이터 청킹(Chunking) 및 벡터(Vector) 추출 후 저장

4. FastAPI를 이용한 Ollama LLM 통신

5. Streamlit을 활용하여 Q&A Chatbot 구현

 

바로 진행해보겠습니다!

---

 

이 사이드 프로젝트(?)를 진행한 이유

사실, 티스토리에도 검색 기능이 충분히 있습니다. 하지만, 제가 필요한 검색이 잘 안되거나, 다시 읽거나 해야하는 경우가 많더라구요.

그래서 제 블로그 Q&A 자체가 제 스스로 필요하다는 생각이 들어서, 하루 시간을 사용해 만들어보았습니다.

그리고 이왕 만드는 것, 그 과정을 블로그에도 공유합니다!

 

1. 데이터베이스 테이블 구성

가장 먼저, 데이터베이스를 준비해야 합니다. 데이터베이스를 준비하는 이유는, 제 티스토리 블로그 글을 크롤링 한 다음 저장할 때도 필요하며, 허깅페이스(HuggingFace) 모델을 이용해 텍스트 벡터를 추출한 후 벡터 값을 저장할 때도 필요하기 때문입니다.

저는 관계형 데이터베이스(RDB)로도 사용할 수 있으면서도 동시에 벡터 데이터베이스(Vector database)로도 사용할 수 있는 PostgreSQL을 사용했습니다. 혹시 PostgreSQL에 대한 설치 방법과 개념이 익숙하지 않다면 제가 일전에 작성한 글을 참조해주세요. 

 

- PostgreSQL 설치 :  https://lsjsj92.tistory.com/675 

PostgreSQL PGVector 설치 및 사용하기(Feat. 벡터 데이터베이스(Vector Database) 구축)

 

- PostgreSQL PGVector 사용하기: https://lsjsj92.tistory.com/677

PGVector와 Python FastAPI를 연동하여 벡터 데이터 저장 및 유사도 기반 조회하기

 

제가 구성한 전체 테이블 구조도는 다음 사진과 같습니다.

 

 

각 테이블을 설명하자면,

 

- blog_posts 테이블: 제 블로그의 원본 글에 대한 데이터입니다. 즉, 제 티스토리 블로그 글을 크롤링할 때 제목, 발행일자, 콘텐츠 내용, 블로그 url, 태그, 카테고리 정보 등을 가지고 오는데요. 그 중 제목, 발행일, 콘텐츠 내용, 블로그 url 정보를 저장합니다.

- post_tags 테이블: blog_posts에 해당하는 블로그 글의 태그 정보입니다. 저는 블로그 글에 태그를 달아두는 습관이 있어, #으로 시작하는 태그 정보들을 넣어두었습니다.

- tags 테이블: 전체 태그 정보를 담아두고 있습니다. 이미 기존에 사용된 태그 값이 있다면 그 태그를 사용할 수 있도록 관계를 구성했습니다.

- processing_status 테이블: 블로그 포스팅 글이 청킹(Chunking)과정이나 벡터 임베딩(vector embedding) 과정을 수행했는지 체크하는 테이블입니다. 만약, 블로그 포스트 글이 있는데 청킹을 수행하지 않았다면 그 체크 값을 활용해 chunking 및 vector embedding 과정을 수행합니다.

- post_categories 테이블: tags와 비슷하게 카테고리 정보를 담아두는 테이블입니다. 블로그에 해당되는 카테고리를 저장합니다.

- categories 테이블: 전체 카테고리 정보를 담고 있습니다.

- content_chunks 테이블: blog_posts에서 content를 기준으로 chunking을 수행하고 그 결과를 저장한 테이블입니다. 이때, embedding_id와도 연계되어서 해당 임베딩이 어떤 chunking 결과인지 알 수 있도록 합니다.

- embeddings 테이블: chunking을 수행한 텍스트의 vector embedding 값입니다. 저는 huggingface의 embedding 모델을 사용했습니다.

 

저는 기본적으로 파이썬(Python) 코드 안에서 테이블을 생성할 수 있도록 아래 코드와 같이 미리 구성해두었습니다.

 

class ContentChunk(Base):
    __tablename__ = 'content_chunks'
    
    chunk_id = Column(Integer, primary_key=True)
    post_id = Column(Integer, ForeignKey('blog_posts.post_id', ondelete='CASCADE'), nullable=False)
    chunk_index = Column(Integer, nullable=False)
    chunk_text = Column(Text, nullable=False)
    chunk_hash = Column(String(64), unique=True)
    chunk_metadata = Column(JSON, nullable=True)  # metadata -> chunk_metadata로 변경
    embedding_id = Column(String(36), nullable=True)
    
    # 관계 정의
    post = relationship("BlogPost", back_populates="chunks")
    embedding = relationship("Embedding", back_populates="chunk", uselist=False, cascade="all, delete-orphan")
    
    def __repr__(self):
        return f"<ContentChunk(chunk_id={self.chunk_id}, post_id={self.post_id}, index={self.chunk_index})>"

class Embedding(Base):
    __tablename__ = 'embeddings'
    
    embedding_id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    chunk_id = Column(Integer, ForeignKey('content_chunks.chunk_id', ondelete='CASCADE'), unique=True)

    embedding = Column(Vector(1024), nullable=False)

    model_name = Column(String(255), nullable=False)
    created_at = Column(DateTime, nullable=False, server_default='now()')
    
    chunk = relationship("ContentChunk", back_populates="embedding")
    
    def __repr__(self):
        return f"<Embedding(embedding_id='{self.embedding_id}', chunk_id={self.chunk_id})>"
        

async def init_db():
    """데이터베이스 초기화"""
    async with async_engine.begin() as conn:
        # pgvector 확장 활성화
        try:
            await conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
            logger.info("pgvector 확장이 활성화되었습니다.")
        except Exception as e:
            logger.warning(f"pgvector 확장 활성화 중 오류 발생: {e}")
        
        # 테이블 생성
        await conn.run_sync(Base.metadata.create_all)

 

이와 같이 수행하게 되면 데이터베이스를 초기화 시킬 때 필요한 테이블을 만들 수 있습니다.

이제, 이렇게 구성된 테이블을 기준으로 데이터를 수집하였습니다. 

---

2. 티스토리 블로그 크롤링 및 PostgreSQL 데이터베이스 저장

테이블을 구성했으니, 데이터를 넣어야겠죠? 블로그 Q&A 챗봇을 만들기 위해서는 당연히 기본적으로 블로그 글 내용이 필요할겁니다. 저는 제 티스토리 블로그(지금 이 블로그입니다.) 글을 기준으로 크롤링을 진행했습니다.

크롤링 시 저는 포스팅의 제목, 포스팅 내용, 포스팅 생성일, 태그, 카테고리 등의 정보를 수집하도록 했습니다. 

클로링은 Python의 beautifulsoup4을 사용했으며, 아래와 같이 파이썬 코드를 구성하였습니다.

 

url = f"https://lsjsj92.tistory.com/{post_number}"
# 이미 크롤링된 URL인지 확인
if check_exists_func:
    if await check_exists_func(url):
        print(f"포스트 {post_number} (URL: {url})는 이미 크롤링되었습니다. 건너뜁니다.")
        return None

try:
    res = requests.get(url)
    if res.status_code == 404:
        return None  # 게시글이 존재하지 않음
    soup = BeautifulSoup(res.text, 'html.parser')

    # 카테고리 추출
    category_element = soup.select_one('.area_title .tit_category a')
    category = category_element.text if category_element else "카테고리 없음"

    # 제목 추출
    title_element = soup.select_one('.area_title h3.tit_post')
    title = title_element.text if title_element else "제목 없음"

 

이렇게 구성된 크롤링 코드를 python main.py와 같이 실행시키면 아래 사진과 같이 크롤링이 진행됩니다.

 

제가 지정한 url 범위에서 데이터를 하나씩 수집하고 그것이 완료되는 과정을 볼 수 있습니다. 

이렇게 상태를 보려고 했던 것은 여러 개의 블로그 글을 수집하다보니, 혹시라도 발생하는 오류를 빠르게 캐치하는 것 뿐만 아니라, 진항 상황을 모니터링 할 수 있게 하기 위함입니다. 

 

모든 크롤링 과정이 종료되었으면 위와 같이 마지막에 크롤링 완료가 나오게 해놨습니다. 그리고 실제 DB를 확인해보면 정상적으로 데이터가 저장이 되었음을 확인할 수 있습니다.

또한, 새로 추가된 포스트가 몇 개인지, 기존에 수집한 포스팅은 몇 개인지 등도 체크하도록 해두었습니다.

이렇게 데이터가 수집되었으면 이제 Q&A 챗봇을 만들 준비가 50%는 끝났다고 볼 수 있습니다!

---

3. 데이터 청킹(Chunking) 및 벡터(vector) 추출 후 저장

다음은 데이터 청킹(Chunking)하는 부분과 이 청킹된 데이터를 벡터로 추출해 PostgreSQL PGVector를 사용해서 저장하는 과정입니다.

데이터를 청킹하는 과정은 블로그 포스팅 글이 매우 길기 때문에 이를 전부 벡터로 변환시키는 것은 효과적이지 못하기 때문입니다. 이에, 긴 텍스트를 특정 조건 + 크기로 쪼갠 뒤 이를 벡터로 변환시키고 저장하도록 합니다.

 

3-1. 데이터 청킹 과정

텍스트 chunking 과정은 다른 말로 텍스트 분할(text split) 과정이라고도 표현합니다. 저는 이 Text split 과정은 langchain을 이용해 진행했습니다. Langchain에서 제공해주는 텍스트 청킹은 다양한 방법이 있는데요. 저는 그 중 RecursiveCharacterTextSplitter( https://python.langchain.com/docs/how_to/recursive_text_splitter/ )를 사용했습니다. 사용한 Python langchain 코드는 다음과 같습니다.

 

from langchain.text_splitter import RecursiveCharacterTextSplitter
from config.settings import CHUNK_SIZE, CHUNK_OVERLAP


async def chunk_blog_post(post: BlogPost) -> List[int]:
    """
    블로그 포스트를 청킹하고 데이터베이스에 저장합니다.
    RecursiveCharacterTextSplitter를 사용하여 자연스러운 분할점에서 텍스트를 청킹합니다.
    """
    # 이미 청킹되었는지 확인
    if await is_post_chunked(post.post_id):
        print(f"Post {post.post_id} has already been chunked.")
        return []
    
    # 텍스트 청킹
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        length_function=len,
        separators=["\n\n", "\n", ".", " ", ""]
    )
    
    # 청킹할 내용 준비
    content = post.content
    if not content:
        return []
    
    # 메타데이터 준비
    metadata = {
        "title": post.title,
        "url": post.url,
        "publication_date": post.publication_date.isoformat() if post.publication_date else None,
    }
    
    # 청킹 수행 - RecursiveCharacterTextSplitter가 내부적으로 재귀적 분할을 처리
    chunks = text_splitter.create_documents([content], [metadata])
    chunk_ids = []
    
    # 청크 저장
    for i, chunk in enumerate(chunks):
        chunk_text = chunk.page_content
        
        # 청크 해시 생성
        chunk_hash = generate_chunk_hash(chunk_text)
        
        # 청크 저장
        chunk_id = await save_chunk(
            post_id=post.post_id,
            chunk_text=chunk_text,
            chunk_index=i,
            chunk_hash=chunk_hash,
            metadata=chunk.metadata
        )
        
        chunk_ids.append(chunk_id)

 

RecursiveCharacterTextAplitter를 사용해 텍스트 청킹을 진행하며, Chunk_size와 overlap을 기준으로 크기와 중복되는 영역을 설정하였습니다. 그리고 이렇게 분리된 텍스트는 chunking content_chunking에 저장되도록 구성해놨습니다.

 

3-2. 텍스트 임베딩 벡터 추출 및 벡터데이터베이스 저장 과정

이렇게 chunking된 텍스트를 기준으로 이제 텍스트 임베딩 벡터(Text embedding vector)를 추출하고 이를 embeddings 테이블에 저장합니다. 저는 embedding vector로 허깅페이스(huggingface)에 올려와있는 오픈된 모델을 사용했는데요. 그 중 nlpai-lab/KURE-v1 모델( https://huggingface.co/nlpai-lab/KURE-v1 ) 을 사용했습니다. 해당 모델은 KoE5 모델보다 긴 시퀀스 길이를 가지고 있으며, 차원수는 1024, 성능도 괜찮다고 알려진 모델입니다. 

 

해당 모델을 사용하려면 sentence_transformer를 이용하면 됩니다. 아래는 제가 사용한 예제 코드입니다.

반응형

EMBEDDING_MODEL_NAME = "nlpai-lab/KURE-v1"

# 모델 로드
model = None

def load_embedding_model():
    """임베딩 모델을 로드합니다."""
    global model
    if model is None:
        model = SentenceTransformer(EMBEDDING_MODEL_NAME)

def create_embedding(text: str) -> List[float]:
    """텍스트에 대한 임베딩 벡터를 생성합니다."""
    # 모델이 로드되지 않았다면 로드
    if model is None:
        load_embedding_model()
    
    # SentenceTransformer를 사용한 임베딩 생성
    embedding_vector = model.encode(text).tolist()
    
    return embedding_vector

async def embed_chunk(chunk: ContentChunk) -> Optional[str]:
    """청크에 대한 임베딩을 생성하고 저장합니다."""
    # 이미 임베딩이 있으면 건너뛰기
    if chunk.embedding_id:
        return chunk.embedding_id
    
    # 임베딩 생성
    embedding_vector = create_embedding(chunk.chunk_text)
    if embedding_vector is None:
        return None
    
    # 임베딩 저장
    embedding_id = await save_embedding(
        chunk_id=chunk.chunk_id,
        embedding_vector=embedding_vector,
        model_name=EMBEDDING_MODEL_NAME
    )
    
    return embedding_id

 

미리 임베딩 모델 이름 KURE-v1을 설정해두고, sentence_transformer를 이용해 모델을 load합니다. 이 load된 model을 활용해서 model.encode(text)를 통해 텍스트에서 임베딩 벡터를 추출할 수 있는데요. chunking된 텍스트를 가지고 온 후 임베딩 모델을 활용해 벡터를 추출한 뒤, 추출된 정보를 embeddings  테이블에 저장하는 프로세스로 진행됩니다. 이때 chunk_id도 같이 저장하여 어떤 chunk에 해당되는 임베딩 벡터 정보인지 확인할 수 있도록 foreign key 값으로 설정해주었습니다. 

 

아래 사진은 블로그 본문 데이터를 청킹하고 임베딩을 넣는 과정을 보여줍니다. 

 

로그를 확인한 결과 completed embedding이 나오면서 모든 chunk가 수행이 완료된 것을 확인할 수 있습니다.

 

또한, 실제 DB를 확인해보면 왼쪽 사진과 같이 블로그 데이터가 chunking되어 저장된 것을 확인할 수 있습니다. 그리고 오른쪽 사진은 chunking을 수행했는지 유무를 체크하는 DB인데요. 정상적으로 데이터가 체크가 된 것을 확인할 수 있습니다.

---

 

4. FastAPI를 이용한 Ollama LLM 통신

이제 Q&A 챗봇 RAG 구축이 80% 완료되었습니다. 앞선 과정을 정리하자면, 티스토리에서 블로그 글을 크롤링한 후 저장하였고, 그 저장된 데이터를 본문을 기준으로 chunking(text split)을 진행하였으며, chunking된 text를 huggingface embedding model을 활용해 text embedding vector를 추출한 뒤 저장하였습니다.

이제 RAG의 역할을 수행하는 LLM을 붙이면 되는데요. 저는 Ollama를 활용해서 간단히 local LLM 환경으로 동작시켰습니다. 

 

저는 LLM 모델로 llama3.2-bllossom-3b 모델( https://huggingface.co/Bllossom/llama-3.2-Korean-Bllossom-3B )을 사용하였습니다. 또한, 요청을 하는 클라이언트가 바로 Ollama와 통신하는 것이 아니라, 중간에 FastAPI 서버를 두어, FastAPI가 Ollama와 통신하도록 구성했습니다. 제가 구성한 Python FastAPI 코드는 아래 예시와 같습니다.

 

router = APIRouter()

@router.post("/query", response_model=QueryResponse)
async def query_blog(
    request: QueryRequest, 
    session: AsyncSession = Depends(get_db_session)
):
    """블로그 내용에 대한 질문에 답변합니다."""
    try:
        # 유사한 청크 검색
        search_results = await search_similar_chunks(
            query=request.query,
            session=session, 
            top_k=request.top_k, 
            threshold=request.threshold
        )
        
        # 검색 결과 포맷팅
        formatted_context = await format_search_results(search_results)
        
        # 관련 정보 유무 확인
        has_relevant_info = formatted_context is not None
        
        # LLM으로 응답 생성
        response, success = await generate_response(
            query=request.query,
            context=formatted_context
        )
        
        return QueryResponse(
            response=response,
            has_relevant_info=has_relevant_info if success else False,
            search_results=search_results if (has_relevant_info and success) else None
        )
        
    except Exception as e:
        # logger.error(f"쿼리 처리 중 예상치 못한 오류 발생: {str(e)}")
        # 최종 폴백 응답
        return QueryResponse(
            response="죄송합니다. 요청을 처리하는 중에 오류가 발생했습니다. 잠시 후 다시 시도해 주세요.",
            has_relevant_info=False,
            search_results=None
        )

 

 

이 FastAPI 코드는 사용자의 질문(query, request)이 들어오면 

 

1. 가장 먼저, 질문을 벡터로 변환한 뒤 가장 유사한 chunk를 찾습니다. 이때, top k개수 만큼 찾으며, threshold 이상의 유사도를 가진 chunk를 찾습니다.

2. 결과를 LLM이 읽을 수 있게 format을 변경합니다. format_search_result라는 함수에서 진행하며, 이 함수에서는 단순히 LLM이 읽을 수 있도록 markdown 형식으로 변환하는 구조를 가지고 있습니다.

3. generate_response 함수를 통해 Ollama와 통신할 수 있도록 합니다. 이때, 2번 과정에서 만든 정보를 함께 Ollama에게 제공하여 Ollama에 올라가서 서빙되는 LLM이 응답을 생성할 수 있도록 합니다.

---

5. Streamlit을 활용하여 Q&A Chatbot 구현

자, 이제 LLM 연동까지 끝냈으니 Q&A Chatbot 화면만 구현하면 되겠죠? 챗봇 화면은 그렇게 어렵지 않습니다.

단순히 사용자의 질문을 받을 수 있는 챗봇 형태의 UI를 python streamlit 등을 활용해 구성하면 되고, FastAPI server에 정보를 request한 후 response 받은 정보를 화면에 뿌려주기만하면 됩니다.

 

 

위 사진은 제가 구성한 Chatbot 형식의 UI를 가진 Python streamlit 화면입니다. 질문을 입력하도록 되어있고, 이러한 Q&A가 계속 반본적으로 이어질 수 있도록 수행합니다. 

만약, 사용자가 어떤 질의사항이 있다고 하면 질문을 입력하는 text input 란에 입력하면 되는데요.

저는 제가 일전에 제 블로그에 업로드한 프롬프트 엔지니어링 기법을 검색하기 위해서 '프롬프트 기법 중 ReAct 프롬프팅이나 one-shot, few-shot prompt에 대해서 소개한 자료가 있을까?'를 검색해서 물어보았습니다.

 

이제 저 질문을 FastAPI 서버에서 받고 아래와 같은 프로세스로 응답을 처리합니다.

 

1.  '프롬프트 기법 중 ReAct 프롬프팅이나 one-shot, few-shot prompt에 대해서 소개한 자료가 있을까?' 질문이 request로 들어옵니다.

2. request로 들어온 텍스트를 embedding vector 모델( 임베딩을 수행했던 모델과 동일한 모델 )을 활용해 벡터로 변환합니다.

3. PostgreSQL의 PGVector를 활용해 저장된 embeddings 테이블에서 코사인 유사도로 유사도 검색을 수행합니다.

4. 유사도가 높은 chunk_id에 따라 post_id까지 join하여 원본 블로그 글을 가져옵니다.

5. 해당 글의 정보를 LLM에게 넘겨주어, 응답을 생성하게 합니다.

6. reference 자료와 llm의 결과를 client에게 넘겨줍니다.(response)

7. 클라이언트는 해당 값을 받아 화면에 출력합니다.

 

 

위 사진은 제가 질문한 질의에 따라 LLM이 생성해준 결과가 streamlit 화면에 출력되는 사진입니다.

제가 이전에 작성한 프롬프트 기법 글이 있고, 거기에 ReAct와 one-shot, few-shot과 관련된 글이 있기에 그 글들을 조합해 LLM이 응답을 생성해준 것입니다.

 

 

또한 저는 위 사진처럼 각 chunk마다 참조한 원 본 글(레퍼런스 글, reference)가 나오도록 했습니다. 즉, 검색된 chunk text가 어디 블로그에 있는 것이고, 그 chunk가 무슨 내용인지 참조 정보로 보여주는 것이죠. 

---

마무리

이번 포스팅은 Python을 활용해 내 블로그 Q&A 챗봇 RAG를 간단하게 만들어보았습니다. 벡터 데이터베이스로는 PostgreSQL PGVector를 사용했고 Python의 FastAPI, Langchain을 활용하였으며, LLM은 Ollama를 사용했습니다.

제 블로그이지만 제 스스로 블로그 Q&A 챗봇이 필요해서 만들어보았는데요. 나름 쏠쏠하게 잘 쓰고 있습니다 ㅎㅎ

여러분들도 블로그를 하고 계시다면 한 번 만들어보시는 것은 어떠실까요?