멀티턴 대화 시스템 구현 - 진행중

npc_social_network/manager/conversation_manager.py

@@ -0,0 +1,97 @@
+# portfolio/npc_social_network/manager/conversation_manager.py
+# 대화의 시작, 진행, 종료를 총괄하는 역할
+from .. import simulation_core
+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
+
+    def add_utterance(self, speaker: "NPC", utterance: str):
+        """대화 내역을 추가"""
+        self.conversation_history.append(f"[{speaker.korean_name}]: {utterance}")
+
+    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 start_conversation(self, initiator: "NPC", target: "NPC",
+                           topic: Optional[str] = None):
+        """새로운 대화를 시작"""
+        if self.is_conversation_active():
+            return # 이미 다른 대화가 진행중이면 시작 x
+        
+        conv_topic = topic or "가벼운 인사"
+        self.active_conversation = Conversation(initiator, target, conv_topic)
+        print(f"--[대화 시작]--")
+        print(f"주제: {conv_topic} / 참여자: {initiator.korean_name}, {target.korean_name}")
+        
+        # 1. 첫 마디 생성
+        first_utterance = initiator.generate_first_utterance(conv_topic, target)
+        if "..." in first_utterance:
+            self.end_conversation("첫 마디 생성 실패")
+            return
+        
+        # 2. 첫 마디를 기록에 추가
+        self.active_conversation.add_utterance(initiator, first_utterance)
+
+        # 3. 턴을 상대방에게 넘김 (이제 상대방이 대답할 차례)
+        self.active_conversation.switch_turn()
+
+    def end_conversation(self, reason: str = "자연스럽게"):
+        """현재 대화를 종료"""
+        if self.is_conversation_active():
+            # 대화가 끝나면, 전체 대화 내용을 바탕으로 기억을 생성하고 관계를 업데이트
+            conv = self.active_conversation
+            initiator = conv.participants[0]
+            target = conv.participants[1]
+            full_conversation = "\n".join(conv.conversation_history)
+
+            with simulation_core.simulation_lock:
+                # 양쪽 모두에게 대화 기억을 저장
+                initiator.remember(content=f"{target.korean_name}와 '{conv.topic}'에 대해 대화했다:\n{full_conversation}", importance=7, memory_type="Conversation")
+                target.remember(content=f"{initiator.korean_name}와 '{conv.topic}'에 대해 대화했다:\n{full_conversation}", importance=7, memory_type="Conversation")
+
+            print(f"---[대화 종료: {reason}]---\n")
+            self.active_conversation = None
+
+    def next_turn(self):
+        """대화의 다음 턴을 진행"""
+        if not self.is_conversation_active():
+            return
+        
+        conv = self.active_conversation
+        speaker = conv.get_current_speaker()
+
+        # 대답 생성 로직 (다음 단계에서 수정)
+        utterance, action = speaker.generate_dialogue_turn(conv)
+
+        if "[LLM Error]" in utterance:
+            self.end_conversation("오류 발생")
+            return
+        
+        conv.add_utterance(speaker, utterance)
+
+        if action == "END" or len(conv.conversation_history) >= 10: #대화가 너무 길어지면 강제 종료
+             self.end_conversation()
+        else:
+            conv.switch_turn()

npc_social_network/models/llm_helper.py

