d01 - 04. Web Search - 02. domain filter and sources v2

02. 도메인 필터링과 소스 추적 v2

이 노트북은 웹 검색 범위를 신뢰할 수 있는 도메인으로 제한하는 방법을 다룹니다. 의료/정책처럼 근거 품질이 중요한 주제에서는 검색 범위를 코드로 제한하는 것이 중요합니다.

# 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",
}

# 현재 커널에서 import할 수 없는 패키지만 설치하는 함수입니다.
def ensure_package(import_name: str, package_name: str) -> None:
    # 이미 import 가능한 패키지는 설치를 건너뜁니다.
    if importlib.util.find_spec(import_name) is not None:
        # 설치가 필요 없음을 호출자에게 조용히 알리고 돌아갑니다.
        return
    # Colab에서는 현재 노트북 커널의 Python에 패키지를 설치해야 합니다.
    subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", package_name])

# Colab 기본 런타임에는 일부 OpenAI 실습 패키지가 없을 수 있으므로 먼저 준비합니다.
if IN_COLAB:
    # 실습 전체에서 필요한 패키지를 하나씩 확인하고 부족한 것만 설치합니다.
    for import_name, package_name in REQUIRED_PACKAGES.items():
        # 누락된 패키지를 현재 Colab 런타임에 설치합니다.
        ensure_package(import_name, package_name)

# 패키지 설치 이후 .env 로딩 기능을 사용할 수 있도록 load_dotenv를 가져옵니다.
from dotenv import load_dotenv

# 현재 작업 폴더의 .env 파일이 있으면 로컬 실행처럼 환경 변수로 올립니다.
load_dotenv(Path.cwd() / ".env")
# OPENAI_API_KEY가 이미 있으면 그대로 사용하고, 없으면 Colab Secrets 또는 입력으로 보완합니다.
if not os.getenv("OPENAI_API_KEY"):
    # Colab Secrets에서 OPENAI_API_KEY를 읽어 볼 변수를 준비합니다.
    secret_key = None
    # Colab이 아닌 로컬 환경에서는 google.colab import가 실패할 수 있으므로 예외를 허용합니다.
    try:
        # Colab Secrets의 userdata API를 가져옵니다.
        from google.colab import userdata
        # Secrets에 저장된 OPENAI_API_KEY 값을 읽습니다.
        secret_key = userdata.get("OPENAI_API_KEY")
    except Exception:
        # Colab Secrets를 사용할 수 없으면 이후 수동 입력 단계로 넘어갑니다.
        secret_key = None
    # Secrets에서 키를 찾았다면 현재 런타임 환경 변수로 설정합니다.
    if secret_key:
        # OpenAI SDK가 자동으로 읽을 수 있도록 표준 환경 변수 이름에 저장합니다.
        os.environ["OPENAI_API_KEY"] = secret_key
    else:
        # 키가 없으면 노트북 실행자가 직접 입력하도록 요청합니다.
        entered_key = getpass.getpass("OPENAI_API_KEY를 입력하세요: ").strip()
        # 빈 문자열이 아닌 값을 입력한 경우에만 환경 변수로 등록합니다.
        if entered_key:
            # OpenAI SDK가 자동으로 읽을 수 있도록 표준 환경 변수 이름에 저장합니다.
            os.environ["OPENAI_API_KEY"] = entered_key

# API 키가 끝까지 없으면 다음 OpenAI API 호출이 실패하므로 명확한 오류를 냅니다.
if not os.getenv("OPENAI_API_KEY"):
    # Colab에서는 Secrets 또는 입력, 로컬에서는 .env 또는 환경 변수를 설정해야 함을 알려 줍니다.
    raise RuntimeError("OPENAI_API_KEY가 없습니다. Colab Secrets, 수동 입력, 또는 .env 파일로 설정해 주세요.")

# 이후 셀에서 Colab 여부를 참고할 수 있도록 간단히 출력합니다.
print(f"개발환경: {'Colab' if IN_COLAB else '로컬'}")
# API 키를 직접 출력하지 않고 준비 완료 여부만 알려 줍니다.
print("OPENAI_API_KEY 준비 완료")

 

 

1단계. 실행 환경 준비

웹 검색 실습은 모델 응답뿐 아니라 검색 호출, 검색 질의, 출처 URL을 함께 확인해야 합니다. 먼저 클라이언트, 저장 폴더, 응답 해석 함수를 준비해 이후 단계에서 검색 근거를 추적할 수 있게 합니다.

