"""background_classify — Phase 7 Step 1. raw locations → building_groups[{group_id, members[], anchor_loc, kind}] LLM 1회 호출로 clustering + 분류 동시 수행 (spec D1). 흐름: 1. build_classify_user_prompt(locations, visual_world_rules) — 입력 직렬화 2. run_background_classify — call_structured_fn(3회 retry) 3. validate_classify_output — partition cover + anchor + kind heuristic 검증 """ from __future__ import annotations import logging import re import time from typing import Any, Callable, Dict, List, Optional, Set from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) _NON_ASCII_TEXT_RE = re.compile( r"[ㄱ-ㆎ가-힣" r"一-鿿㐀-䶿豈-﫿" r"぀-ヿ]" ) _SAFE_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_]*$") _LOC_ID_RE = re.compile(r"^[A-Za-z0-9_]+$") _MODULE = "background_classify" class ClassifyError(Exception): """retry 한도까지 실패.""" def build_classify_user_prompt( locations: List[Dict[str, Any]], visual_world_rules: str, ) -> str: """LLM에 보낼 user prompt — 모든 location 정보 truncation 없이 포함. Args: locations: List of {loc_id, label, shot_count, is_indoor, summary} (raw, NOT pre-grouped). LLM이 직접 cluster + classify. visual_world_rules: 에피소드 visual_world_rules 전문 (절대 자르지 않음). NOTE: visual_world_rules와 locations_block에 포함된 사용자/LLM 원문에 ``{`` ``}`` 가 있을 수 있어 .format() 충돌 회피용으로 placeholder만 직접 replace. """ template = load_prompt(_MODULE, "user_template") lines: List[str] = [] for loc in locations: loc_id = loc.get("loc_id", "") label = loc.get("label", "") or "" shot_count = loc.get("shot_count", 0) summary = (loc.get("summary", "") or "").strip() # is_indoor placeholder는 입력에 표시하지 않는다 — LLM이 description+name으로 # 자율 판단해 응답에 채우도록 위임 (entity_detail kind 누락 케이스 대응). head = f"- {loc_id} ({shot_count} shots): {label}" lines.append(head) if summary: lines.append(f" summary: {summary}") locations_block = "\n".join(lines) or "(none)" rules = visual_world_rules or "(none)" return ( template .replace("{visual_world_rules}", rules) .replace("{locations_block}", locations_block) ) def validate_classify_output( output: Dict[str, Any], all_loc_ids: Set[str], *, indoor_loc_ids: Optional[Set[str]] = None, shot_counts: Optional[Dict[str, int]] = None, ) -> None: """semantic invariants: - group_id : ASCII snake_case (schema 보강) - members[] non-empty; 각 loc_id ∈ all_loc_ids - partition cover: 모든 all_loc_ids가 정확히 한 group의 members에 등장 (LLM이 location을 누락/중복 cover하면 안 됨) - anchor_loc ∈ that group's members' loc_ids (snake_case 강제 안 됨, ASCII만) - kind ∈ {"chain_bg", "prev_shot_ref"} — schema가 잡지만 방어적 - heuristic: chain_bg 는 sum(shot_count) ≥ 3 AND any(member.is_indoor) 충족해야 Args: output: LLM 응답 dict. {"building_groups": [...]} all_loc_ids: 입력으로 준 모든 loc_id의 set (partition cover 검증). indoor_loc_ids: indoor인 loc_id의 set. 안 주어지면 LLM의 member.is_indoor 사용. shot_counts: loc_id → shot_count map. 안 주어지면 LLM의 member.shot_count 사용. """ groups = output.get("building_groups") or [] if not isinstance(groups, list): raise ValueError(f"building_groups must be list, got {type(groups).__name__}") seen_locs: Dict[str, str] = {} # loc_id → group_id seen_group_ids: Set[str] = set() indoor_set = set(indoor_loc_ids or set()) shot_map = dict(shot_counts or {}) for g in groups: gid = g.get("group_id", "") if not _SAFE_ID_RE.match(gid): raise ValueError( f"group_id {gid!r} contains non-ASCII or unsafe chars" ) if _NON_ASCII_TEXT_RE.search(gid): raise ValueError(f"group_id {gid!r} contains non-ASCII text") if gid in seen_group_ids: raise ValueError(f"duplicate group_id {gid!r}") seen_group_ids.add(gid) members = g.get("members") or [] if not isinstance(members, list) or not members: raise ValueError(f"group {gid!r} has empty/invalid members") member_loc_ids: List[str] = [] for m in members: mid = m.get("loc_id", "") if not mid or not _LOC_ID_RE.match(mid): raise ValueError(f"group {gid!r} member loc_id {mid!r} invalid") if mid not in all_loc_ids: raise ValueError( f"group {gid!r} member loc_id {mid!r} not in input locations" ) if mid in seen_locs: raise ValueError( f"loc_id {mid!r} appears in both groups {seen_locs[mid]!r} and {gid!r} " f"(partition cover violation)" ) seen_locs[mid] = gid member_loc_ids.append(mid) anchor = g.get("anchor_loc", "") if not anchor or not _LOC_ID_RE.match(anchor): raise ValueError(f"anchor_loc {anchor!r} for group {gid!r} invalid format") if _NON_ASCII_TEXT_RE.search(anchor): raise ValueError(f"anchor_loc {anchor!r} contains non-ASCII") if anchor not in member_loc_ids: raise ValueError( f"anchor_loc {anchor!r} for group {gid!r} not in its members {member_loc_ids}" ) kind = g.get("kind", "") if kind not in {"chain_bg", "prev_shot_ref"}: raise ValueError(f"unknown kind {kind!r} for group {gid!r}") # heuristic: chain_bg는 ≥3 shots + indoor anchor 필요. if kind == "chain_bg": # Authoritative shot/indoor 정보가 있으면 그것 사용, 아니면 member 필드 사용. total_shots = sum( shot_map.get(mid, m.get("shot_count", 0) or 0) for mid, m in zip(member_loc_ids, members) ) has_indoor = any( (mid in indoor_set) if indoor_set else bool(m.get("is_indoor", False)) for mid, m in zip(member_loc_ids, members) ) if total_shots < 3: raise ValueError( f"group {gid!r} kind=chain_bg but total_shots={total_shots} (< 3)" ) if not has_indoor: raise ValueError( f"group {gid!r} kind=chain_bg but no indoor member" ) # partition cover: 모든 input loc_id가 정확히 한 그룹에 covered missing = all_loc_ids - set(seen_locs.keys()) if missing: raise ValueError( f"missing loc_ids in cover: {sorted(missing)} (partition cover violation)" ) def run_background_classify( *, user_prompt: str, all_loc_ids: List[str], indoor_loc_ids: Optional[List[str]] = None, shot_counts: Optional[Dict[str, int]] = None, call_structured_fn: Callable[..., Dict[str, Any]], project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, max_retries: int = 3, backoff_base_sec: float = 2.0, sleep_fn: Callable[[float], None] = time.sleep, ) -> Dict[str, Any]: system = load_prompt(_MODULE, "system") schema = load_schema(_MODULE, "schema") loc_set = set(all_loc_ids) indoor_set = set(indoor_loc_ids or []) last_err: Optional[Exception] = None for attempt in range(max_retries): try: result = call_structured_fn( step="background_classify", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=project_config, schema_name="background_classify", opik_metadata=opik_metadata, ) validate_classify_output( result, loc_set, indoor_loc_ids=indoor_set, shot_counts=shot_counts, ) return result except Exception as exc: # T20 I3: LLM client errors (litellm/openai/httpx) 포함 last_err = exc logger.warning( "background_classify attempt %d/%d failed: %s", attempt + 1, max_retries, exc, ) if attempt + 1 < max_retries: sleep_fn(backoff_base_sec * (attempt + 1)) raise ClassifyError( f"background_classify exhausted {max_retries} retries: {last_err}" )