Private
Public Access
Per spec FR4 + Phase 3.4: Persona dataclass + properties (provider/model/ temperature/top_p/max_output_tokens) + to_dict/from_dict move from src/models.py into src/personas.py (which already has the PersonaManager ops layer). Re-export at top of models.py preserves 'from src.models import Persona'.
171 lines
5.6 KiB
Python
171 lines
5.6 KiB
Python
"""Personas module: Persona dataclass + PersonaManager CRUD.
|
|
|
|
Per module_taxonomy_refactor_20260627 Phase 3.4, the Persona dataclass
|
|
moved from src/models.py into this module. PersonaManager (the ops layer
|
|
that loads/saves Persona instances to TOML) was already here.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import tomllib
|
|
import tomli_w
|
|
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Dict, Any, Optional
|
|
|
|
from src import paths
|
|
from src.type_aliases import Metadata
|
|
|
|
|
|
@dataclass
|
|
class Persona:
|
|
name: str
|
|
preferred_models: list[Metadata] = field(default_factory=list)
|
|
system_prompt: str = ''
|
|
tool_preset: Optional[str] = None
|
|
bias_profile: Optional[str] = None
|
|
context_preset: Optional[str] = None
|
|
aggregation_strategy: Optional[str] = None
|
|
|
|
@property
|
|
def provider(self) -> str:
|
|
if not self.preferred_models: return ""
|
|
return self.preferred_models[0].get("provider") or ""
|
|
|
|
@property
|
|
def model(self) -> str:
|
|
if not self.preferred_models: return ""
|
|
return self.preferred_models[0].get("model") or ""
|
|
|
|
@property
|
|
def temperature(self) -> float:
|
|
if not self.preferred_models: return 0.0
|
|
return float(self.preferred_models[0].get("temperature") or 0.0)
|
|
|
|
@property
|
|
def top_p(self) -> float:
|
|
if not self.preferred_models: return 1.0
|
|
return float(self.preferred_models[0].get("top_p") or 1.0)
|
|
|
|
@property
|
|
def max_output_tokens(self) -> int:
|
|
if not self.preferred_models: return 0
|
|
return int(self.preferred_models[0].get("max_output_tokens") or 0)
|
|
|
|
def to_dict(self) -> Metadata:
|
|
res = {"system_prompt": self.system_prompt}
|
|
if self.preferred_models:
|
|
processed = []
|
|
for m in self.preferred_models:
|
|
if isinstance(m, str):
|
|
processed.append({"model": m})
|
|
else:
|
|
processed.append(m)
|
|
res["preferred_models"] = processed
|
|
if self.tool_preset is not None: res["tool_preset"] = self.tool_preset
|
|
if self.bias_profile is not None: res["bias_profile"] = self.bias_profile
|
|
if self.context_preset is not None: res["context_preset"] = self.context_preset
|
|
if self.aggregation_strategy is not None: res["aggregation_strategy"] = self.aggregation_strategy
|
|
return res
|
|
|
|
@classmethod
|
|
def from_dict(cls, name: str, data: Metadata) -> "Persona":
|
|
raw_models = data.get("preferred_models", [])
|
|
parsed_models = []
|
|
for m in raw_models:
|
|
if isinstance(m, str):
|
|
parsed_models.append({"model": m})
|
|
else:
|
|
parsed_models.append(m)
|
|
legacy = {}
|
|
for k in ["provider", "model", "temperature", "top_p", "max_output_tokens"]:
|
|
if data.get(k) is not None:
|
|
legacy[k] = data[k]
|
|
if legacy:
|
|
if not parsed_models:
|
|
parsed_models.append(legacy)
|
|
else:
|
|
for k, v in legacy.items():
|
|
if k not in parsed_models[0] or parsed_models[0][k] is None:
|
|
parsed_models[0][k] = v
|
|
return cls(
|
|
name = name,
|
|
preferred_models = parsed_models,
|
|
system_prompt = data.get("system_prompt", ""),
|
|
tool_preset = data.get("tool_preset"),
|
|
bias_profile = data.get("bias_profile"),
|
|
context_preset = data.get("context_preset"),
|
|
aggregation_strategy = data.get("aggregation_strategy"),
|
|
)
|
|
|
|
|
|
class PersonaManager:
|
|
"""Manages Persona profiles across global and project-specific files."""
|
|
|
|
def __init__(self, project_root: Optional[Path] = None):
|
|
self.project_root = project_root
|
|
|
|
def _get_path(self, scope: str) -> Path:
|
|
if scope == "global":
|
|
return paths.get_global_personas_path()
|
|
elif scope == "project":
|
|
if not self.project_root:
|
|
raise ValueError("Project root is not set, cannot resolve project scope.")
|
|
return paths.get_project_personas_path(self.project_root)
|
|
else:
|
|
raise ValueError("Invalid scope, must be 'global' or 'project'")
|
|
|
|
def load_all(self) -> Dict[str, Persona]:
|
|
personas = {}
|
|
global_path = paths.get_global_personas_path()
|
|
global_data = self._load_file(global_path)
|
|
for name, data in global_data.get("personas", {}).items():
|
|
personas[name] = Persona.from_dict(name, data)
|
|
if self.project_root:
|
|
project_path = paths.get_project_personas_path(self.project_root)
|
|
project_data = self._load_file(project_path)
|
|
for name, data in project_data.get("personas", {}).items():
|
|
personas[name] = Persona.from_dict(name, data)
|
|
return personas
|
|
|
|
def save_persona(self, persona: Persona, scope: str = "project") -> None:
|
|
path = self._get_path(scope)
|
|
data = self._load_file(path)
|
|
if "personas" not in data:
|
|
data["personas"] = {}
|
|
data["personas"][persona.name] = persona.to_dict()
|
|
self._save_file(path, data)
|
|
|
|
def get_persona_scope(self, name: str) -> str:
|
|
"""Returns the scope ('global' or 'project') of a persona by name."""
|
|
if self.project_root:
|
|
project_path = paths.get_project_personas_path(self.project_root)
|
|
project_data = self._load_file(project_path)
|
|
if name in project_data.get("personas", {}):
|
|
return "project"
|
|
global_path = paths.get_global_personas_path()
|
|
global_data = self._load_file(global_path)
|
|
if name in global_data.get("personas", {}):
|
|
return "global"
|
|
return "project"
|
|
|
|
def delete_persona(self, name: str, scope: str = "project") -> None:
|
|
path = self._get_path(scope)
|
|
data = self._load_file(path)
|
|
if "personas" in data and name in data["personas"]:
|
|
del data["personas"][name]
|
|
self._save_file(path, data)
|
|
|
|
def _load_file(self, path: Path) -> Dict[str, Any]:
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
with open(path, "rb") as f:
|
|
return tomllib.load(f)
|
|
except Exception:
|
|
return {}
|
|
|
|
def _save_file(self, path: Path, data: Dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(path, "wb") as f:
|
|
tomli_w.dump(data, f) |