app.py

# ---------- 1) Imports & Config ----------
import os, uuid, json, random
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Any
import gradio as gr

# 叙事后端选择: "hf" (CPU, falcon-1b-instruct) 或 "openai"
NARRATION_BACKEND = os.getenv("NARRATION_BACKEND", "hf")

# ---------- 9) Narration Backend ----------
if NARRATION_BACKEND == "hf":
    from transformers import pipeline
    generator = pipeline("text-generation", model="tiiuae/falcon-1b-instruct", device=-1)
    def narrate(prompt:str) -> str:
        out = generator(
            "你是合欢宗的旁白，写狗血八卦日记，中文为主，轻混英文术语（jealousy, scandal, oath），不得露骨：\n"+prompt,
            max_new_tokens=180, do_sample=True, temperature=0.9, top_p=0.95
        )[0]["generated_text"]
        return out[-600:].strip()
else:
    from openai import OpenAI
    client = OpenAI()
    def narrate(prompt:str) -> str:
        res = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role":"system","content":"你是合欢宗的旁白，要写成狗血八卦日记，中文为主，轻混英文术语（jealousy, scandal, oath），不得露骨。每次 2-4 句。"},
                {"role":"user","content": prompt}
            ],
            temperature=0.9, max_tokens=300
        )
        return res.choices[0].message.content.strip()

# ---------- 2) Data Models ----------
Day = int

@dataclass
class Agent:
    name: str
    role: str
    cultivation: int = 50
    willpower: int = 50
    face: int = 0
    jealousy: int = 0
    scandal: int = 0
    memory: Dict[str, Any] = field(default_factory=dict)  # private KnowledgeItems by fact_key
    belief_about_others: Dict[str, Dict[str, Any]] = field(default_factory=dict)  # ToM

@dataclass
class Event:
    id: str
    day: Day
    type: str
    actors: List[str]
    payload: Dict[str, Any]
    truth_strength: float = 1.0

@dataclass
class Observation:
    event_id: str
    observer: str
    day: Day
    mode: str         # "direct" | "overheard" | "told"
    noise: float
    cred: float

@dataclass
class KnowledgeItem:
    id: str
    fact_key: str
    content: Dict[str, Any]
    first_seen_day: Day
    last_update_day: Day
    confidence: float
    sources: List[Dict[str, Any]] = field(default_factory=list)
    is_public: bool = False
    visibility: str = "private"     # "private"|"public"

@dataclass
class World:
    day: Day = 0
    agents: List[Agent] = field(default_factory=list)
    relations: Dict[str, Dict[str, int]] = field(default_factory=dict)  # [-100,100]
    events: Dict[str, Event] = field(default_factory=dict)
    observations: List[Observation] = field(default_factory=list)
    public_board: List[KnowledgeItem] = field(default_factory=list)
    rumor_level: int = 10
    season: str = "normal"

    def snapshot(self):
        return {
            "day": self.day,
            "agents": [asdict(a) for a in self.agents],
            "rumor_level": self.rumor_level,
            "season": self.season,
            "recent_public": [asdict(x) for x in self.public_board[-8:]],
        }

# ---------- 3) Init ----------
ROLES = ["Saintess","Senior","Enforcer","Alchemist","ArrayMaster","Bard","Matchmaker","Shadow","Guest","Head"]

def init_world(seed=42) -> World:
    random.seed(seed)
    ags = [Agent(name=f"A{i+1}", role=ROLES[i]) for i in range(10)]
    names = [a.name for a in ags]
    rel = {u:{v:(0 if u==v else random.randint(-10,10)) for v in names} for u in names}
    return World(day=0, agents=ags, relations=rel)

WORLD = init_world()

# ---------- 4) Info System: helpers ----------
def make_event(day, typ, actors, payload, truth_strength=1.0):
    return Event(id=str(uuid.uuid4())[:8], day=day, type=typ, actors=actors, payload=payload, truth_strength=truth_strength)

def sample_observers(world:World, event:Event, base_p=0.2):
    observers = []
    for ag in world.agents:
        if ag.name in event.actors: continue
        # 关系平均值影响观测概率
        rel = sum(world.relations[ag.name][x] for x in event.actors)/max(1,len(event.actors))
        p = min(0.85, max(0.05, base_p + rel/200))
        if random.random() < p:
            observers.append(Observation(
                event_id=event.id, observer=ag.name, day=event.day,
                mode=random.choice(["direct","overheard"]), noise=random.uniform(0,0.25),
                cred=random.uniform(0.6,0.9)
            ))
    return observers

