类class agentscope.middleware

ReplyBudgetControlMiddleware

class ReplyBudgetControlMiddleware(MiddlewareBase)

源码Source agentscope/middleware/_budget.py:29–233

回复预算 Control 中间件，在智能体执行流程中提供横切扩展。位于 agentscope.middleware 包，继承自 MiddlewareBase。核心成员包括 on_reply、on_reasoning。

Tracks cumulative weighted token usage across all reasoning steps within a single reply. The weighted cost is computed as::

cost = input_token_weight * input_tokens + output_token_weight * output_tokens

Once the accumulated cost reaches token_budget, a hint message is injected into the agent's context before the next reasoning step, and tool_choice is forced to "none" so the agent wraps up without invoking any further tools. A reply that still owes a structured output is offered only the structured output tool instead.

Budget state is stored in :attr:~agentscope.agent.AgentState.middle_context keyed by the middleware key, so it persists across human-in-the-loop (HITL) interruptions and resumptions. State is reset when the next reply starts via a :class:~agentscope.event.ReplyStartEvent.

Example::

from agentscope.middleware import ReplyBudgetControlMiddleware

agent = Agent(
    ...,
    middlewares=[
        ReplyBudgetControlMiddleware(
            token_budget=10000,
            input_token_weight=1.0,
            output_token_weight=2.0,
        )
    ],
)

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(token_budget: float, input_token_weight: float = 1, output_token_weight: float = 1, hint_message: str = _D…`	初始化 `ReplyBudgetControlMiddleware` 实例。 Initialize the budget control middleware.

方法摘要Method Summary

方法Method	说明Description
`async def on_reply(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., AsyncGenerator]) -> AsyncGenerator`async	拦截并扩展回复流程。 Manage per-reply budget state in ``agent.state.middle_context``.
`async def on_reasoning(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., AsyncGenerator]) -> AsyncGenerat…`async	拦截并扩展推理阶段。 Intercept each reasoning step to enforce the token budget.

构造器详情Constructor Detail

ReplyBudgetControlMiddleware

def __init__(token_budget: float, input_token_weight: float = 1, output_token_weight: float = 1, hint_message: str = _DEFAULT_HINT_MESSAGE) -> None

初始化 ReplyBudgetControlMiddleware 实例。

Initialize the budget control middleware.

参数Parameters

名称Name	类型Type	说明Description
`token_budget`	float	令牌预算（类型 `float`）。Maximum weighted token cost allowed per reply. The cost for each model call is computed as `input_token_weight * input_tokens + output_token_weight * output_tokens`. Once the accumulated cost reaches this threshold, the agent is instructed to wrap up without calling any more tools.
`input_token_weight`默认default `1`	float	输入令牌权重（类型 `float`）。Multiplier applied to input tokens when computing the weighted cost. Defaults to `1`.
`output_token_weight`默认default `1`	float	输出令牌权重（类型 `float`）。Multiplier applied to output tokens when computing the weighted cost. Defaults to `1`. Set this higher than `input_token_weight` to reflect that output tokens are typically more expensive.
`hint_message`默认default `_DEFAULT_HINT_MESSAGE`	str	hint 消息（类型 `str`）。The message injected into the agent's context when the budget is exceeded. Defaults to a built-in wrap-up prompt.

源码Source agentscope/middleware/_budget.py:72–105

方法详情Method Detail

on_reply

async def on_reply(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., AsyncGenerator]) -> AsyncGenerator

async

拦截并扩展回复流程。

Resets the weighted cost counter to the new reply on :class:~agentscope.event.ReplyStartEvent and accumulates cost on each :class:~agentscope.event.ModelCallEndEvent.

参数Parameters

名称Name	类型Type	说明Description
`agent`	'Agent'	智能体（类型 `'Agent'`）。The agent instance executing this middleware.
`input_kwargs`	dict	输入关键字参数（类型 `dict`）。Reply input kwargs (passed through unchanged).
`next_handler`	Callable[..., AsyncGenerator]	下一个处理器（类型 `Callable[..., AsyncGenerator]`）。Callable that executes the next middleware or `_reply`.

返回Returns AsyncGenerator

返回AsyncGenerator。Events from the reply process.

源码Source agentscope/middleware/_budget.py:107–149

on_reasoning

async def on_reasoning(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., AsyncGenerator]) -> AsyncGenerator

async

拦截并扩展推理阶段。

Before forwarding to the next handler, reads the accumulated weighted cost for the current reply from agent.state.middle_context.

参数Parameters

名称Name	类型Type	说明Description
`agent`	'Agent'	智能体（类型 `'Agent'`）。The agent instance executing this middleware.
`input_kwargs`	dict	输入关键字参数（类型 `dict`）。Dictionary containing `tool_choice` and other reasoning kwargs forwarded to the next handler.
`next_handler`	Callable[..., AsyncGenerator]	下一个处理器（类型 `Callable[..., AsyncGenerator]`）。Callable that executes the next middleware or `_reasoning_impl`.

返回Returns AsyncGenerator

返回AsyncGenerator。Events from the reasoning process.

源码Source agentscope/middleware/_budget.py:151–233

继承成员Inherited Members

继承自Inherited from MiddlewareBase

is_implemented, on_acting, on_check_permission, on_model_call, on_compress_context, on_system_prompt, list_tools, get_middleware_key