Compare commits

...
Author SHA1 Message Date
6ecfdf6ae0 feat: Zielwert-Zerlegung als Reparaturgrund, SOLVER_TIMEOUT, Summenzeilen-Fehlalarm behoben
- check_objective_decomposition: bei LP/MILP/CP muss eine objective_column den Zielwert
  belegen, eine abweichende Beitragsspalte ist überall ein Befund -> Solver-Reparatur,
  bleibt es: Warnung. 90 von 90 linearen Läufen seit 17.09. hätten bestanden.
- Summenzeilen: „Gesamt…“/„Nutzen…“ nur noch in der letzten Zeile, einzelne Zeile nie,
  Zwischensummen nicht gegen die Gesamtsumme. Vorher 15 Rucksack-Läufe mit falschem
  „Summe (0) weicht vom Zielwert (51) ab“; von 17 Abweichungen bleibt 1 echte.
- SOLVER_TIMEOUT (Default 60) für alle Programmläufe und als {ZEITLIMIT} im Solver-Prompt
- Benchmark: dirty zählt nur versionierte Dateien; SOLVER_TIMEOUT in der Konfiguration

Ändert die Prompt-Kennung (Solver-Prompt.txt).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-18 13:24:46 +02:00
8dd5073cd1 security: Sandbox verbirgt API-Schlüssel und Home-Verzeichnis
Erzeugter Code las über pandas.io.common.os – ohne import os, an validate_imports
vorbei – den OPENROUTER_API_KEY und das gesamte Home-Verzeichnis (nachgewiesen).
Kein Netz, aber alles Gedruckte landet in Ergebnis-JSON und Report. In gespeicherten
Ergebnissen fand sich kein Schlüssel.

- Umgebung des Programms ist eine Positivliste (solver_env), auch bei SOLVER_SANDBOX=aus
- bubblewrap verdeckt das Home per --tmpfs; nur sys.prefix/sys.base_prefix wieder lesbar

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-18 13:23:53 +02:00
3c4d2fb92c fix: Nebenpfade behalten Kontext – Gegenprobe, --von, --resolve, UNDERSTAND_RETRIES=-1
- UNDERSTAND_RETRIES=-1 (A/B-Arm der Verstehensstufe) stürzte seit dem Ablehnungstor ab
- Unlösbar-Gegenprobe bekommt Verständnis, Datenprofil und Datendateien; ihre Lösung
  durchläuft dieselbe Abnahme (_assess_run)
- --von pruefung behält Verständnis, Abbildungen und Datenprofil
- --resolve überschreibt frühere Befunde nicht mehr (nur Zulässigkeit neu)
- --daten wirkt auch mit --resolve und --von
- Prüfung und Korrektur sehen das Datenprofil
- Reparaturmeldung nennt alle Befundarten zugleich
- Test der Balkenhöhe prüft die Höhe tatsächlich; unbenutzte Imports entfernt

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-18 13:22:53 +02:00
15 changed files with 483 additions and 119 deletions

View file

@ -375,6 +375,8 @@ Ist das Modell gut, aber das Programm gescheitert, muss nicht alles neu bezahlt
- Der Aufgabentext kommt aus dem gespeicherten Ergebnis; keine Aufgabendatei angeben. - Der Aufgabentext kommt aus dem gespeicherten Ergebnis; keine Aufgabendatei angeben.
- Die deterministischen Checks laufen immer neu. Passt die Zielgröße laut Prüfbericht nicht, startet auch hier kein Solver. - Die deterministischen Checks laufen immer neu. Passt die Zielgröße laut Prüfbericht nicht, startet auch hier kein Solver.
- `--von` und `--resolve` schließen sich aus; eine abgelehnte Aufgabe lässt sich nicht fortsetzen. - `--von` und `--resolve` schließen sich aus; eine abgelehnte Aufgabe lässt sich nicht fortsetzen.
- Verständnis (Teilfragen) und Abbildungen bleiben erhalten. **Beigelegte Datendateien** sind im Ergebnis nur als Profil gespeichert: bei `--von` und `--resolve` mit `--daten datei.csv …` erneut angeben, sonst findet das Programm sie nicht.
- `--resolve` prüft nur die Zulässigkeit neu; frühere Befunde (z. B. falsche Zielgröße) bleiben bestehen.
### Report nachträglich erzeugen ### Report nachträglich erzeugen
@ -444,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_TIMEOUT` | Zeitlimit je LLM-Aufruf in Sekunden | 420 |
| `LLM_REASONING_EFFORT` | `low`, `medium` oder `high` (OpenRouter) | Modellvorgabe | | `LLM_REASONING_EFFORT` | `low`, `medium` oder `high` (OpenRouter) | Modellvorgabe |
| `OPENROUTER_MIN_THROUGHPUT` | bevorzugte Mindestgeschwindigkeit des Anbieters in Tokens/s | 75 | | `OPENROUTER_MIN_THROUGHPUT` | bevorzugte Mindestgeschwindigkeit des Anbieters in Tokens/s | 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` | | `SOLVER_SANDBOX` | `auto` (bubblewrap, wenn vorhanden), `bwrap` (Pflicht), `aus` (ungeschützt) | `auto` |
Beispiel für einen einzelnen Lauf mit geänderten Einstellungen: Beispiel für einen einzelnen Lauf mit geänderten Einstellungen:
@ -514,7 +517,7 @@ Mit dem Gemini-Gratiskontingent ist der Benchmark praktisch nicht durchführbar;
## 14. Sicherheit und Datenschutz ## 14. Sicherheit und Datenschutz
- **Der Aufgabentext und das Datenprofil werden an den LLM-Anbieter gesendet** (OpenRouter und die dahinterliegenden Modellanbieter bzw. Google). Keine vertraulichen oder personenbezogenen Daten eingeben, die diesen Weg nicht gehen dürfen. Datendateien selbst verlassen den Rechner nicht, wohl aber ihr Profil mit Beispielwerten. - **Der Aufgabentext und das Datenprofil werden an den LLM-Anbieter gesendet** (OpenRouter und die dahinterliegenden Modellanbieter bzw. Google). Keine vertraulichen oder personenbezogenen Daten eingeben, die diesen Weg nicht gehen dürfen. Datendateien selbst verlassen den Rechner nicht, wohl aber ihr Profil mit Beispielwerten.
- **Der Solver-Code stammt von einem LLM.** Er läuft in einer bubblewrap-Sandbox: kein Netzzugang, Dateisystem nur lesbar, beschreibbar nur das Temp-Verzeichnis des Laufs. Zusätzlich sind nur Mathematik- und Solver-Bibliotheken erlaubt, `open`, `eval` und `exec` verboten, und es gilt ein Zeitlimit. - **Der Solver-Code stammt von einem LLM.** Er läuft in einer bubblewrap-Sandbox: kein Netzzugang, Dateisystem nur lesbar, das Home-Verzeichnis unsichtbar, beschreibbar nur das Temp-Verzeichnis des Laufs. API-Schlüssel und andere Umgebungsvariablen bekommt das Programm nicht zu sehen. Zusätzlich sind nur Mathematik- und Solver-Bibliotheken erlaubt, `open`, `eval` und `exec` verboten, und es gilt ein Zeitlimit.
- **Ohne bubblewrap läuft der Code ungeschützt mit Ihren Benutzerrechten – und zwar ohne Warnung** (Standard `SOLVER_SANDBOX=auto`). Prüfen mit `which bwrap`. Wer sichergehen will, setzt `export SOLVER_SANDBOX=bwrap`: Dann wird ein Lauf ohne bubblewrap abgelehnt, statt ungeschützt ausgeführt. - **Ohne bubblewrap läuft der Code ungeschützt mit Ihren Benutzerrechten – und zwar ohne Warnung** (Standard `SOLVER_SANDBOX=auto`). Prüfen mit `which bwrap`. Wer sichergehen will, setzt `export SOLVER_SANDBOX=bwrap`: Dann wird ein Lauf ohne bubblewrap abgelehnt, statt ungeschützt ausgeführt.
- **Ein exportiertes Programm (`--code`) läuft außerhalb der Sandbox**, wenn Sie es selbst starten. Vorher kurz ansehen. - **Ein exportiertes Programm (`--code`) läuft außerhalb der Sandbox**, wenn Sie es selbst starten. Vorher kurz ansehen.
- **API-Schlüssel** nur als Umgebungsvariable; `.env`-Dateien sind von Git ausgeschlossen, gehören aber trotzdem nicht ins Projekt. - **API-Schlüssel** nur als Umgebungsvariable; `.env`-Dateien sind von Git ausgeschlossen, gehören aber trotzdem nicht ins Projekt.

View file

@ -115,6 +115,7 @@ Exit-Codes: `0` Ergebnis zuverlässig, `2` Ergebnis vorhanden, aber unzuverläss
| `OPENROUTER_MIN_THROUGHPUT` | bevorzugte Mindestgeschwindigkeit des OpenRouter-Anbieters (Tokens/s); Präferenz, kein harter Filter | 75 | | `OPENROUTER_MIN_THROUGHPUT` | bevorzugte Mindestgeschwindigkeit des OpenRouter-Anbieters (Tokens/s); Präferenz, kein harter Filter | 75 |
| `LLM_TIMEOUT` | Timeout je LLM-Aufruf (s) | 420 | | `LLM_TIMEOUT` | Timeout je LLM-Aufruf (s) | 420 |
| `LLM_REASONING_EFFORT` | Denkaufwand des Modells über OpenRouter: `low`, `medium` oder `high` | Modellvorgabe | | `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` |
| `SOLVER_SANDBOX` | Isolierung des erzeugten Codes: `auto` (bubblewrap, wenn vorhanden), `bwrap` (Pflicht), `aus` | `auto` | | `SOLVER_SANDBOX` | Isolierung des erzeugten Codes: `auto` (bubblewrap, wenn vorhanden), `bwrap` (Pflicht), `aus` | `auto` |
| `LLM_MODEL_UNDERSTANDING`, `LLM_MODEL_EXTRACTION`, `LLM_MODEL_REVIEW`, `LLM_MODEL_REVISION`, `LLM_MODEL_SOLVER` | Modell einzelner Pipeline-Stufen, z. B. ein schnelles für den Solver-Code; verschiedene Anbieter sind erlaubt | Modell aus `--model` | | `LLM_MODEL_UNDERSTANDING`, `LLM_MODEL_EXTRACTION`, `LLM_MODEL_REVIEW`, `LLM_MODEL_REVISION`, `LLM_MODEL_SOLVER` | Modell einzelner Pipeline-Stufen, z. B. ein schnelles für den Solver-Code; verschiedene Anbieter sind erlaubt | Modell aus `--model` |

View file

