text_to_solver_program/model_schema.py
dschlueter b74445b8d6 Erste öffentliche Fassung: Math Model Generator & Solver
Übersetzt natürlichsprachige Optimierungsaufgaben (Deutsch) per LLM in
eine geprüfte Modellspezifikation, erzeugt Solver-Code und führt ihn in
einer bubblewrap-Sandbox aus. Deterministische Prüfungen ohne LLM,
Report als Markdown/HTML, Benchmark mit 19 verifizierten Fällen.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-17 11:42:58 +02:00

439 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from enum import Enum
from typing import Dict, List, Literal, Optional, Union
from pydantic import BaseModel, Field
class ProblemClass(str, Enum):
LP = "Linear Programming (LP)"
MILP = "Mixed-Integer Linear Programming (MILP)"
CP = "Constraint Programming (CP)"
QP = "Quadratic Programming (QP)"
SOCP = "Second-Order Cone Programming (SOCP)"
MIQP = "Mixed-Integer Quadratic Programming (MIQP)"
NLP = "Nonlinear Programming (NLP)"
SHORTEST_PATH = "Kürzeste Wege im Graphen (Shortest Path)"
class TargetStack(str, Enum):
HIGHS = "HiGHS"
OR_TOOLS_CPSAT = "Google OR-Tools (CP-SAT)"
CVXPY = "CVXPY"
SCIPY = "SciPy (scipy.optimize.linprog)"
SCIP = "SCIP (pyscipopt)"
IPOPT = "Ipopt (cyipopt)"
NETWORKX = "NetworkX (Dijkstra)"
class ObjectiveSense(str, Enum):
MINIMIZE = "minimize"
MAXIMIZE = "maximize"
class VariableType(str, Enum):
CONTINUOUS = "continuous"
INTEGER = "integer"
BINARY = "binary"
class VariableDefinition(BaseModel):
name: str = Field(description="Mathematischer Variablenname, z. B. x_{i,j}")
var_type: VariableType
domain: str = Field(description="Definitionsbereich, z. B. >= 0, [0, 1] oder {0, 1}")
description: str = Field(description="Semantische Bedeutung für Entscheider")
class ParameterDefinition(BaseModel):
symbol: str = Field(description="Mathematisches Symbol, z. B. c_j oder \\Sigma")
dimension: str = Field(description="Dimension, z. B. Skalar, Vektor (n,), Matrix (n x n)")
unit: str = Field(
description="Physikalische Einheit des Werts, z. B. '€', '€/t', 'h', 'Stück', '%' oder 'dimensionslos'. "
"Wird maschinell geprüft: Terme der Zielfunktion dürfen Geldbeträge nicht mit "
"dimensionslosen Größen mischen"
)
description: str = Field(description="Bedeutung (z. B. Kosten pro Tonne, Renditen)")
extracted_values: Optional[str] = Field(
default=None, description="Explizite Werte aus der Aufgabe, falls vorhanden"
)
class ConstraintDefinition(BaseModel):
identifier: str = Field(description="Kurzname, z. B. 'Kapazitaetsgrenze_Lager'")
latex_formula: str = Field(description="Formel in LaTeX-Syntax, z. B. '\\sum_{j} a_j x_j \\le C'")
description: str = Field(description="Erklärung der Restriktion in Geschäftssprache")
is_hard_constraint: bool = Field(default=True)
class AssumptionDefinition(BaseModel):
unclear_point: str = Field(description="Unklare oder fehlende Stelle in der Aufgabenstellung")
assumption: str = Field(description="Getroffene Annahme, auf der das Modell beruht")
alternative: str = Field(
description="Die zweite plausible Lesart derselben Stelle, ausformuliert, und der Grund, warum das Modell "
"ihr nicht folgt. Nur wenn die Stelle wirklich eindeutig ist: 'keine zweite Lesart'"
)
impact: str = Field(description="Auswirkung der Annahme auf Variablen, Parameter oder Restriktionen")
class ColumnProfile(BaseModel):
"""Eine Spalte einer beigelegten Datei – gerechnet, nicht geschätzt."""
name: str
dtype: str
missing: int = 0
minimum: Optional[float] = None
maximum: Optional[float] = None
mean: Optional[float] = None
samples: List[str] = Field(default_factory=list)
class DataFileProfile(BaseModel):
"""Profil einer beigelegten Datendatei (siehe data_profile.py)."""
name: str
rows: int
columns: List[ColumnProfile] = Field(default_factory=list)
truncated_columns: int = 0
error: Optional[str] = None
class SubQuestion(BaseModel):
"""Eine Teilfrage der Aufgabe. Jede muss am Ende beantwortet werden."""
id: str = Field(description="Kennung wie im Text: 'a', 'b', 'Teil 2', '1'")
quote: str = Field(description="Die Frage wörtlich aus dem Aufgabentext")
kind: Literal["formulieren", "rechnen", "vergleichen", "erklaeren", "zeichnen"] = Field(
description="Was verlangt wird: ein Modell aufstellen, eine Zahl berechnen, Varianten vergleichen, "
"etwas begründen oder eine Skizze/einen Graphen")
needs_solver: bool = Field(description="True, wenn die Antwort eine berechnete Zahl braucht")
class Fact(BaseModel):
"""Ein Zahlenwert aus dem Aufgabentext, mit seiner Fundstelle."""
value: str = Field(description="Der Wert genau wie im Text, z. B. '8', '0,05', '12 %'")
unit: str = Field(description="Einheit, z. B. '€', 'h', 'Stück', '%', 'dimensionslos'")
meaning: str = Field(description="Wofür die Zahl steht, in der Sprache der Aufgabe")
quote: str = Field(description="Die Textstelle, in der die Zahl steht (Zeile oder Tabellenzeile)")
class IgnoredNumber(BaseModel):
"""Eine Zahl im Text, die keine Modelldaten sind – mit Begründung."""
value: str
reason: str = Field(description="Warum das keine Eingabegröße ist, z. B. 'Aufgabennummer', "
"'Teil der Einheit (in 1000 Stunden)', 'Seitenzahl'")
class TaskUnderstanding(BaseModel):
"""Was in der Aufgabe steht – vor jeder Modellierung: Fragen, Zahlen, Machbarkeit."""
is_optimization_task: bool = Field(
description="False, wenn der Text gar keine Optimierungs- oder Entscheidungsaufgabe stellt "
"(dann wird nicht modelliert)")
reason: str = Field(description="Begründung in einem Satz, worum es geht bzw. warum es keine Aufgabe ist")
sub_questions: List[SubQuestion] = Field(
description="Alle Teilfragen in der Reihenfolge des Textes; bei einer einzigen Frage genau ein Eintrag")
facts: List[Fact] = Field(description="Alle Zahlenwerte des Textes, die Eingabegrößen des Modells sind")
ignored_numbers: List[IgnoredNumber] = Field(
default_factory=list, description="Zahlen im Text, die keine Eingabegrößen sind, mit Begründung")
class ObjectiveDefinition(BaseModel):
sense: ObjectiveSense
latex_formula: str = Field(description="Zielfunktion in LaTeX, z. B. '\\min \\sum_i c_i x_i' oder '\\min w^T \\Sigma w'")
description: str = Field(description="Ökonomisches Ziel (z. B. Gesamtkosten senken)")
class OptimizationModelSpecification(BaseModel):
"""Vollständige Zerlegung und Routing-Spezifikation einer Optimierungsaufgabe."""
# 1. Routing & Taxonomie
problem_class: ProblemClass
target_stack: TargetStack
routing_rationale: str = Field(
description="Präzise Begründung, warum dieser Stack für das Problem optimal ist."
)
# 2. Didaktische Zusammenfassung
executive_summary: str = Field(
description="Verständliche Zusammenfassung des Entscheidungsproblems für Manager."
)
# 3. Mathematische Bausteine
indices: List[str] = Field(
description="Mengen und Indizes, z. B. ['i in I (Mitarbeiter)', 't in T (Zeitschritte)']"
)
parameters: List[ParameterDefinition]
variables: List[VariableDefinition]
objective: ObjectiveDefinition
constraints: List[ConstraintDefinition]
# 4. Offene Annahmen
open_assumptions: List[AssumptionDefinition] = Field(
default_factory=list,
description="Annahmen zu unklaren oder fehlenden Angaben; leer, wenn die Aufgabe eindeutig ist",
)
# 5. Technische Schnittstelle
suggested_export_format: str = Field(
description="Zieldateityp: '.lp', '.mps', 'Python (CP-SAT)', oder 'Matrix-Dict'"
)
# --- Deterministische Checks (ohne LLM, siehe model_checks.py) ---
class CheckSeverity(str, Enum):
ERROR = "error" # löst Korrekturrunde aus
WARNING = "warning" # nur Report
class CheckIssue(BaseModel):
severity: CheckSeverity
code: str
message: str
# --- Solver-Stufe (siehe solver_stage.py) ---
class SolverStatus(str, Enum):
OPTIMAL = "optimal"
FEASIBLE = "feasible"
INFEASIBLE = "infeasible"
UNBOUNDED = "unbounded"
ERROR = "error" # Laufzeitfehler, Timeout, fehlendes RESULT_JSON, Import-Verstoß
class SolverCode(BaseModel):
code: str = Field(description="Vollständiges, eigenständig ausführbares Python-Skript ohne Markdown-Codeblock")
data_assumptions: List[str] = Field(
default_factory=list,
description="Alle Datenwerte, die nicht im Aufgabentext stehen und für die Lösung angenommen wurden",
)
class ResultTable(BaseModel):
"""Verständliche Ergebnistabelle, vom Solver-Programm aus der tatsächlichen Lösung berechnet."""
title: str
columns: List[str]
rows: List[List[Union[float, int, str, None]]]
objective_column: Optional[str] = Field(
default=None, description="Spalte mit dem Beitrag je Zeile zur Zielfunktion (im Maßstab von objective_value); Summe muss dem Zielwert entsprechen")
unit: Optional[str] = Field(default=None, description="Einheit der objective_column, z. B. '€' oder '%'")
note: Optional[str] = None
class ConstraintValue(BaseModel):
"""Eine Restriktion, ausgewertet an der gefundenen Lösung (vom Solver-Programm gemeldet)."""
name: str = Field(description="Name der Restriktion, beginnt mit dem identifier aus der Spezifikation")
lhs: float = Field(description="Wert der linken Seite an der Lösung")
sense: Literal["<=", ">=", "=="]
rhs: float
class VariableValue(BaseModel):
"""Eine Variable mit Wert, Typ und Schranken – Grundlage für Ganzzahligkeits- und Schrankenprüfung."""
name: str
value: float
var_type: VariableType = VariableType.CONTINUOUS
lb: Optional[float] = None
ub: Optional[float] = None
class Sensitivity(BaseModel):
"""Wirkung einer Lockerung/Verschärfung einer bindenden Restriktion, durch Neulösen ermittelt."""
name: str = Field(description="Name der Restriktion wie in constraints")
delta: float = Field(description="Verschiebung der rechten Seite, > 0")
objective_plus: Optional[float] = Field(default=None, description="Zielwert bei rechter Seite + delta; None wenn dann unlösbar")
objective_minus: Optional[float] = Field(default=None, description="Zielwert bei rechter Seite - delta")
unit: Optional[str] = Field(default=None, description="Einheit des Zielwerts je Einheit der rechten Seite")
class Answer(BaseModel):
"""Die Antwort auf eine Teilfrage – in der Sprache der Frage, nicht im Modelljargon."""
question_id: str = Field(description="Kennung der Teilfrage wie im Verständnis ('a', 'b', 'Teil 2')")
text: str = Field(description="Antwortsatz, der die Frage beantwortet; nur tatsächlich berechnete Zahlen")
class Scenario(BaseModel):
"""Eine Konfiguration im Vergleich – als Abweichung der Eingabedaten, nicht als eigenes Modell."""
name: str = Field(description="Kurzer Name, z. B. 'Reine Varianzminimierung'")
description: str = Field(description="Was an dieser Konfiguration anders ist, in einem Satz")
data_override: dict = Field(
description="Die abweichenden Einträge von DATA, z. B. {'lambda_1': 1, 'lambda_2': 0}")
objective_value: Optional[float] = Field(
default=None, description="Zielwert dieser Konfiguration, vom Programm berechnet")
verified_value: Optional[float] = Field(
default=None, description="Vom Runner durch erneutes Ausführen bestätigter Zielwert (ohne LLM)")
class Multistart(BaseModel):
"""Streuung über mehrere Startpunkte – bei nichtkonvexen Problemen der einzige Hinweis darauf,
ob das gefundene lokale Optimum stabil ist (Ipopt, siehe Solver-Prompt)."""
starts: int = Field(description="Anzahl versuchter Startpunkte")
converged: int = Field(description="Wie viele davon eine zulässige Lösung erreicht haben")
best: float = Field(description="Bester Zielwert (der gemeldete)")
worst: Optional[float] = Field(default=None, description="Schlechtester Zielwert der erfolgreichen Läufe")
mean: Optional[float] = Field(default=None, description="Mittelwert der erfolgreichen Zielwerte")
stddev: Optional[float] = Field(default=None, description="Standardabweichung der erfolgreichen Zielwerte")
class GraphArc(BaseModel):
"""Ein Bogen des zugrunde liegenden Graphen, mit seiner fachlichen Bedeutung."""
source: str = Field(description="Startknoten")
target: str = Field(description="Zielknoten")
weight: float = Field(description="Bogengewicht (Kosten, Dauer, Entfernung)")
class DijkstraStep(BaseModel):
"""Ein Schritt des Markierungsverfahrens: welcher Knoten dauerhaft wurde und mit welcher Marke."""
node: str
label: float = Field(description="Endgültige Entfernung dieses Knotens von der Quelle")
class GraphSolution(BaseModel):
"""Der Graph, sein optimaler Weg und der Markierungsverlauf – Grundlage für Zeichnung und Prüfung."""
nodes: List[str]
arcs: List[GraphArc]
path: List[str] = Field(default_factory=list, description="Knotenfolge des kürzesten Weges, von der Quelle zur Senke")
steps: List[DijkstraStep] = Field(
default_factory=list, description="Reihenfolge, in der Dijkstra die Knoten dauerhaft markiert hat")
node_meaning: str = Field(description="Was ein Knoten in der Aufgabe bedeutet, in einem Satz ohne Fachjargon")
arc_meaning: str = Field(description="Was ein Bogen in der Aufgabe bedeutet, in einem Satz ohne Fachjargon")
class SolverRun(BaseModel):
status: SolverStatus
objective_value: Optional[float] = None
headline: Optional[str] = Field(default=None, description="Ein Satz mit dem Ergebnis, nur berechnete Zahlen")
solution: dict = Field(default_factory=dict)
tables: List[ResultTable] = Field(default_factory=list)
constraints: List[ConstraintValue] = Field(default_factory=list)
variables: List[VariableValue] = Field(default_factory=list)
sensitivity: List[Sensitivity] = Field(default_factory=list)
data: dict = Field(default_factory=dict, description="Eingabedaten (DATA) des Programms – Grundlage für --resolve")
relaxation_bound: Optional[float] = Field(
default=None, description="Zielwert der stetigen Relaxation (nur bei ganzzahligen Modellen)")
multistart: Optional[Multistart] = Field(
default=None, description="Streuung über mehrere Startpunkte (nur bei lokalen Verfahren wie Ipopt)")
graph: Optional[GraphSolution] = Field(
default=None, description="Graph, kürzester Weg und Markierungsverlauf (nur bei Graphenproblemen)")
answers: List[Answer] = Field(
default_factory=list, description="Eine Antwort je Teilfrage der Aufgabe (Rückprojektion des Ergebnisses)")
scenarios: List[Scenario] = Field(
default_factory=list, description="Verglichene Konfigurationen (Datenvarianten desselben Modells)")
feasibility: List[str] = Field(
default_factory=list, description="Befunde der deterministischen Zulässigkeitsprüfung; leer = Lösung hält alles ein")
undeclared_data: List[str] = Field(
default_factory=list, description="Selbst erzeugte, nicht ausgewiesene Eingabedaten; leer = alles ausgewiesen")
warnings: List[str] = Field(default_factory=list, description="Deterministische Prüfungen der Ergebnistabellen")
error: Optional[str] = None
attempts: int = 1
# --- Prüfschritt: Abgleich der Annahmen mit dem Aufgabentext ---
class AssumptionVerdict(str, Enum):
JUSTIFIED = "justified" # Text ist unklar, Annahme plausibel und textkonform
UNNECESSARY = "unnecessary" # Text ist eindeutig, Annahme überflüssig (aber nicht falsch)
CONTRADICTS_TASK = "contradicts_task" # Annahme steht im Widerspruch zum Text
STRICTER_THAN_TASK = "stricter_than_task" # Modell schränkt stärker ein als der Text verlangt
WEAKER_THAN_TASK = "weaker_than_task" # Modell lässt mehr zu als der Text erlaubt
class AssumptionCheck(BaseModel):
unclear_point: str = Field(description="Die geprüfte Stelle (aus open_assumptions übernommen)")
verdict: AssumptionVerdict
task_evidence: str = Field(description="Wörtliches Zitat aus dem Aufgabentext, auf das sich das Urteil stützt")
explanation: str = Field(description="Begründung des Urteils, inkl. logischer Analyse (z. B. Implikation vs. Äquivalenz)")
correction: Optional[str] = Field(
default=None,
description="Korrigierte Annahme bzw. Formel in LaTeX; nur bei contradicts_task, stricter_than_task oder weaker_than_task",
)
class ObjectiveCheck(BaseModel):
task_evidence: str = Field(description="Wörtliches Zitat aus dem Aufgabentext, das die zu optimierende bzw. anzugebende Größe benennt")
asked_quantity: str = Field(description="Gefragte Größe mit Einheit/Maßstab, z. B. 'Standardabweichung der Portfoliorendite' oder 'Gesamtkosten in €'")
modeled_quantity: str = Field(description="Größe, die objective.latex_formula tatsächlich berechnet")
matches: bool = Field(
description="True nur, wenn die Zielfunktion genau die gefragte Größe (inkl. Richtung min/max und Maßstab) liefert; "
"eine monoton transformierte Größe (z. B. Varianz statt Standardabweichung) ist False"
)
units_consistent: bool = Field(
description="True nur, wenn alle Terme der Zielfunktion dieselbe Einheit haben (z. B. alle in Euro oder alle relativ); "
"ein dimensionsloser Term neben Euro-Beträgen ist False")
unit_explanation: str = Field(description="Einheit je Term der Zielfunktion, z. B. 'Rendite €, Risiko dimensionslos, Kosten €'")
explanation: str
correction: Optional[str] = Field(default=None, description="Korrigierte Zielfunktion in LaTeX; nur wenn matches oder units_consistent False ist")
class AssumptionReview(BaseModel):
"""Prüfbericht: Zielgröße und offene Annahmen einer Modellspezifikation gegen den Aufgabentext."""
objective_check: ObjectiveCheck
checks: List[AssumptionCheck]
missing_assumptions: List[AssumptionDefinition] = Field(
default_factory=list,
description="Unklare Stellen im Text, die das Modell stillschweigend entschieden hat, ohne sie auszuweisen",
)
@property
def needs_revision(self) -> bool:
faulty = {
AssumptionVerdict.CONTRADICTS_TASK,
AssumptionVerdict.STRICTER_THAN_TASK,
AssumptionVerdict.WEAKER_THAN_TASK,
}
return (
not self.objective_check.matches
or not self.objective_check.units_consistent
or bool(self.missing_assumptions)
or any(c.verdict in faulty for c in self.checks)
)
class CallRecord(BaseModel):
"""Ein LLM-Aufruf – für die Frage, welche Pipeline-Stufe wie lange dauert."""
step: str
seconds: float
completion_tokens: int = 0
reasoning_tokens: int = 0
provider: Optional[str] = None
model: Optional[str] = Field(default=None, description="Tatsächlich benutztes Modell (kann je Stufe abweichen)")
class LLMUsage(BaseModel):
model: str
provider: str
calls: int = 0
prompt_tokens: int = 0
completion_tokens: int = Field(default=0, description="inkl. Reasoning-Tokens")
reasoning_tokens: int = 0
cost_usd: Optional[float] = Field(default=None, description="nur OpenRouter liefert Kosten")
providers: Dict[str, int] = Field(default_factory=dict, description="Antwortender Anbieter -> Anzahl Aufrufe (OpenRouter)")
call_records: List[CallRecord] = Field(default_factory=list)
def by_step(self) -> Dict[str, dict]:
"""Sekunden und Tokens je Pipeline-Stufe."""
summary: Dict[str, dict] = {}
for record in self.call_records:
entry = summary.setdefault(record.step, {"calls": 0, "seconds": 0.0, "completion_tokens": 0, "reasoning_tokens": 0})
entry["calls"] += 1
entry["seconds"] += record.seconds
entry["completion_tokens"] += record.completion_tokens
entry["reasoning_tokens"] += record.reasoning_tokens
return summary
class PipelineResult(BaseModel):
task: str = ""
data_files: List[DataFileProfile] = Field(
default_factory=list, description="Profile der beigelegten Datendateien (--daten)")
understanding: Optional[TaskUnderstanding] = Field(
default=None, description="Ergebnis der Verstehensstufe (Teilfragen, Fakten); None bei alten Ergebnissen")
specification: OptimizationModelSpecification
review: AssumptionReview
revisions: int = Field(description="Anzahl durchgeführter Korrekturrunden")
selected_revision: int = Field(default=0, description="Runde, deren Spezifikation verwendet wurde (0 = Erstentwurf)")
restarts: int = Field(default=0, description="Anzahl kompletter Neustarts der Modellierung")
reliable: bool = Field(default=True, description="False, wenn nach allen Runden/Neustarts Zielgröße oder Konsistenz nicht stimmen")
infeasible_confirmed: bool = Field(
default=False, description="True, wenn 'unlösbar' durch eine zweite, unabhängige Modellierung bestätigt wurde")
reliability_issues: List[str] = Field(default_factory=list)
checks: List[CheckIssue] = Field(default_factory=list)
solver_code: Optional[SolverCode] = None
solver_run: Optional[SolverRun] = None
usage: Optional[LLMUsage] = None