d02 - 07. embeddings firesearch retrieval - 03. retrieval v2

03. Vector Store Retrieval 직접 호출 v2

이 노트북은 이전 단계에서 만든 vector_store_id를 읽고 client.vector_stores.search(...)를 직접 호출합니다. 기본 검색, 필터 검색, 임베딩 기반 재정렬을 순서대로 확인합니다.

# Colab 또는 로컬 노트북 실행 환경을 구분하기 위해 sys를 가져옵니다.
import sys
# 패키지 설치 명령을 현재 Python 커널에서 실행하기 위해 subprocess를 가져옵니다.
import subprocess
# 패키지 설치 여부를 확인하기 위해 importlib.util을 가져옵니다.
import importlib.util
# 환경 변수에서 API 키를 읽고 설정하기 위해 os를 가져옵니다.
import os
# API 키를 화면에 노출하지 않고 입력받기 위해 getpass를 가져옵니다.
import getpass
# .env 파일 위치를 다루기 위해 pathlib의 Path를 가져옵니다.
from pathlib import Path

# google.colab 모듈이 있으면 현재 런타임이 Google Colab이라고 판단합니다.
IN_COLAB = "google.colab" in sys.modules
# 노트북에서 사용하는 import 이름과 pip 패키지 이름을 짝지어 둡니다.
REQUIRED_PACKAGES = {
    "openai": "openai>=2.26.0",
    "dotenv": "python-dotenv>=1.2.2",
    "pydantic": "pydantic>=2.11.0",
    "PIL": "pillow>=12.1.1",
    "requests": "requests>=2.32.5",
    "numpy": "numpy>=2.3.3",
    "websockets": "websockets>=15.0.1",
    "websocket": "websocket-client>=1.8.0",
    "nest_asyncio": "nest-asyncio>=1.6.0",
}

def ensure_package(import_name: str, package_name: str) -> None:
    if importlib.util.find_spec(import_name) is not None:
        return
    subprocess.check_call(
        [sys.executable, "-m", "pip", "install", "-q", package_name]
    )

if IN_COLAB:
    for import_name, package_name in REQUIRED_PACKAGES.items():
        ensure_package(import_name, package_name)

from dotenv import load_dotenv

load_dotenv(Path.cwd() / ".env")
if not os.getenv("OPENAI_API_KEY"):
    secret_key = None
    try:
        from google.colab import userdata

        secret_key = userdata.get("OPENAI_API_KEY")
    except Exception:
        secret_key = None
    if secret_key:
        os.environ["OPENAI_API_KEY"] = secret_key
    else:
        entered_key = getpass.getpass("OPENAI_API_KEY를 입력하세요: ").strip()
        if entered_key:
            os.environ["OPENAI_API_KEY"] = entered_key

if not os.getenv("OPENAI_API_KEY"):
    raise RuntimeError(
        "OPENAI_API_KEY가 없습니다. Colab Secrets, 수동 입력, 또는 .env 파일로 설정해 주세요."
    )

print(f"개발환경: {'Colab' if IN_COLAB else '로컬'}")
print("OPENAI_API_KEY 준비 완료")

개발환경: 로컬
OPENAI_API_KEY 준비 완료

 

실습 전에 보는 핵심 개념과 API

 

핵심 개념

• Retrieval(검색)은 벡터 스토어 안에서 질의와 의미가 비슷한 문서 조각을 찾는 단계입니다.
• 단순 검색 결과를 그대로 쓰는 것보다, 메타데이터 필터나 재랭킹을 추가하면 더 정밀한 결과를 만들 수 있습니다.
• 하이브리드 재랭킹은 서버가 준 검색 점수와 로컬에서 계산한 임베딩 유사도를 함께 반영하는 방식입니다.

 

API 소개

• client.vector_stores.search(...)는 벡터 스토어를 직접 검색하는 API입니다.
• client.embeddings.create(...)는 검색 결과를 다시 임베딩해 재랭킹할 때 보조적으로 사용합니다.
• 이 노트북은 기본 검색, 필터 검색, 하이브리드 재랭킹을 비교합니다.

 

API 호출부와 주요 매개변수

search_page = client.vector_stores.search(
    vector_store_id=vector_store_id,
    query="재택근무 신청 절차를 알려 주세요.",
    max_num_results=5,
    rewrite_query=True,
    filters={"type": "eq", "key": "category", "value": "policy"},
)

