Spaces:
Sleeping
Sleeping
# 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) |