feat: TextToModel.py als CLI-Programm mit main(), gruppierten Optionen und Defaults
- main(argv) -> int, build_parser() mit Gruppen Eingabe/Modus/Ablauf/LLM/Solver/Ausgabe, Hilfe mit Defaults, Umgebungsvariablen, Beispielen und Exit-Codes - Optionen für bisher reine Umgebungsvariablen (--solver-timeout, --sandbox, --max-tokens, --llm-timeout, --reasoning, --solver-retries, --understand-retries, --infeasible-recheck) - -o/--json und -a/--ausgabe PRÄFIX; im Terminal Kurzfassung statt JSON - Exit-Codes: 1 Eingabefehler, 64 Aufruffehler (nicht argparse' 2 = unzuverlässig), 130 Abbruch - solver_stage: Zeitlimit-Default zur Aufrufzeit auflösen, damit --solver-timeout wirkt - tests/test_cli.py Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
f101b63947
commit
422360ffa5
1 changed files with 382 additions and 70 deletions
284
TextToModel.py
284
TextToModel.py
|
|
@ -7,6 +7,7 @@ from typing import List
|
|||
|
||||
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 core import llm
|
||||
from core.llm import DEFAULT_MODEL, LLMSession
|
||||
from core.model_schema import (
|
||||
SolverStatus,
|
||||
|
|
@ -18,8 +19,9 @@ from core.model_schema import (
|
|||
from data import data_profile
|
||||
from modeling.extraction import revise_model, text_to_model
|
||||
from reporting.report import write_report
|
||||
from solving import solver_stage
|
||||
from solving.script_export import build_solver_script
|
||||
from solving.solver_stage import (SOLVER_TIMEOUT, check_feasibility, check_multistart, check_sensitivity,
|
||||
from solving.solver_stage import (check_feasibility, check_multistart, check_sensitivity,
|
||||
run_solver_code, solve)
|
||||
from understanding.figures import FigureError, normalize_figure_ids, split_figures, with_transcriptions
|
||||
from understanding.understand import read_figures, understand_task
|
||||
|
|
@ -386,7 +388,7 @@ def continue_pipeline(
|
|||
return result
|
||||
|
||||
|
||||
def resolve_result(result_json: str, data_file: str | None = None, solver_timeout: int = SOLVER_TIMEOUT,
|
||||
def resolve_result(result_json: str, data_file: str | None = None, solver_timeout: int | None = None,
|
||||
data_files=()) -> PipelineResult:
|
||||
"""Gespeichertes Ergebnis mit (geänderten) Daten neu rechnen – ohne einen einzigen LLM-Aufruf.
|
||||
|
||||
|
|
@ -420,86 +422,250 @@ def resolve_result(result_json: str, data_file: str | None = None, solver_timeou
|
|||
|
||||
|
||||
def read_task(task_file: str | None) -> str:
|
||||
if task_file:
|
||||
"""Aufgabe aus Datei, aus stdin ('-' oder umgeleitet) oder – ohne beides – die Beispielaufgabe."""
|
||||
if task_file and task_file != "-":
|
||||
task = Path(task_file).read_text(encoding="utf-8")
|
||||
elif not sys.stdin.isatty():
|
||||
elif task_file == "-" or not sys.stdin.isatty():
|
||||
task = sys.stdin.read()
|
||||
else:
|
||||
task = ""
|
||||
# Leere Eingabe würde das LLM ein beliebiges Problem erfinden lassen
|
||||
return task if task.strip() else task_description
|
||||
if not task.strip():
|
||||
log("Keine Aufgabe angegeben – verwende die Beispielaufgabe (Schichtplanung).")
|
||||
return task_description
|
||||
return task
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(description="Textaufgabe → Optimierungsmodell → Prüfung → Lösung")
|
||||
parser.add_argument("task_file", nargs="?", help="Datei mit der Aufgabe (sonst stdin, sonst Beispielaufgabe)")
|
||||
parser.add_argument("--no-solve", action="store_true", help="nur modellieren und prüfen, keinen Solver-Code ausführen")
|
||||
parser.add_argument("--max-revisions", type=int, default=MAX_REVISIONS)
|
||||
parser.add_argument("--max-restarts", type=int, default=MAX_RESTARTS, help="komplette Neustarts, wenn das Modell nach allen Korrekturrunden fehlerhaft ist")
|
||||
parser.add_argument(
|
||||
"--model", default=DEFAULT_MODEL,
|
||||
help=f"Modell (Default: {DEFAULT_MODEL}, Umgebung: LLM_MODEL); mit '/' -> OpenRouter, sonst Gemini",
|
||||
# Exit-Codes für Skripte. Aufruffehler bewusst nicht argparse' 2 – das hieße hier „unzuverlässig“.
|
||||
EXIT_OK, EXIT_ERROR, EXIT_UNRELIABLE, EXIT_REJECTED, EXIT_USAGE, EXIT_INTERRUPTED = 0, 1, 2, 3, 64, 130
|
||||
|
||||
EPILOG = f"""\
|
||||
Beispiele:
|
||||
%(prog)s aufgabe.md --ausgabe output/aufgabe # schreibt output/aufgabe.json, .html und .py
|
||||
%(prog)s aufgabe.md --daten kurse.csv -o output/aufgabe.json --report output/aufgabe.html
|
||||
echo "Ein Betrieb …" | %(prog)s - --report output/modell.md
|
||||
%(prog)s aufgabe.md --no-solve --max-revisions 0 -m deepseek/deepseek-v4-pro-0813
|
||||
%(prog)s --resolve output/aufgabe.json --data neue_werte.json -o output/variante.json # ohne LLM
|
||||
%(prog)s --von solver output/aufgabe.json -o output/neu.json # nur Solver-Stufe neu
|
||||
|
||||
Ausgabe: Ohne -o/--ausgabe geht das Ergebnis-JSON nach stdout, wenn stdout umgeleitet ist; im Terminal
|
||||
erscheint stattdessen eine Kurzfassung. Der Fortschritt steht immer auf stderr.
|
||||
|
||||
Exit-Codes: {EXIT_OK} zuverlässiges Ergebnis · {EXIT_ERROR} Fehler (Datei, Abbildung, gespeichertes Ergebnis) ·
|
||||
{EXIT_UNRELIABLE} Ergebnis unzuverlässig · {EXIT_REJECTED} keine Optimierungsaufgabe · {EXIT_USAGE} falscher Aufruf · \
|
||||
{EXIT_INTERRUPTED} abgebrochen
|
||||
|
||||
Optionen ohne Angabe übernehmen die Umgebungsvariable in Klammern bzw. deren Default.
|
||||
Modell je Stufe nur über die Umgebung: LLM_MODEL_UNDERSTANDING, _EXTRACTION, _REVIEW, _REVISION, _SOLVER."""
|
||||
|
||||
|
||||
class _Parser(argparse.ArgumentParser):
|
||||
def error(self, message):
|
||||
self.print_usage(sys.stderr)
|
||||
self.exit(EXIT_USAGE, f"{self.prog}: Fehler: {message}\n")
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = _Parser(
|
||||
prog="TextToModel.py", description="Textaufgabe → Optimierungsmodell → Prüfung → Lösung",
|
||||
epilog=EPILOG, formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
)
|
||||
parser.add_argument("--resolve", metavar="ERGEBNIS.JSON",
|
||||
help="gespeichertes Ergebnis erneut rechnen, ohne LLM (z. B. nach Datenänderung)")
|
||||
parser.add_argument("--von", nargs=2, metavar=("STUFE", "ERGEBNIS.JSON"),
|
||||
help="gespeichertes Ergebnis ab einer Stufe fortsetzen, statt alles zu wiederholen: "
|
||||
"'pruefung' (Spezifikation behalten, neu prüfen und korrigieren) oder "
|
||||
"'solver' (auch den Prüfbericht behalten, nur Solver-Code neu erzeugen)")
|
||||
parser.add_argument("--daten", nargs="+", metavar="DATEI",
|
||||
help="beigelegte Datendateien (CSV/Excel); ihr Profil geht in die Prompts, "
|
||||
"die Dateien selbst ins Arbeitsverzeichnis des Programms "
|
||||
"(mit --resolve/--von erneut angeben, gespeichert ist nur ihr Profil)")
|
||||
parser.add_argument("--data", metavar="DATEN.JSON",
|
||||
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("--code", metavar="DATEI", help="erzeugten Solver-Code als eigenständiges Python-Skript speichern")
|
||||
args = parser.parse_args()
|
||||
|
||||
eingabe = parser.add_argument_group("Eingabe")
|
||||
eingabe.add_argument("aufgabe", nargs="?", metavar="AUFGABE",
|
||||
help="Datei mit der Aufgabe (Text/Markdown); '-' = stdin. Fehlt sie und ist stdin nicht "
|
||||
"umgeleitet, läuft die Beispielaufgabe")
|
||||
eingabe.add_argument("--daten", nargs="+", metavar="DATEI",
|
||||
help="beigelegte Datendateien (CSV/Excel); ihr Profil geht in die Prompts, die Dateien "
|
||||
"ins Arbeitsverzeichnis des Programms (bei --resolve/--von erneut angeben)")
|
||||
|
||||
modus = parser.add_argument_group("Modus (höchstens einer; ohne Angabe: kompletter Lauf)")
|
||||
exklusiv = modus.add_mutually_exclusive_group()
|
||||
exklusiv.add_argument("--no-solve", action="store_true",
|
||||
help="nur verstehen, modellieren und prüfen – kein Solver-Code")
|
||||
exklusiv.add_argument("--von", nargs=2, metavar=("STUFE", "ERGEBNIS.JSON"),
|
||||
help="gespeichertes Ergebnis fortsetzen: 'pruefung' (Spezifikation behalten, neu prüfen "
|
||||
"und korrigieren) oder 'solver' (auch Prüfbericht behalten, nur Solver-Code neu)")
|
||||
exklusiv.add_argument("--resolve", metavar="ERGEBNIS.JSON",
|
||||
help="gespeichertes Ergebnis ohne LLM neu rechnen (z. B. nach Datenänderung)")
|
||||
modus.add_argument("--data", metavar="DATEN.JSON",
|
||||
help="mit --resolve: Eingabewerte (DATA) des Programms ganz oder teilweise ersetzen")
|
||||
|
||||
ablauf = parser.add_argument_group("Ablauf")
|
||||
ablauf.add_argument("--max-revisions", type=int, default=MAX_REVISIONS, metavar="N",
|
||||
help=f"Korrekturrunden je Modellierung (MAX_REVISIONS, {MAX_REVISIONS})")
|
||||
ablauf.add_argument("--max-restarts", type=int, default=MAX_RESTARTS, metavar="N",
|
||||
help=f"Neustarts der Modellierung, wenn sie fehlerhaft bleibt (MAX_RESTARTS, {MAX_RESTARTS})")
|
||||
ablauf.add_argument("--solver-retries", type=int, default=SOLVER_RETRIES, metavar="N",
|
||||
help=f"Reparaturen des Solver-Codes (SOLVER_RETRIES, {SOLVER_RETRIES})")
|
||||
ablauf.add_argument("--understand-retries", type=int, metavar="N",
|
||||
help=f"Nachfassen beim Verstehen, -1 = Stufe aus (UNDERSTAND_RETRIES, {UNDERSTAND_RETRIES})")
|
||||
ablauf.add_argument("--infeasible-recheck", type=int, choices=(0, 1), metavar="{0,1}",
|
||||
help=f"Gegenprobe bei 'unlösbar' (INFEASIBLE_RECHECK, {INFEASIBLE_RECHECK})")
|
||||
|
||||
ki = parser.add_argument_group("LLM")
|
||||
ki.add_argument("-m", "--model", default=DEFAULT_MODEL,
|
||||
help=f"Modell; mit '/' → OpenRouter, sonst Gemini (LLM_MODEL, {DEFAULT_MODEL})")
|
||||
ki.add_argument("--reasoning", choices=("low", "medium", "high"),
|
||||
help="Denkaufwand, nur OpenRouter (LLM_REASONING_EFFORT, Modellvorgabe)")
|
||||
ki.add_argument("--max-tokens", type=int, metavar="N",
|
||||
help=f"Antwortlänge je Aufruf (LLM_MAX_TOKENS, {llm.LLM_MAX_TOKENS})")
|
||||
ki.add_argument("--llm-timeout", type=int, metavar="SEK",
|
||||
help=f"Zeitlimit je Aufruf (LLM_TIMEOUT, {llm.LLM_TIMEOUT})")
|
||||
|
||||
solver = parser.add_argument_group("Solver")
|
||||
solver.add_argument("--solver-timeout", type=int, metavar="SEK",
|
||||
help=f"Zeitlimit je Programmlauf, geht auch als Zeitbudget ans LLM "
|
||||
f"(SOLVER_TIMEOUT, {solver_stage.SOLVER_TIMEOUT})")
|
||||
solver.add_argument("--sandbox", choices=("auto", "bwrap", "aus"),
|
||||
help=f"Isolierung des erzeugten Codes (SOLVER_SANDBOX, {solver_stage.SANDBOX})")
|
||||
|
||||
ausgabe = parser.add_argument_group("Ausgabe")
|
||||
ausgabe.add_argument("-o", "--json", metavar="DATEI", help="Ergebnis-JSON in diese Datei statt nach stdout")
|
||||
ausgabe.add_argument("--report", metavar="DATEI", help="Report mit Formeln und Solver-Code (.md oder .html)")
|
||||
ausgabe.add_argument("--code", metavar="DATEI", help="Solver-Code als eigenständiges Python-Skript")
|
||||
ausgabe.add_argument("-a", "--ausgabe", metavar="PRÄFIX",
|
||||
help="Kurzform für -o PRÄFIX.json --report PRÄFIX.html --code PRÄFIX.py "
|
||||
"(ausdrückliche Angaben haben Vorrang)")
|
||||
return parser
|
||||
|
||||
|
||||
def _check_args(parser: argparse.ArgumentParser, args: argparse.Namespace) -> None:
|
||||
if args.ausgabe:
|
||||
args.json = args.json or f"{args.ausgabe}.json"
|
||||
args.report = args.report or f"{args.ausgabe}.html"
|
||||
if not args.no_solve:
|
||||
args.code = args.code or f"{args.ausgabe}.py"
|
||||
if args.report and Path(args.report).suffix.lower() not in (".md", ".html", ".htm"):
|
||||
parser.error("--report braucht die Endung .md oder .html")
|
||||
if args.code and args.no_solve:
|
||||
parser.error("--code ist mit --no-solve nicht möglich (es wird kein Solver-Code erzeugt)")
|
||||
if args.data and not args.resolve:
|
||||
parser.error("--data wirkt nur zusammen mit --resolve")
|
||||
if args.von:
|
||||
if args.resolve:
|
||||
parser.error("--von und --resolve schließen sich aus")
|
||||
if args.von[0] not in STAGES:
|
||||
parser.error(f"--von braucht eine Stufe aus {{{', '.join(STAGES)}}}, nicht '{args.von[0]}'")
|
||||
if args.task_file:
|
||||
parser.error("--von nimmt die Aufgabe aus dem gespeicherten Ergebnis – keine Aufgabendatei angeben")
|
||||
if args.no_solve:
|
||||
parser.error("--von endet immer mit der Solver-Stufe; --no-solve wäre wirkungslos")
|
||||
if args.von and args.von[0] not in STAGES:
|
||||
parser.error(f"--von braucht eine Stufe aus {{{', '.join(STAGES)}}}, nicht '{args.von[0]}'")
|
||||
if (args.von or args.resolve) and args.aufgabe:
|
||||
parser.error("--von/--resolve nehmen die Aufgabe aus dem gespeicherten Ergebnis – keine Aufgabendatei angeben")
|
||||
for name, wert in (("--max-revisions", args.max_revisions), ("--max-restarts", args.max_restarts),
|
||||
("--solver-retries", args.solver_retries)):
|
||||
if wert < 0:
|
||||
parser.error(f"{name} darf nicht negativ sein")
|
||||
for name, wert in (("--max-tokens", args.max_tokens), ("--llm-timeout", args.llm_timeout),
|
||||
("--solver-timeout", args.solver_timeout)):
|
||||
if wert is not None and wert <= 0:
|
||||
parser.error(f"{name} muss positiv sein")
|
||||
eingaben = [args.aufgabe if args.aufgabe != "-" else None, args.von[1] if args.von else None,
|
||||
args.resolve, args.data, *(args.daten or [])]
|
||||
for datei in eingaben:
|
||||
if datei and not Path(datei).is_file():
|
||||
parser.error(f"Datei nicht gefunden: {datei}")
|
||||
|
||||
if args.resolve:
|
||||
result = resolve_result(args.resolve, args.data, data_files=args.daten or ())
|
||||
elif args.von:
|
||||
result = continue_pipeline(args.von[1], args.von[0], model=args.model, max_revisions=args.max_revisions,
|
||||
data_files=args.daten or ())
|
||||
|
||||
def apply_settings(args: argparse.Namespace) -> None:
|
||||
"""Optionen, die sonst nur als Umgebungsvariable wirken, für diesen Prozess setzen.
|
||||
|
||||
Die Module lesen die Umgebung beim Import; deshalb werden ihre Konstanten direkt ersetzt – sie werden
|
||||
zur Aufrufzeit gelesen. Modell je Stufe liest `LLMSession` aus der Umgebung und bleibt dort.
|
||||
"""
|
||||
global UNDERSTAND_RETRIES, INFEASIBLE_RECHECK
|
||||
if args.understand_retries is not None:
|
||||
UNDERSTAND_RETRIES = args.understand_retries
|
||||
if args.infeasible_recheck is not None:
|
||||
INFEASIBLE_RECHECK = args.infeasible_recheck
|
||||
if args.reasoning:
|
||||
llm.REASONING_EFFORT = args.reasoning
|
||||
if args.max_tokens:
|
||||
llm.LLM_MAX_TOKENS = args.max_tokens
|
||||
if args.llm_timeout:
|
||||
llm.LLM_TIMEOUT = args.llm_timeout
|
||||
if args.solver_timeout:
|
||||
solver_stage.SOLVER_TIMEOUT = args.solver_timeout
|
||||
if args.sandbox:
|
||||
solver_stage.SANDBOX = args.sandbox
|
||||
|
||||
|
||||
def summary(result: PipelineResult) -> str:
|
||||
"""Kurzfassung fürs Terminal: Status, Zielwert, Antworten, Zuverlässigkeit."""
|
||||
if result.rejected:
|
||||
return "Abgelehnt – " + "; ".join(result.reliability_issues)
|
||||
zeilen = []
|
||||
run = result.solver_run
|
||||
if run is None:
|
||||
spec = result.specification
|
||||
zeilen.append(f"Modell: {spec.problem_class.value} → {spec.target_stack.value} (kein Solver-Lauf)")
|
||||
else:
|
||||
try:
|
||||
result = run_pipeline(
|
||||
read_task(args.task_file), model=args.model,
|
||||
max_revisions=args.max_revisions, solve_model=not args.no_solve, max_restarts=args.max_restarts,
|
||||
data_files=args.daten or (), task_dir=Path(args.task_file).parent if args.task_file else None,
|
||||
)
|
||||
except FigureError as exc:
|
||||
parser.error(str(exc))
|
||||
print(result.model_dump_json(indent=2))
|
||||
zeilen.append(f"Status: {run.status.value}" + (f" · Zielwert: {run.objective_value:g}"
|
||||
if run.objective_value is not None else ""))
|
||||
if run.headline:
|
||||
zeilen.append(run.headline)
|
||||
zeilen += [f" {a.question_id}) {a.text}" for a in run.answers]
|
||||
zeilen.append("Zuverlässig." if result.reliable
|
||||
else "UNZUVERLÄSSIG: " + "; ".join(result.reliability_issues))
|
||||
return "\n".join(zeilen)
|
||||
|
||||
|
||||
def _write_outputs(args: argparse.Namespace, result: PipelineResult) -> None:
|
||||
daten = result.model_dump_json(indent=2)
|
||||
if args.json:
|
||||
Path(args.json).parent.mkdir(parents=True, exist_ok=True)
|
||||
Path(args.json).write_text(daten + "\n", encoding="utf-8")
|
||||
log(f"Ergebnis geschrieben: {args.json}")
|
||||
elif sys.stdout.isatty():
|
||||
print(summary(result))
|
||||
log("(JSON: mit -o DATEI speichern oder stdout umleiten)")
|
||||
else:
|
||||
print(daten)
|
||||
if args.report:
|
||||
Path(args.report).parent.mkdir(parents=True, exist_ok=True)
|
||||
write_report(result, args.report)
|
||||
log(f"Report geschrieben: {args.report}")
|
||||
if args.code and result.solver_code:
|
||||
Path(args.code).parent.mkdir(parents=True, exist_ok=True)
|
||||
Path(args.code).write_text(build_solver_script(result), encoding="utf-8")
|
||||
Path(args.code).chmod(0o755)
|
||||
log(f"Solver-Code geschrieben: {args.code} (ausführen: .venv/bin/python {args.code})")
|
||||
elif args.code:
|
||||
log(f"Kein Solver-Code geschrieben ({args.code}): Solver wurde nicht gestartet.")
|
||||
if result.rejected:
|
||||
sys.exit(3) # für Skripte: keine Optimierungsaufgabe, nichts gerechnet
|
||||
if not result.reliable:
|
||||
sys.exit(2) # für Skripte: Ergebnis vorhanden, aber unzuverlässig
|
||||
|
||||
# Erwartet für das Beispiel:
|
||||
# result.specification.target_stack -> TargetStack.OR_TOOLS_CPSAT
|
||||
# result.specification.problem_class -> ProblemClass.CP
|
||||
|
||||
def main(argv: List[str] | None = None) -> int:
|
||||
parser = build_parser()
|
||||
args = parser.parse_args(argv)
|
||||
_check_args(parser, args)
|
||||
apply_settings(args)
|
||||
daten = args.daten or ()
|
||||
try:
|
||||
if args.resolve or args.von:
|
||||
try:
|
||||
if args.resolve:
|
||||
result = resolve_result(args.resolve, args.data, data_files=daten)
|
||||
else:
|
||||
result = continue_pipeline(args.von[1], args.von[0], model=args.model,
|
||||
max_revisions=args.max_revisions,
|
||||
solver_retries=args.solver_retries, data_files=daten)
|
||||
except ValueError as exc: # auch pydantic.ValidationError: gespeichertes Ergebnis passt nicht
|
||||
log(f"Fehler in {args.resolve or args.von[1]}: {exc}")
|
||||
return EXIT_ERROR
|
||||
else:
|
||||
task_dir = Path(args.aufgabe).parent if args.aufgabe and args.aufgabe != "-" else None
|
||||
result = run_pipeline(
|
||||
read_task(args.aufgabe), model=args.model, max_revisions=args.max_revisions,
|
||||
solve_model=not args.no_solve, solver_retries=args.solver_retries, max_restarts=args.max_restarts,
|
||||
data_files=daten, task_dir=task_dir,
|
||||
)
|
||||
except KeyboardInterrupt:
|
||||
log("Abgebrochen.")
|
||||
return EXIT_INTERRUPTED
|
||||
except (FigureError, OSError) as exc:
|
||||
log(f"Fehler: {exc}")
|
||||
return EXIT_ERROR
|
||||
_write_outputs(args, result)
|
||||
if result.rejected:
|
||||
return EXIT_REJECTED # keine Optimierungsaufgabe, nichts gerechnet
|
||||
if not result.reliable:
|
||||
return EXIT_UNRELIABLE # Ergebnis vorhanden, aber unzuverlässig
|
||||
return EXIT_OK
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
|
|||
Loading…
Reference in a new issue