def upsert_memory(agent:Agent, fact_key:str, content:dict, day:int, source:dict, base_conf:float):
    it = agent.memory.get(fact_key)
    if it is None:
        it = KnowledgeItem(
            id=str(uuid.uuid4())[:8], fact_key=fact_key, content=content,
            first_seen_day=day, last_update_day=day, confidence=base_conf,
            sources=[source], is_public=False, visibility="private"
        )
        agent.memory[fact_key] = it
    else:
        c0, c1 = it.confidence, base_conf
        it.confidence = 1 - (1-c0)*(1-c1)  # 概率互补合并
        it.last_update_day = day
        it.sources.append(source)
    return it

def publish_public(world:World, fact_key:str, content:dict, day:int, base_cred:float, source_tag:str):
    ki = KnowledgeItem(
        id=str(uuid.uuid4())[:8], fact_key=fact_key, content=content,
        first_seen_day=day, last_update_day=day, confidence=base_cred,
        sources=[{"src":source_tag,"cred":base_cred}], is_public=True, visibility="public"
    )
    world.public_board.append(ki); return ki

def daily_decay(world:World, decay=0.98):
    for a in world.agents:
        for it in a.memory.values():
            age = world.day - it.last_update_day
            if age>0:
                it.confidence = max(0.01, it.confidence*(decay**age))
    for it in world.public_board:
        age = world.day - it.last_update_day
        if age>0:
            it.confidence = max(0.01, it.confidence*(decay**age))

def inform(world:World, speaker:str, listener:str, fact_key:str, boost=0.15):
    s = next(a for a in world.agents if a.name==speaker)
    l = next(a for a in world.agents if a.name==listener)
    it = s.memory.get(fact_key)
    if not it: return "speaker knows nothing."
    rel = world.relations[speaker][listener]
    cred = min(0.95, max(0.2, it.confidence + boost + rel/200))
    upsert_memory(l, fact_key, it.content, world.day, {"src":f"told_by_{speaker}","cred":cred}, cred)
    # ToM: 我认为你知道了
    s.belief_about_others.setdefault(listener, {})[fact_key] = {
        "id": str(uuid.uuid4())[:8], "conf":cred, "content":it.content, "last_update_day": world.day
    }
    return "ok"

# ---------- 5) Actions（合欢宗简化版） ----------
def act_gift(world:World, giver:str, receiver:str, place="藏书阁"):
    ev = make_event(world.day, "gift", [giver, receiver], {"place":place,"item":"清心茶"})
    world.events[ev.id] = ev
    # 当事人记忆
    for nm in [giver, receiver]:
        ag = next(a for a in world.agents if a.name==nm)
        fk = f"gift_{giver}_{receiver}_d{world.day}"
        upsert_memory(ag, fk, {"type":"gift","giver":giver,"receiver":receiver,"place":place},
                      world.day, {"src":"self_participant","cred":0.9}, 0.9)
    # 旁观者
    obs = sample_observers(world, ev, base_p=0.18)
    world.observations += obs
    for o in obs:
        ag = next(a for a in world.agents if a.name==o.observer)
        fk = f"gift_{giver}_{receiver}_d{world.day}"
        conf = max(0.2, o.cred*(1-o.noise))
        upsert_memory(ag, fk, {"type":"gift_hint","giver":giver,"receiver":receiver,"place":place},
                      world.day, {"src":o.mode,"cred":conf}, conf)
    # 流言上板
    if random.random() < 0.25:
        publish_public(world, f"gift_{giver}_{receiver}",
                       {"hint":"有人在藏书阁递了东西"}, world.day, 0.55, "rumor")

def act_confide(world:World, a:str, b:str):
    ev = make_event(world.day, "confide", [a,b], {"topic":"心法共鸣"})
    world.events[ev.id] = ev
    for nm in [a,b]:
        ag = next(x for x in world.agents if x.name==nm)
        fk = f"confide_{a}_{b}_d{world.day}"
        upsert_memory(ag, fk, {"type":"confide","pair":[a,b]}, world.day, {"src":"private_talk","cred":0.95}, 0.95)
    for o in sample_observers(world, ev, 0.12):
        ag = next(x for x in world.agents if x.name==o.observer)
        fk = f"confide_{a}_{b}_d{world.day}"
        conf = max(0.15, o.cred*(1-o.noise))
        upsert_memory(ag, fk, {"type":"confide_hint","pair":[a,b]}, world.day, {"src":o.mode,"cred":conf}, conf)

