hgzero/rag/src/api/main.py
2025-10-29 15:31:29 +09:00

644 lines
20 KiB
Python

"""
Vector DB 통합 시스템 FastAPI 애플리케이션
"""
from fastapi import FastAPI, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from typing import List, Dict, Any
import logging
from pathlib import Path
from ..models.term import (
Term,
TermSearchRequest,
TermSearchResult,
TermExplainRequest,
TermExplanation,
TermStats
)
from ..models.document import (
DocumentSearchRequest,
DocumentSearchResult,
DocumentStats
)
from ..models.minutes import (
MinutesSearchRequest,
MinutesSearchResult,
RelatedMinutesRequest,
RelatedMinutesResponse
)
from ..db.postgres_vector import PostgresVectorDB
from ..db.azure_search import AzureAISearchDB
from ..db.rag_minutes_db import RagMinutesDB
from ..services.claude_service import ClaudeService
from ..utils.config import load_config, get_database_url
from ..utils.embedding import EmbeddingGenerator
from ..utils.text_processor import extract_nouns_as_query
from ..utils.redis_cache import RedisCache
# 로깅 설정
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# FastAPI 앱 생성
app = FastAPI(
title="Vector DB 통합 시스템",
description="회의록 작성 시스템을 위한 Vector DB 기반 용어집 및 관련자료 검색 API",
version="1.0.0"
)
# CORS 설정
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 전역 변수 (의존성 주입용)
_config = None
_term_db = None
_doc_db = None
_rag_minutes_db = None
_embedding_gen = None
_claude_service = None
_redis_cache = None
def get_config():
"""설정 로드"""
global _config
if _config is None:
config_path = Path(__file__).parent.parent.parent / "config.yaml"
_config = load_config(str(config_path))
return _config
def get_term_db():
"""용어집 DB 연결"""
global _term_db
if _term_db is None:
config = get_config()
db_url = get_database_url(config)
_term_db = PostgresVectorDB(db_url)
return _term_db
def get_doc_db():
"""관련자료 DB 연결"""
global _doc_db
if _doc_db is None:
config = get_config()
azure_search = config["azure_search"]
_doc_db = AzureAISearchDB(
endpoint=azure_search["endpoint"],
api_key=azure_search["api_key"],
index_name=azure_search["index_name"],
api_version=azure_search["api_version"]
)
return _doc_db
def get_rag_minutes_db():
"""RAG 회의록 DB 연결"""
global _rag_minutes_db
if _rag_minutes_db is None:
config = get_config()
db_url = get_database_url(config)
_rag_minutes_db = RagMinutesDB(db_url)
return _rag_minutes_db
def get_embedding_gen():
"""임베딩 생성기"""
global _embedding_gen
if _embedding_gen is None:
config = get_config()
azure_openai = config["azure_openai"]
_embedding_gen = EmbeddingGenerator(
api_key=azure_openai["api_key"],
endpoint=azure_openai["endpoint"],
model=azure_openai["embedding_model"],
dimension=azure_openai["embedding_dimension"],
api_version=azure_openai["api_version"]
)
return _embedding_gen
def get_claude_service():
"""Claude 서비스"""
global _claude_service
if _claude_service is None:
config = get_config()
claude = config["claude"]
_claude_service = ClaudeService(
api_key=claude["api_key"],
model=claude["model"],
max_tokens=claude["max_tokens"],
temperature=claude["temperature"]
)
return _claude_service
def get_redis_cache():
"""Redis 캐시"""
global _redis_cache
if _redis_cache is None:
config = get_config()
redis_config = config["redis"]
_redis_cache = RedisCache(
host=redis_config["host"],
port=redis_config["port"],
db=redis_config["db"],
password=redis_config.get("password"),
decode_responses=redis_config.get("decode_responses", True)
)
return _redis_cache
# ============================================================================
# 용어집 API
# ============================================================================
@app.get("/")
async def root():
"""루트 엔드포인트"""
return {
"service": "Vector DB 통합 시스템",
"version": "1.0.0",
"endpoints": {
"용어집": "/api/terms/*",
"관련자료": "/api/documents/*"
}
}
@app.post("/api/rag/terms/search", response_model=List[TermSearchResult])
async def search_terms(
request: TermSearchRequest,
term_db: PostgresVectorDB = Depends(get_term_db),
embedding_gen: EmbeddingGenerator = Depends(get_embedding_gen)
):
"""
용어 검색 (Hybrid: Keyword + Vector)
Args:
request: 검색 요청
Returns:
검색 결과 리스트
"""
try:
config = get_config()
# 명사 추출하여 검색 쿼리 생성
search_query = extract_nouns_as_query(request.query)
logger.info(f"검색 쿼리 변환: '{request.query}''{search_query}'")
if request.search_type == "keyword":
# 키워드 검색
results = term_db.search_by_keyword(
query=search_query,
top_k=request.top_k,
confidence_threshold=request.confidence_threshold
)
elif request.search_type == "vector":
# 벡터 검색 (임베딩은 원본 쿼리 사용)
query_embedding = embedding_gen.generate_embedding(search_query)
results = term_db.search_by_vector(
query_embedding=query_embedding,
top_k=request.top_k,
confidence_threshold=request.confidence_threshold
)
else: # hybrid
# 하이브리드 검색
keyword_results = term_db.search_by_keyword(
query=search_query,
top_k=request.top_k,
confidence_threshold=request.confidence_threshold
)
query_embedding = embedding_gen.generate_embedding(search_query)
vector_results = term_db.search_by_vector(
query_embedding=query_embedding,
top_k=request.top_k,
confidence_threshold=request.confidence_threshold
)
# RRF 통합
keyword_weight = config["term_glossary"]["search"]["keyword_weight"]
vector_weight = config["term_glossary"]["search"]["vector_weight"]
# 간단한 가중합
results = []
seen_ids = set()
for result in keyword_results:
term_id = result["term"].term_id
if term_id not in seen_ids:
result["relevance_score"] *= keyword_weight
result["match_type"] = "hybrid"
results.append(result)
seen_ids.add(term_id)
for result in vector_results:
term_id = result["term"].term_id
if term_id not in seen_ids:
result["relevance_score"] *= vector_weight
result["match_type"] = "hybrid"
results.append(result)
seen_ids.add(term_id)
# 점수 기준 재정렬
results.sort(key=lambda x: x["relevance_score"], reverse=True)
results = results[:request.top_k]
# 응답 형식으로 변환
return [
TermSearchResult(
term=result["term"],
relevance_score=result["relevance_score"],
match_type=result["match_type"]
)
for result in results
]
except Exception as e:
logger.error(f"용어 검색 실패: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/rag/terms/{term_id}", response_model=Term)
async def get_term(
term_id: str,
term_db: PostgresVectorDB = Depends(get_term_db)
):
"""
용어 상세 조회
Args:
term_id: 용어 ID
Returns:
용어 객체
"""
try:
term = term_db.get_term_by_id(term_id)
if not term:
raise HTTPException(status_code=404, detail="용어를 찾을 수 없습니다")
return term
except HTTPException:
raise
except Exception as e:
logger.error(f"용어 조회 실패: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/rag/terms/{term_id}/explain", response_model=TermExplanation)
async def explain_term(
term_id: str,
request: TermExplainRequest,
term_db: PostgresVectorDB = Depends(get_term_db),
claude_service: ClaudeService = Depends(get_claude_service)
):
"""
용어 맥락 기반 설명 생성 (Claude AI)
Args:
term_id: 용어 ID
request: 설명 요청
Returns:
용어 설명
"""
try:
# 용어 조회
term = term_db.get_term_by_id(term_id)
if not term:
raise HTTPException(status_code=404, detail="용어를 찾을 수 없습니다")
# Claude AI 호출
result = claude_service.explain_term(
term_name=term.term_name,
definition=term.definition,
context=term.context,
meeting_context=request.meeting_context
)
return TermExplanation(
term=term,
explanation=result["explanation"],
context_documents=[],
generated_by=result["generated_by"],
cached=result["cached"]
)
except HTTPException:
raise
except Exception as e:
logger.error(f"용어 설명 생성 실패: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/rag/terms/stats", response_model=TermStats)
async def get_term_stats(term_db: PostgresVectorDB = Depends(get_term_db)):
"""용어 통계 조회"""
try:
stats = term_db.get_stats()
return TermStats(
total_terms=stats["total_terms"],
by_category=stats["by_category"],
by_source_type={},
avg_confidence=stats["avg_confidence"]
)
except Exception as e:
logger.error(f"통계 조회 실패: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
# ============================================================================
# 관련자료 API
# ============================================================================
@app.post("/api/rag/documents/search", response_model=List[DocumentSearchResult])
async def search_documents(
request: DocumentSearchRequest,
doc_db: AzureAISearchDB = Depends(get_doc_db),
embedding_gen: EmbeddingGenerator = Depends(get_embedding_gen)
):
"""
관련 문서 검색 (Hybrid Search + Semantic Ranking)
Args:
request: 검색 요청
Returns:
검색 결과 리스트
"""
try:
# 쿼리 임베딩 생성
query_embedding = embedding_gen.generate_embedding(request.query)
# Hybrid Search 실행
results = doc_db.hybrid_search(
query=request.query,
query_embedding=query_embedding,
top_k=request.top_k,
folder=request.folder,
document_type=request.document_type,
semantic_ranking=request.semantic_ranking
)
# 관련도 임계값 필터링
filtered_results = [
r for r in results
if r["relevance_score"] >= request.relevance_threshold
]
# 응답 형식으로 변환
return [
DocumentSearchResult(**result)
for result in filtered_results
]
except Exception as e:
logger.error(f"문서 검색 실패: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/rag/documents/stats", response_model=DocumentStats)
async def get_document_stats(doc_db: AzureAISearchDB = Depends(get_doc_db)):
"""문서 통계 조회"""
try:
stats = doc_db.get_stats()
return DocumentStats(
total_documents=stats["total_documents"],
by_type=stats["by_type"],
by_domain={},
total_chunks=stats["total_chunks"]
)
except Exception as e:
logger.error(f"통계 조회 실패: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
# ============================================================================
# RAG 회의록 API
# ============================================================================
@app.post("/api/rag/minutes/search", response_model=List[MinutesSearchResult])
async def search_related_minutes(
request: MinutesSearchRequest,
rag_minutes_db: RagMinutesDB = Depends(get_rag_minutes_db),
embedding_gen: EmbeddingGenerator = Depends(get_embedding_gen)
):
"""
연관 회의록 검색 (Vector Similarity)
Args:
request: 검색 요청
Returns:
유사 회의록 리스트
"""
try:
# 쿼리 임베딩 생성
logger.info(f"회의록 검색 시작: {request.query[:50]}...")
query_embedding = embedding_gen.generate_embedding(request.query)
# 벡터 유사도 검색
results = rag_minutes_db.search_by_vector(
query_embedding=query_embedding,
top_k=request.top_k,
similarity_threshold=request.similarity_threshold
)
# 응답 형식으로 변환
search_results = [
MinutesSearchResult(
minutes=result["minutes"],
similarity_score=result["similarity_score"]
)
for result in results
]
logger.info(f"회의록 검색 완료: {len(search_results)}개 결과")
return search_results
except Exception as e:
logger.error(f"회의록 검색 실패: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/rag/minutes/{minutes_id}")
async def get_minutes(
minutes_id: str,
rag_minutes_db: RagMinutesDB = Depends(get_rag_minutes_db)
):
"""
회의록 상세 조회
Args:
minutes_id: 회의록 ID
Returns:
회의록 객체
"""
try:
minutes = rag_minutes_db.get_minutes_by_id(minutes_id)
if not minutes:
raise HTTPException(status_code=404, detail="회의록을 찾을 수 없습니다")
return minutes
except HTTPException:
raise
except Exception as e:
logger.error(f"회의록 조회 실패: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/rag/minutes/stats")
async def get_minutes_stats(rag_minutes_db: RagMinutesDB = Depends(get_rag_minutes_db)):
"""회의록 통계 조회"""
try:
stats = rag_minutes_db.get_stats()
return stats
except Exception as e:
logger.error(f"통계 조회 실패: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/rag/minutes/related", response_model=List[RelatedMinutesResponse])
async def get_related_minutes(
request: RelatedMinutesRequest,
rag_minutes_db: RagMinutesDB = Depends(get_rag_minutes_db),
embedding_gen: EmbeddingGenerator = Depends(get_embedding_gen),
redis_cache: RedisCache = Depends(get_redis_cache)
):
"""
연관 회의록 조회 (Option A: DB 조회 후 벡터 검색 + Redis 캐싱)
Args:
request: 연관 회의록 조회 요청
- minute_id: 기준 회의록 ID
- meeting_title: 회의 제목 (미사용)
- summary: 회의록 요약 (미사용)
- top_k: 반환할 최대 결과 수
- similarity_threshold: 최소 유사도 임계값
Returns:
연관 회의록 리스트
Process:
1. Redis 캐시에서 연관 회의록 결과 조회
2. 캐시 MISS 시:
a. minute_id로 rag_minutes 테이블에서 회의록 조회 (캐싱)
b. full_content를 벡터 임베딩으로 변환
c. 벡터 DB에서 유사도 검색 (자기 자신 제외)
d. 결과를 Redis에 캐싱
3. 연관 회의록 목록 반환
"""
try:
config = get_config()
cache_config = config.get("rag_minutes", {}).get("cache", {})
cache_prefix = cache_config.get("prefix", "minutes:")
minutes_ttl = cache_config.get("ttl", 1800)
related_ttl = cache_config.get("related_ttl", 3600)
logger.info(f"연관 회의록 조회 시작: minute_id={request.minute_id}")
# 1. 캐시 키 생성
related_cache_key = (
f"{cache_prefix}related:{request.minute_id}:"
f"{request.top_k}:{request.similarity_threshold}"
)
# 2. 캐시 조회
cached_results = redis_cache.get(related_cache_key)
if cached_results:
logger.info(f"연관 회의록 캐시 HIT: {related_cache_key}")
return [
RelatedMinutesResponse(**result)
for result in cached_results
]
# 3. 캐시 MISS - DB 조회
logger.info(f"연관 회의록 캐시 MISS: {related_cache_key}")
# 3-1. 회의록 조회 (캐싱)
minutes_cache_key = f"{cache_prefix}{request.minute_id}"
base_minutes = redis_cache.get(minutes_cache_key)
if base_minutes:
logger.info(f"회의록 캐시 HIT: {minutes_cache_key}")
# RagMinutes 객체로 변환
from ..models.minutes import RagMinutes
base_minutes = RagMinutes(**base_minutes)
else:
logger.info(f"회의록 캐시 MISS: {minutes_cache_key}")
base_minutes = rag_minutes_db.get_minutes_by_id(request.minute_id)
if not base_minutes:
raise HTTPException(
status_code=404,
detail=f"회의록을 찾을 수 없습니다: {request.minute_id}"
)
# 캐시 저장
redis_cache.set(minutes_cache_key, base_minutes.dict(), minutes_ttl)
logger.info(f"기준 회의록 조회 완료: {base_minutes.title}")
# 3-2. full_content를 벡터 임베딩으로 변환
query_embedding = embedding_gen.generate_embedding(base_minutes.full_content)
logger.info(f"임베딩 생성 완료: {len(query_embedding)}차원")
# 3-3. 벡터 유사도 검색 (자기 자신 제외)
results = rag_minutes_db.search_by_vector(
query_embedding=query_embedding,
top_k=request.top_k,
similarity_threshold=request.similarity_threshold,
exclude_minutes_id=request.minute_id
)
# 4. 응답 형식으로 변환
related_minutes = [
RelatedMinutesResponse(
minutes=result["minutes"],
similarity_score=result["similarity_score"]
)
for result in results
]
# 5. 결과 캐싱
redis_cache.set(
related_cache_key,
[r.dict() for r in related_minutes],
related_ttl
)
logger.info(f"연관 회의록 조회 완료: {len(related_minutes)}개 결과")
return related_minutes
except HTTPException:
raise
except Exception as e:
logger.error(f"연관 회의록 조회 실패: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)