141 lines
3.9 KiB
Python
141 lines
3.9 KiB
Python
"""解析模型返回为记忆/世界抽取草稿（纯函数，无 IO）。
|
|
|
|
模型输出对象 ``{"memories": [...], "entities": [...]}``。缺省取**安全悲观默认**：
|
|
记忆 ``sensitivity=low``、世界 ``state_type=inference``（非 fact，召回降权，见
|
|
[世界模型设计] ../../docs/domain/core-domains.md §8）。
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class MemoryCandidateDraft:
|
|
type: str
|
|
title: str
|
|
content: str
|
|
confidence: float = 0.5
|
|
importance: float = 0.5
|
|
sensitivity: str = "low"
|
|
reason: str = ""
|
|
|
|
|
|
@dataclass
|
|
class WorldEntityDraft:
|
|
type: str
|
|
name: str
|
|
attributes: dict[str, Any] = field(default_factory=dict)
|
|
state_type: str = "inference"
|
|
confidence: float = 0.5
|
|
|
|
|
|
_MEMORY_TYPES = {
|
|
"user",
|
|
"preference",
|
|
"event",
|
|
"relationship",
|
|
"skill",
|
|
"goal",
|
|
"other",
|
|
}
|
|
_ENTITY_TYPES = {
|
|
"self",
|
|
"person",
|
|
"object",
|
|
"place",
|
|
"device",
|
|
"project",
|
|
"task",
|
|
"concept",
|
|
}
|
|
_STATE_TYPES = {"fact", "observation", "inference", "hypothesis", "stale"}
|
|
_SENSITIVITY = {"low", "medium", "high"}
|
|
|
|
|
|
def parse_extraction(
|
|
raw: str,
|
|
) -> tuple[list[MemoryCandidateDraft], list[WorldEntityDraft]]:
|
|
data = extract_json_object(raw)
|
|
if not isinstance(data, dict):
|
|
return [], []
|
|
return parse_memory_drafts(data.get("memories")), _parse_entities(data.get("entities"))
|
|
|
|
|
|
def parse_memory_drafts(value: Any) -> list[MemoryCandidateDraft]:
|
|
if not isinstance(value, list):
|
|
return []
|
|
result: list[MemoryCandidateDraft] = []
|
|
for item in value:
|
|
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
|
|
result.append(
|
|
MemoryCandidateDraft(
|
|
type=_enum(item.get("type"), _MEMORY_TYPES, "other"),
|
|
title=title,
|
|
content=content,
|
|
confidence=_clamp(item.get("confidence"), 0.5),
|
|
importance=_clamp(item.get("importance"), 0.5),
|
|
sensitivity=_enum(item.get("sensitivity"), _SENSITIVITY, "low"),
|
|
reason=str(item.get("reason", "")).strip(),
|
|
)
|
|
)
|
|
return result
|
|
|
|
|
|
def _parse_entities(value: Any) -> list[WorldEntityDraft]:
|
|
if not isinstance(value, list):
|
|
return []
|
|
result: list[WorldEntityDraft] = []
|
|
for item in value:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
name = str(item.get("name", "")).strip()
|
|
if not name:
|
|
continue
|
|
attrs = item.get("attributes")
|
|
result.append(
|
|
WorldEntityDraft(
|
|
type=_enum(item.get("type"), _ENTITY_TYPES, "concept"),
|
|
name=name,
|
|
attributes=attrs if isinstance(attrs, dict) else {},
|
|
state_type=_enum(item.get("state_type"), _STATE_TYPES, "inference"),
|
|
confidence=_clamp(item.get("confidence"), 0.5),
|
|
)
|
|
)
|
|
return result
|
|
|
|
|
|
def extract_json_object(raw: str) -> dict[str, Any]:
|
|
text = raw.strip()
|
|
if text.startswith("```"):
|
|
text = re.sub(r"^```(?:json)?\s*", "", text)
|
|
text = re.sub(r"\s*```$", "", text)
|
|
start = text.find("{")
|
|
end = 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, dict) else {}
|
|
|
|
|
|
def _enum(value: Any, valid: set[str], default: str) -> str:
|
|
text = str(value or "").strip().lower()
|
|
return text if text in valid else default
|
|
|
|
|
|
def _clamp(value: Any, default: float) -> float:
|
|
try:
|
|
num = float(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
return max(0.0, min(1.0, num))
|