def act_expose(world:World, accuser:str, target:str, evidence:str):
    ev = make_event(world.day, "expose", [accuser,target], {"evidence":evidence})
    world.events[ev.id] = ev
    fk = f"expose_{target}_d{world.day}"
    pub = publish_public(world, fk, {"type":"expose","target":target,"evidence":evidence}, world.day, 0.8, f"expose_by_{accuser}")
    for ag in world.agents:
        upsert_memory(ag, fk, pub.content, world.day, {"src":"public_board","cred":pub.confidence}, pub.confidence)

# 其他动作：rumor/oath/sabotage/mediate/spar ...（按需补全）

# ---------- 6) Policy ----------
def agent_policy(world:World, actor:Agent):
    # 简化：按角色偏好
    names = [a.name for a in world.agents if a.name != actor.name]
    tgt = random.choice(names)
    if actor.role in ["Matchmaker"]: return ("oath", tgt)
    if actor.role in ["Bard"]:       return ("rumor", tgt)
    if actor.role in ["Shadow"]:     return ("expose", tgt)
    if actor.role in ["Saintess"]:   return ("mediate", tgt)
    return random.choice([("gift", tgt), ("confide", tgt), ("spar", tgt)])

def apply_action(world:World, actor:str, action:str, target:str):
    if action=="gift":    act_gift(world, actor, target);         return f"{actor}→{target} 赠礼"
    if action=="confide": act_confide(world, actor, target);      return f"{actor}↔{target} 推心置腹"
    if action=="expose":  act_expose(world, actor, target, "残页证据"); return f"{actor} 公揭 {target}"
    # TODO: rumor/oath/sabotage/mediate/spar...
    return f"{actor} 今日独自修行"

# ---------- 7) Tick (推进一天) ----------
def tick_one_day(world:World):
    world.day += 1
    daily_events = []
    # 每人 1 动作
    for ag in world.agents:
        act, tgt = agent_policy(world, ag)
        daily_events.append(apply_action(world, ag.name, act, tgt))
    # 信息衰减
    daily_decay(world, decay=0.98)
    # Narration
    snap = json.dumps(world.snapshot(), ensure_ascii=False)[:2000]
    prompt = f"DAY={world.day}\nWORLD_SNAPSHOT={snap}\nKEY_EVENTS={daily_events}\n请写今日纪要（2-4句）：1) 核心八卦；2) 关系走向；3) 明日伏笔（可选）。"
    narration = narrate(prompt)
    return narration

# ---------- 8) God Mode ----------
def god_action(world:World, action:str, target:str="", value:int=0):
    if action=="peach_blossom_trial" and target:
        # 让所有人对 target 好感上升（用 relations 近似）
        for ag in world.agents:
            if ag.name!=target:
                world.relations[ag.name][target] = min(100, world.relations[ag.name][target] + random.randint(2,6))
        world.rumor_level += 10
        return f"桃花劫降临 {target}"
    if action=="inner_demon" and target:
        ag = next(a for a in world.agents if a.name==target)
        diff = random.randint(0,100) - (ag.willpower + value)
        if diff>0:
            ag.jealousy = min(100, ag.jealousy+12)
            ag.scandal  = min(100, ag.scandal+8)
            return f"{target} 心魔试炼失败"
        return f"{target} 心神稳固"
    if action=="seclusion" and target:
        # 简化：只记一条 public 提示
        publish_public(world, f"seclusion_{target}", {"type":"seclusion","target":target,"days":value or 3}, world.day, 0.7, "decree")
        return f"{target} 闭关 {value or 3} 日"
    if action=="grand_banquet":
        world.season="banquet"; world.rumor_level+=15
        return "宗门盛会开启"
    if action=="rumor_storm":
        for ag in world.agents: ag.scandal = min(100, ag.scandal+5)
        world.rumor_level += 12; return "流言四起，众心不安"
    if action=="grant_fate" and target:
        # 造一条“赐缘”公共事实（示意）
        publish_public(world, f"grant_{target}", {"type":"grant_fate","target":target}, world.day, 0.75, "heaven")
        return f"天机点化 {target}"
    return "unknown god action"

