[이미지기반 카페추천 프로젝트] ADK 기반 에이전트 챗봇 구현

def run(self, user_query: str) -> dict:
    # 1. LLM 호출
    prompt_result = self._call_llm(user_query)

    # 2. JSON 파싱 (실패 시 {}~{} 패턴 추출)
    try:
        result_json = json.loads(prompt_result)
    except JSONDecodeError:
        start = prompt_result.find('{')
        end = prompt_result.rfind('}') + 1
        result_json = json.loads(prompt_result[start:end])

    # 3. 결과 반환
    return {
        "is_location_query": result_json.get("is_location_query", False),
        "region_names": result_json.get("region_names", []),
        "original_query": user_query
    }

class AgentState(BaseModel):
    session_id: str              # 세션 고유 ID
    user_query: str              # 현재 사용자 쿼리

    # === Shared Memory (턴 간 유지) ===
    history_place_ids: List[int]       # 이미 추천된 장소 ID (중복 방지)
    last_recommendations: List[Dict]   # 최근 추천 결과 (QnA 참조용)
    current_location_filter: List[str] # 현재 동네 필터 ["서교", "연남"]
    previous_hyde_query: str           # 직전 HyDE 쿼리 (분위기 맥락 유지)
    file_path: str                     # 업로드된 이미지 경로

    # === 현재 턴 상태 ===
    intent: AgentIntent          # SEARCH | FEEDBACK | QNA
    final_response: Dict         # 최종 응답

def save_state(state: AgentState):
    state_json = state.model_dump_json()

    con.execute("""
        INSERT INTO sessions (session_id, state_json, updated_at)
        VALUES (?, ?, CURRENT_TIMESTAMP)
        ON CONFLICT (session_id) DO UPDATE SET
            state_json = excluded.state_json,
            updated_at = excluded.updated_at
    """, (state.session_id, state_json))

@router.post("/", response_model=RecommendedPlacesResponse)
async def create_chat(session_id, user_query, file):
    # 1. 상태 로드/생성
    init_state = context_mgr.get_or_create_state(session_id, user_query)
    init_state.file_path = file_path_disk
    context_mgr.save_state(init_state)

    # 2. ADK Runner 설정 & 실행
    runner = InMemoryRunner(agent=adk_app, app_name="vibe_pick")
    user_msg = types.Content(role="user", parts=[types.Part(text=user_query)])

    async for event in runner.run_async(user_id, session_id, new_message=user_msg):
        pass  # 에이전트 순차 실행

    # 3. 결과 추출
    final_state = context_mgr.get_state(session_id)
    recommendations = final_state.final_response.get("recommendations", [])

    # 4. 추천 질문 생성 (Suggestion)
    suggestions = generate_suggestions(user_query, place_names[:3], hyde_query)

    # 5. 응답 반환
    return RecommendedPlacesResponse(
        recommendations=recommendations,
        suggestions=suggestions,
        ...
    )

from google.adk.agents.loop_agent import LoopAgent

router = ADKRouterAgent(name="router")
location = ADKLocationAgent(name="location")
search = ADKSearchAgent(name="search")
qna = ADKQnAAgent(name="qna")

app = LoopAgent(
    name="vibe_pick_agent",
    sub_agents=[router, location, search, qna],
    max_iterations=5  # 무한루프 방지를 위한 안전장치
)

async def _run_async_impl(self, ctx: InvocationContext):
    session_id = ctx.session.id

    # 1. 텍스트/이미지 추출
    user_query = ""
    image_path = None
    for part in ctx.user_content.parts:
        if getattr(part, "text", None):
            user_query += part.text
        if getattr(part, "inline_data", None):
            # 이미지 저장: temp_images/{uuid}.jpg
            image_path = save_image(part.inline_data)

    # 2. 상태 로드
    state = context_mgr.get_or_create_state(session_id, user_query)
    state.file_path = image_path

    # 3. 라우팅 실행
    intent = self.router_logic.route(state)  # Router.route() 호출
    state.intent = intent  # SEARCH | FEEDBACK | QNA

    # 4. 상태 저장
    context_mgr.save_state(state)

    yield Event(author=self.name, custom_metadata={"type": "routed"})

async def _run_async_impl(self, ctx: InvocationContext):
    state = context_mgr.get_state(ctx.session.id)

    # Intent가 SEARCH/FEEDBACK일 때만 실행
    if state.intent not in [AgentIntent.SEARCH, AgentIntent.FEEDBACK]:
        return  # 조기 반환

    # 위치 파싱
    loc_result = self.loc_agent.run(state.user_query)
    # {"is_location_query": true, "region_names": ["서교", "연남"]}

    new_filters = loc_result.get("region_names", [])
    if new_filters:
        state.current_location_filter.extend(new_filters)
        state.current_location_filter = list(set(state.current_location_filter))

    context_mgr.save_state(state)
    yield Event(author=self.name, custom_metadata={"type": "location_parsed"})

async def _run_async_impl(self, ctx: InvocationContext):
    state = context_mgr.get_state(ctx.session.id)

    if state.intent not in [AgentIntent.SEARCH, AgentIntent.FEEDBACK]:
        return

    # 이전 HyDE 컨텍스트 참조 (FEEDBACK 시 분위기 누적)
    prev_context = state.previous_hyde_query

    # 검색 실행
    hyde_query, recommendations = await hyde_search(
        state.user_query,           # "더 조용한 곳"
        state.file_path,            # 업로드된 이미지
        state.current_location_filter,  # ["서교", "연남"...]
        state.history_place_ids,    # 이미 추천한 장소 제외
        prev_context                # "홍대 모던한 인테리어..." (이전 HyDE)
    )

    # 상태 업데이트
    state.previous_hyde_query = hyde_query
    context_mgr.update_history(state, recommendations)
    state.final_response = {"hyde_query": hyde_query, "recommendations": recommendations}
    context_mgr.save_state(state)

    # 루프 종료 (escalate=True)
    yield Event(
        author=self.name,
        content=types.Content(parts=[types.Part(text=output_text)]),
        actions=EventActions(escalate=True)
    )

def run(self, state: AgentState) -> dict:
    query = state.user_query  # "첫 번째 카페 주차 돼?"

    # 1. 대상 장소 식별 (LLM 사용)
    target_places = self._identify_target_places(query, state.last_recommendations)
    # [{"place_id": 101, "place_name": "블루보틀 연남점"}]

    if not target_places:
        return {"message": "참고할 추천 결과가 없어서 답변드리기 어려워요."}

    # 2. 장소 상세 정보 조회 (DB)
    all_details = []
    for place in target_places:
        details = self._fetch_place_details(place['place_id'])
        # places: address, business_hours, contact
        # amenities: 주차, 예약, 와이파이 등 10개 항목
        # reviews: summary, keywords
        # menus: name, price
        all_details.append(f"=={place['place_name']} ==\n{details}")

    # 3. LLM 답변 생성
    answer = self._generate_answer(query, place_names, combined_details)
    # "네, 주차 가능합니다. 건물 지하주차장 이용 가능해요."

    return {"message": answer, "related_place_id": target_places[0]['place_id']}

def generate_suggestions(user_query, place_names, hyde_query=None) -> list[str]:
    # 1. Context 포맷팅
    place_str = ', '.join(place_names[:3])
    user_content = f"검색어:{user_query}\n추천 장소:{place_str}\n분위기:{hyde_query or '없음'}"

    # 2. 시스템 프롬프트
    system_prompt = """너는 카페 추천 서비스의 대화 어시스턴트야.
    사용자가 카페 추천을 받았어. 사용자가 다음에 물어볼 법한 질문 4개를 생성해줘.

    [질문 유형 - 반드시 다양하게 섞어서]
    1. 장소 상세 질문: 주차, 메뉴, 영업시간 등
    2. 결과 조정 요청: 더 조용한 곳, 더 넓은 곳, 분위기 다른 곳 등

    [규칙]
    - 각 질문은 30자 이내로 짧게
    - 자연스러운 구어체 ("첫 번째 카페 주차 가능해?", "더 아늑한 곳 없어?")
    - 순수 JSON 배열만 출력: ["질문1", "질문2", "질문3", "질문4"]
    """

    # 3. Clova API 호출 (temperature=0.7 다양성 확보)
    payload = {
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_content}
        ],
        "temperature": 0.7,
        "maxTokens": 100
    }

    res = requests.post(clova_api_url, headers=headers, json=payload)
    content = res.json()['result']['message']['content']

    # 4. JSON 파싱 (마크다운 코드블록 제거)
    suggestions = extract_json_array(content)

    # 5. Fallback (4개 미만 시 기본 질문으로 채움)
    if len(suggestions) < 4:
        defaults = ["주차 가능해?", "메뉴 추천해줘", "더 조용한 곳은?", "영업시간 알려줘"]
        return (suggestions + defaults)[:4]

    return suggestions[:4]

# 검색 완료 후 suggestions 생성
suggestions = None
if recommendations:
    place_names = [p.place_name for p in recommendations[:3]]
    try:
        suggestions = generate_suggestions(
            user_query=user_query,
            place_names=place_names,
            hyde_query=hyde_query
        )
    except Exception as e:
        suggestions = ["주차 가능해?", "메뉴 추천해줘", "더 조용한 곳은?", "영업시간 알려줘"]

return RecommendedPlacesResponse(
    ...
    suggestions=suggestions
)