text_to_solver_program/TextToModel.py

import argparse
import json
import os
import sys
from pathlib import Path
from typing import List

from checks.model_checks import check_spec, check_understanding, has_errors
from checks.review import needs_work, review_assumptions, select_best
from core.llm import DEFAULT_MODEL, LLMSession
from core.model_schema import (
    AssumptionReview,
    SolverStatus,
    CheckIssue,
    CheckSeverity,
    OptimizationModelSpecification,
    PipelineResult,
    TaskUnderstanding,
)
from data import data_profile
from modeling.extraction import revise_model, text_to_model
from reporting.report import write_report
from solving.script_export import build_solver_script
from solving.solver_stage import check_feasibility, check_multistart, check_sensitivity, run_solver_code, solve
from understanding.understand import understand_task

MAX_REVISIONS = int(os.environ.get("MAX_REVISIONS", "2"))
SOLVER_RETRIES = int(os.environ.get("SOLVER_RETRIES", "2"))
MAX_RESTARTS = int(os.environ.get("MAX_RESTARTS", "1"))
# "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.
INFEASIBLE_RECHECK = int(os.environ.get("INFEASIBLE_RECHECK", "1"))
# Ein Nachfassen, wenn die Verstehensstufe Zahlen oder Teilfragen übersehen hat (0 schaltet die Stufe ab)
UNDERSTAND_RETRIES = int(os.environ.get("UNDERSTAND_RETRIES", "1"))

task_description = """
Ein Logistikzentrum muss 12 Mitarbeiter auf 3 Acht-Stunden-Schichten verteilen.
In Schicht 1 werden mindestens 3 Mitarbeiter benötigt, in Schicht 2 mindestens 5
und in Schicht 3 genau 2. Kein Mitarbeiter darf zwei aufeinanderfolgende
Schichten arbeiten. Zudem gilt: Wenn Mitarbeiter A arbeitet, muss Mitarbeiter B
in derselben Schicht sein. Die Gesamtkosten für Schichtzulagen sollen minimiert werden.
"""


def log(msg: str) -> None:
    # stderr, damit stdout reines JSON bleibt
    print(msg, file=sys.stderr)


def _understand(session, task: str, label: str, data_prompt: str = "") -> tuple:
    """Verstehensstufe samt Prüfung; ein Nachfassen, wenn Zahlen oder Teilfragen fehlen."""
    if UNDERSTAND_RETRIES < 0:
        return None, []
    log(f"[{label}] Lese die Aufgabe (Teilfragen, Zahlen) ...")
    understanding = understand_task(session, task, data_prompt=data_prompt)
    issues = check_understanding(understanding, task)
    for versuch in range(UNDERSTAND_RETRIES):
        if not issues or not understanding.is_optimization_task:
            break      # eine Nicht-Aufgabe wird abgelehnt, nicht vervollständigt
        log(f"[{label}] Verständnis unvollständig – fasse nach ({versuch + 1}/{UNDERSTAND_RETRIES}): "
            + "; ".join(i.code for i in issues))
        understanding = understand_task(session, task, issues, data_prompt)
        issues = check_understanding(understanding, task)
    if understanding.sub_questions:
        log(f"  {len(understanding.sub_questions)} Teilfrage(n): "
            + ", ".join(q.id for q in understanding.sub_questions)
            + f"; {len(understanding.facts)} Zahlenwerte erfasst")
    return understanding, issues


def _model_attempt(session, task: str, max_revisions: int, label: str, hint: str = "",
                   spec: OptimizationModelSpecification | None = None,
                   understanding: TaskUnderstanding | None = None, data_prompt: str = "") -> dict:
    """Ein Modellierungsversuch: Extraktion, Checks, Prüfung, Korrekturrunden, beste Runde.

    Mit `spec` wird die Extraktion übersprungen – Einstieg ab der Prüfung (siehe continue_pipeline).
    """
    if spec is None:
        log(f"[{label}] Extrahiere Modell ...")
        spec = text_to_model(session, task + hint, understanding, data_prompt)
    else:
        log(f"[{label}] Übernehme gespeicherte Spezifikation ({len(spec.constraints)} Restriktionen) ...")
    issues = check_spec(spec, task, understanding)
    log(f"[{label}] Prüfe Annahmen ...")
    review = review_assumptions(session, task, spec)

    candidates = [(spec, issues, review)]
    revisions = 0
    while needs_work(review, issues) and revisions < max_revisions:
        revisions += 1
        log(f"[{label}] Probleme gefunden – Korrekturrunde {revisions}/{max_revisions} ...")
        spec = revise_model(session, task, spec, review, issues)
        issues = check_spec(spec, task, understanding)
        review = review_assumptions(session, task, spec)
        candidates.append((spec, issues, review))

    selected = select_best(candidates)
    if selected != len(candidates) - 1:
        log(f"[{label}] Spätere Korrekturrunden haben das Modell verschlechtert – verwende Stand aus Runde {selected}.")
    spec, issues, review = candidates[selected]
    return {"spec": spec, "issues": issues, "review": review, "revisions": revisions, "selected": selected}


