feat(llm): migrate from OpenRouter to OpusAgent

Replace OpenRouter (openai SDK) with OpusAgent REST API (httpx + polling).
LLMClient now creates topics per system prompt, submits async requests,
and polls for completion. Model tiers map to Claude model IDs
(opus-4-7, sonnet-4-6, haiku-4-5) configurable via env vars.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-05-26 09:09:54 +00:00
parent a702b2090d
commit 646c64dacd
10 changed files with 342 additions and 324 deletions
+1 -1
View File
@@ -23,7 +23,7 @@
# ./state contiene runs.db (GA) + strategy_crypto.db (paper) + WAL/SHM # ./state contiene runs.db (GA) + strategy_crypto.db (paper) + WAL/SHM
# ./src/strategy_crypto/strategy_crypto/strategies JSON freezate (ro) # ./src/strategy_crypto/strategy_crypto/strategies JSON freezate (ro)
# #
# Secrets (token Cerbero + OpenRouter): caricati da .env via env_file. # Secrets (token Cerbero + OpusAgent): caricati da .env via env_file.
networks: networks:
traefik: traefik:
+4 -4
View File
@@ -159,8 +159,8 @@ def parse_args() -> argparse.Namespace:
default=1, default=1,
help=( help=(
"Numero di propose() LLM concorrenti per generazione (default 1 = " "Numero di propose() LLM concorrenti per generazione (default 1 = "
"serial). 6-10 tipicamente accettati da OpenRouter qwen-2.5 senza " "serial). OpusAgent processa in coda FIFO; concurrency > 1 accoda "
"rate-limit; riduce wall time GA loop di 5-8x." "piu' richieste in parallelo."
), ),
) )
return p.parse_args() return p.parse_args()
@@ -200,13 +200,13 @@ def main() -> None:
print(f"OHLCV loaded: {len(ohlcv)} bars from {ohlcv.index[0]} to {ohlcv.index[-1]}") print(f"OHLCV loaded: {len(ohlcv)} bars from {ohlcv.index[0]} to {ohlcv.index[-1]}")
llm = LLMClient( llm = LLMClient(
openrouter_api_key=settings.openrouter_api_key.get_secret_value(), opus_agent_api_key=settings.opus_agent_api_key.get_secret_value(),
opus_agent_base_url=settings.opus_agent_base_url,
model_tier_s=settings.llm_model_tier_s, model_tier_s=settings.llm_model_tier_s,
model_tier_a=settings.llm_model_tier_a, model_tier_a=settings.llm_model_tier_a,
model_tier_b=settings.llm_model_tier_b, model_tier_b=settings.llm_model_tier_b,
model_tier_c=settings.llm_model_tier_c, model_tier_c=settings.llm_model_tier_c,
model_tier_d=settings.llm_model_tier_d, model_tier_d=settings.llm_model_tier_d,
openrouter_base_url=settings.openrouter_base_url,
) )
cfg = RunConfig( cfg = RunConfig(
+2 -4
View File
@@ -410,16 +410,14 @@ def main() -> None:
) )
loader = CerberoOHLCVLoader(client=cerbero, cache_dir=settings.series_dir) loader = CerberoOHLCVLoader(client=cerbero, cache_dir=settings.series_dir)
# LLM client (qwen-2.5-72b tier C come da spec progetto - vedi MEMORY:
# cambiare modello senza ricalibrare = regressione dimostrata).
llm = LLMClient( llm = LLMClient(
openrouter_api_key=settings.openrouter_api_key.get_secret_value(), opus_agent_api_key=settings.opus_agent_api_key.get_secret_value(),
opus_agent_base_url=settings.opus_agent_base_url,
model_tier_s=settings.llm_model_tier_s, model_tier_s=settings.llm_model_tier_s,
model_tier_a=settings.llm_model_tier_a, model_tier_a=settings.llm_model_tier_a,
model_tier_b=settings.llm_model_tier_b, model_tier_b=settings.llm_model_tier_b,
model_tier_c=settings.llm_model_tier_c, model_tier_c=settings.llm_model_tier_c,
model_tier_d=settings.llm_model_tier_d, model_tier_d=settings.llm_model_tier_d,
openrouter_base_url=settings.openrouter_base_url,
) )
# Setup DB winners (separato dal GA core DB). # Setup DB winners (separato dal GA core DB).
@@ -4,11 +4,9 @@ import re
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any from typing import Any
import openai
from ..genome.hypothesis import HypothesisAgentGenome from ..genome.hypothesis import HypothesisAgentGenome
from ..genome.prompt_library import PromptLibrary from ..genome.prompt_library import PromptLibrary
from ..llm.client import CompletionResult, EmptyCompletionError, LLMClient from ..llm.client import CompletionResult, EmptyCompletionError, LLMClient, OpusAgentError, OpusAgentTransientError
from ..protocol.parser import ParseError, Strategy, parse_strategy from ..protocol.parser import ParseError, Strategy, parse_strategy
from ..protocol.validator import ValidationError, validate_strategy from ..protocol.validator import ValidationError, validate_strategy
@@ -446,15 +444,15 @@ class HypothesisAgent:
try: try:
completion = self._llm.complete(genome, system=system, user=user) completion = self._llm.complete(genome, system=system, user=user)
except EmptyCompletionError as e: except EmptyCompletionError as e:
# LLM esaurito retry tenacity senza una risposta. Tratta come
# parse-fail "empty" e ritenta nel loop esterno (max_attempts).
errors.append(f"empty_completion: {e}") errors.append(f"empty_completion: {e}")
last_raw = "" last_raw = ""
continue continue
except openai.RateLimitError as e: except OpusAgentTransientError as e:
# Provider upstream rate limited oltre i retry tenacity. errors.append(f"transient_error: {e}")
# Marca genome come fallito senza propagare l'eccezione al run. last_raw = ""
errors.append(f"rate_limit: {e}") continue
except OpusAgentError as e:
errors.append(f"opus_agent_error: {e}")
last_raw = "" last_raw = ""
continue continue
completions.append(completion) completions.append(completion)
@@ -23,14 +23,14 @@ class Settings(BaseSettings):
cerbero_mainnet_token: SecretStr | None = None cerbero_mainnet_token: SecretStr | None = None
cerbero_bot_tag: str = "swarm-poc-phase1" cerbero_bot_tag: str = "swarm-poc-phase1"
openrouter_api_key: SecretStr opus_agent_api_key: SecretStr
opus_agent_base_url: str = "https://opus-agent.tielogic.xyz"
llm_model_tier_s: str = "google/gemini-3-flash-preview" llm_model_tier_s: str = "claude-opus-4-7"
llm_model_tier_a: str = "deepseek/deepseek-v4-flash" llm_model_tier_a: str = "claude-opus-4-7"
llm_model_tier_b: str = "deepseek/deepseek-v4-flash" llm_model_tier_b: str = "claude-sonnet-4-6"
llm_model_tier_c: str = "qwen/qwen-2.5-72b-instruct" llm_model_tier_c: str = "claude-sonnet-4-6"
llm_model_tier_d: str = "openai/gpt-oss-20b" llm_model_tier_d: str = "claude-haiku-4-5-20251001"
openrouter_base_url: str = "https://openrouter.ai/api/v1"
run_name: str = "phase1-spike-001" run_name: str = "phase1-spike-001"
data_dir: Path = Field(default=Path("./data")) data_dir: Path = Field(default=Path("./data"))
@@ -8,11 +8,11 @@ from typing import Any
class ModelTier(StrEnum): class ModelTier(StrEnum):
S = "S" # top-tier reasoning (Opus / equivalent) via Anthropic S = "S" # top-tier reasoning → opus via OpusAgent
A = "A" # premium override via Anthropic A = "A" # premium → opus via OpusAgent
B = "B" # Sonnet 4.6 via Anthropic B = "B" # standard → sonnet via OpusAgent
C = "C" # Qwen 2.5 72B via OpenRouter C = "C" # default GA → sonnet via OpusAgent
D = "D" # ultra-economic (Llama / cheap models) via OpenRouter D = "D" # economic → haiku via OpusAgent
@dataclass @dataclass
@@ -1,10 +1,12 @@
from __future__ import annotations from __future__ import annotations
import hashlib
import logging
import threading
import time
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any
import openai import httpx
from openai import OpenAI
from tenacity import ( from tenacity import (
retry, retry,
retry_if_exception_type, retry_if_exception_type,
@@ -14,26 +16,33 @@ from tenacity import (
from ..genome.hypothesis import HypothesisAgentGenome, ModelTier from ..genome.hypothesis import HypothesisAgentGenome, ModelTier
# Modelli configurati per Phase 1 — tutti via OpenRouter logger = logging.getLogger(__name__)
MODEL_TIER_S = "google/gemini-3-flash-preview"
MODEL_TIER_A = "deepseek/deepseek-v4-flash" MODEL_TIER_MAP: dict[ModelTier, str] = {
MODEL_TIER_B = "deepseek/deepseek-v4-flash" ModelTier.S: "claude-opus-4-7",
MODEL_TIER_C = "qwen/qwen-2.5-72b-instruct" ModelTier.A: "claude-opus-4-7",
MODEL_TIER_D = "openai/gpt-oss-20b" ModelTier.B: "claude-sonnet-4-6",
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1" ModelTier.C: "claude-sonnet-4-6",
ModelTier.D: "claude-haiku-4-5-20251001",
}
class EmptyCompletionError(RuntimeError): class EmptyCompletionError(RuntimeError):
pass pass
# Errori transient: retry. Auth/InvalidRequest: NO retry. class OpusAgentError(RuntimeError):
# RateLimitError (HTTP 429) ora retryable: provider OpenRouter come DeepInfra pass
# applicano rate limit upstream temporaneo, recuperabile con backoff.
class OpusAgentTransientError(RuntimeError):
pass
_RETRYABLE_EXCEPTIONS: tuple[type[BaseException], ...] = ( _RETRYABLE_EXCEPTIONS: tuple[type[BaseException], ...] = (
openai.APIConnectionError, httpx.ConnectError,
openai.APITimeoutError, httpx.TimeoutException,
openai.InternalServerError, OpusAgentTransientError,
openai.RateLimitError,
EmptyCompletionError, EmptyCompletionError,
) )
@@ -48,47 +57,89 @@ class CompletionResult:
class LLMClient: class LLMClient:
# Provider OpenRouter da escludere di default. Novita rifiuta /completions
# endpoint per modelli Qwen 2.x — vedi bug 2026-05-12.
DEFAULT_PROVIDER_IGNORE: tuple[str, ...] = ("Novita",)
def __init__( def __init__(
self, self,
openrouter_api_key: str, opus_agent_api_key: str,
model_tier_s: str = MODEL_TIER_S, opus_agent_base_url: str = "https://opus-agent.tielogic.xyz",
model_tier_a: str = MODEL_TIER_A, model_tier_s: str = "claude-opus-4-7",
model_tier_b: str = MODEL_TIER_B, model_tier_a: str = "claude-opus-4-7",
model_tier_c: str = MODEL_TIER_C, model_tier_b: str = "claude-sonnet-4-6",
model_tier_d: str = MODEL_TIER_D, model_tier_c: str = "claude-sonnet-4-6",
openrouter_base_url: str = OPENROUTER_BASE_URL, model_tier_d: str = "claude-haiku-4-5-20251001",
provider_ignore: tuple[str, ...] | None = None, poll_interval: float = 3.0,
poll_timeout: float = 180.0,
) -> None: ) -> None:
self.model_tier_s = model_tier_s self._base_url = opus_agent_base_url.rstrip("/")
self.model_tier_a = model_tier_a self._api_key = opus_agent_api_key
self.model_tier_b = model_tier_b self._tier_map: dict[ModelTier, str] = {
self.model_tier_c = model_tier_c
self.model_tier_d = model_tier_d
self.openrouter_base_url = openrouter_base_url
self._provider_ignore = (
provider_ignore if provider_ignore is not None else self.DEFAULT_PROVIDER_IGNORE
)
self._tier_models: dict[ModelTier, str] = {
ModelTier.S: model_tier_s, ModelTier.S: model_tier_s,
ModelTier.A: model_tier_a, ModelTier.A: model_tier_a,
ModelTier.B: model_tier_b, ModelTier.B: model_tier_b,
ModelTier.C: model_tier_c, ModelTier.C: model_tier_c,
ModelTier.D: model_tier_d, ModelTier.D: model_tier_d,
} }
# Timeout esplicito (60s) per prevenire hang infinito su connessioni self._poll_interval = poll_interval
# stallate. Tenacity retry su APITimeoutError gestisce il recovery. self._poll_timeout = poll_timeout
self._client = OpenAI( self._topic_cache: dict[str, str] = {}
api_key=openrouter_api_key, self._topic_lock = threading.Lock()
base_url=openrouter_base_url, self._client = httpx.Client(
timeout=60.0, base_url=self._base_url,
headers={"X-Api-Key": self._api_key, "Content-Type": "application/json"},
timeout=30.0,
)
def _get_or_create_topic(self, system_prompt: str) -> str:
prompt_hash = hashlib.sha256(system_prompt.encode()).hexdigest()[:16]
if prompt_hash in self._topic_cache:
return self._topic_cache[prompt_hash]
with self._topic_lock:
if prompt_hash in self._topic_cache:
return self._topic_cache[prompt_hash]
topic_name = f"swarm-{prompt_hash}"
resp = self._client.post("/api/topics", json={
"name": topic_name,
"system_prompt": system_prompt,
})
if resp.status_code == 409:
list_resp = self._client.get("/api/topics")
list_resp.raise_for_status()
for topic in list_resp.json()["data"]:
if topic["name"] == topic_name:
self._topic_cache[prompt_hash] = topic["id"]
return topic["id"]
raise OpusAgentError(f"Topic {topic_name} conflict but not found")
if resp.status_code >= 500:
raise OpusAgentTransientError(f"Server error {resp.status_code}")
resp.raise_for_status()
topic_id = resp.json()["data"]["id"]
self._topic_cache[prompt_hash] = topic_id
logger.debug("Created topic %s -> %s", topic_name, topic_id)
return topic_id
def _poll_result(self, request_id: str) -> dict:
deadline = time.monotonic() + self._poll_timeout
while time.monotonic() < deadline:
resp = self._client.get(f"/api/requests/{request_id}")
resp.raise_for_status()
data = resp.json()["data"]
status = data["status"]
if status == "completed":
return data
if status == "failed":
error = data.get("error") or "unknown error"
raise OpusAgentError(f"Request {request_id} failed: {error}")
time.sleep(self._poll_interval)
raise OpusAgentTransientError(
f"Request {request_id} timed out after {self._poll_timeout}s"
) )
@retry( @retry(
stop=stop_after_attempt(5), stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=2.0, min=2.0, max=30.0), wait=wait_exponential(multiplier=2.0, min=2.0, max=30.0),
retry=retry_if_exception_type(_RETRYABLE_EXCEPTIONS), retry=retry_if_exception_type(_RETRYABLE_EXCEPTIONS),
reraise=True, reraise=True,
@@ -100,28 +151,33 @@ class LLMClient:
user: str, user: str,
max_tokens: int = 2000, max_tokens: int = 2000,
) -> CompletionResult: ) -> CompletionResult:
model = self._tier_models[genome.model_tier] model = self._tier_map[genome.model_tier]
extra_body: dict[str, Any] = {} topic_id = self._get_or_create_topic(system)
if self._provider_ignore:
extra_body["provider"] = {"ignore": list(self._provider_ignore)} resp = self._client.post("/api/requests", json={
resp = self._client.chat.completions.create( "topic_id": topic_id,
model=model, "prompt": user,
messages=[ "model": model,
{"role": "system", "content": system}, })
{"role": "user", "content": user},
], if resp.status_code == 429:
temperature=genome.temperature, raise OpusAgentTransientError("Rate limited")
top_p=genome.top_p, if resp.status_code >= 500:
max_tokens=max_tokens, raise OpusAgentTransientError(f"Server error {resp.status_code}")
extra_body=extra_body or None, if resp.status_code != 202:
) raise OpusAgentError(f"Unexpected status {resp.status_code}: {resp.text}")
if not resp.choices or resp.choices[0].message.content is None:
raise EmptyCompletionError(f"empty response from {model}") request_id = resp.json()["data"]["id"]
usage = resp.usage result = self._poll_result(request_id)
text = result.get("result") or ""
if not text:
raise EmptyCompletionError(f"empty response from OpusAgent ({model})")
return CompletionResult( return CompletionResult(
text=resp.choices[0].message.content, text=text,
input_tokens=usage.prompt_tokens if usage else 0, input_tokens=0,
output_tokens=usage.completion_tokens if usage else 0, output_tokens=0,
tier=genome.model_tier, tier=genome.model_tier,
model=model, model=model,
) )
@@ -75,8 +75,8 @@ class RunConfig:
# strategy_crypto/prompts.json via PromptLibrary.from_json(). # strategy_crypto/prompts.json via PromptLibrary.from_json().
prompt_library: PromptLibrary | None = None prompt_library: PromptLibrary | None = None
# Numero di propose() LLM concorrenti per generazione. 1 = sequenziale (default, # Numero di propose() LLM concorrenti per generazione. 1 = sequenziale (default,
# backward compat). 6-10 tipicamente accettati da OpenRouter qwen-2.5 senza # backward compat). OpusAgent processa in coda FIFO; concurrency > 1
# rate-limit. Riduce wall time GA loop di 5-8x su tier C. # accoda piu' richieste in parallelo ma il throughput dipende dal server.
llm_concurrency: int = 1 llm_concurrency: int = 1
@@ -90,7 +90,7 @@ def _parallel_propose(
``n_workers <= 1`` mantiene il comportamento serial originale (ordine fisso, ``n_workers <= 1`` mantiene il comportamento serial originale (ordine fisso,
determinismo data un seed). ``n_workers > 1`` usa un thread pool: l'order determinismo data un seed). ``n_workers > 1`` usa un thread pool: l'order
dei risultati e' preservato (1:1 con ``genomes``). OpenAI/openrouter client dei risultati e' preservato (1:1 con ``genomes``). LLMClient (OpusAgent)
e' thread-safe; ``PromptLibrary`` e ``HypothesisAgent`` non hanno stato mutabile. e' thread-safe; ``PromptLibrary`` e ``HypothesisAgent`` non hanno stato mutabile.
""" """
if n_workers <= 1 or len(genomes) <= 1: if n_workers <= 1 or len(genomes) <= 1:
+16 -32
View File
@@ -18,13 +18,14 @@ def test_settings_loads_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("CERBERO_TESTNET_TOKEN", "tok-test") monkeypatch.setenv("CERBERO_TESTNET_TOKEN", "tok-test")
monkeypatch.setenv("CERBERO_MAINNET_TOKEN", "tok-main") monkeypatch.setenv("CERBERO_MAINNET_TOKEN", "tok-main")
monkeypatch.setenv("CERBERO_BOT_TAG", "swarm-poc-phase1") monkeypatch.setenv("CERBERO_BOT_TAG", "swarm-poc-phase1")
monkeypatch.setenv("OPENROUTER_API_KEY", "or-key") monkeypatch.setenv("OPUS_AGENT_API_KEY", "oa-key")
monkeypatch.setenv("RUN_NAME", "test-run") monkeypatch.setenv("RUN_NAME", "test-run")
s = Settings() # type: ignore[call-arg] s = Settings() # type: ignore[call-arg]
assert s.cerbero_base_url == "http://test:9000" assert s.cerbero_base_url == "http://test:9000"
assert s.cerbero_testnet_token.get_secret_value() == "tok-test" assert s.cerbero_testnet_token.get_secret_value() == "tok-test"
assert s.opus_agent_api_key.get_secret_value() == "oa-key"
assert s.run_name == "test-run" assert s.run_name == "test-run"
assert s.data_dir.name == "data" assert s.data_dir.name == "data"
assert s.db_path.name == "runs.db" assert s.db_path.name == "runs.db"
@@ -32,50 +33,33 @@ def test_settings_loads_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
def test_settings_requires_tokens(monkeypatch: pytest.MonkeyPatch) -> None: def test_settings_requires_tokens(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("CERBERO_TESTNET_TOKEN", raising=False) monkeypatch.delenv("CERBERO_TESTNET_TOKEN", raising=False)
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) monkeypatch.delenv("OPUS_AGENT_API_KEY", raising=False)
from pydantic import ValidationError from pydantic import ValidationError
with pytest.raises(ValidationError): with pytest.raises(ValidationError):
# Disable .env loading to keep the test deterministic regardless of
# whether a developer's local .env exists and is populated.
Settings(_env_file=None) # type: ignore[call-arg] Settings(_env_file=None) # type: ignore[call-arg]
def test_settings_loads_llm_model_overrides(monkeypatch: pytest.MonkeyPatch) -> None: def test_settings_opus_agent_defaults(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("CERBERO_TESTNET_TOKEN", "tok-test") monkeypatch.setenv("CERBERO_TESTNET_TOKEN", "tok-test")
monkeypatch.setenv("OPENROUTER_API_KEY", "or-key") monkeypatch.setenv("OPUS_AGENT_API_KEY", "oa-key")
monkeypatch.setenv("LLM_MODEL_TIER_S", "claude-mega-x") monkeypatch.delenv("OPUS_AGENT_BASE_URL", raising=False)
monkeypatch.setenv("LLM_MODEL_TIER_A", "claude-premium-y")
monkeypatch.setenv("LLM_MODEL_TIER_B", "claude-opus-4-7")
monkeypatch.setenv("LLM_MODEL_TIER_C", "deepseek/deepseek-chat")
monkeypatch.setenv("LLM_MODEL_TIER_D", "mistralai/mistral-7b")
monkeypatch.setenv("OPENROUTER_BASE_URL", "https://example.com/api/v1")
s = Settings(_env_file=None) # type: ignore[call-arg] s = Settings(_env_file=None) # type: ignore[call-arg]
assert s.llm_model_tier_s == "claude-mega-x" assert s.opus_agent_base_url == "https://opus-agent.tielogic.xyz"
assert s.llm_model_tier_a == "claude-premium-y" assert s.llm_model_tier_s == "claude-opus-4-7"
assert s.llm_model_tier_b == "claude-opus-4-7" assert s.llm_model_tier_a == "claude-opus-4-7"
assert s.llm_model_tier_c == "deepseek/deepseek-chat" assert s.llm_model_tier_b == "claude-sonnet-4-6"
assert s.llm_model_tier_d == "mistralai/mistral-7b" assert s.llm_model_tier_c == "claude-sonnet-4-6"
assert s.openrouter_base_url == "https://example.com/api/v1" assert s.llm_model_tier_d == "claude-haiku-4-5-20251001"
def test_settings_llm_model_defaults(monkeypatch: pytest.MonkeyPatch) -> None: def test_settings_opus_agent_base_url_override(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("CERBERO_TESTNET_TOKEN", "tok-test") monkeypatch.setenv("CERBERO_TESTNET_TOKEN", "tok-test")
monkeypatch.setenv("OPENROUTER_API_KEY", "or-key") monkeypatch.setenv("OPUS_AGENT_API_KEY", "oa-key")
monkeypatch.delenv("LLM_MODEL_TIER_S", raising=False) monkeypatch.setenv("OPUS_AGENT_BASE_URL", "https://custom.example.com")
monkeypatch.delenv("LLM_MODEL_TIER_A", raising=False)
monkeypatch.delenv("LLM_MODEL_TIER_B", raising=False)
monkeypatch.delenv("LLM_MODEL_TIER_C", raising=False)
monkeypatch.delenv("LLM_MODEL_TIER_D", raising=False)
monkeypatch.delenv("OPENROUTER_BASE_URL", raising=False)
s = Settings(_env_file=None) # type: ignore[call-arg] s = Settings(_env_file=None) # type: ignore[call-arg]
assert s.llm_model_tier_s == "google/gemini-3-flash-preview" assert s.opus_agent_base_url == "https://custom.example.com"
assert s.llm_model_tier_a == "deepseek/deepseek-v4-flash"
assert s.llm_model_tier_b == "deepseek/deepseek-v4-flash"
assert s.llm_model_tier_c == "qwen/qwen-2.5-72b-instruct"
assert s.llm_model_tier_d == "openai/gpt-oss-20b"
assert s.openrouter_base_url == "https://openrouter.ai/api/v1"
+174 -192
View File
@@ -1,7 +1,14 @@
import httpx
import pytest import pytest
from multi_swarm_core.genome.hypothesis import HypothesisAgentGenome, ModelTier from multi_swarm_core.genome.hypothesis import HypothesisAgentGenome, ModelTier
from multi_swarm_core.llm.client import CompletionResult, LLMClient from multi_swarm_core.llm.client import (
CompletionResult,
EmptyCompletionError,
LLMClient,
OpusAgentError,
OpusAgentTransientError,
)
def make_genome(tier: ModelTier) -> HypothesisAgentGenome: def make_genome(tier: ModelTier) -> HypothesisAgentGenome:
@@ -16,217 +23,192 @@ def make_genome(tier: ModelTier) -> HypothesisAgentGenome:
) )
def test_completion_tier_c_uses_openrouter(mocker): TOPIC_RESPONSE = {
fake_openai = mocker.MagicMock() "success": True,
fake_response = mocker.MagicMock() "data": {"id": "topic-123", "name": "swarm-test", "system_prompt": "sys"},
fake_response.choices = [mocker.MagicMock(message=mocker.MagicMock(content="(strategy ...)"))] }
fake_response.usage = mocker.MagicMock(prompt_tokens=100, completion_tokens=200)
fake_openai.chat.completions.create.return_value = fake_response
mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai) REQUEST_ACCEPTED = {
"success": True,
"data": {"id": "req-456", "session_id": None, "status": "pending"},
}
client = LLMClient(openrouter_api_key="or-x")
g = make_genome(ModelTier.C) def _completed_response(text: str = "(strategy ...)") -> dict:
out = client.complete(g, system="sys", user="usr") return {
"success": True,
"data": {
"id": "req-456",
"status": "completed",
"result": text,
},
}
def _mock_transport(responses: list[httpx.Response]) -> httpx.MockTransport:
call_idx = {"i": 0}
def handler(request: httpx.Request) -> httpx.Response:
idx = call_idx["i"]
call_idx["i"] += 1
if idx < len(responses):
return responses[idx]
return responses[-1]
return httpx.MockTransport(handler)
def _make_client(transport: httpx.MockTransport) -> LLMClient:
client = LLMClient(opus_agent_api_key="test-key", poll_interval=0.01, poll_timeout=5.0)
client._client = httpx.Client(
base_url="https://opus-agent.tielogic.xyz",
headers={"X-Api-Key": "test-key", "Content-Type": "application/json"},
transport=transport,
)
return client
def test_complete_tier_c_uses_sonnet():
transport = _mock_transport([
httpx.Response(201, json=TOPIC_RESPONSE),
httpx.Response(202, json=REQUEST_ACCEPTED),
httpx.Response(200, json=_completed_response()),
])
client = _make_client(transport)
out = client.complete(make_genome(ModelTier.C), system="sys", user="usr")
assert isinstance(out, CompletionResult) assert isinstance(out, CompletionResult)
assert out.text == "(strategy ...)" assert out.text == "(strategy ...)"
assert out.input_tokens == 100 assert out.model == "claude-sonnet-4-6"
assert out.output_tokens == 200
assert out.tier == ModelTier.C assert out.tier == ModelTier.C
fake_openai.chat.completions.create.assert_called_once()
def test_completion_tier_b_uses_openrouter_with_anthropic_model(mocker): def test_complete_tier_s_uses_opus():
fake_openai = mocker.MagicMock() transport = _mock_transport([
fake_response = mocker.MagicMock() httpx.Response(201, json=TOPIC_RESPONSE),
fake_response.choices = [mocker.MagicMock(message=mocker.MagicMock(content="(strategy ...)"))] httpx.Response(202, json=REQUEST_ACCEPTED),
fake_response.usage = mocker.MagicMock(prompt_tokens=80, completion_tokens=150) httpx.Response(200, json=_completed_response("(strategy s)")),
fake_openai.chat.completions.create.return_value = fake_response ])
mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai) client = _make_client(transport)
out = client.complete(make_genome(ModelTier.S), system="sys", user="usr")
client = LLMClient(openrouter_api_key="or-x") assert out.model == "claude-opus-4-7"
g = make_genome(ModelTier.B)
out = client.complete(g, system="sys", user="usr")
assert out.text == "(strategy ...)"
assert out.input_tokens == 80
assert out.output_tokens == 150
assert out.tier == ModelTier.B
call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
assert call_kwargs["model"] == "deepseek/deepseek-v4-flash"
assert out.model == "deepseek/deepseek-v4-flash"
@pytest.mark.slow
def test_completion_retries_on_connection_error(mocker):
"""Retry esegue 3 tentativi su APIConnectionError, poi rilancia."""
import openai
fake_openai = mocker.MagicMock()
fake_openai.chat.completions.create.side_effect = openai.APIConnectionError(
request=mocker.MagicMock()
)
mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
client = LLMClient(openrouter_api_key="or-x")
g = make_genome(ModelTier.C)
with pytest.raises(openai.APIConnectionError):
client.complete(g, system="sys", user="usr")
assert fake_openai.chat.completions.create.call_count == 5
def test_completion_uses_custom_model_tier_c(mocker):
fake_openai = mocker.MagicMock()
fake_response = mocker.MagicMock()
fake_response.choices = [
mocker.MagicMock(message=mocker.MagicMock(content="(strategy ...)"))
]
fake_response.usage = mocker.MagicMock(prompt_tokens=10, completion_tokens=20)
fake_openai.chat.completions.create.return_value = fake_response
mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
client = LLMClient(
openrouter_api_key="or-x",
model_tier_c="deepseek/deepseek-chat",
)
g = make_genome(ModelTier.C)
out = client.complete(g, system="sys", user="usr")
fake_openai.chat.completions.create.assert_called_once()
call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
assert call_kwargs["model"] == "deepseek/deepseek-chat"
assert out.model == "deepseek/deepseek-chat"
def test_completion_uses_custom_model_tier_b(mocker):
fake_openai = mocker.MagicMock()
fake_response = mocker.MagicMock()
fake_response.choices = [
mocker.MagicMock(message=mocker.MagicMock(content="(strategy ...)"))
]
fake_response.usage = mocker.MagicMock(prompt_tokens=10, completion_tokens=20)
fake_openai.chat.completions.create.return_value = fake_response
mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
client = LLMClient(
openrouter_api_key="or-x",
model_tier_b="anthropic/claude-opus-4-7",
)
g = make_genome(ModelTier.B)
out = client.complete(g, system="sys", user="usr")
fake_openai.chat.completions.create.assert_called_once()
call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
assert call_kwargs["model"] == "anthropic/claude-opus-4-7"
assert out.model == "anthropic/claude-opus-4-7"
def test_completion_tier_s_uses_openrouter_with_anthropic_model(mocker):
fake_openai = mocker.MagicMock()
fake_response = mocker.MagicMock()
fake_response.choices = [mocker.MagicMock(message=mocker.MagicMock(content="(strategy s)"))]
fake_response.usage = mocker.MagicMock(prompt_tokens=50, completion_tokens=100)
fake_openai.chat.completions.create.return_value = fake_response
mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
client = LLMClient(openrouter_api_key="or-x")
g = make_genome(ModelTier.S)
out = client.complete(g, system="sys", user="usr")
fake_openai.chat.completions.create.assert_called_once()
call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
assert call_kwargs["model"] == "google/gemini-3-flash-preview"
assert out.tier == ModelTier.S assert out.tier == ModelTier.S
assert out.model == "google/gemini-3-flash-preview" assert out.text == "(strategy s)"
def test_completion_tier_a_uses_openrouter_with_anthropic_model(mocker): def test_complete_tier_d_uses_haiku():
fake_openai = mocker.MagicMock() transport = _mock_transport([
fake_response = mocker.MagicMock() httpx.Response(201, json=TOPIC_RESPONSE),
fake_response.choices = [mocker.MagicMock(message=mocker.MagicMock(content="(strategy a)"))] httpx.Response(202, json=REQUEST_ACCEPTED),
fake_response.usage = mocker.MagicMock(prompt_tokens=40, completion_tokens=80) httpx.Response(200, json=_completed_response("(strategy d)")),
fake_openai.chat.completions.create.return_value = fake_response ])
mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai) client = _make_client(transport)
out = client.complete(make_genome(ModelTier.D), system="sys", user="usr")
client = LLMClient(openrouter_api_key="or-x") assert out.model == "claude-haiku-4-5-20251001"
g = make_genome(ModelTier.A)
out = client.complete(g, system="sys", user="usr")
fake_openai.chat.completions.create.assert_called_once()
call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
assert call_kwargs["model"] == "deepseek/deepseek-v4-flash"
assert out.tier == ModelTier.A
assert out.model == "deepseek/deepseek-v4-flash"
def test_completion_tier_d_uses_openrouter_with_llama(mocker):
fake_openai = mocker.MagicMock()
fake_response = mocker.MagicMock()
fake_response.choices = [
mocker.MagicMock(message=mocker.MagicMock(content="(strategy d)"))
]
fake_response.usage = mocker.MagicMock(prompt_tokens=30, completion_tokens=70)
fake_openai.chat.completions.create.return_value = fake_response
mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
client = LLMClient(openrouter_api_key="or-x")
g = make_genome(ModelTier.D)
out = client.complete(g, system="sys", user="usr")
fake_openai.chat.completions.create.assert_called_once()
call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
assert call_kwargs["model"] == "openai/gpt-oss-20b"
assert out.tier == ModelTier.D assert out.tier == ModelTier.D
assert out.model == "openai/gpt-oss-20b"
def test_completion_uses_custom_model_tier_s(mocker): def test_topic_cached_on_second_call():
fake_openai = mocker.MagicMock() transport = _mock_transport([
fake_response = mocker.MagicMock() httpx.Response(201, json=TOPIC_RESPONSE),
fake_response.choices = [ httpx.Response(202, json=REQUEST_ACCEPTED),
mocker.MagicMock(message=mocker.MagicMock(content="(strategy custom-s)")) httpx.Response(200, json=_completed_response()),
] httpx.Response(202, json=REQUEST_ACCEPTED),
fake_response.usage = mocker.MagicMock(prompt_tokens=10, completion_tokens=20) httpx.Response(200, json=_completed_response("second")),
fake_openai.chat.completions.create.return_value = fake_response ])
mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai) client = _make_client(transport)
client.complete(make_genome(ModelTier.C), system="sys", user="usr1")
out2 = client.complete(make_genome(ModelTier.C), system="sys", user="usr2")
client = LLMClient( assert out2.text == "second"
openrouter_api_key="or-x",
model_tier_s="anthropic/claude-future-mega",
)
g = make_genome(ModelTier.S)
out = client.complete(g, system="sys", user="usr")
call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
assert call_kwargs["model"] == "anthropic/claude-future-mega"
assert out.model == "anthropic/claude-future-mega"
@pytest.mark.slow def test_topic_conflict_409_recovers():
def test_completion_succeeds_after_one_retry(mocker): transport = _mock_transport([
"""Dopo 1 fallimento transient, il retry riesce al 2 tentativo.""" httpx.Response(409, json={"success": False, "error": {"code": "CONFLICT"}}),
import openai httpx.Response(200, json={"success": True, "data": [
{"id": "topic-existing", "name": "swarm-wrong"},
]}),
])
client = _make_client(transport)
fake_response = mocker.MagicMock() with pytest.raises(OpusAgentError, match="conflict but not found"):
fake_response.choices = [ client.complete(make_genome(ModelTier.C), system="sys", user="usr")
mocker.MagicMock(message=mocker.MagicMock(content="(strategy ...)"))
]
fake_response.usage = mocker.MagicMock(prompt_tokens=100, completion_tokens=200)
fake_openai = mocker.MagicMock()
fake_openai.chat.completions.create.side_effect = [
openai.APITimeoutError(request=mocker.MagicMock()),
fake_response,
]
mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
client = LLMClient(openrouter_api_key="or-x") def test_empty_response_raises():
g = make_genome(ModelTier.C) transport = _mock_transport([
out = client.complete(g, system="sys", user="usr") httpx.Response(201, json=TOPIC_RESPONSE),
httpx.Response(202, json=REQUEST_ACCEPTED),
httpx.Response(200, json={
"success": True,
"data": {"id": "req-456", "status": "completed", "result": ""},
}),
])
client = _make_client(transport)
assert isinstance(out, CompletionResult) with pytest.raises(EmptyCompletionError):
assert out.text == "(strategy ...)" client.complete(make_genome(ModelTier.C), system="sys", user="usr")
assert fake_openai.chat.completions.create.call_count == 2
def test_request_failed_raises_opus_agent_error():
transport = _mock_transport([
httpx.Response(201, json=TOPIC_RESPONSE),
httpx.Response(202, json=REQUEST_ACCEPTED),
httpx.Response(200, json={
"success": True,
"data": {"id": "req-456", "status": "failed", "error": "model overloaded"},
}),
])
client = _make_client(transport)
with pytest.raises(OpusAgentError, match="failed"):
client.complete(make_genome(ModelTier.C), system="sys", user="usr")
def test_rate_limit_429_is_retryable():
transport = _mock_transport([
httpx.Response(201, json=TOPIC_RESPONSE),
httpx.Response(429, json={"success": False}),
httpx.Response(202, json=REQUEST_ACCEPTED),
httpx.Response(200, json=_completed_response("after retry")),
])
client = _make_client(transport)
out = client.complete(make_genome(ModelTier.C), system="sys", user="usr")
assert out.text == "after retry"
def test_polling_waits_for_completion():
transport = _mock_transport([
httpx.Response(201, json=TOPIC_RESPONSE),
httpx.Response(202, json=REQUEST_ACCEPTED),
httpx.Response(200, json={
"success": True,
"data": {"id": "req-456", "status": "processing"},
}),
httpx.Response(200, json={
"success": True,
"data": {"id": "req-456", "status": "processing"},
}),
httpx.Response(200, json=_completed_response("done")),
])
client = _make_client(transport)
out = client.complete(make_genome(ModelTier.C), system="sys", user="usr")
assert out.text == "done"
def test_tokens_are_zero():
transport = _mock_transport([
httpx.Response(201, json=TOPIC_RESPONSE),
httpx.Response(202, json=REQUEST_ACCEPTED),
httpx.Response(200, json=_completed_response()),
])
client = _make_client(transport)
out = client.complete(make_genome(ModelTier.C), system="sys", user="usr")
assert out.input_tokens == 0
assert out.output_tokens == 0