d02 - 07. embeddings firesearch retrieval - 01. embeddings

01. Embeddings 직접 호출 v2

이 노트북은 client.embeddings.create(...)를 직접 호출해 텍스트를 벡터로 바꿉니다. 단일 입력, 배치 입력, dimensions 옵션, 코사인 유사도 계산을 순서대로 확인합니다.

# 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

 

핵심 개념

• 임베딩(Embedding)은 텍스트 의미를 숫자 벡터로 바꾸는 작업입니다.
• 의미가 비슷한 문장은 벡터 공간에서도 서로 가깝게 배치되는 경향이 있습니다.
• 단일 입력과 배치 입력은 같은 API를 사용하지만, 응답 data 배열의 길이와 해석 방식이 달라집니다.
• dimensions를 줄이면 저장 공간과 계산 비용을 줄일 수 있지만, 표현력이 일부 줄어들 수 있습니다.

 

API 소개

• client.embeddings.create(...)는 텍스트를 벡터로 변환하는 전용 API입니다.
• 이 노트북은 단일 임베딩, 배치 임베딩, 차원 축소, 코사인 유사도 계산을 순서대로 다룹니다.

 

API 호출부와 주요 매개변수

response = client.embeddings.create(
    model="text-embedding-3-small",
    input=["문장 A", "문장 B"],
    encoding_format="float",
    dimensions=512,
)

• model
  • 보통 text-embedding-3-small, text-embedding-3-large 같은 임베딩 모델을 넣습니다.
  • 기본값: 실질적인 기본값이 없으므로 명시하는 것이 안전합니다.
• input
  • 문자열 하나 또는 문자열 리스트를 넣습니다.
  • 배치 입력이면 응답 data[] 길이도 입력 개수만큼 늘어납니다.
  • 기본값: 없습니다.
• encoding_format
  • 실습에서는 보통 float를 사용해 바로 Python 숫자 배열로 받습니다.
  • 저장 효율이나 전송 최적화를 위해 다른 인코딩 표현을 고려할 수 있지만, 학습용으로는 float가 가장 읽기 쉽습니다.
  • 기본값: SDK와 사용 방식에 따라 기대 형식이 달라질 수 있으므로 이 노트북처럼 명시하는 것이 안전합니다.
• dimensions
  • 기본 임베딩 길이보다 작은 양의 정수를 넣어 벡터 길이를 줄일 수 있습니다.
  • 최신 문서 기준 text-embedding-3-small의 기본 길이는 1536, text-embedding-3-large는 3072입니다.
  • 기본값: 생략하면 모델의 기본 전체 차원을 사용합니다.

 

요청/응답 필드에서 볼 것

• 임베딩 실습에서는 먼저 요청의 input 개수를 보고, 이번 호출이 단일 문장용인지 배치 비교용인지부터 구분해 두면 응답 해석이 쉬워집니다.
• 응답을 볼 때는 data[n].embedding 전체를 다 보려 하기보다 길이와 앞부분 몇 개 값만 미리보기로 저장해 두는 방식이 가장 실습 친화적입니다.
• 차원 축소를 비교할 때는 len(embedding) 결과를 기본 차원과 나란히 적어 보고, 저장 비용과 검색 비용이 왜 줄어드는지 연결해서 이해해 보세요.
• 결과가 기대와 다르면 model, dimensions, encoding_format이 요청에서 실제로 어떤 값이었는지 다시 보는 것이 가장 빠른 점검 방법입니다.
• 이후 검색 노트북으로 넘어갈 때는 질의 벡터와 문서 벡터를 어떻게 재사용할지 생각하면서, usage와 결과 JSON을 함께 남겨 두면 흐름이 자연스럽게 이어집니다.

 

1단계. 실행 환경 준비

임베딩은 이후 검색과 RAG의 바탕이 됩니다. 먼저 모델명과 결과 저장 위치를 준비합니다.

# 운영체제 환경 변수를 읽기 위해 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단계. 단일 문장 임베딩

한 문장이 하나의 숫자 벡터로 바뀌는 가장 기본 흐름입니다. 응답의 data[0].embedding을 직접 확인합니다.

# 임베딩할 단일 문장을 준비합니다.
text = "OpenAI API 교육 과정에서는 직접 API 요청과 응답 구조를 이해하는 것이 중요합니다."
# 임베딩 API를 직접 호출합니다.
response = client.embeddings.create(
    # 사용할 임베딩 모델을 지정합니다.
    model=EMBEDDING_MODEL,
    # 임베딩할 입력 텍스트를 전달합니다.
    input=text,
    # Python float 배열로 받기 위해 인코딩 형식을 지정합니다.
    encoding_format="float",
)
# 첫 번째 결과의 벡터를 꺼냅니다.
embedding = response.data[0].embedding
# 벡터 차원을 계산합니다.
dimension = len(embedding)
# 앞쪽 일부 값을 미리보기로 저장합니다.
preview = embedding[:5]
# 결과 요약을 구성합니다.
summary = {
    "text": text,
    "dimension": dimension,
    "preview": preview,
}
# 결과를 JSON으로 저장합니다.
(OUTPUT_DIR / "01-02_embedding_basic.json").write_text(
    json.dumps(summary, ensure_ascii=False, indent=2) + "\n",
    encoding="utf-8",
)
# 결과 요약을 출력합니다.
print(json.dumps(summary, ensure_ascii=False, indent=2))

