251 lines
12 KiB
Python
251 lines
12 KiB
Python
import asyncio
|
||
import json
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional, Set
|
||
|
||
try:
|
||
from config import (
|
||
OUTPUT_FORMATS, DATA_DIR, PROMPTS_DIR, NEED_CONFIRMATION,
|
||
MAX_TERMINALS, TERMINAL_BUFFER_SIZE, TERMINAL_DISPLAY_SIZE,
|
||
MAX_READ_FILE_CHARS, READ_TOOL_DEFAULT_MAX_CHARS,
|
||
READ_TOOL_DEFAULT_CONTEXT_BEFORE, READ_TOOL_DEFAULT_CONTEXT_AFTER,
|
||
READ_TOOL_MAX_CONTEXT_BEFORE, READ_TOOL_MAX_CONTEXT_AFTER,
|
||
READ_TOOL_DEFAULT_MAX_MATCHES, READ_TOOL_MAX_MATCHES,
|
||
READ_TOOL_MAX_FILE_SIZE,
|
||
TERMINAL_SANDBOX_MOUNT_PATH,
|
||
TERMINAL_SANDBOX_MODE,
|
||
TERMINAL_SANDBOX_CPUS,
|
||
TERMINAL_SANDBOX_MEMORY,
|
||
PROJECT_MAX_STORAGE_MB,
|
||
CUSTOM_TOOLS_ENABLED,
|
||
)
|
||
except ImportError:
|
||
import sys
|
||
project_root = Path(__file__).resolve().parents[2]
|
||
if str(project_root) not in sys.path:
|
||
sys.path.insert(0, str(project_root))
|
||
from config import (
|
||
OUTPUT_FORMATS, DATA_DIR, PROMPTS_DIR, NEED_CONFIRMATION,
|
||
MAX_TERMINALS, TERMINAL_BUFFER_SIZE, TERMINAL_DISPLAY_SIZE,
|
||
MAX_READ_FILE_CHARS, READ_TOOL_DEFAULT_MAX_CHARS,
|
||
READ_TOOL_DEFAULT_CONTEXT_BEFORE, READ_TOOL_DEFAULT_CONTEXT_AFTER,
|
||
READ_TOOL_MAX_CONTEXT_BEFORE, READ_TOOL_MAX_CONTEXT_AFTER,
|
||
READ_TOOL_DEFAULT_MAX_MATCHES, READ_TOOL_MAX_MATCHES,
|
||
READ_TOOL_MAX_FILE_SIZE,
|
||
TERMINAL_SANDBOX_MOUNT_PATH,
|
||
TERMINAL_SANDBOX_MODE,
|
||
TERMINAL_SANDBOX_CPUS,
|
||
TERMINAL_SANDBOX_MEMORY,
|
||
PROJECT_MAX_STORAGE_MB,
|
||
CUSTOM_TOOLS_ENABLED,
|
||
)
|
||
|
||
from modules.file_manager import FileManager
|
||
from modules.search_engine import SearchEngine
|
||
from modules.terminal_ops import TerminalOperator
|
||
from modules.memory_manager import MemoryManager
|
||
from modules.terminal_manager import TerminalManager
|
||
from modules.todo_manager import TodoManager
|
||
from modules.sub_agent_manager import SubAgentManager
|
||
from modules.webpage_extractor import extract_webpage_content, tavily_extract
|
||
from modules.ocr_client import OCRClient
|
||
from modules.easter_egg_manager import EasterEggManager
|
||
from modules.personalization_manager import (
|
||
load_personalization_config,
|
||
build_personalization_prompt,
|
||
)
|
||
from modules.skills_manager import (
|
||
get_skills_catalog,
|
||
build_skills_list,
|
||
merge_enabled_skills,
|
||
build_skills_prompt,
|
||
)
|
||
from modules.custom_tool_registry import CustomToolRegistry, build_default_tool_category
|
||
from modules.custom_tool_executor import CustomToolExecutor
|
||
|
||
try:
|
||
from config.limits import THINKING_FAST_INTERVAL
|
||
except ImportError:
|
||
THINKING_FAST_INTERVAL = 10
|
||
|
||
from modules.container_monitor import collect_stats, inspect_state
|
||
from core.tool_config import TOOL_CATEGORIES
|
||
from utils.api_client import DeepSeekClient
|
||
from utils.context_manager import ContextManager
|
||
from utils.tool_result_formatter import format_tool_result_for_context
|
||
from utils.logger import setup_logger
|
||
from config.model_profiles import (
|
||
get_model_profile,
|
||
get_model_prompt_replacements,
|
||
get_model_context_window,
|
||
)
|
||
|
||
logger = setup_logger(__name__)
|
||
DISABLE_LENGTH_CHECK = True
|
||
|
||
class MainTerminalToolsPolicyMixin:
|
||
def apply_personalization_preferences(self, config: Optional[Dict[str, Any]] = None):
|
||
"""Apply persisted personalization settings that affect runtime behavior."""
|
||
try:
|
||
effective_config = config or load_personalization_config(self.data_dir)
|
||
except Exception:
|
||
effective_config = {}
|
||
|
||
# 工具意图开关
|
||
self.tool_intent_enabled = bool(effective_config.get("tool_intent_enabled"))
|
||
# Skill 强约束开关
|
||
self.skill_strict_terminal_enabled = bool(effective_config.get("skill_strict_terminal_enabled", False))
|
||
self.skill_strict_sub_agent_enabled = bool(effective_config.get("skill_strict_sub_agent_enabled", False))
|
||
|
||
# 解析当前启用的 skills（用于强约束“仅对已启用 skill 生效”）
|
||
try:
|
||
skills_catalog = get_skills_catalog()
|
||
enabled_skills = merge_enabled_skills(
|
||
effective_config.get("enabled_skills") if isinstance(effective_config, dict) else None,
|
||
skills_catalog,
|
||
effective_config.get("skills_catalog_snapshot") if isinstance(effective_config, dict) else None,
|
||
)
|
||
self.enabled_skill_ids = set(enabled_skills or [])
|
||
except Exception:
|
||
self.enabled_skill_ids = set()
|
||
|
||
interval = effective_config.get("thinking_interval")
|
||
if isinstance(interval, int) and interval > 0:
|
||
self.thinking_fast_interval = interval
|
||
else:
|
||
self.thinking_fast_interval = THINKING_FAST_INTERVAL
|
||
|
||
disabled_categories = []
|
||
raw_disabled = effective_config.get("disabled_tool_categories")
|
||
if isinstance(raw_disabled, list):
|
||
disabled_categories = [
|
||
key for key in raw_disabled
|
||
if isinstance(key, str) and key in self.tool_categories_map
|
||
]
|
||
self.default_disabled_tool_categories = disabled_categories
|
||
|
||
# 图片压缩模式传递给上下文
|
||
img_mode = effective_config.get("image_compression")
|
||
if isinstance(img_mode, str):
|
||
self.context_manager.image_compression_mode = img_mode
|
||
|
||
# Reset category states to defaults before applying overrides
|
||
for key, category in self.tool_categories_map.items():
|
||
self.tool_category_states[key] = False if key in disabled_categories else category.default_enabled
|
||
self._refresh_disabled_tools()
|
||
|
||
# 默认模型偏好（优先应用，再处理运行模式）
|
||
preferred_model = effective_config.get("default_model")
|
||
if isinstance(preferred_model, str) and preferred_model != self.model_key:
|
||
try:
|
||
self.set_model(preferred_model)
|
||
except Exception as exc:
|
||
logger.warning("忽略无效默认模型: %s (%s)", preferred_model, exc)
|
||
|
||
preferred_mode = effective_config.get("default_run_mode")
|
||
if isinstance(preferred_mode, str):
|
||
normalized_mode = preferred_mode.strip().lower()
|
||
if normalized_mode in {"fast", "thinking", "deep"} and normalized_mode != self.run_mode:
|
||
try:
|
||
self.set_run_mode(normalized_mode)
|
||
except ValueError:
|
||
logger.warning("忽略无效默认运行模式: %s", preferred_mode)
|
||
|
||
# 静默禁用工具提示
|
||
self.silent_tool_disable = bool(effective_config.get("silent_tool_disable"))
|
||
|
||
def set_tool_category_enabled(self, category: str, enabled: bool) -> None:
|
||
"""设置工具类别的启用状态 / Toggle tool category enablement."""
|
||
categories = self.tool_categories_map
|
||
if category not in categories:
|
||
raise ValueError(f"未知的工具类别: {category}")
|
||
forced = self.admin_forced_category_states.get(category)
|
||
if isinstance(forced, bool) and forced != enabled:
|
||
raise ValueError("该类别被管理员强制为启用/禁用，无法修改")
|
||
self.tool_category_states[category] = bool(enabled)
|
||
self._refresh_disabled_tools()
|
||
|
||
def set_admin_policy(
|
||
self,
|
||
categories: Optional[Dict[str, "ToolCategory"]] = None,
|
||
forced_category_states: Optional[Dict[str, Optional[bool]]] = None,
|
||
disabled_models: Optional[List[str]] = None,
|
||
) -> None:
|
||
"""应用管理员策略（工具分类、强制开关、模型禁用）。"""
|
||
if categories:
|
||
self.tool_categories_map = dict(categories)
|
||
# 保证自定义工具分类存在（仅当功能启用）
|
||
if self.custom_tools_enabled and "custom" not in self.tool_categories_map:
|
||
self.tool_categories_map["custom"] = type(next(iter(TOOL_CATEGORIES.values())))(
|
||
label="自定义工具",
|
||
tools=[],
|
||
default_enabled=True,
|
||
silent_when_disabled=False,
|
||
)
|
||
# 重新构建启用状态映射，保留已有值
|
||
new_states: Dict[str, bool] = {}
|
||
for key, cat in self.tool_categories_map.items():
|
||
if key in self.tool_category_states:
|
||
new_states[key] = self.tool_category_states[key]
|
||
else:
|
||
new_states[key] = cat.default_enabled
|
||
self.tool_category_states = new_states
|
||
# 清理已被移除的类别
|
||
for removed in list(self.tool_category_states.keys()):
|
||
if removed not in self.tool_categories_map:
|
||
self.tool_category_states.pop(removed, None)
|
||
|
||
self.admin_forced_category_states = forced_category_states or {}
|
||
self.admin_disabled_models = disabled_models or []
|
||
self._refresh_disabled_tools()
|
||
|
||
def get_tool_settings_snapshot(self) -> List[Dict[str, object]]:
|
||
"""获取工具类别状态快照 / Return tool category states snapshot."""
|
||
snapshot: List[Dict[str, object]] = []
|
||
categories = self.tool_categories_map
|
||
for key, category in categories.items():
|
||
forced = self.admin_forced_category_states.get(key)
|
||
enabled = self.tool_category_states.get(key, category.default_enabled)
|
||
if isinstance(forced, bool):
|
||
enabled = forced
|
||
snapshot.append({
|
||
"id": key,
|
||
"label": category.label,
|
||
"enabled": enabled,
|
||
"tools": list(category.tools),
|
||
"locked": isinstance(forced, bool),
|
||
"locked_state": forced if isinstance(forced, bool) else None,
|
||
})
|
||
return snapshot
|
||
|
||
def _refresh_disabled_tools(self) -> None:
|
||
"""刷新禁用工具列表 / Refresh disabled tool set."""
|
||
disabled: Set[str] = set()
|
||
notice: Set[str] = set()
|
||
categories = self.tool_categories_map
|
||
for key, category in categories.items():
|
||
state = self.tool_category_states.get(key, category.default_enabled)
|
||
forced = self.admin_forced_category_states.get(key)
|
||
if isinstance(forced, bool):
|
||
state = forced
|
||
if not state:
|
||
disabled.update(category.tools)
|
||
if not getattr(category, "silent_when_disabled", False):
|
||
notice.update(category.tools)
|
||
self.disabled_tools = disabled
|
||
self.disabled_notice_tools = notice
|
||
|
||
def _format_disabled_tool_notice(self) -> Optional[str]:
|
||
"""生成禁用工具提示信息 / Format disabled tool notice."""
|
||
if getattr(self, "silent_tool_disable", False):
|
||
return None
|
||
if not self.disabled_notice_tools:
|
||
return None
|
||
|
||
lines = ["=== 工具可用性提醒 ==="]
|
||
for tool_name in sorted(self.disabled_notice_tools):
|
||
lines.append(f"{tool_name}：已被用户禁用")
|
||
lines.append("=== 提示结束 ===")
|
||
return "\n".join(lines)
|