71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
from collections.abc import Callable
|
|
|
|
from .tools.builtin_tools import BUILTIN_TOOLS
|
|
|
|
|
|
class ToolRegistry:
|
|
"""
|
|
统一管理工具注册、查询和执行。
|
|
|
|
设计目标：
|
|
- 让 Orchestrator 只关心“声明了哪些工具”，不关心工具如何存放。
|
|
- 固化统一的工具调用结果结构，便于页面展示和审计日志保存。
|
|
- 后续新增业务工具时，只需要注册函数，不必改调用协议。
|
|
"""
|
|
|
|
def __init__(self, initial_tools: dict[str, Callable] | None = None):
|
|
self._tools: dict[str, Callable] = dict(initial_tools or {})
|
|
|
|
def register(self, tool_name: str, tool_func: Callable) -> None:
|
|
"""注册一个可通过名称调用的工具函数。"""
|
|
self._tools[tool_name] = tool_func
|
|
|
|
def get(self, tool_name: str) -> Callable | None:
|
|
"""按名称返回工具函数；未注册时返回 None。"""
|
|
return self._tools.get(tool_name)
|
|
|
|
def run(self, tool_name: str, **kwargs) -> dict:
|
|
"""
|
|
执行单个工具，并返回统一结果结构。
|
|
|
|
统一返回值是审计日志、页面展示和后续 Agent 编排共享的协议。
|
|
即使工具不存在或执行失败，也返回可消费的失败结果，而不是抛异常。
|
|
"""
|
|
tool = self.get(tool_name)
|
|
if tool is None:
|
|
return {
|
|
"tool_name": tool_name,
|
|
"success": False,
|
|
"arguments": kwargs,
|
|
"result": {},
|
|
"error": "工具未注册",
|
|
}
|
|
try:
|
|
return {
|
|
"tool_name": tool_name,
|
|
"success": True,
|
|
"arguments": kwargs,
|
|
"result": tool(**kwargs),
|
|
"error": "",
|
|
}
|
|
except Exception as exc:
|
|
return {
|
|
"tool_name": tool_name,
|
|
"success": False,
|
|
"arguments": kwargs,
|
|
"result": {},
|
|
"error": str(exc),
|
|
}
|
|
|
|
|
|
# 默认注册表承载项目内置工具，便于当前 V1 直接复用。
|
|
DEFAULT_TOOL_REGISTRY = ToolRegistry(BUILTIN_TOOLS)
|
|
|
|
|
|
def run_declared_tools(tool_names: list[str], user_input: str) -> list[dict]:
|
|
"""按场景声明顺序执行工具，保证结果顺序与配置顺序一致。"""
|
|
return [
|
|
DEFAULT_TOOL_REGISTRY.run(tool_name, user_input=user_input)
|
|
for tool_name in tool_names
|
|
]
|