# 환경 변수와 모델명을 읽기 위해 os를 가져옵니다.
import os
# 웹 검색 호출 기록과 메타데이터를 JSON으로 저장하기 위해 json을 가져옵니다.
import json
# 오늘 날짜 문자열을 만들기 위해 datetime을 가져옵니다.
from datetime import datetime
# 파일 경로를 안전하게 다루기 위해 Path를 가져옵니다.
from pathlib import Path
# 응답 객체를 유연하게 해석하기 위해 Any 타입을 가져옵니다.
from typing import Any
# URL에서 도메인을 추출하기 위해 urlparse를 가져옵니다.
from urllib.parse import urlparse
# .env 파일의 OPENAI_API_KEY를 읽기 위해 load_dotenv를 가져옵니다.
from dotenv import load_dotenv
# OpenAI API를 호출하는 공식 SDK 클라이언트입니다.
from openai import OpenAI

# 현재 실행 위치를 확인합니다.
CURRENT_DIR = Path.cwd()
# 루트에서 실행 중이면 04.web_search 폴더를 선택합니다.
if (CURRENT_DIR / "04.web_search").exists():
    CURRICULUM_ROOT = CURRENT_DIR / "04.web_search"
# 실습 폴더 안에서 실행 중이면 현재 위치를 사용합니다.
elif CURRENT_DIR.name == "04.web_search":
    CURRICULUM_ROOT = CURRENT_DIR
# 그 밖의 위치에서는 상대 경로로 실습 폴더를 찾습니다.
else:
    CURRICULUM_ROOT = Path("04.web_search").resolve()
# 워크스페이스 루트는 실습 폴더의 상위 폴더입니다.
WORKSPACE_ROOT = CURRICULUM_ROOT.parent
# 결과 저장 폴더 경로입니다.
OUTPUT_DIR = CURRICULUM_ROOT / "output"
# 결과 저장 폴더가 없으면 생성합니다.
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
# .env 파일에서 환경 변수를 로드합니다.
load_dotenv(WORKSPACE_ROOT / ".env")
# API 키가 없으면 명확하게 중단합니다.
if not os.getenv("OPENAI_API_KEY"):
    raise RuntimeError("OPENAI_API_KEY가 없습니다. 워크스페이스 루트의 .env 파일을 확인해 주세요.")
# OpenAI SDK 클라이언트를 생성합니다.
client = OpenAI()
# 웹 검색 실습에 사용할 모델명입니다.
WEB_SEARCH_MODEL = os.getenv("OPENAI_WEB_SEARCH_MODEL", "gpt-5-mini")
# 필요할 때 사용할 보조 모델명입니다.
FALLBACK_MODEL = os.getenv("OPENAI_WEB_SEARCH_FALLBACK_MODEL", "gpt-5-nano")

# 워크스페이스 기준 상대 경로를 반환합니다.
def workspace_path(path: Path) -> str:
    # resolve와 as_posix로 OS 차이를 줄입니다.
    return path.resolve().relative_to(WORKSPACE_ROOT.resolve()).as_posix()

# usage 객체를 JSON 저장 가능한 dict로 바꿉니다.
def usage_to_dict(usage: Any) -> dict[str, Any]:
    # usage가 없으면 빈 dict를 반환합니다.
    if usage is None:
        return {}
    # Pydantic 객체면 model_dump()로 변환합니다.
    if hasattr(usage, "model_dump"):
        return usage.model_dump()
    # 이미 dict면 그대로 사용합니다.
    if isinstance(usage, dict):
        return usage
    # 예상 밖 형태는 빈 dict로 처리합니다.
    return {}

# 긴 텍스트를 짧은 미리보기로 줄입니다.
def preview_text(value: str, limit: int = 230) -> str:
    # 줄바꿈과 연속 공백을 한 칸으로 정리합니다.
    compact = " ".join(str(value).split())
    # 길이가 제한 이하이면 그대로 반환합니다.
    if len(compact) <= limit:
        return compact
    # 길면 앞부분만 남기고 말줄임표를 붙입니다.
    return compact[: limit - 3] + "..."

# 현재 날짜를 한국어 형식으로 만듭니다.
def today_string() -> str:
    # 웹 검색은 날짜에 따라 결과가 바뀌므로 프롬프트에 기준 날짜를 넣습니다.
    return datetime.now().strftime("%Y년 %m월 %d일")

