15차시. 기본 RAG 앱 만들기
5. 실습 ③ — rag.py (핵심)
출처: 원본 PDF 15차시_기본RAG앱만들기.pdf 10쪽 · 쪽 그림 10쪽
검색과 생성을 담당합니다. 외부에는 ask() 하나만 노출합니다.
📝 rag.py
import re
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
import config
from indexer import get_store
from prompts import PROMPTS
load_dotenv()
_store = get_store()
_llm = ChatOpenAI(model=config.LLM_MODEL,
temperature=config.TEMPERATURE)
_chain = PROMPTS[config.PROMPT_VER] | _llm | StrOutputParser()
def _search(question, k=None):
k = k or config.TOP_K
pairs = _store.similarity_search_with_relevance_scores(
question, k=k)
# 기준(MIN_SCORE)을 넘는 것만 골라 담기
docs = []
for d, s in pairs:
if s >= config.MIN_SCORE:
docs.append(d)
return docs
def _build_context(docs):
parts = []
for i, d in enumerate(docs, 1):
parts.append(f"[{i}] {d.metadata['filename']} "
f"p.{d.metadata['page_no']}\n{d.page_content}")
return "\n\n---\n\n".join(parts)
def ask(question: str, k: int = None) -> dict:
# 외부에 노출되는 유일한 함수
if not question or not question.strip():
return {"answer": "질문을 입력해주세요.",
"sources": [], "ok": False}
try:
docs = _search(question, k)
except Exception as e:
print("[검색 오류]", e)
return {"answer": config.MSG_ERROR, "sources": [], "ok": False}
if not docs:
return {"answer": config.MSG_NO_DOC, "sources": [], "ok": True}
try:
answer = _chain.invoke({"context": _build_context(docs),
"question": question})
except Exception as e:
print("[생성 오류]", e)
return {"answer": config.MSG_ERROR, "sources": [], "ok": False}
# 인용 번호 [1][2]... 를 뽑아 문서 개수 범위 안인지 확인
found = re.findall(r"\[(\d+)\]", answer)
nums = []
for n in found:
nums.append(int(n))
cited = len(nums) > 0
for n in nums:
if n < 1 or n > len(docs):
cited = False
sources = []
for d in docs:
sources.append({"file": d.metadata["filename"],
"page": d.metadata["page_no"]})
return {"answer": answer, "sources": sources,
"cited": cited, "ok": True}실습 코드 · rag_app/rag.py
실습 코드: 실습/src/rag_app/rag.py · 위의 원본 PDF 15차시 10쪽 「5️⃣ 실습 ③ — rag.py (핵심)」과 이어집니다.
# ==============================================================
# 15차시. PPT 15차시_기본RAG앱만들기. 코드 폴더는 src/rag_app
# PPT: 15차시 문서 질의응답
# 역할: 한 함수 ask() 안에 검색(Retriever)과 답 생성(Generator), 인용 검사가 순서대로 들어 있습니다.
# 읽는 키: 인자 question, k. LangGraph State 키는 없습니다.
# 쓰는 키: 반환 딕셔너리 answer, sources, cited, ok. 실패 시 sources는 빈 리스트
# 그래프:
# question
# |
# v
# _search (FAISS, MIN_SCORE 이상만)
# |-- 문서 없음 --> MSG_NO_DOC
# +-- 문서 있음 --> _build_context --> LLM --> 인용 번호 검사 --> 반환
# --------------------------------------------------------------
# 파이썬만 아는 분을 위한 말
# State는 노드들이 함께 쓰는 딕셔너리입니다. 함수는 state를 읽고, 바꿀 키만 return 합니다.
# return에 적지 않은 키는 이전 값이 남습니다.
# log와 tried_queries는 리스트라서, 새로 return한 항목이 뒤에 이어 붙습니다.
# Retriever: 검색 담당. 질문과 비슷한 문서 조각만 고릅니다. 답 문장은 만들지 않습니다.
# Generator: 고른 문서를 근거로 답 문장을 만듭니다. 문서가 없으면 안내 문장만 돌려줍니다.
# Verifier: 답이 근거와 맞는지 검사하고 grade를 적습니다.
# grade=pass 답을 확정하고 그래프를 끝냅니다.
# grade=retry 같은 문서로 답 문장만 다시 씁니다. retries가 1 늘어납니다.
# grade=research 문서가 부족합니다. 검색어를 바꾼 뒤 다시 찾습니다. rewrites가 늘어납니다.
# grade=giveup 더 진행하지 않고 fallback으로 안내 문장만 돌려줍니다.
# fallback: 실패했을 때 답을 지어내지 않고 안내 문장을 반환하는 노드입니다.
# intent: 질문 종류. greeting 인사, calc 계산, scope 문서 밖, document 문서 검색.
# rewrite: 사용자가 친 question은 유지하고, 검색에 쓰는 query만 바꿉니다.
# safe_node: 노드를 try로 감쌉니다. 오류가 나도 그래프는 멈추지 않고 node_error에 적습니다.
# 이 파일은 LangGraph를 쓰지 않습니다. State 딕셔너리 대신 ask()의 인자 question과 반환 dict를 씁니다.
# _search가 Retriever, chain.invoke가 Generator 역할입니다.
# cited는 답의 [숫자]가 하나 이상 있고, 모두 1 이상 문서 개수 이하일 때 True입니다.
# sources의 각 항목은 file, page 입니다.
# 파일을 import하면 load_dotenv와 get_store, ChatOpenAI가 바로 실행되므로 이 교안에서는 실행하지 않습니다.
# ==============================================================
# ============================================================
# 사용자의 질문과 관련된 문서를 FAISS에서 검색하고,
# 검색된 문서를 근거로 OpenAI LLM이 최종 답변을 생성합니다.
# 답변의 출처와 인용 번호가 올바른지도 함께 확인합니다.
# ============================================================
import re
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
import config
from indexer import get_store
from prompts import PROMPTS
# ============================================================
# 1. 기본 환경 준비
# ============================================================
# .env 파일의 OPENAI_API_KEY를 환경변수로 불러옵니다.
load_dotenv()
# FAISS 벡터 저장소를 준비합니다.
store = get_store()
# OpenAI LLM을 준비합니다.
llm = ChatOpenAI(
model=config.LLM_MODEL,
temperature=config.TEMPERATURE
)
# 프롬프트 → LLM → 문자열 변환 순서로 실행 체인을 만듭니다.
chain = (
PROMPTS[config.PROMPT_VER]
| llm
| StrOutputParser()
)
# ============================================================
# 2. 검색 함수
# ============================================================
# 질문과 관련된 문서를 벡터 저장소에서 검색합니다.
# MIN_SCORE 이상의 문서만 최종 검색 결과로 사용합니다.
# ============================================================
def _search(question, k=None):
# 검색 개수가 지정되지 않으면 기본값을 사용합니다.
if k is None:
k = config.TOP_K
# 질문과 유사한 문서를 검색합니다.
results = store.similarity_search_with_relevance_scores(
question,
k=k
)
# 기준 점수를 통과한 문서만 저장합니다.
docs = []
for doc, score in results:
if score >= config.MIN_SCORE:
docs.append(doc)
return docs
# ============================================================
# 3. 컨텍스트 생성 함수
# ============================================================
# 검색된 문서들을 LLM에게 전달하기 좋은 문자열로 만듭니다.
# 각 문서 앞에 번호, 파일명, 페이지 번호를 표시합니다.
# ============================================================
def _build_context(docs):
# 문서별 문자열을 저장할 리스트입니다.
parts = []
# 검색된 문서를 하나씩 처리합니다.
for number, doc in enumerate(docs, 1):
# 원본 파일 이름을 가져옵니다.
filename = doc.metadata.get(
"filename",
"unknown"
)
# 페이지 번호를 가져옵니다.
page_no = doc.metadata.get("page_no")
# page_no가 없으면 기본 page 값을 사용합니다.
if page_no is None:
page = doc.metadata.get("page", 0)
page_no = page + 1
# 출처 정보와 문서 내용을 하나의 문자열로 만듭니다.
part = (
f"[{number}] {filename} p.{page_no}\n"
f"{doc.page_content}"
)
parts.append(part)
# 문서 사이에 구분선을 넣어 하나의 문자열로 만듭니다.
context = "\n\n---\n\n".join(parts)
return context
# ============================================================
# 4. 질문 처리 함수
# ============================================================
# 질문을 받아 관련 문서를 검색하고,
# 검색 결과를 근거로 LLM에게 답변을 생성시킵니다.
# ============================================================
def ask(question: str, k: int = None) -> dict:
# --------------------------------------------------------
# 질문 입력 확인
# --------------------------------------------------------
# 질문이 없거나 공백만 있으면 종료합니다.
if not question or not question.strip():
return {
"answer": "질문을 입력해주세요.",
"sources": [],
"ok": False
}
# --------------------------------------------------------
# 관련 문서 검색
# --------------------------------------------------------
try:
docs = _search(question, k)
except Exception as error:
print("[검색 오류]", error)
return {
"answer": config.MSG_ERROR,
"sources": [],
"ok": False
}
# --------------------------------------------------------
# 검색 결과 확인
# --------------------------------------------------------
# 관련 문서를 찾지 못했다면 LLM을 호출하지 않습니다.
if not docs:
return {
"answer": config.MSG_NO_DOC,
"sources": [],
"ok": True
}
# --------------------------------------------------------
# LLM에게 전달할 근거 문서 만들기
# --------------------------------------------------------
context = _build_context(docs)
# --------------------------------------------------------
# LLM 답변 생성
# --------------------------------------------------------
try:
answer = chain.invoke({
"context": context,
"question": question
})
except Exception as error:
print("[생성 오류]", error)
return {
"answer": config.MSG_ERROR,
"sources": [],
"ok": False
}
# --------------------------------------------------------
# 답변의 인용 번호 확인
# --------------------------------------------------------
# 답변에서 [1], [2] 같은 인용 번호를 찾습니다.
found = re.findall(
r"\[(\d+)\]",
answer
)
# 찾은 문자열 번호를 정수로 변환합니다.
citation_numbers = []
for number in found:
citation_numbers.append(int(number))
# 인용 번호가 하나라도 있으면 True로 시작합니다.
cited = False
if len(citation_numbers) > 0:
cited = True
# 실제 문서 범위를 벗어난 인용이 있는지 확인합니다.
for number in citation_numbers:
if number < 1 or number > len(docs):
cited = False
# --------------------------------------------------------
# 출처 정보 정리
# --------------------------------------------------------
sources = []
# 검색에 사용된 문서의 출처를 하나씩 저장합니다.
for doc in docs:
# 파일 이름을 가져옵니다.
filename = doc.metadata.get(
"filename",
"unknown"
)
# 페이지 번호를 가져옵니다.
page_no = doc.metadata.get("page_no")
# page_no가 없으면 기본 page 값을 사용합니다.
if page_no is None:
page = doc.metadata.get("page", 0)
page_no = page + 1
# 출처 정보를 저장합니다.
source = {
"file": filename,
"page": page_no
}
sources.append(source)
# --------------------------------------------------------
# 최종 결과 반환
# --------------------------------------------------------
return {
"answer": answer,
"sources": sources,
"cited": cited,
"ok": True
}실행 결과 샘플 (교안 기록. API 키가 필요한 코드는 키 없이 맞춘 예시이고, 모델의 답은 실행할 때마다 달라집니다)
===== 실행 결과 샘플 =====
예시 출력 (API 키 필요, 실행하지 않음)
반환 모양:
answer: LLM이 만든 문장 (원문은 기록에 없음)
sources: [{file, page}, ...]
cited: 인용 번호가 문서 범위 안이면 True
ok: 빈 질문이거나 오류면 False, 안내 문장만 돌려준 경우도 검색 실패 안내는 True
기록된 같은 종류의 문서 질문:
환불은 며칠 이내에 신청해야 하나요? elapsed=6.69, 다른 실행 elapsed=2.39 n_docs=2 top_score=0.056 grade=pass
대표이사가 누구인가요? grade=giveup rewrites=2 elapsed=3.41 fail_reason=low_score top_score=-0.219
intent, retries, rewrites는 이 ask() 반환 키가 아닙니다. 이후 그래프 로그의 값입니다.코드 해설
한 함수 ask() 안에 검색(Retriever)과 답 생성(Generator), 인용 검사가 순서대로 들어 있습니다.
코드 앞에서 고정해 두는 말입니다.
- 읽는 키: 인자 question, k. LangGraph State 키는 없습니다.
- 쓰는 키: 반환 딕셔너리 answer, sources, cited, ok. 실패 시 sources는 빈 리스트
- 그래프:
- question
- |
- v
- _search (FAISS, MIN_SCORE 이상만)
- |-- 문서 없음 --> MSG_NO_DOC
함수와 클래스로 나누어 읽습니다.
_search: 이름 그대로 한 가지 일만 합니다. 본문 주석을 같이 읽습니다._build_context: 이름 그대로 한 가지 일만 합니다. 본문 주석을 같이 읽습니다.ask: 이름 그대로 한 가지 일만 합니다. 본문 주석을 같이 읽습니다.
파일 맨 아래 실행 결과 샘플 주석은 실행되지 않습니다. 화면 출력은 다음 절에 다시 적어 두었습니다.