exAichatbot_agent/exAiChatBot-chatbot2.0-agent/scripts/handlers/response_handler.py

"""
응답 핸들러 모듈
──────────────
응답 생성, 저장, 로깅 처리
"""

import json
from pathlib import Path
from typing import List, Dict, Any, Optional


class ResponseHandler:
    """응답 처리 및 저장 핸들러"""
    
    def __init__(self, data_dir: Path, chat_manager=None, config=None):
        self.data_dir = data_dir
        self.chat_manager = chat_manager
        self.config = config

    def build_references(self, top_results: List[Dict], scores: List[float]) -> List[Dict[str, Any]]:
        """클라이언트 응답/로그용 참고자료 메타데이터 구성"""
        references = []
        for idx, result in enumerate(top_results):
            references.append({
                "question": result.get("q"),
                "answer": result.get("a"),
                "score": scores[idx] if idx < len(scores) else None,
                "category": result.get("category"),
                "source": result.get("source"),
                "source_id": result.get("source_id"),
                "quality": result.get("quality"),
                "url": result.get("url"),
            })
        return references

    def extract_faq_urls(self, top_results: List[Dict]) -> List[str]:
        """상위 참고자료에서 중복 없는 URL 목록 추출"""
        urls = []
        seen = set()
        for result in top_results:
            url = result.get("url")
            if url and url not in seen:
                urls.append(url)
                seen.add(url)
        return urls
    
    def log_failed_query(self, query: str, ts: str):
        """실패한 질문 로깅"""
        try:
            fail_path = self.data_dir / "qa_failed.jsonl"
            fail_path.parent.mkdir(parents=True, exist_ok=True)
            with fail_path.open("a", encoding="utf-8") as fout:
                rec = {"q": query, "ts": ts}
                json.dump(rec, fout, ensure_ascii=False)
                fout.write("\n")
        except Exception as e:
            print(f"[ResponseHandler] 실패 로깅 오류: {e}")
    
    def log_success(
        self,
        original_query: str,
        rewritten_query: Optional[str],
        top_results: List[Dict],
        scores: List[float],
        answer: str,
        ts: str
    ):
        """성공한 질문 로깅"""
        try:
            succ_path = self.data_dir / "qa_successed.jsonl"
            succ_path.parent.mkdir(parents=True, exist_ok=True)
            with succ_path.open("a", encoding="utf-8") as fout:
                rec = {
                    "user_q": original_query,
                    "top_matches": [{"q": r["q"], "score": s} for r, s in zip(top_results, scores)],
                    "llm_answer": answer,
                    "ts": ts
                }
                if rewritten_query:
                    rec["rewritten_q"] = rewritten_query
                json.dump(rec, fout, ensure_ascii=False)
                fout.write("\n")
        except Exception as e:
            print(f"[ResponseHandler] 성공 로깅 오류: {e}")
    
    def save_to_mongodb(
        self,
        bot_id: Optional[str],
        user_query: str,
        ai_response: str,
        matched_questions: List[str],
        scores: List[float],
        metadata: Dict[str, Any],
        ts: str
    ) -> bool:
        """MongoDB에 대화 저장
        
        Returns:
            저장 성공 여부
        """
        if not self.chat_manager:
            print(f"[ResponseHandler] {ts} ⏭️  MongoDB 비활성화 → 저장 스킵")
            return False
        
        try:
            print(f"[ResponseHandler] {ts} ✍️  MongoDB 저장 중... (bot_id={bot_id})")
            result_id = self.chat_manager.save_conversation(
                bot_id=bot_id,
                user_query=user_query,
                ai_response=ai_response,
                matched_questions=matched_questions,
                scores=scores,
                metadata=metadata
            )
            print(f"[ResponseHandler] {ts} ✅ MongoDB 저장 성공! doc_id={result_id}")
            return True
        except Exception as e:
            import traceback
            print(f"[ResponseHandler] {ts} ❌ MongoDB 저장 실패!")
            print(f"[ResponseHandler] {ts} 오류: {e}")
            traceback.print_exc()
            return False
    
    def print_console_log(
        self,
        original_query: str,
        rewritten_query: Optional[str],
        top_results: List[Dict],
        scores: List[float],
        answer: str,
        ts: str
    ):
        """콘솔 로그 출력"""
        try:
            preview = answer.replace("\n", " ")[:120]
            top_match = top_results[0]["q"] if top_results else "N/A"
            top_score = scores[0] if scores else None
            
            query_log = f"'{original_query}'"
            if rewritten_query:
                query_log = f"'{original_query}' → '{rewritten_query}'"
            
            print(f"[ResponseHandler] {ts} query={query_log}")
            print(f"[ResponseHandler] {ts} top_match='{top_match}' score={top_score}")
            print(f"[ResponseHandler] {ts} answer_preview='{preview}'")
        except Exception as e:
            print(f"[ResponseHandler] 콘솔 로그 오류: {e}")
    
    def build_response(
        self,
        answer: str,
        matched_questions: List[str],
        scores: List[float],
        bot_id: Optional[str],
        rerank_info: Dict[str, Any],
        references: List[Dict[str, Any]] = None,
        faq_urls: List[str] = None,
        faiss_scores: List[float] = None
    ) -> Dict[str, Any]:
        """최종 응답 생성"""
        if faiss_scores is None:
            faiss_scores = []
        if references is None:
            references = []
        if faq_urls is None:
            faq_urls = []
        
        return {
            "answer": answer,
            "matched_questions": matched_questions,
            "scores": scores,
            "num_references": len(matched_questions),
            "botId": bot_id,
            "references": references,
            "faq_urls": faq_urls,
            "rerank_info": rerank_info
        }
    
    def build_no_match_response(
        self,
        answer: str,
        bot_id: Optional[str],
        top_k: int
    ) -> Dict[str, Any]:
        """매칭 실패 응답 생성"""
        return {
            "answer": answer,
            "matched_questions": [],
            "scores": [],
            "num_references": 0,
            "botId": bot_id,
            "references": [],
            "faq_urls": [],
            "rerank_info": {
                "used": False,
                "faiss_top_k": top_k,
                "rerank_top_n": 0,
                "faiss_scores": [],
                "rerank_scores": [],
                "detail": "No matching documents found (threshold not met)"
            }
        }