# 응답 아이템이 dict든 객체든 같은 방식으로 필드를 읽습니다.
def item_field(item: Any, key: str, default: Any = None) -> Any:
    # dict이면 get으로 안전하게 읽습니다.
    if isinstance(item, dict):
        return item.get(key, default)
    # 객체이면 getattr로 안전하게 읽습니다.
    return getattr(item, key, default)

# Responses API 응답에서 web_search_call 항목을 추출합니다.
def extract_web_search_calls(response: Any) -> list[dict[str, Any]]:
    # response.output은 메시지와 도구 호출이 섞인 리스트입니다.
    output_items = getattr(response, "output", [])
    # 추출한 웹 검색 호출을 담을 리스트입니다.
    calls: list[dict[str, Any]] = []
    # output 항목을 순서대로 확인합니다.
    for index, item in enumerate(output_items):
        # web_search_call 항목만 처리합니다.
        if item_field(item, "type", "") != "web_search_call":
            continue
        # 검색 행동 정보가 들어 있는 action 객체를 읽습니다.
        action = item_field(item, "action", None)
        # 단일 query 필드를 읽습니다.
        query = str(item_field(action, "query", "")).strip()
        # 복수 queries 필드를 읽습니다.
        queries_raw = item_field(action, "queries", None)
        # queries가 list이면 각 검색어를 문자열로 정리합니다.
        queries = [str(value).strip() for value in queries_raw if str(value).strip()] if isinstance(queries_raw, list) else ([query] if query else [])
        # sources 필드를 읽습니다.
        sources_raw = item_field(action, "sources", None)
        # 정규화된 source 목록을 담습니다.
        sources: list[dict[str, str]] = []
        # sources가 list일 때만 순회합니다.
        if isinstance(sources_raw, list):
            # 각 source에서 URL과 제목을 꺼냅니다.
            for source in sources_raw:
                sources.append({"url": str(item_field(source, "url", "")).strip(), "title": str(item_field(source, "title", "")).strip(), "type": str(item_field(source, "type", "url")).strip() or "url"})
        # 하나의 웹 검색 호출 정보를 저장합니다.
        calls.append({"index": index, "id": item_field(item, "id", None), "status": str(item_field(item, "status", "")), "action_type": str(item_field(action, "type", "")), "query": query, "queries": queries, "sources": sources})
    # 수집한 웹 검색 호출 목록을 반환합니다.
    return calls

# 응답 본문의 URL 인용 annotation을 추출합니다.
def extract_url_citations(response: Any) -> list[dict[str, str]]:
    # response.output 리스트를 읽습니다.
    output_items = getattr(response, "output", [])
    # 인용 정보를 담을 리스트입니다.
    citations: list[dict[str, str]] = []
    # output 항목을 순회합니다.
    for item in output_items:
        # message 항목의 content 배열을 읽습니다.
        content = item_field(item, "content", None)
        # content가 list가 아니면 건너뜁니다.
        if not isinstance(content, list):
            continue
        # 각 content block을 순회합니다.
        for block in content:
            # block 안의 annotations 배열을 읽습니다.
            annotations = item_field(block, "annotations", None)
            # annotations가 list가 아니면 건너뜁니다.
            if not isinstance(annotations, list):
                continue
            # annotation을 하나씩 확인합니다.
            for annotation in annotations:
                # URL 인용만 수집합니다.
                if item_field(annotation, "type", "") == "url_citation":
                    citations.append({"url": str(item_field(annotation, "url", "")).strip(), "title": str(item_field(annotation, "title", "")).strip(), "type": "url_citation"})
    # 수집한 인용 목록을 반환합니다.
    return citations

# URL에서 도메인만 추출합니다.
def domain_from_url(url: str) -> str:
    # urlparse로 URL을 분해합니다.
    parsed = urlparse(str(url))
    # netloc이 도메인에 해당하므로 소문자로 정리합니다.
    return parsed.netloc.lower()

