58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
"""单个 Agent Run 可见的工具视图。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
from typing import Any
|
|
|
|
from app.application.tool_runtime.runtime import ToolRuntime
|
|
from app.modules.tool_runtime.definition import ToolInput, ToolResult, ToolSpec
|
|
|
|
|
|
class RunScopedToolView:
|
|
"""保存 Run 级工具别名；实际执行始终委托共享 ToolRuntime。"""
|
|
|
|
def __init__(self, runtime: ToolRuntime) -> None:
|
|
self._runtime = runtime
|
|
self._aliases: dict[str, tuple[str, ToolSpec]] = {}
|
|
|
|
def bind_alias(self, alias: str, target: str, *, description_prefix: str = "") -> bool:
|
|
"""将已有工具作为 Run 级别名暴露；目标不存在时返回 False。"""
|
|
spec = next((item for item in self._runtime.list_specs() if item.name == target), None)
|
|
if spec is None:
|
|
return False
|
|
description = f"{description_prefix}{spec.description}" if description_prefix else spec.description
|
|
self._aliases[alias] = (target, replace(spec, name=alias, description=description))
|
|
return True
|
|
|
|
def list_specs(self) -> list[ToolSpec]:
|
|
return [*self._runtime.list_specs(), *(spec for _, spec in self._aliases.values())]
|
|
|
|
def specs_by_name(self) -> dict[str, ToolSpec]:
|
|
return {spec.name: spec for spec in self.list_specs()}
|
|
|
|
async def execute(
|
|
self,
|
|
tool: str,
|
|
input: ToolInput,
|
|
*,
|
|
run_id: str | None = None,
|
|
character_id: str | None = None,
|
|
trace_id: str | None = None,
|
|
confirmed: bool = False,
|
|
context: dict[str, Any] | None = None,
|
|
) -> ToolResult:
|
|
target = self._aliases.get(tool, (tool,))[0]
|
|
return await self._runtime.execute(
|
|
target,
|
|
input,
|
|
run_id=run_id,
|
|
character_id=character_id,
|
|
trace_id=trace_id,
|
|
confirmed=confirmed,
|
|
context=context,
|
|
)
|
|
|
|
|
|
__all__ = ["RunScopedToolView"]
|