{
  "text": "OpenAI API 교육 과정에서는 직접 API 요청과 응답 구조를 이해하는 것이 중요합니다.",
  "dimension": 1536,
  "preview": [
    -0.035980224609375,
    -0.0026226043701171875,
    0.0009775161743164062,
    -0.0026531219482421875,
    0.051422119140625
  ]
}

 

3단계. 배치 입력과 dimensions 옵션

여러 문장을 한 번에 보내면 API 호출 수를 줄일 수 있습니다. dimensions는 벡터 길이를 줄여 저장 비용과 검색 비용을 조절할 때 사용합니다.

# 배치로 임베딩할 문장 목록을 준비합니다.
texts = [
    "휴가 신청은 사내 포털에서 진행합니다.",
    "재택근무는 보안 교육 이수 후 신청할 수 있습니다.",
    "경비 처리는 영수증 첨부가 필요합니다.",
]
# 기본 차원 배치 임베딩을 요청합니다.
batch_response = client.embeddings.create(
    # 임베딩 모델을 지정합니다.
    model=EMBEDDING_MODEL,
    # 여러 텍스트를 리스트로 전달합니다.
    input=texts,
    # float 배열로 결과를 받습니다.
    encoding_format="float",
)
# 512차원 축소 임베딩을 요청합니다.
small_response = client.embeddings.create(
    # 같은 임베딩 모델을 사용합니다.
    model=EMBEDDING_MODEL,
    # 같은 입력을 사용해 차원 차이만 비교합니다.
    input=texts,
    # float 배열로 결과를 받습니다.
    encoding_format="float",
    # 벡터 차원을 512로 줄입니다.
    dimensions=512,
)
# 차원 비교 결과를 구성합니다.
comparison = {
    "default_dimension": len(batch_response.data[0].embedding),
    "reduced_dimension": len(small_response.data[0].embedding),
    "item_count": len(texts),
}
# 비교 결과를 JSON으로 저장합니다.
(OUTPUT_DIR / "01-03_embedding_dimensions.json").write_text(
    json.dumps(comparison, ensure_ascii=False, indent=2) + "\n",
    encoding="utf-8",
)
# 비교 결과를 출력합니다.
print(json.dumps(comparison, ensure_ascii=False, indent=2))

{
  "default_dimension": 1536,
  "reduced_dimension": 512,
  "item_count": 3
}

 

4단계. 코사인 유사도 검색 미리보기

벡터 검색의 직관을 얻기 위해 작은 문서 목록과 질문을 같은 임베딩 공간에 넣고, 코사인 유사도로 직접 순위를 계산합니다.

# 검색 질문을 준비합니다.
query = "재택근무를 신청하려면 무엇이 필요한가요?"
# 질문과 문서를 한 번에 임베딩하기 위해 리스트를 구성합니다.
all_inputs = [query] + texts
# 질문과 문서 임베딩을 요청합니다.
sim_response = client.embeddings.create(
    # 임베딩 모델을 지정합니다.
    model=EMBEDDING_MODEL,
    # 질문과 문서를 한 번에 전달합니다.
    input=all_inputs,
    # float 배열로 결과를 받습니다.
    encoding_format="float",
)
# 첫 번째 벡터를 질문 벡터로 사용합니다.
query_vector = sim_response.data[0].embedding
# 나머지 벡터를 문서 벡터로 사용합니다.
document_vectors = [item.embedding for item in sim_response.data[1:]]
# 문서별 유사도 결과를 담을 리스트를 만듭니다.
ranked = []
# 문서 텍스트와 벡터를 함께 순회합니다.
for document, vector in zip(texts, document_vectors):
    # 질문 벡터와 문서 벡터의 코사인 유사도를 계산합니다.
    score = cosine_similarity(query_vector, vector)
    # 결과 리스트에 문서와 점수를 추가합니다.
    ranked.append(
        {
            "document": document,
            "score": round(score, 4),
        }
    )
# 점수가 높은 순서로 정렬합니다.
ranked.sort(key=lambda item: item["score"], reverse=True)
# 순위 결과를 JSON으로 저장합니다.
(OUTPUT_DIR / "01-04_embedding_similarity_preview.json").write_text(
    json.dumps({"query": query, "ranked": ranked}, ensure_ascii=False, indent=2)
    + "\n",
    encoding="utf-8",
)
# 순위 결과를 출력합니다.
print(json.dumps(ranked, ensure_ascii=False, indent=2))

[
  {
    "document": "재택근무는 보안 교육 이수 후 신청할 수 있습니다.",
    "score": 0.677
  },
  {
    "document": "경비 처리는 영수증 첨부가 필요합니다.",
    "score": 0.3023
  },
  {
    "document": "휴가 신청은 사내 포털에서 진행합니다.",
    "score": 0.2987
  }
]