d02 - 07. embeddings firesearch retrieval - 02. files vector stores

02. Files와 Vector Stores 직접 호출 v2

이 노트북은 샘플 정책 문서를 만들고 Files API와 Vector Stores API를 직접 호출해 검색 인덱스를 구성합니다. 생성된 vector_store_id는 다음 노트북들이 재사용합니다.

# 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

 

핵심 개념

• 벡터 스토어(Vector Store)는 임베딩된 문서 조각을 검색 가능한 형태로 저장하는 공간입니다.
• 파일 업로드와 벡터 스토어 연결은 별도 단계이므로, 업로드 성공과 인덱싱 완료를 모두 확인해야 실제 검색에 사용할 수 있습니다.
• 파일에 attributes를 붙여 두면 나중에 메타데이터 필터 검색에 활용할 수 있습니다.

 

API 소개

• client.vector_stores.create(...)는 새 벡터 스토어를 만듭니다.
• client.files.create(...)는 문서 파일을 업로드합니다.
• client.vector_stores.files.create(...)는 업로드한 파일을 특정 벡터 스토어에 연결합니다.
• 이 노트북은 생성, 업로드/연결, 상태 점검을 차례로 보여줍니다.

 

API 호출부와 주요 매개변수

vector_store = client.vector_stores.create(
    name="workshop-docs",
    expires_after={"anchor": "last_active_at", "days": 7},
)

uploaded = client.files.create(
    file=open(file_path, "rb"),
    purpose="assistants",
)

attachment = client.vector_stores.files.create(
    vector_store_id=vector_store.id,
    file_id=uploaded.id,
    attributes={"category": "policy", "version": "v1"},
)

• name
  • 벡터 스토어를 식별하기 쉬운 이름입니다.
  • 프로젝트명, 실습명, 데이터셋 성격을 함께 넣어 두면 관리가 쉬워집니다.
  • 기본값: 실질적인 기본값은 없으므로 명시하는 편이 좋습니다.
• expires_after
  • 사용하지 않는 스토어를 자동 정리하기 위한 만료 정책입니다.
  • anchor에는 보통 last_active_at 같은 기준을 두고, days에는 유지 일수를 정수로 넣습니다.
  • 기본값: 생략하면 별도 만료 정책이 없는 리소스로 남길 수 있으므로 실습에서는 명시가 더 안전합니다.
• file
  • 업로드할 문서 파일 객체입니다.
  • 보통 텍스트, PDF, 문서 파일을 바이너리 모드로 열어 전달합니다.
  • 기본값: 없습니다.
• purpose
  • 업로드 파일의 사용 목적입니다.
  • 이 노트북에서는 벡터 스토어와 연계되는 파일 업로드 흐름을 보여 주기 위해 assistants를 사용합니다.
  • 기본값: 없습니다. 명시가 필요합니다.
• vector_store_id
  • 어떤 스토어에 파일을 연결할지 지정하는 ID입니다.
  • 기본값: 없습니다.
• file_id
  • Files API가 반환한 업로드 파일 식별자입니다.
  • 기본값: 없습니다.
• attributes
  • 검색 필터에 사용할 메타데이터 객체입니다.
  • 예를 들어 category, department, version, filename 같은 키를 둘 수 있습니다.
  • 기본값: 생략하면 별도 메타데이터 없이 인덱싱합니다.

 

요청/응답 필드에서 볼 것

• 이 실습에서는 생성 직후 vector_store.id를 가장 먼저 기록해 두는 것이 중요합니다. 이후 모든 검색과 첨부 작업이 이 값에 의존합니다.
• 파일 업로드 단계에서는 uploaded.id와 원본 파일명을 함께 저장해 두어야, 어떤 로컬 파일이 어떤 OpenAI 파일 ID로 올라갔는지 추적할 수 있습니다.
• 연결 뒤에는 상태를 한 번만 보지 말고 in_progress, completed, failed, cancelled 중 어디에 머무는지 polling 결과를 기록해 두는 것이 좋습니다.
• 검색이 비어 나오면 가장 먼저 인덱싱 상태가 정말 completed였는지 확인하고, 그다음 attributes가 기대한 대로 저장됐는지 점검해 보세요.
• 설정 파일에는 vector_store_id, 업로드 파일 ID, 벡터 스토어 파일 ID, 메타데이터를 함께 남겨 두는 방식이 다음 노트북과 연결하기 가장 쉽습니다.

 

