115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
"""解析模型返回为演化候选草稿（纯函数，无 IO）。
|
||
|
||
模型可能返回纯 JSON、markdown 代码块或夹杂解释文本；本模块容错提取首个 JSON
|
||
数组，并归一化为演化候选草稿。``evidence`` 序号由 application 映射为记忆 ID。
|
||
|
||
缺省字段取**安全悲观默认**：``level=L4``（必须人工确认）、``auto_apply_allowed=False``，
|
||
避免模型漏字段时误自动应用高风险演化。
|
||
"""
|
||
|
||
import json
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
|
||
@dataclass
|
||
class EvolutionCandidateDraft:
|
||
type: str
|
||
title: str
|
||
proposal: str
|
||
level: str = "L4"
|
||
scope: str = "character_scope"
|
||
source: str = "big_cycle"
|
||
delta: dict[str, Any] = field(default_factory=dict)
|
||
confidence: float = 0.5
|
||
benefit_score: float = 0.5
|
||
risk_score: float = 0.5
|
||
drift_score: float = 0.0
|
||
auto_apply_allowed: bool = False
|
||
evidence: list[int] = field(default_factory=list)
|
||
|
||
|
||
def parse_evolution_candidates(raw: str) -> list[EvolutionCandidateDraft]:
|
||
drafts: list[EvolutionCandidateDraft] = []
|
||
for item in _extract_json_array(raw):
|
||
if not isinstance(item, dict):
|
||
continue
|
||
type_ = str(item.get("type", "")).strip()
|
||
title = str(item.get("title", "")).strip()
|
||
proposal = str(item.get("proposal", "")).strip()
|
||
if not type_ or not title or not proposal:
|
||
continue
|
||
drafts.append(
|
||
EvolutionCandidateDraft(
|
||
type=type_,
|
||
title=title,
|
||
proposal=proposal,
|
||
level=_level(item.get("level")),
|
||
scope=_scope(item.get("scope")),
|
||
source=str(item.get("source", "big_cycle")).strip() or "big_cycle",
|
||
delta=_dict(item.get("delta")),
|
||
confidence=_clamp_float(item.get("confidence"), 0.5),
|
||
benefit_score=_clamp_float(item.get("benefit_score"), 0.5),
|
||
risk_score=_clamp_float(item.get("risk_score"), 0.5),
|
||
drift_score=_clamp_float(item.get("drift_score"), 0.0),
|
||
auto_apply_allowed=bool(item.get("auto_apply_allowed", False)),
|
||
evidence=_int_list(item.get("evidence")),
|
||
)
|
||
)
|
||
return drafts
|
||
|
||
|
||
def _extract_json_array(raw: str) -> list[Any]:
|
||
text = raw.strip()
|
||
matched = re.search(r"```(?:json)?\s*(\[.*?\])\s*```", text, re.DOTALL)
|
||
if matched:
|
||
text = matched.group(1)
|
||
else:
|
||
start, end = text.find("["), text.rfind("]")
|
||
if start != -1 and end > start:
|
||
text = text[start : end + 1]
|
||
try:
|
||
data = json.loads(text)
|
||
except json.JSONDecodeError:
|
||
return []
|
||
return data if isinstance(data, list) else []
|
||
|
||
|
||
_VALID_LEVELS = {"L0", "L1", "L2", "L3", "L4"}
|
||
_VALID_SCOPES = {"character_scope", "user_scope", "task_scope", "global_scope"}
|
||
|
||
|
||
def _level(value: Any) -> str:
|
||
text = str(value or "").strip().upper()
|
||
return text if text in _VALID_LEVELS else "L4"
|
||
|
||
|
||
def _scope(value: Any) -> str:
|
||
text = str(value or "").strip()
|
||
return text if text in _VALID_SCOPES else "character_scope"
|
||
|
||
|
||
def _dict(value: Any) -> dict[str, Any]:
|
||
return value if isinstance(value, dict) else {}
|
||
|
||
|
||
def _clamp_float(value: Any, default: float) -> float:
|
||
try:
|
||
num = float(value)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
return max(0.0, min(1.0, num))
|
||
|
||
|
||
def _int_list(value: Any) -> list[int]:
|
||
if not isinstance(value, list):
|
||
return []
|
||
result: list[int] = []
|
||
for item in value:
|
||
try:
|
||
result.append(int(item))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
return result
|