munshi007/aura-statePython-Framework zum Erstellen von LLM-Workflows als Zustandsmaschinen mit formaler Verifikation durch Z3-Theorembeweise, CTL-Modellprüfung und konforme Vorhersage für nachweislich korrekte Datenextraktion.
Ein Python-Framework zur Erstellung von LLM-Workflows als State Machines mit integrierter formaler Verifikation.
pip install git+https://github.com/munshi007/Aura-State.git
Die meisten LLM-Frameworks lassen Sie API-Aufrufe verketten und hoffen auf das Beste. Aura-State verfolgt einen anderen Ansatz: Sie definieren Ihren Workflow als einen Graphen von Knoten, jeder mit einer bestimmten Aufgabe, und das Framework kümmert sich um Extraktion, Verifikation und Routing.
Der entscheidende Unterschied liegt darin, was zwischen den Knoten passiert:
from aura_state import AuraEngine, Node, CompiledTransition
from pydantic import BaseModel, Field
from openai import OpenAI
# Definieren Sie, was extrahiert werden soll
class LeadData(BaseModel):
name: str = Field(description="Vollständiger Name")
budget: int = Field(description="Budget in USD")
timeline: str = Field(description="Kaufzeitplan")
# Definieren Sie einen Knoten, der es extrahiert
class ExtractLead(Node):
system_prompt = "Extrahiere Lead-Informationen aus einem Verkaufsgespräch-Transkript."
extracts = LeadData
def handle(self, user_text, extracted_data=None, memory=None):
return "QualifyBudget", extracted_data.model_dump()
# Definieren Sie einen Knoten, der deterministische Mathematik durchführt (kein LLM)
class QualifyBudget(Node):
system_prompt = "Bewerte den Lead."
sandbox_rule = "result = budget > 100000" # läuft im abgesicherten AST, nicht im LLM
def handle(self, user_text, extracted_data=None, memory=None):
return "END", memory
# Verdrahten Sie es
engine = AuraEngine(llm_client=OpenAI())
engine.register(ExtractLead, QualifyBudget)
engine.connect([
CompiledTransition(from_node=ExtractLead, to_node=QualifyBudget),
])
# Ausführen
next_state, data = engine.process("ExtractLead", user_text="Hallo, ich bin Sarah. Budget ist 450.000$.")
Wenn Sie engine.process() aufrufen, durchläuft es diese Schritte in dieser Reihenfolge:
1. Adaptiver DAG-Health-Check → Soll dieser Knoten übersprungen oder wiederholt werden?
2. GraphRAG-Cache-Lookup → Haben wir diese genaue Eingabe schon gesehen? LLM überspringen.
3. Few-Shot-Injektion → Ähnliche frühere Erfolge finden, als Beispiele einfügen.
4. LLM-Extraktion + Verifikation → Daten extrahieren, mit Z3 verifizieren, bei Fehlern wiederholen.
5. Ihre handle()-Methode des Knotens → Ihre Geschäftslogik läuft hier.
6. MCTS-Routing (UCB1) → Zweige mit UCB1 + AdaptiveDAG-Metriken bewerten.
7. Zustandsserialisierung → Zustand für Time-Travel-Debugging speichern.
8. Spekulative Ausführung → Wahrscheinliche nächste Knoten parallel vorausberechnen.
Das ist es, was Aura-State tatsächlich von anderen Frameworks unterscheidet.
Ihr Knoten-Graph wird in eine Kripke-Struktur kompiliert und gegen temporallogische Eigenschaften geprüft:
from aura_state import verify_engine, reachability, mutual_exclusion, eventual_completion
results = verify_engine(engine, [
{"description": "QualifyBudget ist erreichbar", "formula": reachability("QualifyBudget")},
{"description": "Alle Pfade terminieren", "formula": eventual_completion("QualifyBudget")},
])
# Ergebnis: PROVEN oder VIOLATED, mit den exakten Zuständen, die erfüllen/verletzen
Dies ist die gleiche Technik, die zur Verifikation von Hardware-Schaltkreisen und Flugsteuerungssystemen verwendet wird (CTL-Modellprüfung, Clarke et al. 1986).
Nachdem das LLM Werte extrahiert hat, kann Z3 (ein Theorembewerber von Microsoft Research) formal beweisen, dass sie Ihre Bedingungen erfüllen:
from aura_state import prove_extraction
result = prove_extraction(
{"budget": 450000, "cost_per_sqft": 3, "total": 1350000},
obligations=["budget > 0", "total == budget * cost_per_sqft"],
)
# result.verified = True
# Wenn False, gibt Z3 ein Gegenbeispiel, das genau zeigt, was gebrochen ist
Führen Sie die Extraktion mehrmals durch und erhalten Sie verteilungsfreie Konfidenzintervalle:
from aura_state import conformal_interval
budgets = [450000, 452000, 448000, 450000, 451000]
ci = conformal_interval(budgets, confidence=0.95)
# ci.lower = 447800, ci.upper = 452200
Dies verwendet konforme Vorhersage (Vovk et al., 2005) — keine Verteilungsannahmen erforderlich.
Wir haben 10 Immobilien-Verkaufsgespräch-Transkripte durch eine 4-Knoten-Pipeline mit GPT-4o-mini laufen lassen (insgesamt 30 API-Aufrufe):
Feld Genauigkeit
────────────── ──────────
name 100%
budget 100%
bedrooms 100%
pre_approved 90%
timeline 90%
city 80%
Temporeigenschaften: 3/3 nachgewiesen
Z3-Proof-Obligations: 20/20 bestanden
Routing-Genauigkeit: 90%
Durchschn. Latenz: 1,4s
# Probieren Sie es selbst — kein API-Schlüssel erforderlich
python examples/benchmark/run_benchmark.py
# Mit echten LLM-Aufrufen (benötigt OPENAI_API_KEY in .env)
python examples/benchmark/run_live.py --model gpt-4o-mini --runs 3
aura_state/
├── core/
│ ├── engine.py # Haupt-Engine — process() + MCTS/UCB1-Routing
│ ├── adaptive_graph.py # Knoten-Health-Monitoring
│ ├── verification_loop.py # Extraktion → Verifikation → Wiederholungs-Schleife
│ └── providers.py # Multi-Modell-Routing + Kostenverfolgung
├── compiler/
│ ├── schema_compiler.py # JSON-Schema → Knotenklassen
│ └── dspy_tuner.py # KNN-Few-Shot-Auswahl
├── verification/
│ ├── temporal_verifier.py # Kripke + CTL-Modellprüfung
│ ├── conformal.py # Konforme Vorhersageintervalle
│ └── proof_engine.py # Z3-Beweise
├── execution/
│ ├── tracer.py # Zustandsserialisierung (Time-Travel-Debug)
│ └── sandbox.py # Sichere Mathematik-Ausführung (AST-validiert)
├── memory/
│ ├── trajectory_cache.py # Subgraph-Isomorphie-Cache
│ └── pruner.py # Kontextfenster-Optimierung
└── consensus/
└── auto_vote.py # Multi-Run-Extraktion mit Abstimmung
pip install git+https://github.com/munshi007/Aura-State.git
Python 3.10+ erforderlich. Abhängigkeiten: pydantic, instructor, openai, networkx, pyModelChecking, z3-solver, pyyaml.
python -m pytest tests/ -v
# 65 Tests erfolgreich
MIT