• vector_store_id
  • 검색 대상 스토어입니다.
  • 기본값: 없습니다.
• query
  • 사용자의 자연어 검색 질문입니다.
  • 검색 품질은 질의 문장 표현에 크게 영향을 받습니다.
  • 기본값: 없습니다.
• max_num_results
  • 최대 반환 개수입니다.
  • 실습에서는 보통 3, 5, 10처럼 작은 수로 결과를 읽기 쉽게 제한합니다.
  • 기본값: 서비스가 내부 기본 개수를 사용할 수 있으므로, 학습용 비교에서는 명시하는 것이 안전합니다.
• rewrite_query
  • 가능한 값은 true, false입니다.
  • true이면 서버가 검색에 더 유리한 형태로 질의를 보정할 수 있습니다.
  • 기본값: 생략하면 원래 질의를 그대로 사용하는 흐름으로 이해하는 편이 안전합니다.
• filters
  • 메타데이터 조건으로 검색 범위를 제한하는 객체입니다.
  • 이 노트북처럼 {"type": "eq", "key": ..., "value": ...} 구조를 자주 사용합니다.
  • 기본값: 생략하면 전체 스토어를 대상으로 검색합니다.

 

요청/응답 필드에서 볼 것

• 검색 실습에서는 요청의 query, filters, max_num_results, rewrite_query를 먼저 읽고, 왜 이런 결과 개수와 범위가 나와야 하는지 먼저 예상해 보는 것이 좋습니다.
• 응답의 data[]를 볼 때는 각 결과에서 파일명, 점수, 본문 일부를 함께 정리해 두어야 검색이 잘 맞았는지 빠르게 판단할 수 있습니다.
• 필터 검색이 기대와 다르면 filters 키와 값이 업로드 때 저장한 attributes와 정확히 일치하는지 먼저 확인해 보세요.
• 재랭킹 단계에서는 서버가 준 점수와 로컬에서 다시 계산한 코사인 유사도를 나란히 저장해 두면 순위가 왜 바뀌었는지 설명하기 쉽습니다.
• 실습 기록에는 질의 문장, 적용한 필터, 반환 결과 수, 상위 결과 JSON을 함께 남겨 두는 방식이 가장 재현성이 좋습니다.

 

1단계. 실행 환경 준비

검색 단계는 이미 만들어진 벡터 스토어 ID가 필요합니다. 설정 파일을 읽어 어떤 인덱스를 검색할지 명확히 합니다.

# 운영체제 환경 변수를 읽기 위해 os 모듈을 가져옵니다.
import os
# JSON 설정과 결과를 저장하기 위해 json 모듈을 가져옵니다.
import json
# 실행 시간과 인덱싱 대기 시간을 다루기 위해 time 모듈을 가져옵니다.
import time
# 벡터 유사도 계산에 사용할 수학 함수를 가져옵니다.
import math
# 파일 경로를 안전하게 다루기 위해 Path를 가져옵니다.
from pathlib import Path
# 타입 힌트에서 유연한 응답 구조를 표현하기 위해 Any를 가져옵니다.
from typing import Any
# .env 파일의 API 키를 로드하기 위해 load_dotenv를 가져옵니다.
from dotenv import load_dotenv
# OpenAI API를 직접 호출하기 위해 OpenAI 클라이언트를 가져옵니다.
from openai import OpenAI
# 현재 실행 위치를 저장합니다.
CURRENT_DIR = Path.cwd()
# 루트에서 실행하는 경우 retrieval 트랙 폴더를 찾습니다.
if (CURRENT_DIR / "07.embeddings_firesearch_retrieval").exists():
    # 워크스페이스 루트에서 실행 중이면 해당 폴더를 실습 루트로 사용합니다.
    RETRIEVAL_ROOT = CURRENT_DIR / "07.embeddings_firesearch_retrieval"
# 하위 폴더에서 실행하는 경우 경로 조각에서 트랙 폴더를 찾습니다.
elif "07.embeddings_firesearch_retrieval" in [part for part in CURRENT_DIR.parts]:
    # 트랙 폴더까지의 경로를 복원합니다.
    RETRIEVAL_ROOT = Path(*CURRENT_DIR.parts[: CURRENT_DIR.parts.index("07.embeddings_firesearch_retrieval") + 1])
