"""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_default_model_key, get_registered_model_keys ALLOWED_RUN_MODES = {"fast", "thinking", "deep"} ALLOWED_PERMISSION_MODES = {"readonly", "approval", "auto_approval", "unrestricted"} ALLOWED_THEMES = {"classic", "light", "dark"} ALLOWED_COMMUNICATION_STYLES = {"default", "human_like", "auto"} ALLOWED_CONVERSATION_CONTINUITY = {"low", "medium", "high"} ALLOWED_GOAL_REVIEW_MODES = {"readonly", "active"} ALLOWED_GOAL_END_CONDITIONS = {"max_turns", "max_tokens"} GOAL_MAX_TURNS_MIN = 1 GOAL_MAX_TURNS_MAX = 100 GOAL_MAX_TURNS_DEFAULT = 5 GOAL_MAX_TOKENS_MIN = 1_000 GOAL_MAX_TOKENS_MAX = 100_000_000 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 RECENT_CONVERSATIONS_PROMPT_LIMIT_MIN = 1 RECENT_CONVERSATIONS_PROMPT_LIMIT_MAX = 30 RECENT_CONVERSATIONS_PROMPT_LIMIT_DEFAULT = 10 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 MIN_SHALLOW_KEEP_USER_TURN_TOOLS = 0 MAX_SHALLOW_KEEP_USER_TURN_TOOLS = 50 DEFAULT_SHALLOW_KEEP_USER_TURN_TOOLS = 3 DEFAULT_PERSONALIZATION_CONFIG: Dict[str, Any] = { "enabled": False, "communication_style": "default", # default / human_like / auto "conversation_continuity": "medium", # high / medium / low "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": "deep", "default_permission_mode": "unrestricted", "auto_generate_title": True, "recent_conversations_prompt_enabled": False, "recent_conversations_prompt_limit": RECENT_CONVERSATIONS_PROMPT_LIMIT_DEFAULT, "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": None, "image_compression": "original", # original / 1080p / 720p / 540p "auto_shallow_compress_enabled": False, "auto_deep_compress_enabled": True, "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, "shallow_compress_keep_user_turn_tools": None, # 保留最近N次用户输入后的工具不压缩（默认3） "deep_compress_trigger_tokens": None, "deep_compress_form": "file", # 深压缩形式：file-生成文件 / inject-直接注入文件全文 "deep_compress_behavior": "continue", # 手动压缩行为：continue-注入并触发请求 / wait-仅插入等待用户 "silent_tool_disable": True, # 禁用工具时不向模型插入提示（默认开启） "enhanced_tool_display": True, # 增强工具显示 "compact_message_display": "full", # 简略消息显示：full-完整原始内容 / brief-一行概要 "stacked_hide_borders": False, # 堆叠块隐藏边线 "show_git_status_bar": True, # 是否显示输入栏上方 Git 状态栏 "versioning_restore_mode": "overwrite", # 版本回溯模式固定为 overwrite "agents_md_auto_inject": False, # AGENTS.md 自动注入开关 "allow_root_file_creation": False, # 允许在根目录创建文件开关 "default_hide_workspace": False, # 默认隐藏工作区 "theme": "classic", # 主题配色: classic-经典/light-明亮/dark-暗黑 # 目标模式（Goal Mode） "goal_review_mode": "readonly", # readonly-仅读对话判断 / active-允许审核智能体跑只读命令取证 "goal_end_conditions": ["max_turns"], # 结束方式，可多选：max_turns / max_tokens "goal_max_turns": GOAL_MAX_TURNS_DEFAULT, # 最多自动续命轮数 "goal_max_tokens": None, # 累计(输入+输出)token 上限；None 表示不启用 } __all__ = [ "PERSONALIZATION_FILENAME", "DEFAULT_PERSONALIZATION_CONFIG", "TONE_PRESETS", "MAX_CONSIDERATION_ITEMS", "RECENT_CONVERSATIONS_PROMPT_LIMIT_MIN", "RECENT_CONVERSATIONS_PROMPT_LIMIT_MAX", "RECENT_CONVERSATIONS_PROMPT_LIMIT_DEFAULT", "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"])) # 交流风格: default / human_like / auto _comm_style = data.get("communication_style", base.get("communication_style", "default")) base["communication_style"] = _comm_style if _comm_style in ALLOWED_COMMUNICATION_STYLES else "default" _conversation_continuity = data.get( "conversation_continuity", base.get("conversation_continuity", "medium"), ) base["conversation_continuity"] = ( _conversation_continuity if _conversation_continuity in ALLOWED_CONVERSATION_CONTINUITY else "medium" ) base["auto_generate_title"] = bool(data.get("auto_generate_title", base["auto_generate_title"])) base["recent_conversations_prompt_enabled"] = bool( data.get("recent_conversations_prompt_enabled", base.get("recent_conversations_prompt_enabled", False)) ) base["recent_conversations_prompt_limit"] = ( _sanitize_optional_int( data.get("recent_conversations_prompt_limit", base.get("recent_conversations_prompt_limit")), min_value=RECENT_CONVERSATIONS_PROMPT_LIMIT_MIN, max_value=RECENT_CONVERSATIONS_PROMPT_LIMIT_MAX, ) or RECENT_CONVERSATIONS_PROMPT_LIMIT_DEFAULT ) 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: try: base["default_model"] = get_default_model_key(visible_only=True) except Exception: base["default_model"] = None # 图片压缩模式 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_keep_user_turn_tools" in data: base["shallow_compress_keep_user_turn_tools"] = _sanitize_optional_int( data.get("shallow_compress_keep_user_turn_tools"), min_value=MIN_SHALLOW_KEEP_USER_TURN_TOOLS, max_value=MAX_SHALLOW_KEEP_USER_TURN_TOOLS, ) else: base["shallow_compress_keep_user_turn_tools"] = _sanitize_optional_int( base.get("shallow_compress_keep_user_turn_tools"), min_value=MIN_SHALLOW_KEEP_USER_TURN_TOOLS, max_value=MAX_SHALLOW_KEEP_USER_TURN_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, ) # 深压缩形式：file（生成文件，引导语提示位置）/ inject（直接注入文件全文） deep_form = data.get("deep_compress_form", base.get("deep_compress_form")) if isinstance(deep_form, str) and deep_form.strip().lower() in ("file", "inject"): base["deep_compress_form"] = deep_form.strip().lower() else: base["deep_compress_form"] = "file" # 手动压缩行为：continue（注入并触发请求）/ wait（仅插入等待用户） deep_behavior = data.get("deep_compress_behavior", base.get("deep_compress_behavior")) if isinstance(deep_behavior, str) and deep_behavior.strip().lower() in ("continue", "wait"): base["deep_compress_behavior"] = deep_behavior.strip().lower() else: base["deep_compress_behavior"] = "continue" 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)) # 简略消息显示：full（完整原始内容）/ brief（一行概要） compact_msg = data.get("compact_message_display", base.get("compact_message_display")) if isinstance(compact_msg, str) and compact_msg.strip().lower() in ("full", "brief"): base["compact_message_display"] = compact_msg.strip().lower() else: base["compact_message_display"] = "full" # 堆叠块隐藏边线 if "stacked_hide_borders" in data: base["stacked_hide_borders"] = bool(data.get("stacked_hide_borders")) else: base["stacked_hide_borders"] = bool(base.get("stacked_hide_borders", False)) # Git 状态栏显示开关 if "show_git_status_bar" in data: base["show_git_status_bar"] = bool(data.get("show_git_status_bar")) else: base["show_git_status_bar"] = bool(base.get("show_git_status_bar", 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)) # 默认隐藏工作区 if "default_hide_workspace" in data: base["default_hide_workspace"] = bool(data.get("default_hide_workspace")) else: base["default_hide_workspace"] = bool(base.get("default_hide_workspace", False)) # 主题配色 theme_value = data.get("theme", base.get("theme")) if isinstance(theme_value, str) and theme_value in ALLOWED_THEMES: base["theme"] = theme_value elif base.get("theme") not in ALLOWED_THEMES: base["theme"] = "classic" # 目标模式：审核模式 goal_review_mode = data.get("goal_review_mode", base.get("goal_review_mode")) if isinstance(goal_review_mode, str) and goal_review_mode in ALLOWED_GOAL_REVIEW_MODES: base["goal_review_mode"] = goal_review_mode elif base.get("goal_review_mode") not in ALLOWED_GOAL_REVIEW_MODES: base["goal_review_mode"] = "readonly" # 目标模式：结束方式（多选） if "goal_end_conditions" in data: base["goal_end_conditions"] = _sanitize_goal_end_conditions(data.get("goal_end_conditions")) else: base["goal_end_conditions"] = _sanitize_goal_end_conditions(base.get("goal_end_conditions")) # 目标模式：最大轮数 if "goal_max_turns" in data: base["goal_max_turns"] = _sanitize_optional_int( data.get("goal_max_turns"), min_value=GOAL_MAX_TURNS_MIN, max_value=GOAL_MAX_TURNS_MAX ) or GOAL_MAX_TURNS_DEFAULT else: base["goal_max_turns"] = _sanitize_optional_int( base.get("goal_max_turns"), min_value=GOAL_MAX_TURNS_MIN, max_value=GOAL_MAX_TURNS_MAX ) or GOAL_MAX_TURNS_DEFAULT # 目标模式：累计 token 上限（可空） if "goal_max_tokens" in data: base["goal_max_tokens"] = _sanitize_optional_int( data.get("goal_max_tokens"), min_value=GOAL_MAX_TOKENS_MIN, max_value=GOAL_MAX_TOKENS_MAX ) else: base["goal_max_tokens"] = _sanitize_optional_int( base.get("goal_max_tokens"), min_value=GOAL_MAX_TOKENS_MIN, max_value=GOAL_MAX_TOKENS_MAX ) return base def _sanitize_goal_end_conditions(value: Any) -> list: """清洗目标模式结束方式列表，保证至少包含 max_turns。""" cleaned: list = [] if isinstance(value, list): for item in value: if isinstance(item, str) and item in ALLOWED_GOAL_END_CONDITIONS and item not in cleaned: cleaned.append(item) if "max_turns" not in cleaned: cleaned.insert(0, "max_turns") return cleaned 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_keep_user_turn = _sanitize_optional_int( data.get("shallow_compress_keep_user_turn_tools"), min_value=MIN_SHALLOW_KEEP_USER_TURN_TOOLS, max_value=MAX_SHALLOW_KEEP_USER_TURN_TOOLS, ) if shallow_keep_user_turn is None: shallow_keep_user_turn = DEFAULT_SHALLOW_KEEP_USER_TURN_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_keep_user_turn_tools": shallow_keep_user_turn, "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 _load_human_like_prompt() -> str: """加载拟人化风格提示词文件。""" prompt = _load_prompt_file("human_like_style.txt") if prompt: return prompt # 默认提示词 return ( "不要「像AI一样说话」，而是尽可能的模仿真人的说话方式或是模仿用户输入的说话方式。\n" "尽可能不要用markdown格式，尤其是：\n" "- ###、##、#等一二三级标题\n" "- 等列表形式\n" "- 不必要的** **加粗\n" "- 刻意的分点分层次的回答\n" "用户如果不输入emoji，就不要使用emoji回答。" ) def _load_auto_style_prompt() -> str: """加载自动交流风格提示词文件。""" return _load_prompt_file("auto_style.txt") def _load_prompt_file(filename: str) -> str: from config.paths import PROMPTS_DIR try: prompt_path = Path(PROMPTS_DIR) / filename if prompt_path.exists(): return prompt_path.read_text(encoding="utf-8").strip() except Exception: pass return "" 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}") conversation_continuity = str(config.get("conversation_continuity") or "medium").strip().lower() if conversation_continuity == "high": lines.append("\n【对话连续性：高】") lines.append( "当前对话应积极延续用户长期背景。回答时优先参考用户过往对话、长期记忆和项目背景；" "遇到可能相关的问题，应主动搜索/回顾历史对话；发现值得长期保留的偏好、项目事实、" "稳定经验或用户明确要求记住的内容，应主动调用记忆工具记录。" ) elif conversation_continuity == "low": lines.append("\n【对话连续性：低】") lines.append( "当前对话应尽量独立。不要主动回顾之前的对话，不要主动调用记忆或历史对话工具，" "不要主动参考用户历史记忆来回答；只有用户明确要求、当前任务必须依赖历史，" "或安全/连续性需要时，才可搜索历史或读取记忆。" ) else: lines.append("\n【对话连续性：中】") lines.append( "当前对话优先，但可以在适当时候参考历史。若用户问题明显与过往偏好、项目背景或未完成事项有关，" "可以调用记忆或历史对话工具；不要为了普通问题主动翻历史。" ) communication_style = str(config.get("communication_style") or "default").strip().lower() if communication_style == "auto": auto_style_prompt = _load_auto_style_prompt() if auto_style_prompt: lines.append("\n【交流风格：自动】") lines.append(auto_style_prompt) elif communication_style == "human_like": human_like_prompt = _load_human_like_prompt() if human_like_prompt: lines.append("\n【交流风格要求】") lines.append(human_like_prompt) 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 = [] dynamic_prefixes = ("mcp_server__",) for item in value: if not isinstance(item, str): continue candidate = item.strip() if not candidate: continue dynamic_ok = candidate == "custom" or any(candidate.startswith(prefix) for prefix in dynamic_prefixes) if candidate not in allowed and not dynamic_ok: 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