[RAG Playground] Quest 2: 공공데이터 API 수집 파이프라인 구축 계획

본격적인 RAG 시스템 구축에 앞서, 파이프라인의 핵심인 “데이터”를 모으는 작업을 시작해 봅니다. 이번 튜토리얼에서 활용할 공공데이터는 부산광역시_가족사랑카드 참여업체 현황 API입니다. 해당 API는 아래 문서를 기준으로 호출을 진행합니다.

부산광역시_가족사랑카드 참여업체 현황 OpenAPI공공데이터포털에서 제공하는 부산광역시 가족사랑카드 참여업체 현황 API 문서

www.data.go.kr

해당 링크로 이동해 먼저 OpenAPI 활용 신청을 진행합니다. 활용 목적을 작성하고 신청하면 즉시 일반 인증키가 발급되며, 저는 이 중 Decoding 키를 활용하여 API를 호출할 예정입니다. (참고로 Encoding 키로는 호출이 안 되더라고요 🤔 설명서를 읽어보니 이유가 있던데, 이 부분은 빠르게 패스하겠습니다!)

왜 이 데이터를 먼저 고르나?

구조화된 정보(지역, 업종 등)와 비구조화된 텍스트(할인 혜택 상세내용)가 섞여 있어, 단순한 단어 매칭을 넘어서는 입체적인 질의응답 실험을 진행하기에 매우 좋습니다.

예를 들어, 나중에 구축할 RAG 시스템에 다음과 같은 다채로운 형태의 사용자 패턴을 테스트해 볼 수 있습니다:

지역 기반 필터링: “해운대구에서 가족사랑카드로 할인되는 음식점 3곳만 알려줘”
조건 기반 추론: “우리아이가 학원을 다니려 하는데, 강서구에 있는 학원 중 세자녀 할인이 큰 곳은 어디야?”
다중 조건 검색: “서구 주변에 있는 병원이나 안경점 중에서 10% 이상 할인되는 곳을 찾아줘”

즉, 메타데이터(지역/업종) 필터링 + 의미 기반(할인율/혜택 내용) 하이브리드 검색의 뼈대를 잡고 성능을 측정해보기에 완벽한 튜토리얼 데이터셋입니다 👍

이번 목표

이번 Quest의 목표는 실제 구현 기준으로 아래 3가지입니다.

httpx.AsyncClient 기반으로 API를 안정적으로 호출하기
응답 JSON을 RAG 친화 포맷으로 정제하기
수집 결과를 data/raw/family_card_shops.json으로 저장하기

구현 내용

1) API 클라이언트 (`family_card_api.py`)

src/rag_playground/data/family_card_api.py에서 아래 두 함수를 구현했습니다.

먼저 API를 호출하는 fetch_card_shops 함수입니다. httpx.AsyncClient로 비동기 호출을 하고, 정상적으로 발급받은 API 키는 Decoding 키를 사용해야 합니다 (Encoding 키는 %로 시작하는 이중 인코딩 문제가 생기더라고요 🤔)

import os
import httpx

BASE_URL = "https://apis.data.go.kr/6260000/FamilyCardService/getFamilyCardInfo"

async def fetch_card_shops(
    page_no: int = 1,
    num_of_rows: int = 100,
    cp_compname: str = "",
    cp_hgu: str = "",
    cp_class: str = "",
) -> dict:
    api_key = os.getenv("DATA_GO_KR_API_KEY")
    if not api_key:
        raise EnvironmentError("DATA_GO_KR_API_KEY 환경 변수가 설정되지 않았습니다.")

    params = {
        "serviceKey": api_key,
        "pageNo": page_no,
        "numOfRows": num_of_rows,
        "resultType": "json",
    }
    if cp_compname:
        params["cpCompname"] = cp_compname
    if cp_hgu:
        params["cpHgu"] = cp_hgu
    if cp_class:
        params["cpClass"] = cp_class

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(BASE_URL, params=params)
        response.raise_for_status()

    # 간헐적으로 JSON 대신 XML 에러 응답이 오는 경우가 있어서 별도 처리
    if response.headers.get("content-type", "").startswith("text/xml"):
        raise ValueError(f"XML 에러 응답 수신: {response.text[:200]}")

    return response.json()

응답 JSON을 RAG에 맞게 정제하는 parse_shops_data 함수입니다. 눈여겨볼 점은 API가 결과가 1개일 때 item을 리스트가 아닌 단일 객체로 내려보내는데, 이를 보정하는 로직이 필요했습니다.

from datetime import datetime, timezone, timedelta

KST = timezone(timedelta(hours=9))

