Create app.py

app.py

@@ -0,0 +1,370 @@
+# ---------- 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()