utils/context_compression.py

"""

Context Compression Utilities

AI 프롬프트 토큰 절감을 위한 압축 유틸리티



@module context_compression

@description

- 스팟 데이터를 압축 형식으로 변환하여 프롬프트 토큰 절감

- 목표: 프롬프트 토큰 5,000 -> 500 (90% 감소)

- API 비용 30% 절감, 응답 속도 20-30% 개선



@changelog

- v1.0.0 (2026-01-25): 초기 구현

  - compress_spot: 단일 스팟 압축 (JSON -> 파이프 구분 문자열)

  - compress_spots: 스팟 리스트 압축

  - decompress_course_spots: AI 응답 spot_id로 원본 스팟 복원

  - create_compression_guide: 프롬프트용 압축 형식 가이드



@example

Before (약 500 토큰):

{

  "id": "vj_123",

  "name": "하귀포구",

  "category": "포구",

  "location": {"lat": 33.456, "lng": 126.789},

  "tags": ["역사", "바다", "사진"],

  "story_preview": "400년 전 왜구의 침략..."

}



After (약 50 토큰):

vj_123|하귀포구|포구|33.4560,126.7890|역사,바다,사진|15

"""
from typing import List, Dict, Any, Optional
import logging

logger = logging.getLogger(__name__)


def compress_spot(spot: Dict[str, Any]) -> str:
    """

    단일 스팟을 압축 형식으로 변환



    Args:

        spot: 스팟 딕셔너리 (id, name, category, location, tags, meta 등)



    Returns:

        압축된 문자열 (파이프 구분)

        형식: spot_id|이름|카테고리|위도,경도|태그1,태그2|체류시간



    Example:

        Input:  {"id": "vj_123", "name": "하귀포구", ...}

        Output: "vj_123|하귀포구|포구|33.4560,126.7890|역사,바다|15"

    """
    # 기본 필드 추출
    spot_id = spot.get("id", "")
    name = spot.get("name", "")
    category = spot.get("category", "")

    # 위치 정보 추출 (소수점 4자리로 제한)
    location = spot.get("location", {})
    lat = location.get("lat", 0)
    lng = location.get("lng", 0)
    loc_str = f"{lat:.4f},{lng:.4f}"

    # 태그 (최대 5개로 제한)
    tags = spot.get("tags", [])[:5]
    tags_str = ",".join(tags) if tags else ""

    # 체류 시간 (meta에서 추출)
    meta = spot.get("meta", {})
    stay_duration = meta.get("stay_duration_min", 15) if meta else 15

    return f"{spot_id}|{name}|{category}|{loc_str}|{tags_str}|{stay_duration}"


def compress_spots(spots: List[Dict[str, Any]]) -> str:
    """

    스팟 리스트를 압축 형식으로 변환



    Args:

        spots: 스팟 딕셔너리 리스트



    Returns:

        줄바꿈으로 구분된 압축 문자열



    Example:

        vj_001|하귀포구|포구|33.4560,126.7890|역사,바다|15

        vj_002|곽지해변|해변|33.4567,126.7891|자연,사진|20

    """
    compressed_lines = [compress_spot(spot) for spot in spots]
    return "\n".join(compressed_lines)


def create_compression_guide() -> str:
    """

    압축 형식 설명 (프롬프트에 포함)

    AI가 압축된 데이터를 이해할 수 있도록 형식 설명 제공



    Returns:

        프롬프트에 삽입할 형식 가이드 문자열

    """
    return """**스팟 데이터 형식 (압축)**

각 줄 형식: spot_id|이름|카테고리|위도,경도|태그들|체류시간(분)

예시: vj_123|하귀포구|포구|33.4560,126.7890|역사,바다|15



**중요**: 응답의 spot_id는 반드시 위 목록에 있는 ID만 사용하세요.

"""


def decompress_course_spots(

    compressed_spot_ids: List[str],

    original_spots: List[Dict[str, Any]]

) -> List[Dict[str, Any]]:
    """

    AI가 반환한 spot_id 리스트를 원본 스팟 데이터로 복원



    Args:

        compressed_spot_ids: AI가 반환한 spot_id 리스트

        original_spots: 원본 스팟 데이터 리스트



    Returns:

        복원된 스팟 딕셔너리 리스트



    Example:

        Input:  ["vj_001", "vj_003", "vj_005"]

        Output: [{"id": "vj_001", ...}, {"id": "vj_003", ...}, ...]

    """
    # spot_id -> 원본 스팟 매핑
    spot_map = {spot["id"]: spot for spot in original_spots}

    decompressed = []
    for spot_id in compressed_spot_ids:
        if spot_id in spot_map:
            decompressed.append(spot_map[spot_id])
        else:
            logger.warning(f"[decompress] spot_id not found: {spot_id}")

    return decompressed


def calculate_compression_ratio(

    original_json: str,

    compressed_str: str

) -> Dict[str, Any]:
    """

    압축률 계산 및 통계 반환



    Args:

        original_json: 원본 JSON 문자열

        compressed_str: 압축된 문자열



    Returns:

        압축 통계 딕셔너리

    """
    original_len = len(original_json)
    compressed_len = len(compressed_str)

    # 대략적인 토큰 수 추정 (한글 기준 약 2자당 1토큰)
    original_tokens_est = original_len // 2
    compressed_tokens_est = compressed_len // 2

    ratio = (compressed_len / original_len * 100) if original_len > 0 else 0
    savings = 100 - ratio

    return {
        "original_chars": original_len,
        "compressed_chars": compressed_len,
        "original_tokens_est": original_tokens_est,
        "compressed_tokens_est": compressed_tokens_est,
        "ratio_percent": round(ratio, 1),
        "savings_percent": round(savings, 1)
    }