"""Deterministische Plausibilitätsprüfungen einer Modellspezifikation (ohne LLM). error -> löst in run_pipeline eine Korrekturrunde aus warning -> wird nur berichtet (Heuristiken mit möglichen Fehlalarmen) """ import re from typing import Iterable, List, Set from core.model_schema import ( CheckIssue, CheckSeverity, OptimizationModelSpecification, ProblemClass, TargetStack, VariableType, ) # Zulässige Ziel-Stacks je Problemklasse (muss zu den Routing-Regeln im Prompt passen) ALLOWED_STACKS = { ProblemClass.LP: {TargetStack.SCIPY, TargetStack.HIGHS}, ProblemClass.MILP: {TargetStack.HIGHS}, ProblemClass.CP: {TargetStack.OR_TOOLS_CPSAT}, ProblemClass.QP: {TargetStack.CVXPY}, ProblemClass.SOCP: {TargetStack.CVXPY}, ProblemClass.MIQP: {TargetStack.SCIP}, ProblemClass.NLP: {TargetStack.IPOPT, TargetStack.SCIP}, # Kürzeste Wege: spezialisierter Graphalgorithmus, alternativ als Flussmodell (MILP) ProblemClass.SHORTEST_PATH: {TargetStack.NETWORKX, TargetStack.HIGHS}, } LINEAR_CLASSES = {ProblemClass.LP, ProblemClass.MILP, ProblemClass.CP, ProblemClass.SHORTEST_PATH} # Quadrat, Transponierte (w^T \Sigma w) oder Norm NONLINEAR_PATTERN = re.compile(r"\^\s*\{?\s*(2|T)\b|\\\||\\lVert|\|\|") # Grenzen von Summen/Produkten/Integralen: \sum_{i=0}^{2} ist keine Potenz, sondern ein Laufbereich LIMIT_PATTERN = re.compile(r"\\(sum|prod|int|iint|oint|bigcup|bigcap|max|min|argmax|argmin)" r"(\s*_\s*(\{[^{}]*\}|\\?\w))?(\s*\^\s*(\{[^{}]*\}|\\?\w))?") GREEK = { "alpha", "beta", "gamma", "delta", "epsilon", "varepsilon", "zeta", "eta", "theta", "vartheta", "iota", "kappa", "lambda", "mu", "nu", "xi", "pi", "rho", "sigma", "tau", "upsilon", "phi", "varphi", "chi", "psi", "omega", "Gamma", "Delta", "Theta", "Lambda", "Xi", "Pi", "Sigma", "Upsilon", "Phi", "Psi", "Omega", } # Wörter, die in Formeln vorkommen dürfen, ohne deklariert zu sein IGNORED_TOKENS = {"sum", "min", "max", "abs", "log", "exp", "sqrt", "for", "all", "if", "then", "and", "or", "not"} # Einheiten je Parameter (ParameterDefinition.unit) – nur zwei Klassen werden gegeneinander geprüft: # Geldbeträge und dimensionslose Größen. Beobachtet: Risiko als \sum w_i^2 \sigma_i^2 (dimensionslos) # neben Kosten in Euro – das Risiko war um den Faktor Kapital zu klein und damit wirkungslos. CURRENCY_UNIT = re.compile(r"€|\bEUR\b|Euro|\$|\bUSD\b|Geldeinheit|Kosteneinheit|\bKE\b|\bGE\b", re.IGNORECASE) RELATIVE_UNIT = re.compile(r"^\s*(-|–|1)?\s*$|%|Prozent|Anteil|Quote|Faktor|Verhältnis|dimensionslos|" r"dimensionless|ohne Einheit|einheitenlos", re.IGNORECASE) # Fachbegriffe, die ein Leser ohne OR-Ausbildung nicht mitbringt. Tauchen sie dort auf, wo er sie sieht # (Restriktionsnamen, Tabellenspalten, Antwortsatz), müssen sie in executive_summary erklärt sein. # Je Begriff mehrere Wortstämme, damit Beugungen ("Bogen"/"Bögen") als derselbe Begriff zählen. JARGON = { "Bogen": ("bogen", "bög"), "Knoten": ("knoten",), "Kante": ("kante",), "Fluss": ("fluss", "flüss"), "Quelle/Senke": ("senke",), "Relaxation": ("relaxation", "relaxier"), "Schlupf": ("schlupf", "slack"), "Kardinalität": ("kardinal",), "Epigraph": ("epigraph",), "Schattenpreis": ("schattenpreis", "dualwert", "dualvariable"), "Konvexität": ("konvex",), # nicht "markier": „die mit ja markierten Zeilen“ ist Alltagssprache (Fehlalarm in 20260917-160618) "Markierung": ("marke", "markierung"), "Inzidenz": ("inzidenz",), "Big-M": ("big-m", "big m"), "Subtour": ("subtour",), "Lagrange": ("lagrange",), } def unexplained_jargon(sichtbar: str, erklaerung: str) -> List[str]: """Fachbegriffe, die im sichtbaren Text vorkommen, in der Erklärung aber nicht.""" sichtbar, erklaerung = sichtbar.lower(), erklaerung.lower() return sorted(name for name, staemme in JARGON.items() if any(s in sichtbar for s in staemme) and not any(s in erklaerung for s in staemme)) def check_jargon(spec: OptimizationModelSpecification) -> List[CheckIssue]: """Fachbegriffe in Namen des Modells, die die Zusammenfassung nicht einführt (siehe Block ERKLÄREN).""" sichtbar = " ".join([*(c.identifier for c in spec.constraints), *(v.name for v in spec.variables), *spec.indices]) offen = unexplained_jargon(sichtbar, spec.executive_summary) if not offen: return [] return [CheckIssue( severity=CheckSeverity.WARNING, code="UNEXPLAINED_JARGON", message=f"Fachbegriffe im Modell, die executive_summary nicht erklärt: {', '.join(offen)}. " f"Der Leser trifft sie in Restriktions- und Variablennamen, ohne zu wissen, was sie bedeuten.", )] # --- Verstehensstufe: Teilfragen und Zahlen des Aufgabentexts (ohne LLM) --- # Zahl im Fließtext: 1.234,56 / 0,05 / 12 / 3.8 – ohne Zahlen, die Teil eines Wortes sind (A02, x1) NUMBER_PATTERN = re.compile(r"(?]*\**\s*(?:\(?([a-h])\)|(?:Teil|Aufgabe|Frage)\s+(\d+)|([1-9])\.)\s*[:.)]?\s+\S", re.MULTILINE) def normalize_number(roh: str) -> str: """'1.234,56' -> '1234.56', '3.8' -> '3.8', '1.234' -> '1234'. Der Punkt ist mehrdeutig: Tausendertrenner (deutsch) oder Dezimalpunkt (englisch, Nummerierung). Entschieden wird am Komma (dann ist der Punkt Tausendertrenner) und sonst an der Gruppenlänge: genau drei Ziffern hinter dem letzten Punkt heißt Tausender, alles andere Dezimalpunkt. """ if "," in roh: roh = roh.replace(".", "").replace(",", ".") elif "." in roh: teile = roh.split(".") if all(len(t) == 3 for t in teile[1:]) and len(teile[0]) <= 3: roh = "".join(teile) # 1.234 bzw. 12.345.678 wert = float(roh) return str(int(wert)) if wert.is_integer() else f"{wert:g}" # große Zahlen exakt, nicht 1.23457e+06 def numbers_in(text: str) -> Set[str]: """Zahlen eines Textes in einheitlicher Schreibweise.""" gefunden = set() for roh in NUMBER_PATTERN.findall(text or ""): try: gefunden.add(normalize_number(roh)) except ValueError: continue return gefunden def question_ids_in(text: str) -> Set[str]: """Kennungen der Teilfragen, die im Aufgabentext erkennbar sind.""" return {(a or b or c).lower() for a, b, c in QUESTION_PATTERN.findall(text or "")} def _covers(genannt: str, erkannt: str) -> bool: """'2a', '2.1', 'a1' decken die Teilfrage '2' bzw. 'a' ab – '21' deckt '2' nicht ab. Beobachtet an „Portfolio Nr. 2": Teil 2 wurde in 2a und 2b zerlegt, gemeldet wurde „es fehlt: 2". """ if genannt == erkannt: return True if not genannt.startswith(erkannt): return False naechstes = genannt[len(erkannt)] return naechstes.isdigit() != erkannt[-1].isdigit() or not naechstes.isalnum() def check_understanding(understanding, task: str) -> List[CheckIssue]: """Prüft das Verständnis gegen den Aufgabentext: sind alle Zahlen und alle Teilfragen erfasst? Zahlen gelten als erfasst, wenn sie irgendwo im Verständnis auftauchen (Wert, Zitat, Bedeutung) oder ausdrücklich als irrelevant erklärt sind – ein Zitat der Tabellenzeile deckt so eine ganze Datenzeile ab, ohne dass jede Zahl einzeln aufgeführt werden muss. """ issues = [] erfasst = numbers_in(understanding.model_dump_json()) fehlend = sorted(numbers_in(task) - erfasst, key=float) if fehlend: issues.append(CheckIssue( severity=CheckSeverity.WARNING, code="UNCAPTURED_NUMBERS", message=f"Zahlen aus dem Aufgabentext, die weder als Fakt noch als irrelevant erfasst sind: " f"{', '.join(fehlend[:12])}{' …' if len(fehlend) > 12 else ''}. " f"Sie fehlen damit in der Spezifikation und können nicht geprüft werden.", )) erkannt = question_ids_in(task) genannt = {q.id.lower().removeprefix("teil ").removeprefix("aufgabe ").strip(" ().") for q in understanding.sub_questions} offen = sorted(e for e in erkannt if not any(_covers(g, e) for g in genannt)) if offen: issues.append(CheckIssue( severity=CheckSeverity.WARNING, code="MISSING_SUB_QUESTIONS", message=f"Der Aufgabentext enthält die Teilfragen {', '.join(sorted(erkannt))}, " f"das Verständnis nennt {', '.join(sorted(genannt)) or 'keine'} – es fehlen: {', '.join(offen)}.", )) if not understanding.sub_questions: issues.append(CheckIssue( severity=CheckSeverity.WARNING, code="NO_SUB_QUESTIONS", message="Keine einzige Teilfrage erfasst – auch eine Aufgabe mit nur einer Frage braucht einen Eintrag.", )) return issues def _strip_scripts(text: str) -> str: """Entfernt Indizes/Exponenten: x_{i,j} -> x, C^{prod}_{p} -> C, x_1 -> x.""" text = text.replace(r"\_", "_") # Schriftbefehle samt Inhalt: \mathbb{Z} ist die Menge der ganzen Zahlen, kein Symbol namens Z text = re.sub(r"\\(text|mathrm|operatorname|mbox|mathbb|mathcal|mathbf|mathsf|boldsymbol)\s*\{[^{}]*\}", " ", text) previous = None while previous != text: # verschachtelte Klammern von innen nach außen previous = text text = re.sub(r"[_^]\s*\{[^{}]*\}", "", text) # einzelnes Zeichen als Index, aber nicht innerhalb von Namen wie min_staff text = re.sub(r"[_^]\s*\\?[A-Za-z0-9](?![A-Za-z0-9])", "", text) return text def _symbols(text: str) -> Set[str]: """Basis-Symbole einer LaTeX-Formel (Greek als Symbol, andere Befehle ignoriert).""" text = _strip_scripts(text) tokens = set() for match in re.finditer(r"\\([A-Za-z]+)|([A-Za-z][A-Za-z0-9_]*)", text): command, name = match.groups() if command: if command in GREEK: tokens.add("\\" + command) else: tokens.add(name.rstrip("_")) return tokens def _declared_symbols(spec: OptimizationModelSpecification) -> Set[str]: declared: Set[str] = set() for text in _texts_declaring(spec): declared |= _symbols(text) return declared def _texts_declaring(spec: OptimizationModelSpecification) -> Iterable[str]: yield from (v.name for v in spec.variables) yield from (p.symbol for p in spec.parameters) for index in spec.indices: # Formel-Begrenzer entfernen ("Tage: \(d \in D\)"), dann erläuternde Klammer am Ende # ("i in I (Mitarbeiter)"); früher wurde am ersten "(" geschnitten – das traf auch "\(" text = re.sub(r"\\[()\[\]]|\$", " ", index) yield re.sub(r"\s\([^()]*\)\s*$", "", text) def _objective_without_label(formula: str) -> str: """'Z = 12 x': der Name links vom Gleichheitszeichen ist eine Beschriftung, kein Symbol des Modells. Nur für die Zielfunktion – bei Restriktionen ist die linke Seite Teil des Modells. """ # Nur ein Gleichheitszeichen auf oberster Klammerebene zählt: \sum_{i=0}^{2} enthält ebenfalls eines, # und daran zerschnitten blieb von der Formel „0}^{2}\sum…“ übrig (führte zu NONLINEAR_TERM). tiefe = 0 for pos, zeichen in enumerate(formula): if zeichen in "{([": tiefe += 1 elif zeichen in "})]": tiefe -= 1 elif zeichen == "=" and tiefe == 0: left, right = formula[:pos], formula[pos + 1:] return right if len(_symbols(left) - IGNORED_TOKENS) <= 1 else formula return formula def _formulas(spec: OptimizationModelSpecification) -> Iterable[tuple]: yield "Zielfunktion", _objective_without_label(spec.objective.latex_formula) for c in spec.constraints: yield c.identifier, c.latex_formula def check_routing(spec: OptimizationModelSpecification) -> List[CheckIssue]: issues = [] var_types = {v.var_type for v in spec.variables} has_discrete = bool(var_types & {VariableType.INTEGER, VariableType.BINARY}) has_continuous = VariableType.CONTINUOUS in var_types allowed = ALLOWED_STACKS[spec.problem_class] if spec.target_stack not in allowed: issues.append(CheckIssue( severity=CheckSeverity.ERROR, code="STACK_MISMATCH", message=f"Problemklasse '{spec.problem_class.value}' passt nicht zu Stack '{spec.target_stack.value}' " f"(erlaubt: {', '.join(s.value for s in allowed)}).", )) if has_discrete and spec.problem_class == ProblemClass.LP: issues.append(CheckIssue( severity=CheckSeverity.ERROR, code="LP_WITH_DISCRETE_VARS", message="Problemklasse LP, aber ganzzahlige/binäre Variablen vorhanden (→ MILP oder CP).", )) if has_discrete and spec.target_stack == TargetStack.CVXPY: issues.append(CheckIssue( severity=CheckSeverity.ERROR, code="CVXPY_WITH_DISCRETE_VARS", message="CVXPY ist hier nur für kontinuierliche QP/SOCP vorgesehen; quadratisch + ganzzahlig/binär " "ist MIQP (Stack SCIP).", )) if spec.problem_class == ProblemClass.MIQP and not has_discrete: issues.append(CheckIssue( severity=CheckSeverity.WARNING, code="MIQP_WITHOUT_DISCRETE_VARS", message="Problemklasse MIQP, aber keine ganzzahligen/binären Variablen (→ eigentlich QP/SOCP mit CVXPY).", )) if has_discrete and spec.target_stack == TargetStack.IPOPT: issues.append(CheckIssue( severity=CheckSeverity.ERROR, code="IPOPT_WITH_DISCRETE_VARS", message="Ipopt löst nur stetige Probleme; mit ganzzahligen/binären Variablen ist SCIP zuständig.", )) if has_discrete and spec.target_stack == TargetStack.SCIPY: issues.append(CheckIssue( severity=CheckSeverity.ERROR, code="SCIPY_WITH_DISCRETE_VARS", message="scipy.optimize.linprog unterstützt keine ganzzahligen/binären Variablen.", )) if has_continuous and spec.target_stack == TargetStack.OR_TOOLS_CPSAT: issues.append(CheckIssue( severity=CheckSeverity.ERROR, code="CPSAT_WITH_CONTINUOUS_VARS", message="CP-SAT unterstützt nur ganzzahlige/binäre Variablen, es gibt aber kontinuierliche.", )) if spec.problem_class == ProblemClass.MILP and not has_discrete: issues.append(CheckIssue( severity=CheckSeverity.WARNING, code="MILP_WITHOUT_DISCRETE_VARS", message="Problemklasse MILP, aber keine ganzzahligen/binären Variablen (→ eigentlich LP).", )) return issues def check_nonlinearity(spec: OptimizationModelSpecification) -> List[CheckIssue]: if spec.problem_class not in LINEAR_CLASSES: return [] return [ CheckIssue( severity=CheckSeverity.WARNING, code="NONLINEAR_TERM", message=f"'{name}' enthält einen möglicherweise nichtlinearen Term, " f"Problemklasse ist aber '{spec.problem_class.value}': {formula}", ) for name, formula in _formulas(spec) if NONLINEAR_PATTERN.search(LIMIT_PATTERN.sub(" ", formula)) ] def _is_implicit_product(symbol: str, declared: Set[str]) -> bool: """'cx' in \\sin(cx) ist das Produkt c·x, kein eigener Name – wenn c und x deklariert sind.""" return len(symbol) > 1 and all(zeichen in declared for zeichen in symbol) def check_undeclared_symbols(spec: OptimizationModelSpecification) -> List[CheckIssue]: declared = _declared_symbols(spec) undeclared: dict = {} for name, formula in _formulas(spec): for symbol in _symbols(formula) - declared - IGNORED_TOKENS: if _is_implicit_product(symbol, declared): continue undeclared.setdefault(symbol, []).append(name) return [ CheckIssue( severity=CheckSeverity.WARNING, code="UNDECLARED_SYMBOL", message=f"Symbol '{symbol}' wird in {', '.join(places)} verwendet, ist aber weder Variable, Parameter noch Index.", ) for symbol, places in sorted(undeclared.items()) ] def _unit_class(unit: str) -> str: """'currency', 'relative' oder 'unbekannt'; nur die ersten beiden werden gegeneinander geprüft.""" text = (unit or "").strip() if CURRENCY_UNIT.search(text): return "currency" if RELATIVE_UNIT.search(text): return "relative" return "unbekannt" def _additive_terms(formula: str) -> List[str]: """Summanden einer Formel: Trennung an + und - auf oberster Klammerebene. Indizes stehen in Klammern (\\sum_{i \\in I}) und werden dadurch nicht zerschnitten. """ body = re.sub(r"\\operatorname\*?\s*\{[^{}]*\}|\\(?:min|max|left|right|quad|qquad)\b|\\[,;!]", " ", formula) if "=" in body: # "\min z = \sum ..." -> nur die rechte Seite ist die Zielfunktion body = body.rsplit("=", 1)[1] terms, current, depth = [], "", 0 for char in body: if char in "{([": depth += 1 elif char in "})]": depth -= 1 if depth == 0 and char in "+-" and current.strip(): terms.append(current) current = "" continue current += char return [t for t in (*terms, current) if t.strip()] def _term_unit_class(term: str, parameter_units: dict) -> str: """Einheit eines Summanden: ein Geldbetrag mal dimensionslosem Faktor bleibt ein Geldbetrag.""" classes = {c for symbol in _symbols(term) for c in parameter_units.get(symbol, ())} if "currency" in classes: return "currency" if classes == {"relative"}: return "relative" return "unbekannt" def check_objective_units(spec: OptimizationModelSpecification) -> List[CheckIssue]: """Mischt die Zielfunktion Geldbeträge mit dimensionslosen Größen? (aus ParameterDefinition.unit)""" parameter_units: dict = {} for parameter in spec.parameters: for symbol in _symbols(parameter.symbol): parameter_units.setdefault(symbol, set()).add(_unit_class(parameter.unit)) by_class: dict = {} for term in _additive_terms(spec.objective.latex_formula): unit_class = _term_unit_class(term, parameter_units) if unit_class != "unbekannt": by_class.setdefault(unit_class, []).append(term.strip()) if "currency" not in by_class or "relative" not in by_class: return [] return [CheckIssue( severity=CheckSeverity.ERROR, code="MIXED_OBJECTIVE_UNITS", message=f"Die Zielfunktion mischt Geldbeträge ({'; '.join(by_class['currency'])}) mit dimensionslosen " f"Termen ({'; '.join(by_class['relative'])}). Der dimensionslose Term ist gegenüber den " f"Geldbeträgen praktisch wirkungslos – mit dem Kapital bzw. Mengengerüst multiplizieren.", )] # Nennt der Aufgabentext ein Verfahren beim Namen, ist es Teil der Aufgabenstellung – ein gleichwertiges # Modell erfüllt sie dann nicht (beobachtet: „Lösen Sie mit dem Dijkstra-Algorithmus“ wurde als # MILP-Flussmodell gelöst, mathematisch richtig, als Aufgabe aber verfehlt). REQUESTED_METHOD = { "dijkstra": {TargetStack.NETWORKX}, "bellman-ford": {TargetStack.NETWORKX, TargetStack.HIGHS}, "kürzeste-wege-problem": {TargetStack.NETWORKX, TargetStack.HIGHS}, "simplex": {TargetStack.SCIPY, TargetStack.HIGHS}, "branch-and-bound": {TargetStack.HIGHS, TargetStack.SCIP}, "branch-and-cut": {TargetStack.HIGHS, TargetStack.SCIP}, "constraint programming": {TargetStack.OR_TOOLS_CPSAT}, } def check_requested_method(spec: OptimizationModelSpecification, task: str) -> List[CheckIssue]: """Ein im Aufgabentext ausdrücklich verlangtes Verfahren muss auch verwendet werden.""" text = (task or "").lower() issues = [] for name, stacks in REQUESTED_METHOD.items(): if name in text and spec.target_stack not in stacks: issues.append(CheckIssue( severity=CheckSeverity.ERROR, code="REQUESTED_METHOD_IGNORED", message=f"Die Aufgabe verlangt ausdrücklich '{name}', das Modell nutzt aber " f"'{spec.target_stack.value}' (passend wäre: {', '.join(sorted(s.value for s in stacks))}). " f"Ein gleichwertiges Modell erfüllt die Aufgabenstellung nicht.", )) return issues def check_data_transfer(spec: OptimizationModelSpecification, understanding) -> List[CheckIssue]: """Sind die erfassten Zahlen auch im Modell angekommen? Beobachtet: Die Verstehensstufe erfasste die vollständige Kostenmatrix, die Spezifikation führte `extracted_values: None` – die Werte betraten die Pipeline erst im Solver-Code, also hinter allen Prüfungen. Ohne Werte in der Spezifikation kann nichts Numerisches geprüft werden. """ if understanding is None or not understanding.facts: return [] # Nur die erklärten Werte, nicht die Zitate: eine zitierte Tabellenzeile enthält auch # Zeilenbeschriftungen, die keine Parameterwerte sind. erfasst = numbers_in(" ".join(f.value for f in understanding.facts)) uebernommen = numbers_in(" ".join(p.extracted_values or "" for p in spec.parameters)) if not erfasst or (erfasst & uebernommen): return [] # bewusst tolerant: gemeldet wird nur die Spezifikation ganz ohne Zahlen return [CheckIssue( severity=CheckSeverity.WARNING, code="UNUSED_FACTS", message=f"Die Verstehensstufe hat {len(erfasst)} Zahlenwerte erfasst " f"({', '.join(sorted(erfasst, key=float)[:8])}), in extracted_values steht keiner davon. " f"Ohne Werte ist die Spezifikation nicht nachrechenbar – die Zahlen erreichen die Pipeline " f"sonst erst im Solver-Code, hinter allen Prüfungen.", )] def check_missing_data(spec: OptimizationModelSpecification) -> List[CheckIssue]: missing = [p.symbol for p in spec.parameters if not (p.extracted_values or "").strip()] if not missing: return [] return [CheckIssue( severity=CheckSeverity.WARNING, code="MISSING_DATA", message=f"Keine Werte im Aufgabentext für: {', '.join(missing)}. Lösung nur mit angenommenen Daten möglich.", )] def check_structure(spec: OptimizationModelSpecification) -> List[CheckIssue]: issues = [] if not spec.variables: issues.append(CheckIssue(severity=CheckSeverity.ERROR, code="NO_VARIABLES", message="Keine Entscheidungsvariablen definiert.")) if not spec.constraints: issues.append(CheckIssue(severity=CheckSeverity.ERROR, code="NO_CONSTRAINTS", message="Keine Restriktionen definiert.")) return issues def check_spec(spec: OptimizationModelSpecification, task: str = "", understanding=None) -> List[CheckIssue]: """Alle deterministischen Prüfungen; `task` und `understanding` nur für Prüfungen, die sie brauchen.""" return ( check_structure(spec) + check_routing(spec) + check_nonlinearity(spec) + check_undeclared_symbols(spec) + check_objective_units(spec) + check_jargon(spec) + check_requested_method(spec, task) + check_data_transfer(spec, understanding) + check_missing_data(spec) ) def has_errors(issues: List[CheckIssue]) -> bool: return any(i.severity == CheckSeverity.ERROR for i in issues)