"""Cognition:LLM 认知层——agent 真正"思考后行动"(docs/agent-runtime-v2.md §1.5)。 - `decide()` 是**纯函数式**的:输入感知快照(ThinkInput),输出决策(Decision), 不触碰共享世界状态——LLM 阻塞调用因此可以丢进线程池,世界变更由引擎在事件循环里单线程执行; - L2 慢想:全量 prompt(人设+共识+躯体+所见+记忆检索+近期事件)→ 重定意图与下一步动作; - L1 快想:被人搭话时的对话模式,短 prompt 接一句话(流畅 & 省钱); - 输出统一为 JSON,解析失败降级为观望(世界不因单次失败停摆)。 """ from __future__ import annotations import json import re import time from collections import deque from dataclasses import dataclass, field from genesis.obs.logging_setup import get_logger from genesis.runtime.bulletin import dynamic_bulletin from genesis.runtime.soma import Soma logger = get_logger("runtime.cognition") # —— LLM 调用耗时埋点(定位思索长尾:多分钟≠单次5-7s,多半是超时重试/调用堆叠)—— _LATENCY: deque = deque(maxlen=3000) # (elapsed_s, aid, tag, think, in_chars, out_chars) SLOW_CALL_S = 20.0 # 超过即告警落日志(带上下文供定位) def latency_stats() -> dict: """LLM 单次调用耗时统计:n/avg/p50/p90/p99/max + 最慢 8 次(含 agent/tag/think/输入输出字数)。""" data = list(_LATENCY) if not data: return {"n": 0} lats = sorted(e[0] for e in data) n = len(lats) pct = lambda p: lats[min(n - 1, int(p * n))] by_tag: dict = {} for e in data: by_tag.setdefault(e[2], []).append(e[0]) slow = sorted(data, key=lambda e: -e[0])[:8] return { "n": n, "avg": round(sum(lats) / n, 2), "p50": round(pct(0.5), 2), "p90": round(pct(0.9), 2), "p99": round(pct(0.99), 2), "max": round(lats[-1], 2), "by_tag": {t: {"n": len(v), "avg": round(sum(v) / len(v), 2), "max": round(max(v), 2)} for t, v in by_tag.items()}, "slowest": [{"s": round(e[0], 1), "aid": e[1], "tag": e[2], "think": e[3], "in": e[4], "out": e[5]} for e in slow], } FACT_RULE = ("【事实纪律】凡涉及具体事件,只能引用『你确切知道的事』或你的记忆;" "你可以出于策略**撒谎**,但不许无意识地编造事实/人物/物品;不知道就说不知道。") INFERENCE_RULE = ( "【分辨信息】你亲眼所见/亲历的才是事实;别人告诉你的只是『他的说法』——你确知的仅仅是『他这么说了』,而非这话为真。" "先掂量他的动机与可信度、再和你已知的对照,然后决定信几分,别人一开口就全信(凶手尤其爱用半真半假的话带节奏)。") # 软引导:不规定该做什么,只把人立住——让智能从"他是谁 + 此刻处境"里自然涌现 SELF_GUIDE = ( "下面会把此刻你身处的世界、你看到听到的一切、和你心里的盘算如实摆给你。" "请就以**你这个人**的性子、来意与那桩不可告人的秘密,对眼前发生的一切**自然地反应**——" "该追查就追查、该试探就试探、该结盟或独行、该隐藏或撒谎、出了事该惊慌该呼救就惊慌呼救。" "你不是按固定流程走的 NPC,而是个有私心、有情绪、会随机应变的活人;没有谁规定你这一步必须做什么,凭你自己判断。") ACTION_DOC = """你的每次回应都是一个 JSON(只输出 JSON,别的都不要): {"scheme": "一句话盘算:眼下这局面,为了你真正想要的,你打算怎么做(权衡处境、在场的人、风险与机会)", "thought": "内心想法(简短)", "intent": "当前意图(一句话)", "speech": {"to": "对方名字,或空字符串=自言自语/对在场所有人", "text": "口语台词,别夹动作描写"} 或 null, "gesture": "神态或小动作(可空;不是台词,别放进 speech)", "action": {"kind": "动作", "arg": "参数"} 或 null, "minutes": 动作预计分钟数, "goal_update": {"index": 目标序号, "status": "doing|done|blocked", "note": "一句话"} 或 null, "goals_replan": ["你心里的打算变了就重列接下来想做的2-4件事;没变就给 null"] 或 null} 各字段都尽量短(scheme/thought/intent 各≤25字,speech 一两句)——你说得越短,世界转得越快。 你能做的事(action.kind,按需自取,没人规定你必须用哪个): - move 去某地(arg=地点,会沿路逐屋走过去) · search 仔细搜当前地点(翻出藏着的线索) · take 拿走可见物品(arg=物品名) - destroy 销毁随身物品(arg=物品名) · use 动用现场的设施(arg=你想做的事,具体看现场描述与提示) - examine 查验当前地点的尸体(arg=死者名) · eavesdrop 偷听同处他人的交谈 - assemble 召集全员当面对质 · accuse 对质中指认(arg=名字,留空=弃权) · shout 放声大喊、全堡可闻(arg=内容) - sleep 回自己房间睡 · idle 原地停留观察(arg 留空) 说话和动作可以同时给。你刚答应别人的事,这一拍就用 action 真去兑现(除非你心里清楚那是骗他的缓兵之计)——说一套做一套会被当场看穿。""" KILLER_DOC = ("【你还有一重谁都不知道的身份】你不是莽撞的凶手,而是布下这场局的人——你要把该清算的人一个个了结," "同时全身而退、活到天亮(你为什么这么做、想先动谁,都在你自己的人设与心事里)。" "沉得住气、看准时机、把首尾收拾干净,远比一时痛快重要;旁边只要还有第三个人,就先忍着。\n" "除了大家都有的动作,你独有两个:\n" "- lure:把同一地点的某人用一句**正中他私欲**的话(『我知道你要的那样东西就在某处』)支去僻静角落单独相处,arg=名字。" "哪怕你俩正夹在一堆人里,也能凑到他耳边低声下饵、把他一个人勾走;这句话还会在他心里种下一个日后仍勾着他往那儿去的念头。" "通常的次序:他在你身边时先 lure → 跟过去会合 → 只剩你俩时再下手。\n" "- kill:对同一地点的目标下杀手,arg=名字。**只在四下无人、只剩你和他、且在僻静处时动手**——有旁人在场就是当场暴露。" "停电、风雨都是天赐的掩护。") DIALOGUE_DOC = """你正在交谈。输出一个 JSON(只输出 JSON): {"thought": "内心想法(简短)", "speech": "你接的话(口语一两句,别夹动作描写)", "gesture": "神态或小动作(可空)", "leave": false} 就像你这个人那样接话——可以套话、结盟、试探、敷衍、安抚或撒谎,也可以问你真想知道的。 聊够了、谈崩了、或心里另有更要紧的事,就把 leave 设 true(道一句别再走),去做你想做的。 对方说的只是『他的说法』、未必当真;答应了又打算做到的事,离开后就真去兑现,别空口许诺。""" @dataclass class ThinkInput: """一次思考的感知快照(由引擎在事件循环线程组装,decide 在线程池消费)。""" now_h: float time_label: str place: str place_desc: str # 地点描述 + 客观状态快照 occupants: list[str] events: list[str] # 收件箱新事件(已转文字) soma_lines: list[str] candidates: list[str] # L0 候选(饿了→吃饭 等) bulletin: str # 动态公告 memories: list[str] # 检索到的相关记忆 intent: str # 当前意图 exits: list[str] = field(default_factory=list) # 从当前地点可直接去的相邻地点(感知世界+合法移动+鼓励探索) inventory: list[str] = field(default_factory=list) # 你此刻身上带着的东西(含用途)——别忘了自己拿了什么、能拿它做什么 facts: list[str] = field(default_factory=list) # 事实锚:确切知道的近期事实(防编造) hearsay: list[str] = field(default_factory=list) # 传闻锚:别人说的话(说法≠事实,须自行推理掂量) dialogue: list[str] = field(default_factory=list) # 会话共享上下文(若在交谈) dialogue_with: str = "" # 会话其他成员(可多人,顿号分隔) dialogue_rounds: int = 0 # 会话已进行的发言数 dialogue_budget: int = 0 # 会话发言预算(超了必须收尾) arriving: str = "" # 预思考:正赶往何处(到达前预想下一步) private_note: str = "" # 私有压力(凶手的剧本钟点等,只进本人 prompt) lure_bait: str = "" # 被引诱者收到的诱饵提示(命中私欲,只进被诱者 prompt) planted_leads: list[str] = field(default_factory=list) # 被植入的"假任务"持久念头(凶手种,本人当真线索,可识破) stage_directive: str = "" # 阶段聚焦(导演按幕注入:此刻处境/凶手本幕任务,导演层↔认知层接口) @dataclass class Decision: scheme: str = "" # 盘算:手段-目的推理(为达成终极目的、此刻最该做什么),决策前置步 thought: str = "" intent: str = "" speech_to: str = "" speech: str = "" gesture: str = "" action_kind: str = "" action_arg: str = "" minutes: float = 10.0 leave_dialogue: bool = False goal_update: dict | None = None replan: list[str] | None = None # 局势剧变时重排的分目标(动机不变,路径随机应变) raw: str = "" @property def empty(self) -> bool: """完全空的决策(解析失败/模型哑火)→ 引擎按"出神"处理(G12)。""" return not (self.speech or self.action_kind or self.thought or self.gesture) def parse_json_block(text: str) -> dict: """从模型输出中稳健地抠出第一个 JSON 对象。""" m = re.search(r"\{.*\}", text, re.S) if not m: return {} try: return json.loads(m.group(0)) except json.JSONDecodeError: try: # 常见瑕疵:尾逗号 return json.loads(re.sub(r",\s*([}\]])", r"\1", m.group(0))) except json.JSONDecodeError: return {} class CognitiveMind: """一个 agent 的 LLM 心智。与引擎的契约:decide(snapshot) → Decision(纯函数,无副作用)。""" def __init__(self, aid: str, persona_card: str, *, is_killer: bool, home: str, llm, memory, static_ground: str) -> None: self.aid = aid self.persona_card = persona_card self.is_killer = is_killer self.home = home self.llm = llm self.memory = memory self.static_ground = static_ground self.soma = Soma(last_meal_h=19.5, woke_at_h=8.0) self.intent = "被渡船独自留在暴雨围困的孤岛上,心里七上八下" # 心智接管前的初始状态(序幕在放映之前) self.dialogue: deque[str] = deque(maxlen=12) # 对话窗口 self.dialogue_with = "" self.last_speech = "" self.last_thought = "" self.goals: list[dict] | None = None # 今晚目标清单(G9/G10):首次思考时自举 self.agenda_seed: dict | None = None # 引擎注入的私罪议程({want,place}):自举时落成一条绑定地点的开场目标(#3 探索动力) self.insights: deque[str] = deque(maxlen=5) # 反思结论(G7):直接注入后续决策 self.relations: dict[str, dict] = {} # 关系/信任(G8):{名:{impression,trust}} self.imp_acc = 0.0 # 重要度累积(达阈值触发反思) # —— 目标自举(G9/G10):开局把 secret 落成 2-4 条可执行目标 —— def _bootstrap_goals(self) -> None: system, user = self._compose_bootstrap() goals = self._run_bootstrap(system, user) if not goals: goals = ["弄清今晚到底是谁设的局", "保住自己的秘密不被揭穿"] # #3 探索动力:把私罪议程落成一条**绑定地点**的开场目标,排在最前(确定性锚,不靠 LLM 是否提及)。 # 这样每个角色开局就有"非去某地不可"的理由,地图/道具才被真正用起来。 seeded = self._agenda_goal() rest = [{"desc": g, "status": "todo", "note": ""} for g in goals] self.goals = (seeded + rest)[:4] if seeded else rest def _agenda_goal(self) -> list[dict]: """私罪议程 → 开场目标(若引擎注入了 agenda_seed)。空则返回 [](向前兼容:无 dossier 的世界不受影响)。""" a = self.agenda_seed or {} want, place = (a.get("want") or "").strip(), (a.get("place") or "").strip() if not (want and place): return [] return [{"desc": f"赶在别人翻出我的旧账之前,去{place}{want}", "status": "todo", "note": ""}] def _compose_bootstrap(self) -> tuple[str, str]: # 事实锚(G2):喂入地图常识+在岛名册,掐死"老周/档案室"这类幻觉实体在目标层出生 a = self.agenda_seed or {} agenda = "" if (a.get("want") or "").strip() and (a.get("place") or "").strip(): # 把"你最怕被翻出的旧账物证在哪"喂给自举,让其余目标围着它长(开场锚由 _agenda_goal 确定性补上) agenda = (f"\n你心底最怕被人翻出的那桩旧账,死穴物证是【{a['want']}】,就在【{a['place']}】——" "今晚你最该抢在别人之前去处理掉它;其余目标围绕『保住这个秘密、查清是谁设的局、活到天亮』来定。") return ("\n\n".join([f"你是{self.aid}。{self.persona_card}", self.static_ground, FACT_RULE]), "今夜你被困在这座与世隔绝的孤岛上(暴风雨封海,要熬到天亮/救援)。根据你的秘密与处境,列出你**今晚必须办成的私人目标**," "2-4条,具体可执行(写明去哪/做什么),按优先级排序。" + agenda + "\n【硬性约束】目标里出现的地点必须取自上面的地图常识、人物必须取自在岛人员名册;" "岛上没有管家、仆人、船夫、档案室等任何额外的人或场所,不存在的东西一个字也不许写。" '输出 JSON:{"goals": ["...", "..."]}') def _run_bootstrap(self, system: str, user: str) -> list[str]: """默认= litellm 直答 + 文本解析(LangGraph 大脑覆写为结构化)。""" d = parse_json_block(self._call(system, user, tag="L2")) return [str(g) for g in (d.get("goals") or [])][:4] def goals_block(self) -> str: if not self.goals: return "" lines = [f"{i+1}. {g['desc']}" + (f"({g['note']})" if g['note'] else "") + (" ✓已了" if g['status'] == "done" else "") for i, g in enumerate(self.goals)] return ("你心里盘算着的几件事(只是此刻的打算,随时能改;真正驱使你的是你那个终极目的):\n" + "\n".join(lines)) # —— 反思(G7):把近期记忆沉淀成判断,并修正对人的信任;长期记忆在此沉淀/召回(G6)—— def reflect(self, time_label: str, longterm=None) -> None: system, user = self._compose_reflect(time_label, longterm) insights, relations = self._run_reflect(system, user) from genesis.memory.record import MemoryKind for ins in insights[:3]: ins = str(ins) self.insights.append(ins) self.memory.observe(f"{time_label},我的判断:{ins}", 0.0, kind=MemoryKind.REFLECTION, importance=0.9) if longterm is not None: longterm.deposit(self.aid, f"{time_label} {ins}", 0.0) # 沉淀长期(慢调用在慢时刻) for r in relations: if not isinstance(r, dict) or not r.get("name"): continue name = str(r["name"]) cur = self.relations.setdefault(name, {"impression": "", "trust": 0.0}) if r.get("impression"): cur["impression"] = str(r["impression"])[:40] try: cur["trust"] = max(-1.0, min(1.0, cur["trust"] + float(r.get("trust", 0)))) except (TypeError, ValueError): pass def _compose_reflect(self, time_label: str, longterm=None) -> tuple[str, str]: recent = self.memory.retrieve("今晚 重要 可疑", 0.0, top_k=12) mems = "\n".join(f"- {m.content}" for m in recent) lt = longterm.recall(self.aid, self.intent + " " + mems[:120]) if longterm else [] lt_block = ("\n你长期记忆里浮现的旧判断:\n" + "\n".join(f"- {x}" for x in lt)) if lt else "" return (f"你是{self.aid}。{self.persona_card}", f"现在是{time_label}。回顾你今晚的经历,沉淀出判断(简短)。\n你的记忆:\n{mems}{lt_block}\n" '输出 JSON:{"insights": ["对人或局势的判断,2-3条,每条≤25字"],' ' "relations": [{"name": "某人", "impression": "一句话新印象", "trust": -1.0到1.0之间的增减}]}') def _run_reflect(self, system: str, user: str) -> tuple[list[str], list[dict]]: """默认= litellm 直答 + 文本解析。返回 (insights, relations dict 列表)。""" # 不开 thinking 模式(分钟级长尾主因);限输出,反思也要快 d = parse_json_block(self._call(system, user, tag="reflect", max_tokens=320)) insights = [str(i) for i in (d.get("insights") or [])][:3] relations = [r for r in (d.get("relations") or []) if isinstance(r, dict) and r.get("name")] return insights, relations def relations_block(self, occupants: list[str]) -> str: if not self.relations: return "" lines = [] for raw in occupants: name = raw.split("(")[0] # 在场者可能带"(动作,神色)"装饰 r = self.relations.get(name) if r and (r["impression"] or abs(r["trust"]) > 0.05): lines.append(f"- {name}:{r['impression']}(信任 {r['trust']:+.1f})") return "【你对在场者的看法】\n" + "\n".join(lines) if lines else "" # —— L2 慢想 —— def decide(self, snap: ThinkInput) -> Decision: if self.goals is None: self._bootstrap_goals() # 首次思考先把 secret 落成目标清单(G9/G10) if snap.dialogue_with: system, user = self._compose_dialogue(snap) return self._run_dialogue(system, user, snap) system, user, deep, max_tokens = self._compose_action(snap) return self._run_action(system, user, deep, max_tokens, snap) # —— 推断步骤(可被 LangGraph 大脑覆写;默认= litellm 直答 + 文本解析)—— def _run_action(self, system: str, user: str, deep: bool, max_tokens: int, snap: ThinkInput) -> Decision: out = self._call(system, user, max_tokens=max_tokens) return self._to_decision(out) def _run_dialogue(self, system: str, user: str, snap: ThinkInput) -> Decision: out = self._call(system, user, tag="L1", max_tokens=220) d = parse_json_block(out) return Decision(thought=str(d.get("thought", "")), speech=str(d.get("speech", "")), gesture=str(d.get("gesture", "") or ""), speech_to=snap.dialogue_with, leave_dialogue=bool(d.get("leave", False)), intent=self.intent, raw=out) # —— prompt 组装(与推断分离:旧大脑与 LangGraph 大脑共用同一份提示词)—— def _compose_action(self, snap: ThinkInput) -> tuple[str, str, bool, int]: # 系统层=把"他是谁"立住(人设+世界基础认知+软引导+能做的事);凶手另有隐藏身份 sys_parts = [f"你是{self.aid}。{self.persona_card}", self.static_ground, SELF_GUIDE, FACT_RULE, INFERENCE_RULE, ACTION_DOC] if self.is_killer: sys_parts.append(KILLER_DOC) sys = "\n\n".join(sys_parts) # 用户层=动态状态模板:把"此刻的世界 / 你的所见所闻所感 / 你自己"如实填进去,不下指令,让他自行判断 p = [snap.bulletin, f"【你在·{snap.place}】{snap.place_desc}"] if snap.exits: p.append("从这里可直接去:" + "、".join(snap.exits) + "。") if snap.occupants: rb = self.relations_block(snap.occupants) p.append("【此刻和你在一起的人】" + "、".join(snap.occupants) + ("\n" + rb if rb else "")) else: p.append("【此刻这里只有你一个人】") if snap.events: # 刚发生的事摆在显眼处:先让你感知,再由你自己决定怎么应——不替你规定反应 p.append("【你眼前刚刚发生】\n" + "\n".join(f"- {e}" for e in snap.events[-8:])) if snap.facts: p.append("【你确凿知道的(亲历亲见)】\n" + "\n".join(f"- {f}" for f in snap.facts[-8:])) if snap.hearsay: p.append("【你听人说的(只是说法,未必真)】\n" + "\n".join(f"- {h}" for h in snap.hearsay[-6:])) if snap.memories: # 记忆流(含经历过的对话):让过往沉淀进当下判断 p.append("【你想起】\n" + "\n".join(f"- {m}" for m in snap.memories)) if snap.inventory: p.append("【你身上带着】" + ";".join(snap.inventory)) if snap.soma_lines: p.append("\n".join(snap.soma_lines)) last_sp = getattr(self, "last_speech", "") if last_sp and (snap.now_h * 3600 - getattr(self, "last_speech_t", 0)) < 25 * 60: p.append(f"【你刚说过】「{last_sp}」") inner = [] # 你的内心活动:意图 + 盘算的事 + 沉淀的判断 if snap.intent: inner.append(f"此刻心里:{snap.intent}") gb = self.goals_block() if gb: inner.append(gb) if self.insights: inner.append("你心里的判断:" + ";".join(self.insights)) if inner: p.append("【你的内心】\n" + "\n".join(inner)) if snap.private_note: # 凶手的私有心事(世界回执+复仇压力) p.append(snap.private_note) if snap.planted_leads: p.append("【你心头还搁着几件事(听来的,未必当真)】\n" + "\n".join(f"- {x}" for x in snap.planted_leads)) if snap.lure_bait: # 有人刚把你往某处引(由你判断要不要上钩) p.append(snap.lure_bait) if snap.stage_directive: # 软导演/凶手剧本(软引导,非硬命令) p.append(snap.stage_directive) if snap.candidates: p.append("(身体的小提醒:" + ";".join(snap.candidates) + ")") if snap.arriving: p.append(f"(你正赶往{snap.arriving},快到了——顺带想好到了先做什么。)") p.append(f"现在,就以{self.aid}的身份,对眼前这一切自然地反应。输出 JSON:") deep = "对质" in snap.bulletin and "投票" in snap.bulletin return sys, "\n\n".join(p), deep, (600 if deep else 520) # —— L1 对话快想(兼容入口:组装 + 推断)—— def _decide_dialogue(self, snap: ThinkInput) -> Decision: system, user = self._compose_dialogue(snap) return self._run_dialogue(system, user, snap) # —— L1 对话快想:prompt 组装 —— def _compose_dialogue(self, snap: ThinkInput) -> tuple[str, str]: sys = "\n\n".join([ f"你是{self.aid}。{self.persona_card}", self.static_ground, # 地图+名册进对话(掐死聊出"剪辑室"这类不存在的地点;静态可缓存) f"现在{snap.time_label},你在{snap.place}和{snap.dialogue_with}交谈。" "口语、简短、符合身份与心机。", FACT_RULE, DIALOGUE_DOC, ]) lines = [*(snap.soma_lines or [])] # 对话中也把你心里惦记的事带上(免得一聊就忘了自己要干嘛) pending = [g["desc"] for g in (self.goals or []) if g.get("status") != "done"][:3] if pending: lines.append("(你心里还惦记着:" + ";".join(pending) + "——聊到想聊的,想抽身就 leave=true。)") if snap.lure_bait: # 突发由头要穿透对话:有人刚把你单独引向某处,别被闲聊黏住而错过(否则诱饵被消费却没人看见) lines.append(snap.lure_bait + "(若这由头值得一试,别耗在这场闲聊里——leave=true 抽身去赴约/查看。)") if snap.stage_directive: # 阶段聚焦也要穿透对话:别让闲聊把当前这一幕的处境/任务冲淡 lines.append(snap.stage_directive + "(若这场谈话无助于眼下这一幕,礼貌收场,leave=true。)") if snap.private_note: # 凶手的剧本钟点也要穿透对话:否则他会被无尽闲聊锁死,错过整幕 # (实测:江离与白聿在厨房荡了1.5小时秋千,第一幕过点无人死亡) lines.append(snap.private_note + "(若这场谈话无助于你的幕,礼貌收场,leave=true。)") h = snap.now_h % 24 if h >= 23.5 or h < 5.0: lines.append("(夜已极深,人人都撑着倦意。)") if snap.facts: lines.append("你确切知道的事:" + ";".join(snap.facts[-6:])) if snap.dialogue_budget and snap.dialogue_rounds >= int(snap.dialogue_budget * 0.45): lines.append(f"(这场已经聊了 {snap.dialogue_rounds} 句了。)") lines += ["对话至今:", *snap.dialogue, "你接话(JSON):"] return sys, "\n".join(lines) def _call(self, system: str, user: str, tag: str = "L2", think: bool = False, max_tokens: int | None = None) -> str: t0 = time.monotonic() out = "{}" try: kw = {"max_tokens": max_tokens} if max_tokens else {} # 快失败:超时 25s(正常 5-7s,留足缓冲)、不重试——失败的决策下一拍重想即可, # 别让一次卡死的调用 35s×重试 拖成 70s+(思索长尾主因之一)。 out = self.llm.complete( [{"role": "system", "content": system}, {"role": "user", "content": user}], temperature=0.85, timeout=25, num_retries=0, _tag=tag, _thinking=think, **kw) except Exception as e: logger.warning("[%s] LLM 调用失败:%s", self.aid, e) dt = time.monotonic() - t0 in_chars = len(system) + len(user) _LATENCY.append((dt, self.aid, tag, think, in_chars, len(out))) if dt > SLOW_CALL_S: # 长尾告警:带上下文定位(单次>20s 已异常,多半超时重试) logger.warning("[慢LLM %.1fs] %s tag=%s think=%s in=%d out=%d", dt, self.aid, tag, think, in_chars, len(out)) return out def _to_decision(self, out: str) -> Decision: d = parse_json_block(out) sp = d.get("speech") or {} act = d.get("action") or {} try: minutes = float(d.get("minutes", 10) or 10) except (TypeError, ValueError): minutes = 10.0 gu = d.get("goal_update") return Decision( scheme=str(d.get("scheme", "") or ""), thought=str(d.get("thought", "")), intent=str(d.get("intent", "")) or self.intent, speech_to=str(sp.get("to", "") if isinstance(sp, dict) else ""), speech=str(sp.get("text", "") if isinstance(sp, dict) else sp or ""), gesture=str(d.get("gesture", "") or ""), action_kind=str(act.get("kind", "") if isinstance(act, dict) else ""), action_arg=str(act.get("arg", "") if isinstance(act, dict) else ""), minutes=max(2.0, min(10.0, minutes)), # 单段动作≤10分钟:世界保持流动,想久了再续 goal_update=gu if isinstance(gu, dict) else None, replan=[str(s) for s in d["goals_replan"]][:4] if isinstance(d.get("goals_replan"), list) and d["goals_replan"] else None, raw=out) def apply_goal_update(self, gu: dict) -> None: """把决策里的目标进度自评落到清单上。""" if not self.goals: return try: idx = int(gu.get("index", 0)) - 1 except (TypeError, ValueError): return if 0 <= idx < len(self.goals): status = str(gu.get("status", "")).strip() if status in ("todo", "doing", "done", "blocked"): self.goals[idx]["status"] = status if gu.get("note"): self.goals[idx]["note"] = str(gu["note"])[:30] def apply_replan(self, steps: list[str]) -> None: """局势剧变时重排分目标:终极动机(persona)不变,保留近期已达成的成就,换上新的行动计划。""" done = [g for g in (self.goals or []) if g["status"] == "done"][-2:] fresh = [{"desc": str(s), "status": "todo", "note": ""} for s in steps if str(s).strip()][:4] if fresh: self.goals = done + fresh def mbti_traits(mbti: str) -> str: """MBTI → 确定性行为倾向(G19):同型号稳定同倾向,不靠每次采样。""" m = (mbti or "XXXX").upper() t = [ "有人作伴更安心,倾向结伴行动、当众表达" if m[0:1] == "E" else "独处不慌,惯于单独行动、观察多于表达", "只信亲眼所见的细节与实证" if m[1:2] == "S" else "善于从蛛丝马迹联想全局,直觉先行", "就事论事,决断冷静甚至显得无情" if m[2:3] == "T" else "顾及人情,决断易被情感与关系牵动", "凡事要有计划与掌控,讨厌变数" if m[3:4] == "J" else "随机应变,走一步看一步", ] return "【性格倾向】" + ";".join(t) + "。" def persona_card_from_manifest(cm) -> str: """把 character manifest 压成人设卡(drive 置顶,secret/role 是私有知识,只进本人 prompt)。""" lines = [] drive = getattr(cm, "drive", "") or "" if drive: # 终极目的永置最顶:任务清单只是达成它的手段,别让 agent 退化成打勾机器 lines.append("【你的终极目的(贯穿今夜、永不改变;下面一切计划/任务都只是达成它的手段)】\n" + drive) lines.append(f"{cm.age}岁,{cm.occupation},MBTI {cm.mbti}。{getattr(cm, 'backstory', '')}") lines.append(mbti_traits(getattr(cm, "mbti", ""))) secret = getattr(cm, "secret", "") or "" if secret: lines.append(f"【你的秘密(绝不轻易示人)】{secret}") role = getattr(cm, "role", "") or "" if role: lines.append(f"【你的真实身份与行动准则】{role}") knows = getattr(cm, "knows", "") or "" if knows: # 你恰好掌握的、可立刻付诸行动的情报/机关线索(别人未必知道)——把它用起来,别干等 lines.append(f"【你知道的关键内情(可立刻行动,别人未必知情)】{knows}") return "\n".join(lines) def build_dynamic(snapshot_time: str, weather: str, power_on: bool, deaths: list[str]) -> str: return dynamic_bulletin(snapshot_time, weather, power_on, deaths)