feat: ABM 시뮬 UX 개선 + 가맹사업법 후보지 영업구역 평가 연계

.gitignore

@@ -176,6 +176,7 @@ sim_*.json
 
 # 법률 RAG 빌드 산출물 (build_parent_articles.py로 재생성 가능)
 backend/data/legal/processed/parent_articles.json
+backend/data/legal/processed/_embedding_cache.pkl
 
 # RAG 검색 trace JSONL (디버깅용, 대용량)
 validation/results/rag_trace/

backend/src/agents/legal/orchestrator.py

@@ -21,21 +21,18 @@
 logger = logging.getLogger(__name__)
 
 
-# tasks 리스트와 1:1 대응 — 예외 처리 시 type 식별에 사용
+# tasks 리스트와 1:1 대응 — 예외 처리 시 type 식별에 사용.
+# 운영(operation) 카테고리(food_hygiene/labor_law/vat_law/privacy_law/sewage_law) 제외 —
+# 사용자 요청에 따라 frontend 미표시 + LLM 호출 절감 (privacy_law specialist skip).
 _RULE_ENGINE_ORDER: list[str] = [
-    "food_hygiene",
     "safety_regulation",
     "fire_safety_law",
     "accessibility_law",
     "school_zone",
     "commercial_lease_law",
-    "labor_law",
-    "vat_law",
-    "sewage_law",
     "franchise_law",
     "fair_trade_law",
     "building_law",
-    "privacy_law",
 ]
 
 
@@ -100,31 +97,29 @@ async def run_legal_evaluation(
     # 룰 9 개 — 동기 pure-Python (~ms). asyncio.to_thread executor 점유 회피 위해
     # 직접 호출 후 dict 로 수집 — 이벤트루프 블로킹 위험 없음.
     rule_results: list[dict] = []
+    # 운영(operation) 카테고리 룰 4개 (food_hygiene/labor/vat/sewage) 제외 —
+    # frontend 미표시 + 비용 절감. _RULE_ENGINE_ORDER 와 1:1 대응.
     for fn, args in (
-        (rules.rule_food_hygiene, (business_type,)),
         (rules.rule_safety_regulation, (business_type, store_area_pyeong)),
         (rules.rule_fire_safety, (business_type, store_area_pyeong)),
         (rules.rule_accessibility, (business_type, store_area_pyeong)),
         (rules.rule_school_zone, (business_type, lat, lon)),
         (rules.rule_commercial_lease, ()),
-        (rules.rule_labor, ()),
-        (rules.rule_vat, ()),
-        (rules.rule_sewage, (business_type,)),
     ):
         try:
             rule_results.append(fn(*args))
         except Exception as e:
             type_name = _RULE_ENGINE_ORDER[len(rule_results)]
             rule_results.append(_fallback_for_type(type_name, e))
 
-    # specialist 4 개 — RAG + LLM (I/O bound) 병렬 실행
+    # specialist 3 개 — RAG + LLM (I/O bound) 병렬 실행.
+    # privacy_law specialist (운영 카테고리) 제외 — LLM 호출 절감 + frontend 미표시.
     # franchise_law 영업지역 분석 = 공실 스팟 좌표 (lat/lon) 기준 500m 반경.
     # 좌표 None 이면 specialist 가 행정동 centroid 폴백 사용.
     specialist_tasks = [
         specialists.specialist_franchise_law(brand, business_type, district, ftc_data, lat, lon, territory_radius_m),
         specialists.specialist_fair_trade_law(brand, business_type, district),
         specialists.specialist_building_law(business_type, district),
-        specialists.specialist_privacy_law(brand, business_type, ftc_data),
     ]
 
     expected_total = len(rule_results) + len(specialist_tasks)

backend/src/agents/legal/specialists.py

@@ -361,6 +361,22 @@ async def specialist_franchise_law(
     precedent_text = _format_precedents(precedents)
     territory_floor, territory_hint = _territory_to_level(territory)
 
+    # 평가 기준 — 사용자 입력 영업구역 우선. 영업구역 미입력 시만 500m 일반 임계값.
+    if territory_radius_m:
+        _territory_criteria = (
+            f"- 사용자 정의 영업구역 = **{territory_radius_m}m**. 이 거리만 침해 판정 기준으로 사용.\n"
+            f"- {territory_radius_m}m 안 동일 브랜드 1개 이상 → danger (가맹사업법 제12조의4 명백 침해)\n"
+            f"- {territory_radius_m}m 안 동일 브랜드 0개 → safe (정보공개서 기준 충족)\n"
+            "- summary/recommendation 에서 '500m' 같은 다른 거리 임계값 인용 금지. "
+            f"오직 {territory_radius_m}m 영업구역 기준만 사용.\n"
+        )
+    else:
+        _territory_criteria = (
+            "- 500m 내 동일 브랜드 1개 이상 + 자기잠식률 ≤-5% → danger (제12조의4 인접 출점)\n"
+            "- 500m 내 동일 브랜드 1개 이상 → caution (영업지역 협의 필요)\n"
+            "- 2km 내 동일 브랜드 3개 이상 → caution (자기잠식 위험)\n"
+        )
+
     user_content = (
         f"브랜드: {brand}\n"
         f"업종: {business_type}\n"
@@ -370,8 +386,7 @@ async def specialist_franchise_law(
         "[평가 기준]\n"
         "- 폐점률 ≥20% → danger 검토\n"
         "- 폐점률 ≥10% → caution\n"
-        "- 500m 내 동일 브랜드 1개 이상 + 자기잠식률 ≤-5% → danger (제12조의4 인접 출점)\n"
-        "- 500m 내 동일 브랜드 1개 이상 → caution (영업지역 협의 필요)\n"
+        f"{_territory_criteria}"
         "- 영업지역 침해(제12조의4)/허위과장(제9조)/필수품목 구입강제(제12조) → danger 후보\n"
         "- 신규 브랜드/직영 → safe~caution\n\n"
         "<<<RAG_CONTEXT>>>\n"

backend/src/agents/nodes/competitor_intel.py

@@ -184,11 +184,29 @@ async def _try_cache_set(cache_key: str, payload: dict) -> None:
                 pass
 
 
-def _run_data_collection(dong_code: str, brand_name: str, keyword: str, ind_code: str | None, ind_label: str) -> dict:
-    """Python 서비스 4개 호출 (동기). asyncio.to_thread 로 래핑해서 사용."""
+def _run_data_collection(
+    dong_code: str,
+    brand_name: str,
+    keyword: str,
+    ind_code: str | None,
+    ind_label: str,
+    spot_lat: float | None = None,
+    spot_lon: float | None = None,
+) -> dict:
+    """Python 서비스 4개 호출 (동기). asyncio.to_thread 로 래핑해서 사용.
+
+    spot_lat/spot_lon 입력 시 카니발리제이션 = 매물 좌표 기준 (legal agent 와 일관).
+    좌표 없으면 행정동 centroid 폴백.
+    """
     comp_500 = analyze_competition(dong_code, keyword, radius_m=500)
     comp_1000 = analyze_competition(dong_code, keyword, radius_m=1000)
-    cannibal = analyze_cannibalization(dong_code, brand_name, radius_m=2000, industry=ind_label)
+    if spot_lat is not None and spot_lon is not None:
+        from src.services.commercial_intelligence import analyze_cannibalization_at
+        cannibal = analyze_cannibalization_at(spot_lat, spot_lon, brand_name, radius_m=2000, industry=ind_label)
+        if not cannibal or "error" in cannibal:
+            cannibal = analyze_cannibalization(dong_code, brand_name, radius_m=2000, industry=ind_label)
+    else:
+        cannibal = analyze_cannibalization(dong_code, brand_name, radius_m=2000, industry=ind_label)
     brand_bench = get_brand_benchmark(brand_name)
     peers: list[dict] = []
     if brand_bench.get("benchmark_available") and brand_bench.get("industry_medium"):
@@ -322,6 +340,25 @@ def _make_competitor_attr(
             "agent_attribution": _attr,
         }
 
+    # winner 매물 좌표 (legal agent 와 동일 기준점) — 일관성 확보.
+    # state.vacancy_spots 에서 winner_district 매칭 spot 추출, 없으면 첫 spot.
+    spot_lat: float | None = None
+    spot_lon: float | None = None
+    _vac_spots = state.get("vacancy_spots") or []
+    if isinstance(_vac_spots, list) and _vac_spots and target_district:
+        _matched = [
+            s for s in _vac_spots if isinstance(s, dict) and s.get("dong_name") == target_district
+        ]
+        _spot = _matched[0] if _matched else _vac_spots[0]
+        try:
+            _slat = _spot.get("lat") if isinstance(_spot, dict) else None
+            _slon = _spot.get("lon") if isinstance(_spot, dict) else None
+            if _slat is not None and _slon is not None:
+                spot_lat = float(_slat)
+                spot_lon = float(_slon)
+        except (TypeError, ValueError):
+            pass
+
     # dong 이름 → code
     dong_code = resolve_dong_code(target_district)
     if not dong_code:
@@ -361,9 +398,10 @@ def _make_competitor_attr(
         }
 
     # Redis 캐시 조회
-    # v2 → v3: brand_mapping_resolver fix (resolve_brand_name 0차 호출 + canonical 통일) 반영.
-    # v2 캐시는 카니발 0 (브랜드 매칭 실패 결과) 저장되어 있어 무효화 필요.
-    cache_key = f"v3:competitor_intel:{dong_code}:{brand_name}"
+    # v3 → v4: 카니발리제이션 기준점 = winner 매물 좌표 (legal agent 와 일관).
+    # 캐시 키에 spot 좌표 포함 — 매물 변경 시 자동 invalidation.
+    _spot_key = f"{spot_lat:.5f},{spot_lon:.5f}" if spot_lat is not None and spot_lon is not None else "centroid"
+    cache_key = f"v4:competitor_intel:{dong_code}:{brand_name}:{_spot_key}"
     cached = await _try_cache_get(cache_key)
     if cached:
         print(f"[competitor_intel] 캐시 히트: {cache_key}")
@@ -387,7 +425,9 @@ def _make_competitor_attr(
 
     # Python 서비스 데이터 수집 (동기 DB 호출이라 별도 스레드)
     try:
-        data = await asyncio.to_thread(_run_data_collection, dong_code, brand_name, keyword, ind_code, ind_label)
+        data = await asyncio.to_thread(
+            _run_data_collection, dong_code, brand_name, keyword, ind_code, ind_label, spot_lat, spot_lon
+        )
     except Exception as e:
         logger.exception(f"[competitor_intel] 서비스 호출 실패: {e}")
         _attr = _make_competitor_attr(

backend/src/agents/nodes/legal.py

@@ -958,20 +958,16 @@ def _safe_ftc(r: object) -> dict:
     )
     ftc_result = _safe_ftc(_ftc_raw)
 
-    # 룰엔진 결과 invariant 검증용 — orchestrator._RULE_ENGINE_ORDER 와 동일 13종.
+    # 룰엔진 결과 invariant 검증용 — orchestrator._RULE_ENGINE_ORDER 와 동일 8종 (location 한정).
+    # 운영(operation) 카테고리 5종 (food_hygiene/labor/vat/privacy/sewage) 제외 — LLM·시간 절감.
     _BATCH_TYPES = [
         "franchise_law",
         "commercial_lease_law",
-        "food_hygiene",
         "safety_regulation",
         "building_law",
         "fire_safety_law",
-        "labor_law",
-        "vat_law",
-        "privacy_law",
         "accessibility_law",
         "school_zone",
-        "sewage_law",
         "fair_trade_law",
     ]
 
@@ -1097,9 +1093,71 @@ def _r(type_name: str) -> dict:
     analysis = dict(state.get("analysis_results") or {})
     analysis["legal_risks"] = risks
     analysis["overall_legal_risk"] = overall_level
-    # 4 vacancy spot 카니발리제이션 사전 스캔 결과 — franchise_law 보완 정보로 함께 노출.
+    # 4 vacancy spot 카니발리제이션 사전 스캔 결과 — 각 spot rank + level + summary 부여.
+    # rank = district_ranking scouting_results 점수 순위 (1등 = 점수 최고).
+    # 임계: territory 안 ≥1 → danger, 0 → safe(territory 입력 시) / caution(territory 미입력 + 500m 안 ≥1).
     if vacancy_spot_analyses:
-        analysis["vacancy_spot_cannibalization"] = vacancy_spot_analyses
+        # 동 → rank 매핑 (district_ranking scouting_results 기반).
+        _scout = state.get("scouting_results") or []
+        _dong_rank: dict[str, int] = {}
+        if isinstance(_scout, list):
+            for _row in _scout:
+                if isinstance(_row, dict):
+                    _d = _row.get("district") or _row.get("dong_name")
+                    _r = _row.get("rank")
+                    if _d and isinstance(_r, int):
+                        _dong_rank[_d] = _r
+
+        spot_evaluations: list[dict] = []
+        for _va in vacancy_spot_analyses:
+            _terr = _va.get("territory_radius_m")
+            _within = _va.get("same_brand_within_territory")
+            _closest = _va.get("closest_m")
+            _2km = _va.get("same_brand_2000m", 0)
+            _500 = _va.get("same_brand_500m") or 0
+            if _terr and _within is not None:
+                if _within >= 1:
+                    _lvl = "danger"
+                    _msg = f"영업구역({_terr}m) 안 동일 브랜드 {_within}개 — 가맹사업법 제12조의4 명백 침해"
+                else:
+                    _lvl = "safe"
+                    _msg = f"영업구역({_terr}m) 안 동일 브랜드 0개 — 정보공개서 기준 침해 없음"
+            else:
+                if _500 >= 1:
+                    _lvl = "caution"
+                    _msg = f"500m 내 동일 브랜드 {_500}개 — 영업지역 협의 필요"
+                elif _2km >= 3:
+                    _lvl = "caution"
+                    _msg = f"2km 내 동일 브랜드 {_2km}개 — 자기잠식 위험"
+                else:
+                    _lvl = "safe"
+                    _msg = "500m 내 동일 브랜드 0개"
+            _dong = _va.get("dong_name")
+            _rank = _dong_rank.get(_dong)
+            spot_evaluations.append({
+                "rank": _rank,
+                "rank_label": f"{_rank}등" if _rank else "순위 미정",
+                "dong_name": _dong,
+                "lat": _va.get("lat"),
+                "lon": _va.get("lon"),
+                "territory_radius_m": _terr,
+                "same_brand_within_territory": _within,
+                "same_brand_500m": _500,
+                "same_brand_2000m": _2km,
+                "closest_m": _closest,
+                "level": _lvl,
+                "summary": _msg,
+            })
+
+        # rank 순 정렬 (1등 → 4등). rank None 은 뒤로.
+        spot_evaluations.sort(key=lambda x: x.get("rank") or 999)
+
+        analysis["vacancy_spot_cannibalization"] = spot_evaluations
+        # franchise_law risk 에 spot_evaluations attach — frontend 가 1등/2등/3등/4등 카드 렌더 가능.
+        for _r in risks:
+            if isinstance(_r, dict) and _r.get("type") == "franchise_law":
+                _r["spot_evaluations"] = spot_evaluations
+                break
 
     # Redis 캐시 저장 — RAG 실패 시 빈 articles가 캐시되는 것을 방지
     # articles가 있는 리스크가 3개 미만이면 캐시하지 않음 (재실행 시 정상 결과 기대)

backend/src/main.py

@@ -964,9 +964,7 @@ async def analyze_location(input_data: SimulationInput, response: Response):
         result["all_competitor_locations"] = await _collect_all_competitor_locations(
             winner, top3, input_data.business_type
         )
-        result["same_brand_locations"] = await _collect_same_brand_locations(
-            winner, top3, input_data.brand_name
-        )
+        result["same_brand_locations"] = await _collect_same_brand_locations(winner, top3, input_data.brand_name)
         return {"status": "success", "data": result}
     except Exception as e:
         print(f"!!! [API ERROR] !!! {str(e)}")
@@ -1031,9 +1029,7 @@ async def analyze_llm(input_data: SimulationInput):
         print(f"[ANALYZE/LLM] all_competitor_locations 수집 실패 (무시): {e}")
         full["all_competitor_locations"] = []
     try:
-        full["same_brand_locations"] = await _collect_same_brand_locations(
-            winner, top3, input_data.brand_name
-        )
+        full["same_brand_locations"] = await _collect_same_brand_locations(winner, top3, input_data.brand_name)
     except Exception as e:
         print(f"[ANALYZE/LLM] same_brand_locations 수집 실패 (무시): {e}")
         full["same_brand_locations"] = []
@@ -1126,9 +1122,7 @@ async def _run() -> None:
                 logger.warning(f"[/analyze/llm/async] all_competitor_locations 실패 (무시): {ce}")
                 full["all_competitor_locations"] = []
             try:
-                full["same_brand_locations"] = await _collect_same_brand_locations(
-                    winner, top3, input_data.brand_name
-                )
+                full["same_brand_locations"] = await _collect_same_brand_locations(winner, top3, input_data.brand_name)
             except Exception as ce:
                 logger.warning(f"[/analyze/llm/async] same_brand_locations 실패 (무시): {ce}")
                 full["same_brand_locations"] = []
@@ -1868,9 +1862,7 @@ async def run_simulation(input_data: SimulationInput, response: Response):
         winner = result.get("winner_district") or input_data.target_district
         top3 = result.get("top_3_candidates") or []
         try:
-            result["same_brand_locations"] = await _collect_same_brand_locations(
-                winner, top3, input_data.brand_name
-            )
+            result["same_brand_locations"] = await _collect_same_brand_locations(winner, top3, input_data.brand_name)
         except Exception as ce:
             logger.warning(f"[/simulate] same_brand_locations 실패 (무시): {ce}")
             result["same_brand_locations"] = []
@@ -2290,8 +2282,10 @@ async def run_abm_simulation(req: AbmSimulationRequest):
     try:
         async with aioredis.from_url(settings.redis_url, decode_responses=True) as r:
             cache_body = {k: v for k, v in response.items() if k != "trajectory"}
-            await r.setex(cache_key, 3600, _json.dumps(cache_body, ensure_ascii=False))
-            logger.info(f"[ABM] cache SET key={cache_key[:16]}... ttl=3600s")
+            # 사용자 피드백 (2026-05-04): TTL 1h → 24h. ABM 시뮬 비용 큼 (gpt-4.1-mini ~$0.05/회) →
+            # 같은 시나리오 재시뮬 회피 위해 캐시 더 오래 유지.
+            await r.setex(cache_key, 86400, _json.dumps(cache_body, ensure_ascii=False))
+            logger.info(f"[ABM] cache SET key={cache_key[:16]}... ttl=86400s")
     except Exception as e:
         logger.warning(f"[ABM] Redis 캐시 저장 실패(무시): {e}")
 

backend/src/services/abm_simulation_service.py

@@ -236,7 +236,8 @@ def _save_to_redis(*, cache_key: str, redis_url: str, response: dict[str, Any])
         # daemon thread 안이라 sync redis 가 안전 (asyncio loop 부재)
         client = _redis_sync.from_url(redis_url, decode_responses=True)
         try:
-            client.setex(cache_key, 3600, _json.dumps(cache_body, ensure_ascii=False))
+            # TTL 1h → 24h (2026-05-04 사용자 피드백) — main.py 와 동기.
+            client.setex(cache_key, 86400, _json.dumps(cache_body, ensure_ascii=False))
             logger.info(f"[ABM async] redis SET key={cache_key[:16]}... ttl=3600s")
         finally:
             client.close()

backend/tests/test_legal_orchestrator.py

@@ -83,7 +83,7 @@ async def test_returns_13_items(patch_specialists):
         store_area_pyeong=20.0,
         ftc_data=None,
     )
-    assert len(risks) == 13
+    assert len(risks) == 8
     types = [r["type"] for r in risks]
     # 순서 확인
     assert types == _RULE_ENGINE_ORDER
@@ -100,8 +100,11 @@ async def test_types_unique_and_complete(patch_specialists):
     )
     types = {r["type"] for r in risks}
     assert types == set(_RULE_ENGINE_ORDER)
-    assert len(types) == 13
+    assert len(types) == 8
     assert "school_zone" in types
+    # 운영 카테고리 5종 제외 검증
+    for excluded in ("food_hygiene", "labor_law", "vat_law", "privacy_law", "sewage_law"):
+        assert excluded not in types
 
 
 @pytest.mark.asyncio
@@ -144,15 +147,15 @@ async def test_specialist_failure_isolated(monkeypatch):
         store_area_pyeong=20.0,
         ftc_data=None,
     )
-    assert len(risks) == 13
+    assert len(risks) == 8
     franchise = next(r for r in risks if r["type"] == "franchise_law")
     assert franchise["level"] == "caution"
     assert franchise.get("is_fallback") is True
     # 나머지 12 항목 중 룰 9 개는 정상 (fallback 아님)
     others = [r for r in risks if r["type"] != "franchise_law"]
     rules_only = [r for r in others if r["type"] in {
-        "food_hygiene", "safety_regulation", "fire_safety_law", "accessibility_law",
-        "school_zone", "commercial_lease_law", "labor_law", "vat_law", "sewage_law",
+        "safety_regulation", "fire_safety_law", "accessibility_law",
+        "school_zone", "commercial_lease_law",
     }]
     for r in rules_only:
         assert not r.get("is_fallback"), f"{r['type']} 이 예기치 않게 fallback 처리됨"
@@ -173,7 +176,7 @@ async def test_multiple_specialist_failures(monkeypatch):
         store_area_pyeong=10.0,
         ftc_data=None,
     )
-    assert len(risks) == 13
+    assert len(risks) == 8
     types = {r["type"] for r in risks}
     assert "franchise_law" in types
     assert "fair_trade_law" in types