# 그 밖의 경우 상대 경로를 사용합니다.
else:
    # 현재 위치 기준 트랙 폴더를 실습 루트로 사용합니다.
    RETRIEVAL_ROOT = CURRENT_DIR / "07.embeddings_firesearch_retrieval"
# 워크스페이스 루트는 retrieval 트랙의 상위 폴더입니다.
WORKSPACE_ROOT = RETRIEVAL_ROOT.parent
# 결과 저장 폴더를 정합니다.
OUTPUT_DIR = RETRIEVAL_ROOT / "output" / "v2"
# 결과 저장 폴더를 생성합니다.
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
# 샘플 문서 폴더를 정합니다.
SAMPLE_DIR = RETRIEVAL_ROOT / "samples" / "v2"
# 샘플 문서 폴더를 생성합니다.
SAMPLE_DIR.mkdir(parents=True, exist_ok=True)
# 벡터 스토어 설정 파일 경로를 정합니다.
CONFIG_PATH = OUTPUT_DIR / "02-04_vector_store_config.json"
# .env 파일을 읽습니다.
load_dotenv(WORKSPACE_ROOT / ".env")
# API 키가 없으면 명확한 오류로 중단합니다.
if not os.getenv("OPENAI_API_KEY"):
    # 필요한 환경 변수 이름을 포함해 안내합니다.
    raise RuntimeError("OPENAI_API_KEY가 없습니다. 워크스페이스 루트의 .env 파일을 확인해 주세요.")
# OpenAI API 클라이언트를 생성합니다.
client = OpenAI()
# 임베딩 모델명을 환경 변수에서 읽습니다.
EMBEDDING_MODEL = os.getenv("OPENAI_EMBEDDING_MODEL", "text-embedding-3-small")
# Responses API 모델명을 환경 변수에서 읽습니다.
RESPONSES_MODEL = os.getenv("OPENAI_RESPONSES_MODEL", "gpt-5-mini")
# 워크스페이스 기준 상대 경로를 반환하는 함수를 정의합니다.
def workspace_path(path: Path) -> str:
    # 절대 경로를 워크스페이스 기준 / 구분 문자열로 변환합니다.
    return path.resolve().relative_to(WORKSPACE_ROOT.resolve()).as_posix()
# 코사인 유사도를 계산하는 함수를 정의합니다.
def cosine_similarity(left: list[float], right: list[float]) -> float:
    # 두 벡터의 내적을 계산합니다.
    dot = sum(a * b for a, b in zip(left, right))
    # 왼쪽 벡터의 길이를 계산합니다.
    left_norm = math.sqrt(sum(a * a for a in left))
    # 오른쪽 벡터의 길이를 계산합니다.
    right_norm = math.sqrt(sum(b * b for b in right))
    # 0으로 나누는 상황을 피하기 위해 길이를 확인합니다.
    if left_norm == 0 or right_norm == 0:
        # 길이가 0이면 유사도를 0으로 처리합니다.
        return 0.0
    # 내적을 두 벡터 길이의 곱으로 나누어 코사인 유사도를 반환합니다.
    return dot / (left_norm * right_norm)
# 실습 루트와 모델 정보를 출력합니다.
print(f"검색 실습 폴더: {RETRIEVAL_ROOT}")
# 결과 저장 위치를 출력합니다.
print(f"결과 저장 폴더: {OUTPUT_DIR}")
# 임베딩 모델을 출력합니다.
print(f"임베딩 모델: {EMBEDDING_MODEL}")
# Responses 모델을 출력합니다.
print(f"Responses 모델: {RESPONSES_MODEL}")

# 설정 파일이 없으면 앞 단계 노트북을 먼저 실행해야 하므로 중단합니다.
if not CONFIG_PATH.exists():
    # 필요한 선행 작업을 오류 메시지로 안내합니다.
    raise RuntimeError("02-04_vector_store_config.json이 없습니다. 02.files_vector_stores/files_vector_stores_v2.ipynb를 먼저 실행해 주세요.")
