"""Utilities for managing per-user personalization settings.""" from __future__ import annotations import json from copy import deepcopy from pathlib import Path from typing import Any, Dict, Iterable, Optional, Union try: from config.limits import THINKING_FAST_INTERVAL except ImportError: THINKING_FAST_INTERVAL = 10 from core.tool_config import TOOL_CATEGORIES from config.model_profiles import get_registered_model_keys ALLOWED_RUN_MODES = {"fast", "thinking", "deep"} ALLOWED_PERMISSION_MODES = {"readonly", "approval", "unrestricted"} PERSONALIZATION_FILENAME = "personalization.json" MAX_SHORT_FIELD_LENGTH = 20 MAX_CONSIDERATION_LENGTH = 50 MAX_CONSIDERATION_ITEMS = 10 TONE_PRESETS = ["健谈", "幽默", "直言不讳", "鼓励性", "诗意", "企业商务", "打破常规", "同理心"] THINKING_INTERVAL_MIN = 1 THINKING_INTERVAL_MAX = 50 DEFAULT_SHALLOW_COMPRESS_TRIGGER_TOKENS = 80_000 DEFAULT_SHALLOW_COMPRESS_KEEP_RECENT_TOOLS = 15 DEFAULT_SHALLOW_COMPRESS_MAX_REPLACE_PER_ROUND = 10 DEFAULT_SHALLOW_COMPRESS_TRIGGER_TOOL_CALLS_INTERVAL = 10 DEFAULT_DEEP_COMPRESS_TRIGGER_TOKENS = 150_000 MIN_COMPRESSION_TRIGGER_TOKENS = 1_000 MAX_COMPRESSION_TRIGGER_TOKENS = 2_000_000 MIN_SHALLOW_KEEP_RECENT_TOOLS = 0 MAX_SHALLOW_KEEP_RECENT_TOOLS = 500 MIN_SHALLOW_MAX_REPLACE_PER_ROUND = 1 MAX_SHALLOW_MAX_REPLACE_PER_ROUND = 200 MIN_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL = 1 MAX_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL = 200 DEFAULT_PERSONALIZATION_CONFIG: Dict[str, Any] = { "enabled": False, "self_identify": "", "user_name": "", "use_custom_names": False, "profession": "", "tone": "", "considerations": [], "thinking_interval": None, "disabled_tool_categories": [], "enabled_skills": None, "skills_catalog_snapshot": None, "default_run_mode": None, "default_permission_mode": "unrestricted", "auto_generate_title": True, "tool_intent_enabled": True, "skill_hints_enabled": False, # Skill 提示系统开关（默认关闭） "skill_strict_terminal_enabled": False, # 强约束：terminal 系列工具需先阅读 terminal-guide "skill_strict_sub_agent_enabled": False, # 强约束：子智能体系列工具需先阅读 sub-agent-guide "skill_strict_run_command_foreground_enabled": False, # 强约束：run_command 前台模式需先阅读 run-command-guide "skill_strict_run_command_background_enabled": False, # 强约束：run_command 后台模式需先阅读 run-command-guide "default_model": "kimi-k2.5", "image_compression": "original", # original / 1080p / 720p / 540p "auto_shallow_compress_enabled": False, "auto_deep_compress_enabled": False, "shallow_compress_trigger_tokens": None, "shallow_compress_keep_recent_tools": None, "shallow_compress_max_replace_per_round": None, "shallow_compress_trigger_tool_calls_interval": None, "deep_compress_trigger_tokens": None, "silent_tool_disable": False, # 禁用工具时不向模型插入提示 "enhanced_tool_display": True, # 增强工具显示 "versioning_restore_mode": "overwrite", # 版本回溯模式固定为 overwrite "agents_md_auto_inject": False, # AGENTS.md 自动注入开关 "allow_root_file_creation": False, # 允许在根目录创建文件开关 } __all__ = [ "PERSONALIZATION_FILENAME", "DEFAULT_PERSONALIZATION_CONFIG", "TONE_PRESETS", "MAX_CONSIDERATION_ITEMS", "load_personalization_config", "save_personalization_config", "ensure_personalization_config", "build_personalization_prompt", "sanitize_personalization_payload", "resolve_context_compression_settings", "validate_context_compression_settings", "ALLOWED_PERMISSION_MODES", ] PathLike = Union[str, Path] def _ensure_parent(path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) def _to_path(base: PathLike) -> Path: base_path = Path(base).expanduser() if base_path.is_dir(): return base_path / PERSONALIZATION_FILENAME return base_path def ensure_personalization_config(base_dir: PathLike) -> Dict[str, Any]: """Ensure the personalization file exists and return its content.""" path = _to_path(base_dir) _ensure_parent(path) if not path.exists(): with open(path, "w", encoding="utf-8") as f: json.dump(DEFAULT_PERSONALIZATION_CONFIG, f, ensure_ascii=False, indent=2) return deepcopy(DEFAULT_PERSONALIZATION_CONFIG) return load_personalization_config(base_dir) def load_personalization_config(base_dir: PathLike) -> Dict[str, Any]: """Load personalization config; fall back to defaults on errors.""" path = _to_path(base_dir) _ensure_parent(path) if not path.exists(): return ensure_personalization_config(base_dir) try: with open(path, "r", encoding="utf-8") as f: raw = json.load(f) sanitized = sanitize_personalization_payload(raw) # 若发现缺失字段（如默认模型）或数据被规范化，主动写回文件，避免下一次读取仍为旧格式 if sanitized != raw: with open(path, "w", encoding="utf-8") as wf: json.dump(sanitized, wf, ensure_ascii=False, indent=2) return sanitized except (json.JSONDecodeError, OSError): # 重置为默认配置，避免错误阻塞 with open(path, "w", encoding="utf-8") as f: json.dump(DEFAULT_PERSONALIZATION_CONFIG, f, ensure_ascii=False, indent=2) return deepcopy(DEFAULT_PERSONALIZATION_CONFIG) def sanitize_personalization_payload( payload: Optional[Dict[str, Any]], fallback: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """Normalize payload structure and clamp field lengths.""" base = deepcopy(DEFAULT_PERSONALIZATION_CONFIG) if fallback: base.update(fallback) data = payload or {} allowed_tool_categories = set(TOOL_CATEGORIES.keys()) allowed_models = set(get_registered_model_keys(visible_only=True)) allowed_image_modes = {"original", "1080p", "720p", "540p"} def _resolve_short_field(key: str) -> str: if key in data: return _sanitize_short_field(data.get(key)) return _sanitize_short_field(base.get(key)) base["enabled"] = bool(data.get("enabled", base["enabled"])) base["auto_generate_title"] = bool(data.get("auto_generate_title", base["auto_generate_title"])) base["self_identify"] = _resolve_short_field("self_identify") base["user_name"] = _resolve_short_field("user_name") base["profession"] = _resolve_short_field("profession") base["tone"] = _resolve_short_field("tone") if "considerations" in data: base["considerations"] = _sanitize_considerations(data.get("considerations")) else: base["considerations"] = _sanitize_considerations(base.get("considerations", [])) if "thinking_interval" in data: base["thinking_interval"] = _sanitize_thinking_interval(data.get("thinking_interval")) else: base["thinking_interval"] = _sanitize_thinking_interval(base.get("thinking_interval")) # 工具意图提示开关 if "tool_intent_enabled" in data: base["tool_intent_enabled"] = bool(data.get("tool_intent_enabled")) else: base["tool_intent_enabled"] = bool(base.get("tool_intent_enabled")) # Skill 提示系统开关 if "skill_hints_enabled" in data: base["skill_hints_enabled"] = bool(data.get("skill_hints_enabled")) else: base["skill_hints_enabled"] = bool(base.get("skill_hints_enabled")) # Skill 强约束开关（仅针对已启用的 skill 生效） if "skill_strict_terminal_enabled" in data: base["skill_strict_terminal_enabled"] = bool(data.get("skill_strict_terminal_enabled")) else: base["skill_strict_terminal_enabled"] = bool(base.get("skill_strict_terminal_enabled", False)) if "skill_strict_sub_agent_enabled" in data: base["skill_strict_sub_agent_enabled"] = bool(data.get("skill_strict_sub_agent_enabled")) else: base["skill_strict_sub_agent_enabled"] = bool(base.get("skill_strict_sub_agent_enabled", False)) if "skill_strict_run_command_foreground_enabled" in data: base["skill_strict_run_command_foreground_enabled"] = bool(data.get("skill_strict_run_command_foreground_enabled")) else: base["skill_strict_run_command_foreground_enabled"] = bool( base.get("skill_strict_run_command_foreground_enabled", False) ) if "skill_strict_run_command_background_enabled" in data: base["skill_strict_run_command_background_enabled"] = bool(data.get("skill_strict_run_command_background_enabled")) else: base["skill_strict_run_command_background_enabled"] = bool( base.get("skill_strict_run_command_background_enabled", False) ) if "disabled_tool_categories" in data: base["disabled_tool_categories"] = _sanitize_tool_categories(data.get("disabled_tool_categories"), allowed_tool_categories) else: base["disabled_tool_categories"] = _sanitize_tool_categories(base.get("disabled_tool_categories"), allowed_tool_categories) if "enabled_skills" in data: base["enabled_skills"] = _sanitize_skills(data.get("enabled_skills")) else: base["enabled_skills"] = _sanitize_skills(base.get("enabled_skills")) if "skills_catalog_snapshot" in data: base["skills_catalog_snapshot"] = _sanitize_skills(data.get("skills_catalog_snapshot")) else: base["skills_catalog_snapshot"] = _sanitize_skills(base.get("skills_catalog_snapshot")) if "default_run_mode" in data: base["default_run_mode"] = _sanitize_run_mode(data.get("default_run_mode")) else: base["default_run_mode"] = _sanitize_run_mode(base.get("default_run_mode")) permission_mode = data.get("default_permission_mode", base.get("default_permission_mode")) if isinstance(permission_mode, str) and permission_mode in ALLOWED_PERMISSION_MODES: base["default_permission_mode"] = permission_mode elif base.get("default_permission_mode") not in ALLOWED_PERMISSION_MODES: base["default_permission_mode"] = "unrestricted" base["versioning_restore_mode"] = "overwrite" # 默认模型 chosen_model = data.get("default_model", base.get("default_model")) if isinstance(chosen_model, str) and chosen_model in allowed_models: base["default_model"] = chosen_model elif allowed_models and base.get("default_model") not in allowed_models: base["default_model"] = "kimi-k2.5" # 图片压缩模式 img_mode = data.get("image_compression", base.get("image_compression")) if isinstance(img_mode, str) and img_mode in allowed_image_modes: base["image_compression"] = img_mode elif base.get("image_compression") not in allowed_image_modes: base["image_compression"] = "original" if "auto_shallow_compress_enabled" in data: base["auto_shallow_compress_enabled"] = bool(data.get("auto_shallow_compress_enabled")) else: base["auto_shallow_compress_enabled"] = bool(base.get("auto_shallow_compress_enabled", False)) if "auto_deep_compress_enabled" in data: base["auto_deep_compress_enabled"] = bool(data.get("auto_deep_compress_enabled")) else: base["auto_deep_compress_enabled"] = bool(base.get("auto_deep_compress_enabled", False)) if "shallow_compress_trigger_tokens" in data: base["shallow_compress_trigger_tokens"] = _sanitize_optional_int( data.get("shallow_compress_trigger_tokens"), min_value=MIN_COMPRESSION_TRIGGER_TOKENS, max_value=MAX_COMPRESSION_TRIGGER_TOKENS, ) else: base["shallow_compress_trigger_tokens"] = _sanitize_optional_int( base.get("shallow_compress_trigger_tokens"), min_value=MIN_COMPRESSION_TRIGGER_TOKENS, max_value=MAX_COMPRESSION_TRIGGER_TOKENS, ) if "shallow_compress_keep_recent_tools" in data: base["shallow_compress_keep_recent_tools"] = _sanitize_optional_int( data.get("shallow_compress_keep_recent_tools"), min_value=MIN_SHALLOW_KEEP_RECENT_TOOLS, max_value=MAX_SHALLOW_KEEP_RECENT_TOOLS, ) else: base["shallow_compress_keep_recent_tools"] = _sanitize_optional_int( base.get("shallow_compress_keep_recent_tools"), min_value=MIN_SHALLOW_KEEP_RECENT_TOOLS, max_value=MAX_SHALLOW_KEEP_RECENT_TOOLS, ) if "shallow_compress_max_replace_per_round" in data: base["shallow_compress_max_replace_per_round"] = _sanitize_optional_int( data.get("shallow_compress_max_replace_per_round"), min_value=MIN_SHALLOW_MAX_REPLACE_PER_ROUND, max_value=MAX_SHALLOW_MAX_REPLACE_PER_ROUND, ) else: base["shallow_compress_max_replace_per_round"] = _sanitize_optional_int( base.get("shallow_compress_max_replace_per_round"), min_value=MIN_SHALLOW_MAX_REPLACE_PER_ROUND, max_value=MAX_SHALLOW_MAX_REPLACE_PER_ROUND, ) if "deep_compress_trigger_tokens" in data: base["deep_compress_trigger_tokens"] = _sanitize_optional_int( data.get("deep_compress_trigger_tokens"), min_value=MIN_COMPRESSION_TRIGGER_TOKENS, max_value=MAX_COMPRESSION_TRIGGER_TOKENS, ) else: base["deep_compress_trigger_tokens"] = _sanitize_optional_int( base.get("deep_compress_trigger_tokens"), min_value=MIN_COMPRESSION_TRIGGER_TOKENS, max_value=MAX_COMPRESSION_TRIGGER_TOKENS, ) if "shallow_compress_trigger_tool_calls_interval" in data: base["shallow_compress_trigger_tool_calls_interval"] = _sanitize_optional_int( data.get("shallow_compress_trigger_tool_calls_interval"), min_value=MIN_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL, max_value=MAX_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL, ) else: base["shallow_compress_trigger_tool_calls_interval"] = _sanitize_optional_int( base.get("shallow_compress_trigger_tool_calls_interval"), min_value=MIN_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL, max_value=MAX_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL, ) # 静默禁用工具提示 if "silent_tool_disable" in data: base["silent_tool_disable"] = bool(data.get("silent_tool_disable")) else: base["silent_tool_disable"] = bool(base.get("silent_tool_disable")) # 增强工具显示 if "enhanced_tool_display" in data: base["enhanced_tool_display"] = bool(data.get("enhanced_tool_display")) else: base["enhanced_tool_display"] = bool(base.get("enhanced_tool_display", True)) # 使用自定义称呼 if "use_custom_names" in data: base["use_custom_names"] = bool(data.get("use_custom_names")) else: base["use_custom_names"] = bool(base.get("use_custom_names")) # AGENTS.md 自动注入开关 if "agents_md_auto_inject" in data: base["agents_md_auto_inject"] = bool(data.get("agents_md_auto_inject")) else: base["agents_md_auto_inject"] = bool(base.get("agents_md_auto_inject", False)) # 允许根目录创建文件开关 if "allow_root_file_creation" in data: base["allow_root_file_creation"] = bool(data.get("allow_root_file_creation")) else: base["allow_root_file_creation"] = bool(base.get("allow_root_file_creation", False)) return base def _sanitize_skills(value: Any) -> Optional[list]: """Sanitize enabled skills list / 清洗启用技能列表。""" if value is None: return None if not isinstance(value, list): return [] cleaned: list = [] seen = set() for item in value: if not isinstance(item, str): continue skill_id = item.strip() if not skill_id or skill_id in seen: continue cleaned.append(skill_id) seen.add(skill_id) return cleaned def save_personalization_config(base_dir: PathLike, payload: Dict[str, Any]) -> Dict[str, Any]: """Persist sanitized personalization config and return it.""" existing = load_personalization_config(base_dir) config = sanitize_personalization_payload(payload, fallback=existing) validate_context_compression_settings(config) path = _to_path(base_dir) _ensure_parent(path) with open(path, "w", encoding="utf-8") as f: json.dump(config, f, ensure_ascii=False, indent=2) return config def _sanitize_optional_int(value: Any, *, min_value: int, max_value: int) -> Optional[int]: if value is None or value == "": return None if isinstance(value, bool): return None try: parsed = int(value) except (TypeError, ValueError): return None if parsed < min_value: return min_value if parsed > max_value: return max_value return parsed def resolve_context_compression_settings(config: Optional[Dict[str, Any]]) -> Dict[str, int]: data = config or {} shallow_trigger = _sanitize_optional_int( data.get("shallow_compress_trigger_tokens"), min_value=MIN_COMPRESSION_TRIGGER_TOKENS, max_value=MAX_COMPRESSION_TRIGGER_TOKENS, ) or DEFAULT_SHALLOW_COMPRESS_TRIGGER_TOKENS shallow_keep_recent = _sanitize_optional_int( data.get("shallow_compress_keep_recent_tools"), min_value=MIN_SHALLOW_KEEP_RECENT_TOOLS, max_value=MAX_SHALLOW_KEEP_RECENT_TOOLS, ) if shallow_keep_recent is None: shallow_keep_recent = DEFAULT_SHALLOW_COMPRESS_KEEP_RECENT_TOOLS shallow_max_replace = _sanitize_optional_int( data.get("shallow_compress_max_replace_per_round"), min_value=MIN_SHALLOW_MAX_REPLACE_PER_ROUND, max_value=MAX_SHALLOW_MAX_REPLACE_PER_ROUND, ) or DEFAULT_SHALLOW_COMPRESS_MAX_REPLACE_PER_ROUND shallow_trigger_interval = _sanitize_optional_int( data.get("shallow_compress_trigger_tool_calls_interval"), min_value=MIN_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL, max_value=MAX_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL, ) or DEFAULT_SHALLOW_COMPRESS_TRIGGER_TOOL_CALLS_INTERVAL deep_trigger = _sanitize_optional_int( data.get("deep_compress_trigger_tokens"), min_value=MIN_COMPRESSION_TRIGGER_TOKENS, max_value=MAX_COMPRESSION_TRIGGER_TOKENS, ) or DEFAULT_DEEP_COMPRESS_TRIGGER_TOKENS if deep_trigger <= shallow_trigger: deep_trigger = shallow_trigger + 1 return { "shallow_trigger_tokens": shallow_trigger, "shallow_keep_recent_tools": shallow_keep_recent, "shallow_max_replace_per_round": shallow_max_replace, "shallow_trigger_tool_calls_interval": shallow_trigger_interval, "deep_trigger_tokens": deep_trigger, } def validate_context_compression_settings(config: Optional[Dict[str, Any]]) -> None: data = config or {} shallow_trigger = _sanitize_optional_int( data.get("shallow_compress_trigger_tokens"), min_value=MIN_COMPRESSION_TRIGGER_TOKENS, max_value=MAX_COMPRESSION_TRIGGER_TOKENS, ) or DEFAULT_SHALLOW_COMPRESS_TRIGGER_TOKENS deep_trigger = _sanitize_optional_int( data.get("deep_compress_trigger_tokens"), min_value=MIN_COMPRESSION_TRIGGER_TOKENS, max_value=MAX_COMPRESSION_TRIGGER_TOKENS, ) or DEFAULT_DEEP_COMPRESS_TRIGGER_TOKENS if deep_trigger <= shallow_trigger: raise ValueError("深压缩触发上下文必须大于浅压缩触发上下文") def build_personalization_prompt( config: Optional[Dict[str, Any]], include_header: bool = True ) -> Optional[str]: """Generate the personalization prompt text based on config.""" if not config or not config.get("enabled"): return None lines = [] if include_header: lines.append("用户的个性化数据，请回答时务必参照这些信息") if config.get("self_identify"): lines.append(f"用户希望你自称：{config['self_identify']}") if config.get("user_name"): lines.append(f"用户希望你称呼为：{config['user_name']}") if config.get("profession"): lines.append(f"用户的职业是：{config['profession']}") if config.get("tone"): lines.append(f"用户希望你使用 {config['tone']} 的语气与TA交流") considerations: Iterable[str] = config.get("considerations") or [] considerations = [item for item in considerations if item] if considerations: lines.append("用户希望你在回答问题时必须考虑的信息是：") for idx, item in enumerate(considerations, 1): lines.append(f"{idx}. {item}") if len(lines) == (1 if include_header else 0): # 没有任何有效内容时不注入 return None return "\n".join(lines) def _sanitize_short_field(value: Optional[str]) -> str: if not value: return "" text = str(value).strip() if not text: return "" return text[:MAX_SHORT_FIELD_LENGTH] def _sanitize_considerations(value: Any) -> list: if not isinstance(value, list): return [] cleaned = [] for item in value: if not isinstance(item, str): continue text = item.strip() if not text: continue cleaned.append(text[:MAX_CONSIDERATION_LENGTH]) if len(cleaned) >= MAX_CONSIDERATION_ITEMS: break return cleaned def _sanitize_thinking_interval(value: Any) -> Optional[int]: if value is None or value == "": return None try: interval = int(value) except (TypeError, ValueError): return None interval = max(THINKING_INTERVAL_MIN, min(THINKING_INTERVAL_MAX, interval)) if interval == THINKING_FAST_INTERVAL: return None return interval def _sanitize_tool_categories(value: Any, allowed: set) -> list: if not isinstance(value, list): return [] result = [] for item in value: if not isinstance(item, str): continue candidate = item.strip() if not candidate or candidate not in allowed: continue if candidate not in result: result.append(candidate) return result def _sanitize_run_mode(value: Any) -> Optional[str]: if value is None: return None if isinstance(value, str): candidate = value.strip().lower() if candidate in ALLOWED_RUN_MODES: return candidate return None