79 lines
3.7 KiB
Python
79 lines
3.7 KiB
Python
"""大周天演化候选 prompt 构建（纯函数，无 IO）。
|
||
|
||
接收 ``EvaluationSnapshot``（由 application 阶段从 ORM 组装的态势快照），产出
|
||
``(system, user)`` 消息对，喂给模型网关。模型返回的 JSON 由 ``extract`` 模块解析。
|
||
|
||
枚举取值以 [契约字典](../../docs/contracts/contract-glossary.md) §12 与
|
||
[自我演化机制设计](../../docs/domain/core-domains.md) §11 为准。
|
||
"""
|
||
|
||
from dataclasses import dataclass, field
|
||
|
||
from app.modules.evolution import LEVELS as EVOLUTION_LEVELS
|
||
from app.modules.evolution import SCOPES as EVOLUTION_SCOPES
|
||
from app.modules.evolution import TYPES as EVOLUTION_TYPES
|
||
|
||
EVOLUTION_TYPES = EVOLUTION_TYPES
|
||
EVOLUTION_LEVELS = EVOLUTION_LEVELS
|
||
EVOLUTION_SCOPES = EVOLUTION_SCOPES
|
||
|
||
|
||
@dataclass
|
||
class EvaluationSnapshot:
|
||
"""大周天自我评估态势快照（application 组装，供候选生成消费）。"""
|
||
|
||
character_id: str
|
||
memory_count: int
|
||
recent_memories: list[tuple[str, str]] # (title, content)
|
||
personality_summary: str
|
||
world_entity_count: int
|
||
conclusion: str
|
||
evidence_ids: list[str] = field(default_factory=list) # 记忆 ID，供 evidence 序号映射
|
||
cumulative_drift: float = 0.0 # 最近大周天的累积漂移值
|
||
drift_budget_remaining: float = 0.0 # 剩余漂移预算
|
||
|
||
|
||
def build_evolution_extraction_messages(
|
||
snapshot: EvaluationSnapshot,
|
||
) -> tuple[str, str]:
|
||
system = (
|
||
"你是魂镜的大周天自我演化引擎。基于本轮自我评估的态势快照，生成轻量、可解释、"
|
||
"可回滚的演化候选——只调整人格表达、程序记忆、工具策略等轻量参数，"
|
||
"不改变核心使命、安全边界或高敏关系判断。"
|
||
f"type 取自：{', '.join(EVOLUTION_TYPES)}。"
|
||
f"level 取自：{', '.join(EVOLUTION_LEVELS)}"
|
||
"（L0-L3 低风险可半自动，L4 核心人格/使命/安全边界必须人工确认）。"
|
||
f"scope 取自：{', '.join(EVOLUTION_SCOPES)}。"
|
||
'以 JSON 数组返回，每个元素：{"type": str, "source": "big_cycle", '
|
||
'"level": str, "scope": str, "title": str, "proposal": str, '
|
||
'"delta": 对象, "confidence": 0~1, "benefit_score": 0~1, '
|
||
'"risk_score": 0~1, "auto_apply_allowed": bool, "evidence": [记忆序号]}。'
|
||
"delta 为具体增量（如人格字段覆盖对象、程序记忆 title/content）。"
|
||
"evidence 为下方记忆摘要的序号（[0]/[1]…）列表。"
|
||
"仅在态势充分支撑时提议，宁缺毋滥。若本轮无需演化返回 []。"
|
||
"只返回 JSON 数组，不要解释或 markdown。"
|
||
)
|
||
user = (
|
||
f"记忆总数：{snapshot.memory_count}\n"
|
||
f"世界实体数：{snapshot.world_entity_count}\n"
|
||
f"自我评估结论：{snapshot.conclusion or '（无明显态势）'}\n"
|
||
f"当前人格摘要：\n{snapshot.personality_summary or '（尚未形成）'}\n"
|
||
f"近期记忆摘要：\n{_format_memories(snapshot.recent_memories)}"
|
||
)
|
||
# 人格权重向量：将漂移信息注入 prompt，引导 LLM 保守生成
|
||
if snapshot.cumulative_drift > 0:
|
||
user += (
|
||
f"\n\n⚠️ 人格漂移警告：累积漂移 {snapshot.cumulative_drift:.2f} / "
|
||
f"预算 {snapshot.drift_budget_remaining + snapshot.cumulative_drift:.1f}。"
|
||
f"剩余漂移预算 {snapshot.drift_budget_remaining:.1f}。"
|
||
f"请尽可能保守，减少高风险 proposal，优先微调而非大幅改动。"
|
||
)
|
||
return system, user
|
||
|
||
|
||
def _format_memories(memories: list[tuple[str, str]]) -> str:
|
||
if not memories:
|
||
return "（暂无）"
|
||
lines = [f"[{idx}] {title}：{content}" for idx, (title, content) in enumerate(memories)]
|
||
return "\n".join(lines)
|