diff --git a/engine-python/app/engine.py b/engine-python/app/engine.py index 7a1b4e0..69aa750 100644 --- a/engine-python/app/engine.py +++ b/engine-python/app/engine.py @@ -5,6 +5,7 @@ from .models import ( GlobalEventResult, GlobalEventInfo ) from .global_events import GLOBAL_EVENT_POOL, GlobalEvent +from .opinions import generate_opinions MAX_EVENTS = 20 MAX_MEMORY_PER_AGENT = 3 @@ -82,6 +83,9 @@ def process_events(state: WorldState, events: List[Event]) -> WorldState: # 更新所有 agent 的 emotion update_emotions(state) + # 生成角色对事件的观点 + generate_opinions(state) + return state @@ -112,6 +116,9 @@ def check_and_trigger_global_event( # 更新 emotion update_emotions(state) + # 生成角色对新事件的观点 + generate_opinions(state) + result = GlobalEventResult( triggered=True, event=global_event.to_info() diff --git a/engine-python/app/models.py b/engine-python/app/models.py index e22fd6c..b7de246 100644 --- a/engine-python/app/models.py +++ b/engine-python/app/models.py @@ -31,10 +31,20 @@ class WorldEffect(BaseModel): mood_modifier: int = 0 +class Opinion(BaseModel): + """角色对事件的观点""" + about: str # 事件类型 (effect_type) + text: str # 观点内容 + tick: int # 生成时的 tick + + class AgentState(BaseModel): emotion: Emotion = Emotion.CALM goal: str = "" memory: List[str] = Field(default_factory=list) + opinion: Optional[Opinion] = None + # 记录已评论过的事件，防止重复生成 + commented_effects: List[str] = Field(default_factory=list) class WorldState(BaseModel): diff --git a/engine-python/app/opinions.py b/engine-python/app/opinions.py new file mode 100644 index 0000000..db7e062 --- /dev/null +++ b/engine-python/app/opinions.py @@ -0,0 +1,163 @@ +"""角色事件评论系统 - 基于规则生成观点""" +from typing import Dict, List, Optional +from .models import WorldState, AgentState, Opinion, Emotion, WorldEffect + + +# 观点模板：effect_type -> emotion -> 观点列表 +OPINION_TEMPLATES: Dict[str, Dict[str, List[str]]] = { + "festival": { + "happy": [ + "这真是个让人开心的日子！", + "节日的气氛太棒了！", + "大家都在庆祝，真好！", + ], + "calm": [ + "节日挺热闹的。", + "看来大家都很高兴。", + "难得的庆典活动。", + ], + "anxious": [ + "我不太确定这是不是好事...", + "人太多了，有点不安。", + "希望一切顺利吧。", + ], + }, + "storm": { + "happy": [ + "暴风雨也有它的美！", + "待在室内也挺好的。", + "雨声其实很治愈。", + ], + "calm": [ + "暴风雨来了，注意安全。", + "等雨停了再出门吧。", + "天气变化无常啊。", + ], + "anxious": [ + "这暴风雨让我很担心...", + "情况看起来不太妙。", + "希望大家都平安。", + ], + }, + "scandal": { + "happy": [ + "不管怎样，生活还要继续！", + "真相总会大白的。", + "保持乐观吧。", + ], + "calm": [ + "这件事确实令人意外。", + "让我们静观其变。", + "事情总会解决的。", + ], + "anxious": [ + "这消息太令人震惊了...", + "不知道接下来会怎样。", + "小镇的未来让人担忧。", + ], + }, + "miracle": { + "happy": [ + "太神奇了！这是奇迹！", + "彩虹真的太美了！", + "今天是幸运的一天！", + ], + "calm": [ + "难得一见的景象。", + "大自然真是神奇。", + "值得记住的时刻。", + ], + "anxious": [ + "希望这是好兆头...", + "美丽的景象，但愿能持续。", + "也许事情会好转吧。", + ], + }, + "fire": { + "happy": [ + "消防队很快就会控制住的！", + "大家齐心协力，一定没问题！", + "相信我们能度过难关！", + ], + "calm": [ + "火灾很危险，大家要小心。", + "希望没有人受伤。", + "消防队正在处理。", + ], + "anxious": [ + "情况变得不妙，我得小心...", + "火灾太可怕了...", + "希望大家都能平安。", + ], + }, +} + +# 默认观点（未知事件类型） +DEFAULT_OPINIONS: Dict[str, List[str]] = { + "happy": ["不管发生什么，保持乐观！"], + "calm": ["让我们看看会怎样。"], + "anxious": ["不知道这意味着什么..."], +} + + +def get_opinion_text( + effect_type: str, + emotion: Emotion, + agent_id: str, + tick: int +) -> str: + """根据事件类型和情绪生成观点文本""" + emotion_key = emotion.value + + # 获取对应模板 + if effect_type in OPINION_TEMPLATES: + templates = OPINION_TEMPLATES[effect_type].get( + emotion_key, + DEFAULT_OPINIONS[emotion_key] + ) + else: + templates = DEFAULT_OPINIONS[emotion_key] + + # 基于 agent_id 和 tick 选择模板（确保可复现） + index = (hash(agent_id) + tick) % len(templates) + return templates[index] + + +def generate_opinions(state: WorldState) -> None: + """为所有 agent 生成对当前活跃事件的观点""" + # 无活跃效果时，清空 opinion + if not state.world_effects: + for agent in state.agents.values(): + agent.opinion = None + return + + # 取第一个活跃效果（优先级最高） + active_effect = state.world_effects[0] + effect_key = f"{active_effect.type}_{active_effect.name}" + + for agent_id, agent in state.agents.items(): + # 检查是否已对此事件发表过观点 + if effect_key in agent.commented_effects: + continue + + # 生成观点 + text = get_opinion_text( + active_effect.type, + agent.emotion, + agent_id, + state.tick + ) + + # 更新 agent.opinion + agent.opinion = Opinion( + about=active_effect.type, + text=text, + tick=state.tick + ) + + # 记录到 memory + memory_entry = f"[{agent_id}对{active_effect.name}的看法] {text}" + agent.memory.append(memory_entry) + + # 标记已评论 + agent.commented_effects.append(effect_key) diff --git a/engine-python/tests/test_step.py b/engine-python/tests/test_step.py index fdb0022..7edec42 100644 --- a/engine-python/tests/test_step.py +++ b/engine-python/tests/test_step.py @@ -171,3 +171,61 @@ def test_global_event_cooldown(client): assert data["global_event"]["triggered"] is False # 冷却递减 assert data["world_state"]["global_meter"]["cooldown"] == 4 + + +def test_opinion_generated_on_world_effect(client): + """测试世界事件触发时生成观点""" + # 触发世界事件 + events = [ + {"type": "chaos", "text": f"chaos{i}", "user": "test", "ts": i} + for i in range(20) + ] + resp = client.post("/step", json={"events": events}) + data = resp.json() + + # 验证事件触发 + assert data["global_event"]["triggered"] is True + assert len(data["world_state"]["world_effects"]) > 0 + + # 验证每个 agent 都有 opinion + for agent_id, agent in data["world_state"]["agents"].items(): + assert agent["opinion"] is not None + assert "about" in agent["opinion"] + assert "text" in agent["opinion"] + assert "tick" in agent["opinion"] + + +def test_opinion_only_once_per_effect(client): + """测试同一事件只生成一次观点""" + # 触发世界事件 + events = [ + {"type": "chaos", "text": f"chaos{i}", "user": "test", "ts": i} + for i in range(20) + ] + resp = client.post("/step", json={"events": events}) + data = resp.json() + + # 记录第一次的观点 + first_opinions = { + aid: agent["opinion"]["text"] + for aid, agent in data["world_state"]["agents"].items() + } + + # 再次调用 step（事件仍在持续） + resp2 = client.post("/step", json={"events": []}) + data2 = resp2.json() + + # 观点应该保持不变（不重新生成） + for aid, agent in data2["world_state"]["agents"].items(): + if agent["opinion"]: + assert agent["opinion"]["text"] == first_opinions[aid] + + +def test_opinion_cleared_when_no_effect(client): + """测试无活跃事件时观点为空""" + resp = client.post("/step", json={"events": []}) + data = resp.json() + + # 无世界事件时，opinion 应为 None + for agent_id, agent in data["world_state"]["agents"].items(): + assert agent["opinion"] is None