JOJO
8a1e17b9a0
Introduce workspace-level goal state persistence, goal prompt injection, and after-turn review handling so an active task can continue until the configured completion conditions are met. Add a dedicated goal review agent with readonly and active evidence modes, configurable model settings, review prompt, token/turn boundaries, idle-no-tool protection, and progress/completed/stopped events. Wire goal_mode through task creation, task restoration, compression handoff, runtime user messages, API message sanitization, and tool-call ordering so goal continuations survive long-running tasks and deep compression. Add Vue UI for arming goal mode from the quick menu, showing running/completed banners, displaying progress metrics, restoring running goal state, and exposing personalization settings for review mode and stop limits. Include goal mode research notes and default goal review configuration.
710 lines
29 KiB
Python
710 lines
29 KiB
Python
"""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_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
|
||
"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,
|
||
"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": 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,
|
||
"shallow_compress_keep_user_turn_tools": None, # 保留最近N次用户输入后的工具不压缩（默认3）
|
||
"deep_compress_trigger_tokens": None,
|
||
"silent_tool_disable": True, # 禁用工具时不向模型插入提示（默认开启）
|
||
"enhanced_tool_display": True, # 增强工具显示
|
||
"versioning_restore_mode": "overwrite", # 版本回溯模式固定为 overwrite
|
||
"agents_md_auto_inject": False, # AGENTS.md 自动注入开关
|
||
"allow_root_file_creation": 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
|
||
_comm_style = data.get("communication_style", base.get("communication_style", "default"))
|
||
base["communication_style"] = "human_like" if _comm_style == "human_like" else "default"
|
||
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,
|
||
)
|
||
|
||
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))
|
||
|
||
# 主题配色
|
||
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:
|
||
"""加载拟人化风格提示词文件。"""
|
||
from config.paths import PROMPTS_DIR
|
||
try:
|
||
prompt_path = Path(PROMPTS_DIR) / "human_like_style.txt"
|
||
if prompt_path.exists():
|
||
return prompt_path.read_text(encoding="utf-8").strip()
|
||
except Exception:
|
||
pass
|
||
# 默认提示词
|
||
return (
|
||
"不要「像AI一样说话」，而是尽可能的模仿真人的说话方式或是模仿用户输入的说话方式。\n"
|
||
"尽可能不要用markdown格式，尤其是：\n"
|
||
"- ###、##、#等一二三级标题\n"
|
||
"- 等列表形式\n"
|
||
"- 不必要的** **加粗\n"
|
||
"- 刻意的分点分层次的回答\n"
|
||
"用户如果不输入emoji，就不要使用emoji回答。"
|
||
)
|
||
|
||
|
||
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 config.get("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
|