1단계. 실행 환경 준비

벡터 스토어는 OpenAI 쪽에 생성되는 리소스이므로 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}")

검색 실습 폴더: 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

 

2단계. 샘플 문서 작성

검색 실습에는 일관된 문서가 필요합니다. 노트북 안에서 작은 정책 문서를 직접 만들면 외부 파일 의존 없이 인덱싱을 반복 검증할 수 있습니다.

# 샘플 문서 사양을 리스트로 준비합니다.
documents = [
    {
        "filename": "02-02_hr_vacation_policy.txt",
        "department": "HR",
        "category": "vacation",
        "text": "연차 휴가는 사내 포털에서 최소 3일 전에 신청합니다. 긴급 휴가는 팀장 승인 후 당일 신청할 수 있습니다.",
    },
    {
        "filename": "02-02_it_remote_work_policy.txt",
        "department": "IT",
        "category": "remote_work",
        "text": "재택근무는 보안 교육을 이수하고 VPN 접근 권한을 받은 직원만 신청할 수 있습니다. 월 최대 8일까지 사용할 수 있습니다.",
    },
    {
        "filename": "02-02_finance_expense_policy.txt",
        "department": "Finance",
        "category": "expense",
        "text": "경비 처리는 결제일로부터 14일 이내에 영수증과 사용 목적을 첨부해 제출해야 합니다.",
    },
]
# 작성한 파일 정보를 담을 리스트를 만듭니다.
local_files = []
# 문서 사양을 하나씩 순회합니다.
for spec in documents:
    # 샘플 파일 경로를 만듭니다.
    path = SAMPLE_DIR / spec["filename"]
    # 문서 내용을 UTF-8 텍스트로 저장합니다.
    path.write_text(spec["text"] + "\n", encoding="utf-8")
    # 파일 경로와 속성을 리스트에 기록합니다.
    local_files.append(
        {
            "path": path,
            "department": spec["department"],
            "category": spec["category"],
        }
    )
# 생성한 파일 목록을 출력합니다.
print([workspace_path(item["path"]) for item in local_files])

['07.embeddings_firesearch_retrieval/samples/v2/02-02_hr_vacation_policy.txt', '07.embeddings_firesearch_retrieval/samples/v2/02-02_it_remote_work_policy.txt', '07.embeddings_firesearch_retrieval/samples/v2/02-02_finance_expense_policy.txt']

 

 

3단계. Vector Store 생성 또는 재사용

매번 새 벡터 스토어를 만들면 관리되지 않는 리소스가 늘어납니다. 설정 파일이 있으면 기존 ID를 재사용하고, 없을 때만 새로 생성합니다.

# 설정 파일이 있는지 확인합니다.
if CONFIG_PATH.exists():
    # 기존 설정 JSON을 읽습니다.
    config = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
    # 기존 벡터 스토어 ID를 가져옵니다.
    vector_store_id = config["vector_store_id"]
    # 기존 벡터 스토어 정보를 OpenAI API에서 조회합니다.
    vector_store = client.vector_stores.retrieve(vector_store_id)
# 설정 파일이 없으면 새 벡터 스토어를 만듭니다.
else:
    # Vector Stores API를 직접 호출해 새 검색 인덱스를 만듭니다.
    vector_store = client.vector_stores.create(
        # 관리하기 쉬운 이름을 지정합니다.
        name="AS.1.1 v2 policy search lab",
        # 실습 리소스가 오래 남지 않도록 만료 정책을 지정합니다.
        expires_after={"anchor": "last_active_at", "days": 7},
    )
    # 새 벡터 스토어 ID를 저장합니다.
    vector_store_id = vector_store.id
    # 설정 dict를 구성합니다.
    config = {
        "vector_store_id": vector_store_id,
        "file_ids": [],
        "files": [],
    }
# 현재 벡터 스토어 ID를 출력합니다.
print(f"vector_store_id: {vector_store_id}")

vector_store_id: vs_6a1dcfe656dc8191a36c70645fd11414

 

 

4단계. 파일 업로드, 첨부, 인덱싱 확인

