DemoGame / app.py
RachelNongyingLI's picture
Update app.py
9e5dfbf verified
# ---------- 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-E-1B-Instruct", # ✅ 正确的 id
device=0
)
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()
import torch
from transformers import pipeline
import time
# 检查 GPU 是否可用
device = 0 if torch.cuda.is_available() else -1
print(f"Device set to {'GPU' if device == 0 else 'CPU'}")
# 加载模型并显示进度
print("Loading model...")
start_time = time.time()
generator = pipeline(
"text-generation",
model="tiiuae/Falcon-E-1B-Instruct",
device=device
)
print(f"Model loaded in {time.time() - start_time:.2f} seconds!")
# 显示是否在使用 GPU
if torch.cuda.is_available():
print(f"Using GPU: {torch.cuda.get_device_name(0)}")
else:
print("Using CPU")
# 测试生成的输出,确保可以推理
print("Testing model inference...")
test_output = generator("Please generate a story.")
print(f"Generated text: {test_output[0]['generated_text']}")
# ---------- 2) Data Models ----------
Day = int
@dataclass
class Agent:
name: str
role: str
relations: Dict[str, Dict[str, float]] = field(default_factory=dict) # 存储与其他角色的关系
memory: Dict[str, Any] = field(default_factory=dict)
belief_about_others: Dict[str, Dict[str, Any]] = field(default_factory=dict)
full_name: str = ""
gender: str = ""
height: int = 170
appearance: str = ""
spirit_root: str = ""
personality: str = "balanced"
is_darkened: bool = False
cultivation: int = 50
willpower: int = 50
face: int = 0
scandal: int = 0
####基础信息生成
ROLES = ["炼气期", "筑基期", "金丹期", "元婴期", "化神期", "合体期", "大乘期"]
#name
SURNAMES = [
"李", "赵", "王", "欧阳", "慕容", "上官", "东方", "南宫", "司马", "诸葛",
"龙", "风", "云", "凤", "陆", "沈", "陆", "白", "周", "秦",
"冯", "段", "朱", "唐", "陆", "花", "凌", "慕", "方", "柏",
"纪", "萧", "沈", "简", "颜", "祁", "孟", "庞", "苏", "施",
"夏", "石", "钟", "梁", "丁", "阮", "黄", "唐", "陈", "范",
"萧", "韦", "何", "冯", "穆", "丘", "薛", "杜", "任", "俞"
]
NAME_PARTS = [
"清", "灵", "雪", "尘", "梦", "羽", "轩", "霖", "珏", "婉",
"澈", "寒", "月", "落", "竹", "珊", "风", "云", "雨", "山", "水",
"青", "霄", "渊", "瑶", "萧", "璇", "月", "辰", "遥", "瑾",
"璃", "星", "凝", "寒", "竹", "飞", "悠", "萧", "星", "琴",
"瑶", "晨", "逸", "岚", "沧", "洛", "晨", "凌", "宸", "熙",
"鸣", "尧", "洛", "彭", "遥", "兰", "琪", "霖", "思", "尘",
"璇", "涵", "泽", "涵", "彬", "雅", "泽", "宇", "琪",
# 新增的2个字名字部分
"紫霄", "白霜", "千秋", "云霁", "玉瑶", "青兰", "月华", "星辰",
"秋水", "长风", "天瑶", "暮雪", "寒月", "紫烟", "白月", "流云",
"醉影", "逍遥", "凌风", "清逸", "皓月", "星云", "红叶", "碧空",
"天雪", "玄霄", "冰心", "凝烟", "云笙", "水月", "风轻", "夜星",
"羽瑶", "千尘", "翠烟", "紫电", "空灵", "雪影", "云起", "凌云",
"独孤", "风华", "清幽", "白兰", "青冥", "羽寒", "天风", "冰澜"
]
def random_name():
surname = random.choice(SURNAMES)
name = "".join(random.choices(NAME_PARTS, k=random.choice([1,2])))
return surname + name
#长相
APPEARANCES = ["清秀","俊逸","冷艳","稚气","邪魅","清冷","英气","娇美","秀雅","潇洒"]
#灵根
SPIRIT_ROOTS = ["金灵根","木灵根","水灵根","火灵根","土灵根","冰灵根","风灵根","雷灵根","混沌灵根"]
def random_gender():
return random.choice(["male","female"])
def random_height(gender):
return random.randint(168,200) if gender=="male" else random.randint(155,175)
def random_appearance():
return random.choice(APPEARANCES)
def random_spirit_root():
return random.choice(SPIRIT_ROOTS)
#性格
PERSONALITY_POOL = [
"aggressive", "scheming", "romantic", "loyal", "jealous", "balanced"
]
DARK_MAP = {
"romantic": "obsessed",
"loyal": "fanatical",
"scheming": "ruthless",
"aggressive": "bloodthirsty",
"jealous": "vengeful",
"balanced": "twisted"
}
def maybe_darkening(agent: Agent):
if not agent.is_darkened:
if agent.jealousy > 70 or agent.scandal > 60 or agent.willpower < 30:
agent.personality = DARK_MAP.get(agent.personality, "twisted")
agent.is_darkened = True
return f"{agent.full_name} 心境失守,黑化成 {agent.personality}!"
return None
@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 ----------
def init_world(seed=42) -> World:
random.seed(seed)
ags = []
for i in range(3):
role = ROLES[i]
gender = random_gender()
ags.append(Agent(
name=f"A{i+1}",
role=role,
full_name=random_name(),
gender=gender,
height=random_height(gender),
appearance=random_appearance(),
spirit_root=random_spirit_root(),
personality=random.choice(PERSONALITY_POOL)
))
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 update_relations(world: World, actor: str, target: str, event_type: str):
"""
更新角色之间的关系,保存好感度、信任度、敌对度和嫉妒心。
"""
# 获取角色的关系字典
actor_relations = world.relations.get(actor, {})
target_relations = world.relations.get(target, {})
# 初始化关系数据(如果不存在)
if target not in actor_relations:
actor_relations[target] = {"liking": 0, "trust": 0, "rivalry": 0, "jealousy": 0}
if actor not in target_relations:
target_relations[actor] = {"liking": 0, "trust": 0, "rivalry": 0, "jealousy": 0}
# 只保存更新后的关系(好感度、嫉妒心等)
world.relations[actor] = actor_relations
world.relations[target] = target_relations
def act_gift(world: World, giver: str, receiver: str):
"""
角色 A 赠送礼物给角色 B,增加好感度并可能引发嫉妒。
"""
giver_agent = next(agent for agent in world.agents if agent.name == giver)
receiver_agent = next(agent for agent in world.agents if agent.name == receiver)
# 增加好感度
giver_agent.relations[receiver]["liking"] += 10
receiver_agent.relations[giver]["liking"] += 10
# 检查是否有第三方嫉妒
other_agents = [agent for agent in world.agents if agent.name != giver and agent.name != receiver]
third_party = max(other_agents, key=lambda x: x.relations[giver]["liking"] if x.relations[giver]["liking"] > x.relations[receiver]["liking"] else x.relations[receiver]["liking"])
# 增加嫉妒心
third_party.relations[giver]["jealousy"] += 10
third_party.relations[receiver]["jealousy"] += 10
# 发布流言
if random.random() < 0.25: # 有一定概率发布流言
publish_public(world, f"gift_{giver}_{receiver}", {"hint": "有人在藏书阁递了东西"}, world.day, 0.55, "rumor")
# 更新关系
update_relations(world, giver, receiver, "gift")
update_relations(world, receiver, third_party.name, "gift")
# 返回事件描述
return f"{giver} 赠送了礼物给 {receiver},增进了好感度,第三方 {third_party.name} 产生了嫉妒。"
def act_confide(world: World, actor: str, target: str):
"""
角色 A 向角色 B 倾诉心事,可能会触发三角恋情节
- 如果角色 A 对 B 表白或倾诉时,可能涉及第三方 C 引发嫉妒等情感纠葛。
"""
actor_agent = next(agent for agent in world.agents if agent.name == actor)
target_agent = next(agent for agent in world.agents if agent.name == target)
# 判断是否涉及三角恋情节,随机选择第三方角色 C
other_agents = [agent for agent in world.agents if agent.name != actor and agent.name != target]
third_party = max(other_agents, key=lambda x: x.relations[actor]["liking"] if x.relations[actor]["liking"] > x.relations[target]["liking"] else x.relations[target]["liking"]) # 第三者选择对两者好感度较高的
# 获取被表白者对表白者的好感度
liking_score = target_agent.relations[actor]["liking"] # B对A的好感度
# 表白是否成功,成功几率与好感度相关
is_successful = random.random() < (liking_score / 100) # 好感度越高,成功概率越大
if is_successful:
# A向B表白并成功,第三方嫉妒
event_description = f"{actor}{target} 表白成功,二人产生了深厚的情感,而 {third_party.name} 看到后感到嫉妒!"
# 更新关系中的嫉妒心
actor_agent.relations[target]["jealousy"] += 10 # A因得到回应产生更多的情感波动
target_agent.relations[actor]["jealousy"] += 5 # B因接受表白产生情感波动
third_party_agent = third_party # C看到表白成功,产生嫉妒
third_party_agent.relations[actor]["jealousy"] += 20 # C因为插足产生嫉妒
third_party_agent.relations[target]["jealousy"] += 20 # C因为插足产生嫉妒
# 发布流言
if random.random() < 0.25: # 有一定概率发布流言
publish_public(world, f"triangle_love_{actor}_{target}", {"hint": "有人偷偷在藏书阁表白"}, world.day, 0.55, "rumor")
# 更新关系
update_relations(world, actor, target, "confide")
update_relations(world, target, third_party_agent.name, "confide")
# 更新角色的记忆
memory_key = f"{actor}_confides_in_{target}_success"
upsert_memory(actor_agent, memory_key, {"event": event_description, "target": target, "third_party": third_party.name})
upsert_memory(target_agent, memory_key, {"event": event_description, "actor": actor, "third_party": third_party.name})
upsert_memory(third_party_agent, memory_key, {"event": event_description, "actor": actor, "target": target})
return event_description # 返回事件描述
else:
# A向B表白失败,第三方窃喜
event_description = f"{actor}{target} 表白失败,{target} 拒绝了 {actor},而 {third_party.name} 看到后感到窃喜!"
# 更新关系中的嫉妒心
actor_agent.relations[target]["jealousy"] += 20 # A因未被接受产生嫉妒
target_agent.relations[actor]["jealousy"] += 5 # B因拒绝A表白产生负面情绪
third_party_agent = third_party # C看到表白失败,窃喜
third_party_agent.relations[actor]["jealousy"] -= 5 # C因表白失败感到窃喜,嫉妒心降低
third_party_agent.relations[target]["jealousy"] -= 5 # C因表白失败感到窃喜,嫉妒心降低
# 发布流言
if random.random() < 0.25: # 有一定概率发布流言
publish_public(world, f"triangle_love_{actor}_{target}_failure", {"hint": "有人偷偷在藏书阁被拒绝了"}, world.day, 0.55, "rumor")
# 更新关系
update_relations(world, actor, target, "confide")
update_relations(world, target, third_party_agent.name, "confide")
# 更新角色的记忆
memory_key = f"{actor}_confides_in_{target}_failure"
upsert_memory(actor_agent, memory_key, {"event": event_description, "target": target, "third_party": third_party.name})
upsert_memory(target_agent, memory_key, {"event": event_description, "actor": actor, "third_party": third_party.name})
upsert_memory(third_party_agent, memory_key, {"event": event_description, "actor": actor, "target": target})
return event_description # 返回事件描述
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)
def act_rumor(world:World, speaker:str, target:str):
ev = make_event(world.day, "rumor", [speaker], {"about": target})
world.events[ev.id] = ev
# 公共板:低 cred 的流言
fk = f"rumor_{target}_d{world.day}"
pub = publish_public(world, fk, {"type":"rumor","about":target}, world.day, 0.55, f"rumor_by_{speaker}")
# 所有人都“看”到——但别全信
for ag in world.agents:
conf = 0.35 if ag.name!=speaker else 0.5
upsert_memory(ag, fk, pub.content, world.day, {"src":"public_board","cred":conf}, conf)
def act_oath(world:World, a:str, b:str):
ev = make_event(world.day, "oath", [a,b], {"form":"同心誓"})
world.events[ev.id] = ev
# 双方记忆加强
for nm in [a,b]:
ag = next(x for x in world.agents if x.name==nm)
fk = f"oath_{a}_{b}_d{world.day}"
upsert_memory(ag, fk, {"type":"oath","pair":[a,b]}, world.day, {"src":"oath","cred":0.95}, 0.95)
# 旁观者吃瓜
for o in sample_observers(world, ev, 0.15):
ag = next(x for x in world.agents if x.name==o.observer)
fk = f"oath_{a}_{b}_d{world.day}"
conf = max(0.2, o.cred*(1-o.noise))
upsert_memory(ag, fk, {"type":"oath_hint","pair":[a,b]}, world.day, {"src":o.mode,"cred":conf}, conf)
# 公共板适度曝光
if random.random()<0.2:
publish_public(world, f"oath_{a}_{b}", {"type":"oath_hint","pair":[a,b]}, world.day, 0.6, "witness")
def act_mediate(world:World, mediator:str, x:str, y:str):
ev = make_event(world.day, "mediate", [mediator,x,y], {})
world.events[ev.id] = ev
# 降低全局 rumor 小幅&双方 private 记忆
world.rumor_level = max(0, world.rumor_level-2)
for t in [x,y]:
ag = next(a for a in world.agents if a.name==t)
fk = f"mediate_{x}_{y}_d{world.day}"
upsert_memory(ag, fk, {"type":"mediate","pair":[x,y],"by":mediator}, world.day, {"src":"mediate","cred":0.8}, 0.8)
def act_sabotage(world:World, actor:str, target:str):
ev = make_event(world.day, "sabotage", [actor,target], {})
world.events[ev.id] = ev
# 目标面子下降;若被抓包,actor 记一笔
tgt = next(a for a in world.agents if a.name==target); tgt.face -= 5
if random.random()<0.3: # 抓包
publish_public(world, f"sabotage_{actor}_{target}", {"type":"sabotage","actor":actor,"target":target}, world.day, 0.7, "caught")
# 旁观者零星观测
for o in sample_observers(world, ev, 0.12):
ag = next(a for a in world.agents if a.name==o.observer)
fk = f"sabotage_{actor}_{target}_d{world.day}"
conf = max(0.2, o.cred*(1-o.noise))
upsert_memory(ag, fk, {"type":"sabotage_hint","actor":actor,"target":target}, world.day, {"src":o.mode,"cred":conf}, conf)
def act_spar(world:World, a:str, b:str):
ev = make_event(world.day, "spar", [a,b], {})
world.events[ev.id] = ev
# 简化胜负
A = next(x for x in world.agents if x.name==a)
B = next(x for x in world.agents if x.name==b)
if A.cultivation + random.randint(-10,10) >= B.cultivation + random.randint(-10,10):
A.cultivation += 4; B.face -= 2
result = "A_win"
else:
B.cultivation += 4; A.face -= 2
result = "B_win"
# 当事人与旁观者记忆
for nm in [a,b]:
ag = next(x for x in world.agents if x.name==nm)
upsert_memory(ag, f"spar_{a}_{b}_d{world.day}", {"type":"spar","result":result,"pair":[a,b]}, world.day, {"src":"spar","cred":0.9}, 0.9)
for o in sample_observers(world, ev, 0.2):
ag = next(x for x in world.agents if x.name==o.observer)
conf = max(0.2, o.cred*(1-o.noise))
upsert_memory(ag, f"spar_{a}_{b}_d{world.day}", {"type":"spar_hint","result":result,"pair":[a,b]}, world.day, {"src":o.mode,"cred":conf}, conf)
# ===== Weighted Policy helpers =====
# 自动感知当前代码里实现了哪些 act_* 动作(防止选到还没写的)
def _discover_supported_actions():
return {name.split("act_")[1] for name, fn in globals().items()
if name.startswith("act_") and callable(fn)}
ACTIONS_SUPPORTED = _discover_supported_actions()
def choose_target_by_memory(world: World, actor: Agent, action: str) -> str:
"""优先挑与我最近记忆有关的人;否则随机一个别人。"""
names = {a.name for a in world.agents if a.name != actor.name}
# 从记忆里抽可能的目标
candidates = []
for it in actor.memory.values():
c = it.content or {}
# 可能出现的人名字段
for key in ("target", "giver", "receiver"):
v = c.get(key)
if isinstance(v, str) and v in names:
candidates.append(v)
if isinstance(c.get("pair"), list):
for v in c["pair"]:
if v in names:
candidates.append(v)
# 若有候选就从中抽,否则随便选一个
return random.choice(candidates) if candidates else random.choice(list(names))
# ---------- 6) Policy
def agent_policy(world: World, actor: Agent):
# 1) 基础权重:所有动作都有 1 的基数
base_weights = {
"gift": 1, "confide": 1, "oath": 1,
"rumor": 1, "expose": 1, "sabotage": 1,
"mediate": 1, "spar": 1
}
# 2) Personality 提供“倾向”,但不是硬性
if getattr(actor, "personality", None) == "aggressive":
base_weights.update({"spar": 3, "expose": 2, "sabotage": 2})
elif actor.personality == "scheming":
base_weights.update({"rumor": 2, "confide": 2, "sabotage": 2})
elif actor.personality == "romantic":
base_weights.update({"gift": 3, "confide": 2, "oath": 2})
elif actor.personality == "loyal":
base_weights.update({"mediate": 2, "gift": 2, "oath": 2})
elif actor.personality == "jealous":
base_weights.update({"expose": 2, "rumor": 2, "sabotage": 2})
# 黑化后的偏好(若你已经实现 is_darkened)
if getattr(actor, "is_darkened", False):
base_weights["sabotage"] += 1
base_weights["expose"] += 1
# 3) 信息修正:根据“我知道的事”微调权重
for fk, item in actor.memory.items():
typ = (item.content or {}).get("type", "")
# 看到/听到别人互送礼 → 更容易吃醋/造谣/搞破坏
if "gift" in fk or typ in ("gift", "gift_hint"):
base_weights["rumor"] += 1
base_weights["sabotage"] += 1
# 看到/听到推心置腹 → 更可能揭发
if "confide" in fk or typ in ("confide", "confide_hint"):
base_weights["expose"] += 1
# 自己曾被曝光 → 倾向 spar/expose
if typ == "expose" and item.content.get("target") == actor.name:
base_weights["spar"] += 2
base_weights["expose"] += 1
# 高可信流言 → 想去调停
if "rumor" in fk and item.confidence > 0.6:
base_weights["mediate"] += 1
# 4) 只保留当前代码里“确实已实现”的动作,防止选到未实现的
weights = {a: w for a, w in base_weights.items() if a in ACTIONS_SUPPORTED and w > 0}
if not weights:
# 兜底:至少保证 gift/confide 存在其一
for a in ("gift", "confide", "expose"):
if a in ACTIONS_SUPPORTED:
weights[a] = 1
if not weights:
return ("gift", random.choice([x.name for x in world.agents if x.name != actor.name]))
# 5) 按权重随机一个动作
actions, probs = zip(*weights.items())
action = random.choices(actions, weights=probs, k=1)[0]
# 6) 选目标:优先与我记忆相关的人;否则随机一个
target = choose_target_by_memory(world, actor, action)
return action, target
def apply_action(world: World, actor: str, action: str, target: str):
if action == "gift":
act_gift(world, actor, target)
update_relations(world, actor, target, "gift") # 更新关系
return f"{actor}{target} 赠礼"
if action == "confide":
act_confide(world, actor, target)
update_relations(world, actor, target, "confide") # 更新关系
return f"{actor}{target} 推心置腹"
if action == "spar":
act_spar(world, actor, target)
update_relations(world, actor, target, "spar") # 更新关系
return f"{actor} ⚔️ {target} 比试"
if action == "expose":
act_expose(world, actor, target, "残页证据")
update_relations(world, actor, target, "expose") # 更新关系
return f"{actor} 公揭 {target}"
# TODO: 其他动作:rumor/oath/sabotage/mediate...
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 = (
"注意:部分人物因掌握某些信息而做出不同寻常的选择。\n"
f"DAY={world.day}\n"
f"WORLD_SNAPSHOT={snap}\n"
f"KEY_EVENTS={daily_events}\n"
"请写今日纪要(2-4句):1) 核心八卦;2) 关系走向;3) 明日伏笔(可选)。"
)
narration = narrate(prompt)
# 改进:让事件更加直观
formatted_events = []
for event in daily_events:
if event.type == "expose":
formatted_events.append(f"🗣️ <b>{event.actors[0]} 揭发了 {event.actors[1]}</b>!这是关于 {event.payload.get('evidence', '残页证据')} 的丑闻。")
elif event.type == "gift":
formatted_events.append(f"🎁 <b>{event.actors[0]} 送给 {event.actors[1]} 一份礼物</b>,暗示着他们之间的关系发生了微妙变化。")
elif event.type == "spar":
formatted_events.append(f"⚔️ <b>{event.actors[0]}{event.actors[1]} 进行了比试</b>,并且 {event.actors[0]} 获胜!")
elif event.type == "confide":
# For confide events, include success or failure of the confession
result = "心意已达" if event.payload.get('result') == 'success' else "表白未果"
formatted_events.append(f"🗣️ <b>{event.actors[0]}{event.actors[1]}</b> 倾诉心事,{result},情感深沉。")
elif event.type == "rumor":
# If the event is a rumor, format the details accordingly
formatted_events.append(f"💬 <b>{event.actors[0]} 传出了关于 {event.payload.get('about', '某人')}</b> 的流言。")
elif event.type == "oath":
formatted_events.append(f"🤝 <b>{event.actors[0]} 向 <b>{event.actors[1]}</b> 立下誓言</b>,誓言让他们的关系更深。")
elif event.type == "sabotage":
formatted_events.append(f"💥 <b>{event.actors[0]} 对 <b>{event.actors[1]}</b> 进行了破坏</b>,引发了一场风波。")
elif event.type == "mediate":
formatted_events.append(f"🕊️ <b>{event.actors[0]} 调解了 <b>{event.actors[1]}{event.actors[2]}</b> 之间的争执</b>,关系暂时恢复平静。")
else:
formatted_events.append(f"🔍 发生了未明确的事件:<b>{event.type}</b>,参与者:<b>{', '.join(event.actors)}</b>")
narration += "\n\n" + "\n".join(formatted_events)
# 黑化事件
dark_events = []
for a in world.agents:
d = maybe_darkening(a)
if d:
dark_events.append(d)
if dark_events:
narration += "\n\n⚠️ 黑化事件:\n" + "\n".join(dark_events)
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) Pretty Gradio UI (Dashboard) ----------
import pandas as pd
import gradio as gr
def world_summary_html(world: World) -> str:
return f"""
<div style="display:flex; gap:16px; flex-wrap:wrap; font-size:14px">
<div style="padding:10px 14px; border-radius:12px; background:#f0f4ff">📅 <b>Day</b>: {world.day}</div>
<div style="padding:10px 14px; border-radius:12px; background:#fff7e6">💬 <b>Rumor</b>: {world.rumor_level}</div>
<div style="padding:10px 14px; border-radius:12px; background:#eafaf1">🎎 <b>Season</b>: {world.season}</div>
</div>
"""
def agents_dataframe(world: World) -> pd.DataFrame:
rows = []
for a in world.agents:
# 小表情让 personality 更直观
pmap = {
"romantic": "romantic ❤️",
"scheming": "scheming 🐍",
"jealous": "jealous 😡",
"loyal": "loyal 🤝",
"aggressive": "aggressive ⚔️",
"balanced": "balanced ⚖️",
"obsessed": "obsessed 💔",
"fanatical": "fanatical 🔥",
"ruthless": "ruthless 🩸",
"bloodthirsty": "bloodthirsty ☠️",
"vengeful": "vengeful 👁️",
"twisted": "twisted 🕷️"
}
pers = pmap.get(a.personality, a.personality)
rows.append({
"ID": a.name,
"姓名": a.full_name,
"修为": a.role,
"灵根": a.spirit_root,
"性格": pers,
"黑化?": "✔ 黑化" if getattr(a, "is_darkened", False) else "❌",
"修为": a.cultivation,
"面子": a.face,
"嫉妒": a.jealousy,
"丑闻": a.scandal,
})
return pd.DataFrame(rows)
def public_board_pretty(world: World) -> str:
if not world.public_board:
return "(暂无公共八卦)"
items = world.public_board[-15:]
lines = []
for x in items:
c = x.content
event_type = c.get("type", "")
# 根据事件类型给出人类可读的描述
if event_type == "gift":
lines.append(f"🎁 <b>{x.actors[0]}</b> 送给 <b>{x.actors[1]}</b> 一份礼物,关系更加微妙。")
elif event_type == "oath":
lines.append(f"🤝 <b>{x.actors[0]}</b> 向 <b>{x.actors[1]}</b> 发誓,誓言让他们的关系更加牢固。")
elif event_type == "expose":
lines.append(f"⚠️ <b>{x.actors[0]}</b> 揭发了 <b>{x.actors[1]}</b> 的丑闻!证据:{c.get('evidence', '残页证据')}")
elif event_type == "confide":
# Check if the confession was successful or failed and express it in classical style
result = "心意已达" if c.get('result') == 'success' else "表白未果"
lines.append(f"🗣️ <b>{x.actors[0]}</b> 向 <b>{x.actors[1]}</b> 倾诉心事,{result},情感深沉。")
else:
lines.append(f"🔍 发生了未明确的事件:<b>{event_type}</b>,涉及人物:<b>{', '.join(x.actors)}</b>")
return "<br>".join(lines)
def memory_pretty(agent: Agent) -> str:
if not agent.memory:
return "(该人物没有私人记忆)"
items = sorted(agent.memory.items(), key=lambda kv: -kv[1].last_update_day)
lines = []
for k, v in items[:30]: # 只显示最新的 30 条
content = v.content
event_type = content.get("type", "")
# 对于每种事件类型,使用更易懂的描述
if event_type == "gift":
lines.append(f"🎁 <b>{content.get('giver')}</b> 送给 <b>{content.get('receiver')}</b> 一份礼物,增进了他们的关系。")
elif event_type == "confide":
lines.append(f"🗣️ <b>{content.get('pair')[0]}</b> 和 <b>{content.get('pair')[1]}</b> 共享了心事,彼此更加信任。")
elif event_type == "oath":
lines.append(f"🤝 <b>{content.get('giver')}</b> 向 <b>{content.get('receiver')}</b> 立下誓言,誓言让他们的关系更深。")
elif event_type == "expose":
lines.append(f"⚠️ <b>{content.get('target')}</b> 被揭发,证据:{content.get('evidence', '残页证据')}")
elif event_type == "rumor":
lines.append(f"💬 <b>{content.get('source')}</b> 传出了关于 <b>{content.get('target')}</b> 的流言。")
else:
lines.append(f"🔍 发生了未明确的事件:<b>{event_type}</b>,参与者:<b>{', '.join(content.get('pair', []))}</b>")
return "<br>".join(lines)
def ui_refresh_dashboard():
html = world_summary_html(WORLD)
df = agents_dataframe(WORLD)
snap = json.dumps(WORLD.snapshot(), ensure_ascii=False, indent=2)
return html, df, snap
def ui_next_day_pretty():
narration = tick_one_day(WORLD)
html, df, snap = ui_refresh_dashboard()
card = f"### 今日纪要\n\n{narration}"
return card, html, df, snap
def ui_auto_pretty(days:int):
days = max(1, min(30, int(days)))
logs = []
for _ in range(days):
logs.append(tick_one_day(WORLD))
html, df, snap = ui_refresh_dashboard()
return "\n\n---\n\n".join(logs), html, df, snap
def ui_public_pretty():
return public_board_pretty(WORLD)
def ui_memory_pretty(agent_name):
ag = next((a for a in WORLD.agents if a.name==agent_name), None)
return memory_pretty(ag) if ag else "no such agent"
def ui_inform_pretty(speaker, listener, fact_key):
return inform(WORLD, speaker.strip(), listener.strip(), fact_key.strip())
theme = gr.themes.Soft(primary_hue="violet", neutral_hue="slate")
with gr.Blocks(theme=theme, analytics_enabled=False) as demo:
gr.Markdown("## 🌸 某某宗 · 多角恋八卦模拟(10 Agents, Day-based)")
with gr.Tab("Dashboard"):
sum_html = gr.HTML(world_summary_html(WORLD))
agents_table = gr.Dataframe(
value=agents_dataframe(WORLD),
interactive=False, wrap=True, label="Agents Overview"
)
snap_box = gr.Code(value=json.dumps(WORLD.snapshot(), ensure_ascii=False, indent=2), label="World Snapshot (JSON)", language="json")
btn_refresh = gr.Button("🔄 Refresh")
btn_refresh.click(fn=ui_refresh_dashboard, inputs=None, outputs=[sum_html, agents_table, snap_box])
with gr.Tab("Run"):
with gr.Row():
btn_next = gr.Button("▶️ Next Day")
auto_days = gr.Slider(1, 30, value=3, step=1, label="AFK days")
btn_auto = gr.Button("⏩ Auto-Run")
narr = gr.Markdown("(点击 Next Day 生成今日纪要)")
btn_next.click(fn=ui_next_day_pretty, inputs=None, outputs=[narr, sum_html, agents_table, snap_box])
btn_auto.click(fn=ui_auto_pretty, inputs=auto_days, outputs=[narr, sum_html, agents_table, snap_box])
with gr.Tab("God Mode"):
gr.Markdown("对世界施法(慎用):")
with gr.Row():
act = gr.Dropdown(
["peach_blossom_trial","inner_demon","seclusion","grand_banquet","rumor_storm","grant_fate"],
value="peach_blossom_trial", label="Action"
)
tgt = gr.Textbox(value="A1", label="Target (可空)")
val = gr.Number(value=0, label="Value")
god_out = gr.Textbox(label="Result")
btn_god = gr.Button("⚡ Cast")
btn_god.click(fn=lambda a,b,c: (god_action(WORLD, a, b.strip(), int(c) if c else 0)),
inputs=[act, tgt, val], outputs=god_out).then(
fn=ui_refresh_dashboard, inputs=None, outputs=[sum_html, agents_table, snap_box]
)
with gr.Tab("Intel"):
with gr.Row():
btn_pub = gr.Button("📣 Public Board")
mem_agent = gr.Dropdown([f"A{i+1}" for i in range(10)], value="A1", label="Agent")
btn_mem = gr.Button("🧠 View Private Memory")
pub_box = gr.HTML()
mem_box = gr.HTML()
btn_pub.click(fn=ui_public_pretty, inputs=None, outputs=pub_box)
btn_mem.click(fn=ui_memory_pretty, inputs=mem_agent, outputs=mem_box)
gr.Markdown("**Inform(把某条 fact 告诉别人)**")
with gr.Row():
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_pretty, inputs=[spk,lst,fk], outputs=inf_res)
if __name__ == "__main__":
demo.launch()