# portfolio/npc_social_network/manager/conversation_manager.py # 대화의 시작, 진행, 종료를 총괄하는 역할 from ..models.llm_helper import (summarize_text, query_llm_for_emotion , evaluate_goal_achievement) from ..npc.npc_manager import get_korean_postposition from ..npc.emotion_config import EMOTION_RELATION_IMPACT, EMOTION_LIST import threading from typing import List, Optional, TYPE_CHECKING if TYPE_CHECKING: from ..npc.npc_base import NPC class Conversation: """단일 대화의 상태를 저장하는 데이터 클래스""" def __init__(self, initiator: "NPC", target: "NPC", topic: str): self.participants: List["NPC"] = [initiator, target] self.conversation_history: List[str] = [] self.topic: str = topic self.turn_index: int = 0 # 0: initiator, 1: target self.is_ending: bool = False self.waiting_for_player: bool = False # 플레이어의 입력 대기 def get_current_speaker(self) -> "NPC": """이번에 말할 차례인 NPC를 반환""" return self.participants[self.turn_index] def switch_turn(self): """대화 턴을 상대방에게 넘김""" self.turn_index = 1 - self.turn_index class ConversationManager: """활성화된 대화를 관리하는 중앙 관리자""" def __init__(self): self.active_conversation: Optional[Conversation] = None def is_conversation_active(self) -> bool: """현재 진행 중인 대화가 있는지 확인""" return self.active_conversation is not None def _add_and_log_utterance(self,speaker: "NPC", utterance: str): """대사를 대화 기록과 UI 로그에 모두 추가""" from .. import simulation_core if not self.is_conversation_active(): return log_message = f"[{speaker.korean_name}]: {utterance}" self.active_conversation.conversation_history.append(log_message) simulation_core.add_log(log_message) def _summarize_and_remember_in_background(self, conv: "Conversation"): """백그라운드에서 대화를 요약하고 기억하며, 관계와 성격에 반영, 목표 달성 평가를 수행하는 함수""" from .. import simulation_core try: if not conv.conversation_history: print("기억할 대화 내용이 없어 종료.") return initiator, target = conv.participants[0], conv.participants[1] full_conversation = "\n".join(conv.conversation_history) # 1. 대화 내용 요약 summary = summarize_text(full_conversation) # 2. 대화의 전반적인 감정 톤 분석 overall_emotion = query_llm_for_emotion(summary) # 3. 목표 달성 여부 평가 initiator = conv.participants[0] evaluation = evaluate_goal_achievement( initiator_name=initiator.korean_name, goal=conv.topic, conversation_transcript=full_conversation ) with simulation_core.simulation_lock: # 4. 평가 결과 로그 추가 simulation_core.add_log(f"[{initiator.korean_name}의 목표 달성 평가]\n" f"'{conv.topic}' -> 달성: {evaluation.get('goal_achieved')}.\n" f"이유: {evaluation.get('reason')}") target_postposition = get_korean_postposition(target.korean_name, "과", "와") # 5. 평가 결과에 따른 감정 업데이트 initiator_emotion = evaluation.get('initiator_emotion') if initiator_emotion and initiator_emotion in EMOTION_LIST: initiator.update_emotion(initiator_emotion, strength=5.0) # 목표 달성 여부는 강한 감정 유발 # 6. 요약된 내용을 바탕으로 기억 생성 memory_content = f"'{conv.topic}'에 대해 대화하며 '{summary}'라는 결론을 내렸다." initiator.remember(content=f"{target.korean_name}{target_postposition} {memory_content}", importance=7, memory_type="Conversation") target.remember(content=f"{initiator.korean_name}{target_postposition} {memory_content}", importance=7, memory_type="Conversation") # 7. 대화의 감정을 바탕으로 관계 상태 업데이트 if overall_emotion and overall_emotion in EMOTION_RELATION_IMPACT: initiator.relationships.update_relationship(target.name, overall_emotion, strength=3.0) target.relationships.update_relationship(initiator.name, overall_emotion, strength=3.0) # 8. 관계 변화를 바탕으로 양쪽 모두의 성격 업데이트 initiator.update_personality() target.update_personality() simulation_core.add_log(f"[{initiator.korean_name}, {target.korean_name}] 대화 경험이 기억과 관계, 성격에 반영되었습니다.") # 9. 모든 백그라운드 작업 종료 후, 현재까지의 시뮬레이션 전체 상태를 즉시 저장. simulation_core.save_simulation(simulation_core.npc_manager) except Exception as e: print(f"[에러] 백그라운드 기억 저장 중 문제 발생: {e}") import traceback traceback.print_exc() def start_conversation(self, initiator: "NPC", target: "NPC", topic: Optional[str] = None): """새로운 대화를 시작, 바로 다음 턴을 호출""" from .. import simulation_core if self.is_conversation_active(): return # 이미 다른 대화가 진행중이면 시작 x conv_topic = topic or "가벼운 인사" self.active_conversation = Conversation(initiator, target, conv_topic) simulation_core.add_log(f"--[대화 시작]--") simulation_core.add_log(f"주제: {conv_topic} / 참여자: {initiator.korean_name}, {target.korean_name}") # 1. 첫 마디 생성 self.next_turn() def end_conversation(self, reason: str = "자연스럽게"): """현재 대화를 종료하고, 대화 내용을 요약하여 기억으로 저장""" from .. import simulation_core if not self.is_conversation_active(): return # 대화가 끝나면, 전체 대화 내용을 바탕으로 기억을 생성하고 관계를 업데이트 conv_to_process = self.active_conversation # 1. 대화 상태를 즉시 '종료'로 변경하여 메인 루프를 해방 self.active_conversation = None simulation_core.add_log(f"---[대화 종료: {reason}]---\n") # 2. 시간이 걸리는 '기억 저장' 작업은 별도의 스레드를 생성하여 백그라운드에서 처리. background_thread = threading.Thread( target=self._summarize_and_remember_in_background, args=(conv_to_process,) ) background_thread.start() def next_turn(self): """대화의 다음 턴을 진행, 생성된 대사 처리""" if not self.is_conversation_active(): return conv = self.active_conversation speaker = conv.get_current_speaker() # 플레이어 턴 처리 로직 if speaker.name == "player": conv.waiting_for_player = True return # 플레이어의 응답이 올 때까지 대화 턴 진행을 멈춤 # 상태 1: 마무리 단계 (작별 인사) if conv.is_ending: utterance, _ = speaker.generate_dialogue_turn(conv, is_final_turn=True) self._add_and_log_utterance(speaker, utterance) self.end_conversation("작별 인사 완료") return # 상태 2: 진행 단계 (일반 대화) utterance, action = speaker.generate_dialogue_turn(conv) self._add_and_log_utterance(speaker, utterance) # 상태 전환 결정 if action != "CONTINUE" or len(conv.conversation_history) >= 10: conv.is_ending = True conv.switch_turn() else: conv.switch_turn()