""" conversation_memory.py의 generate_insight() 기능 통합 테스트 러너. 실제 대화 데이터를 로드하고 정리 기능을 실행하여 결과물을 확인한다. """ import asyncio import json import sys from collections import deque from datetime import datetime from pathlib import Path # --------------------------------------------------------------------------- # 1. sys.path 설정 # --------------------------------------------------------------------------- MULTIMODEL_BOT_DIR = "/home/jay/workspace/services/multimodel-bot" if MULTIMODEL_BOT_DIR not in sys.path: sys.path.insert(0, MULTIMODEL_BOT_DIR) # --------------------------------------------------------------------------- # 상수 # --------------------------------------------------------------------------- CHAT_ID = 99999 JSONL_PATH = Path("/home/jay/workspace/memory/groupchat/test-import-2026-03-15.jsonl") STORAGE_BASE = "/home/jay/workspace/memory/groupchat" INSIGHT_RESULT_PATH = Path("/home/jay/workspace/memory/groupchat/insights/test-insight-result.md") SUMMARIES_DIR = Path("/home/jay/workspace/memory/groupchat/summaries") # --------------------------------------------------------------------------- # 단계별 출력 헬퍼 # --------------------------------------------------------------------------- def step(label: str, ok: bool, detail: str = "") -> None: status = "OK" if ok else "FAIL" msg = f"[{status}] {label}" if detail: msg += f" — {detail}" print(msg) # --------------------------------------------------------------------------- # 2. JSONL 읽기 # --------------------------------------------------------------------------- def load_jsonl(path: Path) -> list[dict]: """JSONL 파일의 각 줄을 파싱하여 dict 목록으로 반환한다.""" records: list[dict] = [] if not path.exists(): step("JSONL 파일 존재 확인", False, f"파일 없음: {path}") return records with open(path, "r", encoding="utf-8") as fh: for lineno, raw in enumerate(fh, start=1): raw = raw.strip() if not raw: continue try: records.append(json.loads(raw)) except json.JSONDecodeError as exc: print(f" [WARN] 줄 {lineno} JSON 파싱 실패, 건너뜀: {exc}") step( "JSONL 파일 로드", bool(records), f"{path.name} — {len(records)}개 레코드", ) return records # --------------------------------------------------------------------------- # 3. ConversationMemory 및 ChatMessage import # --------------------------------------------------------------------------- def import_memory_module(): """conversation_memory 모듈을 import하고 (ConversationMemory, ChatMessage) 반환.""" try: from conversation_memory import ChatMessage, ConversationMemory # type: ignore[import] step("conversation_memory import", True) return ConversationMemory, ChatMessage except ImportError as exc: step("conversation_memory import", False, str(exc)) return None, None # --------------------------------------------------------------------------- # 4. deque에 직접 메시지 로드 (add_message() 우회) # --------------------------------------------------------------------------- def populate_deque(memory, ChatMessage, records: list[dict], chat_id: int) -> int: """JSONL 레코드를 ChatMessage로 변환하여 _messages[chat_id] deque에 직접 삽입. add_message()를 사용하면 JSONL에 다시 append되므로 deque에 직접 넣는다. """ if chat_id not in memory._messages: memory._messages[chat_id] = deque(maxlen=memory._max_messages) loaded = 0 for rec in records: try: ts_raw = rec.get("timestamp", "") try: ts = datetime.fromisoformat(ts_raw) except (ValueError, TypeError): ts = datetime.now() msg = ChatMessage( sender=rec.get("sender", "unknown"), text=rec.get("text", ""), timestamp=ts, is_bot=bool(rec.get("is_bot", False)), ) memory._messages[chat_id].append(msg) loaded += 1 except Exception as exc: print(f" [WARN] 레코드 변환 실패, 건너뜀: {exc} | rec={rec!r}") # _message_count도 동기화 (요약 트리거 계산용) memory._message_count[chat_id] = loaded # lazy load 추적에 등록 (add_message의 중복 load_today 방지) memory._loaded_chats.add(chat_id) step( "deque 직접 로드", loaded > 0, f"chat_id={chat_id} — {loaded}개 메시지 삽입 (maxlen={memory._max_messages})", ) return loaded # --------------------------------------------------------------------------- # 5. generate_insight 호출 # --------------------------------------------------------------------------- async def run_generate_insight(memory, chat_id: int) -> str: """generate_insight(chat_id)를 호출하고 결과 문자열을 반환한다.""" print(f"\n[STEP] generate_insight(chat_id={chat_id}) 호출 중...") try: result = await memory.generate_insight(chat_id=chat_id) ok = bool(result) and "실패" not in result[:30] step("generate_insight 호출", ok, f"{len(result)}자 반환") return result except Exception as exc: msg = f"generate_insight 예외 발생: {exc}" step("generate_insight 호출", False, msg) return msg # --------------------------------------------------------------------------- # 6. insight 결과를 지정 경로에 저장 # --------------------------------------------------------------------------- def save_insight_result(insight_text: str, dest: Path) -> None: """insight 결과를 dest 경로에 저장한다.""" try: dest.parent.mkdir(parents=True, exist_ok=True) dest.write_text(insight_text, encoding="utf-8") step("insight 결과 저장", True, str(dest)) except Exception as exc: step("insight 결과 저장", False, str(exc)) # --------------------------------------------------------------------------- # 7. _do_generate_summary 호출 # --------------------------------------------------------------------------- async def run_do_generate_summary(memory, chat_id: int) -> None: """_do_generate_summary(chat_id)를 직접 호출하여 요약 생성을 테스트한다.""" print(f"\n[STEP] _do_generate_summary(chat_id={chat_id}) 호출 중...") # 요약 중복 실행 방지 플래그 초기화 (generate_insight가 이미 설정했을 수 있음) memory._summary_lock[chat_id] = False # 일일 LLM 예산이 소진되지 않았는지 확인 today = datetime.now().strftime("%Y-%m-%d") budget_remaining = memory._daily_llm_budget - memory._llm_call_count.get(today, 0) print(f" LLM 예산 잔여: {budget_remaining}/{memory._daily_llm_budget}") try: await memory._do_generate_summary(chat_id=chat_id) step("_do_generate_summary 호출", True) except Exception as exc: step("_do_generate_summary 호출", False, str(exc)) # --------------------------------------------------------------------------- # 8. summaries/ 디렉토리 결과 확인 # --------------------------------------------------------------------------- def check_summaries_dir(summaries_dir: Path) -> None: """summaries/ 디렉토리에 오늘 날짜로 생성된 JSON 파일을 확인한다.""" print(f"\n[STEP] summaries/ 디렉토리 확인: {summaries_dir}") if not summaries_dir.exists(): step("summaries/ 디렉토리 존재", False, "디렉토리가 없습니다") return today = datetime.now().strftime("%Y-%m-%d") today_files = sorted(summaries_dir.glob(f"{today}_*.json")) step( "summaries/ 오늘 파일 확인", bool(today_files), f"{len(today_files)}개 파일 발견", ) for fp in today_files: try: data = json.loads(fp.read_text(encoding="utf-8")) summary_preview = data.get("summary", "")[:80].replace("\n", " ") print(f" - {fp.name}") print(f" topic_tag : {data.get('topic_tag', '?')}") print(f" key_topics : {data.get('key_topics', [])}") print(f" consensus : {data.get('consensus_level', '?')}") print(f" summary (앞80): {summary_preview}") except Exception as exc: print(f" - {fp.name} — 읽기 실패: {exc}") # --------------------------------------------------------------------------- # 메인 실행 흐름 # --------------------------------------------------------------------------- async def main() -> None: print("=" * 60) print(" generate_insight() 통합 테스트 러너") print(f" 실행 시각: {datetime.now().isoformat()}") print("=" * 60) # --- Step 1: call_claude import 가능 여부 확인 (비필수, 경고만) --- print("\n[STEP] call_claude import 확인...") try: from engine_v2.bot_api import call_claude # type: ignore[import] # noqa: F401 step("call_claude import", True, "engine_v2.bot_api.call_claude 사용 가능") except ImportError as exc: step("call_claude import", False, f"{exc} — LLM 호출 시 fallback 발동") # --- Step 2: JSONL 로드 --- print(f"\n[STEP] JSONL 읽기: {JSONL_PATH}") records = load_jsonl(JSONL_PATH) if not records: print("[WARN] 로드된 레코드가 없습니다. 빈 대화로 계속 진행합니다.") # --- Step 3: ConversationMemory 인스턴스 생성 --- print("\n[STEP] ConversationMemory 인스턴스 생성...") ConversationMemory, ChatMessage = import_memory_module() if ConversationMemory is None: print("[ERROR] conversation_memory import 실패 — 테스트를 중단합니다.") return memory = ConversationMemory(storage_base=STORAGE_BASE) step( "ConversationMemory 생성", True, f"storage_base={STORAGE_BASE}, max_messages={memory._max_messages}", ) # --- Step 4: deque에 직접 메시지 로드 --- print(f"\n[STEP] 메시지를 deque에 직접 로드 (chat_id={CHAT_ID})...") loaded_count = populate_deque(memory, ChatMessage, records, chat_id=CHAT_ID) actual_in_deque = len(memory._messages.get(CHAT_ID, [])) print(f" deque 실제 크기: {actual_in_deque}개 " f"(maxlen={memory._max_messages}, 로드 시도={loaded_count}개)") # --- Step 5: generate_insight 호출 --- insight_text = await run_generate_insight(memory, chat_id=CHAT_ID) # --- Step 6: insight 결과 지정 경로에 저장 --- print(f"\n[STEP] insight 결과를 {INSIGHT_RESULT_PATH} 에 저장...") save_insight_result(insight_text, INSIGHT_RESULT_PATH) # insight 내용 미리보기 출력 print("\n--- insight 내용 미리보기 (앞 500자) ---") print(insight_text[:500]) if len(insight_text) > 500: print(f" ... (총 {len(insight_text)}자)") print("---") # --- Step 7: _do_generate_summary 직접 호출 --- await run_do_generate_summary(memory, chat_id=CHAT_ID) # --- Step 8: summaries/ 디렉토리 결과 확인 --- check_summaries_dir(SUMMARIES_DIR) # --- 최종 요약 --- print("\n" + "=" * 60) print(" 테스트 완료") print(f" insight 파일: {INSIGHT_RESULT_PATH}") print(f" summaries 디렉토리: {SUMMARIES_DIR}") print("=" * 60) if __name__ == "__main__": asyncio.run(main())