Pinecone

 

순서

먼저 https://app.pinecone.io/ 방문해서 api-key를 생성한다.

 

vs pgvector

• pgvector
  • 이미 PostgreSQL 기반 인프라가 있고, 자체 호스팅을 선호할 때
  • SQL과 벡터를 한 곳에서 관리하며, 커스터마이징·확장성을 직접 책임지고 싶을 때
• Pinecone
  • “운영 부담 없이” 곧바로 대규모 서비스 전환이 필요할 때
  • 메타데이터·하이브리드 검색·자동 스케일링 같은 고급 기능을 즉시 활용

 

실습

 

#!/usr/bin/env python3
"""
demo_pinecone.py ─ Pinecone 4.x + OpenAI 임베딩 실습
"""

import os, sys, time, logging
from dotenv import load_dotenv
import openai
from pinecone import Pinecone, ServerlessSpec      # ⬅️ 새 방식

# ──────── 로그 설정 ────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger(__name__)

# ──────── 환경 변수 읽기 ────────
load_dotenv()
try:
    OPENAI_API_KEY  = os.environ["OPENAI_API_KEY"]
    PINECONE_API_KEY = os.environ["PINECONE_API_KEY"]
    PINECONE_ENV     = os.environ["PINECONE_ENV"]    # ex) us-east-1-aws
except KeyError as e:
    log.error("환경변수 %s 가 없습니다 (.env 확인)", e.args[0]); sys.exit(1)

openai.api_key = OPENAI_API_KEY

# Pinecone 인스턴스 생성
pc = Pinecone(api_key=PINECONE_API_KEY)

# env 문자열을 region / cloud 로 분해 (us-east-1-aws → us-east-1 + aws)
*region, cloud = PINECONE_ENV.split("-")
REGION = "-".join(region)   # us-east-1
CLOUD  = cloud              # aws

INDEX  = "demo-index"
DIM    = 1536

# ──────── 인덱스 생성 / 재사용 ────────
def wait_ready(name):
    while True:
        state = pc.describe_index(name).status.state
        if state == "Ready":
            return
        log.info("   ↳ index status = %s … 대기 중", state)
        time.sleep(2)

if INDEX not in pc.list_indexes().names():
    log.info("새 인덱스 생성: %s", INDEX)
    pc.create_index(
        name=INDEX,
        dimension=DIM,
        metric="cosine",
        spec=ServerlessSpec(cloud=CLOUD, region=REGION),
    )
    wait_ready(INDEX)
else:
    log.info("기존 인덱스 재사용: %s", INDEX)

index = pc.Index(INDEX)

# ──────── 데이터 업서트 ────────
DOCS = [
    "쿠팡은 한국 최대의 전자상거래 기업이다.",
    "파인콘은 벡터 데이터베이스 서비스다.",
    "오픈AI는 GPT-4o 모델을 발표했다.",
    "서울의 여름은 덥고 습하다.",
    "벡터 검색은 의미 기반 유사도를 계산한다.",
]

def embed(texts):
    resp = openai.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
    )
    return [d.embedding for d in resp.data]

log.info("문서 %d개 임베딩 → Pinecone 업서트", len(DOCS))
vecs = embed(DOCS)
index.upsert(
    vectors=[(f"id-{i}", v, {"text": DOCS[i]}) for i, v in enumerate(vecs)]
)

# ──────── 질의 ────────
QUESTION = "유사도 검색을 위한 데이터베이스"
log.info("쿼리: “%s”", QUESTION)
q_vec = embed([QUESTION])[0]

res = index.query(vector=q_vec, top_k=3, include_metadata=True)
log.info("결과 (Top-3):")
for rnk, m in enumerate(res.matches, 1):
    log.info(" %d. %s (score=%.4f)", rnk, m.metadata["text"], m.score)

log.info("완료 ✅")