@@ -114,7 +114,7 @@ Action Types can be "TALK" (for conversing with others) or "SOLO_ACTION" (for ac
         json_match = re.search(r'\{.*\}', response_text, re.DOTALL)
 
         if json_match:
-            return json.load(json_match.group())
+            return json.loads(json_match.group())
         
         # 분석 실패 시 혼자 하는 행동으로 간주
         return {"action_type": "SOLO_ACTION", "target": None, "topic": plan_step}
@@ -173,4 +173,36 @@ Your response MUST be a valid JSON array of strings, where each string is a step
     result = _query_llm_for_json_robustly(prompt)
     if isinstance(result, list) and all(isinstance(s, str) for s in result):
         return result
-    return ["[계획 생성에 실패했습니다]"] # 최종 실패 시 안전한 값 반환
+    return ["[계획 생성에 실패했습니다]"] # 최종 실패 시 안전한 값 반환
+
+def generate_dialogue_action(speaker_name: str, conversation_history: List[str], topic: str) -> dict:
+    """대화 기록을 바탕으로 다음 대사와 행동(계속/종료)을 생성"""
+    history_str = "\n".join(conversation_history)
+
+    prompt = f"""
+# Persona
+당신은 "{speaker_name}"입니다.
+
+# Context
+- 현재 대화 주제: "{topic}"
+- 지금까지의 대화 내용:
+{history_str}
+
+# Instruction
+- 위 대화의 흐름과 주제에 맞춰, 당신이 이번 턴에 할 자연스러운 대사를 생성하셍.
+- 그리고, 이 대화를 계속 이어갈지("CONTINUE"), 아니면 이 대사를 끝으로 대화를 마무리할지("END") 결정하세요.
+- 당신의 응답은 반드시 아래 JSON 형식이어야 합니다.
+- 대사에는 이름, 행동 묘사, 따옴표를 절대 포함하지 마세요.
+
+# Response (JSON format only)
+{{
+    "dialogue": "실제 말할 대사 내용",
+    "action": "CONTINUE or END"
+}}
+"""
+    result = _query_llm_for_json_robustly(prompt)   # 3중 방어 시스템 재활용
+
+    # 결과 포맷 검증 및 기본값 처리
+    if isinstance(result, dict) and "dialogue" in result and "action" in result:
+        return result
+    return {"dialogue": "...", "action": "END"} # 최종 실패 시 안전하게 대화 종료

npc_social_network/npc/npc_base.py

@@ -11,7 +11,8 @@ from .emotion_config import EMOTION_LIST, EMOTION_CATEGORY_MAP, EMOTION_DECAY_RA
 from .emotion_config import POSITIVE_RELATION_EMOTIONS, NEGATIVE_RELATION_EMOTIONS, COGNITIVE_RELATION_EMOTIONS
 from .personality_config import AGE_PROFILE, PERSONALITY_PROFILE
 from .npc_relationship import RelationshipManager
-from ..models.llm_helper import query_llm_with_prompt, query_llm_for_emotion, summarize_memories, analyze_gossip, classify_and_extract_plan_details
+from ..models.llm_helper import query_llm_with_prompt, query_llm_for_emotion, summarize_memories
+from ..models.llm_helper import analyze_gossip, classify_and_extract_plan_details, generate_dialogue_action
 from ..models.llm_prompt_builder import build_npc_prompt
 from .npc_memory_embedder import search_similar_memories, add_memory_to_index
 from .npc_planner import PlannerManager
@@ -141,91 +142,133 @@ class NPC:
         self.personality_stage = None
         self.update_personality_stage()
 
-    def generate_dialogue(self, user_input: str, time_context: str
-                          , target_npc: Optional["NPC"]=None, use_llm: bool=True
-                          , create_memory: bool = True) -> str:
-        """
-        플레이어 또는 NPC와 상호작용 시 사용되는 통합 대사 생성 함수
-        - user_input: 입력 문장 (플레이어나 NPC로부터)
-        - target_npc: NPC 간 대화일 경우 상대 NPC
-        """
-        if use_llm and user_input:
-            # 1. 생각하기
-            # 1-1. 유사 기억 검색
-            _, _, matched_memories = search_similar_memories(self, user_input)
-
-            # 1-2. LLM 프롬프트 생성 및 호출
-            prompt = build_npc_prompt(self, user_input, matched_memories, time_context, target_npc)
-            npc_reply = query_llm_with_prompt(prompt)
-
-            if not npc_reply or "[LLM Error]" in npc_reply:
-                return "[LLM Error]..." # 에러 시 기본 응답
-
-            # 1-3. 자신의 응답에서 감정 추론 및 상태 업데이트
-            emotion_from_reply  = query_llm_for_emotion(npc_reply)
-            # 1-4. 상대방의 말에서 감정 추론 (NPC간 대화일 경우)
-            emotion_from_input = query_llm_for_emotion(user_input) if target_npc else None
-
-            # 2. 행동하기
-            if create_memory:
-                with simulation_core.simulation_lock:
-                    # 2-1. 기억 요약 및 장기 기억화
-                    if len(matched_memories) >= 3:
-                        self.summarize_and_store_memories(matched_memories)
-
-                    # 2-2. 추론한 감정들을 상태에 업데이트
-                    if emotion_from_reply  and emotion_from_reply  in EMOTION_LIST:
-                        self.update_emotion(emotion_from_reply , strength=2.0)
-
-                    # 2-3. 상호작용 기억 저장
-                    interlocutor_name = target_npc.name if target_npc else "player"
-                    interlocutor_korean_name = target_npc.korean_name if target_npc else "플레이어"
-                    memory_content = f"[{interlocutor_korean_name}의 말] '{user_input}' → [나의 응답] '{npc_reply}'"
-                    memory = self.remember(content=memory_content, importance=7, emotion=emotion_from_reply , context_tags=[time_context])
-
-                    # 2-4. 관계 상태 업데이트
-                    if emotion_from_reply  and emotion_from_reply  in EMOTION_RELATION_IMPACT:
-                        self.relationships.update_relationship(interlocutor_name, emotion_from_reply , strength=2.0, memory=memory)
-                    if target_npc and emotion_from_input and emotion_from_input in EMOTION_RELATION_IMPACT:
-                        target_npc.relationships.update_relationship(self.name, emotion_from_input, strength=2.0, memory=memory)
-
-                    # 2-5. 성격 변화 반영
-                    self.update_personality()
-
-            return npc_reply
+    def generate_first_utterance(self, topic: str, target_npc: "NPC") -> str:
+        """대화의 첫 마디를 생성. (대화 기록이 필요 없음)"""
+        prompt = f"""
+# Persona
+당신은 "{self.korean_name}"입니다.
 
-        else:
-            # LLM 미사용 시: 감정 기반 행동 생성
-            behavior_output, behavior_trace = self.behavior.perform_sequence(
-                self.name, self.job, emotion_buffer=self._emotion_buffer, return_trace=True
-            )
+# Context
+- 당신은 지금 "{target_npc.korean_name}"에게 말을 걸려고 합니다.
+- 당신과 상대의 관계: {self.relationships.get_relationship_summary(target_npc.name)}
+- 대화 주제: "{topic}"
 
-            # 행동 기반 감정 추론 및 반영
-            for action, score in behavior_trace:
-                emotion_name = self.behavior.action_to_emotion.get(action)
-                if emotion_name:
-                    self.update_emotion(emotion_name, strength=score)
-
-            # dominant_score 기반 중요도 추정
-            dominant_emotions = self.behavior.get_layer_dominant_emotions(self._emotion_buffer)
-            dominant_sequence = self.behavior.decide_layered_sequence(dominant_emotions)
-            dominant_score = max([score for _, score in dominant_sequence]) if dominant_sequence else 5
-            importance = min(int(dominant_score), 10)
-
-            # Memory 저장
-            memory_entry = Memory(
-                content=f"행동 수행: {behavior_trace}",
-                importance=importance,
-                emotion=self.emotion.get_dominant_emotion(),
-                behavior_trace=behavior_trace,
-                context_tags=[time_context]
-            )
-            self.memory_store.add_memory(memory_entry)
+# Instruction
+- 주어진 상황과 대화 주제에 맞춰, 상대방에게 건넬 자연스러운 첫 대사 **한 문장만** 생성하세요.
+- **절대로** 당신의 이름, 행동 묘사, 주석, 따옴표를 포함하지 마세요.
 
-            # 성격 업데이트
-            self.update_personality()
+{self.korean_name}:
+"""
+        utterance = query_llm_with_prompt(prompt).strip().strip('"')
+        return utterance if utterance and "[LLM Error]" not in utterance else "..."
 
-            return str(behavior_output)
+    def generate_dialogue_turn(self, conversation: "Conversation") -> (str, str):
+        """대화의 현재 턴에 대한 응답과 행동을 생성"""
+        
+        # 생각하기
+        result = generate_dialogue_action(
+            self.korean_name,
+            conversation.conversation_history,
+            conversation.topic
+        )
+        dialogue = result.get("dialogue", "...")
+        action = result.get("action", "END")
+
+        # 행동하기 (잠금과 함께)
+        # 이 부분은 대화가 끝난 후, 전체 대화 내용을 바탕으로 기억/관계를 한 번에 업데이트 하도록 향후 수정.
+        # 지금은 매 턴 감정만 업데이트
+        with simulation_core.simulation_lock:
+            emotion = query_llm_for_emotion(dialogue)
+            if emotion and emotion in EMOTION_LIST:
+                self.update_emotion(emotion, 1.0)
+
+        return dialogue, action
+
+    # def generate_dialogue(self, user_input: str, time_context: str
+    #                       , target_npc: Optional["NPC"]=None, use_llm: bool=True
+    #                       , create_memory: bool = True) -> str:
+    #     """
+    #     플레이어 또는 NPC와 상호작용 시 사용되는 통합 대사 생성 함수
+    #     - user_input: 입력 문장 (플레이어나 NPC로부터)
+    #     - target_npc: NPC 간 대화일 경우 상대 NPC
+    #     """
+    #     if use_llm and user_input:
+    #         # 1. 생각하기
+    #         # 1-1. 유사 기억 검색
+    #         _, _, matched_memories = search_similar_memories(self, user_input)
+
+    #         # 1-2. LLM 프롬프트 생성 및 호출
+    #         prompt = build_npc_prompt(self, user_input, matched_memories, time_context, target_npc)
+    #         npc_reply = query_llm_with_prompt(prompt)
+
+    #         if not npc_reply or "[LLM Error]" in npc_reply:
+    #             return "[LLM Error]..." # 에러 시 기본 응답
+
+    #         # 1-3. 자신의 응답에서 감정 추론 및 상태 업데이트
+    #         emotion_from_reply  = query_llm_for_emotion(npc_reply)
+    #         # 1-4. 상대방의 말에서 감정 추론 (NPC간 대화일 경우)
+    #         emotion_from_input = query_llm_for_emotion(user_input) if target_npc else None
+
+    #         # 2. 행동하기
+    #         if create_memory:
+    #             with simulation_core.simulation_lock:
+    #                 # 2-1. 기억 요약 및 장기 기억화
+    #                 if len(matched_memories) >= 3:
+    #                     self.summarize_and_store_memories(matched_memories)
+
+    #                 # 2-2. 추론한 감정들을 상태에 업데이트
+    #                 if emotion_from_reply  and emotion_from_reply  in EMOTION_LIST:
+    #                     self.update_emotion(emotion_from_reply , strength=2.0)
+
+    #                 # 2-3. 상호작용 기억 저장
+    #                 interlocutor_name = target_npc.name if target_npc else "player"
+    #                 interlocutor_korean_name = target_npc.korean_name if target_npc else "플레이어"
+    #                 memory_content = f"[{interlocutor_korean_name}의 말] '{user_input}' → [나의 응답] '{npc_reply}'"
+    #                 memory = self.remember(content=memory_content, importance=7, emotion=emotion_from_reply , context_tags=[time_context])
+
+    #                 # 2-4. 관계 상태 업데이트
+    #                 if emotion_from_reply  and emotion_from_reply  in EMOTION_RELATION_IMPACT:
+    #                     self.relationships.update_relationship(interlocutor_name, emotion_from_reply , strength=2.0, memory=memory)
+    #                 if target_npc and emotion_from_input and emotion_from_input in EMOTION_RELATION_IMPACT:
+    #                     target_npc.relationships.update_relationship(self.name, emotion_from_input, strength=2.0, memory=memory)
+
+    #                 # 2-5. 성격 변화 반영
+    #                 self.update_personality()
+
+    #         return npc_reply
+
+    #     else:
+    #         # LLM 미사용 시: 감정 기반 행동 생성
+    #         behavior_output, behavior_trace = self.behavior.perform_sequence(
+    #             self.name, self.job, emotion_buffer=self._emotion_buffer, return_trace=True
+    #         )
+
+    #         # 행동 기반 감정 추론 및 반영
+    #         for action, score in behavior_trace:
+    #             emotion_name = self.behavior.action_to_emotion.get(action)
+    #             if emotion_name:
+    #                 self.update_emotion(emotion_name, strength=score)
+
+    #         # dominant_score 기반 중요도 추정
+    #         dominant_emotions = self.behavior.get_layer_dominant_emotions(self._emotion_buffer)
+    #         dominant_sequence = self.behavior.decide_layered_sequence(dominant_emotions)
+    #         dominant_score = max([score for _, score in dominant_sequence]) if dominant_sequence else 5
+    #         importance = min(int(dominant_score), 10)
+
+    #         # Memory 저장
+    #         memory_entry = Memory(
+    #             content=f"행동 수행: {behavior_trace}",
+    #             importance=importance,
+    #             emotion=self.emotion.get_dominant_emotion(),
+    #             behavior_trace=behavior_trace,
+    #             context_tags=[time_context]
+    #         )
+    #         self.memory_store.add_memory(memory_entry)
+
+    #         # 성격 업데이트
+    #         self.update_personality()
+
+    #         return str(behavior_output)
 
     def remember(self, content: str, importance: int = 5, emotion: str = None,
                  strength: float = 1.0, memory_type:str = "Event",
@@ -451,7 +494,7 @@ class NPC:
                     target_npc = self.manager.get_npc_by_korean_name(target_korean_name)
                     if target_npc and target_npc != self:
                         # 주제가 있는 대화 시작
-                        self.manager.initiate_npc_to_npc_interaction(self, target_npc, time_context, topic=topic)
+                        simulation_core.conversation_manager.start_conversation(self, target_npc, topic=topic)
                     else:
                         print(f"[{self.korean_name}] '{target_korean_name}'를 찾을 수 없어 대화 계획을 실행하지 못했습니다.")
                 else:
@@ -520,7 +563,7 @@ class NPC:
                     target_npc = random.choices(potential_targets, weights=weights, k=1)[0]
                      
                     # 주제가 없는 가벼운 대화 시작
-                    self.manager.initiate_npc_to_npc_interaction(self, target_npc, time_context, topic=None)
+                    simulation_core.conversation_manager.start_conversation(self, target_npc)
 
     def create_symbolic_memory(self):
         """

npc_social_network/simulation_core.py

@@ -7,6 +7,9 @@ import random
 from .npc.npc_manager import NPCManager
 from .scenarios.scenario_setup import setup_initial_scenario
 from .manager.simulation_manager import save_simulation, load_simulation
+from .manager.conversation_manager import ConversationManager
+
+from typing import List, Optional
 
 #------------------------------------------
 # 1. 시뮬레이션 상태 (전역 변수)
@@ -14,8 +17,9 @@ from .manager.simulation_manager import save_simulation, load_simulation
 #------------------------------------------
 npc_manager: NPCManager = None
 simulation_paused = True
-event_log = []
+event_log:List[str] = []
 simulation_lock = threading.Lock()      # 데이터 동시 접근을 막는 잠금장치
+conversation_manager: Optional[ConversationManager] = None
 
 #------------------------------------------
 # 2. 시뮬레이션 로직 함수
@@ -45,7 +49,7 @@ def tick_simulation():
         with simulation_lock:
             npc.decay_emotions()
             npc.decay_memories()
-        if random.random() < 0.2:
+        if random.random() < 0.1:
             npc.update_autonomous_behavior("자율 행동 시간")
 
 def simulation_loop():
@@ -57,7 +61,11 @@ def simulation_loop():
     tick_counter = 0
 
     while True:
-        if not simulation_paused:
+        # 대화 중일 때는 대화만 진행
+        if conversation_manager and conversation_manager.is_conversation_active():
+            conversation_manager.next_turn()
+
+        elif not simulation_paused:
             tick_simulation()
 
             # 틱이 발생할 때마다 카운터 증가
@@ -67,12 +75,13 @@ def simulation_loop():
                     save_simulation(npc_manager)
                 tick_counter = 0
                 
-        time.sleep(5)   # 5초에 한 번씩 틱 발생
+        time.sleep(5)   # 5초에 한 번씩 틱 발생 (이것보다 줄이면 Gemini API 사용량 초과 발생 가능)
 
 def initialize_simulation():
     """서버 시작 시 시뮬레이션을 초기화합니다."""
-    global npc_manager
+    global npc_manager, conversation_manager
     npc_manager = load_simulation()
+    conversation_manager = ConversationManager()
     if npc_manager is None:
         npc_manager = setup_initial_scenario()
         save_simulation(npc_manager)