MCP Model Context Protocol
LLM이 외부도구 Tool, 데이터 Data, 시스템 System 과 표준 방식으로 연결하기 위한 프로토콜.
AI가 프로그램과 대화하기 위한 공통 언어.
기존의 서비스마다 연결 방식이 모두 달라 각각 api를 구현해야했던 단점을 mcp로 모든 tool을 동일한 방식으로 사용하여 통일화했다.
MCP는 AI가 실제 시스템에 접근할 수 있게 해줌으로 보안이 매우 중요하고 최근 MCP 보안에 대한 연구와 권장사항도 활발히 나오고 있다.
Host, Client, Tool, Resource, Prompt
https://standout.tistory.com/1907
MCP란? Model Context Protocol: 사용자를 위한 서버가 아니라 AI를 위한 서버
MCP Model Context ProtocolLLM이 외부도구 Tool, 데이터 Data, 시스템 System 과 표준 방식으로 연결하기 위한 프로토콜.AI가 프로그램과 대화하기 위한 공통 언어. 기존의 서비스마다 연결 방식이 모두 달라
standout.tistory.com
샘플 프로젝트 simple_mcp_server_project 를 직접 작성해보자 .
strftime은 날짜(Date)와 시간(Time) 객체를 원하는 문자열(String) 형식으로 변환하는 함수, 시간을 문자열 형식으로 변환한다(String Format Time)
함수를 작성하고 @mcp.tool() 를 붙였다. 이때 독스트링에 신경을 썼다.
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from mcp.server.fastmcp import FastMCP
BASE_DIR = Path(__file__).resolve().parent.parent
MANUAL_FILE = BASE_DIR / "data" / "manual.txt"
mcp = FastMCP("Simple MCP Server")
@mcp.tool()
def hello(name: str) -> str:
"""이름을 전달받아서 인사말을 만들어서 반환합니다.
Args:
name: 인사말에 포함할 사용자 이름입니다.
Returns:
사용자 이름이 포함된 인사말 문자열입니다.
"""
return f"안녕하세요. {name}님!"
@mcp.tool()
def add(a: int, b: int) -> int:
"""
숫자 2개를 전달받아서 더하기 한 결과를 반환합니다.
Args:
a: 첫번째 숫자
b: 두번째 숫자
Returns:
a와 b를 더한 숫자
"""
return a + b
@mcp.tool()
def get_current_time() -> datetime:
"""
MCP 서버가 실행중인 컴퓨터의 현재 날짜와 시간을 반환합니다.
return:
yyyy-mm-dd hh:mm:ss 형식
"""
current_time = datetime.now()
return current_time.strftime("%Y-%m-%d %H:%M:%S")
리소스의 경우 tool과다르게 url을 지정해준다.
@mcp.resource("manual://guide")
@mcp.resource("manual://guide")
def read_manual() -> str:
"""
data/manual.txt 파일의 내용을 읽어 mcp resource로 제공합니다 .
:return: 사용설명서 파일의 전체 내용 문자열입니다.
"""
if not MANUAL_FILE.exists():
return "manual.txt 파일을 찾읈수없습니다. "
return MANUAL_FILE.read_text(encoding="utf-8")
@mcp.resource("profile://{name}")
def get_profile(name: str) -> str:
"""
URL에 포함된 이름으로 간단 사용자 프로필을 생성합니다.
args:
name: profile://url에 포함된 사용자 이름
:return:
이름이 포함된 간단 르포필 문자열
"""
return (
f"사용자이름: {name}, 학습주제: MCP SERVER"
)
prompt도 지정한다. () 지정해 여러 프롬프트도 만들 수 있다.
@mcp.prompt()
def summarize_document(topic: str, style:str = "쉽게") -> str:
"""
특정 주제 topic을 요약하도록 LLM에 전달할 prompt 생성
args:
:param topic: 요약할 주제
:param style: 요약 문체
:return: LLM에 전달할 최종 prompt 문자열
"""
return (
f"다음주제를 {style} 설명하세요. 핵심개념, 동작과정, 간단한 예제를 포함하세요. 주제: {topic}"
)
run
이 파일을 직접 실행했을 때만 아래 코드를 실행
mcp.run() transport MCP는 Host와 Server가 어떤 통신 방식(Transport) 으로 데이터를 주고받을지 지정
stdio 표준 입력(Standard Input)과 표준 출력(Standard Output)을 이용하여 통신
mcp.run(transport="streamable-http") 웹 서버처럼 실행된다. 회사에서 MCP Server를 운영할경우 사용된다. mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8000
)
- stdio: 로컬에서 AI 애플리케이션(Claude Desktop, Cursor 등)과 MCP Server를 연결할 때 사용.
- streamable-http: 네트워크를 통해 여러 클라이언트가 접속하는 원격 MCP Server를 운영할 때 사용.
if __name__ == "__main__":
mcp.run(transport="stdio")
MCP Inspector
MCP Server를 테스트하는 프로그램
MCP Server를 실행하고, MCP Inspector(테스트 도구)로 연결해서 확인하는 명령어
npx -y @modelcontextprotocol/inspector python server.py
환경변수를 편집할 수 있다 . 정보를 복사할 수 있다.
header 설정가능
세션등의 config 설정가능
connect.
resource 확인가능
prompt
Tools 등록한것들 체스트
외 필요에 따라 rotts, auth, metadata등 설정가능하다.
실습프로젝트 mc_rag_project를 분석해보자.
schemas.py
import pydantic basemodel, field 데이터 모델을 정의하고 검증하기 위한 클래스 데이터 모델의 기본 클래스
ChatRequest() question 설정
SearchRequest() query와 top_k 설정 문서 검색
RagRequest() question와 top_k설정. 질문요청 모델
FileReadRequest() filname 설정
KnowledgeCreateRequest() title, content 설정
"""
FastAPI 요청과 응답 검증에 사용하는 Pydantic 모델을 정의합니다.
"""
# Pydantic의 BaseModel과 Field를 가져옵니다.
from pydantic import BaseModel, Field
# 일반 GPT 질문 요청 모델을 정의합니다.
class ChatRequest(BaseModel):
"""일반 GPT 질문 요청입니다."""
# 비어 있지 않은 질문 문자열을 정의합니다.
question: str = Field(min_length=1, description="GPT에 전달할 질문")
# 문서 검색 요청 모델을 정의합니다.
class SearchRequest(BaseModel):
"""Vector Search 요청입니다."""
# 비어 있지 않은 검색어를 정의합니다.
query: str = Field(min_length=1, description="검색할 질문 또는 키워드")
# 1에서 20 사이의 검색 결과 개수를 정의합니다.
top_k: int = Field(default=4, ge=1, le=20)
# RAG 질문 요청 모델을 정의합니다.
class RagRequest(BaseModel):
"""RAG 답변 생성 요청입니다."""
# 비어 있지 않은 질문을 정의합니다.
question: str = Field(min_length=1)
# 검색에 사용할 문서 개수를 정의합니다.
top_k: int = Field(default=4, ge=1, le=20)
# 파일 읽기 요청 모델을 정의합니다.
class FileReadRequest(BaseModel):
"""문서 파일 읽기 요청입니다."""
# docs 폴더 안에서 읽을 파일명을 정의합니다.
filename: str = Field(min_length=1)
# MySQL 데이터 등록 요청 모델을 정의합니다.
class KnowledgeCreateRequest(BaseModel):
"""MySQL 지식 데이터 등록 요청입니다."""
# 데이터 제목을 정의합니다.
title: str = Field(min_length=1, max_length=200)
# 데이터 본문을 정의합니다.
content: str = Field(min_length=1)
app.llm.embedding_service
from __future__ import, hashlib, numpy, openai,
app.config.settings 가져오기
EmbeddingService 임베딩 생성 클래스 정의
__init__() settings와 client 객체 초기화.
@property dimension() 임베딩 백엔드가 openai일 경우, 1536 차원을 return openai의 text-embedding-3-small 모델이 생성하는 임베딩의 벡텅 차원의 수.
아닐경우에는 설정에서 정의한 차원 반환
embed_documents() 여러 텍스트를 벡터로 변환. backend가 openai일 경우 AI API를 이용해 텍스트를 임베딩하는 메서드를 호출, 아닐경우 기본 api키가 필요없는 로컬 임베딩 생성.
embed_query() 매개변수 text를 embed_documents() 를 활용해 첫번째 결과 반환
_embed_openai() client가 없으면 RuntimeError() 있으면 client embeddings.create() 객체 만들기 return.for문으로 돌려 각 임베딩 배열만 추출해 반환한다.
_embed_local() np.zero() 배열을 생성하고 text.lower().split() 한뒤 tokens를 순회해 hashlib.sha256() 해시값을 생성, index를 반환하고 양수, 음수부호를 결정해 위치에 부호값을 누적한다.
norm 벡터 길이 계산
0으로 나누지않게 값이 있을때만 정규화. return tolist
"""
OpenAI 임베딩과 API 키가 필요 없는 로컬 임베딩을 제공합니다.
"""
# 미래 타입 힌트 평가 방식을 사용합니다.
from __future__ import annotations
# 해시 기반 로컬 임베딩을 만들기 위해 hashlib을 가져옵니다.
import hashlib
# 벡터 정규화와 배열 계산을 위해 NumPy를 가져옵니다.
import numpy as np
# OpenAI 임베딩 API를 호출하기 위해 OpenAI 클라이언트를 가져옵니다.
from openai import OpenAI
# 프로젝트 설정을 가져옵니다.
from app.config.settings import Settings
# 임베딩 생성 기능을 담당하는 클래스를 정의합니다.
class EmbeddingService:
"""설정에 따라 로컬 또는 OpenAI 임베딩을 생성합니다."""
# 서비스 생성 시 설정 객체를 전달받습니다.
def __init__(self, settings: Settings) -> None:
# 전달받은 설정을 인스턴스 변수에 저장합니다.
self.settings = settings
# OpenAI API 키가 설정된 경우에만 OpenAI 클라이언트를 생성합니다.
self.client = OpenAI(api_key=settings.openai_api_key) if settings.openai_api_key else None
# 현재 사용하는 임베딩 차원을 반환합니다.
@property
def dimension(self) -> int:
"""현재 임베딩 벡터의 차원을 반환합니다."""
# OpenAI text-embedding-3-small 모델의 기본 차원을 반환합니다.
if self.settings.embedding_backend.lower() == "openai":
return 1536
# 로컬 임베딩 설정에서 정의한 차원을 반환합니다.
return self.settings.local_embedding_dimension
# 여러 텍스트를 벡터로 변환합니다.
def embed_documents(self, texts: list[str]) -> list[list[float]]:
"""문서 목록을 임베딩 벡터 목록으로 변환합니다."""
# OpenAI 백엔드를 선택한 경우 OpenAI 임베딩을 생성합니다.
if self.settings.embedding_backend.lower() == "openai":
return self._embed_openai(texts)
# 기본값으로 API 키가 필요 없는 로컬 임베딩을 생성합니다.
return [self._embed_local(text) for text in texts]
# 검색 질문 하나를 벡터로 변환합니다.
def embed_query(self, text: str) -> list[float]:
"""검색 질문을 하나의 임베딩 벡터로 변환합니다."""
# 기존 문서 임베딩 함수를 재사용하여 첫 번째 결과를 반환합니다.
return self.embed_documents([text])[0]
# OpenAI 임베딩 API를 호출합니다.
def _embed_openai(self, texts: list[str]) -> list[list[float]]:
"""OpenAI API를 사용하여 실제 의미 기반 임베딩을 생성합니다."""
# OpenAI 클라이언트가 없으면 명확한 오류를 발생시킵니다.
if self.client is None:
raise RuntimeError("OPENAI_API_KEY가 설정되지 않았습니다.")
# OpenAI Embeddings API에 모델 이름과 텍스트 목록을 전달합니다.
response = self.client.embeddings.create(
model=self.settings.openai_embedding_model,
input=texts,
)
# API 응답에서 각 임베딩 배열만 추출하여 반환합니다.
return [item.embedding for item in response.data]
# 로컬에서 결정적인 해시 임베딩을 생성합니다.
def _embed_local(self, text: str) -> list[float]:
"""실습용 로컬 임베딩을 생성합니다."""
# 설정된 차원만큼 0으로 채운 NumPy 배열을 생성합니다.
vector = np.zeros(self.settings.local_embedding_dimension, dtype=np.float32)
# 공백을 기준으로 텍스트를 단어 토큰으로 나눕니다.
tokens = text.lower().split()
# 각 토큰을 순회하며 벡터의 특정 위치에 값을 누적합니다.
for token in tokens:
# 토큰의 SHA-256 해시값을 생성합니다.
digest = hashlib.sha256(token.encode("utf-8")).digest()
# 해시의 앞 4바이트를 정수로 바꾼 뒤 벡터 인덱스로 변환합니다.
index = int.from_bytes(digest[:4], "little") % len(vector)
# 해시의 다음 바이트를 이용해 양수 또는 음수 부호를 결정합니다.
sign = 1.0 if digest[4] % 2 == 0 else -1.0
# 계산한 위치에 부호 값을 누적합니다.
vector[index] += sign
# 벡터 길이를 계산합니다.
norm = float(np.linalg.norm(vector))
# 0으로 나누는 문제를 막기 위해 값이 있을 때만 정규화합니다.
if norm > 0.0:
vector = vector / norm
# NumPy 배열을 일반 Python 리스트로 변환하여 반환합니다.
return vector.tolist()
app.llm.openai_service
import __future__ annotation, openai, settings가져오기
OpenAIServices
settings, client만들기
answer() client가 없을때 대체응답 설정
client.responses.create() 호출
answer_with_context() 매개변수 prompt_template에 사용자 질문을 삽입하기, client가 없으면 대체 로컬 응답 반환하기, client.response.create() 한번 더 호출 , 최종 text 반환.
"""
OpenAI GPT를 호출하여 일반 질의응답과 RAG 답변을 생성합니다.
"""
# 미래 타입 힌트 평가 방식을 사용합니다.
from __future__ import annotations
# OpenAI Responses API 클라이언트를 가져옵니다.
from openai import OpenAI
# 프로젝트 설정 모델을 가져옵니다.
from app.config.settings import Settings
# OpenAI GPT 호출 기능을 담당하는 클래스를 정의합니다.
class OpenAIService:
"""OpenAI API 호출과 로컬 대체 응답을 제공합니다."""
# 설정 객체를 전달받아 서비스를 초기화합니다.
def __init__(self, settings: Settings) -> None:
# 전달받은 설정을 저장합니다.
self.settings = settings
# API 키가 있을 때만 OpenAI 클라이언트를 생성합니다.
self.client = OpenAI(api_key=settings.openai_api_key) if settings.openai_api_key else None
# 일반 질문에 대한 답변을 생성합니다.
def answer(self, question: str, system_prompt: str) -> str:
"""OpenAI GPT를 사용하여 일반 질문에 답합니다."""
# API 키가 없으면 프로젝트 구조를 확인할 수 있는 대체 응답을 반환합니다.
if self.client is None:
return (
"[로컬 데모 응답]\n"
"OPENAI_API_KEY가 설정되지 않아 실제 GPT 호출은 생략했습니다.\n"
f"질문: {question}"
)
# Responses API에 시스템 지시와 사용자 질문을 전달합니다.
response = self.client.responses.create(
model=self.settings.openai_chat_model,
instructions=system_prompt,
input=question,
)
# 편리한 output_text 속성에서 최종 답변 문자열을 반환합니다.
return response.output_text
# 검색 문맥을 근거로 RAG 답변을 생성합니다.
def answer_with_context(self, question: str, context: str, prompt_template: str) -> str:
"""검색된 문서를 근거로 답변을 생성합니다."""
# Prompt 템플릿에 검색 문맥과 사용자 질문을 삽입합니다.
final_prompt = prompt_template.format(context=context, question=question)
# API 키가 없으면 검색된 문맥을 확인할 수 있는 로컬 응답을 반환합니다.
if self.client is None:
return (
"[로컬 RAG 데모 응답]\n"
"검색은 정상 수행되었지만 OPENAI_API_KEY가 없어 GPT 생성은 생략했습니다.\n\n"
f"질문: {question}\n\n"
f"검색 문맥:\n{context}"
)
# OpenAI Responses API에 근거 제한 지시와 완성된 Prompt를 전달합니다.
response = self.client.responses.create(
model=self.settings.openai_chat_model,
instructions=(
"당신은 근거 중심 RAG Assistant입니다. "
"제공된 검색 문맥에 없는 사실은 추측하지 말고 모른다고 답하세요."
),
input=final_prompt,
)
# 생성된 최종 텍스트를 반환합니다.
return response.output_text
app.service.document_service
import __future__ annotations, path, app.config.settings 가져오기
DocumentService
init() setting과 docs_dir.mkdir 생성
list_files() 지원하는 문서형식 지정 sorted() supported만 된 path만 for문으로 정렬해 return.
load_chunks() chunk 저장할 목록 생성. 파일명을 순회해 경로를 만들어 문서를 읽고 _split_text() 청킹. for문으로 순회해서 cunk에 chunk+index, cunk_text , 출처를 붙어 return
_split_text() 빈 문자열이면 목록 반환, max() 다음 청크시작 step 계산.
chunk []리스트형 목록생성
for문으로 돌려 strip() 앞뒤 공백 제거, 비어있지않으면 chunk.append(), 마지막까지읽으면 break.
chunk 반환
"""
docs 폴더의 문서를 읽고 검색용 청크로 분할합니다.
"""
# 미래 타입 힌트 평가 방식을 사용합니다.
from __future__ import annotations
# 파일 경로 처리를 위해 Path를 가져옵니다.
from pathlib import Path
# 프로젝트 설정을 가져옵니다.
from app.config.settings import Settings
# 문서 파일 처리 서비스를 정의합니다.
class DocumentService:
"""텍스트 계열 문서를 읽고 중첩 청크로 분할합니다."""
# 설정 객체를 전달받습니다.
def __init__(self, settings: Settings) -> None:
# 설정을 저장합니다.
self.settings = settings
# 문서 디렉터리가 없으면 생성합니다.
self.settings.docs_dir.mkdir(parents=True, exist_ok=True)
# docs 폴더에 있는 파일 목록을 반환합니다.
def list_files(self) -> list[str]:
"""지원하는 문서 파일명을 반환합니다."""
# 지원 확장자를 정의합니다.
supported = {".txt", ".md"}
# 지원 확장자를 가진 파일만 정렬하여 반환합니다.
return [
path.name
for path in sorted(self.settings.docs_dir.iterdir())
if path.is_file() and path.suffix.lower() in supported
]
# 모든 문서를 읽어 청크 목록으로 변환합니다.
def load_chunks(self) -> list[dict]:
"""docs 폴더의 전체 문서를 검색용 청크로 반환합니다."""
# 모든 청크를 저장할 목록을 생성합니다.
chunks: list[dict] = []
# 지원 문서 파일명을 순회합니다.
for filename in self.list_files():
# 문서의 전체 경로를 만듭니다.
path = self.settings.docs_dir / filename
# UTF-8 인코딩으로 문서를 읽습니다.
text = path.read_text(encoding="utf-8")
# 문서를 중첩 청크로 분할합니다.
split_texts = self._split_text(text)
# 각 청크에 출처와 순번 메타데이터를 붙입니다.
for chunk_index, chunk_text in enumerate(split_texts):
chunks.append(
{
"content": chunk_text,
"source": filename,
"chunk_index": chunk_index,
}
)
# 전체 문서 청크를 반환합니다.
return chunks
# 긴 텍스트를 일정 크기의 중첩 청크로 분할합니다.
def _split_text(self, text: str) -> list[str]:
"""문자 수 기준으로 문서를 분할합니다."""
# 빈 문자열이면 빈 목록을 반환합니다.
if not text.strip():
return []
# 다음 청크 시작 위치의 이동 크기를 계산합니다.
step = max(1, self.settings.chunk_size - self.settings.chunk_overlap)
# 분할 결과를 저장할 목록을 생성합니다.
chunks: list[str] = []
# step 간격으로 문서 시작 위치를 이동합니다.
for start in range(0, len(text), step):
# 현재 청크의 끝 위치를 계산합니다.
end = start + self.settings.chunk_size
# 현재 범위의 문자열을 잘라 앞뒤 공백을 제거합니다.
chunk = text[start:end].strip()
# 비어 있지 않은 청크만 결과에 추가합니다.
if chunk:
chunks.append(chunk)
# 문서 마지막까지 읽었다면 반복을 종료합니다.
if end >= len(text):
break
# 분할된 청크 목록을 반환합니다.
return chunks
app.services.mysql_service
mysql.connector.connect() 리턴. 예재 테이블을 생성하는 sql try, commit, close()
connection.cursor() 커서를 생성해 select문 실행, return 값 저장 및 close
insert문으로 add_item하는 함수설정.
"""
MySQL 연결과 예제 지식 테이블 조회 기능을 제공합니다.
"""
# MySQL 서버에 연결하기 위해 mysql.connector를 가져옵니다.
import mysql.connector
# 설정 모델을 가져옵니다.
from app.config.settings import Settings
# MySQL 서비스 클래스를 정의합니다.
class MySQLService:
"""MySQL 연결 상태 확인과 안전한 예제 조회를 담당합니다."""
# 설정 객체를 전달받습니다.
def __init__(self, settings: Settings) -> None:
# 설정을 저장합니다.
self.settings = settings
# MySQL 연결 객체를 생성합니다.
def _connect(self):
"""환경설정에 지정된 MySQL 서버에 연결합니다."""
# MySQL 기능이 비활성화되어 있으면 명확한 오류를 발생시킵니다.
if not self.settings.mysql_enabled:
raise RuntimeError("MYSQL_ENABLED=false입니다. .env에서 MySQL 기능을 활성화하세요.")
# 환경설정 값으로 MySQL 연결을 생성하여 반환합니다.
return mysql.connector.connect(
host=self.settings.mysql_host,
port=self.settings.mysql_port,
database=self.settings.mysql_database,
user=self.settings.mysql_user,
password=self.settings.mysql_password,
)
# 예제 테이블을 생성합니다.
def initialize(self) -> dict:
"""지식 저장용 knowledge_items 테이블을 생성합니다."""
# MySQL 연결을 엽니다.
connection = self._connect()
# SQL 실행 후에도 연결이 닫히도록 try/finally를 사용합니다.
try:
# 커서를 생성합니다.
cursor = connection.cursor()
# 테이블이 없을 때만 생성하는 SQL을 실행합니다.
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS knowledge_items (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
title VARCHAR(200) NOT NULL,
content TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# 테이블 생성 작업을 확정합니다.
connection.commit()
# 성공 결과를 반환합니다.
return {"message": "knowledge_items 테이블 준비 완료"}
finally:
# 열린 MySQL 연결을 항상 닫습니다.
connection.close()
# 모든 예제 지식 데이터를 조회합니다.
def list_items(self) -> list[dict]:
"""knowledge_items 테이블의 데이터를 최신 순으로 조회합니다."""
# MySQL 연결을 엽니다.
connection = self._connect()
# 조회 후에도 연결이 닫히도록 try/finally를 사용합니다.
try:
# 조회 결과를 딕셔너리로 받는 커서를 생성합니다.
cursor = connection.cursor(dictionary=True)
# 전체 데이터를 최신 ID 순으로 조회합니다.
cursor.execute(
"SELECT id, title, content, created_at "
"FROM knowledge_items ORDER BY id DESC"
)
# 모든 행을 가져와 반환합니다.
return cursor.fetchall()
finally:
# 열린 MySQL 연결을 항상 닫습니다.
connection.close()
# 예제 지식 데이터를 추가합니다.
def add_item(self, title: str, content: str) -> dict:
"""제목과 본문을 MySQL에 안전하게 저장합니다."""
# MySQL 연결을 엽니다.
connection = self._connect()
# 저장 후에도 연결이 닫히도록 try/finally를 사용합니다.
try:
# 커서를 생성합니다.
cursor = connection.cursor()
# 파라미터 바인딩 방식으로 SQL Injection 위험을 낮춥니다.
cursor.execute(
"INSERT INTO knowledge_items(title, content) VALUES (%s, %s)",
(title, content),
)
# INSERT 작업을 확정합니다.
connection.commit()
# 생성된 기본키 값을 반환합니다.
return {"id": cursor.lastrowid, "title": title, "content": content}
finally:
# 열린 MySQL 연결을 항상 닫습니다.
connection.close()
app.services.prompt_service
prompt기본값 설정, prompt 목록반환, prompt 템플릿 조회 등록되지않은 prompt면 안내
"""
애플리케이션에서 재사용할 Prompt 템플릿을 관리합니다.
"""
# Prompt 템플릿 목록을 상수 딕셔너리로 정의합니다.
PROMPTS: dict[str, str] = {
"rag_answer": (
"아래 검색 문맥만 사용하여 질문에 답하세요.\n"
"답변 마지막에는 사용한 출처 파일명을 표시하세요.\n"
"문맥에 답이 없다면 '제공된 문서에서 확인할 수 없습니다.'라고 답하세요.\n\n"
"[검색 문맥]\n{context}\n\n"
"[질문]\n{question}"
),
"general_assistant": (
"당신은 FastAPI, OpenAI, MCP 및 RAG를 설명하는 기술 교육 도우미입니다. "
"한국어로 명확하고 실용적으로 답하세요."
),
}
# Prompt 관리 서비스를 정의합니다.
class PromptService:
"""이름으로 Prompt를 조회하고 목록을 제공합니다."""
# 전체 Prompt 이름 목록을 반환합니다.
def list_prompts(self) -> list[str]:
"""등록된 Prompt 이름을 반환합니다."""
# PROMPTS 딕셔너리의 키를 리스트로 변환합니다.
return list(PROMPTS.keys())
# 이름으로 Prompt 템플릿을 조회합니다.
def get_prompt(self, name: str) -> str:
"""지정한 Prompt 템플릿을 반환합니다."""
# 등록되지 않은 이름이면 KeyError 대신 이해하기 쉬운 오류를 발생시킵니다.
if name not in PROMPTS:
raise ValueError(f"등록되지 않은 Prompt입니다: {name}")
# 요청한 Prompt 문자열을 반환합니다.
return PROMPTS[name]
app.services.rag_service
문서, 임베딩, 벡터, openai , prompt 서비스 저장
검색용 청크를 읽어 본문 content만 추출해 임베딩 벡터로 변환, vector_store/rebuild() 벡터 저장소 전체 재구축 리턴.
질문 임베딩 벡터로 변환해 벡터 저장소에서 search.한것 return.
질문과 유사한 문서를 검색해 if result가 없으면 안내문구, 있으면 join해 return. rag_answer로도 조회, 답변을 생성해 return. 이때 중복되지않은 출처 파일명 source,을 저장해 함꼐 return.
"""
문서 인덱싱, Vector Search, RAG 답변 생성을 통합합니다.
"""
# 미래 타입 힌트 평가 방식을 사용합니다.
from __future__ import annotations
# 임베딩 서비스를 가져옵니다.
from app.llm.embedding_service import EmbeddingService
# OpenAI 답변 생성 서비스를 가져옵니다.
from app.llm.openai_service import OpenAIService
# 문서 서비스를 가져옵니다.
from app.services.document_service import DocumentService
# Prompt 서비스를 가져옵니다.
from app.services.prompt_service import PromptService
# 벡터 저장소 공통 인터페이스를 가져옵니다.
from app.vectordb.base import VectorStore
# RAG 서비스 클래스를 정의합니다.
class RagService:
"""문서 적재, 검색, 답변 생성을 한 곳에서 처리합니다."""
# 필요한 하위 서비스를 생성자에서 전달받습니다.
def __init__(
self,
document_service: DocumentService,
embedding_service: EmbeddingService,
vector_store: VectorStore,
openai_service: OpenAIService,
prompt_service: PromptService,
) -> None:
# 문서 서비스를 저장합니다.
self.document_service = document_service
# 임베딩 서비스를 저장합니다.
self.embedding_service = embedding_service
# 벡터 저장소를 저장합니다.
self.vector_store = vector_store
# OpenAI 서비스를 저장합니다.
self.openai_service = openai_service
# Prompt 서비스를 저장합니다.
self.prompt_service = prompt_service
# docs 폴더 전체를 다시 인덱싱합니다.
def rebuild_index(self) -> dict:
"""전체 문서를 청크로 분할하고 벡터 저장소를 재구축합니다."""
# docs 폴더에서 검색용 청크를 읽습니다.
documents = self.document_service.load_chunks()
# 각 청크의 본문만 추출합니다.
texts = [document["content"] for document in documents]
# 전체 문서 청크를 임베딩 벡터로 변환합니다.
vectors = self.embedding_service.embed_documents(texts) if texts else []
# 선택한 벡터 저장소를 전체 재구축합니다.
count = self.vector_store.rebuild(documents, vectors)
# 처리 결과를 API 응답용 딕셔너리로 반환합니다.
return {"indexed_chunks": count}
# 질문과 유사한 문서를 검색합니다.
def search(self, query: str, top_k: int) -> list[dict]:
"""Vector Search 결과를 반환합니다."""
# 질문을 임베딩 벡터로 변환합니다.
query_vector = self.embedding_service.embed_query(query)
# 벡터 저장소에서 유사 문서를 검색합니다.
return self.vector_store.search(query_vector, top_k)
# RAG 답변과 출처를 생성합니다.
def ask(self, question: str, top_k: int) -> dict:
"""검색 문맥을 근거로 답변을 생성합니다."""
# 질문과 유사한 문서를 검색합니다.
results = self.search(question, top_k)
# 검색 결과가 없으면 먼저 인덱스를 만들라는 메시지를 반환합니다.
if not results:
return {
"answer": "검색 결과가 없습니다. 먼저 /api/rag/rebuild를 실행하세요.",
"sources": [],
"matches": [],
}
# 각 검색 결과를 출처와 본문이 포함된 문맥 문자열로 변환합니다.
context = "\n\n".join(
f"[출처: {item.get('source', 'unknown')}]\n{item.get('content', '')}"
for item in results
)
# RAG Prompt 템플릿을 조회합니다.
prompt_template = self.prompt_service.get_prompt("rag_answer")
# OpenAI 또는 로컬 대체 방식으로 답변을 생성합니다.
answer = self.openai_service.answer_with_context(
question=question,
context=context,
prompt_template=prompt_template,
)
# 중복되지 않은 출처 파일명을 순서대로 구성합니다.
sources = list(dict.fromkeys(item.get("source", "unknown") for item in results))
# 답변, 출처, 검색 상세 결과를 반환합니다.
return {"answer": answer, "sources": sources, "matches": results}
app.vectordb.faiss_store
AISS는 "문서를 저장하는 데이터베이스"가 아니라 "벡터를 빠르게 검색하는 엔진"
클래스하나가 faiss 저장, 검색만 담당한다 .
FAISS(Facebook AI Similarity Search) 는 Meta(Facebook AI Research)가 만든 벡터(Vector) 검색 라이브러리
임베딩 차원저장, 디렉터리생성 및 정으ㅣ, 빈 인덱스를 생성해 faiss.IndexFlatIP() 인덱스 생성. 파일로 저장, 문서 내용과 출처 메타데이터를 json파일로 저장 return.
search() faiss 인덱스를 읽어 json읽기, np.asarray() 2차원 배열로 변환해 normalize() 질문벡터 정규화하기, min 문서수에 맞춰 실제 검색 개수 결정 검색된 인덱스와 점수를 함께 순환해 position이 0보다 작은 유효하지않은 인덱스는 건너뛰고 해당 문서 를 복사해 score를 float값으로 추가해 return.
"""
FAISS 기반 로컬 벡터 저장소를 구현합니다.
"""
# JSON 메타데이터를 저장하고 읽기 위해 json을 가져옵니다.
import json
# 파일 경로를 처리하기 위해 Path를 가져옵니다.
from pathlib import Path
# FAISS 인덱스를 생성하고 검색하기 위해 faiss를 가져옵니다.
import faiss
# 벡터 배열 계산을 위해 NumPy를 가져옵니다.
import numpy as np
# 공통 벡터 저장소 인터페이스를 가져옵니다.
from app.vectordb.base import SearchResult, VectorStore
# FAISS 저장소 클래스를 정의합니다.
class FaissVectorStore(VectorStore):
"""FAISS 인덱스와 JSON 메타데이터를 디스크에 저장합니다."""
# 저장 디렉터리와 임베딩 차원을 전달받습니다.
def __init__(self, directory: Path, dimension: int) -> None:
# 저장 디렉터리를 저장합니다.
self.directory = directory
# 임베딩 차원을 저장합니다.
self.dimension = dimension
# 저장 디렉터리가 없으면 생성합니다.
self.directory.mkdir(parents=True, exist_ok=True)
# FAISS 인덱스 파일 경로를 정의합니다.
self.index_path = self.directory / "documents.faiss"
# 문서 메타데이터 JSON 파일 경로를 정의합니다.
self.meta_path = self.directory / "documents.json"
# 인덱스를 전체 재구축합니다.
def rebuild(self, documents: list[dict], vectors: list[list[float]]) -> int:
"""문서와 벡터를 이용해 FAISS 인덱스를 새로 만듭니다."""
# 벡터가 없으면 빈 인덱스를 생성합니다.
matrix = np.asarray(vectors, dtype=np.float32)
# 코사인 유사도 검색을 위해 각 벡터를 L2 정규화합니다.
if len(matrix) > 0:
faiss.normalize_L2(matrix)
# 내적 기반의 단순하고 정확한 Flat 인덱스를 생성합니다.
index = faiss.IndexFlatIP(self.dimension)
# 벡터가 존재하는 경우에만 인덱스에 추가합니다.
if len(matrix) > 0:
index.add(matrix)
# 완성된 FAISS 인덱스를 파일로 저장합니다.
faiss.write_index(index, str(self.index_path))
# 문서 내용과 출처 메타데이터를 JSON 파일로 저장합니다.
self.meta_path.write_text(
json.dumps(documents, ensure_ascii=False, indent=2),
encoding="utf-8",
)
# 저장한 문서 수를 반환합니다.
return len(documents)
# 질문 벡터와 유사한 문서를 검색합니다.
def search(self, query_vector: list[float], top_k: int) -> list[SearchResult]:
"""FAISS 인덱스에서 유사 문서를 검색합니다."""
# 인덱스 또는 메타데이터 파일이 없으면 빈 결과를 반환합니다.
if not self.index_path.exists() or not self.meta_path.exists():
return []
# 디스크에서 FAISS 인덱스를 읽습니다.
index = faiss.read_index(str(self.index_path))
# 디스크에서 문서 메타데이터 목록을 읽습니다.
documents = json.loads(self.meta_path.read_text(encoding="utf-8"))
# 질문 벡터를 2차원 float32 배열로 변환합니다.
query = np.asarray([query_vector], dtype=np.float32)
# 코사인 유사도에 맞게 질문 벡터를 정규화합니다.
faiss.normalize_L2(query)
# 저장된 문서 수를 넘지 않도록 실제 검색 개수를 결정합니다.
limit = min(top_k, index.ntotal)
# 인덱스가 비어 있으면 빈 목록을 반환합니다.
if limit == 0:
return []
# FAISS에서 점수와 문서 위치를 검색합니다.
scores, indices = index.search(query, limit)
# 최종 검색 결과를 저장할 목록을 생성합니다.
results: list[SearchResult] = []
# 검색된 인덱스와 점수를 함께 순회합니다.
for position, score in zip(indices[0], scores[0]):
# 유효하지 않은 인덱스는 건너뜁니다.
if position < 0:
continue
# 원본 문서 메타데이터를 복사합니다.
item = dict(documents[position])
# 검색 점수를 float 값으로 추가합니다.
item["score"] = float(score)
# 완성한 항목을 결과 목록에 추가합니다.
results.append(item)
# 유사도 순으로 정렬된 결과를 반환합니다.
return results
app.vectordb.qdrant_store
uuid 고유id만들기,
qdrant_client와 model가져오기 벡터 데이터베이스 서버
app config setting와 bectordb가져오기
로컬 모드일때는 내부 디렉터리 사용하는 client 생성, 아닐경우 서버주소사용하기
rebuild() 기존 컬렉션이 존재하면삭제하고 collection 생성하기 이때 코사인 거리와 임베딩 차원을사용함. size=self.dimension, distance=Distance.COSINE) 문서와 벡터를 Qdrant에 저장할 ㅅ ㅜ있는 형태로 변환하는 코드 Pointstruct()
point가 있을때 client.upsert() qdrant에 저장
컬렉션이 없으면 빈 결과 반환, client.query_points로 유사벡터 검색, 결과목록생성, for문을 순회하며 score에 float형 유사도 점수 추가. append해 return
"""
Qdrant local 또는 server 모드 벡터 저장소를 구현합니다.
"""
# UUID 형태의 고유 문서 ID를 만들기 위해 uuid4를 가져옵니다.
from uuid import uuid4
# Qdrant 클라이언트를 가져옵니다.
from qdrant_client import QdrantClient
# Qdrant 컬렉션과 Point 구조를 정의하는 모델을 가져옵니다.
from qdrant_client.models import Distance, PointStruct, VectorParams
# 설정 모델을 가져옵니다.
from app.config.settings import Settings
# 공통 벡터 저장소 인터페이스를 가져옵니다.
from app.vectordb.base import SearchResult, VectorStore
# Qdrant 저장소 클래스를 정의합니다.
class QdrantVectorStore(VectorStore):
"""Qdrant에 문서 벡터와 payload를 저장합니다."""
# 설정과 벡터 차원을 전달받습니다.
def __init__(self, settings: Settings, dimension: int) -> None:
# 설정을 저장합니다.
self.settings = settings
# 벡터 차원을 저장합니다.
self.dimension = dimension
# 사용할 컬렉션 이름을 저장합니다.
self.collection_name = settings.qdrant_collection
# local 모드이면 프로젝트 내부 디렉터리를 사용하는 클라이언트를 생성합니다.
if settings.qdrant_mode.lower() == "local":
self.client = QdrantClient(path=str(settings.qdrant_dir))
else:
# server 모드이면 외부 Qdrant 서버 주소를 사용합니다.
self.client = QdrantClient(url=settings.qdrant_url)
# 컬렉션 전체를 새로 구성합니다.
def rebuild(self, documents: list[dict], vectors: list[list[float]]) -> int:
"""기존 컬렉션을 교체하고 문서를 저장합니다."""
# 기존 컬렉션이 존재하면 삭제합니다.
if self.client.collection_exists(self.collection_name):
self.client.delete_collection(self.collection_name)
# 코사인 거리와 임베딩 차원을 사용하여 컬렉션을 생성합니다.
self.client.create_collection(
collection_name=self.collection_name,
vectors_config=VectorParams(size=self.dimension, distance=Distance.COSINE),
)
# 문서와 벡터를 Qdrant Point 목록으로 변환합니다.
points = [
PointStruct(id=str(uuid4()), vector=vector, payload=document)
for document, vector in zip(documents, vectors)
]
# Point가 있을 때만 Qdrant에 저장합니다.
if points:
self.client.upsert(collection_name=self.collection_name, points=points)
# 저장한 문서 수를 반환합니다.
return len(points)
# 유사 문서를 검색합니다.
def search(self, query_vector: list[float], top_k: int) -> list[SearchResult]:
"""Qdrant에서 유사 문서를 검색합니다."""
# 컬렉션이 없으면 빈 결과를 반환합니다.
if not self.client.collection_exists(self.collection_name):
return []
# 최신 query_points API로 유사 벡터를 검색합니다.
response = self.client.query_points(
collection_name=self.collection_name,
query=query_vector,
limit=top_k,
with_payload=True,
)
# 최종 결과 목록을 생성합니다.
results: list[SearchResult] = []
# 검색된 Point를 순회합니다.
for point in response.points:
# payload가 없을 수 있으므로 빈 딕셔너리를 기본값으로 복사합니다.
item = dict(point.payload or {})
# Qdrant 유사도 점수를 추가합니다.
item["score"] = float(point.score)
# 완성한 결과를 목록에 추가합니다.
results.append(item)
# 검색 결과를 반환합니다.
return results
container.py
import
functools의 lru_cache 반복호출시 같은 container를 재사용하기 위한 lru_cache
app.config.settings 가져오기
app.llmembedding_service, app.llm.openai_service, app.service.document_service, app.services.mysql_service, app.services.prompt_service, app.services.rag_servic, app.vectordb.faiss_store, app.vectordb.qdrant_store 가져오기
각 서비스 생성, @lru_cache container 재사용할수있도록 설정
"""
서비스 객체 생성과 의존성 조립을 한 곳에서 담당합니다.
"""
# 반복 호출 시 같은 Container를 재사용하기 위해 lru_cache를 가져옵니다.
from functools import lru_cache
# 프로젝트 설정을 가져옵니다.
from app.config.settings import get_settings
# 임베딩 서비스를 가져옵니다.
from app.llm.embedding_service import EmbeddingService
# OpenAI 서비스를 가져옵니다.
from app.llm.openai_service import OpenAIService
# 문서 서비스를 가져옵니다.
from app.services.document_service import DocumentService
# MySQL 서비스를 가져옵니다.
from app.services.mysql_service import MySQLService
# Prompt 서비스를 가져옵니다.
from app.services.prompt_service import PromptService
# RAG 서비스를 가져옵니다.
from app.services.rag_service import RagService
# FAISS 저장소를 가져옵니다.
from app.vectordb.faiss_store import FaissVectorStore
# Qdrant 저장소를 가져옵니다.
from app.vectordb.qdrant_store import QdrantVectorStore
# 애플리케이션에서 공유하는 서비스 묶음을 정의합니다.
class Container:
"""환경설정에 따라 필요한 구현을 생성하고 서로 연결합니다."""
# 모든 서비스 객체를 생성합니다.
def __init__(self) -> None:
# 캐시된 설정 객체를 가져옵니다.
self.settings = get_settings()
# 문서 서비스를 생성합니다.
self.document_service = DocumentService(self.settings)
# 임베딩 서비스를 생성합니다.
self.embedding_service = EmbeddingService(self.settings)
# 설정에 따라 Qdrant 또는 FAISS 저장소를 생성합니다.
if self.settings.vector_backend.lower() == "qdrant":
self.vector_store = QdrantVectorStore(
self.settings,
self.embedding_service.dimension,
)
else:
self.vector_store = FaissVectorStore(
self.settings.faiss_dir,
self.embedding_service.dimension,
)
# OpenAI 서비스를 생성합니다.
self.openai_service = OpenAIService(self.settings)
# Prompt 서비스를 생성합니다.
self.prompt_service = PromptService()
# MySQL 서비스를 생성합니다.
self.mysql_service = MySQLService(self.settings)
# 앞서 생성한 서비스를 연결하여 RAG 서비스를 생성합니다.
self.rag_service = RagService(
document_service=self.document_service,
embedding_service=self.embedding_service,
vector_store=self.vector_store,
openai_service=self.openai_service,
prompt_service=self.prompt_service,
)
# Container를 한 번만 생성하여 재사용합니다.
@lru_cache
def get_container() -> Container:
"""애플리케이션 공용 Container를 반환합니다."""
# 새 Container 객체를 생성하여 캐시에 저장하고 반환합니다.
return Container()
file_tools.py
허용된 루트 폴더안의 경로인지 검사한다.
파일목록 반환텍스트 파일읽기
"""
프로젝트 docs 폴더에 한정된 안전한 파일 시스템 Tool을 제공합니다.
"""
# 파일 경로 처리를 위해 Path를 가져옵니다.
from pathlib import Path
# 허용된 루트 폴더 안의 경로인지 검사합니다.
def _safe_path(base_dir: Path, filename: str) -> Path:
"""경로 탈출 공격을 방지하면서 파일 경로를 반환합니다."""
# 기준 폴더의 절대 경로를 계산합니다.
base = base_dir.resolve()
# 사용자 파일명을 기준 폴더와 결합한 뒤 절대 경로로 변환합니다.
target = (base / filename).resolve()
# 대상 경로가 기준 폴더 내부가 아니면 접근을 거부합니다.
if base not in target.parents and target != base:
raise ValueError("docs 폴더 밖의 경로에는 접근할 수 없습니다.")
# 검증이 끝난 경로를 반환합니다.
return target
# docs 폴더의 파일 목록을 반환합니다.
def list_doc_files(base_dir: Path) -> list[str]:
"""docs 폴더의 파일명을 반환합니다."""
# 폴더가 없으면 생성합니다.
base_dir.mkdir(parents=True, exist_ok=True)
# 실제 파일만 정렬하여 파일명 목록으로 반환합니다.
return [path.name for path in sorted(base_dir.iterdir()) if path.is_file()]
# docs 폴더의 텍스트 파일을 읽습니다.
def read_doc_file(base_dir: Path, filename: str) -> str:
"""검증된 문서 파일의 내용을 UTF-8로 읽습니다."""
# 안전한 파일 경로를 계산합니다.
target = _safe_path(base_dir, filename)
# 파일이 없으면 오류를 발생시킵니다.
if not target.exists() or not target.is_file():
raise FileNotFoundError(f"파일을 찾을 수 없습니다: {filename}")
# UTF-8 인코딩으로 파일 전체를 반환합니다.
return target.read_text(encoding="utf-8")
api.py
import fastapi, apirouter, httpexception url을 묶어주는라우터와 에러
app.routers.schemes에서 필요한 함수가져오기
app.services.container에서 함수가져오기
app.tools.file_tools에서 함수가져오기
@router.get("/health") 서버와 설정 상태 확인
@router.post("/chat") 질문정의 공통 container를 가져와 assistant prompt를 조회해 gpt답변을 생성해 return.
@router.post("/rag/rebuild") container에서 rag_service, revuild_index 재구축하기
@router.post("/rag/search") 유사문서 검색하기 결과목록 반환
@router.post("/rag/ask") rag 답변 생성해 반환하기
@router.get("/prompts") 프롬프트 목록 반환
@router.get("/prompts/{name}")요청한 프롬프트 반환
@router.get("/files") 파일목록 반환
@router.post("/files/read") 파일을 읽어 반환
@router.post("/mysql/init") 예제 테이블 포기화,
@router.get("/mysql/items") 목록 아이템 조회
@router.post("/mysql/items") 요청 데이터 저장하고 결과반환
"""
프로젝트의 FastAPI REST API 엔드포인트를 정의합니다.
"""
# FastAPI Router와 HTTPException을 가져옵니다.
from fastapi import APIRouter, HTTPException
# 요청 데이터 모델을 가져옵니다.
from app.routers.schemas import (
ChatRequest,
FileReadRequest,
KnowledgeCreateRequest,
RagRequest,
SearchRequest,
)
# 서비스 Container를 가져옵니다.
from app.services.container import get_container
# 파일 Tool 함수를 가져옵니다.
from app.tools.file_tools import list_doc_files, read_doc_file
# /api 접두어를 사용하는 Router를 생성합니다.
router = APIRouter(prefix="/api", tags=["RAG Assistant"])
# 서버와 설정 상태를 확인하는 API를 정의합니다.
@router.get("/health")
def health() -> dict:
"""현재 앱과 선택된 백엔드 상태를 반환합니다."""
# 공용 Container를 가져옵니다.
container = get_container()
# API 키와 비밀번호는 노출하지 않고 활성화 상태만 반환합니다.
return {
"status": "ok",
"app_name": container.settings.app_name,
"embedding_backend": container.settings.embedding_backend,
"vector_backend": container.settings.vector_backend,
"openai_configured": bool(container.settings.openai_api_key),
"mysql_enabled": container.settings.mysql_enabled,
}
# 일반 GPT 질문 API를 정의합니다.
@router.post("/chat")
def chat(request: ChatRequest) -> dict:
"""OpenAI GPT 또는 로컬 대체 응답을 반환합니다."""
# 공용 Container를 가져옵니다.
container = get_container()
# 일반 Assistant Prompt를 조회합니다.
system_prompt = container.prompt_service.get_prompt("general_assistant")
# GPT 답변을 생성합니다.
answer = container.openai_service.answer(request.question, system_prompt)
# JSON 응답 형태로 반환합니다.
return {"answer": answer}
# 문서 인덱스 전체 재구축 API를 정의합니다.
@router.post("/rag/rebuild")
def rebuild_rag_index() -> dict:
"""docs 폴더 전체를 벡터 인덱스로 재구축합니다."""
# 오류를 HTTP 500 응답으로 변환하기 위해 try 블록을 사용합니다.
try:
# RAG 인덱스를 재구축하고 결과를 반환합니다.
return get_container().rag_service.rebuild_index()
except Exception as exc:
# 실제 오류 내용을 포함하는 HTTP 예외를 발생시킵니다.
raise HTTPException(status_code=500, detail=str(exc)) from exc
# Vector Search API를 정의합니다.
@router.post("/rag/search")
def search_documents(request: SearchRequest) -> dict:
"""질문과 유사한 문서 청크를 반환합니다."""
# 오류를 HTTP 응답으로 변환하기 위해 try 블록을 사용합니다.
try:
# 유사 문서를 검색합니다.
results = get_container().rag_service.search(request.query, request.top_k)
# 검색 결과 수와 결과 목록을 반환합니다.
return {"count": len(results), "results": results}
except Exception as exc:
# 검색 오류를 HTTP 500으로 반환합니다.
raise HTTPException(status_code=500, detail=str(exc)) from exc
# RAG 질의응답 API를 정의합니다.
@router.post("/rag/ask")
def ask_rag(request: RagRequest) -> dict:
"""검색된 문서를 근거로 GPT 답변을 생성합니다."""
# 오류를 HTTP 응답으로 변환하기 위해 try 블록을 사용합니다.
try:
# RAG 답변을 생성하여 반환합니다.
return get_container().rag_service.ask(request.question, request.top_k)
except Exception as exc:
# RAG 처리 오류를 HTTP 500으로 반환합니다.
raise HTTPException(status_code=500, detail=str(exc)) from exc
# 등록된 Prompt 목록 API를 정의합니다.
@router.get("/prompts")
def list_prompts() -> dict:
"""사용 가능한 Prompt 이름을 반환합니다."""
# Prompt 이름 목록을 반환합니다.
return {"prompts": get_container().prompt_service.list_prompts()}
# 개별 Prompt 조회 API를 정의합니다.
@router.get("/prompts/{name}")
def get_prompt(name: str) -> dict:
"""이름으로 Prompt 템플릿을 반환합니다."""
# 존재하지 않는 Prompt 오류를 처리하기 위해 try 블록을 사용합니다.
try:
# 요청한 Prompt를 조회하여 반환합니다.
return {"name": name, "template": get_container().prompt_service.get_prompt(name)}
except ValueError as exc:
# 존재하지 않는 Prompt는 HTTP 404로 반환합니다.
raise HTTPException(status_code=404, detail=str(exc)) from exc
# docs 폴더의 파일 목록 API를 정의합니다.
@router.get("/files")
def list_files() -> dict:
"""MCP 파일 Tool과 동일한 문서 목록 기능을 REST로 제공합니다."""
# 설정에서 docs 경로를 가져옵니다.
docs_dir = get_container().settings.docs_dir
# 파일 목록을 반환합니다.
return {"files": list_doc_files(docs_dir)}
# docs 폴더 파일 읽기 API를 정의합니다.
@router.post("/files/read")
def read_file(request: FileReadRequest) -> dict:
"""MCP 파일 Tool과 동일한 읽기 기능을 REST로 제공합니다."""
# 파일 오류를 HTTP 응답으로 변환하기 위해 try 블록을 사용합니다.
try:
# 지정한 문서를 읽어 반환합니다.
content = read_doc_file(get_container().settings.docs_dir, request.filename)
# 파일명과 내용을 반환합니다.
return {"filename": request.filename, "content": content}
except (ValueError, FileNotFoundError) as exc:
# 잘못된 경로나 없는 파일은 HTTP 400으로 반환합니다.
raise HTTPException(status_code=400, detail=str(exc)) from exc
# MySQL 예제 테이블 초기화 API를 정의합니다.
@router.post("/mysql/init")
def initialize_mysql() -> dict:
"""MySQL 지식 테이블을 생성합니다."""
# MySQL 오류를 HTTP 응답으로 변환합니다.
try:
# 테이블 초기화 결과를 반환합니다.
return get_container().mysql_service.initialize()
except Exception as exc:
# 연결 또는 SQL 오류를 HTTP 500으로 반환합니다.
raise HTTPException(status_code=500, detail=str(exc)) from exc
# MySQL 지식 목록 API를 정의합니다.
@router.get("/mysql/items")
def list_mysql_items() -> dict:
"""MySQL에 저장된 지식 데이터를 조회합니다."""
# MySQL 오류를 HTTP 응답으로 변환합니다.
try:
# 데이터를 조회합니다.
items = get_container().mysql_service.list_items()
# 결과 수와 데이터 목록을 반환합니다.
return {"count": len(items), "items": items}
except Exception as exc:
# 연결 또는 SQL 오류를 HTTP 500으로 반환합니다.
raise HTTPException(status_code=500, detail=str(exc)) from exc
# MySQL 지식 등록 API를 정의합니다.
@router.post("/mysql/items")
def create_mysql_item(request: KnowledgeCreateRequest) -> dict:
"""MySQL에 새 지식 데이터를 저장합니다."""
# MySQL 오류를 HTTP 응답으로 변환합니다.
try:
# 요청 데이터를 저장하고 결과를 반환합니다.
return get_container().mysql_service.add_item(request.title, request.content)
except Exception as exc:
# 연결 또는 SQL 오류를 HTTP 500으로 반환합니다.
raise HTTPException(status_code=500, detail=str(exc)) from exc
settings.py
import __future__ annotation 타입힌트
functools lru_cache 반복사용한 객체를 캐시하기 위한 lru_cache
pathlib,
pydantic_settings에서의 basesettings, settingsconfigdict env 파일을 읽는다.
PROJECT_ROOT 프로젝트 루트 설정
Settings() env파일을 읽어 변수설정 SettingsConfigDict() env 파일을 읽어 utf-8로 읽는다.
extra="ignore" 모델에 정의되지않는 추가 필드를 ignore한다 .
@lru_cache 설정 객체를 한번만 생성해 재사용하도록 캐시한다. Settings()
"""
프로젝트 전체에서 사용하는 환경설정을 정의합니다.
"""
# 미래 버전의 타입 힌트 평가 방식을 사용합니다.
from __future__ import annotations
# 반복 사용한 설정 객체를 캐시하기 위해 lru_cache를 가져옵니다.
from functools import lru_cache
# 운영체제와 무관하게 파일 경로를 처리하기 위해 Path를 가져옵니다.
from pathlib import Path
# .env 파일과 환경변수를 읽는 BaseSettings를 가져옵니다.
from pydantic_settings import BaseSettings, SettingsConfigDict
# 현재 파일을 기준으로 프로젝트 루트 경로를 계산합니다.
PROJECT_ROOT = Path(__file__).resolve().parents[2]
# 프로젝트 환경설정 모델을 정의합니다.
class Settings(BaseSettings):
"""환경변수와 .env 파일의 값을 타입 안전하게 관리합니다."""
# 애플리케이션 화면과 문서에 표시할 이름을 정의합니다.
app_name: str = "FastAPI + OpenAI + MCP 기반 RAG Assistant"
# FastAPI 서버가 사용할 호스트 주소를 정의합니다.
app_host: str = "127.0.0.1"
# FastAPI 서버가 사용할 포트 번호를 정의합니다.
app_port: int = 8000
# OpenAI API 키를 저장합니다.
openai_api_key: str = ""
# OpenAI 질의응답에 사용할 모델 이름을 정의합니다.
openai_chat_model: str = "gpt-4.1-mini"
# OpenAI 임베딩에 사용할 모델 이름을 정의합니다.
openai_embedding_model: str = "text-embedding-3-small"
# 임베딩 구현을 local 또는 openai 중에서 선택합니다.
embedding_backend: str = "local"
# 벡터 저장소 구현을 faiss 또는 qdrant 중에서 선택합니다.
vector_backend: str = "faiss"
# Qdrant를 local 또는 server 방식으로 실행할지 정의합니다.
qdrant_mode: str = "local"
# 외부 Qdrant 서버 주소를 정의합니다.
qdrant_url: str = "http://127.0.0.1:6333"
# Qdrant 컬렉션 이름을 정의합니다.
qdrant_collection: str = "mcp_rag_documents"
# MySQL 서버 주소를 정의합니다.
mysql_host: str = "127.0.0.1"
# MySQL 서버 포트를 정의합니다.
mysql_port: int = 3306
# MySQL 데이터베이스 이름을 정의합니다.
mysql_database: str = "mcp_rag_db"
# MySQL 사용자 이름을 정의합니다.
mysql_user: str = "mcp_user"
# MySQL 비밀번호를 정의합니다.
mysql_password: str = "1234"
# MySQL 기능을 활성화할지 정의합니다.
mysql_enabled: bool = False
# RAG 검색에서 반환할 기본 문서 수를 정의합니다.
rag_top_k: int = 4
# 문서를 분할할 때 사용할 최대 문자 수를 정의합니다.
chunk_size: int = 700
# 문서 청크 사이에서 겹칠 문자 수를 정의합니다.
chunk_overlap: int = 100
# 로컬 임베딩 벡터의 차원을 정의합니다.
local_embedding_dimension: int = 384
# 원본 학습 문서가 저장되는 디렉터리를 정의합니다.
docs_dir: Path = PROJECT_ROOT / "docs"
# FAISS 인덱스가 저장되는 디렉터리를 정의합니다.
faiss_dir: Path = PROJECT_ROOT / "data" / "faiss"
# Qdrant local 데이터가 저장되는 디렉터리를 정의합니다.
qdrant_dir: Path = PROJECT_ROOT / "data" / "qdrant"
# .env 파일을 읽고 환경변수 이름의 대소문자를 구분하지 않도록 설정합니다.
model_config = SettingsConfigDict(
env_file=PROJECT_ROOT / ".env",
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore",
)
# 설정 객체를 한 번만 생성하여 재사용하도록 캐시합니다.
@lru_cache
def get_settings() -> Settings:
"""프로젝트 전역에서 사용할 Settings 객체를 반환합니다."""
# Settings 인스턴스를 생성하여 반환합니다.
return Settings()
main.py
import fastapi
fastapi.responses htmlresponse json대신 html을 반환하기 위한 htmlresponse
작성한 app. routers.api에서의 router와 app.config.settings의 get_settings 가져오기
case_sensitive 대소문자 구분하징낳도록.
fastapi 객체 생성, router 설정
@app.get("/", response_class=HTMLResponse) html객체 return
if __name__ == "__main__": uvicorn asgi 서버 실행 모듈을 가져와 run
uvicorn.run()은 FastAPI(또는 ASGI 애플리케이션)를 실행하는 함수
"""
FastAPI 애플리케이션의 실행 진입점입니다.
"""
# FastAPI 클래스를 가져옵니다.
from fastapi import FastAPI
# 브라우저에서 JSON 대신 간단한 안내 HTML을 반환하기 위해 HTMLResponse를 가져옵니다.
from fastapi.responses import HTMLResponse
# 프로젝트 API Router를 가져옵니다.
from app.routers.api import router
# 설정 객체를 가져옵니다.
from app.config.settings import get_settings
# 캐시된 설정 객체를 가져옵니다.
settings = get_settings()
# FastAPI 애플리케이션 객체를 생성합니다.
app = FastAPI(
title=settings.app_name,
description="OpenAI GPT, MCP, FAISS/Qdrant, MySQL을 학습하는 RAG Assistant API",
version="1.0.0",
)
# 프로젝트 REST API Router를 애플리케이션에 등록합니다.
app.include_router(router)
# 루트 URL에서 프로젝트 안내 화면을 반환합니다.
@app.get("/", response_class=HTMLResponse)
def home() -> str:
"""실행 확인과 주요 URL을 알려주는 간단한 HTML을 반환합니다."""
# 브라우저에서 바로 확인할 수 있는 HTML 문자열을 반환합니다.
return """
<!doctype html>
<html lang="ko">
<head>
<meta charset="utf-8">
<title>MCP RAG Assistant</title>
<style>
body { font-family: Arial, sans-serif; max-width: 900px; margin: 40px auto; line-height: 1.7; }
code { background: #f2f4f7; padding: 3px 7px; border-radius: 5px; }
.box { border: 1px solid #d0d7de; border-radius: 10px; padding: 20px; }
</style>
</head>
<body>
<h1>FastAPI + OpenAI + MCP 기반 RAG Assistant</h1>
<div class="box">
<p>FastAPI 서버가 정상적으로 실행 중입니다.</p>
<p>Swagger: <a href="/docs">/docs</a></p>
<p>상태 확인: <a href="/api/health">/api/health</a></p>
<p>먼저 Swagger에서 <code>POST /api/rag/rebuild</code>를 실행하세요.</p>
<p>MCP 서버는 별도 터미널에서 <code>python -m mcp_server.server</code>로 실행합니다.</p>
</div>
</body>
</html>
"""
# 이 파일을 직접 실행했을 때 Uvicorn 서버를 시작합니다.
if __name__ == "__main__":
# ASGI 서버 실행 모듈을 가져옵니다.
import uvicorn
# 문자열 import 방식으로 FastAPI 앱을 실행합니다.
uvicorn.run(
"app.main:app",
host=settings.app_host,
port=settings.app_port,
reload=True,
)
base.py
abc 추상클래스 정의
typing 임의 형태 메타데이터 타입 표현하기위한 any
자료형 정의
VectorStore 추상클래스, @abstractmethod rebuild(), search()
문서벡터와 저장하는 메서드, 질문벡터와 유사한 문서를 검색하는 메서드를 추상메서드로 정의한다 .
"""
벡터 저장소가 공통으로 구현해야 할 인터페이스를 정의합니다.
"""
# 추상 클래스를 정의하기 위해 ABC와 abstractmethod를 가져옵니다.
from abc import ABC, abstractmethod
# 임의 형태의 메타데이터 타입을 표현하기 위해 Any를 가져옵니다.
from typing import Any
# 벡터 검색 결과 자료형을 정의합니다.
SearchResult = dict[str, Any]
# 벡터 저장소 공통 인터페이스를 정의합니다.
class VectorStore(ABC):
"""FAISS와 Qdrant가 동일한 방식으로 호출되도록 하는 추상 클래스입니다."""
# 문서와 벡터를 저장하는 메서드를 추상 메서드로 정의합니다.
@abstractmethod
def rebuild(self, documents: list[dict], vectors: list[list[float]]) -> int:
"""기존 인덱스를 교체하고 전체 문서를 새로 저장합니다."""
# 질문 벡터와 유사한 문서를 검색하는 메서드를 추상 메서드로 정의합니다.
@abstractmethod
def search(self, query_vector: list[float], top_k: int) -> list[SearchResult]:
"""질문 벡터와 가까운 문서를 반환합니다."""
resources.py
container를 가져와 설정 딕셔너리로 구성해 return.
카탈로그 return
"""
MCP Resource가 반환할 데이터를 정의합니다.
"""
# JSON 문자열 생성을 위해 json을 가져옵니다.
import json
# 서비스 Container를 가져옵니다.
from app.services.container import get_container
# 파일 목록 Tool을 가져옵니다.
from app.tools.file_tools import list_doc_files
# 실행 설정을 Resource 문자열로 반환합니다.
def runtime_config() -> str:
"""민감정보를 제외한 현재 실행 설정을 JSON으로 반환합니다."""
# 공용 Container를 가져옵니다.
container = get_container()
# 사용자에게 공개해도 되는 설정만 딕셔너리로 구성합니다.
data = {
"app_name": container.settings.app_name,
"embedding_backend": container.settings.embedding_backend,
"vector_backend": container.settings.vector_backend,
"qdrant_mode": container.settings.qdrant_mode,
"mysql_enabled": container.settings.mysql_enabled,
"openai_configured": bool(container.settings.openai_api_key),
}
# 한글을 유지하는 들여쓰기 JSON 문자열로 반환합니다.
return json.dumps(data, ensure_ascii=False, indent=2)
# 문서 카탈로그를 Resource 문자열로 반환합니다.
def document_catalog() -> str:
"""docs 폴더에 있는 문서 목록을 JSON으로 반환합니다."""
# docs 폴더의 파일 목록을 가져옵니다.
files = list_doc_files(get_container().settings.docs_dir)
# 파일 목록을 JSON 문자열로 변환하여 반환합니다.
return json.dumps({"files": files}, ensure_ascii=False, indent=2)
tools.py
서비스 container, tools구현을 가져와 함수를 정의하기
"""
MCP Server에서 제공할 Tool 함수를 정의합니다.
"""
# 서비스 Container를 가져옵니다.
from app.services.container import get_container
# 파일 Tool 구현을 가져옵니다.
from app.tools.file_tools import list_doc_files, read_doc_file
# 두 숫자를 더합니다.
def add_numbers(a: float, b: float) -> float:
"""두 숫자의 합을 반환합니다."""
# 두 숫자를 더한 결과를 반환합니다.
return a + b
# docs 폴더의 파일 목록을 반환합니다.
def list_files() -> list[str]:
"""MCP Client가 사용할 수 있는 문서 파일 목록을 반환합니다."""
# Container 설정에서 docs 경로를 가져옵니다.
docs_dir = get_container().settings.docs_dir
# 공통 파일 Tool로 목록을 반환합니다.
return list_doc_files(docs_dir)
# docs 폴더의 파일을 읽습니다.
def read_file(filename: str) -> str:
"""지정한 문서 파일 내용을 반환합니다."""
# Container 설정에서 docs 경로를 가져옵니다.
docs_dir = get_container().settings.docs_dir
# 안전한 공통 파일 Tool을 호출합니다.
return read_doc_file(docs_dir, filename)
# 벡터 문서 검색을 수행합니다.
def search_documents(query: str, top_k: int = 4) -> list[dict]:
"""FAISS 또는 Qdrant에서 유사 문서를 검색합니다."""
# RAG 서비스의 검색 기능을 호출하여 결과를 반환합니다.
return get_container().rag_service.search(query, top_k)
# 문서 인덱스를 재구축합니다.
def rebuild_index() -> dict:
"""docs 폴더 전체를 벡터 저장소에 다시 적재합니다."""
# RAG 서비스의 인덱스 재구축 기능을 호출합니다.
return get_container().rag_service.rebuild_index()
# RAG 답변을 생성합니다.
def ask_rag(question: str, top_k: int = 4) -> dict:
"""검색 문서를 근거로 답변과 출처를 반환합니다."""
# RAG 서비스의 질의응답 기능을 호출합니다.
return get_container().rag_service.ask(question, top_k)
# MySQL 지식 목록을 조회합니다.
def list_mysql_knowledge() -> list[dict]:
"""MySQL knowledge_items 테이블 데이터를 반환합니다."""
# MySQL 서비스의 조회 기능을 호출합니다.
return get_container().mysql_service.list_items()
server.py
기존에 만들어진 python함수들을 mcp tool로 노출하는 서버 코드
함수구현은 이미 있다. mcp가 이해할 수 있도록 어너테이션을 추가해 등록해 ai가 호출가능하도록 만들었다 .Decorator가 실행되면서 MCP Server에 등록되어 사용하게 된다. 어차피 개발자가 다 구현하는데 기존이랑 뭐가 다를까? AI Agent가 표준 방식으로 기능을 발견(discovery)하고 호출할 수 있게 만드는 규격.
기존 함수를 가져와서 @mcp.tool()로 감싸면 AI가 사용할 수 있는 Tool이 된다를 이해하자 .mcp.
FastMCP()객체를 만들어
@mcp.tool() 로 명시하기, mcp.run(transport = stdio)
from mcp.server.fastmcp import FastMCP
from app.routers.schemas import RagRequest
from mcp_server.resources import document_catalog, runtime_config
from mcp_server.tools import (
add_numbers as add_numbers_tool,
ask_rag as ask_rag_tool,
list_files as list_files_tool,
list_mysql_knowledge as list_mysql_knowledge_tool,
read_file as read_file_tool,
rebuild_index as rebuild_index_tool,
search_documents as search_documents_tool,
)
mcp = FastMCP("MCP RAG Assistant")
@mcp.tool()
def add_numbers(a: float, b: float) -> float:
return add_numbers_tool(a, b)
@mcp.tool()
def ask_rag(request: RagRequest) -> dict:
return ask_rag_tool(request)
@mcp.tool()
def list_files() -> list[str]:
return list_files_tool()
@mcp.tool()
def mysql_knowledge() -> list[dict]:
return list_mysql_knowledge_tool()
@mcp.tool()
def read_file(request: FileReadRequest) -> dict:
return read_file_tool(request)
@mcp.tool()
def rebuild_index() -> dict:
return rebuild_index_tool()
@mcp.tool()
def search_documents(query: str) -> list[dict]:
return search_documents_tool(query)
@mcp.resource("config://runtime")
def config_resource() -> str:
return runtime_config()
@mcp.resource("docs://catalog")
def docs_resource() -> str:
return document_catalog()
@mcp.prompt()
def grounded_rag_prompt(question:str) -> str:
return (
f"먼저 vector_search 또는 rag_question_answer Tool을 사용하세요, 검색결과에 포함된 문서만 근거로 답하세요, "
f"확인할 수 없는 내용은 추측하지마세효, 사용자질문 {question}"
)
if __name__ == "__main__":
mcp.run(transport="stdio")
사용자 - FastAPI 애플리케이션 서버 - Openai API(LLM) - MCP Client - MCP Server - tool / resource / prompt
사용자가 자연어로 요청하면 서비스서버 fastapi에서 사용자과 ai시스템 사이의 웹서버로http요청을 받아 인증처리, 요청을 전달하고 응답을 반환한다. openai은 llm으로 생각하고 판단하는 모델. tool을 선택하고 실행을 판단해 답변을 생성한다.
이때 mcp client는 llm과 mcp server사이으 ㅣ통신 담당자로 mcp server에서 tool목록을 가져와 llm에게 정보를 전달하고 mcp server는 실제코드가 있고 tool은 llm이 실행할수잇는함수가잇다. 이때 mcp server는 server라는 이름때문에 웹서버라고 생각하기 쉽지만 반드시 거대한 별도의 서버 프로그램일 필요가 없으며 mcp규격으로 tool resource prompt를 제공하는 프로그램일 뿐이다 .파일하나일수도있고 별도의 서비스일수도 있는것. 중요한점은 역할이 서버이기 때문인데 일반적인 웹서버와 비교했을때 요청을 받고 기능을 제공한ㄴ다는 의미에서 server이다 .
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My Server")
@mcp.tool()
def add(a: int, b: int):
"""두 숫자를 더합니다."""
return a + b
@mcp.tool()
def search_document(query: str):
"""문서를 검색합니다."""
return "검색 결과"
if __name__ == "__main__":
mcp.run(transport="stdio")
실제 서비스에서는 아래와같이 분리하며 mcp server는 엡서버와다르다.
mcp_server
│
├── main.py # MCP 실행
│
├── tools
│ ├── mysql.py
│ ├── search.py
│ └── email.py
│
├── resources
│ ├── docs.py
│ └── config.py
│
├── services
│ └── rag_service.py
│
└── vectorstore
└── faiss.py
'Personal > SK 네트웍스 AI 캠프' 카테고리의 다른 글
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day47_LangGraph 기반 멀티 에이전트 구조와 응용 (0) | 2026.07.16 |
|---|---|
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day46_LangGraph 기반 멀티 에이전트 구조와 응용 (1) | 2026.07.15 |
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day44_Vector DB와 LLM을 결합한 RAG 아키텍처 (0) | 2026.07.13 |
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day43_Tools를 활용한 ReAct 에이전트 구현 (0) | 2026.07.10 |
| [SK네트웍스 Family AI 캠프] 32기 11주차 회고: Day40 ~ Day43 (0) | 2026.07.08 |