# 웹 검색 호출과 인용 목록을 요약 지표로 바꿉니다.
def collect_search_overview(web_calls: list[dict[str, Any]], citations: list[dict[str, str]]) -> dict[str, Any]:
    # 모든 검색 질의를 담을 리스트입니다.
    all_queries: list[str] = []
    # 모든 검색 출처 URL을 담을 리스트입니다.
    source_urls: list[str] = []
    # 각 웹 검색 호출을 순회합니다.
    for call in web_calls:
        # 질의 목록을 누적합니다.
        all_queries.extend(call.get("queries", []))
        # 출처 URL을 누적합니다.
        source_urls.extend([source.get("url", "") for source in call.get("sources", [])])
    # 최종 응답에서 실제 인용된 URL 목록입니다.
    citation_urls = [item.get("url", "") for item in citations]
    # 출처 도메인을 중복 제거해 정렬합니다.
    source_domains = sorted({domain_from_url(url) for url in source_urls if domain_from_url(url)})
    # 인용 도메인을 중복 제거해 정렬합니다.
    citation_domains = sorted({domain_from_url(url) for url in citation_urls if domain_from_url(url)})
    # 학습자가 확인할 요약 지표를 반환합니다.
    return {"web_search_call_count": len(web_calls), "query_count": len([query for query in all_queries if query]), "unique_queries": sorted({query for query in all_queries if query}), "source_url_count": len([url for url in source_urls if url]), "source_domain_count": len(source_domains), "source_domains": source_domains, "citation_count": len(citations), "citation_domain_count": len(citation_domains), "citation_domains": citation_domains}

# 실행 조건을 출력합니다.
print(f"실습 폴더: {CURRICULUM_ROOT}")
# 저장 위치를 출력합니다.
print(f"결과 저장 폴더: {OUTPUT_DIR}")
# 요청 모델을 출력합니다.
print(f"요청 모델: {WEB_SEARCH_MODEL}")

실습 폴더: c:\Users\ai4nu\2026-ai-tech\AS.1.1_0601\04.web_search
결과 저장 폴더: c:\Users\ai4nu\2026-ai-tech\AS.1.1_0601\04.web_search\output
요청 모델: gpt-5-mini

 

2단계. 검색 과제와 도구 설정

이 단계에서는 모델이 어떤 검색을 해야 하는지와 어떤 웹 검색 옵션을 사용할지 정합니다. 도구 설정을 코드로 직접 보면 web_search, filters, user_location, reasoning이 요청 안에 어떻게 들어가는지 분명해집니다.

# 검색을 허용할 신뢰 도메인 목록입니다.
ALLOWED_DOMAINS = ["pubmed.ncbi.nlm.nih.gov", "clinicaltrials.gov", "www.who.int", "www.cdc.gov", "www.fda.gov"]
# 의료 정보 검증 역할 지시문입니다.
INSTRUCTIONS = "당신은 의료 정보 검증을 돕는 교육용 리서치 어시스턴트입니다. 반드시 웹 검색을 수행하고, 허용된 도메인 기반으로 요약하세요. 답변은 쉬운 한국어로 작성하고 핵심 용어는 짧게 풀어서 설명하세요."
# 도메인 제한 검색용 사용자 프롬프트입니다.
PROMPT = "반드시 웹 검색을 수행해 주세요. 주제: 당뇨병 치료에서 세마글루타이드(semaglutide)의 최신 임상 동향. 요구사항: 1) 근거 3개, 2) 기대 효과, 3) 주의사항, 4) 출처 URL을 정리해 주세요."
# allowed_domains 필터가 포함된 웹 검색 도구 설정입니다.
WEB_TOOLS = [{"type": "web_search", "filters": {"allowed_domains": ALLOWED_DOMAINS}}]

# 최대 출력 토큰 수입니다.
MAX_OUTPUT_TOKENS = 1600
# reasoning effort 값입니다.
REASONING_EFFORT = 'low'

 

 

3단계. client.responses.create(...)로 웹 검색 실행

이 셀이 핵심입니다. tools=[...], tool_choice='required', include=[...]가 실제 API 요청에 들어가며, 응답의 output에서 검색 호출과 최종 메시지를 함께 확인합니다.

# 이번 실습의 프롬프트를 준비합니다.
prompt = PROMPT
# API 호출 시작 시간을 기록합니다.
started = datetime.now()
# Responses API를 직접 호출하며 web_search 도구를 전달합니다.
response = client.responses.create(
    # 웹 검색 도구를 사용할 모델입니다.
    model=WEB_SEARCH_MODEL,
    # 모델의 역할과 답변 형식을 지정합니다.
    instructions=INSTRUCTIONS,
    # 사용자 검색 요청입니다.
    input=[{"role": "user", "content": prompt}],
    # 웹 검색 도구 설정입니다.
    tools=WEB_TOOLS,
    # 학습 재현성을 위해 검색 실행을 요구합니다.
    tool_choice="required",
    # 출처 URL을 추적하기 위해 sources 필드를 포함합니다.
    include=["web_search_call.action.sources"],
    # 검색과 요약 결과가 잘리지 않도록 출력 토큰 한도를 지정합니다.
    max_output_tokens=MAX_OUTPUT_TOKENS,
    # 추론 깊이를 조절합니다.
    reasoning={"effort": REASONING_EFFORT},
)
# 최종 응답 텍스트를 읽습니다.
final_text = response.output_text or ""
# 텍스트가 비어 있으면 상태를 포함해 중단합니다.
if not final_text:
    raise RuntimeError(f"최종 응답 텍스트가 비어 있습니다. status={response.status}")
