AI 벡터 데이터베이스 Pinecone: RAG 시스템 및 고성능 벡터 검색을 위한 데이터 플랫폼

# =================================================================
# 설정: API KEY, HOST, 임베딩 모델 정보 설정
# =================================================================
API_KEY = "YOUR_PINECONE_API_KEY"
PINECONE_HOST = "https://your-pinecone-instance.svc..pinecone.io"
EMBEDDING_MODEL_NAME = 'intfloat/multilingual-e5-large' # 고성능 다국어 임베딩 모델
BLOG_DOMAIN = "https:블로그 주소"
EMBEDDING_DIMENSION = 1024 # 모델이 생성하는 벡터 차원 (multilingual-e5-large 기준)

# 임베딩 모델 초기화
model = SentenceTransformer(EMBEDDING_MODEL_NAME)

# =================================================================
# 사이트맵에서 포스트 URL 수집 함수
# =================================================================
import xml.etree.ElementTree as ET

def parse_post_urls_from_xml_number(url):
urls = set()
try:
response = requests.get(url, timeout=15)
response.raise_for_status()
content = response.content.decode('utf-8')
content = re.sub(r'xmlns="[^"]+"', '', content)
root = ET.fromstring(content)

for element in root.findall('.//loc'):
loc_url = element.text
if loc_url:
# /숫자 형식의 URL만 필터링 (게시글 식별)
if re.search(r'/d+$', loc_url) and not any(keyword in loc_url for keyword in ['/category', '/pages', '/tag', '/guestbook']):
urls.add(loc_url.split('?')[0])
return urls
except Exception as e:
print(f" 사이트맵 파싱 오류 ({url}): {e}")
return set()

def get_post_urls_from_sitemap(blog_domain):
sitemap_url = f"{blog_domain}/sitemap.xml"
print(f"[{sitemap_url}]에서 RAG 데이터 소스 URL을 수집합니다...")
post_urls = parse_post_urls_from_xml_number(sitemap_url)
print(f" 사이트맵 파싱에서 총 {len(post_urls)}개의 게시글 URL을 찾았습니다.")
return list(post_urls)

# =================================================================
# HTML에서 콘텐츠 파싱 (데이터 전처리) 함수
# =================================================================
def scrape_blog_content(url):
title = "제목 없음"
content = ""
canonical_url = url

try:
response = requests.get(url, timeout=15)
response.raise_for_status()
html = response.text

soup = BeautifulSoup(html, 'html.parser')

# canonical URL 추출
canonical_tag = soup.find('link', rel='canonical')
if canonical_tag and 'href' in canonical_tag.attrs:
canonical_url = canonical_tag['href'].split('?')[0]

# 콘텐츠가 포함된 핵심 div 추출 (데이터 전처리)
main_div = soup.find('div', class_='tt_article_useless_p_margin contents_style')
if main_div:
# 제목 추출 및 제거
h1_tag = main_div.find('h1')
h2_tag = main_div.find('h2')
if h1_tag:
title = h1_tag.get_text(strip=True)
h1_tag.decompose()
elif h2_tag:
title = h2_tag.get_text(strip=True)
h2_tag.decompose()

# 정리된 텍스트 추출
content = main_div.get_text(separator='
', strip=True)
else:
print(f"⚠ 콘텐츠 div 없음: {url}")

return title, content, canonical_url

except Exception as e:
print(f" HTML 파싱 오류 ({url}): {e}")
return "오류 발생", "", url

# =================================================================
# Pinecone REST API를 이용한 벡터 배치 업로드 (Upsert) 함수
# =================================================================
import json
import requests

def upsert_batch_via_rest(vectors_batch):
"""벡터 배치를 Pinecone 인덱스에 업로드합니다."""
url = f"{PINECONE_HOST}/vectors/upsert"
# 네임스페이스는 필요에 따라 설정 가능 (RAG 등에서 데이터 분리 용도)
payload = {"vectors": vectors_batch, "namespace": ""}
headers = {"Api-Key": API_KEY, "Content-Type": "application/json"}
try:
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"
Pinecone Upsert 오류 발생: {e}")
return None

# =================================================================
# 메인 함수 (전체 실행 로직)
# =================================================================
def main():
post_urls = get_post_urls_from_sitemap(BLOG_DOMAIN)
if not post_urls:
print(" 수집할 포스트 URL이 없습니다. 스크립트를 종료합니다.")
return

upserts = []

for url in tqdm(post_urls, desc="포스트 처리 및 임베딩 생성 중"):
title, content, canonical_url = scrape_blog_content(url)

if content and title != "오류 발생":
# 벡터 ID는 URL의 마지막 숫자로 설정 (고유 식별자)
vector_id = canonical_url.split('/')[-1]
try:
# Sentence Transformer를 사용하여 텍스트를 벡터로 인코딩
embedding = model.encode(content).tolist()
except Exception as e:
print(f"임베딩 생성 오류 (벡터화 실패): {e}")
continue

# Pinecone 업로드용 데이터 형식
upserts.append({
'id': vector_id,
'values': embedding,
'metadata': { # 메타데이터는 검색 결과에 함께 반환됨 (RAG에서 필수)
'text': content,
'title': title,
'url': canonical_url
}
})

if upserts:
print(f"
총 {len(upserts)}개의 임베딩 벡터를 Pinecone에 업로드합니다...")
# 100개 단위로 배치 업로드하여 효율성 높이기
for i in tqdm(range(0, len(upserts), 100), desc="Pinecone 벡터 데이터 업로드 중"):
batch = upserts[i:i + 100]
upsert_batch_via_rest(batch)
print("
AI 벡터 데이터 업로드 완료! 이제 유사도 검색을 수행할 수 있습니다.")
else:
print("업로드할 유효한 데이터가 없습니다.")