Spaces:
Runtime error
Runtime error
File size: 5,482 Bytes
065853d f369cb4 065853d add744f 065853d add744f f369cb4 065853d f369cb4 065853d f369cb4 065853d f369cb4 065853d f369cb4 065853d f369cb4 065853d f369cb4 065853d f369cb4 065853d f369cb4 add744f f369cb4 add744f 065853d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """
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 |