# 웹 검색 호출 기록을 추출합니다.
web_calls = extract_web_search_calls(response)
# 최종 응답의 URL 인용을 추출합니다.
citations = extract_url_citations(response)
# 검색 호출과 인용을 요약 지표로 바꿉니다.
overview = collect_search_overview(web_calls, citations)
# 최종 응답 미리보기를 출력합니다.
print(preview_text(final_text, 260))
# 검색 호출 수를 출력합니다.
print(f"웹 검색 호출 수: {overview['web_search_call_count']}")
# 검색 질의 수를 출력합니다.
print(f"검색 질의 수: {overview['query_count']}")
# 인용 수를 출력합니다.
print(f"인용 출처 수: {overview['citation_count']}")

알겠습니다 — 요청하신 대로 반드시 웹 검색(허용 도메인: PubMed, ClinicalTrials.gov 등)을 수행하여 정리합니다. 아래는 쉬운 한국어 요약(핵심 용어는 간단 풀이), 근거 3개, 기대효과, 주의사항, 그리고 출처 URL입니다. 요약(한줄) - 세마글루타이드(semaglutide)는 GLP‑1 수용체 작용제로, 혈당을 낮추고 체중을 줄이며 심혈관 위험 지표에 유리한 영향을 보이는 약물입니다. (근거: 무작위시험·메타분석/임상시험 문헌)....
웹 검색 호출 수: 3
검색 질의 수: 6
인용 출처 수: 6

 

 

4단계. 결과와 출처 추적 저장

웹 검색 답변은 최신성과 근거가 중요합니다. 최종 응답만 저장하지 않고, 모델이 만든 검색 질의와 수집된 출처를 JSON으로 함께 저장해 검증 가능한 실습 결과를 만듭니다.

# 최종 응답 텍스트 파일 경로입니다.
output_text_path = OUTPUT_DIR / "02_domain_filter_v2_output.txt"
# 검색 호출과 출처 추적 정보를 저장할 파일 경로입니다.
trace_path = OUTPUT_DIR / "02_domain_filter_v2_trace.json"
# 메타데이터 파일 경로입니다.
metadata_path = OUTPUT_DIR / "02_domain_filter_v2_meta.json"
# 최종 응답을 저장합니다.
output_text_path.write_text(final_text + "\n", encoding="utf-8")
# 추적 정보를 구성합니다.
trace = {"instructions": INSTRUCTIONS, "prompt": prompt, "actual_model": response.model, "web_search_calls": web_calls, "citations": citations, "overview": overview}
# 추적 정보를 JSON으로 저장합니다.
trace_path.write_text(json.dumps(trace, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
# usage를 dict로 변환합니다.
usage = usage_to_dict(response.usage)
# 메타데이터를 구성합니다.
metadata = {"example": "02_domain_filter_v2", "requested_model": WEB_SEARCH_MODEL, "actual_model": response.model, "reasoning_effort": REASONING_EFFORT, "web_search_call_count": overview["web_search_call_count"], "query_count": overview["query_count"], "unique_query_count": len(overview["unique_queries"]), "source_domain_count": overview["source_domain_count"], "citation_count": overview["citation_count"], "input_tokens": usage.get("input_tokens"), "output_tokens": usage.get("output_tokens"), "total_tokens": usage.get("total_tokens"), "output_path": workspace_path(output_text_path), "trace_path": workspace_path(trace_path)}
# 메타데이터를 저장합니다.
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
# 저장 위치를 출력합니다.
print(f"최종 응답 저장: {workspace_path(output_text_path)}")
print(f"추적 JSON 저장: {workspace_path(trace_path)}")
print(f"메타데이터 저장: {workspace_path(metadata_path)}")

최종 응답 저장: 04.web_search/output/02_domain_filter_v2_output.txt
추적 JSON 저장: 04.web_search/output/02_domain_filter_v2_trace.json
메타데이터 저장: 04.web_search/output/02_domain_filter_v2_meta.json