def _unresolved(attempt: dict) -> List[str]:
    """Probleme, die nach allen Korrekturrunden einen Neustart rechtfertigen."""
    problems = []
    if not attempt["review"].objective_check.matches:
        oc = attempt["review"].objective_check
        problems.append(f"Zielfunktion passt nicht zur Frage (gefragt: {oc.asked_quantity}; modelliert: {oc.modeled_quantity})")
    problems += [f"[{i.code}] {i.message}" for i in attempt["issues"] if i.severity == CheckSeverity.ERROR]
    return problems


INFEASIBLE_HINT = (
    "\n\nHINWEIS: Eine erste, unabhängige Modellierung dieser Aufgabe ergab '{status}'. "
    "Modelliere die Aufgabe daher völlig neu und prüfe dabei besonders, ob eine Bedingung zu streng ausgelegt "
    "wurde (z. B. eine Gleichung, wo der Text einen Spielraum zulässt, oder eine Grenze, die der Text nicht "
    "verlangt). Ist die Aufgabe wirklich unlösbar, modelliere sie trotzdem getreu."
)


def _recheck_infeasible(session, task: str, model: str, result: PipelineResult,
                        max_revisions: int, solver_retries: int) -> PipelineResult:
    """'unlösbar' nicht ungeprüft glauben: unabhängig neu modellieren und erneut lösen."""
    status = result.solver_run.status.value
    if INFEASIBLE_RECHECK < 1:
        result.reliable = False
        result.reliability_issues.append(f"Solver meldet '{status}' – ohne Gegenprobe (INFEASIBLE_RECHECK=0)")
        return result

    log(f"[{model}] Solver meldet '{status}' – Gegenprobe mit unabhängiger Modellierung ...")
    zweiter = _model_attempt(session, task, max_revisions, f"{model} Gegenprobe",
                             hint=INFEASIBLE_HINT.format(status=status))
    if not zweiter["review"].objective_check.matches:
        log("  Gegenprobe verworfen: Zielfunktion der neuen Modellierung passt nicht zur Frage.")
        result.reliable = False
        result.reliability_issues.append(f"Solver meldet '{status}'; die Gegenprobe war unbrauchbar")
        return result

    spec = zweiter["spec"]
    log(f"[{model}] Gegenprobe: Solver-Code erzeugen und starten ({spec.target_stack.value}) ...")
    code, run = solve(session, task, spec, retries=solver_retries, log=log)

    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.")
        result.specification, result.review, result.checks = spec, zweiter["review"], check_spec(spec, task)
        result.solver_code, result.solver_run = code, run
        result.restarts += 1
        return result

    if run.status.value == status:
        log(f"  Gegenprobe bestätigt '{status}' mit einer unabhängigen Modellierung.")
        result.infeasible_confirmed = True
        return result

    log(f"  Gegenprobe nicht verwertbar (Status '{run.status.value}'), erste Modellierung bleibt.")
    result.reliable = False
    result.reliability_issues.append(f"Solver meldet '{status}'; die Gegenprobe endete mit '{run.status.value}'")
    return result