# 벡터 스토어 설정 파일을 읽습니다.
config = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
# 검색에 사용할 벡터 스토어 ID를 꺼냅니다.
vector_store_id = config["vector_store_id"]
# 벡터 스토어 ID를 출력합니다.
print(f"vector_store_id: {vector_store_id}")

검색 실습 폴더: c:\Users\ai4nu\2026-ai-tech\AS.1.1_0601\07.embeddings_firesearch_retrieval
결과 저장 폴더: c:\Users\ai4nu\2026-ai-tech\AS.1.1_0601\07.embeddings_firesearch_retrieval\output\v2
임베딩 모델: text-embedding-3-small
Responses 모델: gpt-5-mini
vector_store_id: vs_6a1dcfe656dc8191a36c70645fd11414

 

 

2단계. 기본 vector store search

기본 검색은 질문과 가장 가까운 문서 조각을 반환합니다. max_num_results로 결과 수를 직접 제한합니다.

# 검색 질문을 준비합니다.
query = "재택근무를 신청하려면 어떤 조건이 필요한가요?"
# Vector Store Search API를 직접 호출합니다.
search_results = client.vector_stores.search(
    # 검색할 벡터 스토어 ID를 전달합니다.
    vector_store_id=vector_store_id,
    # 자연어 검색 질문을 전달합니다.
    query=query,
    # 상위 5개 결과만 받습니다.
    max_num_results=5,
)
# 정규화된 결과를 담을 리스트를 만듭니다.
normalized = []
# 검색 결과 data 배열을 순회합니다.
for result in search_results.data:
    # 결과의 텍스트 조각을 합칩니다.
    content_text = "\n".join(
        part.text for part in result.content if hasattr(part, "text")
    )
    # 파일명, 점수, 내용을 dict로 저장합니다.
    normalized.append(
        {
            "filename": result.filename,
            "score": result.score,
            "content": content_text,
        }
    )
# 검색 결과를 JSON으로 저장합니다.
(OUTPUT_DIR / "03-02_retrieval_basic_search.json").write_text(
    json.dumps({"query": query, "results": normalized}, ensure_ascii=False, indent=2)
    + "\n",
    encoding="utf-8",
)
# 검색 결과를 출력합니다.
print(json.dumps(normalized, ensure_ascii=False, indent=2))

[
  {
    "filename": "02-02_it_remote_work_policy.txt",
    "score": 0.5285193440349892,
    "content": "재택근무는 보안 교육을 이수하고 VPN 접근 권한을 받은 직원만 신청할 수 있습니다. 월 최대 8일까지 사용할 수 있습니다."
  },
  {
    "filename": "02-02_hr_vacation_policy.txt",
    "score": 0.4497045810056439,
    "content": "연차 휴가는 사내 포털에서 최소 3일 전에 신청합니다. 긴급 휴가는 팀장 승인 후 당일 신청할 수 있습니다."
  },
  {
    "filename": "02-02_finance_expense_policy.txt",
    "score": 0.32825904935539557,
    "content": "경비 처리는 결제일로부터 14일 이내에 영수증과 사용 목적을 첨부해 제출해야 합니다."
  }
]

 

 

3단계. metadata 필터 검색

파일을 첨부할 때 넣은 attributes는 검색 필터로 사용할 수 있습니다. 이 단계는 HR 문서만 검색하는 식으로 범위를 줄이는 방법을 보여 줍니다.

# 필터 검색 질문을 준비합니다.
filtered_query = "휴가 신청 절차를 알려 주세요."
# HR 부서 문서만 검색하도록 필터를 구성합니다.
filters = {"type": "eq", "key": "department", "value": "HR"}
# 필터가 포함된 Vector Store Search API를 호출합니다.
filtered_results = client.vector_stores.search(
    # 검색할 벡터 스토어 ID입니다.
    vector_store_id=vector_store_id,
    # 검색 질문입니다.
    query=filtered_query,
    # metadata 필터입니다.
    filters=filters,
    # 검색 결과 수를 제한합니다.
    max_num_results=5,
    # 검색 질문을 모델이 더 좋은 검색어로 바꿀 수 있게 허용합니다.
    rewrite_query=True,
)
# 필터 검색 결과를 담을 리스트를 만듭니다.
filtered_normalized = []
# 검색 결과를 순회합니다.
for result in filtered_results.data:
    # 결과 텍스트를 합칩니다.
    content_text = "\n".join(
        part.text for part in result.content if hasattr(part, "text")
    )
    # 결과 정보를 추가합니다.
    filtered_normalized.append(
        {
            "filename": result.filename,
            "score": result.score,
            "content": content_text,
        }
    )
