대규모 언어 모델(LLM)을 프로덕션 환경에서 운영할 때, RAG(Retrieval-Augmented Generation)는 필수 아키텍처입니다. 하지만 데이터가 복잡해지고 도메인 지식이 깊어질수록, 전통적인 벡터 검색만으로는 한계에 부딪히게 됩니다.
이 글에서는 Traditional RAG에서 GraphRAG로 마이그레이션하는 방법, HolySheep AI 플랫폼을 활용한 전환 과정, 그리고 실제 프로덕션 환경에서 발생할 수 있는 문제들을 상세히 다룹니다. 저는 3년 동안 RAG 시스템을 구축하며 Traditional RAG의 한계를 뼈저리게 느꼈고, GraphRAG 전환 후 문서 이해율이 40% 이상 향상된 경험을 공유합니다.
Traditional RAG와 GraphRAG: 핵심 차이점 이해
마이그레이션을 시작하기 전에, 두 접근법의 근본적인 차이를 이해해야 합니다. Traditional RAG는 시맨틱 벡터 유사도에 의존하는 반면, GraphRAG는 지식 그래프(Knowledge Graph)의 구조적 관계를 활용합니다.
Traditional RAG 동작 원리
전통적인 RAG 시스템은 문서를 청크(chunks)로 분할하고, 각 청크를 임베딩 모델로 벡터화한 후 벡터 데이터베이스에 저장합니다. 사용자 질문이 들어오면, 질문도 벡터화하여 유사도 기반 검색을 수행합니다.
# Traditional RAG 검색 파이프라인 예시
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
class TraditionalRAG:
def __init__(self, embeddings_model, vector_db):
self.embeddings_model = embeddings_model
self.vector_db = vector_db
def retrieve(self, query: str, top_k: int = 5):
"""
사용자의 질문에서 가장 관련성 높은 문서 청크 검색
Traditional RAG는 단순 벡터 유사도만 사용
"""
# 질문 벡터화
query_vector = self.embeddings_model.encode(query)
# 벡터 데이터베이스에서 유사도 검색
results = self.vector_db.search(
vector=query_vector,
top_k=top_k,
include_values=True
)
# 단순 유사도 점수로 정렬
return [
{
"content": doc.text,
"score": similarity,
"source": doc.metadata.get("source", "unknown")
}
for doc, similarity in results
]
def generate(self, query: str, context: str):
"""
HolySheep AI를 통한 응답 생성
"""
import requests
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "당신은 도움되는 AI 어시스턴트입니다."},
{"role": "user", "content": f"질문: {query}\n\n참고 문서:\n{context}"}
],
"temperature": 0.3,
"max_tokens": 1000
}
)
return response.json()["choices"][0]["message"]["content"]
Traditional RAG의 한계점
1. 단순 키워드 매칭에 의존하여 관련성 높은 문서를 놓칠 수 있음
2. 문서 간 관계나 맥락을 이해하지 못함
3. 복잡한 도메인 질문에서 정확한 답변 생성 어려움
GraphRAG 동작 원리
GraphRAG는 문서에서 엔티티(Entities)와 관계(Relationships)를 추출하여 지식 그래프를 구축합니다. 검색 시 벡터 유사도와 그래프 구조적 관계를 결합하여 훨씬 정교한 정보를 제공합니다.
# GraphRAG 검색 파이프라인 예시
import json
from typing import List, Dict, Set
import requests
class GraphRAG:
def __init__(
self,
embeddings_model,
graph_db,
entity_extraction_prompt: str,
relation_extraction_prompt: str
):
self.embeddings_model = embeddings_model
self.graph_db = graph_db
self.entity_prompt = entity_extraction_prompt
self.relation_prompt = relation_extraction_prompt
def extract_entities_and_relations(self, document: str) -> Dict:
"""
HolySheep AI를 사용하여 문서에서 엔티티와 관계 추출
GraphRAG의 핵심: 구조화된 지식抽出
"""
# 엔티티 추출
entity_response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": self.entity_prompt},
{"role": "user", "content": document}
],
"temperature": 0.1
}
)
entities = json.loads(entity_response.json()["choices"][0]["message"]["content"])
# 관계 추출
relation_response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": self.relation_prompt},
{"role": "user", "content": document}
],
"temperature": 0.1
}
)
relations = json.loads(relation_response.json()["choices"][0]["message"]["content"])
return {"entities": entities, "relations": relations}
def build_graph(self, documents: List[str]):
"""
다중 문서에서 지식 그래프 구축
GraphRAG는 개별 문서가 아닌 전체 지식 베이스를 이해
"""
all_entities = []
all_relations = []
for doc in documents:
extracted = self.extract_entities_and_relations(doc)
all_entities.extend(extracted["entities"])
all_relations.extend(extracted["relations"])
# 중복 엔티티 병합
unique_entities = self.deduplicate_entities(all_entities)
# 그래프 데이터베이스에 저장
for entity in unique_entities:
self.graph_db.create_node(
id=entity["id"],
properties={
"name": entity["name"],
"type": entity["type"],
"description": entity.get("description", ""),
"embedding": self.embeddings_model.encode(entity["name"])
}
)
for relation in all_relations:
self.graph_db.create_edge(
source_id=relation["source"],
target_id=relation["target"],
properties={
"type": relation["type"],
"description": relation.get("description", "")
}
)
def community_detection_retrieve(self, query: str, top_k: int = 5) -> List[Dict]:
"""
커뮤니티 감지 기반 검색: 관련 엔티티 그룹 전체를 반환
Traditional RAG와의 핵심 차이점
"""
query_vector = self.embeddings_model.encode(query)
# 1단계: 관련 엔티티 찾기
initial_entities = self.graph_db.find_similar_entities(
vector=query_vector,
top_k=10
)
# 2단계: 관련 커뮤니티 찾기 (그래프 구조 활용)
related_communities = set()
for entity in initial_entities:
community_id = self.graph_db.get_community(entity["id"])
related_communities.add(community_id)
# 3단계: 커뮤니티 내 모든 관련 엔티티와 관계 수집
community_context = []
for community_id in related_communities:
community_entities = self.graph_db.get_community_members(community_id)
community_relations = self.graph_db.get_community_relations(community_id)
# 커뮤니티 전체를 맥락으로 반환
context = self._format_community_context(
community_entities,
community_relations,
top_k=top_k
)
community_context.append(context)
return community_context
def _format_community_context(
self,
entities: List[Dict],
relations: List[Dict],
top_k: int
) -> Dict:
"""지식 그래프 컨텍스트를 읽기 쉬운 형식으로 변환"""
return {
"summary": f"엔티티 {len(entities)}개, 관계 {len(relations)}개 포함",
"entities": entities[:top_k],
"relations": relations[:top_k * 2],
"formatted_text": self._create_readable_summary(entities, relations)
}
def _create_readable_summary(self, entities: List[Dict], relations: List[Dict]) -> str:
"""엔티티와 관계를 자연어로 변환"""
lines = ["[지식 그래프 컨텍스트]"]
for entity in entities:
lines.append(f"- {entity['name']} ({entity['type']}): {entity.get('description', '')}")
lines.append("\n[관계]")
for relation in relations:
lines.append(f"- {relation['source']} --[{relation['type']}]--> {relation['target']}")
return "\n".join(lines)
GraphRAG vs Traditional RAG: 언제 무엇을 선택해야 하는가
| 평가 기준 | Traditional RAG | GraphRAG |
|---|---|---|
| 적합 데이터 유형 | 단순 문서, FAQ, 구조 없는 텍스트 | 복잡한 관계, 도메인 지식, 다중 문서 상호작용 |
| 검색 정확도 | 키워드/의미 유사도 기반, 맥락 이해 제한 | 엔티티 관계 이해, 글로벌 맥락 파악 가능 |
| 구현 복잡도 | 낮음 (단순 벡터 DB만 필요) | 높음 (그래프 DB + 파이프라인 필요) |
| 인프라 비용 | 벡터 DB만 ($50-200/월) | 벡터 DB + 그래프 DB ($150-500/월) |
| 응답 생성 품질 | 평균적, 단편적 정보 제공 | 높음, 구조화된 관계 정보 제공 |
| 스케일링 | 문서 수 증가 시 품질 저하 가능 | 관계 복잡도에 비례하여 품질 유지 |
| 홀루시네이션 감소 | 제한적 | 엔티티-관계 검증으로 크게 개선 |
| 개발 시간 | 1-2주 | 4-8주 (초기 구축) |
이런 팀에 적합 / 비적합
GraphRAG가 적합한 팀
- 도메인 전문가 집합: 법률, 의료, 금융 등 복잡한 관계 기반 지식이 필요한 분야
- 대규모 문서庫 운영: 수천 개 이상의 문서를 관리하고 정확한 정보 검색이 필요한 팀
- 높은 정확도 요구: 사실 확인이 중요하고 잘못된 답변 허용 범위가 좁은 시스템
- 멀티 홉 추론 필요: "A와 B의 관계를 통해 C를 추론" 같은 복잡한 질문이 빈번한 경우
- 지식 베이스 자주 업데이트: 그래프 구조로 변경 사항을 효율적으로 반영해야 하는 경우
- AI 기술 역량 갖춘 팀: 그래프 알고리즘과 LLM 통합에 대한 이해가 있는 개발자 보유
Traditional RAG가 적합한 팀
- 빠른 프로토타입 필요: 1-2주 내 POC完成이 목표인 경우
- 제한된 예산: 인프라 비용을 최소화해야 하는 초기 스타트업
- 단순 Q&A 중심: FAQ, 문서 검색 등 비교적 단순한 검색만 필요한 경우
- 제한된 개발リソース: 그래프 기술 전문 지식이 부족한 팀
- 소규모 문서集: 수백 개 미만의 문서만 다루는 경우
- 정형화된 데이터: 이미 구조화된 데이터베이스가 있고 단순 검색만 필요한 경우
Traditional RAG에서 GraphRAG로 마이그레이션 계획
1단계: 현재 시스템 평가 (1-2주)
저는 마이그레이션的第一步으로 기존 Traditional RAG 시스템의 성능 병목 지점을 정확히 파악하는 것이 중요합니다. 단순히 "느리다"는 말이 아니라, 어떤 유형의 질문에서 정확도가 떨어지는지 정량적으로 측정해야 합니다.
# Traditional RAG 성능 평가 스크립트
import json
import time
from typing import List, Dict, Tuple
import requests
class RAGPerformanceEvaluator:
def __init__(self, rag_system, evaluation_llm: str = "gpt-4.1"):
self.rag = rag_system
self.evaluation_llm = evaluation_llm
def evaluate_retrieval_precision(
self,
test_queries: List[Dict]
) -> Dict[str, float]:
"""
검색 정밀도 평가: 관련 문서가 상위 결과에 포함되었는가?
"""
results = {
"top1_precision": 0,
"top3_precision": 0,
"top5_precision": 0,
"total_queries": len(test_queries)
}
for query_data in test_queries:
query = query_data["query"]
relevant_docs = set(query_data["relevant_doc_ids"])
retrieved = self.rag.retrieve(query, top_k=5)
retrieved_ids = {doc["doc_id"] for doc in retrieved}
# Top-K 정밀도 계산
if retrieved_ids & relevant_docs:
results["top1_precision"] += 1 if list(retrieved_ids)[0] in relevant_docs else 0
results["top3_precision"] += 1 if bool(retrieved_ids & relevant_docs) else 0
results["top5_precision"] += 1 if bool(retrieved_ids & relevant_docs) else 0
# 평균 계산
n = results["total_queries"]
results["top1_precision"] /= n
results["top3_precision"] /= n
results["top5_precision"] /= n
return results
def evaluate_answer_quality(
self,
test_queries: List[Dict]
) -> Dict[str, any]:
"""
HolySheep AI를 사용한 답변 품질 평가
"""
evaluation_results = []
for query_data in test_queries:
query = query_data["query"]
expected_answer = query_data["expected_answer"]
# RAG 응답 생성
context = self._get_context(query)
generated_answer = self.rag.generate(query, context)
# 품질 평가 (LLM-as-Judge)
evaluation = self._evaluate_with_llm(
question=query,
expected=expected_answer,
generated=generated_answer
)
evaluation_results.append({
"query": query,
"generated": generated_answer,
"evaluation": evaluation,
"response_time": time.time()
})
return {
"average_score": sum(r["evaluation"]["score"] for r in evaluation_results) / len(evaluation_results),
"results": evaluation_results
}
def _get_context(self, query: str) -> str:
"""검색 결과를 컨텍스트 문자열로 변환"""
retrieved = self.rag.retrieve(query, top_k=5)
return "\n\n".join([
f"[문서 {i+1}] {doc['content']}"
for i, doc in enumerate(retrieved)
])
def _evaluate_with_llm(
self,
question: str,
expected: str,
generated: str
) -> Dict:
"""LLM을 사용한 답변 품질 평가"""
prompt = f"""다음 질문과 예상 답변, 생성된 답변을 평가해주세요.
질문: {question}
예상 답변: {expected}
생성된 답변: {generated}
0-100 점수로 정확도, 완전성, 일관성을 평가하고, 구체적인 피드백을 제공해주세요.
JSON 형식으로 응답해주세요:
{{"score": 숫자, "accuracy": 숫자, "completeness": 숫자, "coherence": 숫자, "feedback": "피드백 텍스트"}}
"""
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": self.evaluation_llm,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
"max_tokens": 500
}
)
return json.loads(response.json()["choices"][0]["message"]["content"])
def identify_failure_modes(self, test_queries: List[Dict]) -> Dict[str, List]:
"""
실패 모드 분석: 어떤 유형의 질문에서 Traditional RAG가 실패하는가?
GraphRAG 전환이 필요한지 판단하는 핵심 근거
"""
failure_modes = {
"multi_hop_questions": [], # 멀티 홉 질문
"relationship_questions": [], # 관계 질문
"comparative_questions": [], # 비교 질문
"inference_questions": [], # 추론 질문
"ambiguous_queries": [] # 모호한 질문
}
for query_data in test_queries:
query = query_data["query"]
evaluation = self._classify_query_type(query)
if evaluation["requires_graph_reasoning"]:
category = evaluation["primary_category"]
failure_modes[category].append({
"query": query,
"reason": evaluation["reason"],
"current_accuracy": query_data.get("accuracy", "unknown")
})
return failure_modes
def _classify_query_type(self, query: str) -> Dict:
"""질문 유형 분류"""
classification_prompt = f"""다음 질문이 어떤 유형인지 분류해주세요:
질문: {query}
분류 기준:
- multi_hop_questions: "A와 B의 관계를 통해 C를 추론" 같은 다단계 추론 필요
- relationship_questions: "A와 B의 차이점/유사점" 같은 관계 질문
- comparative_questions: "어떤 것이 더 좋은가" 같은 비교 질문
- inference_questions: 명시적으로 언급되지 않은 정보 추론 필요
- ambiguous_queries: 키워드가 모호하여 벡터 검색이 부정확할 수 있음
JSON으로 응답:
{{"requires_graph_reasoning": true/false, "primary_category": "카테고리", "reason": "판단 이유"}}
"""
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": classification_prompt}],
"temperature": 0.1
}
)
return json.loads(response.json()["choices"][0]["message"]["content"])
마이그레이션 결정 기준
이 결과에서 multi_hop_questions + relationship_questions가 30% 이상이면
GraphRAG 전환을 권장합니다
2단계: 마이그레이션 실행 (4-8주)
평가를 완료했다면, 실제 마이그레이션 단계를 진행합니다. 저는 병렬 운영 전략을 권장합니다. 기존 Traditional RAG를 완전히 대체하지 않고, GraphRAG를 병렬로 운영하며 점진적으로 트래픽을 전환합니다.
# GraphRAG 마이그레이션 파이프라인
import hashlib
from datetime import datetime
from typing import List, Dict, Optional
import requests
class GraphRAGMigrationPipeline:
def __init__(
self,
traditional_rag: TraditionalRAG,
graph_rag: GraphRAG,
hybrid_router: callable = None
):
self.traditional_rag = traditional_rag
self.graph_rag = graph_rag
self.hybrid_router = hybrid_router or self._default_router
self.migration_state = {
"documents_processed": 0,
"total_documents": 0,
"graph_entities": 0,
"graph_relations": 0,
"errors": [],
"start_time": None,
"last_checkpoint": None
}
def _default_router(self, query: str, traditional_score: float, graph_score: float) -> str:
"""기본 라우팅 로직: 점수 기반으로 선택"""
# 복잡한 질문은 GraphRAG, 단순 질문은 Traditional RAG
if self._is_complex_query(query):
return "graph"
return "graph" if graph_score > traditional_score else "traditional"
def _is_complex_query(self, query: str) -> bool:
"""질문 복잡도 판단"""
complexity_keywords = [
"관계", "차이", "비교", "추론", "왜", "어떻게",
"어느 것이", "무엇이", "어떤", "관련된"
]
return any(keyword in query for keyword in complexity_keywords)
def migrate_documents(
self,
documents: List[Dict],
checkpoint_interval: int = 100,
checkpoint_path: str = "./migration_checkpoint.json"
) -> Dict:
"""
문서 마이그레이션: Traditional RAG에서 GraphRAG로
체크포인트 기반 증분 마이그레이션 지원
"""
self.migration_state["start_time"] = datetime.now().isoformat()
self.migration_state["total_documents"] = len(documents)
# 체크포인트에서 재개
checkpoint = self._load_checkpoint(checkpoint_path)
start_index = checkpoint.get("processed_count", 0)
print(f"[마이그레이션 시작] {start_index}/{len(documents)} 문서 처리됨")
for i, doc in enumerate(documents[start_index:], start=start_index):
try:
# 1. 기존 Traditional RAG 데이터 확인
traditional_chunks = self.traditional_rag.get_chunks(doc["id"])
# 2. GraphRAG용 엔티티/관계 추출
extracted = self.graph_rag.extract_entities_and_relations(doc["content"])
# 3. 그래프에 저장
self.graph_rag.build_graph([doc["content"]])
# 4. 상태 업데이트
self.migration_state["documents_processed"] += 1
self.migration_state["graph_entities"] += len(extracted["entities"])
self.migration_state["graph_relations"] += len(extracted["relations"])
# 체크포인트 저장
if (i + 1) % checkpoint_interval == 0:
self._save_checkpoint(checkpoint_path, i + 1)
print(f"[체크포인트] {i + 1}/{len(documents)} 완료")
except Exception as e:
error_info = {
"document_id": doc.get("id", "unknown"),
"error": str(e),
"timestamp": datetime.now().isoformat()
}
self.migration_state["errors"].append(error_info)
print(f"[오류] 문서 {doc.get('id')}: {e}")
self._save_checkpoint(checkpoint_path, len(documents), final=True)
return {
"status": "completed" if len(self.migration_state["errors"]) == 0 else "completed_with_errors",
"statistics": self.migration_state,
"migration_time": (
datetime.fromisoformat(datetime.now().isoformat()) -
datetime.fromisoformat(self.migration_state["start_time"])
).total_seconds()
}
def _load_checkpoint(self, path: str) -> Dict:
"""체크포인트 로드"""
try:
with open(path, 'r') as f:
return json.load(f)
except FileNotFoundError:
return {"processed_count": 0, "is_final": False}
def _save_checkpoint(self, path: str, count: int, final: bool = False):
"""체크포인트 저장"""
checkpoint = {
"processed_count": count,
"is_final": final,
"timestamp": datetime.now().isoformat(),
"state": self.migration_state.copy()
}
with open(path, 'w') as f:
json.dump(checkpoint, f, indent=2)
def run_ab_test(
self,
test_queries: List[str],
traffic_split: float = 0.5,
duration_hours: int = 24
) -> Dict[str, any]:
"""
A/B 테스트: Traditional RAG vs GraphRAG 성능 비교
"""
results = {
"traditional": {"queries": [], "avg_latency": 0, "avg_score": 0},
"graph": {"queries": [], "avg_latency": 0, "avg_score": 0}
}
for query in test_queries:
# Traditional RAG 테스트
start = time.time()
traditional_result = self.traditional_rag.retrieve(query)
traditional_latency = time.time() - start
# GraphRAG 테스트
start = time.time()
graph_result = self.graph_rag.community_detection_retrieve(query)
graph_latency = time.time() - start
# 결과 저장
results["traditional"]["queries"].append({
"query": query,
"latency": traditional_latency,
"result": traditional_result
})
results["graph"]["queries"].append({
"query": query,
"latency": graph_latency,
"result": graph_result
})
# 평균 계산
n = len(test_queries)
results["traditional"]["avg_latency"] = sum(
q["latency"] for q in results["traditional"]["queries"]
) / n
results["graph"]["avg_latency"] = sum(
q["latency"] for q in results["graph"]["queries"]
) / n
return results
def gradual_traffic_migration(
self,
initial_percentage: int = 10,
increment: int = 10,
increment_interval_hours: int = 24
) -> Dict:
"""
점진적 트래픽 마이그레이션
HolySheep AI 모니터링 통합
"""
migration_plan = []
current_percentage = initial_percentage
while current_percentage <= 100:
migration_plan.append({
"target_percentage": current_percentage,
"action": "router_config_update",
"config": {
"graph_rag_weight": current_percentage / 100,
"traditional_rag_weight": (100 - current_percentage) / 100
},
"estimated_completion": f"+{increment_interval_hours}h"
})
current_percentage += increment
return {
"migration_plan": migration_plan,
"total_duration_hours": len(migration_plan) * increment_interval_hours,
"recommendation": (
"각 단계에서 100개 이상의 쿼리 샘플로 성능을 검증한 후 "
"다음 단계로 진행하세요. 오류율이 5%를 초과하면 롤백하세요."
)
}
3단계: 롤백 계획 수립
저는 어떤 마이그레이션이든 롤백 계획 없이 진행하지 않습니다. 특히 GraphRAG로의 전환은 인프라 변경이 크기 때문에, 완전한 롤백 메커니즘을 사전에 구축해야 합니다.
# GraphRAG 롤백 시스템
import sqlite3
from datetime import datetime, timedelta
from typing import Optional, List
import json
class GraphRAGRollbackManager:
def __init__(self, backup_db_path: str = "./rollback_backup.db"):
self.backup_db_path = backup_db_path
self._init_backup_database()
def _init_backup_database(self):
"""롤백용 백업 데이터베이스 초기화"""
conn = sqlite3.connect(self.backup_db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS rollback_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
snapshot_name TEXT NOT NULL,
traditional_rag_state BLOB,
graph_rag_state BLOB,
router_config TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
expires_at TIMESTAMP,
is_active BOOLEAN DEFAULT 0
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS rollback_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
action TEXT NOT NULL,
snapshot_id INTEGER,
executed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
status TEXT,
error_message TEXT,
FOREIGN KEY (snapshot_id) REFERENCES rollback_snapshots(id)
)
""")
conn.commit()
conn.close()
def create_snapshot(
self,
name: str,
traditional_rag_state: Dict,
graph_rag_state: Dict,
router_config: Dict,
retention_hours: int = 72
) -> int:
"""롤백 스냅샷 생성"""
conn = sqlite3.connect(self.backup_db_path)
cursor = conn.cursor()
expires_at = datetime.now() + timedelta(hours=retention_hours)
cursor.execute("""
INSERT INTO rollback_snapshots
(snapshot_name, traditional_rag_state, graph_rag_state, router_config, expires_at)
VALUES (?, ?, ?, ?, ?)
""", (
name,
json.dumps(traditional_rag_state),
json.dumps(graph_rag_state),
json.dumps(router_config),
expires_at.isoformat()
))
snapshot_id = cursor.lastrowid
conn.commit()
conn.close()
self._log_action("create_snapshot", snapshot_id, "success")
return snapshot_id
def restore_snapshot(self, snapshot_id: int) -> Dict:
"""스냅샷으로 롤백"""
conn = sqlite3.connect(self.backup_db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT traditional_rag_state, graph_rag_state, router_config, snapshot_name
FROM rollback_snapshots
WHERE id = ? AND expires_at > ?
""", (snapshot_id, datetime.now().isoformat()))
result = cursor.fetchone()
if not result:
error_msg = f"유효한 스냅샷을 찾을 수 없습니다: {snapshot_id}"
self._log_action("restore_snapshot", snapshot_id, "failed", error_msg)
conn.close()
raise ValueError(error_msg)
traditional_state = json.loads(result[0])
graph_state = json.loads(result[1])
router_config = json.loads(result[2])
snapshot_name = result[3]
# 실제 롤백 수행
rollback_result = {
"status": "rollback_initiated",
"snapshot_id": snapshot_id,
"snapshot_name": snapshot_name,
"actions": []
}
# 1. 라우터 설정을 Traditional RAG Only로 변경
router_config_backup = self._update_router_config(
weight_traditional=1.0,
weight_graph=0.0
)
rollback_result["actions"].append({
"action": "router_config_updated",
"status": "completed"
})
# 2. GraphRAG 상태 백업 (현재 상태 보존)
self._backup_current_state("pre_rollback_backup")
rollback_result["actions"].append({
"action": "current_state_backed_up",
"status": "completed"
})
# 3. Traditional RAG 상태 복원
self._restore_traditional_rag(traditional_state)
rollback_result["actions"].append({
"action": "traditional_rag_restored",
"status": "completed"
})
conn.close()
self._log_action("restore_snapshot", snapshot_id, "success")
rollback_result["status"] = "rollback_completed"
rollback_result["rollback_time"] = datetime.now().isoformat()
return rollback_result
def _update_router_config(self, weight_traditional: float, weight_graph: float) -> Dict:
"""라우터 설정 업데이트"""
# HolySheep AI API를 통한 설정 변경
import requests
response = requests.patch(
"https://api.holysheep.ai/v1/router/config",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"routing_weights": {
"traditional_rag": weight_traditional,
"graph_rag": weight_graph
},
"failover_mode": "traditional_rag"
}
)
return response.json()