""" 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) }