mirror of
https://github.com/agent0ai/agent-zero.git
synced 2026-08-25 08:14:19 +00:00
5ff106a2d4
Set the default consecutive unusable response threshold to five and align the Agent Settings copy, helper DOX, and regression assertion.
99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from helpers.errors import HandledException
|
|
from helpers.files import read_prompt_file
|
|
from helpers.settings import get_default_settings, normalize_settings
|
|
from extensions.python._functions.agent.Agent.hist_add_warning.end import (
|
|
_90_stop_unusable_response_loop as response_loop,
|
|
)
|
|
|
|
|
|
class FakeLog:
|
|
def __init__(self):
|
|
self.entries = []
|
|
|
|
def log(self, **entry):
|
|
self.entries.append(entry)
|
|
|
|
|
|
def _agent():
|
|
prompts = {
|
|
"fw.msg_misformat.md": "misformatted",
|
|
"fw.msg_repeat.md": "repeated",
|
|
}
|
|
|
|
def read_prompt(name, **kwargs):
|
|
if name == "fw.msg_unusable_response_limit.md":
|
|
return f"stopped at {kwargs['limit']}"
|
|
return prompts[name]
|
|
|
|
return SimpleNamespace(
|
|
loop_data=SimpleNamespace(iteration=0, params_persistent={}),
|
|
context=SimpleNamespace(log=FakeLog()),
|
|
read_prompt=read_prompt,
|
|
)
|
|
|
|
|
|
def _run(extension, agent, message):
|
|
data = {"args": (agent, message), "kwargs": {}, "exception": None}
|
|
extension.execute(data=data)
|
|
return data
|
|
|
|
|
|
def test_stops_at_configured_failure_limit(monkeypatch):
|
|
monkeypatch.setattr(
|
|
response_loop,
|
|
"get_settings",
|
|
lambda: {"max_consecutive_unusable_responses": 3},
|
|
)
|
|
agent = _agent()
|
|
extension = response_loop.StopUnusableResponseLoop(agent=agent)
|
|
|
|
assert _run(extension, agent, "misformatted")["exception"] is None
|
|
|
|
agent.loop_data.iteration = 1
|
|
assert _run(extension, agent, "repeated")["exception"] is None
|
|
|
|
agent.loop_data.iteration = 2
|
|
data = _run(extension, agent, "repeated")
|
|
|
|
assert isinstance(data["exception"], HandledException)
|
|
assert agent.loop_data.params_persistent[response_loop.STATE_KEY]["count"] == 3
|
|
assert agent.context.log.entries == [
|
|
{"type": "warning", "content": "stopped at 3"}
|
|
]
|
|
|
|
|
|
def test_nonconsecutive_failure_starts_a_new_recovery_window(monkeypatch):
|
|
monkeypatch.setattr(
|
|
response_loop,
|
|
"get_settings",
|
|
lambda: {"max_consecutive_unusable_responses": 2},
|
|
)
|
|
agent = _agent()
|
|
extension = response_loop.StopUnusableResponseLoop(agent=agent)
|
|
|
|
assert _run(extension, agent, {"structured": "warning"})["exception"] is None
|
|
_run(extension, agent, "misformatted")
|
|
agent.loop_data.iteration = 2
|
|
data = _run(extension, agent, "repeated")
|
|
|
|
assert data["exception"] is None
|
|
assert agent.loop_data.params_persistent[response_loop.STATE_KEY]["count"] == 1
|
|
|
|
|
|
def test_general_settings_expose_the_default_failure_limit():
|
|
settings = get_default_settings()
|
|
assert settings["max_consecutive_unusable_responses"] == 5
|
|
settings["max_consecutive_unusable_responses"] = 0
|
|
assert normalize_settings(settings)["max_consecutive_unusable_responses"] == 1
|
|
|
|
html = Path("webui/components/settings/agent/agent.html").read_text()
|
|
assert (
|
|
'x-model.number="$store.settings.settings.max_consecutive_unusable_responses"'
|
|
in html
|
|
)
|
|
assert "after 3 consecutive" in read_prompt_file(
|
|
"fw.msg_unusable_response_limit.md", ["prompts"], limit=3
|
|
)
|