feat: Prompt-Cache nutzbar und messbar; Grundsatzfrage Einzelaufruf gegen Stufenkette

OpenRouter-Aufrufe eines Laufs tragen eine gemeinsame session_id (Sticky Routing,
damit der Anbieter-Cache greifen kann); Cache-Treffer werden je Aufruf und Stufe
erfasst und im Log als Anteil ausgewiesen. docs/Konzeptionelle_Fragen_und_Antworten.md
hält Pipeline-Stufen und die Abwägung ein Aufruf vs. Kette mit Messdaten und Quellen fest.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Dieter Schlüter 2026-09-17 15:44:37 +02:00
commit eadf0608ce
5 changed files with 221 additions and 6 deletions

View file

@ -202,11 +202,14 @@ def _log_usage(session, result: PipelineResult) -> None:
result.usage = session.usage result.usage = session.usage
cost = f", Kosten ${session.usage.cost_usd:.4f}" if session.usage.cost_usd is not None else "" cost = f", Kosten ${session.usage.cost_usd:.4f}" if session.usage.cost_usd is not None else ""
providers = f", Anbieter: {session.usage.providers}" if session.usage.providers else "" providers = f", Anbieter: {session.usage.providers}" if session.usage.providers else ""
log(f"LLM-Nutzung: {session.usage.calls} Aufrufe, {session.usage.prompt_tokens} Input-/" usage = session.usage
f"{session.usage.completion_tokens} Output-Tokens (davon {session.usage.reasoning_tokens} Reasoning){cost}{providers}") cache = (f" (davon {usage.cached_tokens} aus dem Cache, {100 * usage.cached_tokens / usage.prompt_tokens:.0f} %)"
if usage.prompt_tokens else "")
log(f"LLM-Nutzung: {usage.calls} Aufrufe, {usage.prompt_tokens} Input-{cache}/"
f"{usage.completion_tokens} Output-Tokens (davon {usage.reasoning_tokens} Reasoning){cost}{providers}")
for step, data in session.usage.by_step().items(): for step, data in session.usage.by_step().items():
log(f" {step:18} {data['calls']}x {data['seconds']:6.1f} s {data['completion_tokens']:6} Output-Tokens " log(f" {step:18} {data['calls']}x {data['seconds']:6.1f} s {data['completion_tokens']:6} Output-Tokens "
f"(davon {data['reasoning_tokens']} Reasoning)") f"(davon {data['reasoning_tokens']} Reasoning) {data['prompt_tokens']:6} Input (Cache {data['cached_tokens']})")
def run_pipeline( def run_pipeline(

View file

@ -8,6 +8,7 @@ import os
import sys import sys
import threading import threading
import time import time
import uuid
import instructor import instructor
@ -98,6 +99,10 @@ class LLMSession:
self.client = self._clients[self.provider] self.client = self._clients[self.provider]
self.usage = LLMUsage(model=model, provider=self.provider) self.usage = LLMUsage(model=model, provider=self.provider)
self._lock = threading.Lock() self._lock = threading.Lock()
# Eine Kennung je Pipeline-Lauf: OpenRouter leitet alle Aufrufe dann an denselben Anbieter (Sticky
# 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
def overrides(self) -> dict: def overrides(self) -> dict:
"""Stufen mit abweichendem Modell – für die Fortschrittsausgabe.""" """Stufen mit abweichendem Modell – für die Fortschrittsausgabe."""
@ -113,7 +118,7 @@ class LLMSession:
kwargs["extra_body"] = {"provider": { kwargs["extra_body"] = {"provider": {
"require_parameters": True, "require_parameters": True,
"preferred_min_throughput": {"p50": OPENROUTER_MIN_THROUGHPUT}, "preferred_min_throughput": {"p50": OPENROUTER_MIN_THROUGHPUT},
}} }, "session_id": self.session_id}
if REASONING_EFFORT: if REASONING_EFFORT:
kwargs["extra_body"]["reasoning"] = {"effort": REASONING_EFFORT} kwargs["extra_body"]["reasoning"] = {"effort": REASONING_EFFORT}
@ -130,7 +135,8 @@ class LLMSession:
time.sleep(wait) time.sleep(wait)
def _record(self, completion, step: str = "unbenannt", seconds: float = 0.0, model: str = None) -> None: def _record(self, completion, step: str = "unbenannt", seconds: float = 0.0, model: str = None) -> None:
before = (self.usage.completion_tokens, self.usage.reasoning_tokens) before = (self.usage.completion_tokens, self.usage.reasoning_tokens, self.usage.prompt_tokens,
self.usage.cached_tokens)
with self._lock: with self._lock:
self.usage.calls += 1 self.usage.calls += 1
provider = getattr(completion, "provider", None) # OpenRouter: tatsächlich antwortender Anbieter provider = getattr(completion, "provider", None) # OpenRouter: tatsächlich antwortender Anbieter
@ -142,6 +148,8 @@ class LLMSession:
self.usage.completion_tokens += usage.completion_tokens or 0 self.usage.completion_tokens += usage.completion_tokens or 0
details = getattr(usage, "completion_tokens_details", None) details = getattr(usage, "completion_tokens_details", None)
self.usage.reasoning_tokens += getattr(details, "reasoning_tokens", 0) or 0 self.usage.reasoning_tokens += getattr(details, "reasoning_tokens", 0) or 0
prompt_details = getattr(usage, "prompt_tokens_details", None)
self.usage.cached_tokens += getattr(prompt_details, "cached_tokens", 0) or 0
cost = getattr(usage, "cost", None) cost = getattr(usage, "cost", None)
if cost is not None: if cost is not None:
self.usage.cost_usd = (self.usage.cost_usd or 0.0) + cost self.usage.cost_usd = (self.usage.cost_usd or 0.0) + cost
@ -152,6 +160,7 @@ class LLMSession:
self.usage.prompt_tokens += meta.prompt_token_count or 0 self.usage.prompt_tokens += meta.prompt_token_count or 0
self.usage.completion_tokens += meta.candidates_token_count or 0 self.usage.completion_tokens += meta.candidates_token_count or 0
self.usage.reasoning_tokens += meta.thoughts_token_count or 0 self.usage.reasoning_tokens += meta.thoughts_token_count or 0
self.usage.cached_tokens += getattr(meta, "cached_content_token_count", 0) or 0
self._log_call(step, seconds, before, getattr(completion, "provider", None), model) self._log_call(step, seconds, before, getattr(completion, "provider", None), model)
def _log_call(self, step, seconds, before, provider, model=None) -> None: def _log_call(self, step, seconds, before, provider, model=None) -> None:
@ -159,4 +168,6 @@ class LLMSession:
step=step, seconds=round(seconds, 1), provider=provider, model=model, step=step, seconds=round(seconds, 1), provider=provider, model=model,
completion_tokens=self.usage.completion_tokens - before[0], completion_tokens=self.usage.completion_tokens - before[0],
reasoning_tokens=self.usage.reasoning_tokens - before[1], reasoning_tokens=self.usage.reasoning_tokens - before[1],
prompt_tokens=self.usage.prompt_tokens - before[2],
cached_tokens=self.usage.cached_tokens - before[3],
)) ))