@ -5,13 +5,11 @@ import sys
from pathlib import Path from pathlib import Path
from typing import List from typing import List
from checks.model_checks import check_figure_readings, check_spec, check_understanding, has_errors from checks.model_checks import check_figure_readings, check_spec, check_understanding
from checks.review import needs_work, review_assumptions, revision_reasons, select_best from checks.review import needs_work, review_assumptions, revision_reasons, select_best
from core.llm import DEFAULT_MODEL, LLMSession from core.llm import DEFAULT_MODEL, LLMSession
from core.model_schema import ( from core.model_schema import (
AssumptionReview,
SolverStatus, SolverStatus,
CheckIssue,
CheckSeverity, CheckSeverity,
OptimizationModelSpecification, OptimizationModelSpecification,
PipelineResult, PipelineResult,
@ -21,7 +19,8 @@ from data import data_profile
from modeling.extraction import revise_model, text_to_model from modeling.extraction import revise_model, text_to_model
from reporting.report import write_report from reporting.report import write_report
from solving.script_export import build_solver_script from solving.script_export import build_solver_script
from solving.solver_stage import check_feasibility, check_multistart, check_sensitivity, run_solver_code, solve from solving.solver_stage import (SOLVER_TIMEOUT, check_feasibility, check_multistart, check_sensitivity,
run_solver_code, solve)
from understanding.figures import FigureError, normalize_figure_ids, split_figures, with_transcriptions from understanding.figures import FigureError, normalize_figure_ids, split_figures, with_transcriptions
from understanding.understand import read_figures, understand_task from understanding.understand import read_figures, understand_task
@ -31,7 +30,7 @@ MAX_RESTARTS = int(os.environ.get("MAX_RESTARTS", "1"))
# "unlösbar" kann ein Modellierungsfehler sein (beobachtet bei qwen3-235b: Aufgabe war lösbar). # "unlösbar" kann ein Modellierungsfehler sein (beobachtet bei qwen3-235b: Aufgabe war lösbar).
# 1 = einmal unabhängig neu modellieren und erneut lösen, bevor "unlösbar" berichtet wird. # 1 = einmal unabhängig neu modellieren und erneut lösen, bevor "unlösbar" berichtet wird.
INFEASIBLE_RECHECK = int(os.environ.get("INFEASIBLE_RECHECK", "1")) INFEASIBLE_RECHECK = int(os.environ.get("INFEASIBLE_RECHECK", "1"))
# Ein Nachfassen, wenn die Verstehensstufe Zahlen oder Teilfragen übersehen hat (0 schaltet die Stufe ab) # Nachfassen, wenn die Verstehensstufe Zahlen oder Teilfragen übersehen hat (0 = kein Nachfassen, -1 = Stufe aus)
UNDERSTAND_RETRIES = int(os.environ.get("UNDERSTAND_RETRIES", "1")) UNDERSTAND_RETRIES = int(os.environ.get("UNDERSTAND_RETRIES", "1"))
task_description = """ task_description = """
@ -107,7 +106,7 @@ def _model_attempt(session, task: str, max_revisions: int, label: str, hint: str
log(f"[{label}] Übernehme gespeicherte Spezifikation ({len(spec.constraints)} Restriktionen) ...") log(f"[{label}] Übernehme gespeicherte Spezifikation ({len(spec.constraints)} Restriktionen) ...")
issues = check_spec(spec, task, understanding) issues = check_spec(spec, task, understanding)
log(f"[{label}] Prüfe Annahmen ...") log(f"[{label}] Prüfe Annahmen ...")
review = review_assumptions(session, task, spec) review = review_assumptions(session, task, spec, data_prompt)
candidates = [(spec, issues, review)] candidates = [(spec, issues, review)]
revisions = 0 revisions = 0
@ -116,9 +115,9 @@ def _model_attempt(session, task: str, max_revisions: int, label: str, hint: str
log(f"[{label}] Probleme gefunden – Korrekturrunde {revisions}/{max_revisions} ...") log(f"[{label}] Probleme gefunden – Korrekturrunde {revisions}/{max_revisions} ...")
for grund in revision_reasons(review, issues): for grund in revision_reasons(review, issues):
log(f" Grund: {grund}") log(f" Grund: {grund}")
spec = revise_model(session, task, spec, review, issues) spec = revise_model(session, task, spec, review, issues, data_prompt)
issues = check_spec(spec, task, understanding) issues = check_spec(spec, task, understanding)
review = review_assumptions(session, task, spec) review = review_assumptions(session, task, spec, data_prompt)
candidates.append((spec, issues, review)) candidates.append((spec, issues, review))
selected = select_best(candidates) selected = select_best(candidates)
@ -147,7 +146,7 @@ INFEASIBLE_HINT = (
def _recheck_infeasible(session, task: str, model: str, result: PipelineResult, def _recheck_infeasible(session, task: str, model: str, result: PipelineResult,
max_revisions: int, solver_retries: int) -> PipelineResult: max_revisions: int, solver_retries: int, data_files=(), data_prompt: str = "") -> PipelineResult:
"""'unlösbar' nicht ungeprüft glauben: unabhängig neu modellieren und erneut lösen.""" """'unlösbar' nicht ungeprüft glauben: unabhängig neu modellieren und erneut lösen."""
status = result.solver_run.status.value status = result.solver_run.status.value
if INFEASIBLE_RECHECK < 1: if INFEASIBLE_RECHECK < 1:
@ -157,7 +156,8 @@ def _recheck_infeasible(session, task: str, model: str, result: PipelineResult,
log(f"[{model}] Solver meldet '{status}' – Gegenprobe mit unabhängiger Modellierung ...") log(f"[{model}] Solver meldet '{status}' – Gegenprobe mit unabhängiger Modellierung ...")
zweiter = _model_attempt(session, task, max_revisions, f"{model} Gegenprobe", zweiter = _model_attempt(session, task, max_revisions, f"{model} Gegenprobe",
hint=INFEASIBLE_HINT.format(status=status)) hint=INFEASIBLE_HINT.format(status=status),
understanding=result.understanding, data_prompt=data_prompt)
if not zweiter["review"].objective_check.matches: if not zweiter["review"].objective_check.matches:
log(" Gegenprobe verworfen: Zielfunktion der neuen Modellierung passt nicht zur Frage.") log(" Gegenprobe verworfen: Zielfunktion der neuen Modellierung passt nicht zur Frage.")
result.reliable = False result.reliable = False
@ -166,13 +166,16 @@ def _recheck_infeasible(session, task: str, model: str, result: PipelineResult,
spec = zweiter["spec"] spec = zweiter["spec"]
log(f"[{model}] Gegenprobe: Solver-Code erzeugen und starten ({spec.target_stack.value}) ...") log(f"[{model}] Gegenprobe: Solver-Code erzeugen und starten ({spec.target_stack.value}) ...")
code, run = solve(session, task, spec, retries=solver_retries, log=log) code, run = solve(session, task, spec, retries=solver_retries, log=log, understanding=result.understanding,
data_files=data_files, data_prompt=data_prompt)
if run.status in (SolverStatus.OPTIMAL, SolverStatus.FEASIBLE) and not run.feasibility: if run.status in (SolverStatus.OPTIMAL, SolverStatus.FEASIBLE) and not run.feasibility:
log(f" Gegenprobe widerlegt '{status}': Zielwert {run.objective_value} – diese Lösung wird verwendet.") log(f" Gegenprobe widerlegt '{status}': Zielwert {run.objective_value} – diese Lösung wird verwendet.")
result.specification, result.review, result.checks = spec, zweiter["review"], check_spec(spec, task) result.specification, result.review = spec, zweiter["review"]
result.checks = check_spec(spec, task, result.understanding)
result.solver_code, result.solver_run = code, run result.solver_code, result.solver_run = code, run
result.restarts += 1 result.restarts += 1
_assess_run(result) # die neue Lösung durchläuft dieselbe Abnahme wie die erste
return result return result
if run.status.value == status: if run.status.value == status:
@ -199,6 +202,15 @@ def _solver_stage(session, task: str, result: PipelineResult, model: str,
result.solver_code, result.solver_run = solve(session, task, spec, retries=solver_retries, log=log, result.solver_code, result.solver_run = solve(session, task, spec, retries=solver_retries, log=log,
understanding=result.understanding, understanding=result.understanding,
data_files=data_files, data_prompt=data_prompt) data_files=data_files, data_prompt=data_prompt)
_assess_run(result)
if result.solver_run.status in (SolverStatus.INFEASIBLE, SolverStatus.UNBOUNDED):
result = _recheck_infeasible(session, task, model, result, max_revisions, solver_retries,
data_files=data_files, data_prompt=data_prompt)
return result
def _assess_run(result: PipelineResult) -> None:
"""Abnahme eines Solver-Laufs: Befunde, die das Ergebnis unzuverlässig machen, in `result` übernehmen."""
run = result.solver_run run = result.solver_run
log(f"Solver: {run.status.value}, Zielwert: {run.objective_value}" + (f"\n {run.error}" if run.error else "")) log(f"Solver: {run.status.value}, Zielwert: {run.objective_value}" + (f"\n {run.error}" if run.error else ""))
if run.feasibility: if run.feasibility:
@ -220,10 +232,6 @@ def _solver_stage(session, task: str, result: PipelineResult, model: str,
if result.solver_code.data_assumptions: if result.solver_code.data_assumptions:
log(f" Mit angenommenen Daten gelöst ({len(result.solver_code.data_assumptions)} Annahmen).") log(f" Mit angenommenen Daten gelöst ({len(result.solver_code.data_assumptions)} Annahmen).")
if run.status in (SolverStatus.INFEASIBLE, SolverStatus.UNBOUNDED):
result = _recheck_infeasible(session, task, model, result, max_revisions, solver_retries)
return result
def _log_usage(session, result: PipelineResult) -> None: def _log_usage(session, result: PipelineResult) -> None:
result.usage = session.usage result.usage = session.usage
@ -268,7 +276,7 @@ def run_pipeline(
if figur.uncertain: if figur.uncertain:
log(f" {figur.id} – unsicher gelesen: " + "; ".join(figur.uncertain)) log(f" {figur.id} – unsicher gelesen: " + "; ".join(figur.uncertain))
task = with_transcriptions(task, understanding.figures) task = with_transcriptions(task, understanding.figures)
if not understanding.is_optimization_task: if understanding is not None and not understanding.is_optimization_task:
# Keine Entscheidung verlangt: jedes Modell wäre erfunden, jede Zahl ohne Frage dahinter # Keine Entscheidung verlangt: jedes Modell wäre erfunden, jede Zahl ohne Frage dahinter
grund = f"Keine Optimierungsaufgabe: {understanding.reason}" grund = f"Keine Optimierungsaufgabe: {understanding.reason}"
log(f"ABGELEHNT – {grund} Es wird weder modelliert noch gerechnet.") log(f"ABGELEHNT – {grund} Es wird weder modelliert noch gerechnet.")
@ -319,12 +327,14 @@ STAGES = ("pruefung", "solver")
def continue_pipeline( def continue_pipeline(
result_json: str, stage: str, model: str = DEFAULT_MODEL, max_revisions: int = MAX_REVISIONS, result_json: str, stage: str, model: str = DEFAULT_MODEL, max_revisions: int = MAX_REVISIONS,
solver_retries: int = SOLVER_RETRIES, session: LLMSession | None = None, solver_retries: int = SOLVER_RETRIES, session: LLMSession | None = None, data_files=(),
) -> PipelineResult: ) -> PipelineResult:
"""Gespeichertes Ergebnis ab einer Stufe fortsetzen – die davor liegenden LLM-Aufrufe entfallen. """Gespeichertes Ergebnis ab einer Stufe fortsetzen – die davor liegenden LLM-Aufrufe entfallen.
'pruefung' behält die Spezifikation und prüft/korrigiert sie neu; 'solver' übernimmt auch den 'pruefung' behält die Spezifikation und prüft/korrigiert sie neu; 'solver' übernimmt auch den
Prüfbericht und erzeugt nur den Solver-Code neu. Aufgabentext kommt aus dem gespeicherten Ergebnis. Prüfbericht und erzeugt nur den Solver-Code neu. Aufgabentext, Verständnis und Abbildungen kommen aus
dem gespeicherten Ergebnis; beigelegte Daten (`data_files`) sind dort nur als Profil gespeichert und
müssen erneut angegeben werden, sonst findet das Programm seine Dateien nicht.
""" """
if stage not in STAGES: if stage not in STAGES:
raise ValueError(f"Unbekannte Stufe '{stage}' (erlaubt: {', '.join(STAGES)})") raise ValueError(f"Unbekannte Stufe '{stage}' (erlaubt: {', '.join(STAGES)})")
@ -332,39 +342,45 @@ def continue_pipeline(
if stored.rejected or stored.specification is None: if stored.rejected or stored.specification is None:
raise ValueError(f"{result_json} ist eine abgelehnte Aufgabe (keine Optimierungsaufgabe) – nichts fortzusetzen") raise ValueError(f"{result_json} ist eine abgelehnte Aufgabe (keine Optimierungsaufgabe) – nichts fortzusetzen")
session = session or LLMSession(model) session = session or LLMSession(model)
task = stored.task task, understanding = stored.task, stored.understanding
profile = data_profile.profile_files(data_files) if data_files else stored.data_files
data_prompt = data_profile.to_prompt(profile)
if stored.data_files and not data_files:
log("WARNUNG – das gespeicherte Ergebnis hatte beigelegte Daten; ohne --daten findet das Programm sie nicht.")
log(f"Setze {result_json} ab Stufe '{stage}' fort – Extraktion entfällt.") log(f"Setze {result_json} ab Stufe '{stage}' fort – Extraktion entfällt.")
if stage == "solver": if stage == "solver":
result = stored.model_copy(update={ checks = check_spec(stored.specification, task, understanding) # deterministisch, kostet nichts
"checks": check_spec(stored.specification, task), # deterministisch, kostet nichts problems = _unresolved({"review": stored.review, "issues": checks})
"solver_code": None, "solver_run": None, "usage": None, update = {"checks": checks}
"reliable": True, "reliability_issues": [],
})
problems = _unresolved({"review": stored.review, "issues": result.checks})
result.reliable, result.reliability_issues = not problems, problems
else: else:
attempt = _model_attempt(session, task, max_revisions, model, spec=stored.specification) attempt = _model_attempt(session, task, max_revisions, model, spec=stored.specification,
understanding=understanding, data_prompt=data_prompt)
problems = _unresolved(attempt) problems = _unresolved(attempt)
result = PipelineResult( update = {"specification": attempt["spec"], "review": attempt["review"], "revisions": attempt["revisions"],
task=task, specification=attempt["spec"], review=attempt["review"], revisions=attempt["revisions"], "selected_revision": attempt["selected"], "restarts": 0, "checks": attempt["issues"]}
selected_revision=attempt["selected"], checks=attempt["issues"], # Verständnis, Abbildungen und Datenprofil bleiben erhalten; alles ab der Fortsetzungsstufe entsteht neu
reliable=not problems, reliability_issues=problems, result = stored.model_copy(update={
) **update, "data_files": profile, "solver_code": None, "solver_run": None, "usage": None,
"infeasible_confirmed": False, "reliable": not problems, "reliability_issues": problems,
})
for issue in result.checks: for issue in result.checks:
log(f" {issue.severity.value.upper()} [{issue.code}] {issue.message}") log(f" {issue.severity.value.upper()} [{issue.code}] {issue.message}")
if result.reliability_issues: if result.reliability_issues:
log("ERGEBNIS UNZUVERLÄSSIG: " + "; ".join(result.reliability_issues)) log("ERGEBNIS UNZUVERLÄSSIG: " + "; ".join(result.reliability_issues))
result = _solver_stage(session, task, result, model, max_revisions, solver_retries) result = _solver_stage(session, task, result, model, max_revisions, solver_retries,
data_files=data_files, data_prompt=data_prompt)
_log_usage(session, result) _log_usage(session, result)
return result return result
def resolve_result(result_json: str, data_file: str | None = None, solver_timeout: int = 60) -> PipelineResult: def resolve_result(result_json: str, data_file: str | None = None, solver_timeout: int = SOLVER_TIMEOUT,
data_files=()) -> PipelineResult:
"""Gespeichertes Ergebnis mit (geänderten) Daten neu rechnen – ohne einen einzigen LLM-Aufruf. """Gespeichertes Ergebnis mit (geänderten) Daten neu rechnen – ohne einen einzigen LLM-Aufruf.
Das erzeugte Programm hält alle Eingabezahlen in DATA; nur diese werden ersetzt. Das erzeugte Programm hält alle Eingabezahlen in DATA; nur diese werden ersetzt. Liest es beigelegte
Dateien, müssen sie über `data_files` erneut angegeben werden.
""" """
result = PipelineResult.model_validate_json(Path(result_json).read_text(encoding="utf-8")) result = PipelineResult.model_validate_json(Path(result_json).read_text(encoding="utf-8"))
if not result.solver_code: if not result.solver_code:
@ -372,14 +388,16 @@ def resolve_result(result_json: str, data_file: str | None = None, solver_timeou
data = json.loads(Path(data_file).read_text(encoding="utf-8")) if data_file else None data = json.loads(Path(data_file).read_text(encoding="utf-8")) if data_file else None
log(f"Rechne {result_json} neu" + (f" mit Daten aus {data_file}" if data_file else "") + " – ohne LLM ...") log(f"Rechne {result_json} neu" + (f" mit Daten aus {data_file}" if data_file else "") + " – ohne LLM ...")
run = run_solver_code(result.solver_code.code, timeout=solver_timeout, data=data) run = run_solver_code(result.solver_code.code, timeout=solver_timeout, data=data, data_files=data_files)
run.feasibility = check_feasibility(run, result.specification) run.feasibility = check_feasibility(run, result.specification)
run.warnings += check_sensitivity(run, result.specification.objective.sense.value) run.warnings += check_sensitivity(run, result.specification.objective.sense.value)
run.warnings += check_multistart(run, result.specification.objective.sense.value) run.warnings += check_multistart(run, result.specification.objective.sense.value)
result.solver_run = run result.solver_run = run
result.reliable = not run.feasibility # Frühere Befunde (Zielgröße, Abbildung, Datenherkunft) gelten weiter – nur die Zulässigkeit ist neu geprüft
result.reliability_issues = [f"Zulässigkeitsprüfung: {p}" for p in run.feasibility] result.reliability_issues = ([p for p in result.reliability_issues if not p.startswith("Zulässigkeitsprüfung: ")]
+ [f"Zulässigkeitsprüfung: {p}" for p in run.feasibility])
result.reliable = not result.reliability_issues
result.usage = None # keine LLM-Nutzung in diesem Lauf result.usage = None # keine LLM-Nutzung in diesem Lauf
log(f"Solver: {run.status.value}, Zielwert: {run.objective_value}" + (f"\n {run.error}" if run.error else "")) log(f"Solver: {run.status.value}, Zielwert: {run.objective_value}" + (f"\n {run.error}" if run.error else ""))
for problem in run.feasibility: for problem in run.feasibility:
@ -416,7 +434,8 @@ if __name__ == "__main__":
"'solver' (auch den Prüfbericht behalten, nur Solver-Code neu erzeugen)") "'solver' (auch den Prüfbericht behalten, nur Solver-Code neu erzeugen)")
parser.add_argument("--daten", nargs="+", metavar="DATEI", parser.add_argument("--daten", nargs="+", metavar="DATEI",
help="beigelegte Datendateien (CSV/Excel); ihr Profil geht in die Prompts, " help="beigelegte Datendateien (CSV/Excel); ihr Profil geht in die Prompts, "
"die Dateien selbst ins Arbeitsverzeichnis des Programms") "die Dateien selbst ins Arbeitsverzeichnis des Programms "
"(mit --resolve/--von erneut angeben, gespeichert ist nur ihr Profil)")
parser.add_argument("--data", metavar="DATEN.JSON", parser.add_argument("--data", metavar="DATEN.JSON",
help="mit --resolve: Eingabewerte (DATA) des Programms ganz oder teilweise ersetzen") help="mit --resolve: Eingabewerte (DATA) des Programms ganz oder teilweise ersetzen")
parser.add_argument("--report", metavar="DATEI", help="Report mit gerenderten Formeln und Solver-Code (.md oder .html)") parser.add_argument("--report", metavar="DATEI", help="Report mit gerenderten Formeln und Solver-Code (.md oder .html)")
@ -439,9 +458,10 @@ if __name__ == "__main__":
parser.error("--von endet immer mit der Solver-Stufe; --no-solve wäre wirkungslos") parser.error("--von endet immer mit der Solver-Stufe; --no-solve wäre wirkungslos")
if args.resolve: if args.resolve:
result = resolve_result(args.resolve, args.data) result = resolve_result(args.resolve, args.data, data_files=args.daten or ())
elif args.von: elif args.von:
result = continue_pipeline(args.von[1], args.von[0], model=args.model, max_revisions=args.max_revisions) result = continue_pipeline(args.von[1], args.von[0], model=args.model, max_revisions=args.max_revisions,
data_files=args.daten or ())
else: else:
try: try:
result = run_pipeline( result = run_pipeline(

View file

@ -36,7 +36,7 @@ PROMPT_FILES = ("understanding/Verstehen-Prompt.txt", "understanding/Abbildungs-
# Umgebungsvariablen, die das Verhalten ändern – ohne sie ließe sich ein A/B-Lauf nicht zuordnen # Umgebungsvariablen, die das Verhalten ändern – ohne sie ließe sich ein A/B-Lauf nicht zuordnen
CONFIG_VARS = ("LLM_MODEL", "LLM_MODEL_EXTRACTION", "LLM_MODEL_REVIEW", "LLM_MODEL_REVISION", "LLM_MODEL_SOLVER", CONFIG_VARS = ("LLM_MODEL", "LLM_MODEL_EXTRACTION", "LLM_MODEL_REVIEW", "LLM_MODEL_REVISION", "LLM_MODEL_SOLVER",
"UNDERSTAND_RETRIES", "MAX_REVISIONS", "SOLVER_RETRIES", "MAX_RESTARTS", "INFEASIBLE_RECHECK", "UNDERSTAND_RETRIES", "MAX_REVISIONS", "SOLVER_RETRIES", "MAX_RESTARTS", "INFEASIBLE_RECHECK",
"LLM_REASONING_EFFORT", "SOLVER_SANDBOX") "LLM_REASONING_EFFORT", "SOLVER_SANDBOX", "SOLVER_TIMEOUT")
def prompt_digest(paths=None) -> str: def prompt_digest(paths=None) -> str:
@ -63,7 +63,8 @@ def prompt_state() -> dict:
text=True, timeout=10).stdout.strip() text=True, timeout=10).stdout.strip()
try: try:
state["commit"] = git("rev-parse", "--short", "HEAD") or "unbekannt" state["commit"] = git("rev-parse", "--short", "HEAD") or "unbekannt"
state["dirty"] = bool(git("status", "--porcelain")) # nur versionierte Dateien: eine abgelegte Aufgabe in tasks/ ändert nichts am gemessenen Stand
state["dirty"] = bool(git("status", "--porcelain", "--untracked-files=no"))
except (OSError, subprocess.SubprocessError): except (OSError, subprocess.SubprocessError):
state["commit"], state["dirty"] = "unbekannt", None state["commit"], state["dirty"] = "unbekannt", None
return state return state

View file

@ -9,14 +9,18 @@ from core.model_schema import AssumptionReview, AssumptionVerdict, CheckIssue, C
REVIEW_PROMPT_FILE = Path(__file__).parent / "Pruef-Prompt.txt" REVIEW_PROMPT_FILE = Path(__file__).parent / "Pruef-Prompt.txt"
def review_assumptions(session: LLMSession, task: str, spec: OptimizationModelSpecification) -> AssumptionReview: def review_assumptions(session: LLMSession, task: str, spec: OptimizationModelSpecification,
data_prompt: str = "") -> AssumptionReview:
# Mit beigelegten Daten gehört ihr Profil zum Aufgabentext – sonst beurteilt die Prüfung Annahmen
# über Spalten, die sie nie gesehen hat.
aufgabe = f"{task}\n\n{data_prompt}" if data_prompt else task
return session.create( return session.create(
AssumptionReview, AssumptionReview,
[ [
{"role": "system", "content": REVIEW_PROMPT_FILE.read_text(encoding="utf-8")}, {"role": "system", "content": REVIEW_PROMPT_FILE.read_text(encoding="utf-8")},
{ {
"role": "user", "role": "user",
"content": f"AUFGABENTEXT:\n{task}\n\nMODELLSPEZIFIKATION:\n{spec.model_dump_json(indent=2)}", "content": f"AUFGABENTEXT:\n{aufgabe}\n\nMODELLSPEZIFIKATION:\n{spec.model_dump_json(indent=2)}",
}, },
], ],
step="Prüfung", step="Prüfung",

View file

@ -35,7 +35,7 @@ def text_to_model(session: LLMSession, task: str,
def revise_model( def revise_model(
session: LLMSession, task: str, spec: OptimizationModelSpecification, review: AssumptionReview, session: LLMSession, task: str, spec: OptimizationModelSpecification, review: AssumptionReview,
issues: List[CheckIssue] = (), issues: List[CheckIssue] = (), data_prompt: str = "",
) -> OptimizationModelSpecification: ) -> OptimizationModelSpecification:
errors = [i for i in issues if i.severity == CheckSeverity.ERROR] errors = [i for i in issues if i.severity == CheckSeverity.ERROR]
check_text = "\n".join(f"- [{i.code}] {i.message}" for i in errors) or "keine" check_text = "\n".join(f"- [{i.code}] {i.message}" for i in errors) or "keine"
@ -43,6 +43,7 @@ def revise_model(
# sollen sie mit behoben werden – sonst erreicht z. B. unerklärter Fachjargon das LLM nie. # sollen sie mit behoben werden – sonst erreicht z. B. unerklärter Fachjargon das LLM nie.
hinweise = [i for i in issues if i.severity == CheckSeverity.WARNING] hinweise = [i for i in issues if i.severity == CheckSeverity.WARNING]
hinweis_text = "\n".join(f"- [{i.code}] {i.message}" for i in hinweise) or "keine" hinweis_text = "\n".join(f"- [{i.code}] {i.message}" for i in hinweise) or "keine"
aufgabe = f"{task}\n\n{data_prompt}" if data_prompt else task # Datenprofil wie bei der Extraktion
return session.create( return session.create(
OptimizationModelSpecification, OptimizationModelSpecification,
[ [
@ -50,7 +51,7 @@ def revise_model(
{ {
"role": "user", "role": "user",
"content": ( "content": (
f"AUFGABENTEXT:\n{task}\n\n" f"AUFGABENTEXT:\n{aufgabe}\n\n"
f"BISHERIGE MODELLSPEZIFIKATION:\n{spec.model_dump_json(indent=2)}\n\n" f"BISHERIGE MODELLSPEZIFIKATION:\n{spec.model_dump_json(indent=2)}\n\n"
f"PRÜFBERICHT:\n{review.model_dump_json(indent=2)}\n\n" f"PRÜFBERICHT:\n{review.model_dump_json(indent=2)}\n\n"
f"FEHLER DER AUTOMATISCHEN KONSISTENZPRÜFUNG:\n{check_text}\n\n" f"FEHLER DER AUTOMATISCHEN KONSISTENZPRÜFUNG:\n{check_text}\n\n"

View file

@ -256,12 +256,12 @@ def _chart_for_table(table):
return "" return ""
col = table.columns.index(table.objective_column) col = table.columns.index(table.objective_column)
labels, values, total = [], [], None labels, values, total = [], [], None
for row in table.rows: for n, row in enumerate(table.rows):
value = row[col] if col < len(row) else None value = row[col] if col < len(row) else None
if not isinstance(value, (int, float)) or isinstance(value, bool): if not isinstance(value, (int, float)) or isinstance(value, bool):
continue continue
label = str(row[0]) if row else "" label = str(row[0]) if row else ""
if _is_sum_row(row): if _is_sum_row(row, last=n == len(table.rows) - 1):
total = (label, value) total = (label, value)
else: else:
labels.append(label) labels.append(label)

View file

@ -78,7 +78,7 @@ Ergebnisausgabe:
- sensitivity: Wirkung der bindenden Restriktionen, durch NEULÖSEN ermittelt (nicht aus Solver-Attributen ablesen). Für höchstens 8 bindende Restriktionen (linke Seite erreicht die Grenze genau), am besten die inhaltlich wichtigsten: - sensitivity: Wirkung der bindenden Restriktionen, durch NEULÖSEN ermittelt (nicht aus Solver-Attributen ablesen). Für höchstens 8 bindende Restriktionen (linke Seite erreicht die Grenze genau), am besten die inhaltlich wichtigsten:
{"name": "<wie in constraints>", "delta": <Verschiebung > 0>, "objective_plus": <Zielwert mit rhs+delta oder null>, "objective_minus": <Zielwert mit rhs-delta oder null>, "unit": "€ je Einheit"} {"name": "<wie in constraints>", "delta": <Verschiebung > 0>, "objective_plus": <Zielwert mit rhs+delta oder null>, "objective_minus": <Zielwert mit rhs-delta oder null>, "unit": "€ je Einheit"}
- delta klein wählen (etwa 1 % der rechten Seite, mindestens eine sinnvolle Einheit), das Modell je Richtung einmal neu lösen und den Zielwert eintragen; bei unlösbar null. - delta klein wählen (etwa 1 % der rechten Seite, mindestens eine sinnvolle Einheit), das Modell je Richtung einmal neu lösen und den Zielwert eintragen; bei unlösbar null.
- Nur ausführen, wenn das Neulösen schnell geht (Gesamtlaufzeit unter 60 s), sonst leere Liste. - Nur ausführen, wenn das Neulösen schnell geht (Gesamtlaufzeit unter {ZEITLIMIT} s), sonst leere Liste.
- relaxation_bound: bei ganzzahligen/binären Variablen der Zielwert der stetigen Relaxation (alle Ganzzahligkeitsforderungen aufgehoben), sonst null. Zeigt, was die Ganzzahligkeit kostet. - relaxation_bound: bei ganzzahligen/binären Variablen der Zielwert der stetigen Relaxation (alle Ganzzahligkeitsforderungen aufgehoben), sonst null. Zeigt, was die Ganzzahligkeit kostet.
- answers: Die Rückprojektion auf die gestellte Aufgabe – genau ein Eintrag je Teilfrage aus der Liste im Auftrag, mit deren question_id. Das gilt auch dann, wenn es nur EINE Teilfrage gibt: answers bleibt nie leer. - answers: Die Rückprojektion auf die gestellte Aufgabe – genau ein Eintrag je Teilfrage aus der Liste im Auftrag, mit deren question_id. Das gilt auch dann, wenn es nur EINE Teilfrage gibt: answers bleibt nie leer.
[{"question_id": "a", "text": "…"}, {"question_id": "b", "text": "…"}] [{"question_id": "a", "text": "…"}, {"question_id": "b", "text": "…"}]
@ -94,7 +94,7 @@ Ergebnisausgabe:
- data_override enthält genau die Einträge von DATA, die sich ändern. Das Programm rechnet jedes Szenario selbst und trägt seinen objective_value ein. - data_override enthält genau die Einträge von DATA, die sich ändern. Das Programm rechnet jedes Szenario selbst und trägt seinen objective_value ein.
- Die Pipeline führt dein Programm anschließend mit jedem data_override ERNEUT aus und vergleicht – abweichende Werte fallen auf. Kein LLM ist daran beteiligt. [→ verify_scenarios] - Die Pipeline führt dein Programm anschließend mit jedem data_override ERNEUT aus und vergleicht – abweichende Werte fallen auf. Kein LLM ist daran beteiligt. [→ verify_scenarios]
- Damit das funktioniert, muss das Programm ausschließlich aus DATA lesen; höchstens fünf Szenarien werden nachgerechnet. - Damit das funktioniert, muss das Programm ausschließlich aus DATA lesen; höchstens fünf Szenarien werden nachgerechnet.
- Jedes Szenario braucht einen objective_value – null nur, wenn es bewiesen unlösbar ist. Unterscheiden sich Szenarien nur in Zielgewichten, ist die Lösung eines Szenarios für die anderen zulässig: übergib sie als Startlösung, dann endet keines im Zeitlimit ohne Ergebnis. Zeitlimits so wählen, dass alle Läufe zusammen unter 60 s bleiben. [→ check_scenarios_solved] - Jedes Szenario braucht einen objective_value – null nur, wenn es bewiesen unlösbar ist. Unterscheiden sich Szenarien nur in Zielgewichten, ist die Lösung eines Szenarios für die anderen zulässig: übergib sie als Startlösung, dann endet keines im Zeitlimit ohne Ergebnis. Zeitlimits so wählen, dass alle Läufe zusammen unter {ZEITLIMIT} s bleiben. [→ check_scenarios_solved]
- graph: NUR bei Graphenproblemen (Stack NetworkX), sonst null. Grundlage für die Zeichnung im Report: - graph: NUR bei Graphenproblemen (Stack NetworkX), sonst null. Grundlage für die Zeichnung im Report:
{"nodes": ["0", "1", "2", "3"], {"nodes": ["0", "1", "2", "3"],
"arcs": [{"source": "0", "target": "1", "weight": 8}, ...], "arcs": [{"source": "0", "target": "1", "weight": 8}, ...],
@ -120,7 +120,7 @@ Ergebnisausgabe:
- Entscheidungswerte als formatierte Texte, Beiträge als Zahlen; Dezimalkomma im Text ist erlaubt. - Entscheidungswerte als formatierte Texte, Beiträge als Zahlen; Dezimalkomma im Text ist erlaubt.
- unit: Einheit der objective_column für die Diagrammbeschriftung (z. B. " €", " %"), sonst weglassen. - unit: Einheit der objective_column für die Diagrammbeschriftung (z. B. " €", " %"), sonst weglassen.
- note: ein Satz, wie die Tabelle zu lesen ist. - note: ein Satz, wie die Tabelle zu lesen ist.
Erzeuge nach Möglichkeit diese drei Tabellen (in dieser Reihenfolge): Erzeuge nach Möglichkeit diese drei Tabellen (in dieser Reihenfolge). Bei LP, MILP und CP ist die Zielwert-Zerlegung Pflicht (die Zielfunktion ist dort immer eine Summe), und jede objective_column muss sich zum Zielwert summieren – sonst wird das Programm zur Reparatur zurückgegeben. [→ check_objective_decomposition]
1. Die Objekttabelle wie oben beschrieben. 1. Die Objekttabelle wie oben beschrieben.
2. "Zielwert-Zerlegung": Zeilen für die Bestandteile der Zielfunktion (z. B. "Erwartete Rendite", "Risiko (λ·Varianz)", "Transaktionskosten") mit Spalten ["Bestandteil", "Wert", "Einheit"], Vorzeichen wie in der Zielfunktion (Abzüge negativ), objective_column "Wert", danach die Summenzeile "Zielwert". Alle Bestandteile müssen dieselbe Einheit haben; steht dort Unterschiedliches (z. B. "€" neben "dimensionslos"), ist das Modell fehlerhaft. 2. "Zielwert-Zerlegung": Zeilen für die Bestandteile der Zielfunktion (z. B. "Erwartete Rendite", "Risiko (λ·Varianz)", "Transaktionskosten") mit Spalten ["Bestandteil", "Wert", "Einheit"], Vorzeichen wie in der Zielfunktion (Abzüge negativ), objective_column "Wert", danach die Summenzeile "Zielwert". Alle Bestandteile müssen dieselbe Einheit haben; steht dort Unterschiedliches (z. B. "€" neben "dimensionslos"), ist das Modell fehlerhaft.
3. "Bindende Restriktionen": welche Grenzen die Lösung ausreizt, mit Spalten ["Restriktion", "Wert", "Grenze", "bindend"] und Werten aus der Lösung (z. B. "Höchstgewicht A02" | "30,00 %" | "30,00 %" | "ja"). Nur Tatsachen aus der Lösung, keine Bewertung; ohne objective_column. Die note erklärt „bindend" in Alltagssprache („die Lösung schöpft diese Grenze voll aus") – ohne Fachbegriffe. [→ check_text_quality] 3. "Bindende Restriktionen": welche Grenzen die Lösung ausreizt, mit Spalten ["Restriktion", "Wert", "Grenze", "bindend"] und Werten aus der Lösung (z. B. "Höchstgewicht A02" | "30,00 %" | "30,00 %" | "ja"). Nur Tatsachen aus der Lösung, keine Bewertung; ohne objective_column. Die note erklärt „bindend" in Alltagssprache („die Lösung schöpft diese Grenze voll aus") – ohne Fachbegriffe. [→ check_text_quality]

View file

@ -19,6 +19,7 @@ from core.llm import LLMSession
from checks.model_checks import numbers_in, unexplained_jargon from checks.model_checks import numbers_in, unexplained_jargon
from core.model_schema import ( from core.model_schema import (
Answer, Answer,
ProblemClass,
TargetStack, TargetStack,
Scenario, Scenario,
ConstraintValue, ConstraintValue,
@ -40,6 +41,9 @@ RESULT_MARKER = "RESULT_JSON:"
# Isolierung des erzeugten Codes: "auto" nutzt bubblewrap, wenn vorhanden; "bwrap" verlangt es # Isolierung des erzeugten Codes: "auto" nutzt bubblewrap, wenn vorhanden; "bwrap" verlangt es
# (Fehler, wenn es fehlt); "aus" führt ungeschützt aus (nur für Systeme ohne bubblewrap). # (Fehler, wenn es fehlt); "aus" führt ungeschützt aus (nur für Systeme ohne bubblewrap).
SANDBOX = os.environ.get("SOLVER_SANDBOX", "auto").strip().lower() SANDBOX = os.environ.get("SOLVER_SANDBOX", "auto").strip().lower()
# Zeitlimit je Programmlauf in Sekunden. Geht auch als Zeitbudget in den Solver-Prompt ({ZEITLIMIT}): ein höheres
# Limit nützt nur, wenn das Programm seine internen Solver-Zeitlimits danach wählt (SCIP-Szenarien, Portfolio Nr. 2).
SOLVER_TIMEOUT = int(os.environ.get("SOLVER_TIMEOUT", "60"))
ALLOWED_MODULES = {"json", "math", "itertools", "functools", "collections", "numpy", "scipy", "highspy", "ortools", "cvxpy", "pyscipopt", "cyipopt", "networkx", "pandas"} ALLOWED_MODULES = {"json", "math", "itertools", "functools", "collections", "numpy", "scipy", "highspy", "ortools", "cvxpy", "pyscipopt", "cyipopt", "networkx", "pandas"}
FORBIDDEN_CALLS = {"__import__", "eval", "exec", "compile", "open", "input", "breakpoint"} FORBIDDEN_CALLS = {"__import__", "eval", "exec", "compile", "open", "input", "breakpoint"}
@ -260,13 +264,21 @@ def _tail(text: str, limit: int = 2000) -> str:
SUM_ROW_LABELS = ("summe", "gesamt", "total", "σ", "sum", "zielwert", "ergebnis", "nutzen", "insgesamt") SUM_ROW_LABELS = ("summe", "gesamt", "total", "σ", "sum", "zielwert", "ergebnis", "nutzen", "insgesamt")
# Vor der letzten Zeile zählt nur eine eindeutige Beschriftung: „Nutzen der eingepackten Gegenstände“ und
# „Gesamtnutzen …“ sind dort Bestandteile. Beobachtet: 15 Rucksack-Läufe meldeten „Summe (0) weicht vom
# Zielwert (51) ab“, weil der einzige Bestandteil als Summenzeile galt und aus der Summe fiel.
_INNER_SUM_ROW = re.compile(r"(zwischen)?summe\b.*|(gesamt|total|insgesamt|zielwert|σ|sum)\s*:?\s*", re.IGNORECASE)
def _is_sum_row(row, value=None, others=None) -> bool: def _is_sum_row(row, value=None, others=None, last: bool = True) -> bool:
"""Summenzeile: am Text erkennbar oder (letzte Zeile) am Wert = Summe der übrigen Zeilen.""" """Summenzeile: am Text erkennbar oder (letzte Zeile) am Wert = Summe der übrigen Zeilen.
if row and isinstance(row[0], str) and row[0].strip().lower().startswith(SUM_ROW_LABELS):
Wortanfänge wie „Gesamt…“ oder „Nutzen…“ zählen nur in der letzten Zeile, davor nur eindeutige Beschriftungen.
"""
label = row[0].strip().lower() if row and isinstance(row[0], str) else ""
if label and (label.startswith(SUM_ROW_LABELS) if last else _INNER_SUM_ROW.fullmatch(label)):
return True return True
if value is None or not others: if not last or value is None or not others:
return False return False
return _close(value, sum(others)) return _close(value, sum(others))
@ -286,6 +298,60 @@ def _close(a: float, b: float) -> bool:
return abs(a - b) <= 1e-3 * max(abs(a), abs(b)) + 1e-6 return abs(a - b) <= 1e-3 * max(abs(a), abs(b)) + 1e-6
def _is_number(value) -> bool:
return isinstance(value, (int, float)) and not isinstance(value, bool)
def check_table(table: ResultTable, objective_value) -> tuple:
"""Eine Tabelle gegen den Zielwert prüfen: (belegt, abweichung, Warnungen).
`belegt` heißt: die Beitragsspalte summiert sich zum gemeldeten Zielwert – der Zielwert ist damit aus
Bestandteilen nachgerechnet und nicht bloß behauptet. `abweichung` ist die Meldung, wenn sie es nicht tut.
"""
if not table.objective_column:
return False, None, []
if table.objective_column not in table.columns:
return False, None, [f"Tabelle '{table.title}': Spalte '{table.objective_column}' existiert nicht."]
col = table.columns.index(table.objective_column)
cells = [(row, row[col] if col < len(row) else None) for row in table.rows]
numeric = [v for _, v in cells if _is_number(v)]
letzte = len(cells) - 1
def summenzeile(n, row, v):
# letzte Zeile kann auch ohne Schlüsselwort die Summe sein (z. B. Beschriftung "Zielwert");
# eine einzige Zeile ist nie eine Summe („Gesamtnutzen …“ als einziger Bestandteil)
if letzte == 0:
return False
return _is_sum_row(row, v if n == letzte else None, numeric[:-1] if n == letzte else None, last=n == letzte)
values = [v for n, (row, v) in enumerate(cells) if not summenzeile(n, row, v)]
if not all(_is_number(v) for v in values):
return False, None, [f"Tabelle '{table.title}': Spalte '{table.objective_column}' enthält nicht nur "
f"Zahlen – Summe nicht prüfbar."]
total = sum(values)
if _is_comparison(values, len(values) == len(cells), total, objective_value):
return False, None, []
warnings, abweichung = [], None
belegt = objective_value is not None and bool(values) and _close(total, objective_value)
if objective_value is not None and not _close(total, objective_value):
abweichung = (f"Tabelle '{table.title}': Summe der Spalte '{table.objective_column}' ({total:.6g}) "
f"weicht vom Zielwert ({objective_value:.6g}) ab – Tabelle bitte nicht verwenden.")
warnings.append(abweichung)
if len(values) > 1 and max(abs(v) for v in values) > 0:
largest = max(abs(v) for v in values)
tiny = [row[0] for row, v in cells if _is_number(v) and v in values and 0 < abs(v) < largest * 1e-4]
if tiny:
warnings.append(
f"Tabelle '{table.title}': {', '.join(str(t) for t in tiny)} ist um Größenordnungen kleiner "
f"als die übrigen Bestandteile – vermutlich uneinheitliche Einheiten (z. B. relativ statt absolut)."
)
# nur die Schlusszeile: eine Zwischensumme weicht von der Gesamtsumme naturgemäß ab
if cells and _is_number(cells[-1][1]) and summenzeile(letzte, *cells[-1]) and not _close(cells[-1][1], total):
warnings.append(f"Tabelle '{table.title}': Summenzeile ({cells[-1][1]:.6g}) passt nicht zu den Zeilen "
f"({total:.6g}).")
return belegt, abweichung, warnings
def parse_result_tables(raw_tables: list, objective_value) -> tuple: def parse_result_tables(raw_tables: list, objective_value) -> tuple:
"""Ergebnistabellen validieren und deterministisch gegen den Zielwert prüfen. """Ergebnistabellen validieren und deterministisch gegen den Zielwert prüfen.
@ -299,47 +365,37 @@ def parse_result_tables(raw_tables: list, objective_value) -> tuple:
warnings.append(f"Ergebnistabelle ungültig und verworfen: {str(exc)[:200]}") warnings.append(f"Ergebnistabelle ungültig und verworfen: {str(exc)[:200]}")
continue continue
tables.append(table) tables.append(table)
if not table.objective_column: warnings += check_table(table, objective_value)[2]
continue
if table.objective_column not in table.columns:
warnings.append(f"Tabelle '{table.title}': Spalte '{table.objective_column}' existiert nicht.")
continue
col = table.columns.index(table.objective_column)
cells = [(row, row[col] if col < len(row) else None) for row in table.rows]
numeric = [v for _, v in cells if isinstance(v, (int, float)) and not isinstance(v, bool)]
# letzte Zeile kann auch ohne Schlüsselwort die Summe sein (z. B. Beschriftung "Zielwert")
values = [v for n, (row, v) in enumerate(cells)
if not _is_sum_row(row, v if n == len(cells) - 1 else None, numeric[:-1] if n == len(cells) - 1 else None)]
if not all(isinstance(v, (int, float)) and not isinstance(v, bool) for v in values):
warnings.append(f"Tabelle '{table.title}': Spalte '{table.objective_column}' enthält nicht nur Zahlen – Summe nicht prüfbar.")
continue
total = sum(values)
if _is_comparison(values, len(values) == len(cells), total, objective_value):
continue
if objective_value is not None and not _close(total, objective_value):
warnings.append(
f"Tabelle '{table.title}': Summe der Spalte '{table.objective_column}' ({total:.6g}) "
f"weicht vom Zielwert ({objective_value:.6g}) ab – Tabelle bitte nicht verwenden."
)
if len(values) > 1 and max(abs(v) for v in values) > 0:
largest = max(abs(v) for v in values)
tiny = [row[0] for row, v in cells
if isinstance(v, (int, float)) and not isinstance(v, bool) and v in values
and 0 < abs(v) < largest * 1e-4]
if tiny:
warnings.append(
f"Tabelle '{table.title}': {', '.join(str(t) for t in tiny)} ist um Größenordnungen kleiner "
f"als die übrigen Bestandteile – vermutlich uneinheitliche Einheiten (z. B. relativ statt absolut)."
)
for n, (row, value) in enumerate(cells):
if not isinstance(value, (int, float)) or isinstance(value, bool):
continue
if _is_sum_row(row, value if n == len(cells) - 1 else None, numeric[:-1] if n == len(cells) - 1 else None) \
and not _close(value, total):
warnings.append(f"Tabelle '{table.title}': Summenzeile ({value:.6g}) passt nicht zu den Zeilen ({total:.6g}).")
return tables, warnings return tables, warnings
# Lineare Zielfunktionen sind immer eine Summe von Bestandteilen – dort ist die Zerlegung Pflicht. Bei
# QP/SOCP/NLP nicht (Volatilität = Wurzel ist nicht additiv); eine vorhandene Beitragsspalte muss aber stimmen.
DECOMPOSITION_REQUIRED = {ProblemClass.LP, ProblemClass.MILP, ProblemClass.CP}
DECOMPOSITION_MISSING = "Zielwert nicht aus Bestandteilen belegt"
def check_objective_decomposition(run, spec: OptimizationModelSpecification) -> list:
"""Ist der gemeldete Zielwert aus Bestandteilen nachgerechnet – oder nur behauptet?
`check_feasibility` prüft die Restriktionen, nicht den Zielwert. Den belegt erst eine Beitragsspalte,
deren Summe ihn ergibt. Gemessen an 90 Läufen linearer Klassen seit dem 17.09.2026: jede hatte eine –
die Pflicht kostet also kaum Reparaturen, fängt aber Rückfälle. Bleibt der Befund nach den Reparaturen,
ist er eine Warnung: die Lösung selbst ist geprüft zulässig.
"""
if run.status not in (SolverStatus.OPTIMAL, SolverStatus.FEASIBLE) or run.objective_value is None:
return []
geprueft = [check_table(t, run.objective_value) for t in run.tables]
befunde = [abweichung for _, abweichung, _ in geprueft if abweichung]
if spec.problem_class in DECOMPOSITION_REQUIRED and not any(belegt for belegt, _, _ in geprueft):
befunde.append(
f"{DECOMPOSITION_MISSING}: keine Ergebnistabelle hat eine objective_column, deren Summe den Zielwert "
f"{run.objective_value:.6g} ergibt. Die Zielfunktion ist linear, also eine Summe – melde die Tabelle "
f"„Zielwert-Zerlegung“ mit einer Zeile je Bestandteil, objective_column \"Wert\" und der Summenzeile "
f"„Zielwert“.")
return befunde
def _parse_list(raw, model_type) -> list: def _parse_list(raw, model_type) -> list:
"""Fehlerhafte Einträge verwerfen statt den Lauf scheitern zu lassen (die Lösung ist davon unberührt).""" """Fehlerhafte Einträge verwerfen statt den Lauf scheitern zu lassen (die Lösung ist davon unberührt)."""
parsed = [] parsed = []
@ -482,9 +538,18 @@ def sandbox_command(script: Path, tmp: str) -> List[str]:
if SANDBOX == "bwrap": if SANDBOX == "bwrap":
raise CodeRejected("SOLVER_SANDBOX=bwrap verlangt bubblewrap, es ist aber nicht installiert.") raise CodeRejected("SOLVER_SANDBOX=bwrap verlangt bubblewrap, es ist aber nicht installiert.")
return [sys.executable, str(script)] return [sys.executable, str(script)]
# Das Home-Verzeichnis verdecken (Schlüsseldateien, .env, andere Projekte) – `validate_imports` hält den
# Zugriff nicht auf: beobachtet, dass `pandas.io.common.os` ohne `import os` das ganze Home auflistete.
# Wieder lesbar werden nur die Python-Installationen, die der Solver-Lauf braucht (.venv, Basis-Python).
home = Path.home().resolve()
python = []
for pfad in dict.fromkeys(Path(p).resolve() for p in (sys.prefix, sys.base_prefix)):
if pfad.is_relative_to(home):
python += ["--ro-bind", str(pfad), str(pfad)]
return [ return [
"bwrap", "--unshare-all", "--die-with-parent", "bwrap", "--unshare-all", "--die-with-parent",
"--ro-bind", "/", "/", # alles lesbar (Python, Solver-Bibliotheken), nichts beschreibbar "--ro-bind", "/", "/", # alles lesbar (Python, Solver-Bibliotheken), nichts beschreibbar
"--tmpfs", str(home), *python, # außer dem Home-Verzeichnis
"--bind", tmp, tmp, # einzige Ausnahme: das Arbeitsverzeichnis dieses Laufs "--bind", tmp, tmp, # einzige Ausnahme: das Arbeitsverzeichnis dieses Laufs
"--dev", "/dev", "--proc", "/proc", "--dev", "/dev", "--proc", "/proc",
"--setenv", "HOME", tmp, "--setenv", "TMPDIR", tmp, "--setenv", "HOME", tmp, "--setenv", "TMPDIR", tmp,
@ -493,7 +558,17 @@ def sandbox_command(script: Path, tmp: str) -> List[str]:
] ]
def run_solver_code(code: str, timeout: int = 60, data: dict | None = None, # Umgebung des Solver-Programms: nur diese Variablen, nie die ganze Umgebung – sonst liest der erzeugte Code
# OPENROUTER_API_KEY/GEMINI_API_KEY (nachgewiesen) und kann sie in Ergebnis-JSON oder Report schreiben.
SOLVER_ENV_VARS = ("PATH", "LANG", "LC_ALL", "LC_CTYPE", "OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS")
def solver_env(tmp: str) -> dict:
env = {name: os.environ[name] for name in SOLVER_ENV_VARS if name in os.environ}
return {**env, "HOME": tmp, "TMPDIR": tmp, "PYTHONDONTWRITEBYTECODE": "1"}
def run_solver_code(code: str, timeout: int = SOLVER_TIMEOUT, data: dict | None = None,
data_files=()) -> SolverRun: data_files=()) -> SolverRun:
try: try:
validate_imports(code) validate_imports(code)
@ -506,7 +581,7 @@ def run_solver_code(code: str, timeout: int = 60, data: dict | None = None,
# Beigelegte Daten ins Laufverzeichnis: es ist das einzige, das die Sandbox beschreibbar bindet # Beigelegte Daten ins Laufverzeichnis: es ist das einzige, das die Sandbox beschreibbar bindet
for quelle in data_files or (): for quelle in data_files or ():
shutil.copy2(quelle, Path(tmp) / Path(quelle).name) shutil.copy2(quelle, Path(tmp) / Path(quelle).name)
env = dict(os.environ) env = solver_env(tmp)
if data is not None: # geänderte Eingabedaten ohne neuen LLM-Aufruf if data is not None: # geänderte Eingabedaten ohne neuen LLM-Aufruf
daten_datei = Path(tmp) / "daten.json" daten_datei = Path(tmp) / "daten.json"
daten_datei.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") daten_datei.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
@ -557,6 +632,11 @@ def run_solver_code(code: str, timeout: int = 60, data: dict | None = None,
return SolverRun(status=SolverStatus.ERROR, error=f"Ungültiges Ergebnis von report(): {exc}: {result_lines[0][:500]}") return SolverRun(status=SolverStatus.ERROR, error=f"Ungültiges Ergebnis von report(): {exc}: {result_lines[0][:500]}")
def solver_prompt() -> str:
"""System-Prompt der Solver-Stufe mit dem Zeitbudget dieses Laufs (SOLVER_TIMEOUT)."""
return SOLVER_PROMPT_FILE.read_text(encoding="utf-8").replace("{ZEITLIMIT}", str(SOLVER_TIMEOUT))
def generate_solver_code( def generate_solver_code(
session: LLMSession, task: str, spec: OptimizationModelSpecification, session: LLMSession, task: str, spec: OptimizationModelSpecification,
previous: Optional[SolverCode] = None, previous_error: Optional[str] = None, understanding=None, previous: Optional[SolverCode] = None, previous_error: Optional[str] = None, understanding=None,
@ -578,7 +658,7 @@ def generate_solver_code(
return session.create( return session.create(
SolverCode, SolverCode,
[ [
{"role": "system", "content": SOLVER_PROMPT_FILE.read_text(encoding="utf-8")}, {"role": "system", "content": solver_prompt()},
{"role": "user", "content": content}, {"role": "user", "content": content},
], ],
step="Solver-Reparatur" if previous is not None else "Solver-Code", step="Solver-Reparatur" if previous is not None else "Solver-Code",
@ -727,7 +807,7 @@ def check_text_quality(run, summary: str = "") -> list:
MAX_SCENARIOS = 5 # jedes Szenario kostet einen weiteren Programmlauf (kein LLM-Aufruf) MAX_SCENARIOS = 5 # jedes Szenario kostet einen weiteren Programmlauf (kein LLM-Aufruf)
def verify_scenarios(run, code: str, timeout: int = 60, data_files=()) -> list: def verify_scenarios(run, code: str, timeout: int = SOLVER_TIMEOUT, data_files=()) -> list:
"""Jedes Szenario mit übersteuerten Daten NEU rechnen und den gemeldeten Zielwert bestätigen. """Jedes Szenario mit übersteuerten Daten NEU rechnen und den gemeldeten Zielwert bestätigen.
Ein Szenario ist eine Datenvariante desselben Programms – dieselbe Maschinerie wie `--resolve`. Ein Szenario ist eine Datenvariante desselben Programms – dieselbe Maschinerie wie `--resolve`.
@ -778,7 +858,7 @@ def check_scenarios_solved(run) -> list:
f"{szenario.verified_status or 'nicht ausgeführt'}) – vermutlich ein Zeitlimit ohne Lösung. " f"{szenario.verified_status or 'nicht ausgeführt'}) – vermutlich ein Zeitlimit ohne Lösung. "
f"„Keine Lösung\" nur bei bewiesener Unlösbarkeit. Haben die Szenarien dieselben Restriktionen, die " f"„Keine Lösung\" nur bei bewiesener Unlösbarkeit. Haben die Szenarien dieselben Restriktionen, die "
f"Lösung eines anderen Szenarios als Startlösung übergeben; Zeitlimits so wählen, dass alle Läufe " f"Lösung eines anderen Szenarios als Startlösung übergeben; Zeitlimits so wählen, dass alle Läufe "
f"zusammen unter 60 s bleiben.") f"zusammen unter {SOLVER_TIMEOUT} s bleiben.")
return befunde return befunde
@ -1014,7 +1094,7 @@ def _check_and_run(code: str, timeout: int, spec: OptimizationModelSpecification
def solve(session: LLMSession, task: str, spec: OptimizationModelSpecification, def solve(session: LLMSession, task: str, spec: OptimizationModelSpecification,
retries: int = 2, timeout: int = 60, log=lambda msg: None, understanding=None, retries: int = 2, timeout: int = SOLVER_TIMEOUT, log=lambda msg: None, understanding=None,
data_files=(), data_prompt: str = "") -> tuple: data_files=(), data_prompt: str = "") -> tuple:
"""Code erzeugen und ausführen; bei Fehlern bis zu `retries` Reparaturversuche. """Code erzeugen und ausführen; bei Fehlern bis zu `retries` Reparaturversuche.
@ -1028,16 +1108,19 @@ def solve(session: LLMSession, task: str, spec: OptimizationModelSpecification,
run = _check_and_run(solver_code.code, timeout, spec, understanding, task, data_files, data_prompt) run = _check_and_run(solver_code.code, timeout, spec, understanding, task, data_files, data_prompt)
run.undeclared_data = check_generated_data(solver_code.code, solver_code, spec) run.undeclared_data = check_generated_data(solver_code.code, solver_code, spec)
einheiten = check_answer_units(run, task, data_prompt) einheiten = check_answer_units(run, task, data_prompt)
zerlegung = check_objective_decomposition(run, spec)
attempt = 1 attempt = 1
while (run.status == SolverStatus.ERROR or run.feasibility or run.undeclared_data or run.unsolved_scenarios while (run.status == SolverStatus.ERROR or run.feasibility or run.undeclared_data or run.unsolved_scenarios
or einheiten) and attempt <= retries: or einheiten or zerlegung) and attempt <= retries:
fehler = run.error or (("Die Lösung hält das Modell nicht ein:\n- " + "\n- ".join(run.feasibility)) # Alle Befundarten zugleich melden – sonst kostet jede weitere Art einen eigenen Reparaturversuch
if run.feasibility else fehler = run.error or "\n\n".join(
("Eingabedaten nicht ausgewiesen:\n- " + "\n- ".join(run.undeclared_data)) titel + ":\n- " + "\n- ".join(befunde)
if run.undeclared_data else for titel, befunde in (("Die Lösung hält das Modell nicht ein", run.feasibility),
("Szenario ohne Ergebnis:\n- " + "\n- ".join(run.unsolved_scenarios)) ("Eingabedaten nicht ausgewiesen", run.undeclared_data),
if run.unsolved_scenarios else ("Szenario ohne Ergebnis", run.unsolved_scenarios),
("Einheit in der Antwort erfunden:\n- " + "\n- ".join(einheiten))) ("Einheit in der Antwort erfunden", einheiten),
("Zielwert nicht nachvollziehbar", zerlegung))
if befunde)
log(f"Solver-Code fehlgeschlagen – Reparaturversuch {attempt}/{retries}: {fehler[:200]}") log(f"Solver-Code fehlgeschlagen – Reparaturversuch {attempt}/{retries}: {fehler[:200]}")
solver_code = generate_solver_code(session, task, spec, previous=solver_code, previous_error=fehler, solver_code = generate_solver_code(session, task, spec, previous=solver_code, previous_error=fehler,
understanding=understanding, data_prompt=data_prompt) understanding=understanding, data_prompt=data_prompt)
@ -1045,8 +1128,10 @@ def solve(session: LLMSession, task: str, spec: OptimizationModelSpecification,
run = _check_and_run(solver_code.code, timeout, spec, understanding, task, data_files, data_prompt) run = _check_and_run(solver_code.code, timeout, spec, understanding, task, data_files, data_prompt)
run.undeclared_data = check_generated_data(solver_code.code, solver_code, spec) run.undeclared_data = check_generated_data(solver_code.code, solver_code, spec)
einheiten = check_answer_units(run, task, data_prompt) einheiten = check_answer_units(run, task, data_prompt)
zerlegung = check_objective_decomposition(run, spec)
attempt += 1 attempt += 1
# Bleibt die erfundene Einheit, ist das Ergebnis trotzdem richtig – Warnung, nicht unzuverlässig # Bleiben erfundene Einheit oder fehlende Zerlegung, ist die Lösung trotzdem geprüft zulässig – Warnung, nicht
run.warnings += einheiten # unzuverlässig. Eine abweichende Summe steht schon in den Warnungen (parse_result_tables).
run.warnings += einheiten + [z for z in zerlegung if z.startswith(DECOMPOSITION_MISSING)]
run.attempts = attempt run.attempts = attempt
return solver_code, run return solver_code, run

View file

@ -1,4 +1,5 @@
"""Inline-SVG-Diagramme: gültiges SVG, korrekte Geometrie, Einbindung in den Report.""" """Inline-SVG-Diagramme: gültiges SVG, korrekte Geometrie, Einbindung in den Report."""
import re
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
import pytest import pytest
@ -33,8 +34,12 @@ def test_bars_are_proportional_and_capped_in_thickness():
widths = [float(d.split("H")[1].split(" ")[0].split(",")[0]) for d in widths = [float(d.split("H")[1].split(" ")[0].split(",")[0]) for d in
[p.get("d") for p in parse(svg).findall(".//{http://www.w3.org/2000/svg}path")]] [p.get("d") for p in parse(svg).findall(".//{http://www.w3.org/2000/svg}path")]]
assert widths[0] > widths[1] assert widths[0] > widths[1]
heights = {float(p.get("d").split("V")[1].split(" ")[0]) for p in parse(svg).findall(".//{http://www.w3.org/2000/svg}path")} # Balkenhöhe = unterer Rand (letztes „x,y“ vor „H… Z“) minus oberer Rand (Startpunkt „M x,y“)
assert charts.BAR_HEIGHT <= 24 heights = set()
for p in parse(svg).findall(".//{http://www.w3.org/2000/svg}path"):
ys = [float(y) for y in re.findall(r"[\d.]+,([\d.]+)", p.get("d"))]
heights.add(max(ys) - min(ys))
assert heights == {float(charts.BAR_HEIGHT)} and charts.BAR_HEIGHT <= 24
def test_negative_values_use_the_other_colour_role(): def test_negative_values_use_the_other_colour_role():

View file

@ -412,6 +412,8 @@ def main():
"""Loest und beantwortet.""" """Loest und beantwortet."""
report("optimal", 9.0, {"x": 1.0}, report("optimal", 9.0, {"x": 1.0},
headline="Die Kosten betragen 9 Geldeinheiten.", headline="Die Kosten betragen 9 Geldeinheiten.",
tables=[{"title": "Zielwert-Zerlegung", "columns": ["Bestandteil", "Wert"],
"objective_column": "Wert", "rows": [["Kosten", 9.0]]}],
constraints=[{"name": "Bedarf", "lhs": 4.0, "sense": ">=", "rhs": 4.0}], constraints=[{"name": "Bedarf", "lhs": 4.0, "sense": ">=", "rhs": 4.0}],
variables=[{"name": "x", "value": 1.0, "var_type": "continuous", "lb": 0}], variables=[{"name": "x", "value": 1.0, "var_type": "continuous", "lb": 0}],
answers=[{"question_id": "a", "text": "Das Modell ist ein lineares Programm."}, answers=[{"question_id": "a", "text": "Das Modell ist ein lineares Programm."},
@ -457,7 +459,6 @@ if __name__ == "__main__":
def test_report_shows_data_profile_and_scenarios(tmp_path): def test_report_shows_data_profile_and_scenarios(tmp_path):
from core.model_schema import Scenario
from reporting.report import build_document, to_markdown from reporting.report import build_document, to_markdown
datei = tmp_path / "werte.csv" datei = tmp_path / "werte.csv"
datei.write_text("name,menge\nA,3\nB,4\n") datei.write_text("name,menge\nA,3\nB,4\n")
@ -468,6 +469,8 @@ DATA = {"faktor": 1.0}
def main(): def main():
"""Meldet zwei Datenvarianten.""" """Meldet zwei Datenvarianten."""
report("optimal", 7.0, {"x": 1.0}, report("optimal", 7.0, {"x": 1.0},
tables=[{"title": "Zielwert-Zerlegung", "columns": ["Bestandteil", "Wert"],
"objective_column": "Wert", "rows": [["Kosten", 7.0]]}],
constraints=[{"name": "Bedarf", "lhs": 4.0, "sense": ">=", "rhs": 4.0}], constraints=[{"name": "Bedarf", "lhs": 4.0, "sense": ">=", "rhs": 4.0}],
variables=[{"name": "x", "value": 1.0, "var_type": "continuous", "lb": 0}], variables=[{"name": "x", "value": 1.0, "var_type": "continuous", "lb": 0}],
scenarios=[{"name": "Doppelt", "description": "Faktor verdoppelt", scenarios=[{"name": "Doppelt", "description": "Faktor verdoppelt",
@ -609,3 +612,127 @@ def test_persistent_scenario_without_result_is_unreliable():
result = run_pipeline("Minimiere die Kosten.", session=session, max_revisions=0, max_restarts=0, solver_retries=1) result = run_pipeline("Minimiere die Kosten.", session=session, max_revisions=0, max_restarts=0, solver_retries=1)
assert not result.reliable assert not result.reliable
assert any(p.startswith("Szenario ohne Ergebnis") for p in result.reliability_issues) assert any(p.startswith("Szenario ohne Ergebnis") for p in result.reliability_issues)
# --- Nebenpfade: Verstehensstufe aus, Gegenprobe und --von mit vollem Kontext ---
def test_pipeline_runs_with_understanding_stage_disabled(monkeypatch):
# UNDERSTAND_RETRIES=-1 war der A/B-Arm der Verstehensstufe; das Ablehnungstor griff dann auf None zu
import TextToModel
monkeypatch.setattr(TextToModel, "UNDERSTAND_RETRIES", -1)
session = FakeSession([GOOD], [OK], [SolverCode(code=STRUCTURED_CODE)])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0)
assert result.understanding is None and not result.rejected
assert session.steps[0] == "Extraktion"
assert result.solver_run.status == SolverStatus.OPTIMAL
def test_infeasible_recheck_keeps_understanding_and_data_files(tmp_path, monkeypatch):
# Ohne Datendateien fand das Programm der Gegenprobe seine CSV nicht, ohne Verständnis fehlten die Antworten
import TextToModel
aufrufe = []
echtes_solve = TextToModel.solve
def protokoll(*args, **kwargs):
aufrufe.append(kwargs)
return echtes_solve(*args, **kwargs)
monkeypatch.setattr(TextToModel, "solve", protokoll)
csv = tmp_path / "werte.csv"
csv.write_text("a,b\n1,2\n", encoding="utf-8")
session = FakeSession(specs=[GOOD, GOOD], reviews=[OK, OK],
codes=[SolverCode(code=INFEASIBLE_CODE), SolverCode(code=STRUCTURED_CODE)])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0, data_files=[str(csv)])
assert result.restarts == 1 and len(aufrufe) == 2
gegenprobe = aufrufe[1]
assert gegenprobe["understanding"] is not None
assert list(gegenprobe["data_files"]) == [str(csv)] and "werte.csv" in gegenprobe["data_prompt"]
assert "werte.csv" in session.tasks[1] # auch die neue Extraktion sieht das Datenprofil
def test_continue_from_review_keeps_understanding(tmp_path):
from TextToModel import continue_pipeline
path = stored_result(tmp_path)
session = FakeSession([], [OK], [SolverCode(code=STRUCTURED_CODE)])
result = continue_pipeline(str(path), "pruefung", session=session)
assert result.understanding is not None
assert [q.id for q in result.understanding.sub_questions] == ["1"]
assert "ZU BEANTWORTENDE TEILFRAGEN" in session.letzter_prompt # der Solver bekommt die Teilfragen
def test_review_and_revision_see_the_data_profile():
from checks.review import review_assumptions
from modeling.extraction import revise_model
session = FakeSession([GOOD], [OK], [])
review_assumptions(session, "Aufgabe", GOOD, "BEIGELEGTE DATEN: kurse.csv")
assert "kurse.csv" in session.letzter_prompt
revise_model(session, "Aufgabe", GOOD, OK, data_prompt="BEIGELEGTE DATEN: kurse.csv")
assert "kurse.csv" in session.letzter_prompt
def test_repair_message_names_every_kind_of_finding():
# Vorher erfuhr das LLM nur von der ersten Befundart und verbrauchte für die zweite einen eigenen Versuch
from solving.solver_stage import solve
code = '''
import numpy as np
def main():
"""Verletzt die Restriktion und erzeugt Zahlen per Zufall, ohne sie auszuweisen."""
np.random.seed(0)
report("optimal", 9.0, {"x": 1.0},
constraints=[{"name": "Bedarf", "lhs": 1.0, "sense": ">=", "rhs": 4.0}],
variables=[{"name": "x", "value": 1.0, "var_type": "continuous", "lb": 0, "ub": None}])
if __name__ == "__main__":
main()
'''
session = FakeSession([], [], [SolverCode(code=code), SolverCode(code=code)])
solve(session, "Aufgabe", GOOD, retries=1)
assert session.steps == ["Solver-Code", "Solver-Reparatur"]
assert "hält das Modell nicht ein" in session.letzter_prompt
assert "Eingabedaten nicht ausgewiesen" in session.letzter_prompt
# --- Zielwert-Zerlegung in der Solver-Stufe ---
OHNE_ZERLEGUNG = STRUCTURED_CODE.split(" tables=[")[0] + " )\n\n\nif __name__ == \"__main__\":\n main()\n"
def test_missing_decomposition_is_repaired():
assert "tables" not in OHNE_ZERLEGUNG
session = FakeSession([GOOD], [OK], [SolverCode(code=OHNE_ZERLEGUNG), SolverCode(code=STRUCTURED_CODE)])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0)
assert session.steps.count("Solver-Reparatur") == 1
assert "Zielwert-Zerlegung" in session.letzter_prompt
assert result.solver_run.attempts == 2 and result.reliable
def test_persistently_missing_decomposition_is_only_a_warning():
# Die Lösung ist geprüft zulässig – nur ihr Zielwert ist nicht aus Bestandteilen belegt
from solving.solver_stage import DECOMPOSITION_MISSING
session = FakeSession([GOOD], [OK], [SolverCode(code=OHNE_ZERLEGUNG)] * 3)
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0)
assert result.reliable and result.solver_run.objective_value == 9
assert any(w.startswith(DECOMPOSITION_MISSING) for w in result.solver_run.warnings)
def test_untracked_files_do_not_mark_the_measurement_dirty(monkeypatch):
# Eine abgelegte Aufgabe in tasks/ ließ bisher jede Messung als „nicht eingecheckte Änderungen“ erscheinen
import subprocess
from benchmark import benchmark
aufrufe = []
def git(befehl, **_):
aufrufe.append(befehl)
return subprocess.CompletedProcess(befehl, 0, stdout="abc1234\n" if "rev-parse" in befehl else "")
monkeypatch.setattr(benchmark.subprocess, "run", git)
benchmark.prompt_state.cache_clear()
try:
stand = benchmark.prompt_state()
finally:
benchmark.prompt_state.cache_clear()
assert stand["dirty"] is False
assert any("--untracked-files=no" in befehl for befehl in aufrufe)

View file

@ -63,7 +63,6 @@ def test_infeasible_data_is_reported_not_crashing():
# --- CLI-Modus: gespeichertes Ergebnis neu rechnen --- # --- CLI-Modus: gespeichertes Ergebnis neu rechnen ---
def gespeichertes_ergebnis(tmp_path): def gespeichertes_ergebnis(tmp_path):
import json
from core.model_schema import PipelineResult, SolverCode from core.model_schema import PipelineResult, SolverCode
from test_model_checks import make_spec from test_model_checks import make_spec
from test_review import make_review from test_review import make_review
@ -96,3 +95,17 @@ def test_resolve_keeps_specification_and_checks_feasibility(tmp_path):
neu = resolve_result(str(gespeichertes_ergebnis(tmp_path))) neu = resolve_result(str(gespeichertes_ergebnis(tmp_path)))
assert neu.specification.problem_class.value.startswith("Linear") assert neu.specification.problem_class.value.startswith("Linear")
assert neu.solver_run.feasibility == [] assert neu.solver_run.feasibility == []
def test_resolve_keeps_earlier_reliability_issues(tmp_path):
# Neurechnen prüft nur die Zulässigkeit neu – eine falsche Zielgröße bleibt ein Befund
from core.model_schema import PipelineResult
from TextToModel import resolve_result
pfad = gespeichertes_ergebnis(tmp_path)
alt = PipelineResult.model_validate_json(pfad.read_text(encoding="utf-8"))
alt.reliable = False
alt.reliability_issues = ["Zielfunktion passt nicht zur Frage", "Zulässigkeitsprüfung: alter Befund"]
pfad.write_text(alt.model_dump_json(), encoding="utf-8")
neu = resolve_result(str(pfad))
assert neu.reliability_issues == ["Zielfunktion passt nicht zur Frage"] # alter Zulässigkeitsbefund entfällt
assert not neu.reliable

View file

@ -140,7 +140,7 @@ def test_normal_decomposition_has_no_magnitude_warning():
def test_report_shows_multistart_stability(): def test_report_shows_multistart_stability():
from core.model_schema import Multistart, SolverRun from core.model_schema import Multistart
from reporting.report import build_document, to_markdown from reporting.report import build_document, to_markdown
from test_pipeline import FakeSession from test_pipeline import FakeSession
from TextToModel import run_pipeline from TextToModel import run_pipeline
@ -345,3 +345,54 @@ def test_english_unit_words_cover_german_answers():
aufgabe = "One ton of raw material X costs 3 thousand euros; at least 4 tons are required." aufgabe = "One ton of raw material X costs 3 thousand euros; at least 4 tons are required."
assert check_answer_units(_run_with_answer("Es sollten 1 Tonne X und 3 Tonnen Y gekauft werden, Kosten 9 Tausend Euro."), aufgabe) == [] assert check_answer_units(_run_with_answer("Es sollten 1 Tonne X und 3 Tonnen Y gekauft werden, Kosten 9 Tausend Euro."), aufgabe) == []
assert check_answer_units(_run_with_answer("It takes 12 hours."), "Minimize the cost.") != [] # erfunden bleibt erfunden assert check_answer_units(_run_with_answer("It takes 12 hours."), "Minimize the cost.") != [] # erfunden bleibt erfunden
# --- Summenzeilen und Zielwert-Zerlegung (check_table, check_objective_decomposition) ---
def _tabelle(rows):
from core.model_schema import ResultTable
return ResultTable(title="Zielwert-Zerlegung", columns=["Bestandteil", "Wert"], rows=rows, objective_column="Wert")
@pytest.mark.parametrize("rows", [
[["Nutzen der eingepackten Gegenstände", 51.0], ["Zielwert", 51.0]], # beobachtet: 15 Rucksack-Läufe
[["Gesamtnutzen der eingepackten Gegenstände", 51]], # einzige Zeile ist keine Summe
[["Nutzen Kocher", 13.0], ["Nutzen Schlafsack", 23.0], ["Nutzen Proviant", 15.0], ["Zielwert", 51.0]],
[["Kocher", 13.0], ["Schlafsack", 23.0], ["Zwischensumme", 36.0], ["Proviant", 15.0], ["Summe", 51.0]],
])
def test_components_named_like_a_sum_are_not_dropped(rows):
from solving.solver_stage import check_table
belegt, abweichung, warnungen = check_table(_tabelle(rows), 51.0)
assert belegt and abweichung is None and warnungen == []
def test_wrong_decomposition_is_still_detected():
from solving.solver_stage import check_table
belegt, abweichung, _ = check_table(_tabelle([["Kocher", 13.0], ["Schlafsack", 23.0], ["Summe", 36.0]]), 51.0)
assert not belegt and "weicht vom Zielwert" in abweichung
def _lauf(tables, status="optimal", objective=51.0):
from core.model_schema import SolverRun, SolverStatus
return SolverRun(status=SolverStatus(status), objective_value=objective, tables=tables)
def test_linear_model_without_decomposition_is_a_finding():
from solving.solver_stage import DECOMPOSITION_MISSING, check_objective_decomposition
befunde = check_objective_decomposition(_lauf([]), make_spec())
assert len(befunde) == 1 and befunde[0].startswith(DECOMPOSITION_MISSING)
def test_decomposition_is_optional_for_non_additive_objectives():
# Volatilität = Wurzel: keine Summe, also keine Pflicht – eine vorhandene Beitragsspalte muss aber stimmen
from solving.solver_stage import check_objective_decomposition
spec = make_spec(problem_class="Second-Order Cone Programming (SOCP)", target_stack="CVXPY")
assert check_objective_decomposition(_lauf([]), spec) == []
falsch = _tabelle([["Rendite", 0.1], ["Risiko", -0.02]])
assert "weicht vom Zielwert" in check_objective_decomposition(_lauf([falsch], objective=0.5), spec)[0]
def test_no_decomposition_needed_without_a_solution():
from solving.solver_stage import check_objective_decomposition
assert check_objective_decomposition(_lauf([], status="infeasible", objective=None), make_spec()) == []
assert check_objective_decomposition(_lauf([_tabelle([["Kosten", 51.0]])]), make_spec()) == []

View file

@ -35,6 +35,9 @@ def main():
constraints=[{"name": "Bedarf", "lhs": x + y, "sense": ">=", "rhs": 4}], constraints=[{"name": "Bedarf", "lhs": x + y, "sense": ">=", "rhs": 4}],
variables=[{"name": "x", "value": x, "var_type": "continuous", "lb": 0, "ub": 3}, variables=[{"name": "x", "value": x, "var_type": "continuous", "lb": 0, "ub": 3},
{"name": "y", "value": y, "var_type": "continuous", "lb": 0, "ub": 3}], {"name": "y", "value": y, "var_type": "continuous", "lb": 0, "ub": 3}],
tables=[{"title": "Zielwert-Zerlegung", "columns": ["Bestandteil", "Wert"], "objective_column": "Wert",
"rows": [["Kosten Rohstoff X", COSTS[0] * x], ["Kosten Rohstoff Y", COSTS[1] * y],
["Zielwert", float(res.fun)]]}],
) )

View file

@ -482,7 +482,6 @@ def test_path_must_use_declared_arcs():
def test_path_length_must_match_objective(): def test_path_length_must_match_objective():
from core.model_schema import SolverRun
run = graph_run() run = graph_run()
run.objective_value = 7.0 # 4 + 5 = 9 run.objective_value = 7.0 # 4 + 5 = 9
assert any("weicht vom Zielwert" in w for w in check_graph(run, "minimize")) assert any("weicht vom Zielwert" in w for w in check_graph(run, "minimize"))
@ -757,7 +756,6 @@ def test_ordinary_programs_are_not_suspected():
def test_scenario_failing_without_error_text_gets_a_reason(): def test_scenario_failing_without_error_text_gets_a_reason():
# beobachtet (Portfolio Nr. 2): Szenario ohne Lösung im Zeitlimit, Warnung endete mit leerem Grund # beobachtet (Portfolio Nr. 2): Szenario ohne Lösung im Zeitlimit, Warnung endete mit leerem Grund
from core.model_schema import Scenario
from solving.solver_stage import verify_scenarios from solving.solver_stage import verify_scenarios
code = SZENARIO_CODE.replace(' basis = loesen(DATA)\n', code = SZENARIO_CODE.replace(' basis = loesen(DATA)\n',
' basis = loesen(DATA)\n if DATA["preis"] > 15:\n' ' basis = loesen(DATA)\n if DATA["preis"] > 15:\n'
@ -834,3 +832,55 @@ def test_unprotected_run_carries_the_warning_into_the_result(monkeypatch):
code = 'def main():\n """Test."""\n report("optimal", 1.0, {})\n\n\nif __name__ == "__main__":\n main()\n' code = 'def main():\n """Test."""\n report("optimal", 1.0, {})\n\n\nif __name__ == "__main__":\n main()\n'
run = solver_stage._check_and_run(code, 30, make_spec()) run = solver_stage._check_and_run(code, 30, make_spec())
assert any("Sandbox abgeschaltet" in w for w in run.warnings) assert any("Sandbox abgeschaltet" in w for w in run.warnings)
# Liest Umgebung und Home-Verzeichnis ohne `import os` – über ein Attribut von pandas, das
# validate_imports nicht sieht. Nachgewiesen: vorher waren API-Schlüssel und ~ vollständig lesbar.
PEEK_CODE = '''
import pandas as pd
def main():
"""Späht Umgebung und Home-Verzeichnis aus."""
_os = pd.io.common.os
try:
eintraege = len(_os.listdir({home!r}))
except OSError:
eintraege = -1
report("optimal", 0.0, {{"schluessel": len(_os.environ.get("OPENROUTER_API_KEY", "")),
"home_eintraege": eintraege}})
if __name__ == "__main__":
main()
'''
def test_program_does_not_see_api_keys(monkeypatch):
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-geheim")
run = run_solver_code(PEEK_CODE.format(home="/"))
assert run.status == SolverStatus.OPTIMAL, run.error
assert run.solution["schluessel"] == 0
@pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap nicht installiert")
def test_sandbox_hides_the_home_directory():
from pathlib import Path
run = run_solver_code(PEEK_CODE.format(home=str(Path.home())))
assert run.status == SolverStatus.OPTIMAL, run.error
# Sichtbar bleiben höchstens die Einhängepunkte der Python-Installationen (.venv, Basis-Python)
assert run.solution["home_eintraege"] <= 2
def test_time_budget_in_the_prompt_follows_solver_timeout(monkeypatch):
# Ein höheres Zeitlimit nützt nur, wenn auch das Programm seine internen Solver-Zeitlimits danach wählt
from solving import solver_stage
assert "{ZEITLIMIT}" not in solver_stage.solver_prompt() and "unter 60 s" in solver_stage.solver_prompt()
monkeypatch.setattr(solver_stage, "SOLVER_TIMEOUT", 180)
assert "unter 180 s" in solver_stage.solver_prompt() and "unter 60 s" not in solver_stage.solver_prompt()
def test_timeout_ends_a_slow_program():
code = "def main():\n while True:\n pass\n\n\nif __name__ == '__main__':\n main()\n"
run = run_solver_code(code, timeout=1)
assert run.status == SolverStatus.ERROR and "Timeout nach 1 s" in run.error