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-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 | |
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 | |
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 ---------- | |
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() |