# portfolio/npc_social_network/npc/npc_base.py import random import copy import re from typing import Optional, List from datetime import datetime from .npc_memory import Memory, MemoryStore from .npc_emotion import EmotionManager from .npc_behavior import BehaviorManager from .emotion_config import (EMOTION_LIST, EMOTION_CATEGORY_MAP, EMOTION_DECAY_RATE , PERSONALITY_TEMPLATE, EMOTION_RELATION_IMPACT , 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 , evaluate_social_action_outcome , generate_reflection_topic) from .npc_memory_embedder import search_similar_memories, add_memory_to_index from .npc_manager import get_korean_postposition from .npc_planner import PlannerManager from .. import simulation_core from typing import TYPE_CHECKING, Tuple, List if TYPE_CHECKING: from .npc_manager import NPCManager from ..manager.conversation_manager import Conversation # Personality 데이터를 관리하고 요약 기능을 제공하는 클래스 정의 class PersonalityManager: """ NPC의 성격 데이터를 관리하고 관련 기능을 제공하는 클래스 """ def __init__(self, initial_traits=None): self.traits = copy.deepcopy(initial_traits) if initial_traits else copy.deepcopy(PERSONALITY_TEMPLATE) def get(self, key, default=None): return self.traits.get(key, default) def __getitem__(self, key): return self.traits[key] def __setitem__(self, key, value): self.traits[key] = value def keys(self): return self.traits.keys() def get_narrative_summary(self) -> str: """현재 성격 수치를 바탕으로 서술적인 묘사를 동적으로 생성""" narratives = [] sensitive = self.traits.get("sensitive", 0.5) stoic = self.traits.get("stoic", 0.5) congitive_bias = self.traits.get("cognitive_bias", 0.5) # 각 성격 축에 대한 서술적 묘사 생성 if sensitive > 0.7 and stoic < 0.3: narratives.append("감수성이 매우 풍부하고 감정의 영향을 쉽게 받는 편입니다.") elif stoic > 0.7 and sensitive < 0.3: narratives.append("감정을 거의 드러내지 않는 과묵하고 무던한 성향입니다.") else: narratives.append("상황에 따라 감정을 표현할 줄 아는 균형 잡힌 모습을 보입니다.") if congitive_bias > 0.7: narratives.append("대화를 할 때 논리적이고 사실에 기반하여 판단하려는 경향이 강합니다.") elif congitive_bias < 0.3: narratives.append("직관과 감정에 따라 상황을 판단하는 경향이 있습니다.") if not narratives: return "평범하고 균형 잡힌 성격입니다." return "".join(narratives) # NPC 클래스 정의 class NPC: """ NPC 클래스 NPC에 대한 기본 정보를 객체화 """ def __init__(self, name: str, korean_name: str, job: str, personality: Optional[dict]=None): self.name = name self.korean_name = korean_name self.job = job self.manager: Optional['NPCManager'] = None # npc 기억 self.memory_store = MemoryStore() # 감정 상태 등 감정 관리자 초기화 self.personality = PersonalityManager(personality) self.relationships = RelationshipManager(self) self.reputation = {} # 다른 NPC에 대한 평판 저장 self.emotion = EmotionManager(EMOTION_DECAY_RATE, self.personality) self.knowledge: dict[str, dict] = {} # 다른 NPC에 대한 사실 정보를 저장하는 '지식 베이스' # 행동 관리자 self.behavior = BehaviorManager() self.planner = PlannerManager(self) # 내부: float 감정 수치 관리용 버퍼 self._emotion_buffer = self.emotion.get_buffer() # baseline 감정 값 부여 baseline_buffer = {} for emo in self._emotion_buffer: base_min = 0.05 base_max = 0.3 category = EMOTION_CATEGORY_MAP.get(emo, None) bias_map = { "core": "affect_bias", "social": "social_bias", "cognitive": "cognitive_bias", "complex": "complex_bias" } bias = self.personality.get(bias_map.get(category, "sensitive"), 1.0) baseline_value = base_min + (base_max - base_min) * bias baseline_value *= self.personality.get("sensitive", 1.0) # 전체 민감도 적용 baseline_buffer[emo] = round(baseline_value, 3) # baseline buffer를 EmotionManager에 등록 self.emotion.set_baseline(baseline_buffer) # emotion_buffer 초기화 self._emotion_buffer = self.emotion.get_buffer() # 프로필 지정 → 기본은 "stable", 나중에 NPC마다 다르게 부여 profile_type = "stable" profile = PERSONALITY_PROFILE[profile_type] # Personality 변화 속도 개별 설정 self.personality_change_rate = { "sensitive": random.uniform(*profile["sensitive"]), "stoic": random.uniform(*profile["stoic"]), "cognitive_bias": random.uniform(*profile["cognitive_bias"]), } # Personality baseline 저장 (초기값 복사) self.personality_baseline = copy.deepcopy(self.personality.traits) # 나이 추가 (18 ~ 50세 기본 랜덤 초기화 → 나중에 생성 시 age 지정 가능) self.age = random.randint(18, 50) # 초기 personality_stage는 update_personality_stage()에서 자동 설정 self.personality_stage = None self.update_personality_stage() def generate_dialogue_turn(self, conversation: "Conversation", is_final_turn: bool=False) -> Tuple[str, str]: """대화의 현재 턴에 대한 응답과 행동을 생성 (기억과 관계를 총 동원)""" from ..models.llm_helper import generate_dialogue_action # 1. 생각하기: '첫 마디'와 '대답'에 따라 다른 정보로 기억을 탐색. if not conversation.conversation_history: # 대화 기록이 없으면 '첫 마디' 차례 query_text = conversation.topic else: # 대화 기록이 있으면 '대화' 차례 query_text = "\n".join(conversation.conversation_history[-2:]) # 최근 2턴의 대화로 기억 탐색 _, _, relevant_memories = search_similar_memories(self, query_text, top_k=3) # 2. 대사 생성: 모든 정보를 종합하여 llm_helper에 요청 result = generate_dialogue_action( speaker = self, target = conversation.participants[1-conversation.turn_index], conversation=conversation, memories = relevant_memories, is_final_turn = is_final_turn ) dialogue = result.get("dialogue", "...") action = result.get("action", "END").strip().upper() # 행동하기 (잠금과 함께) 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 remember(self, content: str, importance: int = 5, emotion: str = None, strength: float = 1.0, memory_type:str = "Event", context_tags: Optional[List[str]]=None) -> Memory: """ NPC가 새로운 기억을 저장하고 감정 상태에 반영 """ memory = Memory( content=content, importance = importance, emotion=emotion or "neutral", memory_type=memory_type, context_tags=context_tags ) self.memory_store.add_memory(memory) # 중요도가 3 이상인 기억만 벡터 인덱스에 추가하여 효율성 확보 if memory.importance >= 3: add_memory_to_index(self, memory) if emotion and emotion in EMOTION_LIST: self.update_emotion(emotion, strength) # strength = importance / 5.0으로도 가능 # 만약 기억이 '소문'이라면, 평판을 업데이트 if memory_type == "Gossip": # (간단한 예시: LLM으로 소문의 대상과 긍/부정을 파악해야 더 정확함) # 긍정 소문 예시: "밥이 찰리를 도왔다" # 부정 소문 예시: "엘린이 앨리스와 다퉜다" self.process_gossip_and_update_reputation(memory) return memory # 생성된 기억 객체 반환 def update_emotion(self, emotion:str, strength: float = 1.0): """ 특정 감정을 감정 상태에 반영 """ self.emotion.update_emotion(emotion, strength) self._emotion_buffer = self.emotion.get_buffer() def decay_emotions(self): """ 시간이 지남에 따라 모든 감정이 서서히 감소 """ self.emotion.decay_emotions() self._emotion_buffer = self.emotion.get_buffer() def recall_all_memories(self) -> List[str]: """ 저장된 모든 기억(단기 + 장기)을 리스트로 반환 """ return [m.content for m in self.memory_store.get_all_memories()] def get_composite_emotion_state(self, top_n: int = 3) -> List[tuple]: """ 현재 감정 상태 중 상위 N개를 반환 """ buffer = self.emotion.get_buffer() # EmotionManager에서 최신 감정 상태 복사본 가져오기 sorted_emotions = sorted(buffer.items(), key=lambda x: x[1], reverse=True) composite = [(emo,round(score, 2)) for emo, score in sorted_emotions[:top_n] if score > 0.1] return composite # 예 [('fear', 2.0), ('anger'), 1.5] def summarize_emotional_state(self) -> str: """ 감정 상태 요약 텍스트 생성 """ composite = self.get_composite_emotion_state() dominant = self.emotion.get_dominant_emotion() if not dominant: return "현재 특별한 감정 없이 평온한 상태입니다." return f"현재 {dominant} 등의 감정을 느끼고 있습니다.\n(주요 감정: {composite if composite else '없음'})" def get_relationship_description(self, target_name: str) -> str: """ 특정 대상과의 관계를 자연어로 반환 """ return self.relationships.get_relationship_summary(target_name) def get_top_emotions(self, top_n=6): """ 현재 NPC 감정 상태 상위 N개 반환 """ return self.emotion.get_top_emotions(top_n=top_n) def update_personality(self): """ 경험에 따라 성격을 점진적으로 변화시킴 - MemoryType 별 가중치 + EmotionInfluence 별 가중치 모두 반영 """ recent_memories = [m for m in self.memory_store.get_recent_memories(limit=10) if m.importance >= 5] if not recent_memories: return total_influence = {"sensitive": 0.0, "stoic": 0.0, "cognitive_bias": 0.0} # Personality 변화 반영 for mem in recent_memories: if not mem.emotion or mem.emotion not in EMOTION_RELATION_IMPACT: continue # 각 감정의 고유한 영향력(impact)을 가져옴 emotion_impact = EMOTION_RELATION_IMPACT.get(mem.emotion, 0.0) # 1. 감정의 긍정/부정/인지적 속성에 따른 영향 방향 결정 influence_vector = {"sensitive": 0.0, "stoic": 0.0, "cognitive_bias": 0.0} # 감정 영향력이 양수이면(긍정적 감정), 민감도와 내성을 낮춤 if mem.emotion in POSITIVE_RELATION_EMOTIONS: influence_vector["sensitive"] -= abs(emotion_impact) / 2.0 # 긍정적 경험은 민감도를 낮춤 influence_vector["stoic"] -= abs(emotion_impact) / 4.0 # 긍정적인 경험은 내성적일 필요를 줄임 # 감정 영향력이 음수이면(부정적 감정), 민감도와 내성을 높임 elif mem.emotion in NEGATIVE_RELATION_EMOTIONS: influence_vector["sensitive"] += abs(emotion_impact) / 2.0 # 부정적 경험은 민감도를 높임 influence_vector["stoic"] += abs(emotion_impact) / 4.0 # 부정적 경험은 내성적으로 만듦 # 인지적 감정은 cognitive_bias에 영향 if mem.emotion in COGNITIVE_RELATION_EMOTIONS: influence_vector["cognitive_bias"] += 0.4 # 인지적 경험은 논리적 사고를 강화 # 2. 변화 강도 계산 (기억의 중요도와 나이에 따라) age_factor = AGE_PROFILE.get(self.personality_stage, 1.0) change_strength = (mem.importance / 10.0) * age_factor # 3. 이번 기억으로 인한 최종 변화량 누적 for trait in total_influence.keys(): total_influence[trait] += influence_vector[trait] * change_strength # 4. 누적된 변화량을 성격에 적용 for trait, total_change in total_influence.items(): if abs(total_change) > 0.001: # 의미있는 변화만 적용 rate = self.personality_change_rate.get(trait, 0.003) delta = total_change * rate self.personality[trait] = max(0.0, min(1.0, self.personality[trait] + delta)) # 5. 기준선으로 회귀 (항상성) base_decay = 0.001 for trait in self.personality.keys(): decay_rate = base_decay * (1.0 + self.personality.get("stoic", 0.0) - self.personality.get("sensitive", 0.0)) decay_rate = max(0.0005, min(decay_rate, 0.005)) self.personality[trait] += (self.personality_baseline[trait] - self.personality[trait]) * decay_rate print(f"[Personality Update] {self.korean_name} → sensitive: {self.personality['sensitive']:.3f}, " f"stoic: {self.personality['stoic']:.3f}, cognitive_bias: {self.personality['cognitive_bias']:.3f}", flush=True) def decay_memories(self): """ 시간이 지남에 따라 기억 importance 감소 - MemoryStore의 decay_memories에 self(NPC 객체)를 전달 """ self.memory_store.decay_memories(npc=self) def update_personality_stage(self): """ 현재 나이에 따라 personality_stage 자동 업데이트 """ if self.age < 3: self.personality_stage = "infancy_and_toddlerhood" elif self.age < 6: self.personality_stage = "early_childhood" elif self.age < 12: self.personality_stage = "middle_childhood" elif self.age < 18: self.personality_stage = "adolescence" elif self.age < 30: self.personality_stage = "young_adulthood" elif self.age < 65: self.personality_stage = "middle_adulthood" else: self.personality_stage = "older_adulthood" def get_memory_importance_scaling(self, memory_age_in_days: float) -> float: """ NPC의 나이에 따라 memory_age_in_days 기반 scaling factor 반환 - 젊은 NPC → 최근 기억 더 강조 - 노년 NPC → 오래된 기억 더 강조 """ if self.personality_stage in [ "infancy_and_toddlerhood", "early_childhood", "middle_childhood", "adolescence", "young_adulthood" ]: # 최근 기억 강조 → 오래된 기억 영향 약화 return max(0.5, 1.5 - 0.05 * memory_age_in_days) # 최대 1.5, 점점 감소 elif self.personality_stage in [ "middle_adulthood", "older_adulthood" ]: return min(1.5, 0.5 + 0.05 * memory_age_in_days) # 최소 0.5, 점점 증가 else: # fallback → 정상화 return 1.0 def summarize_and_store_memories(self, memories: List[Memory]): """ 기억 리스트를 요약하고, NPC 장기 기억으로 저장 """ if not memories: return summary = summarize_memories(memories) if "[LLM Error]" in summary: return self.remember( content=f"[요약된 기억] {summary}", importance=9, emotion=self.emotion.get_dominant_emotion(), memory_type="Summary" ) def update_autonomous_behavior(self, time_context: str): """ NPC가 스스로 판단하여 행동을 결정하는 자율 행동 로직 주기적으로 호출되어 목표 설정, 계획 실행 등을 담당. """ # 플레이어가 비활성화 상태일 경우, 자율 행동을 실행하지 않도록 하는 안전장치 if self.name == "player" and not self.manager.player_is_active: return # 1. 계획 실행 로직을 NPC가 직접 처리 if self.planner.has_active_plan(): step = self.planner.get_current_step() if not step: self.planner.clear_plan() return print(f"[{self.korean_name}의 행동 계획 실행] {self.planner.current_plan.current_step_index + 1}/{len(self.planner.current_plan.steps)}: {step}") # LLM으로 계획을 분석하여 행동을 결정 plan_details = classify_and_extract_plan_details(step) action_type = plan_details.get("action_type") target_korean_name = plan_details.get("target") # 행동 실행 전 '조건 확인'단계 can_execute = True if target_korean_name == "플레이어" and not self.manager.player_is_active: # 목표 대상이 '플레이어'인데, 플레이어가 비활성화 상태일 경우 can_execute = False simulation_core.add_log(f"[{self.korean_name}의 생각] 지금은 플레이어가 없으니, '{step}' 계획은 나중에 다시 시도해야겠다.") return # 행동 실행 if action_type == "TALK" and target_korean_name: target_npc = self.manager.get_npc_by_korean_name(target_korean_name) if target_npc and target_npc != self: # 주제가 있는 대화 시작 simulation_core.conversation_manager.start_conversation(self, target_npc, topic=plan_details.get("topic")) elif action_type == "HELP" and target_korean_name: target_npc = self.manager.get_npc_by_korean_name(target_korean_name) target_postposition = get_korean_postposition(target_korean_name, "이", "가") target_postposition_Eul = get_korean_postposition(target_korean_name, "을", "를") if target_npc and target_npc != self: # '돕기' 행동의 결과를 시뮬레이션 outcome = evaluate_social_action_outcome(self, target_npc, "HELP") summary = outcome.get('outcome_summary') # 결과를 바탕으로 양쪽 모두의 상태를 업데이트 with simulation_core.simulation_lock: # 행동자(나)의 기억과 관계 업데이트 self.remember(content=f"{target_korean_name}{target_postposition_Eul} 도왔다. ({summary})", importance=8, emotion=outcome.get('initiator_emotion')) self.relationships.update_relationship(target_npc.name, outcome.get('initiator_emotion'), strength=3.0) # 대상자(상대방)의 기억과 관계 업데이트 target_npc.remember(content=f"{self.korean_name}{target_postposition} 나를 도와주었다. ({summary})") target_npc.relationships.update_relationship(self.name, outcome.get('target_emotion'), strength=3.0) else: # SOLOT_ACTION 또는 기타 # 대화가 아닌 다른 행동은 그대로 기억에만 기록 self.remember( content=f"'{self.planner.current_goal.description}' 목표를 위해 혼자 행동 했다: {step}", importance = 6, emotion="engagement", memory_type="Behavior" ) # 현재 단계 완료 처리 self.planner.complete_step() # 2. 계획이 없으면 새로운 목표를 생성하도록 시도 else: # 소문을 퍼뜨리려는 시도 gossip_memories = [m for m in self.memory_store.get_all_memories() if m.memory_type == "Gossip" and not m.is_shared] if gossip_memories and random.random() < 0.2: # 20% 확률로 소문 전파 시도 gossip_to_spread = random.choice(gossip_memories) # 아직 대화한 적 없는 NPC를 타겟으로 선정 potential_targets = [npc for npc in self.manager.get_interactive_npcs() if npc.name != self.name] if potential_targets: target_npc = random.choice(potential_targets) # LLM을 활용하여 소문을 퍼뜨리는 대화 생성 gossip_dialogue = self.generate_dialogue( user_input=f"'{target_npc.korean_name}'에게 '{gossip_to_spread.content}'에 대해 넌지시 이야기한다.", time_context=time_context, target_npc=target_npc, create_memory=False ) if gossip_dialogue and "[LLM Error]" not in gossip_dialogue: simulation_core.add_log(f"[소문 전파] {self.korean_name} -> {target_npc.korean_name}: {gossip_dialogue}") gossip_to_spread.is_shared = True # 2. '소문 전파'라는 행동에 대한 명확한 기억을 별도로 생성 self.remember( content=f"{target_npc.korean_name}에게 '{gossip_to_spread.content}'라는 소문에 대해 이야기했다.", importance=5, emotion="engagement", # '담합'과 같은 사회적 행동에 대한 감정 memory_type="Gossip_Spreading" # '소문 전파'라는 별도의 타입 지정 ) # 일정 확률로 새로운 목표를 생성 (매번 생성하지 않도록) if random.random() < 0.05: # 5% 확률 self.planner.generate_goal() elif random.random() < 0.1: # 10% 확률로 다른 NPC에게 가벼운 인사 시도 potential_targets = [npc for npc in self.manager.get_interactive_npcs() if npc != self] if potential_targets: weights = [] # 가중치 계산 로직 for target in potential_targets: # 관계 점수를 -100 ~ 100에서 0~2 범위로 변환하여 기본 가중치 1에 더함 # score 100 -> weight 6 / score 0 -> weight 1 / score -100 -> weight 0.1 (최소 확률) score = self.relationships.get_relationship_score(target.name) weight = 1 + (score / 20.0) weights.append(max(0.1, weight)) # 가중치가 0이 되지 않도록 최소값 보장 # 가중치에 따라 대화 상대 1명 선택 target_npc = random.choices(potential_targets, weights=weights, k=1)[0] # 주제가 없는 가벼운 대화 시작 simulation_core.conversation_manager.start_conversation(self, target_npc) def create_symbolic_memory(self): """ 최근 경험과 성격을 바탕으로 동적으로 성찰 주제를 정하고, 가치관을 형성 """ simulation_core.add_log(f"[{self.korean_name}: 깊은 생각에 잠깁니다...]") # 1. 최근 기억 10개를 가져옵니다. recent_memories = self.memory_store.get_all_memories()[-10:] if len(recent_memories) < 8: # 충분한 기억이 쌓엿을 때만 생성 return # 2. 성격 정보를 함께 전달하여 성찰 주제 생성 reflection_topic = generate_reflection_topic( recent_memories, self.personality.get_narrative_summary() ) simulation_core.add_log(f"[{self.korean_name}의 성찰 주제] {reflection_topic}") # 3. 생성된 주제와 가장 관련 있는 기억들을 다시 검색합니다. _, _, relevant_memories = search_similar_memories(self, reflection_topic, top_k=10) if not relevant_memories: relevant_memories = recent_memories # 관련 기억이 없으면 최근 기억을 그대로 사용 # 4. 관련 기억들을 요약 memory_summary = summarize_memories(relevant_memories) # 5. 요약된 기억과 성찰 주제를 바탕으로 최종 깨달음을 얻습니다. prompt = f""" # Instructions - 당신은 '{self.korean_name}'입니다. - 아래의 정보들을 바탕으로, 당신의 인생에 대한 깊은 깨달음이나 신념을 한 문장의 '상징적인 기억'으로 만들어주세요. # Reflection Topic "{reflection_topic}" # Summary of Related Memories {memory_summary} # My Realization (a one-sentence lesson or core belief): """ symbolic_memory_content = query_llm_with_prompt(prompt).strip() if symbolic_memory_content and "[LLM Error]" not in symbolic_memory_content: self.remember( content=symbolic_memory_content, importance=10, memory_type="Symbolic", emotion="realization" # 깨달음이라는 새로운 감정 ) simulation_core.add_log(f"[{self.korean_name}의 새로운 깨달음] {symbolic_memory_content}") def update_reputation(self, target_name: str, score_change: int): """특정 대상에 대한 평판 점수를 변경""" if self.name == target_name: return # 자기 자신에 대한 평판x (자기 자신의 평판에 대한 내용도 성격 형성에 중요할 수 있지 않나?) current_score = self.reputation.get(target_name, 0) self.reputation[target_name] = current_score + score_change target_npc = self.manager.get_npc_by_name(target_name) target_korean_name = target_npc.korean_name if target_npc else target_name print(f"[{self.korean_name}]의 {target_korean_name}에 대한 평판: {self.reputation[target_name]} ({score_change:+.0f})") def process_gossip_and_update_reputation(self, gossip_memory: Memory): """LLM을 이용해 소문을 분석하고 관련 인물들의 평판을 업데이트""" analysis = analyze_gossip(gossip_memory.content) if not analysis: return person1_korean_name = analysis.get("person1") person2_korean_name = analysis.get("person2") sentiment = analysis.get("sentiment") if not all([person1_korean_name, person2_korean_name, sentiment]): return score_change = 0 if sentiment == "positive": score_change = 2 elif sentiment == "negative": score_change = -2 if score_change != 0: # LLM이 반환한 한글 이름으로 NPC 객체를 찾아 영어 ID를 가져온다. p1_npc = self.manager.get_npc_by_korean_name(person1_korean_name) p2_npc = self.manager.get_npc_by_korean_name(person2_korean_name) # 소문의 대상이 된 두 사람에 대한 나의 평판을 변경 if p1_npc: self.update_reputation(p1_npc.name, score_change) if p2_npc: self.update_reputation(p2_npc.name, score_change)