JOJO
478d24c4dd
本次变更是对版本控制功能的大范围重构，核心目标：
- 默认使用浅备份（只追踪 write_file/edit_file 实际修改的文件），避免大工作区首次初始化耗时十几秒；
- 保留可选的完全备份（git write-tree），不污染 git log；
- 回溯时统一选择范围（仅对话 / 对话+工作区）和模式（覆盖 / 复制）；
- 修复复制对话、侧边栏同步、新建对话延迟、diff 统计与渲染等连锁问题。
后端改动：
- modules/shallow_versioning.py（新增）
- 实现 ShallowVersioningManager：track_edit、make_snapshot、rewind、list_snapshots。
- 基于 difflib.SequenceMatcher 计算真实 insertions/deletions，replace 场景正确显示 +N/-M。
- diff 详情使用结构化 add/remove/context 行，不再解析 unified_diff 文本。
- modules/versioning_manager.py
- 完全备份从 git commit 改为 git write-tree，字段 commit/parent_commit 改为 tree_hash/parent_tree_hash。
- 恢复改用 read-tree + checkout-index；create_checkpoint 支持 diff_summary 参数。
- get_checkpoint_detail 保留 API 字段名 last_commit/restored_commit 以兼容前端。
- core/main_terminal_parts/tools_execution.py
- write_file/edit_file 在执行前调用浅备份 track_edit，保留编辑前状态。
- server/chat_flow_task_main.py
- 消息轮次结束时根据 backup_mode 选择完全备份 create_checkpoint 或浅备份 make_snapshot。
- 浅备份模式下同时写入 ConversationVersioningManager 检查点行，让前端能统一展示。
- server/conversation.py
- restore/detail/list API 支持 shallow 模式；list/detail 对浅备份实时重新计算统计。
- 创建对话时读取 personalization 的 versioning_backup_mode，shallow 模式固定 tracking_mode 为 conversation_only。
- core/web_terminal.py
- _ensure_conversation_versioning_enabled 读取默认备份模式并写入 versioning meta。
- modules/file_manager/crud_mixin.py / replace_mixin.py
- write_file / edit_file 返回 original_file / new_file / replacement_details（真实旧行/新行）。
- modules/personalization_manager.py
- 新增 versioning_enabled_by_default、versioning_backup_mode（shallow/full）。
- utils/perf_log.py（新增）
- 创建新对话全链路性能日志，用于定位延迟瓶颈。
- utils/conversation_manager/{index,list_search,crud}_mixin.py
- 加性能日志；index_mixin 主动检查新增对话文件，修复侧边栏不同步。
前端改动：
- static/src/components/overlay/VersioningDialog.vue
- 去掉顶部管理范围下拉；底部新增回溯范围与回溯模式选择。
- diff 详情改为 div.diff-line 网格布局（marker + content），支持横向滚动，状态显示缩写 A/M/D。
- 修复 pre 标签空白黑行、容器被挤窄导致折行等问题。
- static/src/app/methods/versioning.ts
- 移除 mismatch 弹窗逻辑；支持 copy 模式；selectVersioningCheckpoint 增强错误捕获。
- static/src/components/personalization/PersonalizationDrawer.vue
- 「工作区与权限」新增版本控制默认开关与备份方式选择。
- static/src/stores/personalization.ts
- 新增 versioning_enabled_by_default、versioning_backup_mode 字段。
- static/src/components/chat/actions/ToolAction.vue / toolRenderers.ts
- edit_file / write_file 工具展示基于 result.details / result.new_file 的真实替换结果。
- static/src/app/state.ts / watchers.ts / App.vue
- 新增 versioningInitializingBackupToastId，完全备份初始化时显示 toast。
- static/src/app/methods/conversation/action.ts / message/send.ts
- host + 完全备份 + 默认开启时，创建新对话/从 /new 发送首条消息显示「正在初始化备份」toast。
验证：
- /Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/bin/python3.9 -m unittest test.test_server_refactor_smoke 通过。
- npm run build 通过。
- 宿主机与 docker 模式新建对话、编辑文件、回溯、复制对话均验证通过。
820 lines
34 KiB
Python
820 lines
34 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_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-直接注入文件全文
|
||
"silent_tool_disable": True, # 禁用工具时不向模型插入提示（默认开启）
|
||
"enhanced_tool_display": True, # 增强工具显示
|
||
"compact_message_display": "full", # 简略消息显示：full-完整原始内容 / brief-一行概要
|
||
"show_status_avatar": True, # 是否显示助手状态形象
|
||
"stacked_hide_borders": False, # 堆叠块隐藏边线
|
||
"show_git_status_bar": True, # 是否显示输入栏上方 Git 状态栏
|
||
"versioning_enabled_by_default": True, # 新对话是否默认开启版本控制
|
||
"versioning_backup_mode": "shallow", # 文件备份方式：shallow-浅备份（只备份AI编辑的文件）/ full-完全备份（整个工作区）
|
||
"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"
|
||
|
||
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))
|
||
|
||
# 助手状态形象显示开关
|
||
if "show_status_avatar" in data:
|
||
base["show_status_avatar"] = bool(data.get("show_status_avatar"))
|
||
else:
|
||
base["show_status_avatar"] = bool(base.get("show_status_avatar", True))
|
||
|
||
# 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 "versioning_enabled_by_default" in data:
|
||
base["versioning_enabled_by_default"] = bool(data.get("versioning_enabled_by_default"))
|
||
else:
|
||
base["versioning_enabled_by_default"] = bool(base.get("versioning_enabled_by_default", True))
|
||
|
||
# 文件备份方式
|
||
if "versioning_backup_mode" in data:
|
||
mode = str(data.get("versioning_backup_mode") or "shallow").strip().lower()
|
||
base["versioning_backup_mode"] = "full" if mode == "full" else "shallow"
|
||
else:
|
||
base["versioning_backup_mode"] = "shallow" if base.get("versioning_backup_mode") not in ("shallow", "full") else base["versioning_backup_mode"]
|
||
|
||
# 使用自定义称呼
|
||
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
|