agent-Specialization/core/main_terminal_parts/tools_definition.py

790 lines
45 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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 MainTerminalToolsDefinitionMixin:
def _inject_intent(self, properties: Dict[str, Any]) -> Dict[str, Any]:
"""在工具参数中注入 intent简短意图说明仅当开关启用时。
字段含义要求模型用不超过15个中文字符对即将执行的动作做简要说明供前端展示。
"""
if not self.tool_intent_enabled:
return properties
if not isinstance(properties, dict):
return properties
intent_field = {
"intent": {
"type": "string",
"description": "用不超过15个字向用户说明你要做什么例如等待下载完成/创建日志文件"
}
}
# 将 intent 放在最前面以提高模型关注度
return {**intent_field, **properties}
def _apply_intent_to_tools(self, tools: List[Dict]) -> List[Dict]:
"""遍历工具列表,为缺少 intent 的工具补充字段(开关启用时生效)。"""
if not self.tool_intent_enabled:
return tools
intent_field = {
"intent": {
"type": "string",
"description": "用不超过15个字向用户说明你要做什么例如等待下载完成/创建日志文件/搜索最新新闻"
}
}
for tool in tools:
func = tool.get("function") or {}
params = func.get("parameters") or {}
if not isinstance(params, dict):
continue
if params.get("type") != "object":
continue
props = params.get("properties")
if not isinstance(props, dict):
continue
# 补充 intent 属性
if "intent" not in props:
params["properties"] = {**intent_field, **props}
# 将 intent 加入必填
required_list = params.get("required")
if isinstance(required_list, list):
if "intent" not in required_list:
required_list.insert(0, "intent")
params["required"] = required_list
else:
params["required"] = ["intent"]
return tools
def _build_custom_tools(self) -> List[Dict]:
if not (self.custom_tools_enabled and getattr(self, "user_role", "user") == "admin"):
return []
try:
definitions = self.custom_tool_registry.reload()
except Exception:
definitions = self.custom_tool_registry.list_tools()
if not definitions:
# 更新分类为空列表,避免旧缓存
if "custom" in self.tool_categories_map:
self.tool_categories_map["custom"].tools = []
return []
tools: List[Dict] = []
tool_ids: List[str] = []
for item in definitions:
tool_id = item.get("id")
if not tool_id:
continue
if item.get("invalid_id"):
# 跳过不合法的工具 ID避免供应商严格校验时报错
continue
tool_ids.append(tool_id)
params = item.get("parameters") or {"type": "object", "properties": {}}
if isinstance(params, dict) and params.get("type") != "object":
params = {"type": "object", "properties": {}}
required = item.get("required")
if isinstance(required, list):
params = dict(params)
params["required"] = required
tools.append({
"type": "function",
"function": {
"name": tool_id,
"description": item.get("description") or f"自定义工具: {tool_id}",
"parameters": params
}
})
# 覆盖 custom 分类的工具列表
if "custom" in self.tool_categories_map:
self.tool_categories_map["custom"].tools = tool_ids
return tools
def define_tools(self) -> List[Dict]:
"""定义可用工具(添加确认工具)"""
current_time = datetime.now().strftime("%Y-%m-%d %H")
tools = [
{
"type": "function",
"function": {
"name": "sleep",
"description": "等待指定的秒数,用于短暂延迟/节奏控制(例如让终端产生更多输出、或在两次快照之间留出间隔)。命令是否完成必须用 terminal_snapshot 确认;需要强制超时终止请使用 run_command。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"seconds": {
"type": "number",
"description": "等待的秒数可以是小数如0.2秒。建议范围0.1-10秒"
},
"reason": {
"type": "string",
"description": "等待的原因说明(可选)"
}
}),
"required": ["seconds"]
}
}
},
{
"type": "function",
"function": {
"name": "create_file",
"description": "创建新文件(仅创建空文件,正文请使用 write_file 或 edit_file 写入/替换)",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"path": {"type": "string", "description": "文件路径"},
"file_type": {"type": "string", "enum": ["txt", "py", "md"], "description": "文件类型"},
"annotation": {"type": "string", "description": "文件备注"}
}),
"required": ["path", "file_type", "annotation"]
}
}
},
{
"type": "function",
"function": {
"name": "write_file",
"description": "将内容写入本地文件系统append 为 False 时覆盖原文件True 时追加到末尾。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"file_path": {
"type": "string",
"description": "要写入的相对路径"
},
"content": {
"type": "string",
"description": "要写入文件的内容"
},
"append": {
"type": "boolean",
"description": "是否追加到文件而不是覆盖它",
"default": False
}
}),
"required": ["file_path", "content"]
}
}
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "读取/搜索/抽取 UTF-8 文本文件内容。通过 type 参数选择 read阅读、search搜索、extract具体行段支持限制返回字符数。若文件非 UTF-8 或过大,请改用 run_python。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"path": {"type": "string", "description": "文件路径"},
"type": {
"type": "string",
"enum": ["read", "search", "extract"],
"description": "读取模式read=阅读、search=搜索、extract=按行抽取"
},
"max_chars": {
"type": "integer",
"description": "返回内容的最大字符数,默认与 config 一致"
},
"start_line": {
"type": "integer",
"description": "[read] 可选的起始行号1开始"
},
"end_line": {
"type": "integer",
"description": "[read] 可选的结束行号(>=start_line"
},
"query": {
"type": "string",
"description": "[search] 搜索关键词"
},
"max_matches": {
"type": "integer",
"description": "[search] 最多返回多少条命中默认5最大50"
},
"context_before": {
"type": "integer",
"description": "[search] 命中行向上追加的行数默认1最大3"
},
"context_after": {
"type": "integer",
"description": "[search] 命中行向下追加的行数默认1最大5"
},
"case_sensitive": {
"type": "boolean",
"description": "[search] 是否区分大小写,默认 false"
},
"segments": {
"type": "array",
"description": "[extract] 需要抽取的行区间",
"items": {
"type": "object",
"properties": {
"label": {
"type": "string",
"description": "该片段的标签(可选)"
},
"start_line": {
"type": "integer",
"description": "起始行号(>=1"
},
"end_line": {
"type": "integer",
"description": "结束行号(>=start_line"
}
},
"required": ["start_line", "end_line"]
},
"minItems": 1
}
}),
"required": ["path", "type"]
}
}
},
{
"type": "function",
"function": {
"name": "edit_file",
"description": "在文件中执行精确的字符串替换;建议先使用 read_file 获取最新内容以确保精确匹配。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"file_path": {
"type": "string",
"description": "要修改文件的相对路径"
},
"old_string": {
"type": "string",
"description": "要替换的文本(需与文件内容精确匹配,保留缩进)"
},
"new_string": {
"type": "string",
"description": "用于替换的新文本(必须不同于 old_string"
}
}),
"required": ["file_path", "old_string", "new_string"]
}
}
},
{
"type": "function",
"function": {
"name": "vlm_analyze",
"description": "使用大参数视觉语言模型Qwen3.5)理解图片:文字、物体、布局、表格等,仅支持本地路径。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"path": {"type": "string", "description": "项目内的图片相对路径"},
"prompt": {"type": "string", "description": "传递给 VLM 的中文提示词,如“请总结这张图的内容”“表格的总金额是多少”“图中是什么车?”。"}
}),
"required": ["path", "prompt"]
}
}
},
{
"type": "function",
"function": {
"name": "delete_file",
"description": "删除文件",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"path": {"type": "string", "description": "文件路径"}
}),
"required": ["path"]
}
}
},
{
"type": "function",
"function": {
"name": "rename_file",
"description": "重命名文件",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"old_path": {"type": "string", "description": "原文件路径"},
"new_path": {"type": "string", "description": "新文件路径"}
}),
"required": ["old_path", "new_path"]
}
}
},
{
"type": "function",
"function": {
"name": "create_folder",
"description": "创建文件夹",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"path": {"type": "string", "description": "文件夹路径"}
}),
"required": ["path"]
}
}
},
{
"type": "function",
"function": {
"name": "terminal_session",
"description": "管理持久化终端会话,可打开、关闭、列出或切换终端。请在授权工作区内执行命令,禁止启动需要完整 TTY 的程序python REPL、vim、top 等)。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"action": {
"type": "string",
"enum": ["open", "close", "list", "reset"],
"description": "操作类型open-打开新终端close-关闭终端list-列出所有终端reset-重置终端"
},
"session_name": {
"type": "string",
"description": "终端会话名称open、close、reset时需要"
},
"working_dir": {
"type": "string",
"description": "工作目录相对于项目路径open时可选"
}
}),
"required": ["action"]
}
}
},
{
"type": "function",
"function": {
"name": "terminal_input",
"description": "向指定终端发送命令或输入。禁止启动会占用终端界面的程序python/node/nano/vim 等);如遇卡死请结合 terminal_snapshot 并使用 terminal_session 的 reset 恢复。timeout 必填传入数字最大300表示本次等待输出的时长不会封装命令、不会强杀进程在等待窗口内若检测到命令已完成会提前返回否则在超时后返回已产生的输出并保持命令继续运行。需要强制超时终止请使用 run_command。\n若不确定上一条命令是否结束,先用 terminal_snapshot 确认后再继续输入。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"command": {
"type": "string",
"description": "要执行的命令或发送的输入"
},
"session_name": {
"type": "string",
"description": "目标终端会话名称(必填)"
},
"timeout": {
"type": "number",
"description": "等待输出的最长秒数必填最大300不会封装命令、不会中断进程"
}
}),
"required": ["command", "timeout", "session_name"]
}
}
},
{
"type": "function",
"function": {
"name": "terminal_snapshot",
"description": "获取指定终端最近的输出快照用于判断当前状态。默认返回末尾的50行可通过参数调整。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"session_name": {
"type": "string",
"description": "目标终端会话名称(可选,默认活动终端)"
},
"lines": {
"type": "integer",
"description": "返回的最大行数(可选)"
},
"max_chars": {
"type": "integer",
"description": "返回的最大字符数(可选)"
}
})
}
}
},
{
"type": "function",
"function": {
"name": "web_search",
"description": f"当现有资料不足时搜索外部信息(当前时间 {current_time})。调用前说明目的,精准撰写 query并合理设置时间/主题参数;避免重复或无意义的搜索。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"query": {
"type": "string",
"description": "搜索查询内容(不要包含日期或时间范围)"
},
"max_results": {
"type": "integer",
"description": "最大结果数,可选"
},
"topic": {
"type": "string",
"description": "搜索主题可选值general默认/news/finance"
},
"time_range": {
"type": "string",
"description": "相对时间范围,可选 day/week/month/year支持缩写 d/w/m/y与 days 和 start_date/end_date 互斥"
},
"days": {
"type": "integer",
"description": "最近 N 天,仅当 topic=news 时可用;与 time_range、start_date/end_date 互斥"
},
"start_date": {
"type": "string",
"description": "开始日期YYYY-MM-DD必须与 end_date 同时提供,与 time_range、days 互斥"
},
"end_date": {
"type": "string",
"description": "结束日期YYYY-MM-DD必须与 start_date 同时提供,与 time_range、days 互斥"
},
"country": {
"type": "string",
"description": "国家过滤,仅 topic=general 可用,使用英文小写国名"
},
"include_domains": {
"type": "array",
"description": "仅包含这些域名可选最多300个",
"items": {
"type": "string"
}
}
}),
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "extract_webpage",
"description": "在 web_search 结果不够详细时提取网页正文。调用前说明用途,注意提取内容会消耗大量 token超过80000字符将被拒绝。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"url": {"type": "string", "description": "要提取内容的网页URL"}
}),
"required": ["url"]
}
}
},
{
"type": "function",
"function": {
"name": "save_webpage",
"description": "提取网页内容并保存为纯文本文件,适合需要长期留存的长文档。请提供网址与目标路径(含 .txt 后缀),落地后请通过终端命令查看。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"url": {"type": "string", "description": "要保存的网页URL"},
"target_path": {"type": "string", "description": "保存位置,包含文件名,相对于项目根目录"}
}),
"required": ["url", "target_path"]
}
}
},
{
"type": "function",
"function": {
"name": "run_python",
"description": "执行一次性 Python 脚本,可用于处理二进制或非 UTF-8 文件(如 Excel、Word、PDF、图片或进行数据分析与验证。必须提供 timeout最长60秒一旦超时脚本会被打断且无法继续执行需要重新运行并返回已捕获输出。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"code": {"type": "string", "description": "Python代码"},
"timeout": {
"type": "number",
"description": "超时时长必填最大60"
}
}),
"required": ["code", "timeout"]
}
}
},
{
"type": "function",
"function": {
"name": "run_command",
"description": "执行一次性终端命令适合查看文件信息file/ls/stat/iconv 等)、转换编码或调用 CLI 工具。禁止启动交互式程序;对已聚焦文件仅允许使用 grep -n 等定位命令。必须提供 timeout最长30秒一旦超时命令**一定会被打断**且无法继续执行需要重新运行并返回已捕获输出输出超过10000字符将被截断或拒绝。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"command": {"type": "string", "description": "终端命令"},
"timeout": {
"type": "number",
"description": "超时时长必填最大30"
}
}),
"required": ["command", "timeout"]
}
}
},
{
"type": "function",
"function": {
"name": "update_memory",
"description": "按条目更新记忆列表自动编号。append 追加新条目replace 用序号替换delete 用序号删除。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"memory_type": {"type": "string", "enum": ["main", "task"], "description": "记忆类型"},
"content": {"type": "string", "description": "条目内容。append/replace 时必填"},
"operation": {"type": "string", "enum": ["append", "replace", "delete"], "description": "操作类型"},
"index": {"type": "integer", "description": "要替换/删除的序号从1开始"}
}),
"required": ["memory_type", "operation"]
}
}
},
{
"type": "function",
"function": {
"name": "todo_create",
"description": "创建待办列表,最多 8 条任务;若已有列表将被覆盖。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"overview": {"type": "string", "description": "一句话概述待办清单要完成的目标50 字以内。"},
"tasks": {
"type": "array",
"description": "任务列表1~8 条,每条写清“动词+对象+目标”。",
"items": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "单个任务描述,写成可执行的步骤"}
},
"required": ["title"]
},
"minItems": 1,
"maxItems": 8
}
}),
"required": ["overview", "tasks"]
}
}
},
{
"type": "function",
"function": {
"name": "todo_update_task",
"description": "批量勾选或取消任务(支持单个或多个任务);全部勾选时提示所有任务已完成。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"task_index": {"type": "integer", "description": "任务序号1-8兼容旧参数"},
"task_indices": {
"type": "array",
"items": {"type": "integer"},
"minItems": 1,
"maxItems": 8,
"description": "要更新的任务序号列表1-8可一次勾选多个"
},
"completed": {"type": "boolean", "description": "true=打勾false=取消"}
}),
"required": ["completed"]
}
}
},
{
"type": "function",
"function": {
"name": "close_sub_agent",
"description": "强制关闭指定子智能体,适用于长时间无响应、超时或卡死的任务。使用前请确认必要的日志/文件已保留,操作会立即终止该任务。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"task_id": {"type": "string", "description": "子智能体任务ID"},
"agent_id": {"type": "integer", "description": "子智能体编号1~5若缺少 task_id 可用"}
})
}
}
},
{
"type": "function",
"function": {
"name": "create_sub_agent",
"description": "创建新的子智能体任务。适合大规模信息搜集、网页提取与多文档总结等会占用大量上下文的工作需要提供任务摘要、详细要求、交付目录以及参考文件。注意同一时间最多运行5个子智能体。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"agent_id": {"type": "integer", "description": "子智能体代号1~5"},
"summary": {"type": "string", "description": "任务摘要,简要说明目标"},
"task": {"type": "string", "description": "任务详细要求"},
"target_dir": {"type": "string", "description": "项目下用于接收交付的相对目录"},
"reference_files": {
"type": "array",
"description": "提供给子智能体的参考文件列表相对路径禁止在summary和task中直接告知子智能体引用图片的路径必须使用本参数提供",
"items": {"type": "string"},
"maxItems": 10
},
"timeout_seconds": {"type": "integer", "description": "子智能体最大运行秒数:单/双次搜索建议180秒多轮搜索整理建议300秒深度调研或长篇分析可设600秒"}
}),
"required": ["agent_id", "summary", "task", "target_dir"]
}
}
},
{
"type": "function",
"function": {
"name": "wait_sub_agent",
"description": "等待指定子智能体任务结束(或超时)。任务完成后会返回交付目录,并将结果复制到指定的项目文件夹。调用时 `timeout_seconds` 应不少于对应子智能体的 `timeout_seconds`,否则可能提前终止等待。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"task_id": {"type": "string", "description": "子智能体任务ID"},
"agent_id": {"type": "integer", "description": "子智能体代号(可选,用于缺省 task_id 的情况)"},
"timeout_seconds": {"type": "integer", "description": "本次等待的超时时长(秒)"}
}),
"required": []
}
}
},
{
"type": "function",
"function": {
"name": "trigger_easter_egg",
"description": "触发隐藏彩蛋,用于展示非功能性特效。需指定 effect 参数,例如 flood灌水或 snake贪吃蛇",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"effect": {
"type": "string",
"description": "彩蛋标识,目前支持 flood灌水与 snake贪吃蛇"
}
}),
"required": ["effect"]
}
}
}
]
# 视觉模型Qwen3.5 / Kimi-k2.5)自带多模态能力,不再暴露 vlm_analyze改为 view_image / view_video
if getattr(self, "model_key", None) in {"qwen3-vl-plus", "kimi-k2.5"}:
tools = [
tool for tool in tools
if (tool.get("function") or {}).get("name") != "vlm_analyze"
]
tools.append({
"type": "function",
"function": {
"name": "view_image",
"description": "将指定本地图片附加到工具结果中tool 消息携带 image_url便于模型主动查看图片内容。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"path": {
"type": "string",
"description": "项目内的图片相对路径(不要以 /workspace 开头);宿主机模式可用绝对路径。支持 png/jpg/webp/gif/bmp/svg。"
}
}),
"required": ["path"]
}
}
})
tools.append({
"type": "function",
"function": {
"name": "view_video",
"description": "将指定本地视频附加到工具结果中tool 消息携带 video_url便于模型查看视频内容。",
"parameters": {
"type": "object",
"properties": self._inject_intent({
"path": {
"type": "string",
"description": "项目内的视频相对路径(不要以 /workspace 开头);宿主机模式可用绝对路径。支持 mp4/mov/mkv/avi/webm。"
}
}),
"required": ["path"]
}
}
})
# 附加自定义工具(仅管理员可见)
custom_tools = self._build_custom_tools()
if custom_tools:
tools.extend(custom_tools)
if self.disabled_tools:
tools = [
tool for tool in tools
if tool.get("function", {}).get("name") not in self.disabled_tools
]
return self._apply_intent_to_tools(tools)