문서는 먼저 Files API로 업로드하고, 그 file_id를 Vector Store에 첨부합니다. 첨부 후에는 백그라운드 인덱싱 상태를 직접 polling해서 completed가 될 때까지 확인합니다.

# 이미 설정에 기록된 파일명을 set으로 만듭니다.
known_filenames = {item.get("filename") for item in config.get("files", [])}
# 로컬 샘플 파일을 하나씩 순회합니다.
for item in local_files:
    # 파일 이름을 읽습니다.
    filename = item["path"].name
    # 이미 업로드 기록이 있으면 중복 업로드를 피합니다.
    if filename in known_filenames:
        # 다음 파일로 넘어갑니다.
        continue
    # 샘플 파일을 바이너리 읽기 모드로 엽니다.
    with item["path"].open("rb") as file_handle:
        # Files API를 직접 호출해 파일을 업로드합니다.
        uploaded_file = client.files.create(file=file_handle, purpose="assistants")
    # Vector Store Files API를 직접 호출해 업로드 파일을 검색 인덱스에 첨부합니다.
    vector_file = client.vector_stores.files.create(
        # 대상 벡터 스토어 ID입니다.
        vector_store_id=vector_store_id,
        # 업로드된 파일 ID입니다.
        file_id=uploaded_file.id,
        # 검색 필터에 쓸 속성 metadata입니다.
        attributes={
            # 어느 부서 문서인지 기록합니다.
            "department": item["department"],
            # 어떤 카테고리 문서인지 기록합니다.
            "category": item["category"],
            # 원본 파일명을 함께 기록합니다.
            "filename": filename,
        },
    )
    # 설정에 파일 정보를 추가합니다.
    config.setdefault("files", []).append(
        {
            # 나중에 중복 업로드를 피하기 위한 파일명입니다.
            "filename": filename,
            # Files API가 돌려준 업로드 파일 ID입니다.
            "file_id": uploaded_file.id,
            # Vector Store에 연결된 파일 항목 ID입니다.
            "vector_store_file_id": vector_file.id,
            # 검색 필터용 부서 속성입니다.
            "department": item["department"],
            # 검색 필터용 카테고리 속성입니다.
            "category": item["category"],
        }
    )
# 인덱싱 상태를 담을 리스트를 만듭니다.
statuses = []
# 설정에 기록된 파일을 하나씩 확인합니다.
for item in config.get("files", []):
    # 최대 60초 동안 polling합니다.
    for attempt in range(30):
        # 벡터 스토어 파일 상태를 조회합니다.
        current = client.vector_stores.files.retrieve(
            vector_store_id=vector_store_id,
            file_id=item["file_id"],
        )
        # 현재 상태를 읽습니다.
        status = current.status
        # 상태가 completed이면 대기를 멈춥니다.
        if status == "completed":
            break
        # 실패 상태이면 즉시 중단합니다.
        if status in {"failed", "cancelled"}:
            raise RuntimeError(f"인덱싱 실패: {item['filename']} status={status}")
        # 아직 진행 중이면 2초 기다립니다.
        time.sleep(2)
    # 최종 상태를 기록합니다.
    statuses.append(
        {
            # 어떤 파일의 상태인지 기록합니다.
            "filename": item["filename"],
            # polling 종료 시점의 최종 상태를 기록합니다.
            "status": status,
        }
    )
# 설정 파일에 벡터 스토어 ID와 파일 목록을 저장합니다.
CONFIG_PATH.write_text(json.dumps(config, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
# 인덱싱 상태를 JSON으로 저장합니다.
(OUTPUT_DIR / "02-04_vector_store_index_status.json").write_text(
    json.dumps(statuses, ensure_ascii=False, indent=2) + "\n",
    encoding="utf-8",
)
# 상태를 출력합니다.
print(json.dumps(statuses, ensure_ascii=False, indent=2))
# 설정 파일 위치를 출력합니다.
print(f"설정 저장: {workspace_path(CONFIG_PATH)}")

[
  {
    "filename": "02-02_hr_vacation_policy.txt",
    "status": "completed"
  },
  {
    "filename": "02-02_it_remote_work_policy.txt",
    "status": "completed"
  },
  {
    "filename": "02-02_finance_expense_policy.txt",
    "status": "completed"
  }
]
설정 저장: 07.embeddings_firesearch_retrieval/output/v2/02-04_vector_store_config.json