def parse_shops_data(raw_data: dict) -> list[dict]:
    header = raw_data.get("getFamilyCardInfo", {}).get("header", {})
    if str(header.get("resultCode")) != "00":
        raise ValueError(f"API 에러: {header.get('resultMsg')}")

    body = raw_data.get("getFamilyCardInfo", {}).get("body", {})
    items = body.get("items", {}).get("item", [])

    # 결과가 1건일 때 단일 객체로 내려오는 케이스 보정
    if isinstance(items, dict):
        items = [items]

    collected_at = datetime.now(tz=KST).isoformat()

    return [
        {
            "shop_name": item.get("cpCompname", ""),
            "address": item.get("cpAddr", ""),
            "district": item.get("cpHgu", ""),
            "benefit": item.get("cpWoo") or item.get("cpContent", ""),
            "category": item.get("cpClass", ""),
            "phone": item.get("cpTel", ""),
            "source": "부산광역시_가족사랑카드 참여업체 현황",
            "collected_at": collected_at,
        }
        for item in items
    ]

2) 수집 스크립트 (`collect_family_card.py`)

src/rag_playground/data/collect_family_card.py는 페이지네이션 기반 전체 수집기입니다. 마지막 페이지 판단은 “받아온 결과 수 < 요청한 페이지 크기” 조건으로 처리합니다.

import asyncio
import orjson
from pathlib import Path
from family_card_api import fetch_card_shops, parse_shops_data

NUM_OF_ROWS = 100

async def collect_all():
    all_shops = []
    page_no = 1

    while True:
        print(f"[페이지 {page_no}] 수집 중...")
        raw = await fetch_card_shops(page_no=page_no, num_of_rows=NUM_OF_ROWS)

        try:
            shops = parse_shops_data(raw)
        except ValueError as e:
            print(f"파싱 실패: {e}")
            break

        all_shops.extend(shops)
        print(f"  → {len(shops)}건 수집 (누적 {len(all_shops)}건)")

        if len(shops) < NUM_OF_ROWS:
            print("마지막 페이지 도달. 수집 완료.")
            break

        page_no += 1
        await asyncio.sleep(0.5)  # API 서버 부하 방지

    output_dir = Path(__file__).resolve().parents[3] / "data" / "raw"
    output_dir.mkdir(parents=True, exist_ok=True)
    output_path = output_dir / "family_card_shops.json"

    output_path.write_bytes(orjson.dumps(all_shops, option=orjson.OPT_INDENT_2))
    print(f"\n저장 완료: {output_path} ({len(all_shops)}건)")

if __name__ == "__main__":
    asyncio.run(collect_all())

저는 평소에 json 보다 orjson 라이브러리를 더 선호합니다. Rust 기반으로 만들어져서 속도가 확실히 빠르고, 굳이 안 쓸 이유가 없거든요. 물론 아주 작은 데이터는 표준 json으로도 충분하지만, 어느 정도 규모가 있는 데이터라면 orjson을 쓰는 게 습관적으로 더 낫다고 생각합니다 😎

저장 데이터 포맷(예시)

최종적으로는 아래 형태를 목표로 합니다.

{
  "shop_name": "가게명",
  "address": "부산광역시 ...",
  "district": "해운대구",
  "benefit": "결제금액 10% 할인",
  "category": "음식점",
  "phone": "051-000-0000",
  "source": "부산광역시_가족사랑카드 참여업체 현황",
  "collected_at": "2026-03-02T00:00:00+09:00"
}

이렇게 맞춰두면 이후 단계에서:

임베딩용 텍스트 생성(shop_name + address + benefit)
메타데이터 필터링(district, category)
추적성 확보(source, collected_at)

를 깔끔하게 처리할 수 있습니다 👻

검증 계획

자동화/수동 검증

collect_family_card.py 실행 후 페이지별 로그가 정상 출력되는지 확인
최종 저장 파일 data/raw/family_card_shops.json 생성 여부 확인
저장 건수와 페이지 종료 조건(마지막 페이지 판단)이 의도대로 동작했는지 확인
샘플 레코드에서 benefit, district, category, phone 필드 정합성 확인
RAG 적재 전 텍스트/메타데이터 분리 전략 검토

다음 단계 메모

수집이 안정화되면 다음 순서로 이어갈 예정입니다.

정제 데이터 기반 chunking 전략 수립
벡터 DB 적재 스크립트 작성
지역/혜택 질의 세트로 검색 품질 점검

이번 단계는 단순 수집 스크립트를 넘어, 에러 케이스(XML 응답), 단일 객체 응답 보정, 수집 메타데이터(collected_at)까지 포함한 형태로 기본기를 다지는 데 초점을 맞췄습니다.

좋은 RAG는 결국 좋은 데이터 파이프라인에서 시작하니까요 🙆‍♂️

RAG Playground (9부작)