munshi007/aura-stateFramework Python pour construire des workflows LLM en tant que machines à états avec vérification formelle via la preuve de théorèmes Z3, le model checking CTL et la prédiction conforme pour une extraction de données prouvablement correcte.
Un framework Python pour construire des workflows LLM en tant que machines à états, avec vérification formelle intégrée.
pip install git+https://github.com/munshi007/Aura-State.git
La plupart des frameworks LLM vous laissent enchaîner des appels API en espérant le meilleur. Aura-State adopte une approche différente : vous définissez votre workflow comme un graphe de nœuds, chacun avec une tâche spécifique, et le framework gère l'extraction, la vérification et le routage.
La différence clé réside dans ce qui se passe entre les nœuds :
from aura_state import AuraEngine, Node, CompiledTransition
from pydantic import BaseModel, Field
from openai import OpenAI
# Define what you want to extract
class LeadData(BaseModel):
name: str = Field(description="Full name")
budget: int = Field(description="Budget in USD")
timeline: str = Field(description="Buying timeline")
# Define a node that extracts it
class ExtractLead(Node):
system_prompt = "Extract lead info from a sales call transcript."
extracts = LeadData
def handle(self, user_text, extracted_data=None, memory=None):
return "QualifyBudget", extracted_data.model_dump()
# Define a node that does deterministic math (no LLM)
class QualifyBudget(Node):
system_prompt = "Score the lead."
sandbox_rule = "result = budget > 100000" # runs in sandboxed AST, not LLM
def handle(self, user_text, extracted_data=None, memory=None):
return "END", memory
# Wire it up
engine = AuraEngine(llm_client=OpenAI())
engine.register(ExtractLead, QualifyBudget)
engine.connect([
CompiledTransition(from_node=ExtractLead, to_node=QualifyBudget),
])
# Run
next_state, data = engine.process("ExtractLead", user_text="Hi, I'm Sarah. Budget is $450k.")
Lorsque vous appelez engine.process(), les étapes suivantes sont exécutées dans l'ordre :
1. Adaptive DAG health check → Should this node be skipped or retried?
2. GraphRAG cache lookup → Have we seen this exact input before? Skip the LLM.
3. Few-shot injection → Find similar past successes, inject as examples.
4. LLM extraction + verification → Extract data, verify with Z3, retry if wrong.
5. Your node's handle() method → Your business logic runs here.
6. MCTS Routing (UCB1) → Score branches using UCB1 + AdaptiveDAG metrics.
7. State serialization → Save state for time-travel debugging.
8. Speculative execution → Pre-compute likely next nodes in parallel.
C'est ce qui distingue réellement Aura-State des autres frameworks.
Votre graphe de nœuds est compilé en une structure de Kripke et vérifié par rapport à des propriétés de logique temporelle :
from aura_state import verify_engine, reachability, mutual_exclusion, eventual_completion
results = verify_engine(engine, [
{"description": "QualifyBudget is reachable", "formula": reachability("QualifyBudget")},
{"description": "All paths terminate", "formula": eventual_completion("QualifyBudget")},
])
# Result: PROVEN or VIOLATED, with the exact states that satisfy/violate
C'est la même technique que celle utilisée pour vérifier les circuits matériels et les systèmes de contrôle de vol (model checking CTL, Clarke et al. 1986).
Après que le LLM a extrait les valeurs, Z3 (un prouveur de théorèmes de Microsoft Research) peut prouver formellement qu'elles satisfont vos contraintes :
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
# If False, Z3 gives you a counterexample showing exactly what broke
Exécutez l'extraction plusieurs fois et obtenez des intervalles de confiance sans distribution :
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
Ceci utilise la prédiction conforme (Vovk et al., 2005) — aucune hypothèse distributionnelle requise.
Nous avons exécuté 10 transcriptions de ventes immobilières via un pipeline à 4 nœuds en utilisant GPT-4o-mini (30 appels API au total) :
Field Accuracy
────────────── ──────────
name 100%
budget 100%
bedrooms 100%
pre_approved 90%
timeline 90%
city 80%
Temporal properties: 3/3 proven
Z3 proof obligations: 20/20 passed
Routing accuracy: 90%
Avg latency: 1.4s
# Try it yourself — no API key needed
python examples/benchmark/run_benchmark.py
# With real LLM calls (needs OPENAI_API_KEY in .env)
python examples/benchmark/run_live.py --model gpt-4o-mini --runs 3
aura_state/
├── core/
│ ├── engine.py # Main engine — process() + MCTS/UCB1 routing
│ ├── adaptive_graph.py # Node health monitoring
│ ├── verification_loop.py # Extract → verify → retry loop
│ └── providers.py # Multi-model routing + cost tracking
├── compiler/
│ ├── schema_compiler.py # JSON Schema → Node classes
│ └── dspy_tuner.py # KNN few-shot selection
├── verification/
│ ├── temporal_verifier.py # Kripke + CTL model checking
│ ├── conformal.py # Conformal prediction intervals
│ └── proof_engine.py # Z3 proofs
├── execution/
│ ├── tracer.py # State serialization (time-travel debug)
│ └── sandbox.py # Safe math execution (AST validated)
├── memory/
│ ├── trajectory_cache.py # Subgraph isomorphism cache
│ └── pruner.py # Context window optimization
└── consensus/
└── auto_vote.py # Multi-run extraction with voting
pip install git+https://github.com/munshi007/Aura-State.git
Python 3.10+ requis. Dépendances : pydantic, instructor, openai, networkx, pyModelChecking, z3-solver, pyyaml.
python -m pytest tests/ -v
# 65 tests passing
MIT