def _solver_stage(session, task: str, result: PipelineResult, model: str,
                  max_revisions: int, solver_retries: int, data_files=(), data_prompt: str = "") -> PipelineResult:
    """Solver-Code erzeugen, ausführen, prüfen – gemeinsam für run_pipeline und continue_pipeline."""
    spec = result.specification
    if not result.review.objective_check.matches:
        # Der Solver würde ein anderes als das gefragte Problem lösen – ein "optimal" wäre irreführend
        log(f"[{model}] Solver wird nicht gestartet, weil die Zielfunktion nicht zur Frage passt.")
        return result

    log(f"[{model}] Erzeuge und starte Solver-Code ({spec.target_stack.value}) ...")
    result.solver_code, result.solver_run = solve(session, task, spec, retries=solver_retries, log=log,
                                                  understanding=result.understanding,
                                                  data_files=data_files, data_prompt=data_prompt)
    run = result.solver_run
    log(f"Solver: {run.status.value}, Zielwert: {run.objective_value}" + (f"\n  {run.error}" if run.error else ""))
    if run.feasibility:
        result.reliable = False
        result.reliability_issues += [f"Zulässigkeitsprüfung: {p}" for p in run.feasibility]
        log("ERGEBNIS UNZUVERLÄSSIG – die Lösung hält das Modell nicht ein:")
        for problem in run.feasibility:
            log(f"  {problem}")
    elif run.constraints:
        log(f"  Zulässigkeit geprüft: {len(run.constraints)} Restriktionen, {len(run.variables)} Variablen – alles eingehalten.")
    if run.undeclared_data:
        result.reliable = False
        result.reliability_issues += [f"Datenherkunft: {p}" for p in run.undeclared_data]
        log("ERGEBNIS UNZUVERLÄSSIG – selbst erzeugte Eingabedaten sind nicht ausgewiesen.")
    if result.solver_code.data_assumptions:
        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:
    result.usage = session.usage
    cost = f", Kosten ${session.usage.cost_usd:.4f}" if session.usage.cost_usd is not None else ""
    providers = f", Anbieter: {session.usage.providers}" if session.usage.providers else ""
    log(f"LLM-Nutzung: {session.usage.calls} Aufrufe, {session.usage.prompt_tokens} Input-/"
        f"{session.usage.completion_tokens} Output-Tokens (davon {session.usage.reasoning_tokens} Reasoning){cost}{providers}")
    for step, data in session.usage.by_step().items():
        log(f"  {step:18} {data['calls']}x  {data['seconds']:6.1f} s  {data['completion_tokens']:6} Output-Tokens "
            f"(davon {data['reasoning_tokens']} Reasoning)")


def run_pipeline(
    task: str, model: str = DEFAULT_MODEL, max_revisions: int = MAX_REVISIONS,
    solve_model: bool = True, solver_retries: int = SOLVER_RETRIES, max_restarts: int = MAX_RESTARTS,
    session: LLMSession | None = None, data_files=(),
) -> PipelineResult:
    session = session or LLMSession(model)
    profile = data_profile.profile_files(data_files)
    data_prompt = data_profile.to_prompt(profile)
    for p in profile:
        log(f"Datei {p.name}: " + (p.error if p.error else f"{p.rows} Zeilen, {len(p.columns)} Spalten"))
    for step, step_model in session.overrides().items():
        log(f"[{model}] Stufe '{step}' läuft mit abweichendem Modell: {step_model}")

    understanding, verstehen_issues = _understand(session, task, model, data_prompt)
    if not understanding.is_optimization_task:
        # Keine Entscheidung verlangt: jedes Modell wäre erfunden, jede Zahl ohne Frage dahinter
        grund = f"Keine Optimierungsaufgabe: {understanding.reason}"
        log(f"ABGELEHNT – {grund} Es wird weder modelliert noch gerechnet.")
        result = PipelineResult(task=task, understanding=understanding, data_files=profile, rejected=True,
                                reliable=False, reliability_issues=[grund], checks=verstehen_issues)
        _log_usage(session, result)
        return result

    attempts = [_model_attempt(session, task, max_revisions, model, understanding=understanding,
                               data_prompt=data_prompt)]
    # Beobachtet: das LLM modelliert gelegentlich ein Fremdproblem; Korrekturrunden reparieren das nicht,
    # ein frischer Neustart der Extraktion schon eher.
    while _unresolved(attempts[-1]) and len(attempts) <= max_restarts:
        log(f"[{model}] Modell nach allen Korrekturrunden fehlerhaft – Neustart {len(attempts)}/{max_restarts}: "
            + "; ".join(_unresolved(attempts[-1]))[:300])
        attempts.append(_model_attempt(session, task, max_revisions, f"{model} Neustart {len(attempts)}",
                                       understanding=understanding, data_prompt=data_prompt))

    best = attempts[select_best([(a["spec"], a["issues"], a["review"]) for a in attempts])]
    spec, issues, review = best["spec"], best["issues"] + verstehen_issues, best["review"]
    problems = _unresolved(best)

    for issue in issues:
        log(f"  {issue.severity.value.upper()} [{issue.code}] {issue.message}")
    if problems:
        log("ERGEBNIS UNZUVERLÄSSIG: " + "; ".join(problems))
    elif review.needs_revision:
        log("WARNUNG: Die Prüfung hat noch Anmerkungen zu Annahmen.")
    else:
        log("Prüfung ohne Beanstandung.")

    result = PipelineResult(
        task=task, understanding=understanding, data_files=profile, specification=spec, review=review, revisions=best["revisions"],
        selected_revision=best["selected"], restarts=len(attempts) - 1, checks=issues,
        reliable=not problems, reliability_issues=problems,
    )
    if solve_model:
        result = _solver_stage(session, task, result, model, max_revisions, solver_retries,
                               data_files=data_files, data_prompt=data_prompt)

    _log_usage(session, result)
    return result


