어제와 마찬가지로 pdf 자료없이, 코드작성만 했다.
코드 분석 rag을 추가했다.
app추가. rag
"""OpenAI + ChromaDB 기반 RAG 검색 앱 설정입니다."""
from django.apps import AppConfig
class RagConfig(AppConfig):
"""문서와 게시글을 벡터 검색하는 RAG 앱입니다."""
default_auto_field = "django.db.models.BigAutoField"
name = "rag"
verbose_name = "RAG 검색"
url.py
appname은 rag
같은 패키지의 views를 가져온다.
"""RAG 앱 URL 설정입니다."""
from django.urls import path
from . import views
app_name = "rag"
urlpatterns = [
path("", views.rag_search, name="search"),
path("api/search/", views.rag_search_api, name="api_search"),
path("documents/<str:filename>/", views.rag_document, name="document"),
]
forms.py
RagQueryForm 자연어 질문가 검색 범위 입력 받는다.
"""RAG 질문 입력과 검색 범위를 검증하는 폼입니다."""
from django import forms
class RagQueryForm(forms.Form):
"""사용자의 자연어 질문과 검색 범위를 입력받습니다."""
SOURCE_CHOICES = [
("all", "전체: 문서 + 게시글"),
("documents", "근거 문서만"),
("boards", "게시글만"),
]
question = forms.CharField(
label="질문",
max_length=1000,
widget=forms.Textarea(
attrs={
"rows": 4,
"placeholder": "예: 상품을 반품하려면 며칠 안에 신청해야 하나요?",
}
),
)
source_scope = forms.ChoiceField(
label="검색 범위",
choices=SOURCE_CHOICES,
initial="all",
)
top_k = forms.IntegerField(
label="검색 문서 수",
min_value=1,
max_value=10,
initial=5,
)
document_loader
hashlib, dataclass, path, pdfreader import
@dataclass 설정.
sha256_text, 해시반환, split_text 텍스트를 겹치게 분할해 문맥 단절을 줄인다.
load_pdf_chunks 각 페이지 텍스트를 추출해 chnik생성
"""docs 폴더의 PDF를 읽고 RAG용 청크로 분할합니다."""
from __future__ import annotations
import hashlib
from dataclasses import dataclass
from pathlib import Path
from pypdf import PdfReader
@dataclass
class TextChunk:
"""벡터DB에 저장할 하나의 텍스트 청크와 메타데이터입니다."""
chunk_id: str
text: str
metadata: dict
def sha256_text(text: str) -> str:
"""텍스트 변경 여부 확인에 사용할 SHA-256 해시를 반환합니다."""
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def split_text(text: str, chunk_size: int = 1000, overlap: int = 150) -> list[str]:
"""문자 단위로 텍스트를 겹치게 분할하여 문맥 단절을 줄입니다."""
normalized = " ".join(text.split())
if not normalized:
return []
chunks: list[str] = []
start = 0
while start < len(normalized):
end = min(start + chunk_size, len(normalized))
chunks.append(normalized[start:end])
if end >= len(normalized):
break
start = max(end - overlap, start + 1)
return chunks
def load_pdf_chunks(pdf_path: Path) -> list[TextChunk]:
"""PDF 각 페이지의 텍스트를 추출하고 페이지별 청크를 생성합니다."""
reader = PdfReader(str(pdf_path))
title = ((reader.metadata.title if reader.metadata else None) or pdf_path.stem).strip()
source_key = f"document:{pdf_path.name}"
chunks: list[TextChunk] = []
for page_number, page in enumerate(reader.pages, start=1):
page_text = page.extract_text() or ""
for chunk_index, chunk_text in enumerate(split_text(page_text)):
content_hash = sha256_text(chunk_text)
chunk_id = sha256_text(
f"{source_key}:{page_number}:{chunk_index}"
)[:32]
chunks.append(
TextChunk(
chunk_id=f"doc-{chunk_id}",
text=chunk_text,
metadata={
"source_type": "document",
"source_key": source_key,
"title": title,
"filename": pdf_path.name,
"page": page_number,
"chunk_index": chunk_index,
"content_hash": content_hash,
},
)
)
return chunks
service.py
dataclass, path, iteravle, settings, openai, board import
같은 패키지의 document_loder 가져오기
@dataclass 여러 값을 묶어서 가지고 다니기 편한 Python 객체를 만들어주는 기능
화면과 api에 전달할 벡터 검색결과 지정하기 SearchResult, 변수처럼 사용가능하게 해주는 @property에 similarity
RagService, init시 openaikey 및 정보 설정.
_embed 임베딩하기
_batched 배치로 나누기
_stored_hashes hash 전환을 읽기
sync_documents pdf를 읽되 변경된것만 재색인하기
_board_payload 게시글하나를 검색가능한 텍스트와 메타데이터로 변환해
sync_boards chromadb와 동기화하기 Django의 MySQL 데이터와 ChromaDB의 벡터 데이터를 서로 맞춰주기 위해서 rag에는 db가 하나마 ㄴ있는게 아니라 mysql에 원본데이터, chromadb에는 검색을 위한 벡터가 있다. django에서 게시글을 수정해버리는데 chromadb에 예전 내용이 그대로 있으면 불일치하기 ㄸ문.
sync_all pdf 문서와 게시글 모두 동기화하기
search 검색
answer
"""OpenAI 임베딩/LLM과 ChromaDB를 연결하는 RAG 핵심 서비스입니다."""
from __future__ import annotations
import hashlib
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable
import chromadb
from django.conf import settings
from openai import OpenAI
from boards.models import Board
from .document_loader import load_pdf_chunks, sha256_text
@dataclass
class SearchResult:
"""화면과 API에 전달할 벡터 검색 결과입니다."""
text: str
metadata: dict
distance: float | None
@property
def similarity(self) -> float | None:
"""Chroma cosine distance를 사람이 보기 쉬운 유사도로 변환합니다."""
if self.distance is None:
return None
return max(-1.0, min(1.0, 1.0 - float(self.distance)))
class RagService:
"""문서/게시글 색인, 벡터 검색, 근거 기반 답변 생성을 담당합니다."""
COLLECTION_NAME = "django_docs_and_boards"
def __init__(self):
api_key = settings.OPENAI_API_KEY
if not api_key:
raise RuntimeError("OPENAI_API_KEY가 설정되지 않았습니다. .env 파일을 확인하십시오.")
self.openai = OpenAI(api_key=api_key)
self.embedding_model = settings.OPENAI_EMBEDDING_MODEL
self.chat_model = settings.OPENAI_CHAT_MODEL
self.docs_dir = Path(settings.RAG_DOCS_DIR)
self.vector_dir = Path(settings.RAG_VECTOR_DB_DIR)
self.vector_dir.mkdir(parents=True, exist_ok=True)
self.chroma = chromadb.PersistentClient(path=str(self.vector_dir))
self.collection = self.chroma.get_or_create_collection(
name=self.COLLECTION_NAME,
# 최신 Chroma 설정 형식으로 HNSW 거리 함수를 cosine으로 지정합니다.
configuration={"hnsw": {"space": "cosine"}},
)
def _embed(self, texts: list[str]) -> list[list[float]]:
"""OpenAI Embeddings API로 여러 텍스트를 벡터로 변환합니다."""
if not texts:
return []
response = self.openai.embeddings.create(
model=self.embedding_model,
input=texts,
)
ordered = sorted(response.data, key=lambda item: item.index)
return [item.embedding for item in ordered]
def _batched(self, items: list, size: int = 64) -> Iterable[list]:
"""API 한 번에 너무 많은 텍스트를 보내지 않도록 배치로 나눕니다."""
for i in range(0, len(items), size):
yield items[i : i + size]
def _stored_hashes(self, source_type: str) -> dict[str, set[str]]:
"""벡터DB에 저장된 source_key별 content_hash 집합을 읽습니다."""
data = self.collection.get(
where={"source_type": source_type},
include=["metadatas"],
)
result: dict[str, set[str]] = {}
for metadata in data.get("metadatas") or []:
if not metadata:
continue
source_key = str(metadata.get("source_key", ""))
content_hash = str(metadata.get("content_hash", ""))
result.setdefault(source_key, set()).add(content_hash)
return result
def sync_documents(self, force: bool = False) -> dict:
"""docs 폴더 PDF를 읽어 변경된 문서만 ChromaDB에 재색인합니다."""
self.docs_dir.mkdir(parents=True, exist_ok=True)
stored = self._stored_hashes("document")
current_source_keys: set[str] = set()
indexed_files = 0
indexed_chunks = 0
skipped_files = 0
for pdf_path in sorted(self.docs_dir.glob("*.pdf")):
chunks = load_pdf_chunks(pdf_path)
if not chunks:
continue
source_key = chunks[0].metadata["source_key"]
current_source_keys.add(source_key)
current_hashes = {c.metadata["content_hash"] for c in chunks}
if not force and stored.get(source_key) == current_hashes:
skipped_files += 1
continue
self.collection.delete(where={"source_key": source_key})
for batch in self._batched(chunks):
texts = [c.text for c in batch]
embeddings = self._embed(texts)
self.collection.upsert(
ids=[c.chunk_id for c in batch],
documents=texts,
metadatas=[c.metadata for c in batch],
embeddings=embeddings,
)
indexed_chunks += len(batch)
indexed_files += 1
for old_source_key in set(stored) - current_source_keys:
self.collection.delete(where={"source_key": old_source_key})
return {
"indexed_files": indexed_files,
"indexed_chunks": indexed_chunks,
"skipped_files": skipped_files,
}
def _board_payload(self, board: Board) -> tuple[str, dict]:
"""게시글 하나를 검색 가능한 텍스트와 메타데이터로 변환합니다."""
text = (
f"게시글 번호: {board.pk}\n"
f"제목: {board.title}\n"
f"작성자: {board.author.display_name}\n"
f"작성일: {board.created_at:%Y-%m-%d %H:%M}\n"
f"내용: {board.content}"
)
source_key = f"board:{board.pk}"
metadata = {
"source_type": "board",
"source_key": source_key,
"board_id": int(board.pk),
"title": board.title,
"author": board.author.display_name,
"content_hash": sha256_text(text),
}
return text, metadata
def sync_boards(self, force: bool = False) -> dict:
"""Django 게시글을 ChromaDB와 증분 동기화합니다."""
stored = self._stored_hashes("board")
current_source_keys: set[str] = set()
indexed = 0
skipped = 0
boards = Board.objects.select_related("author").all()
for board in boards:
text, metadata = self._board_payload(board)
source_key = metadata["source_key"]
current_source_keys.add(source_key)
current_hash = metadata["content_hash"]
if not force and stored.get(source_key) == {current_hash}:
skipped += 1
continue
embedding = self._embed([text])[0]
vector_id = "board-" + hashlib.sha256(source_key.encode("utf-8")).hexdigest()[:24]
self.collection.delete(where={"source_key": source_key})
self.collection.upsert(
ids=[vector_id],
documents=[text],
metadatas=[metadata],
embeddings=[embedding],
)
indexed += 1
for old_source_key in set(stored) - current_source_keys:
self.collection.delete(where={"source_key": old_source_key})
return {"indexed": indexed, "skipped": skipped}
def sync_all(self, force: bool = False) -> dict:
"""PDF 문서와 게시글을 모두 벡터DB에 동기화합니다."""
return {
"documents": self.sync_documents(force=force),
"boards": self.sync_boards(force=force),
}
def search(self, question: str, source_scope: str = "all", top_k: int = 5) -> list[SearchResult]:
"""질문 임베딩과 가장 가까운 문서/게시글 청크를 검색합니다."""
query_embedding = self._embed([question])[0]
where = None
if source_scope == "documents":
where = {"source_type": "document"}
elif source_scope == "boards":
where = {"source_type": "board"}
kwargs = {
"query_embeddings": [query_embedding],
"n_results": max(1, min(top_k, 10)),
"include": ["documents", "metadatas", "distances"],
}
if where:
kwargs["where"] = where
data = self.collection.query(**kwargs)
docs = (data.get("documents") or [[]])[0]
metadatas = (data.get("metadatas") or [[]])[0]
distances = (data.get("distances") or [[]])[0]
return [
SearchResult(text=text, metadata=metadata or {}, distance=distance)
for text, metadata, distance in zip(docs, metadatas, distances)
]
def answer(self, question: str, results: list[SearchResult]) -> str:
"""검색 결과만 근거로 사용하도록 LLM 프롬프트를 구성하여 답변합니다."""
if not results:
return "검색된 근거가 없어 답변할 수 없습니다."
context_blocks = []
for index, result in enumerate(results, start=1):
metadata = result.metadata
if metadata.get("source_type") == "document":
source_label = f"문서: {metadata.get('title')} / {metadata.get('page')}쪽"
else:
source_label = f"게시글 #{metadata.get('board_id')}: {metadata.get('title')}"
context_blocks.append(f"[근거 {index}] {source_label}\n{result.text}")
context = "\n\n".join(context_blocks)
instructions = (
"당신은 Django 수업 프로젝트의 RAG 답변 도우미입니다. "
"반드시 제공된 검색 근거 안에서만 답하십시오. "
"근거에 없는 사실은 추측하지 말고 '제공된 근거에서 확인되지 않습니다'라고 말하십시오. "
"답변은 자연스러운 한국어로 작성하고, 중요한 주장 뒤에는 [근거 1]처럼 근거 번호를 표시하십시오."
)
user_input = f"질문:\n{question}\n\n검색 근거:\n{context}"
response = self.openai.responses.create(
model=self.chat_model,
instructions=instructions,
input=user_input,
)
return response.output_text.strip()
views.py
path, settings, message, login_required, fileresponse, http404, jsonreponse, render, reverse import
같은 패키지의 form과 service 가져오기
@login_required 로그인 되었을때 folrm에서의 RagQueryForm를 가져다 답변 세팅해 return
rag_search_api 파라미터로 검색결과를 json으로 반환함. rag 검색기능을 다른 프로그램에서 사용할 수 있도로 ㄱapi화 하는것. 검색해달라는 요청을 받으면 chromadb에서 검색하고 결과를 json으로 돌려주는 것. 프론트엔드가 쉽게 사용할 수 있게 json으로 반환하는 것. 이것을 굳이 api로만드는 이유는 검색 로직을 화면과 분리할 수 있기 때문이다.
rag_document 근거문서를 제공하기
"""RAG 검색 화면, 검색 API, 근거 PDF 제공 View입니다."""
from pathlib import Path
from django.conf import settings
from django.contrib import messages
from django.contrib.auth.decorators import login_required
from django.http import FileResponse, Http404, JsonResponse
from django.shortcuts import render
from django.urls import reverse
from .forms import RagQueryForm
from .service import RagService
@login_required
def rag_search(request):
"""질문을 받아 벡터 검색 후 OpenAI가 근거 기반 답변을 생성합니다."""
form = RagQueryForm(request.POST or None)
answer = None
results = []
sync_result = None
if request.method == "POST" and form.is_valid():
try:
service = RagService()
# 새 PDF와 게시글 변경분이 있으면 질문 전에 자동으로 증분 반영합니다.
sync_result = service.sync_all(force=False)
results = service.search(
form.cleaned_data["question"],
source_scope=form.cleaned_data["source_scope"],
top_k=form.cleaned_data["top_k"],
)
answer = service.answer(form.cleaned_data["question"], results)
# 템플릿에서 바로 사용할 수 있도록 근거 링크를 메타데이터에 추가합니다.
for result in results:
if result.metadata.get("source_type") == "document":
result.metadata["url"] = reverse(
"rag:document",
kwargs={"filename": result.metadata.get("filename", "")},
)
elif result.metadata.get("source_type") == "board":
result.metadata["url"] = reverse(
"boards:detail",
kwargs={"pk": result.metadata.get("board_id")},
)
except Exception as exc:
messages.error(request, f"RAG 처리 중 오류가 발생했습니다: {exc}")
return render(
request,
"rag/search.html",
{"form": form, "answer": answer, "results": results, "sync_result": sync_result},
)
@login_required
def rag_search_api(request):
"""GET q 파라미터로 벡터 검색 결과를 JSON으로 반환합니다."""
question = request.GET.get("q", "").strip()
scope = request.GET.get("scope", "all")
try:
top_k = max(1, min(int(request.GET.get("top_k", "5")), 10))
except ValueError:
top_k = 5
if not question:
return JsonResponse({"error": "q 질문 파라미터가 필요합니다."}, status=400)
if scope not in {"all", "documents", "boards"}:
return JsonResponse({"error": "scope는 all, documents, boards 중 하나여야 합니다."}, status=400)
try:
service = RagService()
service.sync_all(force=False)
results = service.search(question, source_scope=scope, top_k=top_k)
return JsonResponse(
{
"question": question,
"scope": scope,
"results": [
{
"text": result.text,
"similarity": result.similarity,
"metadata": result.metadata,
}
for result in results
],
},
json_dumps_params={"ensure_ascii": False},
)
except Exception as exc:
return JsonResponse({"error": str(exc)}, status=500, json_dumps_params={"ensure_ascii": False})
@login_required
def rag_document(request, filename):
"""로그인 사용자에게 docs 폴더의 PDF 근거 문서를 안전하게 제공합니다."""
docs_dir = Path(settings.RAG_DOCS_DIR).resolve()
target = (docs_dir / filename).resolve()
if docs_dir not in target.parents or not target.is_file() or target.suffix.lower() != ".pdf":
raise Http404("문서를 찾을 수 없습니다.")
return FileResponse(open(target, "rb"), content_type="application/pdf")
'Personal > SK 네트웍스 AI 캠프' 카테고리의 다른 글
| [SK네트웍스 Family AI 캠프] 32기 16주차 회고: Day57~Day60 (0) | 2026.08.13 |
|---|---|
| SK 네트웍스 AI 캠프 - 4_AI 활용 애플리케이션 개발 - Day58_Django Member Board MVT (0) | 2026.08.12 |
| SK 네트웍스 AI 캠프 - 4_AI 활용 애플리케이션 개발 - Day57 Django 기반 웹 서버 개발 기초 (0) | 2026.08.10 |
| [SK네트웍스 Family AI 캠프] 32기 15주차 회고: Day56 (0) | 2026.08.05 |
| SK 네트웍스 AI 캠프 - 4_AI 활용 애플리케이션 개발 - Day56_HTML5 (0) | 2026.08.05 |
