Open
Description
Problem Statement
For our use case, we want to be able to set limits to handle when the agent is stuck in a loop/when the task is more complex than anticipated. This is to address a concern with the potential costs of letting the agent dictate how much compute it needs.
With Callback Handlers we can see the information we need with this toy example for max_cycles:
import functools
from strands import Agent
def _callback_handler(max_cycles, **kwargs):
if "event_loop_metrics" in kwargs:
event_loop_metrics = kwargs["event_loop_metrics"]
if event_loop_metrics.cycle_count > max_cycles:
logger.warning(
f"TODO: end run, cycle budget exhausted: {event_loop_metrics.cycle_count} > {max_cycles}"
)
_handler = functools.partial(_callback_handler, max_cycles=100)
agent = Agent(model=model, callback_handler=_handler)
But once we have this information, there is currently no way to act on it.
Proposed Solution
The ideal state for this feature would be that you can pass a limits object at the time of constructing the agent and have the agent handle it:
from pydantic import BaseModel
class AgentLimits(BaseModel):
token_budget: int
cycle_budget: int
runtime: float # in seconds
budget = AgentLimits(token_budget=10000, cycle_budget=100, runtime=1800)
agent = Agent(model=model, callback_handler=_handler limits=budget)
OR,
Provide hooks that we can implement to catch these updates and exit out gracefully.
Use Case
- As a user, I have a limited bedrock budget, and want to make sure my agent doesn't exceed my allocation.
- As a user, I want to cancel out an agent workflow if it takes more than 10 minutes.
Alternatives Solutions
No response
Additional Context
No response