STAGES = ("pruefung", "solver")


def continue_pipeline(
    result_json: str, stage: str, model: str = DEFAULT_MODEL, max_revisions: int = MAX_REVISIONS,
    solver_retries: int = SOLVER_RETRIES, session: LLMSession | None = None,
) -> PipelineResult:
    """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
    Prüfbericht und erzeugt nur den Solver-Code neu. Aufgabentext kommt aus dem gespeicherten Ergebnis.
    """
    if stage not in STAGES:
        raise ValueError(f"Unbekannte Stufe '{stage}' (erlaubt: {', '.join(STAGES)})")
    stored = PipelineResult.model_validate_json(Path(result_json).read_text(encoding="utf-8"))
    if stored.rejected or stored.specification is None:
        raise ValueError(f"{result_json} ist eine abgelehnte Aufgabe (keine Optimierungsaufgabe) – nichts fortzusetzen")
    session = session or LLMSession(model)
    task = stored.task

    log(f"Setze {result_json} ab Stufe '{stage}' fort – Extraktion entfällt.")
    if stage == "solver":
        result = stored.model_copy(update={
            "checks": check_spec(stored.specification, task),  # deterministisch, kostet nichts
            "solver_code": None, "solver_run": None, "usage": None,
            "reliable": True, "reliability_issues": [],
        })
        problems = _unresolved({"review": stored.review, "issues": result.checks})
        result.reliable, result.reliability_issues = not problems, problems
    else:
        attempt = _model_attempt(session, task, max_revisions, model, spec=stored.specification)
        problems = _unresolved(attempt)
        result = PipelineResult(
            task=task, specification=attempt["spec"], review=attempt["review"], revisions=attempt["revisions"],
            selected_revision=attempt["selected"], checks=attempt["issues"],
            reliable=not problems, reliability_issues=problems,
        )
    for issue in result.checks:
        log(f"  {issue.severity.value.upper()} [{issue.code}] {issue.message}")
    if result.reliability_issues:
        log("ERGEBNIS UNZUVERLÄSSIG: " + "; ".join(result.reliability_issues))

    result = _solver_stage(session, task, result, model, max_revisions, solver_retries)
    _log_usage(session, result)
    return result


def resolve_result(result_json: str, data_file: str | None = None, solver_timeout: int = 60) -> PipelineResult:
    """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.
    """
    result = PipelineResult.model_validate_json(Path(result_json).read_text(encoding="utf-8"))
    if not result.solver_code:
        raise ValueError(f"{result_json} enthält keinen Solver-Code (Lauf mit --no-solve?)")
    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 ...")
    run = run_solver_code(result.solver_code.code, timeout=solver_timeout, data=data)
    run.feasibility = check_feasibility(run, result.specification)
    run.warnings += check_sensitivity(run, result.specification.objective.sense.value)
    run.warnings += check_multistart(run, result.specification.objective.sense.value)

    result.solver_run = run
    result.reliable = not run.feasibility
    result.reliability_issues = [f"Zulässigkeitsprüfung: {p}" for p in run.feasibility]
    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 ""))
    for problem in run.feasibility:
        log(f"  {problem}")
    return result


def read_task(task_file: str | None) -> str:
    if task_file:
        task = Path(task_file).read_text(encoding="utf-8")
    elif 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 __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",
    )
    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")
    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()
    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.resolve:
        result = resolve_result(args.resolve, args.data)
    elif args.von:
        result = continue_pipeline(args.von[1], args.von[0], model=args.model, max_revisions=args.max_revisions)
    else:
        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 (),
        )
    print(result.model_dump_json(indent=2))
    if args.report:
        write_report(result, args.report)
        log(f"Report geschrieben: {args.report}")
    if args.code and result.solver_code:
        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