File size: 2,597 Bytes
c641d5f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | import requests
from smolagents import CodeAgent, ToolCallingAgent
from agent import GaiaAgent
from config import Settings
def settings(tmp_path, **overrides):
values = {
"api_url": "https://example.test",
"hf_token": "fake-token",
"model_id": "primary",
"vision_model_id": "vision",
"inference_provider": "provider-a",
"asr_model_id": "asr",
"cache_dir": tmp_path,
"request_timeout": 10,
"retries": 1,
"backoff_seconds": 0,
"max_steps": 2,
"use_cache": False,
"stockfish_path": None,
"model_requests_per_minute": 60_000,
"search_requests_per_minute": 60_000,
}
values.update(overrides)
return Settings(**values)
def test_managed_web_research_architecture_and_model_reuse(tmp_path):
agent = GaiaAgent(settings(tmp_path))
spec = ("primary", "provider-a")
manager = agent._manager(spec)
assert isinstance(manager, CodeAgent)
assert isinstance(manager.managed_agents["web_researcher"], ToolCallingAgent)
assert manager.managed_agents["web_researcher"].max_steps == 10
assert agent._model(spec) is agent._model(spec)
def test_transient_primary_failure_uses_configured_fallback(tmp_path, monkeypatch):
agent = GaiaAgent(
settings(
tmp_path,
fallback_model_id="secondary",
fallback_provider="provider-b",
)
)
class Runner:
def __init__(self, value):
self.value = value
def run(self, prompt, images=None):
if isinstance(self.value, Exception):
raise self.value
return self.value
runners = {
("primary", "provider-a"): Runner(requests.ConnectionError("DNS failure")),
("secondary", "provider-b"): Runner(
'{"candidate_answer":"42","evidence":[],"confidence":1}'
),
}
monkeypatch.setattr(agent, "_manager", lambda spec: runners[spec])
assert "42" in str(agent._run_model("question"))
def test_nontransient_agent_failure_also_tries_fallback(tmp_path, monkeypatch):
agent = GaiaAgent(settings(tmp_path, fallback_model_id="secondary"))
class Runner:
def run(self, prompt, images=None):
raise ValueError("bad tool arguments")
monkeypatch.setattr(agent, "_manager", lambda spec: Runner())
import pytest
with pytest.raises(
RuntimeError, match="All inference model/provider attempts failed"
) as exc:
agent._run_model("question")
assert str(exc.value).count("bad tool arguments") == 2
|