안녕하세요, 저는 5년간 AI 시스템 구축을 해온 개발자입니다. 오늘은 MCP(Model Context Protocol)를 활용해서 RAG(검색 증강 생성) 시스템을 구현하는 방법을 초보자도 이해할 수 있도록 단계별로 알려드리겠습니다. 이 튜토리얼을 마치면, 문서 검색과 AI 생성 결과를 자연스럽게 연결하는 시스템을 만들 수 있게 됩니다.
MCP와 RAG이란 무엇인가요?
MCP(Model Context Protocol)란?
MCP는 AI 모델이 외부 도구나 데이터 소스와 안전하게 통신할 수 있게 해주는 프로토콜입니다. 마치 번역가가 두 나라 사람 사이에서 대화하게 해주는 통역사처럼, MCP는 AI가 데이터베이스, 검색 엔진, 파일 시스템 등 다양한 리소스에 접근할 수 있게 합니다.
RAG(검색 증강 생성)란?
RAG는 AI가 먼저 관련 문서를 검색한 뒤, 그 내용을 기반으로 답변을 생성하는 기법입니다. 예를 들어, 여러 개의 계약서 문서가 있을 때 "특정 조항이 포함된 계약서는?"이라고 물으면, AI가 먼저 관련 문서를 찾고 그 내용을 바탕으로 정확한 답변을 제공합니다.
왜 이 둘을 함께 사용하나요?
저는 실제로 수천 개의 기술 문서를 관리하는 시스템을 구축할 때 이 조합을 사용했습니다. 각 문서가 자주 업데이트되고, 사용자가 자연어로 검색해야 하는 상황이었습니다. MCP를 사용하면 검색 도구를 표준화된 방식으로 연결할 수 있어서, 나중에 검색 엔진을 바꾸거나 추가하더라도 코드를 크게 수정할 필요가 없었습니다.
기본 환경 설정하기
시작하기 전에 필요한 도구들을 설치하겠습니다. 이 튜토리얼에서는 Python 3.9 이상이 필요합니다.
# 필요한 패키지 설치
pip install mcp holysheep-sdk python-dotenv langchain langchain-community
프로젝트 폴더 생성 및 이동
mkdir rag-mcp-tutorial
cd rag-mcp-tutorial
환경 변수 파일 생성
touch .envHolySheep AI API 키 설정
지금 가입해서 HolySheep AI에서 무료 크레딧을 받으세요. HolySheep AI는海外 신용카드 없이 로컬 결제를 지원하며, 단일 API 키로 GPT-4.1, Claude, Gemini 등 모든 주요 모델을 통합할 수 있습니다. 가입 후 받은 API 키를 .env 파일에 저장합니다.
# .env 파일에 다음 내용 작성
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
BASE_URL=https://api.holysheep.ai/v1MCP 서버 구성하기
MCP 서버는 검색 도구와 AI 모델 사이에서 중개자 역할을 합니다. 이번 섹션에서는 파일 기반 검색 서버와 데이터베이스 검색 서버 두 가지를 만들어보겠습니다.
# mcp_server.py
import json
import os
from pathlib import Path
from typing import List, Dict, Any, Optional
class SimpleSearchServer:
"""간단한 문서 검색 MCP 서버"""
def __init__(self, documents_path: str):
self.documents_path = Path(documents_path)
self.document_index: Dict[str, Any] = {}
def initialize(self):
"""문서 인덱스 초기화"""
if not self.documents_path.exists():
self.documents_path.mkdir(parents=True, exist_ok=True)
self._create_sample_documents()
self._build_index()
print(f"인덱스 초기화 완료: {len(self.document_index)}개 문서 로드됨")
def _create_sample_documents(self):
"""샘플 문서 생성"""
documents = {
"product_guide_ko.txt": """HolySheep AI 제품 가이드
- HolySheep AI는 글로벌 AI API 게이트웨이입니다
- GPT-4.1 모델 제공 (비용: 미션당 8달러)
- Claude Sonnet 4.5 모델 제공 (비용: 미션당 15달러)
- Gemini 2.5 Flash 모델 제공 (비용: 미션당 2.5달러)
- DeepSeek V3.2 모델 제공 (비용: 미션당 0.42달러)
- 로컬 결제 지원으로海外 신용카드 불필요
""",
"api_reference.txt": """HolySheep AI API 레퍼런스
- 기본 URL: https://api.holysheep.ai/v1
- 채팅 완성 API: POST /chat/completions
- 임베딩 API: POST /embeddings
- 사용량 조회: GET /usage
- API 키는 HolySheep 대시보드에서 생성 가능
""",
"pricing_info.txt": """요금제 정보
- 월간 구독 또는 종량제 선택 가능
- 무료 크레딧: 신규 가입 시 제공
- 사용량 초과 시 자동 결제
- 한국 원화로 결제 가능
"""
}
for filename, content in documents.items():
filepath = self.documents_path / filename
filepath.write_text(content, encoding='utf-8')
def _build_index(self):
"""문서 인덱스 구축"""
for filepath in self.documents_path.glob("*.txt"):
self.document_index[filepath.name] = filepath.read_text(encoding='utf-8')
def search(self, query: str, max_results: int = 3) -> List[Dict[str, Any]]:
"""문서 검색 수행"""
results = []
query_lower = query.lower()
# 단순 키워드 매칭 기반 검색
for filename, content in self.document_index.items():
content_lower = content.lower()
if any(keyword in content_lower for keyword in query_lower.split()):
# 관련성 점수 계산
score = sum(1 for keyword in query_lower.split() if keyword in content_lower)
results.append({
"filename": filename,
"score": score,
"content": content.strip(),
"preview": content[:200].replace("\n", " ") + "..."
})
# 점수순 정렬
results.sort(key=lambda x: x["score"], reverse=True)
return results[:max_results]
def get_document(self, filename: str) -> Optional[str]:
"""특정 문서 내용 반환"""
return self.document_index.get(filename)
def list_documents(self) -> List[str]:
"""모든 문서 목록 반환"""
return list(self.document_index.keys())
MCP 도구 정의
TOOLS = [
{
"name": "search_documents",
"description": "사용자의 질문과 관련된 문서를 검색합니다. 질문어 포함된 모든 키워드를 검색합니다.",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "검색할 질문"},
"max_results": {"type": "integer", "description": "반환할 최대 결과 수", "default": 3}
},
"required": ["query"]
}
},
{
"name": "get_document_content",
"description": "특정 문서의 전체 내용을 가져옵니다.",
"input_schema": {
"type": "object",
"properties": {
"filename": {"type": "string", "description": "문서 파일명"}
},
"required": ["filename"]
}
},
{
"name": "list_all_documents",
"description": "사용 가능한 모든 문서의 목록을 반환합니다.",
"input_schema": {"type": "object", "properties": {}}
}
]
if __name__ == "__main__":
# 테스트 실행
server = SimpleSearchServer("./documents")
server.initialize()
# 도구 목록 출력
print("\n=== 사용 가능한 MCP 도구 ===")
for tool in TOOLS:
print(f"- {tool['name']}: {tool['description']}")RAG 시스템 구현하기
이제 HolySheep AI API와 MCP 검색 서버를 연결해서 RAG 파이프라인을 구축하겠습니다.
# rag_system.py
import os
import json
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from dotenv import load_dotenv
환경 변수 로드
load_dotenv()
@dataclass
class SearchResult:
"""검색 결과 데이터 클래스"""
filename: str
content: str
preview: str
score: float
@dataclass
class RAGResponse:
"""RAG 응답 데이터 클래스"""
answer: str
sources: List[SearchResult]
model_used: str
tokens_used: int
latency_ms: float
class HolySheepAIClient:
"""HolySheep AI API 클라이언트"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.model = "gpt-4.1" # 기본 모델: GPT-4.1
def chat_completion(
self,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 1000
) -> Dict[str, Any]:
"""채팅 완성 API 호출"""
import requests
url = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()
class RAGPipeline:
"""RAG 파이프라인: 검색 + 생성 통합 시스템"""
def __init__(self, search_server, ai_client: HolySheepAIClient):
self.search_server = search_server
self.ai_client = ai_client
def _format_context(self, results: List[SearchResult]) -> str:
"""검색 결과를 컨텍스트 형식으로 변환"""
context_parts = []
for i, result in enumerate(results, 1):
context_parts.append(
f"[문서 {i}: {result.filename}]\n{result.content}\n"
)
return "\n".join(context_parts)
def _build_prompt(self, query: str, context: str) -> List[Dict[str, str]]:
"""RAG 프롬프트 구성"""
system_prompt = """당신은 문서 기반 질문 답변 어시스턴트입니다.
주어진 문서를 참조해서 정확하고 상세한 답변을 제공하세요.
답변을 만들 때 반드시 참조한 문서의 출처를 명시하세요.
문서에서 답변에 대한 정보를 찾을 수 없으면, "제공된 문서에서 해당 정보를 찾을 수 없습니다"라고 답변하세요."""
user_prompt = f"""=== 참조 문서 ===
{context}
=== 질문 ===
{query}
=== 답변 작성 규칙 ===
1. 참조 문서 내용을 기반으로 답변하세요
2. 답변에 사용된 출처(문서명)를 명시하세요
3. 문서에 없는 내용은 추측하지 마세요"""
return [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
]
def query(self, question: str, max_search_results: int = 3) -> RAGResponse:
"""질문 처리 파이프라인 실행"""
import time
start_time = time.time()
# 1단계: 관련 문서 검색
search_results_raw = self.search_server.search(
query=question,
max_results=max_search_results
)
search_results = [
SearchResult(
filename=r["filename"],
content=r["content"],
preview=r["preview"],
score=r["score"]
)
for r in search_results_raw
]
# 2단계: 컨텍스트 구성
context = self._format_context(search_results)
# 3단계: HolySheep AI API로 답변 생성
messages = self._build_prompt(question, context)
response = self.ai_client.chat_completion(messages)
# 4단계: 결과 정리
latency_ms = (time.time() - start_time) * 1000
return RAGResponse(
answer=response["choices"][0]["message"]["content"],
sources=search_results,
model_used=response.get("model", "gpt-4.1"),
tokens_used=response.get("usage", {}).get("total_tokens", 0),
latency_ms=latency_ms
)
def interactive_mode(self):
"""대화형 질문 모드"""
print("\n=== RAG 검색 시스템에 오신 것을 환영합니다 ===")
print("질문을 입력하세요. '종료'를 입력하면 프로그램이 끝납니다.\n")
while True:
question = input("질문: ").strip()
if question.lower() in ["종료", "exit", "quit"]:
print("감사합니다. 프로그램을 종료합니다.")
break
if not question:
print("질문을 입력해주세요.\n")
continue
try:
result = self.query(question)
print(f"\n📄 답변:\n{result.answer}")
print(f"\n📚 참조 문서 ({len(result.sources)}개):")
for i, source in enumerate(result.sources, 1):
print(f" [{i}] {source.filename} (관련성: {source.score})")
print(f"\n⏱️ 처리 시간: {result.latency_ms:.2f}ms")
print(f"🤖 사용 모델: {result.model_used}\n")
except Exception as e:
print(f"오류 발생: {str(e)}\n")
if __name__ == "__main__":
from mcp_server import SimpleSearchServer
# HolySheep AI 클라이언트 초기화
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key:
print("오류: HOLYSHEEP_API_KEY 환경 변수가 설정되지 않았습니다.")
print(".env 파일을 확인해주세요.")
exit(1)
ai_client = HolySheepAIClient(api_key=api_key)
# 검색 서버 초기화
search_server = SimpleSearchServer("./documents")
search_server.initialize()
# RAG 파이프라인 생성
rag_pipeline = RAGPipeline(search_server, ai_client)
# 대화형 모드 시작
rag_pipeline.interactive_mode()검색 도구와 생성 모델의 자동 협력 구현
더 발전된 형태로, AI가 질문 내용을 분석해서 자동으로 적절한 검색 도구를 선택하고, 필요하면 추가 검색도 수행하는 시스템을 만들어보겠습니다.
# mcp_rag_orchestrator.py
import os
import json
import time
from typing import List, Dict, Any, Optional, Union
from dataclasses import dataclass, field
from enum import Enum
from dotenv import load_dotenv
load_dotenv()
class ToolType(Enum):
"""사용 가능한 도구 유형"""
DOCUMENT_SEARCH = "document_search"
WEB_SEARCH = "web_search"
DATABASE_QUERY = "database_query"
CALCULATOR = "calculator"
@dataclass
class ToolCall:
"""도구 호출 정보"""
tool_name: str
tool_type: ToolType
parameters: Dict[str, Any]
reasoning: str
@dataclass
class Message:
"""대화 메시지"""
role: str # "user", "assistant", "system"
content: str
tool_calls: List[ToolCall] = field(default_factory=list)
class MCPToolRegistry:
"""MCP 도구 레지스트리: 사용 가능한 도구 관리"""
def __init__(self):
self.tools: Dict[str, Dict[str, Any]] = {}
self._register_default_tools()
def _register_default_tools(self):
"""기본 도구 등록"""
self.register_tool(
name="document_search",
description="문서库에서 관련 정보를 검색합니다. 사실 확인, 제품 정보, 가이드 등 문서 기반 질문에 사용됩니다.",
tool_type=ToolType.DOCUMENT_SEARCH,
parameters={
"query": {"type": "string", "required": True},
"max_results": {"type": "integer", "default": 5}
}
)
self.register_tool(
name="web_search",
description="웹에서 실시간 정보를 검색합니다. 뉴스, 최신 동향, 날씨 등 현재 정보가 필요한 질문에 사용됩니다.",
tool_type=ToolType.WEB_SEARCH,
parameters={
"query": {"type": "string", "required": True},
"num_results": {"type": "integer", "default": 5}
}
)
self.register_tool(
name="calculate",
description="수학적 계산이나 데이터 분석을 수행합니다.",
tool_type=ToolType.CALCULATOR,
parameters={
"expression": {"type": "string", "required": True}
}
)
def register_tool(
self,
name: str,
description: str,
tool_type: ToolType,
parameters: Dict[str, Any]
):
"""새 도구 등록"""
self.tools[name] = {
"name": name,
"description": description,
"type": tool_type,
"parameters": parameters
}
def get_tools_for_prompt(self) -> str:
"""프롬프트에 포함할 도구 설명 생성"""
tool_descriptions = []
for name, tool in self.tools.items():
param_info = ", ".join(
f"{p}: {info.get('type', 'any')}"
for p, info in tool["parameters"].items()
)
tool_descriptions.append(
f"- {name}({param_info}): {tool['description']}"
)
return "\n".join(tool_descriptions)
def execute_tool(self, name: str, parameters: Dict[str, Any]) -> Any:
"""도구 실행"""
if name not in self.tools:
raise ValueError(f"도구를 찾을 수 없습니다: {name}")
tool = self.tools[name]
# 도구 유형에 따른 실제 실행 로직
if tool["type"] == ToolType.DOCUMENT_SEARCH:
return self._execute_document_search(**parameters)
elif tool["type"] == ToolType.WEB_SEARCH:
return self._execute_web_search(**parameters)
elif tool["type"] == ToolType.CALCULATOR:
return self._execute_calculate(**parameters)
else:
return {"error": "지원되지 않는 도구 유형"}
def _execute_document_search(self, query: str, max_results: int = 5) -> Dict[str, Any]:
"""문서 검색 실행"""
# 실제 구현에서는 여기서 벡터 데이터베이스나 파일 시스템을 查询
return {
"results": [
{"title": "문서 1", "snippet": f"'{query}' 관련 내용입니다", "score": 0.95},
{"title": "문서 2", "snippet": f"'{query}'에 대한 추가 정보", "score": 0.87}
],
"total": 2
}
def _execute_web_search(self, query: str, num_results: int = 5) -> Dict[str, Any]:
"""웹 검색 실행"""
return {
"results": [
{"title": f"'{query}' 관련 웹 결과", "url": "https://example.com"}
]
}
def _execute_calculate(self, expression: str) -> Dict[str, Any]:
"""계산 실행"""
try:
result = eval(expression) # 실제 환경에서는 안전한 eval 사용 필요
return {"expression": expression, "result": result}
except Exception as e:
return {"error": str(e)}
class IntelligentRAGOrchestrator:
"""지능형 RAG 오케스트레이터: 도구 선택 + 검색 + 생성 자동화"""
def __init__(self, api_key: str, tool_registry: MCPToolRegistry):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.tool_registry = tool_registry
self.conversation_history: List[Message] = []
self.max_iterations = 3 # 최대 도구 호출 반복 횟수
def _call_api(self, messages: List[Dict[str, Any]]) -> Dict[str, Any]:
"""HolySheep AI API 호출"""
import requests
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "g