feat(ai): Single-turn Router → Supervisor 패턴 전환 (#51)

backend-ai/app/api/routes.py

@@ -39,7 +39,7 @@ def agent_chat(request: AgentChatRequest) -> AgentChatResponse:
     }
     result = get_agent_graph().invoke(state)
     return AgentChatResponse(
-        intent=result["intent"],
+        workersCalled=result.get("workers_called", []),
         answer=result["answer"],
         properties=result.get("properties", []),
         legalCards=result.get("legal_cards", []),

backend-ai/app/api/schemas.py

@@ -2,8 +2,6 @@
 
 from pydantic import BaseModel, ConfigDict, Field
 
-from app.graph.state import Intent
-
 
 class RecentMessage(BaseModel):
     role: str
@@ -27,7 +25,7 @@ class AgentChatRequest(BaseModel):
 
 
 class AgentChatResponse(BaseModel):
-    intent: Intent
+    workers_called: list[str] = Field(default_factory=list, alias="workersCalled")
     answer: str
     properties: list[dict[str, Any]] = Field(default_factory=list)
     legal_cards: list[dict[str, Any]] = Field(default_factory=list, alias="legalCards")

backend-ai/app/clients/llm_client.py

@@ -5,7 +5,7 @@
 import httpx
 
 from app.core.config import get_settings
-from app.graph.state import AgentState, Intent
+from app.graph.state import AgentState
 from app.rag.prompts import (
     build_analysis_answer_prompt,
     build_legal_rag_prompt,
@@ -16,6 +16,29 @@
 
 HttpPost = Callable[..., httpx.Response]
 
+SUPERVISOR_PROMPT = """\
+다음 사용자 메시지와 지금까지 실행된 워커 목록을 보고, 다음에 호출할 워커를 결정해줘.
+
+사용자 메시지: {message}
+이미 실행된 워커: {workers_called}
+
+사용 가능한 워커:
+- PROPERTY_SEARCH: 매물 추천·검색·조건 필터링 (지역, 가격, 면적, 타입 등)
+- LEGAL_CONSULT: 임대차 법률, 계약, 보증금, 대항력, 갱신 등 법률 질문
+- PRICE_ANALYSIS: 특정 지역·매물의 시세·실거래가·가격 적정성 분석
+- SAFETY_ANALYSIS: 주변 치안, CCTV, 안전시설, 범죄율 등 생활 안전 분석
+- GENERAL_CHAT: 인사, 잡담, 서비스 소개 등 부동산과 무관한 일반 대화
+- FINISH: 충분한 정보가 모였으므로 답변 생성 단계로 이동
+
+규칙:
+- 이미 실행된 워커는 다시 선택하지 마.
+- 사용자 의도를 처리하기에 충분한 워커가 실행됐으면 FINISH를 선택해.
+- 워커가 하나도 실행되지 않았으면 반드시 워커 하나를 선택해.
+
+JSON만 반환해. 설명 없이:
+{{"next_worker": "...", "reasoning": "이유 한 줄"}}\
+"""
+
 CLASSIFY_INTENT_PROMPT = """\
 다음 사용자 메시지를 읽고, 부동산 AI 어시스턴트 관점에서 의도를 분류해줘.
 
@@ -73,6 +96,41 @@ def __init__(
         self.timeout_seconds = timeout_seconds
         self.http_post = http_post
 
+    def decide_next_worker(self, message: str, workers_called: list[str]) -> str:
+        prompt = SUPERVISOR_PROMPT.format(
+            message=message,
+            workers_called=", ".join(workers_called) if workers_called else "없음",
+        )
+        valid = {"PROPERTY_SEARCH", "LEGAL_CONSULT", "PRICE_ANALYSIS", "SAFETY_ANALYSIS", "GENERAL_CHAT", "FINISH"}
+        try:
+            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": 128,
+                },
+                timeout=self.timeout_seconds,
+            )
+            response.raise_for_status()
+            text = extract_chat_completion_text(response.json()) or "{}"
+            parsed = json.loads(text)
+            next_worker = parsed.get("next_worker", "FINISH")
+            if next_worker not in valid or next_worker in workers_called:
+                return "FINISH"
+            return next_worker
+        except (httpx.HTTPError, json.JSONDecodeError, KeyError, TypeError, ValueError):
+            # 아직 아무 워커도 실행되지 않은 첫 호출에서 장애가 나면 FINISH로 보내면
+            # workers_called=[]인 채로 generate_answer에 도달해 fallback 메시지만 반환됨.
+            # 기본 워커로 라우팅해 최소한의 응답을 보장한다.
+            if not workers_called:
+                return "PROPERTY_SEARCH"
+            return "FINISH"
+
     def classify(self, message: str) -> dict[str, Any] | None:
         prompt = CLASSIFY_INTENT_PROMPT.format(message=message)
         try:
@@ -120,29 +178,72 @@ def extract_property_criteria(self, message: str) -> dict[str, Any]:
             return {}
 
     def generate_answer(self, state: AgentState) -> str:
-        intent = state.get("intent", Intent.FALLBACK)
-        if intent == Intent.LEGAL_CONSULT:
+        workers_called = state.get("workers_called", [])
+
+        if "GENERAL_CHAT" in workers_called:
+            live = self._generate_live_general_chat_answer(state)
+            if live:
+                return live
+            return "안녕하세요! 살만해 부동산 AI입니다. 매물 추천, 법률 상담, 시세 분석, 안전 분석을 도와드릴 수 있습니다."
+
+        parts: list[str] = []
+
+        if "LEGAL_CONSULT" in workers_called:
             live_answer = self._generate_live_legal_answer(state)
-            if live_answer:
-                return live_answer
-            return generate_legal_answer(state)
-        if intent == Intent.PROPERTY_SEARCH:
-            count = len(state.get("properties", []))
-            return f"조건에 맞는 매물 {count}개를 찾았습니다."
-        if intent == Intent.PRICE_ANALYSIS:
-            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:
+            parts.append(live_answer if live_answer else generate_legal_answer(state))
+
+        if "PRICE_ANALYSIS" in workers_called or "SAFETY_ANALYSIS" in workers_called:
             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에서는 관련 조건 안내까지만 제공합니다."
+                parts.append(live_answer)
+            else:
+                if "PRICE_ANALYSIS" in workers_called:
+                    parts.append(AnalysisAnswerService().generate_price_answer(state))
+                if "SAFETY_ANALYSIS" in workers_called:
+                    parts.append(AnalysisAnswerService().generate_safety_answer(state))
+
+        if "PROPERTY_SEARCH" in workers_called:
+            count = len(state.get("properties", []))
+            parts.append(f"조건에 맞는 매물 {count}개를 찾았습니다.")
+
+        if parts:
+            return "\n\n".join(parts)
+
         return "질문 의도를 조금 더 구체화해 주세요. 매물 추천, 법률 상담, 시세 분석, 안전 분석을 도와드릴 수 있습니다."
 
+    def _generate_live_general_chat_answer(self, state: AgentState) -> str | None:
+        if not self.api_key or not self.model:
+            return None
+        try:
+            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": "system",
+                            "content": (
+                                "당신은 살만해 부동산 AI 어시스턴트입니다. "
+                                "매물 추천, 임대차 법률 상담, 시세 분석, 안전 분석을 도와줍니다. "
+                                "일반 대화나 인사에는 친절하게 응답하고, 부동산 관련 질문으로 자연스럽게 유도하세요. "
+                                "한국어로 간결하게 답변하세요."
+                            ),
+                        },
+                        {"role": "user", "content": state["message"]},
+                    ],
+                    "max_completion_tokens": 300,
+                },
+                timeout=self.timeout_seconds,
+            )
+            response.raise_for_status()
+            return extract_chat_completion_text(response.json())
+        except (httpx.HTTPError, KeyError, TypeError, ValueError):
+            return None
+
     def _generate_live_legal_answer(self, state: AgentState) -> str | None:
         legal_cards = state.get("legal_cards", [])
         if not self.api_key or not self.model or not legal_cards:

backend-ai/app/clients/supabase_client.py

@@ -77,10 +77,11 @@ def _similarity_search_legal_documents_pgvector(
             connect_timeout=self.connect_timeout_seconds,
         ) as conn:
             with conn.cursor() as cursor:
-                cursor.execute(
-                    "set local statement_timeout = %s",
-                    (self.statement_timeout_ms,),
-                )
+                cursor.execute(f"SET LOCAL statement_timeout = {int(self.statement_timeout_ms)}")
+                # IVFFlat 인덱스가 lists=100으로 설정돼 있으나 데이터 수가 적을 때
+                # 기본 probes=1이면 대부분의 클러스터를 건너뛰어 결과가 0개가 됨.
+                # probes를 lists 값과 동일하게 설정해 전체 인덱스를 탐색하도록 한다.
+                cursor.execute("SET LOCAL ivfflat.probes = 100")
                 cursor.execute(sql, (vector_literal, vector_literal, top_k))
                 return list(cursor.fetchall())
 

backend-ai/app/graph/builder.py

@@ -1,70 +1,54 @@
 from langgraph.graph import END, START, StateGraph
 
-from app.graph.nodes.classify_intent import classify_intent
+from app.graph.nodes.general_chat import general_chat
 from app.graph.nodes.generate_answer import generate_answer
 from app.graph.nodes.legal_rag import legal_rag
 from app.graph.nodes.price_analysis import price_analysis
 from app.graph.nodes.property_search import property_search
 from app.graph.nodes.safety_analysis import safety_analysis
-from app.graph.state import AgentState, Intent
-
-
-def route_by_intent(state: AgentState) -> str:
-    intent = state.get("intent", Intent.FALLBACK)
-    return {
-        Intent.PROPERTY_SEARCH: "property_search",
-        Intent.LEGAL_CONSULT: "legal_rag",
-        Intent.PRICE_ANALYSIS: "price_analysis",
-        Intent.SAFETY_ANALYSIS: "safety_analysis",
-        Intent.HUG_CALC: "fallback",
-        Intent.GENERAL_CHAT: "fallback",
-        Intent.FALLBACK: "fallback",
-    }[intent]
-
-
-def fallback(state: AgentState) -> AgentState:
-    next_actions = state.get("next_actions", [])
-    next_actions.append(
-        {
-            "type": "ASK_CLARIFYING_QUESTION",
-            "label": "질문 구체화",
-        }
-    )
-    return {
-        **state,
-        "tool_results": {
-            **state.get("tool_results", {}),
-            "fallback": {"reason": "No MVP tool is available for this intent yet."},
-        },
-        "next_actions": next_actions,
+from app.graph.nodes.supervisor import supervisor
+from app.graph.state import AgentState
+
+
+def route_after_supervisor(state: AgentState) -> str:
+    mapping = {
+        "PROPERTY_SEARCH": "property_search",
+        "LEGAL_CONSULT": "legal_rag",
+        "PRICE_ANALYSIS": "price_analysis",
+        "SAFETY_ANALYSIS": "safety_analysis",
+        "GENERAL_CHAT": "general_chat",
+        "FINISH": "generate_answer",
     }
+    return mapping.get(state.get("next_worker", "FINISH"), "generate_answer")
 
 
 def build_agent_graph():
     workflow = StateGraph(AgentState)
-    workflow.add_node("classify_intent", classify_intent)
+
+    workflow.add_node("supervisor", supervisor)
     workflow.add_node("property_search", property_search)
     workflow.add_node("legal_rag", legal_rag)
     workflow.add_node("price_analysis", price_analysis)
     workflow.add_node("safety_analysis", safety_analysis)
-    workflow.add_node("fallback", fallback)
+    workflow.add_node("general_chat", general_chat)
     workflow.add_node("generate_answer", generate_answer)
 
-    workflow.add_edge(START, "classify_intent")
+    workflow.add_edge(START, "supervisor")
     workflow.add_conditional_edges(
-        "classify_intent",
-        route_by_intent,
+        "supervisor",
+        route_after_supervisor,
         {
             "property_search": "property_search",
             "legal_rag": "legal_rag",
             "price_analysis": "price_analysis",
             "safety_analysis": "safety_analysis",
-            "fallback": "fallback",
+            "general_chat": "general_chat",
+            "generate_answer": "generate_answer",
         },
     )
 
-    for node_name in ["property_search", "legal_rag", "price_analysis", "safety_analysis", "fallback"]:
-        workflow.add_edge(node_name, "generate_answer")
+    for node in ["property_search", "legal_rag", "price_analysis", "safety_analysis", "general_chat"]:
+        workflow.add_edge(node, "supervisor")
 
     workflow.add_edge("generate_answer", END)
     return workflow.compile()

backend-ai/app/graph/nodes/general_chat.py

@@ -0,0 +1,5 @@
+from app.graph.state import AgentState
+
+
+def general_chat(state: AgentState) -> AgentState:
+    return state

backend-ai/app/graph/nodes/supervisor.py

@@ -0,0 +1,22 @@
+import logging
+
+from app.clients.llm_client import LLMClient
+from app.graph.state import AgentState
+
+logger = logging.getLogger(__name__)
+
+
+def supervisor(state: AgentState) -> AgentState:
+    workers_called = state.get("workers_called", [])
+    call_no = len(workers_called) + 1
+    logger.info("[supervisor #%d] 호출됨 | workers_called=%s", call_no, workers_called)
+
+    next_worker = LLMClient().decide_next_worker(
+        message=state["message"],
+        workers_called=workers_called,
+    )
+    logger.info("[supervisor #%d] → next_worker=%s", call_no, next_worker)
+
+    if next_worker != "FINISH":
+        workers_called = [*workers_called, next_worker]
+    return {**state, "next_worker": next_worker, "workers_called": workers_called}