[AI] 챗봇 프로젝트: GCP를 활용한 파이썬 챗봇 만들기 (5) RAG 생성 및 백엔드 연결

[AI] 챗봇 프로젝트: GCP를 활용한 파이썬 챗봇 만들기 (5) RAG 생성 및 백엔드 연결

목차

---

• (1) 기획 단계
• (2) BigQuery로 데이터 가공하기
• (3) 임베딩 (Embedding)
• (4) 유사도 검색 (Search)

---

안녕하세요, 거의 두 달만에 블로그를 작성하는 것 같아요 😅 일이 너무 바빠 포스팅에 신경을 쓰지 못했네요.

RAG 구조는 이미 다 생성 한채로 마무리를 하고 포스팅을 뒤늦게 작성하는거라 기억을 더듬어보겠습니다..!

 

앞선 포스팅에서 벡터 검색 기능(Vector Similarity Search)를 구현 했었습니다.

`top_k` 개의 인덱스를 추출하여 리스트에 저장하는 것 까지 해보았는데요.

 

이번 포스팅에서는 해당 유사도를 가지고 사용자의 질문에 대응할 수 있는 프롬프트를 만드는 RAG를 구현 해보고,

백엔드와 연결해서 실제 API 형태로 동작하도록 구성해보겠습니다.

 

📑RAG 

RAG(Retrieval-Augmented Generation)는 이전에 포스팅 한 적이 있었습니다.

[LLM] RAG (Retrieval-Augmented Generation)

RAG는 Retrieval-Augmented Generation의 약자로 검색 기반 생성을 의미합니다.

쉽게 말해 외부 데이터베이스나 지식 그래프에서 관련 정보를 검색한 후 이를 바탕으로 답변을 생성하는 방식을 RAG라고 합니다.

 

지금 상황에 맞는 RAG 구조를 생성하기 위해서 해야하는 과정은 아래와 같습니다.

 

1) 사용자의 질문에 대한 유사도 문서 검색

2) 유사도 문서를 기반으로 AI에 학습 할 프롬프트 생성

3) 모델은 해당 프롬프트 기반으로 답변 

 

+) 추가 기능 (굳이 없어도 되지만 저는 아래와 같은 기능을 추가했습니다.)

• 사용자 대화 히스토리 저장
• 히스토리 초기화
• 스트림 기능

그럼 구현을 시작해봅시다!

 

⚙️ RAG 구조 구현하기

벡터 검색 기반 유사 문서를 활용해, Gemini 모델에게 던질 프롬프트를 조합하고, 스트리밍 방식으로 답변을 받아오는 흐름을 함수 중심으로 풀어본다.

 

 

전체적인 흐름은 다음과 같습니다.

---

1. 사용자의 입력
2. 벡터 검색으로 유사문서 추출
3. 프롬프트 컨텍스트 조합
4. 모델에게 전달
5. 답변을 점진적으로 클라이언트에 전달
6. (필요시) 히스토리에 저장 / 초기화

---

 

해당 흐름을 염두에 두면서, 함수별로 설명을 해보겠습니다.

 

0. (선택사항) 사용자 대화 히스토리 관리

def add_history(self, user_id: str, message: str):
    if user_id not in self.user_histories:
        self.user_histories[user_id] = []
    self.user_histories[user_id].append(message)
    MAX_HISTORY_LEN = 10
    if len(self.user_histories[user_id]) > MAX_HISTORY_LEN:
        self.user_histories[user_id] = self.user_histories[user_id][-MAX_HISTORY_LEN:]

해당 함수는 사용자의 메시지를 받아서 `user_histories` 딕셔너리에 저장합니다.

오래된 대화는 잘라내어 메모리 과부하나 맥락 혼란을 방지합니다.

def clear_history(self, user_id: str):
    if user_id in self.user_histories:
        del self.user_histories[user_id]