View file

@ -395,6 +395,8 @@ class CallRecord(BaseModel):
seconds: float seconds: float
completion_tokens: int = 0 completion_tokens: int = 0
reasoning_tokens: int = 0 reasoning_tokens: int = 0
prompt_tokens: int = 0
cached_tokens: int = Field(default=0, description="Input-Tokens aus dem Anbieter-Cache (billiger)")
provider: Optional[str] = None provider: Optional[str] = None
model: Optional[str] = Field(default=None, description="Tatsächlich benutztes Modell (kann je Stufe abweichen)") model: Optional[str] = Field(default=None, description="Tatsächlich benutztes Modell (kann je Stufe abweichen)")
@ -406,6 +408,7 @@ class LLMUsage(BaseModel):
prompt_tokens: int = 0 prompt_tokens: int = 0
completion_tokens: int = Field(default=0, description="inkl. Reasoning-Tokens") completion_tokens: int = Field(default=0, description="inkl. Reasoning-Tokens")
reasoning_tokens: int = 0 reasoning_tokens: int = 0
cached_tokens: int = Field(default=0, description="davon Input-Tokens aus dem Anbieter-Cache")
cost_usd: Optional[float] = Field(default=None, description="nur OpenRouter liefert Kosten") cost_usd: Optional[float] = Field(default=None, description="nur OpenRouter liefert Kosten")
providers: Dict[str, int] = Field(default_factory=dict, description="Antwortender Anbieter -> Anzahl Aufrufe (OpenRouter)") providers: Dict[str, int] = Field(default_factory=dict, description="Antwortender Anbieter -> Anzahl Aufrufe (OpenRouter)")
call_records: List[CallRecord] = Field(default_factory=list) call_records: List[CallRecord] = Field(default_factory=list)
@ -414,11 +417,14 @@ class LLMUsage(BaseModel):
"""Sekunden und Tokens je Pipeline-Stufe.""" """Sekunden und Tokens je Pipeline-Stufe."""
summary: Dict[str, dict] = {} summary: Dict[str, dict] = {}
for record in self.call_records: for record in self.call_records:
entry = summary.setdefault(record.step, {"calls": 0, "seconds": 0.0, "completion_tokens": 0, "reasoning_tokens": 0}) entry = summary.setdefault(record.step, {"calls": 0, "seconds": 0.0, "completion_tokens": 0,
"reasoning_tokens": 0, "prompt_tokens": 0, "cached_tokens": 0})
entry["calls"] += 1 entry["calls"] += 1
entry["seconds"] += record.seconds entry["seconds"] += record.seconds
entry["completion_tokens"] += record.completion_tokens entry["completion_tokens"] += record.completion_tokens
entry["reasoning_tokens"] += record.reasoning_tokens entry["reasoning_tokens"] += record.reasoning_tokens
entry["prompt_tokens"] += record.prompt_tokens
entry["cached_tokens"] += record.cached_tokens
return summary return summary

