diff --git a/engine-python/app/engine.py b/engine-python/app/engine.py index 69aa750..46c771c 100644 --- a/engine-python/app/engine.py +++ b/engine-python/app/engine.py @@ -6,6 +6,7 @@ from .models import ( ) from .global_events import GLOBAL_EVENT_POOL, GlobalEvent from .opinions import generate_opinions +from .social_influence import apply_social_influence MAX_EVENTS = 20 MAX_MEMORY_PER_AGENT = 3 @@ -86,6 +87,9 @@ def process_events(state: WorldState, events: List[Event]) -> WorldState: # 生成角色对事件的观点 generate_opinions(state) + # 应用社交影响 + apply_social_influence(state) + return state @@ -119,6 +123,9 @@ def check_and_trigger_global_event( # 生成角色对新事件的观点 generate_opinions(state) + # 应用社交影响 + apply_social_influence(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 b7de246..35962ad 100644 --- a/engine-python/app/models.py +++ b/engine-python/app/models.py @@ -38,6 +38,12 @@ class Opinion(BaseModel): tick: int # 生成时的 tick +class Stance(BaseModel): + """角色立场""" + optimism: float = Field(default=0.5, ge=0.0, le=1.0) + fear: float = Field(default=0.5, ge=0.0, le=1.0) + + class AgentState(BaseModel): emotion: Emotion = Emotion.CALM goal: str = "" @@ -45,6 +51,8 @@ class AgentState(BaseModel): opinion: Optional[Opinion] = None # 记录已评论过的事件,防止重复生成 commented_effects: List[str] = Field(default_factory=list) + # 角色立场 + stance: Stance = Field(default_factory=Stance) class WorldState(BaseModel): diff --git a/engine-python/app/opinions.py b/engine-python/app/opinions.py index db7e062..d649b11 100644 --- a/engine-python/app/opinions.py +++ b/engine-python/app/opinions.py @@ -1,6 +1,13 @@ """角色事件评论系统 - 基于规则生成观点""" from typing import Dict, List, Optional -from .models import WorldState, AgentState, Opinion, Emotion, WorldEffect +from .models import WorldState, AgentState, Opinion, Emotion, WorldEffect, Stance + +# 情绪对 stance 的影响 +EMOTION_STANCE_EFFECTS: Dict[str, Dict[str, float]] = { + "happy": {"optimism": 0.1, "fear": -0.05}, + "calm": {"optimism": 0.0, "fear": 0.0}, + "anxious": {"optimism": -0.05, "fear": 0.1}, +} # 观点模板:effect_type -> emotion -> 观点列表 @@ -123,6 +130,20 @@ def get_opinion_text( return templates[index] +def update_stance_from_emotion(agent: AgentState) -> None: + """根据情绪更新 stance(单次变化不超过 ±0.1)""" + emotion_key = agent.emotion.value + effects = EMOTION_STANCE_EFFECTS.get(emotion_key, {}) + + # 更新 optimism + new_optimism = agent.stance.optimism + effects.get("optimism", 0.0) + agent.stance.optimism = max(0.0, min(1.0, new_optimism)) + + # 更新 fear + new_fear = agent.stance.fear + effects.get("fear", 0.0) + agent.stance.fear = max(0.0, min(1.0, new_fear)) + + def generate_opinions(state: WorldState) -> None: """为所有 agent 生成对当前活跃事件的观点""" # 无活跃效果时,清空 opinion @@ -155,6 +176,9 @@ def generate_opinions(state: WorldState) -> None: tick=state.tick ) + # 根据情绪更新 stance + update_stance_from_emotion(agent) + # 记录到 memory memory_entry = f"[{agent_id}对{active_effect.name}的看法] {text}" agent.memory.append(memory_entry) diff --git a/engine-python/app/social_influence.py b/engine-python/app/social_influence.py new file mode 100644 index 0000000..76ed65f --- /dev/null +++ b/engine-python/app/social_influence.py @@ -0,0 +1,126 @@ +"""社交影响系统 - 角色间观点传播""" +from typing import Dict, List, Tuple +from .models import WorldState, AgentState, Stance + +# 单次 stance 变化上限 +MAX_STANCE_CHANGE = 0.1 + +# 影响强度系数 +INFLUENCE_STRENGTH = 0.05 + + +def get_contact_targets( + agent_id: str, + all_agent_ids: List[str], + tick: int +) -> List[str]: + """ + 确定性地选择 1-2 个接触对象 + 基于 agent_id 和 tick 确保可复现 + """ + # 排除自己 + others = [aid for aid in all_agent_ids if aid != agent_id] + if not others: + return [] + + # 基于 hash 确定性选择 + seed = hash(agent_id) + tick + + # 选择 1 或 2 个目标 + num_targets = 1 + (seed % 2) + num_targets = min(num_targets, len(others)) + + targets = [] + for i in range(num_targets): + idx = (seed + i * 7) % len(others) + target = others[idx] + if target not in targets: + targets.append(target) + + return targets + + +def calculate_influence( + agent_stance: Stance, + target_stance: Stance +) -> Tuple[float, float]: + """ + 计算社交影响产生的 stance 变化 + + 逻辑: + - 相近立场 → 强化原有立场 + - 相反立场 → 产生反向影响 + + 返回: (optimism_delta, fear_delta) + """ + # 计算差异 + opt_diff = target_stance.optimism - agent_stance.optimism + fear_diff = target_stance.fear - agent_stance.fear + + # 相近度判断(差异小于 0.3 视为相近) + opt_similar = abs(opt_diff) < 0.3 + fear_similar = abs(fear_diff) < 0.3 + + opt_delta = 0.0 + fear_delta = 0.0 + + if opt_similar: + # 相近 → 强化原有立场(向中间靠拢后再强化) + opt_delta = opt_diff * INFLUENCE_STRENGTH + else: + # 相反 → 反向影响(抵抗) + opt_delta = -opt_diff * INFLUENCE_STRENGTH * 0.5 + + if fear_similar: + fear_delta = fear_diff * INFLUENCE_STRENGTH + else: + fear_delta = -fear_diff * INFLUENCE_STRENGTH * 0.5 + + # 限制单次变化幅度 + opt_delta = max(-MAX_STANCE_CHANGE, min(MAX_STANCE_CHANGE, opt_delta)) + fear_delta = max(-MAX_STANCE_CHANGE, min(MAX_STANCE_CHANGE, fear_delta)) + + return opt_delta, fear_delta + + +def apply_social_influence(state: WorldState) -> None: + """ + 应用社交影响:每个 agent 与 1-2 个其他 agent 接触 + 所有变化写入 world_state + """ + agent_ids = list(state.agents.keys()) + if len(agent_ids) < 2: + return + + # 收集所有变化(先计算,后应用,避免顺序影响) + stance_changes: Dict[str, Tuple[float, float]] = {} + + for agent_id in agent_ids: + agent = state.agents[agent_id] + targets = get_contact_targets(agent_id, agent_ids, state.tick) + + total_opt_delta = 0.0 + total_fear_delta = 0.0 + + for target_id in targets: + target = state.agents[target_id] + opt_delta, fear_delta = calculate_influence( + agent.stance, target.stance + ) + total_opt_delta += opt_delta + total_fear_delta += fear_delta + + # 限制总变化 + total_opt_delta = max(-MAX_STANCE_CHANGE, min(MAX_STANCE_CHANGE, total_opt_delta)) + total_fear_delta = max(-MAX_STANCE_CHANGE, min(MAX_STANCE_CHANGE, total_fear_delta)) + + stance_changes[agent_id] = (total_opt_delta, total_fear_delta) + + # 应用所有变化 + for agent_id, (opt_delta, fear_delta) in stance_changes.items(): + agent = state.agents[agent_id] + new_opt = agent.stance.optimism + opt_delta + new_fear = agent.stance.fear + fear_delta + + agent.stance.optimism = max(0.0, min(1.0, new_opt)) + agent.stance.fear = max(0.0, min(1.0, new_fear)) diff --git a/engine-python/app/state.py b/engine-python/app/state.py index cf8dcb3..c9fa6ed 100644 --- a/engine-python/app/state.py +++ b/engine-python/app/state.py @@ -1,6 +1,6 @@ import json from pathlib import Path -from .models import WorldState, AgentState, GlobalMeter +from .models import WorldState, AgentState, GlobalMeter, Stance STATE_FILE = Path(__file__).parent.parent / "state.json" @@ -12,8 +12,18 @@ def get_default_state() -> WorldState: weather="sunny", town_mood=0, agents={ - "alice": AgentState(emotion="calm", goal="探索小镇", memory=[]), - "bob": AgentState(emotion="calm", goal="与人交流", memory=[]), + "alice": AgentState( + emotion="calm", + goal="探索小镇", + memory=[], + stance=Stance(optimism=0.6, fear=0.4), + ), + "bob": AgentState( + emotion="calm", + goal="与人交流", + memory=[], + stance=Stance(optimism=0.4, fear=0.6), + ), }, events=[], global_meter=GlobalMeter(value=0, threshold=100, cooldown=0), diff --git a/engine-python/tests/test_step.py b/engine-python/tests/test_step.py index 7edec42..23daf76 100644 --- a/engine-python/tests/test_step.py +++ b/engine-python/tests/test_step.py @@ -229,3 +229,69 @@ def test_opinion_cleared_when_no_effect(client): # 无世界事件时,opinion 应为 None for agent_id, agent in data["world_state"]["agents"].items(): assert agent["opinion"] is None + + +def test_stance_exists_in_agent(client): + """测试 agent 包含 stance 字段""" + resp = client.post("/step", json={"events": []}) + data = resp.json() + + for agent_id, agent in data["world_state"]["agents"].items(): + assert "stance" in agent + assert "optimism" in agent["stance"] + assert "fear" in agent["stance"] + assert 0.0 <= agent["stance"]["optimism"] <= 1.0 + assert 0.0 <= agent["stance"]["fear"] <= 1.0 + + +def test_stance_changes_on_world_event(client): + """测试世界事件触发时 stance 变化""" + # 记录初始 stance + resp1 = client.post("/step", json={"events": []}) + initial_stances = { + aid: agent["stance"].copy() + for aid, agent in resp1.json()["world_state"]["agents"].items() + } + + # 触发世界事件 + events = [ + {"type": "chaos", "text": f"chaos{i}", "user": "test", "ts": i} + for i in range(20) + ] + resp2 = client.post("/step", json={"events": events}) + data = resp2.json() + + # 验证事件触发 + assert data["global_event"]["triggered"] is True + + # stance 应该有变化(由于情绪和社交影响) + for aid, agent in data["world_state"]["agents"].items(): + # stance 值仍在有效范围内 + assert 0.0 <= agent["stance"]["optimism"] <= 1.0 + assert 0.0 <= agent["stance"]["fear"] <= 1.0 + + +def test_social_influence_deterministic(client): + """测试社交影响是确定性的(同输入同输出)""" + events = [{"type": "comment", "text": "hello", "user": "test", "ts": 1}] + + # 第一次执行 + save_state(get_default_state()) + resp1 = client.post("/step", json={"events": events}) + stances1 = { + aid: agent["stance"] + for aid, agent in resp1.json()["world_state"]["agents"].items() + } + + # 重置并再次执行 + save_state(get_default_state()) + resp2 = client.post("/step", json={"events": events}) + stances2 = { + aid: agent["stance"] + for aid, agent in resp2.json()["world_state"]["agents"].items() + } + + # 结果应该相同 + for aid in stances1: + assert stances1[aid]["optimism"] == stances2[aid]["optimism"] + assert stances1[aid]["fear"] == stances2[aid]["fear"]