"""解析模型返回为候选草稿(纯函数,无 IO)。 模型可能返回纯 JSON、markdown 代码块或夹杂解释文本;本模块容错提取首个 JSON 数组, 并归一化为候选草稿。evidence 序号由 application 映射为 ``import_items.id``。 """ import json import re from dataclasses import dataclass, field from typing import Any @dataclass class ProfileCandidateDraft: field_name: str value: str confidence: float reason: str evidence: list[int] = field(default_factory=list) @dataclass class MemoryCandidateDraft: type: str title: str content: str importance: float sensitivity: str reason: str evidence: list[int] = field(default_factory=list) @dataclass class PersonalityProposalDraft: field_name: str change: str reason: str evidence: list[int] = field(default_factory=list) def parse_profile_candidates(raw: str) -> list[ProfileCandidateDraft]: drafts: list[ProfileCandidateDraft] = [] for item in _extract_json_array(raw): if not isinstance(item, dict): continue field_name = str(item.get("field", "")).strip() value = str(item.get("value", "")).strip() if not field_name or not value: continue drafts.append( ProfileCandidateDraft( field_name=field_name, value=value, confidence=_clamp_float(item.get("confidence"), 0.5), reason=str(item.get("reason", "")).strip(), evidence=_int_list(item.get("evidence")), ) ) return drafts def parse_memory_candidates(raw: str) -> list[MemoryCandidateDraft]: drafts: list[MemoryCandidateDraft] = [] for item in _extract_json_array(raw): if not isinstance(item, dict): continue title = str(item.get("title", "")).strip() content = str(item.get("content", "")).strip() if not title or not content: continue drafts.append( MemoryCandidateDraft( type=str(item.get("type", "other")).strip() or "other", title=title, content=content, importance=_clamp_float(item.get("importance"), 0.5), sensitivity=_sensitivity(item.get("sensitivity")), reason=str(item.get("reason", "")).strip(), evidence=_int_list(item.get("evidence")), ) ) return drafts def parse_personality_proposals(raw: str) -> list[PersonalityProposalDraft]: drafts: list[PersonalityProposalDraft] = [] for item in _extract_json_array(raw): if not isinstance(item, dict): continue field_name = str(item.get("field", "")).strip() change = str(item.get("change", "")).strip() if not field_name or not change: continue drafts.append( PersonalityProposalDraft( field_name=field_name, change=change, reason=str(item.get("reason", "")).strip(), 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 [] 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 def _sensitivity(value: Any) -> str: text = str(value or "").strip().lower() return text if text in {"low", "medium", "high"} else "low"