Spaces:
Runtime error
Runtime error
| # ---------- 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 | |
| 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 | |
| class Event: | |
| id: str | |
| day: Day | |
| type: str | |
| actors: List[str] | |
| payload: Dict[str, Any] | |
| truth_strength: float = 1.0 | |
| class Observation: | |
| event_id: str | |
| observer: str | |
| day: Day | |
| mode: str # "direct" | "overheard" | "told" | |
| noise: float | |
| cred: float | |
| 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" | |
| 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() |