144 lines
4.2 KiB
Python
144 lines
4.2 KiB
Python
"""解析模型返回为候选草稿（纯函数，无 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"
|