# ---------- 10) Gradio UI ----------
def ui_init(seed):
    global WORLD; WORLD = init_world(int(seed) if seed else 42)
    return f"Reset ok. Day={WORLD.day}", json.dumps(WORLD.snapshot(), ensure_ascii=False, indent=2)

def ui_next_day():
    nar = tick_one_day(WORLD)
    return nar, json.dumps(WORLD.snapshot(), ensure_ascii=False, indent=2)

def ui_auto(days:int):
    days = max(1, min(30, int(days)))
    logs = []
    for _ in range(days): logs.append(tick_one_day(WORLD))
    return "\n\n".join(logs), json.dumps(WORLD.snapshot(), ensure_ascii=False, indent=2)

def ui_god(act, target, value):
    msg = god_action(WORLD, act, target.strip(), int(value) if value else 0)
    return msg, json.dumps(WORLD.snapshot(), ensure_ascii=False, indent=2)

def ui_public():
    data = [asdict(x) for x in WORLD.public_board[-20:]]
    return json.dumps(data, ensure_ascii=False, indent=2)

def ui_memory(agent_name):
    ag = next((a for a in WORLD.agents if a.name==agent_name), None)
    if not ag: return "no such agent"
    data = {k:asdict(v) for k,v in ag.memory.items()}
    return json.dumps(data, ensure_ascii=False, indent=2)

def ui_inform(speaker, listener, fact_key):
    return inform(WORLD, speaker.strip(), listener.strip(), fact_key.strip())

with gr.Blocks() as demo:
    gr.Markdown("## 合欢宗 · 多角恋八卦模拟（10 Agents, Day-based）")

    with gr.Row():
        seed = gr.Number(value=42, label="Seed")
        btn_reset = gr.Button("Reset World")
    init_out = gr.Textbox(label="Init")
    snap_box = gr.Code(label="World Snapshot")
    btn_reset.click(fn=ui_init, inputs=seed, outputs=[init_out, snap_box])

    with gr.Row():
        btn_next = gr.Button("Next Day")
        auto_days = gr.Number(value=3, label="AFK days (1-30)")
        btn_auto = gr.Button("Auto-Run")
    narr = gr.Textbox(label="Narration", lines=6)
    btn_next.click(fn=ui_next_day, inputs=None, outputs=[narr, snap_box])
    btn_auto.click(fn=ui_auto, inputs=auto_days, outputs=[narr, snap_box])

    gr.Markdown("### God Mode")
    act = gr.Dropdown(["peach_blossom_trial","inner_demon","seclusion","grand_banquet","rumor_storm","grant_fate"], value="peach_blossom_trial")
    tgt = gr.Textbox(value="A1", label="Target (可空)")
    val = gr.Number(value=0, label="Value")
    btn_god = gr.Button("Cast")
    god_out = gr.Textbox(label="God Result")
    btn_god.click(fn=ui_god, inputs=[act, tgt, val], outputs=[god_out, snap_box])

    gr.Markdown("### Info System")
    btn_pub = gr.Button("View Public Board")
    pub_box = gr.Code(label="Public Board (recent)")
    btn_pub.click(fn=ui_public, inputs=None, outputs=pub_box)

    mem_agent = gr.Dropdown([f"A{i+1}" for i in range(10)], value="A1", label="Agent")
    btn_mem = gr.Button("View Agent Memory")
    mem_box = gr.Code(label="Private Memory (Agent)")
    btn_mem.click(fn=ui_memory, inputs=mem_agent, outputs=mem_box)

    gr.Markdown("### Inform (tell someone a fact_key)")
    spk = gr.Textbox(value="A1", label="Speaker")
    lst = gr.Textbox(value="A2", label="Listener")
    fk  = gr.Textbox(value="gift_A1_A2_d1", label="fact_key")
    btn_inf = gr.Button("Inform")
    inf_res = gr.Textbox(label="Inform Result")
    btn_inf.click(fn=ui_inform, inputs=[spk,lst,fk], outputs=inf_res)

if __name__ == "__main__":
    demo.launch()