refactor(core): remove tool history manager and integrate cache into tool executor

Removes the `ToolHistoryManager` and its associated functionalities, including history recording, querying, and prompt generation. This simplifies the architecture by decoupling tool execution history from the core logic.

The tool caching mechanism is now directly integrated into the `ToolExecutor` by wrapping the `execute_tool_call` method. This ensures that caching is applied consistently for all tool executions that have it enabled, improving performance and reducing redundant calls.

- Deletes `src/common/tool_history.py`.
- Removes tool history related functions from `prompt_builder.py` and `tool_api.py`.
- Adds a `wrap_tool_executor` function in `cache_manager.py` to apply caching logic directly to the `ToolExecutor`.
This commit is contained in:
minecraft1024a
2025-08-28 18:42:17 +08:00
parent 5153d8f60f
commit 7ba3b7f34d
4 changed files with 81 additions and 573 deletions

View File

@@ -1,9 +1,7 @@
from typing import Any, Dict, List, Optional, Type, Union
from datetime import datetime
from typing import Any, Dict, List, Optional, Type
from src.plugin_system.base.base_tool import BaseTool
from src.plugin_system.base.component_types import ComponentType
from src.common.tool_history import ToolHistoryManager
from src.common.logger import get_logger
logger = get_logger("tool_api")
@@ -33,110 +31,4 @@ def get_llm_available_tool_definitions():
from src.plugin_system.core import component_registry
llm_available_tools = component_registry.get_llm_available_tools()
return [(name, tool_class.get_tool_definition()) for name, tool_class in llm_available_tools.items()]
def get_tool_history(
tool_names: Optional[List[str]] = None,
start_time: Optional[Union[datetime, str]] = None,
end_time: Optional[Union[datetime, str]] = None,
chat_id: Optional[str] = None,
limit: Optional[int] = None,
status: Optional[str] = None
) -> List[Dict[str, Any]]:
"""
获取工具调用历史记录
Args:
tool_names: 工具名称列表,为空则查询所有工具
start_time: 开始时间可以是datetime对象或ISO格式字符串
end_time: 结束时间可以是datetime对象或ISO格式字符串
chat_id: 会话ID用于筛选特定会话的调用
limit: 返回记录数量限制
status: 执行状态筛选("completed""error")
Returns:
List[Dict]: 工具调用记录列表,每条记录包含以下字段:
- tool_name: 工具名称
- timestamp: 调用时间
- arguments: 调用参数
- result: 调用结果
- execution_time: 执行时间
- status: 执行状态
- chat_id: 会话ID
"""
history_manager = ToolHistoryManager()
return history_manager.query_history(
tool_names=tool_names,
start_time=start_time,
end_time=end_time,
chat_id=chat_id,
limit=limit,
status=status
)
def get_tool_history_text(
tool_names: Optional[List[str]] = None,
start_time: Optional[Union[datetime, str]] = None,
end_time: Optional[Union[datetime, str]] = None,
chat_id: Optional[str] = None,
limit: Optional[int] = None,
status: Optional[str] = None
) -> str:
"""
获取工具调用历史记录的文本格式
Args:
tool_names: 工具名称列表,为空则查询所有工具
start_time: 开始时间可以是datetime对象或ISO格式字符串
end_time: 结束时间可以是datetime对象或ISO格式字符串
chat_id: 会话ID用于筛选特定会话的调用
limit: 返回记录数量限制
status: 执行状态筛选("completed""error")
Returns:
str: 格式化的工具调用历史记录文本
"""
history = get_tool_history(
tool_names=tool_names,
start_time=start_time,
end_time=end_time,
chat_id=chat_id,
limit=limit,
status=status
)
if not history:
return "没有找到工具调用记录"
text = "工具调用历史记录:\n"
for record in history:
# 提取结果中的name和content
result = record['result']
if isinstance(result, dict):
name = result.get('name', record['tool_name'])
content = result.get('content', str(result))
else:
name = record['tool_name']
content = str(result)
# 格式化内容
content = content.strip().replace('\n', ' ')
if len(content) > 200:
content = content[:200] + "..."
# 格式化时间
timestamp = datetime.fromisoformat(record['timestamp']).strftime("%Y-%m-%d %H:%M:%S")
text += f"[{timestamp}] {name}\n"
text += f"结果: {content}\n\n"
return text
def clear_tool_history() -> None:
"""
清除所有工具调用历史记录
"""
history_manager = ToolHistoryManager()
history_manager.clear_history()
return [(name, tool_class.get_tool_definition()) for name, tool_class in llm_available_tools.items()]