단순히 특정 사용자의 대화 기록을 모두 지우는 함수입니다.

새로운 세션 시작 버튼이나 초기화 명령어에 연결하면 유용하게 사용할 수 있습니다.

 

1. 검색 문서 + 히스토리를 프롬프트 형태로 묶기

벡터 검색을 통해 얻은 유서문서들의 리스트를 프롬프트 형태로 조합하는 과정입니다.

def build_context(self, docs: list, user_id: str = None, max_len=300) -> str:
    context = ""
    for i, doc in enumerate(docs, 1):
        question = doc['question']
        answer = doc['answer']
        if len(question) > 100:
            question = question[:100] + "..."
        if len(answer) > max_len:
            answer = answer[:max_len] + "..."
        context += f"[{i}]\nQ: {question}\nA: {answer}\n\n"

    if user_id and user_id in self.user_histories:
        history_text = "\n".join(self.user_histories[user_id])
        context += f"\n[사용자 대화 히스토리]\n{history_text}"

    return context.strip()

각 문서에 번호를 붙여 정돈된 형태로 만들며, 사용자 히스토리가 있다면 끝에 붙여 맥락을 더 보강할 여지를 제공합니다.

💡`max_len` 값을 조정해서 답변 부분을 더 많이 넣거나 덜 넣을수도 있습니다.

 

2. 스트리밍 방식으로 답변 전달

async def stream_response(self, prompt: str):
    start = time.time()
    responses = self.client.models.generate_content_stream(
        model=self.model,
        contents=[prompt],
    )

    buffer = ""
    for chunk in responses:
        if hasattr(chunk, "text") and chunk.text:
            buffer += chunk.text
            if "\n" in buffer or len(buffer) > 50:
                yield buffer
                buffer = ""

    if buffer:
        yield buffer

    print(f"✅ Gemini 응답 완료 ({time.time() - start:.2f}초)")

`generate_content_stream` 메서드를 통해 모델의 응답을 스트림 형태로 받아옵니다.

`buffer` 변수에 누적하다가, 줄바꿈이 생기거나 일정 길이를 넘으면 그 부분까지 `yield`로 넘깁니다.

이렇게 사용하면 사용자 입장에서는 답변이 점진적으로 나오는 느낌을 받을 수 있어요

 

아래는 위 함수를 조합한 클래스의 코드입니다. 

import time
from google import genai
from .config import *
from .search import VectorSearchService

class RagService:
    def __init__(self):
        self.client = genai.Client(vertexai=True, project=PROJECT_ID, location="us-central1")
        self.model = "gemini-2.5-pro"
        self.vs = VectorSearchService()
        self.user_histories = {}

    def add_history(self, user_id: str, message: str):
        if user_id not in self.user_histories:
            self.user_histories[user_id] = []
        self.user_histories[user_id].append(message)
        MAX_HISTORY_LEN = 10
        if len(self.user_histories[user_id]) > MAX_HISTORY_LEN:
            self.user_histories[user_id] = self.user_histories[user_id][-MAX_HISTORY_LEN:]

    def build_context(self, docs: list, user_id: str = None, max_len=300) -> str:
        context = ""
        for i, doc in enumerate(docs, 1):
            question = doc['question']
            answer = doc['answer']
            if len(question) > 100:
                question = question[:100] + "..."
            if len(answer) > max_len:
                answer = answer[:max_len] + "..."
            context += f"[{i}]\nQ: {question}\nA: {answer}\n\n"

        if user_id and user_id in self.user_histories:
            history_text = "\n".join(self.user_histories[user_id])
            context += f"\n[사용자 대화 히스토리]\n{history_text}"

        return context.strip()

    def clear_history(self, user_id: str):
        if user_id in self.user_histories:
            del self.user_histories[user_id]

    async def stream_response(self, prompt: str):
        start = time.time()
        responses = self.client.models.generate_content_stream(
            model=self.model,
            contents=[prompt],
        )

        # responses는 동기 generator이므로 async for 안됨 → 일반 for문 사용
        buffer = ""
        for chunk in responses:
            if hasattr(chunk, "text") and chunk.text:
                buffer += chunk.text
                if "\n" in buffer or len(buffer) > 50:
                    yield buffer
                    buffer = ""

        if buffer:
            yield buffer

        print(f"✅ Gemini 응답 완료 ({time.time() - start:.2f}초)")

 

