[Phase 4] feat(ai): 분석 결과 기반 답변 생성

backend-ai/app/clients/llm_client.py

@@ -6,7 +6,12 @@
 
 from app.core.config import get_settings
 from app.graph.state import AgentState, Intent
-from app.rag.prompts import build_legal_rag_prompt, format_legal_context
+from app.rag.prompts import (
+    build_analysis_answer_prompt,
+    build_legal_rag_prompt,
+    format_legal_context,
+)
+from app.services.analysis_answer_service import AnalysisAnswerService
 
 
 HttpPost = Callable[..., httpx.Response]
@@ -125,9 +130,15 @@ def generate_answer(self, state: AgentState) -> str:
             count = len(state.get("properties", []))
             return f"조건에 맞는 매물 {count}개를 찾았습니다."
         if intent == Intent.PRICE_ANALYSIS:
-            return "선택한 매물 또는 지역의 실거래가를 기준으로 시세 적정성을 분석했습니다."
+            live_answer = self._generate_live_analysis_answer(state)
+            if live_answer:
+                return live_answer
+            return AnalysisAnswerService().generate_price_answer(state)
         if intent == Intent.SAFETY_ANALYSIS:
-            return "주변 안전시설 반경과 안전 점수를 기준으로 생활 안전성을 분석했습니다."
+            live_answer = self._generate_live_analysis_answer(state)
+            if live_answer:
+                return live_answer
+            return AnalysisAnswerService().generate_safety_answer(state)
         if intent == Intent.HUG_CALC:
             return "HUG 보증 가입 계산은 1.5차 범위입니다. MVP에서는 관련 조건 안내까지만 제공합니다."
         return "질문 의도를 조금 더 구체화해 주세요. 매물 추천, 법률 상담, 시세 분석, 안전 분석을 도와드릴 수 있습니다."
@@ -167,6 +178,36 @@ def _generate_live_legal_answer(self, state: AgentState) -> str | None:
         except (httpx.HTTPError, KeyError, TypeError, ValueError):
             return None
 
+    def _generate_live_analysis_answer(self, state: AgentState) -> str | None:
+        analysis_cards = state.get("analysis_cards", [])
+        tool_results = state.get("tool_results", {})
+        if not self.api_key or not self.model or not analysis_cards or not tool_results:
+            return None
+
+        try:
+            prompt = build_analysis_answer_prompt(
+                state["message"],
+                analysis_cards,
+                json.dumps(tool_results, ensure_ascii=False),
+            )
+            response = self.http_post(
+                f"{self.base_url}/chat/completions",
+                headers={
+                    "Authorization": f"Bearer {self.api_key}",
+                    "Content-Type": "application/json",
+                },
+                json={
+                    "model": self.model,
+                    "messages": [{"role": "user", "content": prompt}],
+                    "max_completion_tokens": 500,
+                },
+                timeout=self.timeout_seconds,
+            )
+            response.raise_for_status()
+            return extract_chat_completion_text(response.json())
+        except (httpx.HTTPError, KeyError, TypeError, ValueError):
+            return None
+
 
 def extract_chat_completion_text(payload: Any) -> str | None:
     if not isinstance(payload, dict):

backend-ai/app/rag/prompts.py

