第 10 站：执行工具

def log_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"调用 {func.__name__}")
        result = func(*args, **kwargs)
        print(f"完成 {func.__name__}")
        return result
    return wrapper

@log_decorator
def get_weather(city):
    return f"{city}: 晴"

# _toolkit.py:117
class Toolkit(StateModule):
    """Toolkit 是注册、管理、删除工具函数的核心模块"""

# _toolkit.py:157
self.tools: dict[str, RegisteredToolFunction] = {}    # 所有注册的工具
self.groups: dict[str, ToolGroup] = {}                 # 工具分组
self.skills: dict[str, AgentSkill] = {}                # Agent 技能
self._middlewares: list = []                           # 中间件列表

# _types.py:16
@dataclass
class RegisteredToolFunction:
    name: str                  # 函数名
    group: str | Literal["basic"]  # 所属分组
    source: Literal["function", "mcp_server", "function_group"]  # 来源
    original_func: ToolFunction    # 原始 Python 函数
    json_schema: dict              # 自动生成的 JSON Schema
    preset_kwargs: dict            # 预设参数（不暴露给模型）
    postprocess_func: ... | None   # 后处理函数
    async_execution: bool          # 是否异步执行

# _toolkit.py:274
def register_tool_function(
    self,
    tool_func: ToolFunction,
    group_name: str | Literal["basic"] = "basic",
    preset_kwargs: dict | None = None,
    func_name: str | None = None,
    func_description: str | None = None,
    json_schema: dict | None = None,
    ...
) -> None:

# _toolkit.py:853
@trace_toolkit
@_apply_middlewares     # 注意这个装饰器！
async def call_tool_function(
    self,
    tool_call: ToolUseBlock,
) -> AsyncGenerator[ToolResponse, None]:

# 简化的执行逻辑（_toolkit.py:853 之后）

# 1. 检查工具是否存在
if tool_call["name"] not in self.tools:
    return ToolResponse(content=[TextBlock(text="FunctionNotFoundError: ...")])

# 2. 检查工具分组是否激活
if tool_func.group != "basic" and not self.groups[tool_func.group].active:
    return ToolResponse(content=[TextBlock(text="FunctionInactiveError: ...")])

# 3. 合并参数
kwargs = {**tool_func.preset_kwargs, **(tool_call.get("input", {}) or {})}

# 4. 执行函数 → 返回 AsyncGenerator[ToolResponse]

# _response.py:12
@dataclass
class ToolResponse:
    content: list[TextBlock | ImageBlock | AudioBlock | VideoBlock]
    metadata: dict | None
    stream: bool
    is_last: bool
    is_interrupted: bool

# _toolkit.py:187
def create_tool_group(
    self,
    group_name: str,
    description: str,
    active: bool = False,
    ...
) -> None:

toolkit.create_tool_group("advanced", description="高级工具", active=False)
toolkit.register_tool_function(complex_tool, group_name="advanced")

# _toolkit.py:1441
def register_middleware(
    self,
    middleware: Callable,
) -> None:
    """注册洋葱式中间件"""

async def my_middleware(
    kwargs: dict,          # 包含 tool_call
    next_handler: Callable  # 下一个中间件或真正的工具函数
) -> AsyncGenerator[ToolResponse, None]:
    # 前置处理
    print(f"即将调用: {kwargs['tool_call']['name']}")

    # 调用下一层
    async for response in await next_handler(**kwargs):
        # 可以修改每个响应
        yield response

    # 后置处理
    print("工具调用完成")

# _toolkit.py:57
def _apply_middlewares(func):
    """装饰器：在运行时动态构建中间件链"""
    @wraps(func)
    async def wrapper(self, tool_call):
        middlewares = self._middlewares
        if not middlewares:
            return func(self, tool_call)  # 没有中间件，直接执行

        # 构建洋葱链：最内层是真正的函数，外层依次包裹中间件
        chain = func
        for middleware in reversed(middlewares):
            chain = partial(middleware, next_handler=chain)
        return chain(kwargs={"tool_call": tool_call})

print(f"[DEBUG] 调用工具: {tool_call['name']}, 参数: {tool_call.get('input', {})}")

import asyncio
from agentscope.tool import Toolkit
from agentscope.tool._response import ToolResponse
from agentscope.message import TextBlock, ToolUseBlock

def get_weather(city: str) -> ToolResponse:
    """获取指定城市的天气。

    Args:
        city (str): 城市名称

    Returns:
        ToolResponse: 天气信息
    """
    return ToolResponse(
        content=[TextBlock(type="text", text=f"{city}: 晴，25°C")]
    )

async def main():
    toolkit = Toolkit()
    toolkit.register_tool_function(get_weather)

    # 打印自动生成的 JSON Schema
    print("=== 注册的工具 ===")
    for name, func in toolkit.tools.items():
        print(f"工具名: {name}")
        import json
        print(json.dumps(func.json_schema, ensure_ascii=False, indent=2))

    # 模拟模型返回的 ToolUseBlock
    tool_call = ToolUseBlock(
        type="tool_use",
        id="call_test",
        name="get_weather",
        input={"city": "北京"},
    )

    # 调用工具
    print("\n=== 调用工具 ===")
    async for response in toolkit.call_tool_function(tool_call):
        for block in response.content:
            print(f"结果: {block['text']}")

asyncio.run(main())

import time

async def timing_middleware(kwargs, next_handler):
    start = time.time()
    print(f"  [计时] 开始执行 {kwargs.get('name', '?')}")
    async for resp in await next_handler(**kwargs):
        yield resp
    elapsed = time.time() - start
    print(f"  [计时] 耗时 {elapsed:.3f}s")

toolkit.register_middleware(timing_middleware)

print(f"[DEBUG] 注册工具: {func_name} → 分组={group_name}, "
      f"schema参数={list(json_schema.get('function', {}).get('parameters', {}).get('properties', {}).keys())}")

def _apply_middlewares(func):
    @wraps(func)
    async def wrapper(self, tool_call):
        middlewares = self._middlewares
        if not middlewares:
            return func(self, tool_call)

        print(f"[DEBUG] 中间件链: {len(middlewares)} 层, "
              f"顺序={[m.__name__ for m in middlewares]}")
        chain = func
        for middleware in reversed(middlewares):
            chain = partial(middleware, next_handler=chain)
        return chain(kwargs={"tool_call": tool_call})