💻백엔드 연결하기 (FastAPI)

앞에서 만든 RagService 클래스는 RAG 구조의 핵심 로직을 담당합니다.

이번에는 이걸 FastAPI 서버와 연결하여 실제 API 형태로 동작하도록 구성해봅시다.

 

1. FastAPI 기본 세팅

import os
from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from chatbot import rag

load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), "chatbot", ".env"))
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = os.path.join(
    os.path.dirname(__file__), "chatbot", "credentials", "chatbot-key.json"
)

환경 변수를 불러오고 바로 추가 해줍니다.

 

2. FastAPI 앱 객체 생성

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

rag_service = rag.RagService()

객체 생성을 하면서 `rag_service` 인스턴스를 만들어, 아래 엔드포인트에서 계속 사용합니다. 

 

3. 질문 처리 API 

@app.post("/chat")
async def chat(request: Request):
    body = await request.json()
    question = body.get("question", "")
    user_id = body.get("user_id", "anonymous")

    print(f"\n[📩 사용자 질문] ({user_id}): {question}")
    rag_service.add_history(user_id, f"Q: {question}")

`/chat`이라는 질문 처리 API를 생성합니다.

• 클라이언트에서 JSON `{ question, user_id }` 형태로 요청을 보내면 파싱합니다.
• `user_id`가 없으면 `anonymous` 기본 값을 사용합니다.
• 질문을 먼저 히스토리에 추가합니다.

 

3-1. 벡터 검색 실행

    docs = rag_service.vs.search(question)
    print(f"[📄 검색 문서 수]: {len(docs)}")
    for i, d in enumerate(docs, 1):
        print(f"  ({i}) Q: {d['question'][:30]}... / A: {d['answer'][:30]}...")

• `vs.search()`로 유사 문서를 가져옵니다.
• 몇 개의 문서가 검색됐는지 로그로 확인이 가능합니다.

 

3-2. 프롬프트 생성

    context = rag_service.build_context(docs, user_id=user_id)

    prompt = f"""
    # 역할 및 규칙
    당신은 파이썬에 능숙한 개발자입니다. 역할극은 하지 마시고, 친근하고 자연스러운 존댓말로 질문에 바로 답해주세요.
    문서를 먼저 읽고 추론하지 마세요, 사용자의 질문을 먼저 파악하세요

    다음 규칙을 지켜주세요:
    1. 질문이 파이썬 관련이면 실행 가능한 코드를 먼저 보여주시고, 초보자도 이해할 수 있도록 짧고 쉽게 설명해주세요.
    2. 질문이 파이썬과 관련 없는 경우, 코드 얘기는 꺼내지 말고 알고 있는 선에서 간단하게 답해주세요.
    3. 설명은 딱딱하지 않게, 동료에게 말하듯 자연스럽고 부드럽게 해주세요.
    4. 질문이 애매하면 “혹시 이런 의미일까요?”처럼 부드럽게 되물어봐 주세요.
    5. 참고 문서는 필요할 때만 활용하세요. 질문이 명확하다면 문서를 무시해도 괜찮습니다.
    6. 참고 문서만 보고 먼저 추론하거나 설명하지 마세요.
    7. “문서에 따르면” 같은 표현은 사용하지 마세요.
    8. 인사, 자기소개, “질문해주세요” 같은 말은 하지 마세요. 바로 질문에 반응해 주세요.
    9. 당신은 질문을 받는 입장입니다. “질문해 주세요”, “기다릴게요” 같은 말은 하지 마세요.
    10. 문서가 있어도 절대 먼저 추론하지 말고, “무엇이 궁금하신가요?”처럼 질문의 의도를 먼저 확인해 주세요.

    --- 문맥 ---
    {context}

    --- 질문 ---
    {question}
    """.strip()