@@ -10,6 +10,14 @@
 Do not invent listings or legal facts.
 """
 
+ANALYSIS_ANSWER_SYSTEM_PROMPT = """\
+You are the Salmanhae F-4 price and safety analysis assistant.
+Answer in Korean using only the provided analysis cards and Spring Boot tool results.
+Mention concrete numbers from the tool results when available.
+Do not make HUG eligibility conclusions or legal-contract advice.
+If grounded facts are insufficient, say that the analysis data is insufficient.
+"""
+
 
 def format_legal_context(legal_cards: list[dict], max_cards: int = 3) -> str:
     if not legal_cards:
@@ -36,3 +44,19 @@ def build_legal_rag_prompt(question: str, legal_cards: list[dict]) -> str:
             "Add a short explanation and recommend 전문가 검토 for real contracts.",
         ]
     )
+
+
+def build_analysis_answer_prompt(
+    question: str,
+    analysis_cards: list[dict],
+    tool_results_json: str,
+) -> str:
+    return "\n\n".join(
+        [
+            ANALYSIS_ANSWER_SYSTEM_PROMPT.strip(),
+            f"User question:\n{question.strip()}",
+            f"Analysis cards:\n{analysis_cards}",
+            f"Spring Boot tool results JSON:\n{tool_results_json}",
+            "Answer in 2-4 concise Korean sentences. Use only grounded facts from the cards/results.",
+        ]
+    )

backend-ai/app/services/__init__.py

@@ -0,0 +1 @@
+

backend-ai/app/services/analysis_answer_service.py

@@ -0,0 +1,153 @@
+from typing import Any
+
+from app.graph.state import AgentState
+
+
+class AnalysisAnswerService:
+    def generate_price_answer(self, state: AgentState) -> str:
+        result = self._tool_result(state, "priceAnalysis")
+        card = self._analysis_card(state, "PRICE")
+        metrics = self._combined_metrics(result, card)
+        transactions = self._items(result.get("transactions"))
+        price_analysis = result.get("priceAnalysis", {})
+        if not isinstance(price_analysis, dict):
+            price_analysis = {}
+        region_stats = self._items(price_analysis.get("regionStats"))
+
+        facts: list[str] = []
+        comparable_count_metric = metrics.get("comparableTransactionCount")
+        if comparable_count_metric is not None:
+            comparable_count = comparable_count_metric
+        elif transactions:
+            comparable_count = len(transactions)
+        else:
+            comparable_count = None
+        if comparable_count is not None:
+            facts.append(f"최근 실거래 {comparable_count}건을 기준으로 확인했습니다.")
+
+        if transactions:
+            transaction = transactions[0]
+            parts: list[str] = []
+            contract_ym = transaction.get("contractYearMonth")
+            if contract_ym:
+                parts.append(str(contract_ym))
+            deposit = self._format_won(transaction.get("deposit"))
+            if deposit:
+                parts.append(f"보증금 {deposit}")
+            monthly_rent = self._format_won(transaction.get("monthlyRent"))
+            if monthly_rent:
+                parts.append(f"월세 {monthly_rent}")
+            area_m2 = transaction.get("areaM2")
+            if area_m2 is not None:
+                parts.append(f"전용면적 {area_m2}㎡")
+            if parts:
+                facts.append("최근 사례는 " + ", ".join(parts) + "입니다.")
+
+        if region_stats:
+            region_stat = region_stats[0]
+            parts = []
+            avg_deposit = self._format_won(region_stat.get("avgDeposit"))
+            if avg_deposit:
+                parts.append(f"지역 평균 보증금 {avg_deposit}")
+            avg_monthly_rent = self._format_won(region_stat.get("avgMonthlyRent"))
+            if avg_monthly_rent:
+                parts.append(f"지역 평균 월세 {avg_monthly_rent}")
+            transaction_count = region_stat.get("transactionCount")
+            if transaction_count is not None:
+                parts.append(f"통계 표본 {transaction_count}건")
+            if parts:
+                facts.append(", ".join(parts) + "입니다.")
+
+        if not facts:
+            return "분석할 근거 데이터가 부족합니다. 매물을 선택하거나 시세 데이터가 쌓인 뒤 다시 확인해 주세요."
+
+        return " ".join(facts) + " 보증보험 가능 여부나 법적 판단은 포함하지 않습니다."
+
+    def generate_safety_answer(self, state: AgentState) -> str:
+        result = self._tool_result(state, "safetyAnalysis")
+        card = self._analysis_card(state, "SAFETY")
+        safety_summary = result.get("safetySummary", {})
+        if not isinstance(safety_summary, dict):
+            safety_summary = {}
+        result_metrics = result.get("metrics", {})
+        if not isinstance(result_metrics, dict):
+            result_metrics = {}
+        metrics = {
+            **self._combined_metrics(result, card),
+            **safety_summary,
+            **result_metrics,
+        }
+
+        facts: list[str] = []
+        score = result.get("score")
+        if score is None:
+            score = safety_summary.get("safetyScore")
+        if score is None:
+            score = metrics.get("safetyScore")
+        if score is None:
+            score = card.get("score")
+        if score is not None:
+            facts.append(f"안전 점수 {score}점")
+
+        radius = metrics.get("radius")
+        if radius is not None:
+            facts.append(f"반경 {radius}m")
+
+        count_specs = [
+            ("cctvCount300m", "CCTV"),
+            ("bellCount300m", "비상벨"),
+            ("lightCount300m", "보안등"),
+            ("policeCount500m", "파출소"),
+        ]
+        for key, label in count_specs:
+            value = metrics.get(key)
+            if value is not None:
+                facts.append(f"{label} {value}개")
+
+        if not facts:
+            return "분석할 근거 데이터가 부족합니다. 매물을 선택하거나 안전 데이터가 쌓인 뒤 다시 확인해 주세요."
+
+        return "주변 안전 데이터는 " + ", ".join(facts) + "로 확인됩니다. 실제 체감 안전은 현장 환경에 따라 달라질 수 있습니다."
+
+    def _tool_result(self, state: AgentState, key: str) -> dict[str, Any]:
+        tool_results = state.get("tool_results", {})
+        if not isinstance(tool_results, dict):
+            return {}
+        value = tool_results.get(key, {})
+        return value if isinstance(value, dict) else {}
+
+    def _analysis_card(self, state: AgentState, card_type: str) -> dict[str, Any]:
+        analysis_cards = state.get("analysis_cards", [])
+        if not isinstance(analysis_cards, list):
+            return {}
+        for card in reversed(analysis_cards):
+            if isinstance(card, dict) and card.get("type") == card_type:
+                return card
+        return {}
+
+    def _combined_metrics(self, result: dict[str, Any], card: dict[str, Any]) -> dict[str, Any]:
+        result_metrics = result.get("metrics", {})
+        card_metrics = card.get("metrics", {})
+        if not isinstance(result_metrics, dict):
+            result_metrics = {}
+        if not isinstance(card_metrics, dict):
+            card_metrics = {}
+        return {**card_metrics, **result_metrics}
+
+    def _items(self, value: Any) -> list[dict[str, Any]]:
+        if not isinstance(value, list):
+            return []
+        return [item for item in value if isinstance(item, dict)]
+
+    def _format_won(self, value: Any) -> str | None:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            try:
+                value = float(value)
+            except ValueError:
+                return None
+        if not isinstance(value, (int, float)):
+            return None
+        amount = int(value) if float(value).is_integer() else value
+        return f"{amount:,}원"

backend-ai/tests/test_agent_chat.py

@@ -114,6 +114,24 @@ def analyze_price(self, message: str, context: dict) -> dict:
                     "regionStatCount": 1,
                     "buildingStatCount": 1,
                 },
+                "transactions": [
+                    {
+                        "contractYearMonth": "2026-05",
+                        "deposit": 10000000,
+                        "monthlyRent": 520000,
+                        "areaM2": 21.8,
+                    }
+                ],
+                "priceAnalysis": {
+                    "regionStats": [
+                        {
+                            "avgDeposit": 10500000,
+                            "avgMonthlyRent": 520000,
+                            "transactionCount": 3,
+                        }
+                    ],
+                    "buildingStats": [],
+                },
                 "stub": False,
             }
 
@@ -142,6 +160,8 @@ def analyze_price(self, message: str, context: dict) -> dict:
     assert card["metrics"]["selectedPropertyId"] == "1"
     assert card["metrics"]["comparableTransactionCount"] == 2
     assert card["metrics"]["stub"] is False
+    assert "최근 실거래 2건" in body["answer"]
+    assert "지역 평균 보증금 10,500,000원" in body["answer"]
 
 
 def test_agent_chat_returns_price_analysis_error_metric_on_fallback(monkeypatch) -> None:
@@ -187,6 +207,16 @@ def analyze_safety(self, message: str, context: dict) -> dict:
                 "metrics": {
                     "radius": 500,
                     "cctvCount300m": 8,
+                    "bellCount300m": 2,
+                    "lightCount300m": 14,
+                    "policeCount500m": 1,
+                },
+                "safetySummary": {
+                    "radius": 500,
+                    "safetyScore": 78,
+                    "cctvCount300m": 8,
+                    "bellCount300m": 2,
+                    "lightCount300m": 14,
                     "policeCount500m": 1,
                 },
                 "stub": False,
@@ -218,6 +248,8 @@ def analyze_safety(self, message: str, context: dict) -> dict:
     assert card["score"] == 78
     assert card["metrics"]["radius"] == 500
     assert card["metrics"]["stub"] is False
+    assert "안전 점수 78점" in body["answer"]
+    assert "CCTV 8개" in body["answer"]
 
 
 def test_classify_intent_fallback_returns_fallback() -> None: