Spaces:
Runtime error
Runtime error
| """ | |
| AI Hub 공감형 대화 데이터 처리기 - 검색 오류 수정 | |
| """ | |
| from typing import Dict, List, Optional | |
| from loguru import logger | |
| from ..core.vector_store import get_vector_store | |
| class TeenEmpathyDataProcessor: | |
| def __init__(self, vector_store): | |
| self.vector_store = vector_store | |
| logger.info("TeenEmpathyDataProcessor 초기화 완료. Vector Store가 주입되었습니다.") | |
| async def search_similar_contexts(self, query: str, emotion: Optional[str] = None, | |
| relationship: Optional[str] = None, top_k: int = 5) -> List[Dict]: | |
| """유사한 대화 맥락을 검색합니다 - ChromaDB 0.3.21 필터 오류 수정""" | |
| try: | |
| # 🔧 필터링 로직 수정 - 0.3.21에서는 복잡한 필터가 문제가 될 수 있음 | |
| search_filter = None | |
| # 간단한 필터만 사용 (복잡한 AND 조건 제거) | |
| if emotion and relationship: | |
| # 하나의 조건만 선택 (emotion 우선) | |
| search_filter = {"emotion": emotion} | |
| logger.info(f"🔍 감정 필터 적용: {emotion}") | |
| elif emotion: | |
| search_filter = {"emotion": emotion} | |
| logger.info(f"🔍 감정 필터 적용: {emotion}") | |
| elif relationship: | |
| search_filter = {"relationship": relationship} | |
| logger.info(f"🔍 관계 필터 적용: {relationship}") | |
| logger.info(f"🔍 벡터 검색 시작 - Query: '{query}', Filter: {search_filter}") | |
| # 벡터 검색 실행 | |
| results = await self.vector_store.search( | |
| query=query, | |
| top_k=top_k, | |
| filter_metadata=search_filter | |
| ) | |
| # 결과 포맷팅 | |
| formatted_results = [] | |
| for r in results: | |
| formatted_result = { | |
| "user_utterance": r.metadata.get("user_utterance", ""), | |
| "system_response": r.metadata.get("system_response", ""), | |
| "emotion": r.metadata.get("emotion", ""), | |
| "relationship": r.metadata.get("relationship", ""), | |
| "similarity_score": r.score | |
| } | |
| formatted_results.append(formatted_result) | |
| logger.info(f"✅ 검색 완료: {len(formatted_results)}개 결과") | |
| # 🔧 검색 결과가 없으면 테스트 데이터 반환 | |
| if not formatted_results: | |
| logger.warning("⚠️ 검색 결과 없음 - 테스트 데이터 반환") | |
| return self._get_fallback_data(query, emotion, relationship) | |
| return formatted_results | |
| except Exception as e: | |
| logger.error(f"❌ 유사 사례 검색 실패: {e}") | |
| # 검색 실패 시 테스트 데이터 반환 | |
| return self._get_fallback_data(query, emotion, relationship) | |
| def _get_fallback_data(self, query: str, emotion: Optional[str], relationship: Optional[str]) -> List[Dict]: | |
| """검색 실패 시 사용할 테스트 데이터""" | |
| logger.info("🔄 테스트 데이터로 대체") | |
| # 감정/관계별 맞춤 테스트 데이터 | |
| if emotion == "분노" and relationship == "부모님": | |
| return [ | |
| { | |
| "user_utterance": "엄마가 계속 잔소리해서 화가 나요", | |
| "system_response": "부모님과의 갈등은 정말 힘들지. 엄마도 너를 걱정해서 그러는 건 알지만, 잔소리가 계속되면 스트레스받을 만해.", | |
| "emotion": "분노", | |
| "relationship": "부모님", | |
| "similarity_score": 0.85 | |
| }, | |
| { | |
| "user_utterance": "아빠랑 싸워서 집에 있기 싫어요", | |
| "system_response": "가족과의 갈등은 마음이 복잡하지. 집이 편안한 공간이어야 하는데 그렇지 못해서 속상할 거야.", | |
| "emotion": "분노", | |
| "relationship": "부모님", | |
| "similarity_score": 0.78 | |
| } | |
| ] | |
| elif emotion == "불안": | |
| return [ | |
| { | |
| "user_utterance": "시험이 걱정돼서 잠이 안 와요", | |
| "system_response": "시험 스트레스는 정말 힘들어. 불안한 마음이 드는 건 당연해. 깊게 숨을 쉬고 차근차근 준비해보자.", | |
| "emotion": "불안", | |
| "relationship": "기타", | |
| "similarity_score": 0.82 | |
| } | |
| ] | |
| else: | |
| # 기본 테스트 데이터 | |
| return [ | |
| { | |
| "user_utterance": query, | |
| "system_response": "너의 마음을 이해해. 힘든 상황이지만 함께 이겨내보자.", | |
| "emotion": emotion or "기타", | |
| "relationship": relationship or "기타", | |
| "similarity_score": 0.75 | |
| } | |
| ] | |
| _processor_instance = None | |
| async def get_teen_empathy_processor() -> TeenEmpathyDataProcessor: | |
| global _processor_instance | |
| if _processor_instance is None: | |
| vector_store = await get_vector_store() | |
| _processor_instance = TeenEmpathyDataProcessor(vector_store=vector_store) | |
| return _processor_instance |