diff --git a/BEDIENUNGSANLEITUNG.md b/BEDIENUNGSANLEITUNG.md index f5fa52b..e8c2a08 100644 --- a/BEDIENUNGSANLEITUNG.md +++ b/BEDIENUNGSANLEITUNG.md @@ -446,6 +446,7 @@ Wo das Geld hingeht: rund zwei Drittel der Kosten sind **erzeugter Output**, dav | `LLM_TIMEOUT` | Zeitlimit je LLM-Aufruf in Sekunden | 420 | | `LLM_REASONING_EFFORT` | `low`, `medium` oder `high` (OpenRouter) | Modellvorgabe | | `OPENROUTER_MIN_THROUGHPUT` | bevorzugte Mindestgeschwindigkeit des Anbieters in Tokens/s | 75 | +| `OPENROUTER_SWITCH_BELOW` | liefert ein Aufruf (ab 1000 Tokens) weniger Tokens/s, wird der Anbieter für den Rest des Laufs gemieden; `0` schaltet das ab | wie `OPENROUTER_MIN_THROUGHPUT` (75) | | `SOLVER_TIMEOUT` | Zeitlimit je Programmlauf in Sekunden; zugleich das Zeitbudget, das der Solver-Prompt dem Programm nennt | `60` | | `SOLVER_SANDBOX` | `auto` (bubblewrap, wenn vorhanden), `bwrap` (Pflicht), `aus` (ungeschützt) | `auto` | diff --git a/README.md b/README.md index 8308727..bd2db8e 100644 --- a/README.md +++ b/README.md @@ -113,6 +113,7 @@ Exit-Codes: `0` Ergebnis zuverlässig, `2` Ergebnis vorhanden, aber unzuverläss | `LLM_MAX_TOKENS` | Obergrenze je Antwort (OpenRouter) | 32000 | | `--max-restarts` / `MAX_RESTARTS` | komplette Neustarts, wenn das Modell nach allen Korrekturrunden fehlerhaft ist | 1 | | `OPENROUTER_MIN_THROUGHPUT` | bevorzugte Mindestgeschwindigkeit des OpenRouter-Anbieters (Tokens/s); Präferenz, kein harter Filter | 75 | +| `OPENROUTER_SWITCH_BELOW` | liefert ein Aufruf (ab 1000 Tokens) weniger Tokens/s, wird der Anbieter für den Rest des Laufs gemieden; `0` schaltet das ab | wie `OPENROUTER_MIN_THROUGHPUT` (75) | | `LLM_TIMEOUT` | Timeout je LLM-Aufruf (s) | 420 | | `LLM_REASONING_EFFORT` | Denkaufwand des Modells über OpenRouter: `low`, `medium` oder `high` | Modellvorgabe | | `SOLVER_TIMEOUT` | Zeitlimit je Programmlauf in Sekunden; zugleich das Zeitbudget, das der Solver-Prompt dem Programm nennt | `60` | diff --git a/core/llm.py b/core/llm.py index c5ed520..601839c 100644 --- a/core/llm.py +++ b/core/llm.py @@ -35,6 +35,14 @@ OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1" # Es ist eine Präferenz, kein harter Filter: gibt es keinen so schnellen Anbieter mit JSON-Schema, # leitet OpenRouter trotzdem weiter (geprüft mit qwen3-coder -> SiliconFlow, ~25 Tokens/s). OPENROUTER_MIN_THROUGHPUT = int(os.environ.get("OPENROUTER_MIN_THROUGHPUT", "75")) # Tokens/s (p50) +# Sticky Routing (session_id) hält einen Lauf bei einem Anbieter – auch bei einem langsamen. Beobachtet an +# wagner_whitin_dijkstra (18.09.2026): alle Aufrufe bei AtlasCloud mit ~65 Tokens/s, 1575 s statt ~150 s. Liefert +# ein Aufruf mit mindestens SWITCH_MIN_TOKENS weniger als diese Rate, wird der Anbieter für den Rest des Laufs +# gemieden und eine neue session_id vergeben. Gemessen an 1012 Aufrufen: Median 156 Tokens/s, 3,8 % unter 75. +# 0 schaltet das ab. Der Cache der Folgeaufrufe geht dabei verloren – er bringt nur bei wiederholten Stufen etwas. +OPENROUTER_SWITCH_BELOW = int(os.environ.get("OPENROUTER_SWITCH_BELOW", str(OPENROUTER_MIN_THROUGHPUT))) +SWITCH_MIN_TOKENS = 1000 # darunter misst die Rate vor allem die Wartezeit bis zum ersten Token +MAX_IGNORED_PROVIDERS = 3 # sonst bliebe womöglich kein Anbieter mit JSON-Schema übrig LLM_TIMEOUT = int(os.environ.get("LLM_TIMEOUT", "420")) # Sekunden je Aufruf # Ohne Vorgabe schneiden manche OpenRouter-Anbieter die Antwort ab (beobachtet: abgebrochene # Modellspezifikation in einer Korrekturrunde). Großzügig setzen; bezahlt werden nur erzeugte Tokens. @@ -64,6 +72,25 @@ def is_transient_error(message: str) -> bool: return not is_fatal_error(message) and any(marker in message for marker in TRANSIENT_MARKERS) +def discarded_completions(exc: BaseException) -> list: + """Antworten, die ein Fehler verworfen hat – bezahlt, aber ohne Ergebnis. + + instructor hängt sie an: `IncompleteOutputException.last_completion` (an LLM_MAX_TOKENS abgeschnitten) und + `InstructorRetryException.failed_attempts[*].completion` (Schema verletzt). Beobachtet an wagner_whitin_dijkstra + (18.09.2026): eine abgeschnittene Solver-Antwort mit ~32.000 Tokens fehlte in Kosten und Tokenzählung. + """ + gefunden = [] + for fehler in (exc, exc.__cause__): + if fehler is None: + continue + kandidaten = [getattr(fehler, "last_completion", None)] + kandidaten += [getattr(a, "completion", None) for a in getattr(fehler, "failed_attempts", None) or []] + for c in kandidaten: + if c is not None and all(c is not g for g in gefunden): + gefunden.append(c) + return gefunden + + def provider_for(model: str) -> str: return "openrouter" if "/" in model else "gemini" @@ -107,37 +134,70 @@ class LLMSession: # Routing), dessen Prompt-Cache wiederholte Präfixe billiger liest. Ohne sie wechselten die Anbieter # je Aufruf (beobachtet: Modal, Together, Venice in einem Lauf) und kein Cache konnte greifen. self.session_id = uuid.uuid4().hex + self.ignored_providers = [] # in diesem Lauf zu langsame Anbieter (siehe OPENROUTER_SWITCH_BELOW) def overrides(self) -> dict: """Stufen mit abweichendem Modell – für die Fortschrittsausgabe.""" return {step: m for step, m in self.step_models.items() if m != self.model} + def _kwargs(self, model: str, response_model, messages) -> dict: + kwargs = {"model": model, "response_model": response_model, "messages": messages} + if provider_for(model) == "openrouter": + kwargs["max_tokens"] = LLM_MAX_TOKENS + # nur Anbieter wählen, die strukturierte Ausgabe (response_format) wirklich unterstützen + anbieter = {"require_parameters": True, "preferred_min_throughput": {"p50": OPENROUTER_MIN_THROUGHPUT}} + if self.ignored_providers: + anbieter["ignore"] = list(self.ignored_providers) + kwargs["extra_body"] = {"provider": anbieter, "session_id": self.session_id} + if REASONING_EFFORT: + kwargs["extra_body"]["reasoning"] = {"effort": REASONING_EFFORT} + return kwargs + def create(self, response_model, messages, step: str = "unbenannt"): model = self.step_models.get(step, self.model) provider = provider_for(model) - kwargs = {"model": model, "response_model": response_model, "messages": messages} - if provider == "openrouter": - kwargs["max_tokens"] = LLM_MAX_TOKENS - # nur Anbieter wählen, die strukturierte Ausgabe (response_format) wirklich unterstützen - kwargs["extra_body"] = {"provider": { - "require_parameters": True, - "preferred_min_throughput": {"p50": OPENROUTER_MIN_THROUGHPUT}, - }, "session_id": self.session_id} - if REASONING_EFFORT: - kwargs["extra_body"]["reasoning"] = {"effort": REASONING_EFFORT} - - for wait in (*RETRY_WAITS, None): + waits = list(RETRY_WAITS) + while True: + kwargs = self._kwargs(model, response_model, messages) started = time.monotonic() try: result, completion = self._clients[provider].chat.completions.create_with_completion(**kwargs) self._record(completion, step, time.monotonic() - started, model) + if provider == "openrouter": + self._avoid_if_slow() return result except Exception as exc: - if wait is None or not is_transient_error(str(exc)): + # Verworfene Antworten sind bezahlt – sonst fehlen sie in Kosten, Tokens und Aufrufzahl + for verworfen in discarded_completions(exc): + self._record(verworfen, f"{step} (verworfen)", time.monotonic() - started, model) + if "No endpoints found" in str(exc) and self.ignored_providers: + # die Sperrliste hat alle passenden Anbieter ausgeschlossen – lieber langsam als gar nicht + print(f"Kein Anbieter ohne {self.ignored_providers} verfügbar – Sperrliste aufgehoben.", + file=sys.stderr) + self.ignored_providers.clear() + continue + if not waits or not is_transient_error(str(exc)): raise + wait = waits.pop(0) print(f"Vorübergehender LLM-Fehler – neuer Versuch in {wait} s: {str(exc)[:120]}", file=sys.stderr) time.sleep(wait) + def _avoid_if_slow(self) -> None: + """War der letzte Aufruf zu langsam, den Anbieter für den Rest des Laufs meiden (siehe OPENROUTER_SWITCH_BELOW).""" + if OPENROUTER_SWITCH_BELOW <= 0 or not self.usage.call_records: + return + letzter = self.usage.call_records[-1] + if (not letzter.provider or letzter.completion_tokens < SWITCH_MIN_TOKENS or letzter.seconds <= 0 + or letzter.provider in self.ignored_providers or len(self.ignored_providers) >= MAX_IGNORED_PROVIDERS): + return + rate = letzter.completion_tokens / letzter.seconds + if rate >= OPENROUTER_SWITCH_BELOW: + return + self.ignored_providers.append(letzter.provider) + self.session_id = uuid.uuid4().hex # Sticky Routing an den neuen Anbieter binden + print(f"Anbieter {letzter.provider} zu langsam ({rate:.0f} Tokens/s < {OPENROUTER_SWITCH_BELOW}) – " + f"für den Rest des Laufs gemieden.", file=sys.stderr) + def _record(self, completion, step: str = "unbenannt", seconds: float = 0.0, model: str = None) -> None: before = (self.usage.completion_tokens, self.usage.reasoning_tokens, self.usage.prompt_tokens, self.usage.cached_tokens) diff --git a/tests/test_llm.py b/tests/test_llm.py index 5d2ee60..76b16f0 100644 --- a/tests/test_llm.py +++ b/tests/test_llm.py @@ -23,6 +23,7 @@ def fake_session(model="deepseek/deepseek-v4.1-flash", step_models=None): session.usage = LLMUsage(model=model, provider=session.provider) session._lock = threading.Lock() session.session_id = "lauf-1" + session.ignored_providers = [] completions = {} session._clients = {} for provider in {session.provider, *(provider_for(m) for m in session.step_models.values())}: @@ -189,3 +190,111 @@ def test_missing_or_odd_model_field_is_ignored(): session._record(NS(provider=None, model=object(), usage=NS(prompt_tokens=1, completion_tokens=1, cost=None, completion_tokens_details=None, prompt_tokens_details=None))) assert session.usage.responded_models == {} and session.usage.call_records[0].responded_model is None + + +def _antwort_mit_verbrauch(completion_tokens): + from types import SimpleNamespace as NS + return NS(provider="AtlasCloud", model="deepseek/deepseek-v4.1-flash", + usage=NS(prompt_tokens=8000, completion_tokens=completion_tokens, cost=0.02, + completion_tokens_details=NS(reasoning_tokens=completion_tokens - 500), + prompt_tokens_details=NS(cached_tokens=0))) + + +def test_truncated_answer_is_counted_before_the_retry(monkeypatch): + # beobachtet an wagner_whitin_dijkstra: an LLM_MAX_TOKENS abgeschnitten, bezahlt, aber nirgends gezählt + from instructor.core.exceptions import IncompleteOutputException + monkeypatch.setattr(llm, "RETRY_WAITS", (0,)) + session, completions = fake_session() + versuche = iter([IncompleteOutputException(last_completion=_antwort_mit_verbrauch(32000)), + ("ok", _antwort_mit_verbrauch(9000))]) + + def create_with_completion(**kwargs): + naechster = next(versuche) + if isinstance(naechster, Exception): + raise naechster + return naechster + + completions["openrouter"].create_with_completion = create_with_completion + assert session.create(str, [{"role": "user", "content": "x"}], step="Solver-Code") == "ok" + assert session.usage.calls == 2 and session.usage.completion_tokens == 41000 + assert session.usage.cost_usd == pytest.approx(0.04) + assert [r.step for r in session.usage.call_records] == ["Solver-Code (verworfen)", "Solver-Code"] + + +def test_rejected_attempts_are_counted_even_when_the_call_fails(monkeypatch): + from types import SimpleNamespace as NS + from instructor.core.exceptions import InstructorRetryException + session, completions = fake_session() + fehler = InstructorRetryException("Schema verletzt", last_completion=_antwort_mit_verbrauch(3000), n_attempts=1, + total_usage=0, failed_attempts=[NS(completion=_antwort_mit_verbrauch(3000))]) + + def create_with_completion(**kwargs): + raise fehler + + completions["openrouter"].create_with_completion = create_with_completion + with pytest.raises(InstructorRetryException): + session.create(str, [{"role": "user", "content": "x"}], step="Prüfung") + assert session.usage.calls == 2 and session.usage.completion_tokens == 6000 # zwei verschiedene Antworten + + +def test_ordinary_errors_have_no_discarded_answers(): + assert llm.discarded_completions(RuntimeError("Error code: 503")) == [] + + +# --- Langsamen Anbieter nicht festhalten (OPENROUTER_SWITCH_BELOW) --- + +def _aufruf(session, anbieter, tokens, sekunden): + from types import SimpleNamespace as NS + session._record(NS(provider=anbieter, usage=NS(prompt_tokens=100, completion_tokens=tokens, cost=None, + completion_tokens_details=None, prompt_tokens_details=None)), + step="Verstehen", seconds=sekunden) + session._avoid_if_slow() + + +def test_slow_provider_is_avoided_for_the_rest_of_the_run(monkeypatch): + # beobachtet: alle Aufrufe bei AtlasCloud mit ~65 Tokens/s, 1575 s statt ~150 s + monkeypatch.setattr(llm, "OPENROUTER_SWITCH_BELOW", 75) + session, completions = fake_session() + _aufruf(session, "AtlasCloud", 15201, 232.0) + assert session.ignored_providers == ["AtlasCloud"] and session.session_id != "lauf-1" + session.create(str, [{"role": "user", "content": "x"}], step="Extraktion") + body = completions["openrouter"].kwargs["extra_body"] + assert body["provider"]["ignore"] == ["AtlasCloud"] and body["session_id"] == session.session_id + + +@pytest.mark.parametrize("anbieter, tokens, sekunden", [ + ("Together", 15000, 60.0), # schnell + ("Fireworks", 400, 20.0), # zu kurz, um die Rate zu beurteilen (Wartezeit bis zum ersten Token) +]) +def test_fast_or_short_calls_keep_the_provider(monkeypatch, anbieter, tokens, sekunden): + monkeypatch.setattr(llm, "OPENROUTER_SWITCH_BELOW", 75) + session, _ = fake_session() + _aufruf(session, anbieter, tokens, sekunden) + assert session.ignored_providers == [] and session.session_id == "lauf-1" + + +def test_switching_can_be_disabled_and_is_capped(monkeypatch): + monkeypatch.setattr(llm, "OPENROUTER_SWITCH_BELOW", 0) + session, _ = fake_session() + _aufruf(session, "AtlasCloud", 15000, 230.0) + assert session.ignored_providers == [] + monkeypatch.setattr(llm, "OPENROUTER_SWITCH_BELOW", 75) + for anbieter in ("A", "B", "C", "D"): + _aufruf(session, anbieter, 15000, 230.0) + assert session.ignored_providers == ["A", "B", "C"] # höchstens MAX_IGNORED_PROVIDERS + + +def test_no_endpoints_left_lifts_the_ignore_list(): + session, completions = fake_session() + session.ignored_providers = ["AtlasCloud", "Parasail"] + gesendet = [] + + def create_with_completion(**kwargs): + gesendet.append(kwargs["extra_body"]["provider"].get("ignore")) + if len(gesendet) == 1: + raise RuntimeError("Error code: 404 - {'error': {'message': 'No endpoints found for deepseek/x.'}}") + return "ok", object() + + completions["openrouter"].create_with_completion = create_with_completion + assert session.create(str, [{"role": "user", "content": "x"}], step="Prüfung") == "ok" + assert gesendet == [["AtlasCloud", "Parasail"], None] and session.ignored_providers == []