규칙이 굉장히 많죠 😅 처음에 간단한 규칙을 넣어서만 질문을 했더니.. 모델이 다음과 같은 답변을 하는거에요.

🤖 네 보내주신 문서를 참고하여 답변드리겠습니다 ! 

😲 ???...

 

어떠한 문서를 참고하는 지 모르는 사용자의 입장에서는 모델의 답변이 아주 황당할 것입니다.

단순히 검색 문서를 그대로 붙이는 게 아니라, 모델에게 지켜야 할 규칙까지 함께 넣어 프롬프트를 구성했습니다.

 

3-3. 스트리밍 응답 처리

    async def event_generator():
        answer_buffer = []
        async for chunk in rag_service.stream_response(prompt):
            print(f"[🧩 응답 조각]: {repr(chunk)}")
            answer_buffer.append(chunk)
            yield chunk.encode("utf-8")
        rag_service.add_history(user_id, "A: " + "".join(answer_buffer))
        print(f"\n✅ 전체 응답 완료\n")

    return StreamingResponse(event_generator(), media_type="text/plain; charset=utf-8")

• `event_generator()` 안에서 `stream_response()`를 호출하면 모델 응답이 청크 단위로 들어옵니다.
• 들어오는 즉시 `yield`하여 클라이언트사 실시간으로 받을 수 있습니다.
• 최종적으로 합친 답변을 다시 히스토리에 저장합니다.

 

4. 히스토리 초기화 API

@app.post("/clear_history")
async def clear_history(request: Request):
    body = await request.json()
    user_id = body.get("user_id")
    if user_id:
        rag_service.clear_history(user_id)
        print(f"🧹 히스토리 초기화됨: {user_id}")
    return {"status": "ok"}

프론트엔드에서 특정 유저의 대화 기록을 지우고 싶을 때 호출되는 기능입니다.

 

📌전체 동작 요약

1. `/chat` 엔드포인트를 호출하여 질문과 사용자의 ID를 전달합니다.
2. 벡터 검색을 하여 문맥, 히스토리 기반으로 프롬프트를 생성합니다.
3. Gemini의 응답을 스트리밍으로 받아 클라이언트에 실시간으로 전달합니다.
4. 답변을 다시 히스토리에 저장해 다음 대화 맥락을 유지합니다.
5. `/clear_history`로 언제든 초기화가 가능합니다.

 

전체 코드는 아래를 참고해주세요

import os
from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from chatbot import rag

load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), "chatbot", ".env"))
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = os.path.join(
    os.path.dirname(__file__), "chatbot", "credentials", "chatbot-key.json"
)

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

rag_service = rag.RagService()

