第 27 章：高级扩展——限流中间件、场景分组与 Agent Skill

async def my_middleware(kwargs, next_handler):
    # 前置处理
    async for response in await next_handler(**kwargs):
        yield response
    # 后置处理

def create_tool_group(
    self,
    group_name: str,
    description: str,
    active: bool = False,
    notes: str | None = None,
) -> None:

@dataclass
class ToolGroup:
    name: str
    active: bool
    description: str
    notes: str | None

class AgentSkill(TypedDict):
    name: str
    description: str
    dir: str

import asyncio
from agentscope.tool._response import ToolResponse
from agentscope.message import TextBlock


def create_concurrency_limiter(max_concurrent: int = 5):
    """创建并发限制中间件。

    Args:
        max_concurrent: 最大并发数
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    async def concurrency_middleware(kwargs, next_handler):
        async with semaphore:
            async for response in await next_handler(**kwargs):
                yield response

    return concurrency_middleware

toolkit = Toolkit()
toolkit.register_tool_function(query_database)
toolkit.register_middleware(create_concurrency_limiter(max_concurrent=3))

import time
from collections import deque


def create_rate_limiter(max_calls: int = 10, period: float = 1.0):
    """创建速率限制中间件。

    Args:
        max_calls: 时间窗口内的最大调用次数
        period: 时间窗口（秒）
    """
    call_times = deque()

    async def rate_limit_middleware(kwargs, next_handler):
        now = time.time()

        # 清除过期记录
        while call_times and call_times[0] < now - period:
            call_times.popleft()

        # 检查是否超限
        if len(call_times) >= max_calls:
            wait_time = period - (now - call_times[0])
            if wait_time > 0:
                await asyncio.sleep(wait_time)

        call_times.append(time.time())

        async for response in await next_handler(**kwargs):
            yield response

    return rate_limit_middleware

import asyncio


def create_timeout(timeout_seconds: float = 30.0):
    """创建超时中间件。"""
    async def timeout_middleware(kwargs, next_handler):
        try:
            async for response in asyncio.wait_for(
                next_handler(**kwargs),
                timeout=timeout_seconds,
            ):
                yield response
        except asyncio.TimeoutError:
            yield ToolResponse(
                content=[TextBlock(
                    type="text",
                    text=f"工具调用超时（{timeout_seconds}秒），已取消。",
                )],
                is_last=True,
            )

    return timeout_middleware

toolkit.register_middleware(create_rate_limiter(max_calls=20, period=1.0))
toolkit.register_middleware(create_concurrency_limiter(max_concurrent=5))
toolkit.register_middleware(create_timeout(timeout_seconds=30.0))

toolkit = Toolkit()

# 基础工具（始终激活）
toolkit.register_tool_function(search_web)
toolkit.register_tool_function(calculator)

# 高级工具（默认不激活）
toolkit.create_tool_group(
    group_name="advanced_search",
    description="高级搜索工具：学术搜索、专利搜索等",
    active=False,
)
toolkit.register_tool_function(search_academic, group_name="advanced_search")
toolkit.register_tool_function(search_patents, group_name="advanced_search")

# 管理工具（默认不激活）
toolkit.create_tool_group(
    group_name="admin",
    description="管理操作：删除数据、修改配置等",
    active=False,
)
toolkit.register_tool_function(delete_records, group_name="admin")
toolkit.register_tool_function(update_config, group_name="admin")

for name, group in toolkit.groups.items():
    status = "激活" if group.active else "未激活"
    print(f"分组 '{name}': {status} - {group.description}")

# 在 ReActAgent 的循环中，模型可以决定激活 "advanced_search" 分组
# 这通过调用 reset_equipped_tools 工具实现

# _toolkit.py:880（简化）
if tool_func.group != "basic" and not self.groups[tool_func.group].active:
    return ToolResponse(content=[TextBlock(text="FunctionInactiveError: ...")])

class AgentSkill(TypedDict):
    name: str         # 技能名称
    description: str  # 技能描述
    dir: str          # 技能目录

toolkit.register_agent_skill("/path/to/skill_dir")

skill_prompt = toolkit.get_agent_skill_prompt()
# 返回类似：
# "你拥有以下技能：
# 1. data_analysis: 分析 CSV 和 Excel 数据，生成统计报告
# 2. web_search: 搜索互联网获取最新信息
# ..."

import time
from collections import defaultdict

stats = defaultdict(lambda: {"calls": 0, "errors": 0, "total_time": 0.0})

async def monitoring_middleware(kwargs, next_handler):
    tool_name = kwargs.get("tool_call", {}).get("name", "unknown")
    start = time.time()
    stats[tool_name]["calls"] += 1

    try:
        async for response in await next_handler(**kwargs):
            yield response
    except Exception as e:
        stats[tool_name]["errors"] += 1
        raise
    finally:
        stats[tool_name]["total_time"] += time.time() - start


def print_stats():
    for name, s in stats.items():
        avg = s["total_time"] / s["calls"] if s["calls"] > 0 else 0
        print(f"  {name}: {s['calls']} 次调用, {s['errors']} 次错误, 平均 {avg:.2f}s")

toolkit.register_middleware(monitoring_middleware)
# ... 运行几次工具调用 ...
print("=== 工具调用统计 ===")
print_stats()