View file

@ -0,0 +1,152 @@
# Konzeptionelle Fragen und Antworten
Grundsatzfragen zur Bauweise des Projekts – mit Begründung, eigenen Messdaten und Quellen, damit
Entscheidungen später nachvollziehbar bleiben.
---
## Unsere Pipeline (Stand 17.09.2026)
Sieben Stufen; Ablauf und Prüfungen gesteuert von `TextToModel.py` (`run_pipeline`, `_solver_stage`).
Alle LLM-Aufrufe laufen über `core/llm.py` (`LLMSession.create`), alle Datenstrukturen stehen in
`core/model_schema.py`.
| # | Stufe | LLM | Python-Datei (Funktion) | Prompt |
|---|---|---|---|---|
| 1 | **Verstehen** – Teilfragen und alle Zahlen erfassen; Nicht-Aufgaben ablehnen; Vollständigkeit ohne LLM prüfen | ja | `understanding/understand.py` (`understand_task`), Prüfung `checks/model_checks.py` (`check_understanding`) | `understanding/Verstehen-Prompt.txt` |
| 2 | **Extraktion** – Problemklasse, Solver, Indizes, Parameter, Variablen, Zielfunktion, Restriktionen, Annahmen | ja | `modeling/extraction.py` (`text_to_model`) | `modeling/Routing-und-Extraktions-Prompt.txt` |
| 3 | **Deterministische Checks** – Stack passt zur Klasse, Einheiten, Symbole, Datenübernahme, Fachbegriffe | nein | `checks/model_checks.py` (`check_spec`) | – |
| 4 | **LLM-Prüfung** – liefert die Zielfunktion die gefragte Größe? Sind die Annahmen textkonform? | ja | `checks/review.py` (`review_assumptions`, `needs_work`, `select_best`) | `checks/Pruef-Prompt.txt` |
| 5 | **Korrektur** – Schleife 3–5, beste Runde gewinnt, notfalls Neustart | ja | `modeling/extraction.py` (`revise_model`) | `modeling/Routing-und-Extraktions-Prompt.txt` |
| 6 | **Solver** – Python-Code erzeugen, in der Sandbox ausführen, bei Fehlern reparieren, Szenarien nachrechnen | ja | `solving/solver_stage.py` (`solve`, `run_solver_code`, `verify_scenarios`) | `solving/Solver-Prompt.txt` |
| 7 | **Abnahme** – Zulässigkeit, Antworten, Zahlen und Einheiten belegt, Datenherkunft, Szenarien mit Ergebnis | nein | `solving/solver_stage.py` (`_check_and_run`, `check_feasibility`, `check_answers`, `check_answer_units`, `check_generated_data`, `check_scenarios_solved`) | – |
Ausgabe: Report `reporting/report.py` mit Diagrammen aus `reporting/charts.py`, eigenständiges Skript
`solving/script_export.py`. Beigelegte Daten profiliert `data/data_profile.py` ohne LLM. Messung:
`benchmark/benchmark.py` mit den Fällen in `benchmark/cases/`.
---
## Frage 1: Ein oder mehrere Agenten – was ist besser?
### Die Frage (17.09.2026)
> Wir haben festgestellt, dass im GitHub-Repository MathModelAgent (Fall 1) alle Schritte hintereinander
> weg mit nur einer LLM erledigt werden, während im GitHub-Repository LLM-MM-Agent (Fall 2) mit
> verschiedenen Agenten gearbeitet wird, die auch unterschiedliche LLMs benutzen können. Ich unterstelle:
> Vermutlich sind die Ergebnisse ähnlich. Aber welche Methode würde bezogen auf unser Projekt
> zuverlässiger, schneller und günstiger funktionieren? […] Meine Überlegung: Im Fall 1 müssen alle Daten
> und Erkenntnisse nur einmal ins LLM geladen werden. Im Fall 2 müssen sie bei jedem Schritt der Pipeline
> erneut ins LLM geladen werden. Das kostet mehr Geld und Zeit. Im Fall 1 brauchen wir einen möglicherweise
> sehr langen Prompt, der möglicherweise nicht ganz korrekt ausgeführt wird. Im Fall 2 arbeiten wir mit
> verschiedenen kürzeren Prompts, die individueller auf die jeweilige Aufgabe zugeschnitten sein können.
> Sehe ich das richtig? Was habe ich vergessen?
### Kurzantwort
**Für unser Projekt ist die Stufenkette zuverlässiger, ungefähr gleich schnell und kaum teurer als ein
einzelner Aufruf – wir behalten sie.** Die Hebel liegen innerhalb der Kette: Prompt-Caching messbar machen
und nutzen, und für die Prüfung gezielt ein *anderes* Modell einsetzen.
### 1. Die Prämisse ist vertauscht – und beide Projekte sind Ketten
Im Quellcode nachgesehen:
- **MathModelAgent** hat vier Agenten (Coordinator, Modeler, Coder, Writer) mit **je eigenem Modell**
(`COORDINATOR_MODEL`, `MODELER_MODEL`, `CODER_MODEL`, `WRITER_MODEL` in `backend/.env.example`;
`llm_factory.get_all_llms()` in `backend/app/core/workflow.py`). Jeder Agent führt einen eigenen,
**wachsenden Gesprächsverlauf**; übergeben wird ein Objekt (`CoordinatorToModeler`, `ModelerToCoder`).
- **LLM-MM-Agent** nutzt **ein Modell** (`--model_name`, Standard gpt-4o) für alle Agenten, aber in vielen
getrennten Aufrufen (27 Aufrufstellen `llm.generate(prompt, system)`). Jeder Aufruf beginnt frisch,
Vorwissen geht als zusammengefasster Text („memory“) mit, dazu Erzeugen–Kritisieren–Verbessern je Stufe.
- **Wir** arbeiten mit einer Kette frischer Aufrufe, **strukturierter** Übergabe (Schemas), einem Modell
(Modell je Stufe möglich über `LLM_MODEL_EXTRACTION`, `LLM_MODEL_REVIEW`, `LLM_MODEL_REVISION`,
`LLM_MODEL_SOLVER` – nie gemessen) und **Prüfungen ohne LLM zwischen den Stufen**.
„Alles in einem Aufruf“ macht keines der Projekte. Es gibt drei unabhängige Unterscheidungen, nicht eine:
1. ein Aufruf **oder** eine Kette von Aufrufen,
2. wachsender Gesprächsverlauf **oder** frischer Aufruf mit Übergabeobjekt,
3. ein Modell für alles **oder** ein Modell je Stufe.
### 2. Eigene Messdaten
Letzter voller Benchmark `benchmark/results/20260917-043002.json` (26 Läufe, `deepseek/deepseek-v4.1-flash`):
- 5,8 Aufrufe je Lauf; Input 31 000, Output 33 000 Tokens je Lauf, davon **61 % Reasoning**;
$0,045 und Ø 194 s je Lauf.
- Preise per Regression über 228 Läufe geschätzt: Input ≈ $0,50/M, Output ≈ $0,78/M –
**Input ≈ 35 % der Kosten, Output ≈ 65 %**.
- Zeitanteile: Solver-Code 36 %, Extraktion 24 %, Prüfung 15 %, Verstehen 13 %, Solver-Reparatur 9 %,
Korrektur 3 %. Die Zeit folgt dem **erzeugten Output**, nicht dem gelesenen Input.
- Cache-Treffer wurden bis dahin nicht erfasst.
### 3. Die Annahmen im Einzelnen
| Annahme | Urteil | Begründung |
|---|---|---|
| Einmal laden spart Geld | **nur zum Teil** | LLM-Schnittstellen merken sich nichts zwischen Aufrufen; auch ein Gesprächsverlauf wird bei jedem Zug komplett neu gesendet und wächst dabei. Sparen lässt sich höchstens am Input-Anteil (35 %), realistisch ~20 %, und nur ohne Wiederholungen. Prompt-Caching verbilligt Wiederholtes in *beiden* Bauweisen. |
| Einmal laden spart Zeit | **kaum** | Einlesen ist schnell, Erzeugen langsam. Ein Einzelaufruf müsste dieselben ~33 000 Output-Tokens erzeugen. |
| Ein langer Prompt wird schlechter befolgt | **belegt** | IFScale: Befolgung sinkt mit der Zahl gleichzeitiger Regeln, häufigster Fehler ist das Auslassen, frühe Regeln werden bevorzugt. Chroma „Context Rot“: alle 18 getesteten Modelle werden mit wachsender Eingabe schlechter (Herausgeber hat ein Produktinteresse). Neuere Modelle sind deutlich robuster. |
| Kürzere, zugeschnittene Prompts sind besser | **belegt** | ACL 2024: getrennte Aufrufe gewinnen 77-mal, derselbe Ablauf in einem Prompt 53-mal; im Einzel-Prompt „simuliert“ das Modell die Verbesserung und baut dafür Fehler ein. |
| Die Ergebnisse sind ähnlich | **bei leichten Aufgaben ja** | Auf NL4Opt erreicht ein starkes Modell ohne Gerüst laut Sekundärquelle ähnliche oder bessere Werte als Multi-Agent-Frameworks; das Gerüst lohnt bei komplexen Aufgaben. Unser Benchmark ist gesättigt (38 von 38) und kann den Unterschied nicht zeigen. |
### 4. Was in der Überlegung fehlte – die eigentlich entscheidenden Punkte
1. **Prüftore zwischen den Stufen.** Unsere Zuverlässigkeit entsteht durch Code-Prüfungen *zwischen* den
Aufrufen: Ist der Text keine Aufgabe, endet der Lauf nach einem Aufruf; passt die Zielgröße nicht, startet
kein Solver. In einem Einzelaufruf zeigen sich alle Fehler erst am Ende, und jeder kostet den ganzen Lauf.
2. **Reparatur dort, wo der Fehler liegt.** In 26 Läufen gab es 9 Korrekturrunden und 11 Solver-Reparaturen.
Jede wiederholte nur ihre Stufe (~3 000–9 000 Output-Tokens) statt den ganzen Lauf (~30 000).
3. **Harte Grenzen.** `LLM_MAX_TOKENS` ist 32 000, ein Lauf erzeugt ~33 000 Output-Tokens – ein Einzelaufruf
liefe an die Grenze. Ein riesiges Ausgabeschema scheitert außerdem als Ganzes an einer einzigen Stelle.
4. **Selbstprüfung ist schwach.** Modelle finden ihre eigenen Fehler kaum, und Fehler sind modellspezifisch
(nur 9,6 % der festen Fehler eines Modells traten beim anderen auf). Prüfung im selben Aufruf bringt wenig,
ein frischer Aufruf mehr, **ein anderes Modell als Prüfer am meisten** – das stärkste Argument für ein
Modell je Stufe.
5. **Übergabeverlust ist das echte Risiko einer Kette.** Google/MIT 2025: Multi-Agent-Aufbauten wurden bei
aufeinander aufbauenden Aufgaben um 39–70 % schlechter; Fehler verstärkten sich ohne zentrale Prüfung um
das 17-Fache, mit zentraler Prüfung um das 4-Fache. Das betrifft frei zusammenarbeitende Agenten, nicht
feste Ketten mit Prüftoren – wir haben den Effekt aber selbst gemessen: Von 7, 10 und 14 Zahlen der Aufgabe
kam jeweils keine in der Spezifikation an, bis die Verstehensstufe und `check_data_transfer` die Übergabe
prüften.
6. **Multi-Agent ist nicht gleich Kette.** Autonome Agenten verbrauchen laut Anthropic ~15-mal so viele Tokens
wie ein Chat; eine feste Kette nicht.
7. **Messbarkeit.** Kosten und Zeit lassen sich je Stufe zuordnen, `--von solver` braucht 2 statt 5–7 Aufrufe,
Tests gibt es je Stufe.
8. **Caching wurde verschenkt.** Cache-Treffer wurden nicht erfasst, und die Anfragen landeten wechselnd bei
verschiedenen Anbietern (Modal, Together, Venice), die sich nichts teilen.
### 5. Einschätzung für unser Projekt
| Bauweise | Zuverlässigkeit | Zeit | Kosten |
|---|---|---|---|
| Ein Aufruf für alles | schlechter | gleich oder länger | −20 % ohne Fehler, mehr mit Wiederholungen |
| Kette mit wachsendem Verlauf (MathModelAgent) | etwa gleich | gleich | höher, weil der Verlauf wächst |
| **Kette mit frischen Aufrufen (wir)** | gemessen 26/26 | 194 s | $0,045 je Lauf |
| Kette mit anderem Modell für Prüfung und schnellem Modell für Code | vermutlich besser | vermutlich kürzer | ungefähr gleich |
### 6. Folgerungen
- Bauweise behalten.
- **Umgesetzt im Anschluss:** Cache-Treffer je Aufruf erfassen und die Aufrufe eines Laufs per `session_id`
beim selben OpenRouter-Anbieter halten (Sticky Routing). Messen beim nächsten ohnehin fälligen Lauf.
- **Später, mit Freigabe:** `LLM_MODEL_REVIEW` auf eine andere Modellfamilie – als A/B nur mit Aufgaben, die
Kopfraum haben (unordentliche Aufgaben), nicht auf dem gesättigten Benchmark.
### Quellen
- [Prompt Chaining or Stepwise Prompt? (ACL Findings 2024)](https://aclanthology.org/2024.findings-acl.449/)
- [IFScale – How Many Instructions Can LLMs Follow at Once?](https://arxiv.org/abs/2507.11538)
- [Arize: Nachmessung zur Befolgung von Anweisungen 2026](https://arize.com/blog/llm-instruction-following-benchmark-2026/)
- [Chroma: Context Rot](https://www.trychroma.com/research/context-rot)
- [Towards a Science of Scaling Agent Systems (Google/MIT)](https://arxiv.org/abs/2512.08296)
- [Google Research Blog dazu](https://research.google/blog/towards-a-science-of-scaling-agent-systems-when-and-why-agent-systems-work/)
- [Anthropic: Building effective agents](https://www.anthropic.com/news/building-effective-agents)
- [Anthropic Multi-Agent Research System (ZenML-Zusammenfassung)](https://www.zenml.io/llmops-database/building-a-multi-agent-research-system-for-complex-information-tasks)
- [Kamoi et al.: Self-Correction Survey (TACL)](https://direct.mit.edu/tacl/article/doi/10.1162/tacl_a_00713/125177/When-Can-LLMs-Actually-Correct-Their-Own-Mistakes)
- [Too Consistent to Detect: Self-Consistent Errors](https://arxiv.org/pdf/2505.17656)
- [OptiMUS-0.3](https://arxiv.org/pdf/2407.19633)
- [Survey: Optimization Modeling Meets LLMs](https://arxiv.org/html/2508.10047)
- [OpenRouter: Prompt Caching](https://openrouter.ai/docs/guides/best-practices/prompt-caching)
- [OpenRouter Prompt Caching in der Praxis](https://china-llm.com/blog/openrouter-prompt-caching)

View file

@ -22,6 +22,7 @@ def fake_session(model="deepseek/deepseek-v4.1-flash", step_models=None):
session.step_models = step_models or {} session.step_models = step_models or {}
session.usage = LLMUsage(model=model, provider=session.provider) session.usage = LLMUsage(model=model, provider=session.provider)
session._lock = threading.Lock() session._lock = threading.Lock()
session.session_id = "lauf-1"
completions = {} completions = {}
session._clients = {} session._clients = {}
for provider in {session.provider, *(provider_for(m) for m in session.step_models.values())}: for provider in {session.provider, *(provider_for(m) for m in session.step_models.values())}:
@ -114,3 +115,45 @@ def test_create_uses_step_model_and_records_it():
def test_overrides_lists_only_deviating_steps(): def test_overrides_lists_only_deviating_steps():
session, _ = fake_session(model="a/b", step_models={"Extraktion": "a/b", "Solver-Code": "schnell/modell"}) session, _ = fake_session(model="a/b", step_models={"Extraktion": "a/b", "Solver-Code": "schnell/modell"})
assert session.overrides() == {"Solver-Code": "schnell/modell"} assert session.overrides() == {"Solver-Code": "schnell/modell"}
# --- Prompt-Caching: Kennung je Lauf (Sticky Routing) und erfasste Cache-Treffer ---
def test_all_calls_of_a_run_share_one_openrouter_session_id():
session, completions = fake_session(step_models={"Solver-Code": "gemini-2.5-flash"})
session.create(str, [{"role": "user", "content": "x"}], step="Extraktion")
assert completions["openrouter"].kwargs["extra_body"]["session_id"] == "lauf-1"
session.create(str, [{"role": "user", "content": "x"}], step="Solver-Code")
assert "extra_body" not in completions["gemini"].kwargs # Gemini kennt den Parameter nicht
def test_new_session_gets_a_new_id(monkeypatch):
monkeypatch.setattr(llm, "_make_client", lambda provider: object())
assert llm.LLMSession("a/b").session_id != llm.LLMSession("a/b").session_id
def test_cached_tokens_are_recorded_per_call_and_step():
from types import SimpleNamespace as NS
session, _ = fake_session()
def antwort(prompt, cached):
return NS(provider="Modal", usage=NS(prompt_tokens=prompt, completion_tokens=100, cost=0.001,
completion_tokens_details=NS(reasoning_tokens=40),
prompt_tokens_details=NS(cached_tokens=cached)))
session._record(antwort(5000, 0), step="Prüfung")
session._record(antwort(5200, 4800), step="Prüfung") # zweite Runde liest den gleichen Präfix
session._record(NS(provider=None, usage=NS(prompt_tokens=900, completion_tokens=10, cost=None,
completion_tokens_details=None, prompt_tokens_details=None)),
step="Verstehen") # Anbieter ohne Cache-Angabe
assert session.usage.cached_tokens == 4800 and session.usage.prompt_tokens == 11100
assert [(r.prompt_tokens, r.cached_tokens) for r in session.usage.call_records] == [(5000, 0), (5200, 4800), (900, 0)]
stufen = session.usage.by_step()
assert stufen["Prüfung"]["prompt_tokens"] == 10200 and stufen["Prüfung"]["cached_tokens"] == 4800
def test_old_results_without_cache_fields_still_load():
from core.model_schema import LLMUsage
alt = {"model": "a/b", "provider": "openrouter", "calls": 1, "prompt_tokens": 10, "completion_tokens": 5,
"reasoning_tokens": 0, "call_records": [{"step": "Verstehen", "seconds": 1.0, "completion_tokens": 5}]}
usage = LLMUsage.model_validate(alt)
assert usage.cached_tokens == 0 and usage.call_records[0].cached_tokens == 0