@app.post("/chat")
async def chat(request: Request):
    body = await request.json()
    question = body.get("question", "")
    user_id = body.get("user_id", "anonymous")

    print(f"\n[📩 사용자 질문] ({user_id}): {question}")
    rag_service.add_history(user_id, f"Q: {question}")

    docs = rag_service.vs.search(question)
    print(f"[📄 검색 문서 수]: {len(docs)}")
    for i, d in enumerate(docs, 1):
        print(f"  ({i}) Q: {d['question'][:30]}... / A: {d['answer'][:30]}...")

    context = rag_service.build_context(docs, user_id=user_id)

    prompt = f"""
    # 역할 및 규칙
    당신은 파이썬에 능숙한 개발자입니다. 역할극은 하지 마시고, 친근하고 자연스러운 존댓말로 질문에 바로 답해주세요.
    문서를 먼저 읽고 추론하지 마세요, 사용자의 질문을 먼저 파악하세요

    다음 규칙을 지켜주세요:
    1. 질문이 파이썬 관련이면 실행 가능한 코드를 먼저 보여주시고, 초보자도 이해할 수 있도록 짧고 쉽게 설명해주세요.
    2. 질문이 파이썬과 관련 없는 경우, 코드 얘기는 꺼내지 말고 알고 있는 선에서 간단하게 답해주세요.
    3. 설명은 딱딱하지 않게, 동료에게 말하듯 자연스럽고 부드럽게 해주세요.
    4. 질문이 애매하면 “혹시 이런 의미일까요?”처럼 부드럽게 되물어봐 주세요.
    5. 참고 문서는 필요할 때만 활용하세요. 질문이 명확하다면 문서를 무시해도 괜찮습니다.
    6. 참고 문서만 보고 먼저 추론하거나 설명하지 마세요.
    7. “문서에 따르면” 같은 표현은 사용하지 마세요.
    8. 인사, 자기소개, “질문해주세요” 같은 말은 하지 마세요. 바로 질문에 반응해 주세요.
    9. 당신은 질문을 받는 입장입니다. “질문해 주세요”, “기다릴게요” 같은 말은 하지 마세요.
    10. 문서가 있어도 절대 먼저 추론하지 말고, “무엇이 궁금하신가요?”처럼 질문의 의도를 먼저 확인해 주세요.

    --- 문맥 ---
    {context}

    --- 질문 ---
    {question}
    """.strip()

    async def event_generator():
        answer_buffer = []
        async for chunk in rag_service.stream_response(prompt):
            print(f"[🧩 응답 조각]: {repr(chunk)}")
            answer_buffer.append(chunk)
            yield chunk.encode("utf-8")
        rag_service.add_history(user_id, "A: " + "".join(answer_buffer))
        print(f"\n✅ 전체 응답 완료\n")

    return StreamingResponse(event_generator(), media_type="text/plain; charset=utf-8")


@app.post("/clear_history")
async def clear_history(request: Request):
    body = await request.json()
    user_id = body.get("user_id")
    if user_id:
        rag_service.clear_history(user_id)
        print(f"🧹 히스토리 초기화됨: {user_id}")
    return {"status": "ok"}

---

🧩 완성 & 마무리하며

짜잔

로그

 

프론트엔드는 리액트로 구현하였습니다. 다만 제가 프론트엔드쪽에는 지식이 얕아 gpt의 도움을 많이 받았으므로 😅

따로 포스팅에는 작성하지 않았습니다.

 

이번에 진행한 챗봇 프로젝트는 단순히 모델을 불러다 쓰는 수준이 아니라, 기획부터 데이터 전처리, 임베딩, 검색, 그리고 백엔드 서비스 연결까지 한 사이클을 모두 경험한 과정이었어요. 

 

이 과정을 통해 얻은 가장 큰 성과는 AI 챗봇의 전체 아키텍처를 처음부터 끝까지 직접 손으로 구현해봤다는 점입니다. 

단순히 모델을 불러와 돌려본 것이 아니라 데이터 가공 -> 임베딩 -> 검색 -> 응답 생성 -> API 서비스까지 이어지는 전체 플로우를 구축한 경험이 앞으로 큰 자산이 될 것 같아요.

 

다음 단계는 아마도 최적화와 배포일 겁니다.

• 응답 속도 개선 --> 이거 너무 어려워요...ㅠ 지금은 평균 답변 속도가 15초~20초 가량입니다 😥
• 프롬프트 고도화

이런 부분들을 다듬으면 더욱 완성된 챗봇이 될 수 있을 것 같아요

 

한 사이클을 끝까지 완주했다는 게 가장 의미 있는 지점이라, 이번 경험은 앞으로의 모든 프로젝트에서 든든한 기반이 될 것 같습니다. 🚀