ai-town/engine-python/app/engine.py

from __future__ import annotations
from typing import List, Optional, Tuple
from .models import (
    WorldState, AgentState, Event, Action, Emotion, Weather,
    GlobalEventResult, GlobalEventInfo
)
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
COOLDOWN_TICKS = 5

# 事件类型对应的能量值
ENERGY_MAP = {
    "comment": 1,
    "support": 3,
    "chaos": 5,
}
DEFAULT_ENERGY = 1


def update_world_effects(state: WorldState) -> None:
    """更新世界效果生命周期，应用持续影响"""
    # 递减 remaining_ticks，移除过期效果
    active_effects = []
    for effect in state.world_effects:
        effect.remaining_ticks -= 1
        if effect.remaining_ticks > 0:
            active_effects.append(effect)
            # 应用持续情绪影响
            state.town_mood += effect.mood_modifier
            state.town_mood = max(-10, min(10, state.town_mood))
    state.world_effects = active_effects


def process_events(state: WorldState, events: List[Event]) -> WorldState:
    """处理事件并更新世界状态"""
    # tick 递增
    state.tick += 1

    # cooldown 递减
    if state.global_meter.cooldown > 0:
        state.global_meter.cooldown -= 1

    # 更新世界效果（持续影响）
    update_world_effects(state)

    for event in events:
        text = event.text

        # 累计能量
        energy = ENERGY_MAP.get(event.type, DEFAULT_ENERGY)
        state.global_meter.value += energy

        # 天气规则
        if "下雨" in text:
            state.weather = Weather.RAINY
        if "天晴" in text:
            state.weather = Weather.SUNNY

        # 情绪规则
        if "支持" in text:
            state.town_mood = min(10, state.town_mood + 1)
        if "混乱" in text or "讨厌" in text:
            state.town_mood = max(-10, state.town_mood - 2)

        # 添加事件到全局事件列表
        event_str = f"{event.user}: {text}"
        state.events.append(event_str)

        # 添加到每个 agent 的 memory
        for agent in state.agents.values():
            agent.memory.append(event_str)
            # 保持最多3条
            if len(agent.memory) > MAX_MEMORY_PER_AGENT:
                agent.memory = agent.memory[-MAX_MEMORY_PER_AGENT:]

    # 保持全局事件最多20条
    if len(state.events) > MAX_EVENTS:
        state.events = state.events[-MAX_EVENTS:]

    # 更新所有 agent 的 emotion
    update_emotions(state)

    # 生成角色对事件的观点
    generate_opinions(state)

    # 应用社交影响
    apply_social_influence(state)

    return state


def check_and_trigger_global_event(
    state: WorldState
) -> Tuple[WorldState, GlobalEventResult]:
    """检查并触发世界级事件"""
    meter = state.global_meter
    result = GlobalEventResult(triggered=False, event=None)

    # 检查触发条件
    if meter.value >= meter.threshold and meter.cooldown == 0:
        # 基于 tick 选择事件（确保可复现）
        event_index = state.tick % len(GLOBAL_EVENT_POOL)
        global_event = GLOBAL_EVENT_POOL[event_index]

        # 应用即时效果
        global_event.apply_effect(state)

        # 生成持续影响效果
        world_effect = global_event.create_world_effect()
        state.world_effects.append(world_effect)

        # 重置能量条，设置冷却
        state.global_meter.value = 0
        state.global_meter.cooldown = COOLDOWN_TICKS

        # 更新 emotion
        update_emotions(state)

        # 生成角色对新事件的观点
        generate_opinions(state)

        # 应用社交影响
        apply_social_influence(state)

        result = GlobalEventResult(
            triggered=True,
            event=global_event.to_info()
        )

    return state, result


def update_emotions(state: WorldState) -> None:
    """根据 town_mood 更新所有 agent 的 emotion"""
    if state.town_mood >= 4:
        emotion = Emotion.HAPPY
    elif state.town_mood <= -4:
        emotion = Emotion.ANXIOUS
    else:
        emotion = Emotion.CALM

    for agent in state.agents.values():
        agent.emotion = emotion


def generate_actions(state: WorldState) -> List[Action]:
    """根据当前状态生成 agent 的行动"""
    actions = []

    for agent_id in state.agents:
        agent = state.agents[agent_id]
        action = generate_agent_action(agent_id, agent, state)
        actions.append(action)

    return actions


def generate_agent_action(
    agent_id: str, agent: AgentState, state: WorldState
) -> Action:
    """为单个 agent 生成行动（基于规则模板）"""
    weather = state.weather
    emotion = agent.emotion

    # Alice 的行为模板
    if agent_id == "alice":
        if weather == Weather.RAINY:
            if emotion == Emotion.HAPPY:
                say = "虽然下雨了，但心情还是很好呢！"
                do = "在屋檐下哼着歌"
            elif emotion == Emotion.ANXIOUS:
                say = "这雨什么时候才能停啊..."
                do = "焦虑地看着窗外"
            else:
                say = "下雨了，待在室内吧。"
                do = "在室内看书"
        else:  # sunny
            if emotion == Emotion.HAPPY:
                say = "今天天气真好，心情也很棒！"
                do = "在广场上散步"
            elif emotion == Emotion.ANXIOUS:
                say = "天气虽好，但总觉得不安..."
                do = "在小镇边缘徘徊"
            else:
                say = "天气不错，出去走走吧。"
                do = "在街道上闲逛"

    # Bob 的行为模板
    elif agent_id == "bob":
        if weather == Weather.RAINY:
            if emotion == Emotion.HAPPY:
                say = "雨天也有雨天的乐趣！"
                do = "在咖啡馆和人聊天"
            elif emotion == Emotion.ANXIOUS:
                say = "这种天气让人心烦..."
                do = "独自待在角落"
            else:
                say = "下雨天适合思考。"
                do = "在窗边沉思"
        else:  # sunny
            if emotion == Emotion.HAPPY:
                say = "阳光明媚，适合交朋友！"
                do = "主动和路人打招呼"
            elif emotion == Emotion.ANXIOUS:
                say = "人太多了，有点紧张..."
                do = "躲在树荫下观察"
            else:
                say = "今天适合出门。"
                do = "在市场闲逛"

    # 默认行为（其他 agent）
    else:
        say = f"我是{agent_id}，今天感觉{emotion.value}。"
        do = "四处走动"

    return Action(agent_id=agent_id, say=say, do=do)