# 필터 검색 결과를 저장합니다.
(OUTPUT_DIR / "03-03_retrieval_filtered_search.json").write_text(
    json.dumps(
        {
            "query": filtered_query,
            "filters": filters,
            "results": filtered_normalized,
        },
        ensure_ascii=False,
        indent=2,
    )
    + "\n",
    encoding="utf-8",
)
# 결과를 출력합니다.
print(json.dumps(filtered_normalized, ensure_ascii=False, indent=2))

[
  {
    "filename": "02-02_hr_vacation_policy.txt",
    "score": 0.5390274384606489,
    "content": "연차 휴가는 사내 포털에서 최소 3일 전에 신청합니다. 긴급 휴가는 팀장 승인 후 당일 신청할 수 있습니다."
  }
]

 

 

4단계. 임베딩 유사도 기반 재정렬

검색 API의 점수와 별도로, 가져온 결과 텍스트를 다시 임베딩해 질문과의 코사인 유사도를 계산할 수 있습니다. 이 과정을 통해 reranking의 기본 아이디어를 확인합니다.

# 재정렬에 사용할 후보 텍스트 목록을 만듭니다.
candidate_texts = [item["content"] for item in normalized if item["content"]]
# 질문과 후보가 모두 있어야 재정렬을 진행합니다.
if not candidate_texts:
    # 후보가 없으면 명확히 중단합니다.
    raise RuntimeError("재정렬할 검색 결과가 없습니다.")
# 질문과 후보 텍스트를 함께 임베딩합니다.
embedding_response = client.embeddings.create(
    # 임베딩 모델을 지정합니다.
    model=EMBEDDING_MODEL,
    # 질문을 첫 번째, 후보 문서를 뒤쪽에 배치합니다.
    input=[query] + candidate_texts,
    # float 배열로 결과를 받습니다.
    encoding_format="float",
)
# 질문 벡터를 꺼냅니다.
query_vector = embedding_response.data[0].embedding
# 후보 벡터들을 꺼냅니다.
candidate_vectors = [item.embedding for item in embedding_response.data[1:]]
# 재정렬 결과를 담을 리스트를 만듭니다.
reranked = []
# 기존 검색 결과와 후보 벡터를 함께 순회합니다.
for item, vector in zip(normalized, candidate_vectors):
    # 질문과 후보의 코사인 유사도를 계산합니다.
    rerank_score = cosine_similarity(query_vector, vector)
    # 기존 결과에 rerank 점수를 추가합니다.
    reranked.append({**item, "rerank_score": round(rerank_score, 4)})
# rerank 점수가 높은 순서로 정렬합니다.
reranked.sort(key=lambda item: item["rerank_score"], reverse=True)
# 재정렬 결과를 저장합니다.
(OUTPUT_DIR / "03-04_retrieval_hybrid_rerank.json").write_text(
    json.dumps(reranked, ensure_ascii=False, indent=2) + "\n",
    encoding="utf-8",
)
# 재정렬 결과를 출력합니다.
print(json.dumps(reranked, ensure_ascii=False, indent=2))

[
  {
    "filename": "02-02_it_remote_work_policy.txt",
    "score": 0.5285193440349892,
    "content": "재택근무는 보안 교육을 이수하고 VPN 접근 권한을 받은 직원만 신청할 수 있습니다. 월 최대 8일까지 사용할 수 있습니다.",
    "rerank_score": 0.5717
  },
  {
    "filename": "02-02_hr_vacation_policy.txt",
    "score": 0.4497045810056439,
    "content": "연차 휴가는 사내 포털에서 최소 3일 전에 신청합니다. 긴급 휴가는 팀장 승인 후 당일 신청할 수 있습니다.",
    "rerank_score": 0.3395
  },
  {
    "filename": "02-02_finance_expense_policy.txt",
    "score": 0.32825904935539557,
    "content": "경비 처리는 결제일로부터 14일 이내에 영수증과 사용 목적을 첨부해 제출해야 합니다.",
    "rerank_score": 0.2605
  }
]